如何将 Cap'n'Proto 线程与非 Cap'n'Proto 线程集成？答案

【问题标题】：How to integrate Cap'n'Proto threads with non Cap'n'Proto threads?如何将 Cap'n'Proto 线程与非 Cap'n'Proto 线程集成？
【发布时间】：2020-06-30 22:29:27
【问题描述】：

如何将 Cap'n'Proto 客户端使用与周围的多线程代码正确集成？ Cap'n'Proto 文档说每个 Cap'n'Proto 接口都是带有专用事件循环的单线程。此外，他们建议使用 Cap'n'Proto 在线程之间进行通信。但是，文档似乎没有描述非 Cap'n'Proto 线程（例如 UI 循环）如何与之集成。即使可以在某些地方将 Cap'n'Proto 事件循环与 UI 循环集成，其他模型，如线程池（Android Binder、全局 libdispatch 队列）似乎更具挑战性。

我认为解决方案是将客户端线程的线程执行器缓存在非capnp线程将访问它的同步位置。

我相信虽然调用线程也总是需要在自己的事件循环中才能与它们结合，但我只是想确保情况确实如此。我最初尝试在一个简单的单元测试中这样做失败了。我创建了一个 KjLooperEventPort 类（遵循节点 libuv 适配器的结构）以在 Android 上结合 KJ 和 ALooper。

那么我的测试代码是：

TEST(KjLooper, CrossThreadPromise) {
  std::thread::id kjThreadId;
  ConditionVariable<const kj::Executor*> executorCv{nullptr};
  ConditionVariable<std::pair<bool, kj::Promise<void>>> looperThreadFinished{false, nullptr};

  std::thread looperThread([&] {
    auto looper = android::newLooper();
    android::KjLooperEventPort kjEventPort{looper};
    kj::WaitScope waitScope(kjEventPort.getKjLoop());

    auto finished = kj::newPromiseAndFulfiller<void>();
    looperThreadFinished.constructValueAndNotifyAll(true, kj::mv(finished.promise));

    executorCv.waitNotValue(nullptr);

    auto executor = executorCv.readCopy();
    kj::Promise<void> asyncPromise = executor->executeAsync([&] {
      ASSERT_EQ(std::this_thread::get_id(), kjThreadId);
    });
    asyncPromise = asyncPromise.then([tid = std::this_thread::get_id(), kjThreadId, &finished] {
      std::cerr << "Running promise completion on original thread\n";
      ASSERT_NE(tid, kjThreadId);
      ASSERT_EQ(std::this_thread::get_id(), tid);
      std::cerr << "Fulfilling\n";
      finished.fulfiller->fulfill();
      std::cerr << "Fulfilled\n";
    });
    asyncPromise.wait(waitScope);
  });

  std::thread kjThread([&] {
    kj::Promise<void> finished = kj::NEVER_DONE;
    looperThreadFinished.wait([&](auto& promise) {
      finished = kj::mv(promise.second);
      return promise.first;
    });

    auto ioContext = kj::setupAsyncIo();
    kjThreadId = std::this_thread::get_id();
    executorCv.setValueAndNotifyAll(&kj::getCurrentThreadExecutor());
    finished.wait(ioContext.waitScope);
  });

  looperThread.join();
  kjThread.join();
}

这会在履行返回给 kj 线程的承诺时崩溃。

terminating with uncaught exception of type kj::ExceptionImpl: kj/async.c++:1269: failed: expected threadLocalEventLoop == &loop || threadLocalEventLoop == nullptr; Event armed from different thread than it was created in.  You must use
 Executor to queue events cross-thread.

【问题讨论】：

标签： c++ capnproto

【解决方案1】：

大多数 Cap'n Proto RPC 和 KJ Promise 相关对象只能在创建它们的线程中访问。例如，如您所见，解决 promise 跨线程将失败。

您可以解决此问题的一些方法包括：

如果你使用executeSync()，你可以use kj::Executor to schedule code to run on a different thread's event loop.调用线程不需要是一个KJ事件循环线程——但是，这个函数会阻塞，直到另一个线程有机会唤醒并执行该函数。我不确定这在实践中的表现如何。如果有问题，可能还有空间扩展Executor 接口以更有效地处理此用例。
您可以通过管道或套接字对传递消息在线程之间进行通信（但以这种方式发送大消息会涉及到套接字缓冲区/从套接字缓冲区进行大量不必要的复制）。
您可以使用管道、信号或（在 Linux 上）eventfd 向另一个线程的事件循环发出信号以唤醒，然后让它在受互斥体保护的队列中查找消息。（但kj::Executor 基本上已经过时了这种技术。）
虽然不容易，但可以调整 KJ 的事件循环以在其他事件循环之上运行，以便两者可以在同一个线程中运行。例如node-capnp adapts KJ to run on top of libuv.

【讨论】：