如何将帽‘n’threads线程与非帽‘n’threads线程集成？

Question

如何正确地将Cap‘multi客户端的使用与周围的多线程代码集成起来？Cap‘event文档指出每个Cap’n‘event接口都是单线程的，其中有一个专用的事件循环。此外，他们建议使用Cap‘using在线程之间进行通信。然而，文档似乎没有描述非Cap‘could线程(例如UI循环)如何与其集成。即使可以在某些地方将Cap‘pools事件循环与UI循环集成，其他模型(比如线程池(Android，全局lib分派队列))似乎更具有挑战性。

我认为解决方案是将客户端线程的线程执行器缓存到非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.

Answer 1

大多数can RPC和KJ承诺相关的对象只能在创建它们的线程中访问。例如，如你所见，解决一个承诺--交叉线--将失败。

解决这一问题的一些方法包括：

1. 如果使用安排在不同线程的事件循环上运行的代码。，则调用线程不需要是KJ事件循环线程--但是，该函数会阻塞，直到另一个线程有机会唤醒并执行该函数。我不知道这在实践中会有多好；如果这是一个问题，可能有空间扩展Executor接口，以更有效地处理这个用例。
2. 您可以通过管道或套接字对传递消息，从而在线程之间进行通信(但以这种方式发送大消息将涉及到对套接字缓冲区的大量不必要的复制)。
3. 您可以使用管道、信号或(在Linux上) eventfd通知另一个线程的事件循环，然后让它在受互斥锁保护的队列中查找消息。(但kj::Executor基本上淘汰了这一技术。)
4. 调整KJ的事件循环以在其他事件循环之上运行是可能的，尽管并不容易，这样两者就可以在同一个线程中运行。例如，使KJ在libuv之上运行。