c++ - 如何将 Cap'n'Proto 线程与非 Cap'n'Proto 线程集成？

Question

如何正确地将 Cap'n'Proto 客户端使用与周围的多线程代码集成？ Cap'n'Proto 文档表示每个 Cap'n'Proto 接口(interface)都是单线程的，具有专用的事件循环。此外，他们建议使用 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 线程履行 promise 时崩溃。

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.

Answer 1

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

解决此问题的一些方法包括:

1. 您可以 use kj::Executor to schedule code to run on a different thread's event loop.如果您使用executeSync()，则调用线程不需要是KJ事件循环线程——但是，该函数会阻塞，直到另一个线程有机会唤醒并执行该函数。我不确定这在实践中表现如何；如果这是一个问题，可能有空间扩展 Executor 接口(interface)以更有效地处理此用例。

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

3. 您可以使用管道、信号或(在 Linux 上)eventfd 向另一个线程的事件循环发出信号以唤醒。 ，然后让它在互斥锁保护的队列中查找消息。 (但是 kj::Executor 基本上已经废弃了这种技术。)

4. 调整 KJ 的事件循环以在其他事件循环之上运行，以便两者可以在同一线程中运行，尽管并不容易，但还是有可能的。例如，node-capnp adapts KJ to run on top of libuv.