0

如何将 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.
4

1 回答 1

0

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

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

  1. 您可以使用kj::Executor来安排代码在不同线程的事件循环上运行。如果您使用，调用线程不需要是 KJ 事件循环线程executeSync()- 但是，此函数会阻塞，直到另一个线程有机会唤醒并执行该函数。我不确定这在实践中的表现如何。如果这是一个问题，则可能有空间扩展Executor接口以更有效地处理此用例。

  2. 您可以通过管道或套接字对传递消息来在线程之间进行通信（但以这种方式发送大消息会涉及到套接字缓冲区/从套接字缓冲区进行大量不必要的复制）。

  3. 您可以使用管道、信号或 (on Linux) 向另一个线程的事件循环发出信号以唤醒eventfd，然后让它在受互斥体保护的队列中查找消息。（但kj::Executor大多数情况下已经过时了这种技术。）

  4. 虽然不容易，但可以调整 KJ 的事件循环以在其他事件循环之上运行，以便两者都可以在同一个线程中运行。例如，node-capnp 将 KJ 适配为在 libuv 之上运行。

于 2020-07-05T17:10:37.700 回答