CapnProto - 在回调中向其他服务器请求答案

【问题标题】：CapnProto - Request to other server in callbackCapnProto - 在回调中向其他服务器请求
【发布时间】：2022-10-03 06:35:56
【问题描述】：

我是 CapnProto 的新手。
我想在server1的回调中请求server2的一个函数。
但我得到了如下异常。
请帮我解决它。
非常感谢！

* thread #1, queue = \'com.apple.main-thread\', stop reason = EXC_BAD_ACCESS (code=1, address=0x3000000000020)
  * frame #0: 0x000000010035859c libcapnp-rpc-0.10.2.dylib`capnp::VatNetwork<capnp::rpc::twoparty::VatId, capnp::rpc::twoparty::ProvisionId, capnp::rpc::twoparty::RecipientId, capnp::rpc::twoparty::ThirdPartyCapId, capnp::rpc::twoparty::JoinResult>::baseConnect(capnp::AnyStruct::Reader) + 20
    frame #1: 0x0000000100361a2c libcapnp-rpc-0.10.2.dylib`kj::_::TransformPromiseNode<kj::_::Void, kj::Own<kj::AsyncIoStream>, capnp::EzRpcClient::Impl::Impl(kj::StringPtr, unsigned int, capnp::ReaderOptions)::\'lambda\'(kj::Own<kj::AsyncIoStream>&&), kj::_::PropagateException>::getImpl(kj::_::ExceptionOrValue&) + 512
    frame #2: 0x00000001004d2220 libkj-async-0.10.2.dylib`kj::_::RunnableImpl<kj::_::TransformPromiseNodeBase::get(kj::_::ExceptionOrValue&)::$_31>::run() + 32
    frame #3: 0x000000010028aaa4 libkj-0.10.2.dylib`kj::_::runCatchingExceptions(kj::_::Runnable&) + 40
    frame #4: 0x00000001004c7e48 libkj-async-0.10.2.dylib`kj::_::TransformPromiseNodeBase::get(kj::_::ExceptionOrValue&) + 64
    frame #5: 0x00000001004c8684 libkj-async-0.10.2.dylib`kj::_::ForkHubBase::fire() + 60
    frame #6: 0x00000001004c654c libkj-async-0.10.2.dylib`kj::_::waitImpl(kj::Own<kj::_::PromiseNode>&&, kj::_::ExceptionOrValue&, kj::WaitScope&, kj::SourceLocation) + 608
    frame #7: 0x0000000100005198 client`kj::Promise<capnp::Response<SampleServer1::CallbackRegisterResults> >::wait(kj::WaitScope&, kj::SourceLocation) + 120
    frame #8: 0x0000000100004b94 client`main + 344
    frame #9: 0x000000010003d08c dyld`start + 520

源代码实现示例如下

SampleServer1.capnp：

interface SampleServer1 {

    callbackRegister @0 (callback :Callback) -> ();   //to register a callback

    interface Callback {
        calbackFunc @0 (in :Int32) -> ();
    }
}

SampleServer1::Server 实现：

class SampleServer1Impl : public SampleServer1::Server
{
::kj::Promise<void> callbackRegister(CallbackRegisterContext context){
    auto cb = context.getParams().getCallback());

    auto request = cb.calbackFuncRequest();   //Call callback function
    request.setIn(111);
    auto promise = request.send();

    return kj::READY_NOW;
  }
}

SampleServer1::Callback::Server 实现：

class CallbackImpl : public SampleServer1::Callback::Server
{
::kj::Promise<void> calbackFunc(CalbackFuncContext context){

    capnp::EzRpcClient ezClient2(\"unix:/tmp/capnp-server-2\");
    SampleServer2::Client client2 = ezClient2.getMain<SampleServer2>();

    auto& waitScope = ezClient2.getWaitScope();
    {
      auto request = client2.functionSampleRequest();  //Request to SERVER2
      request.setIn(222);
      auto promise = request.send();
     
      promise.wait(waitScope);
      
    }

    return kj::READY_NOW;
  }
}

SampleServer2.capnp：

interface SampleServer2 {

    functionSample @0 (in :Int32) -> ();
}

SampleServer2::Server 实现

class SampleServer2 Impl : public SampleServer2::Server
{
::kj::Promise<void> functionSample(FunctionSampleContext context){
    //Do something
    return kj::READY_NOW;
  }
}

客户端实现

  capnp::EzRpcClient ezClient(\"unix:/tmp/capnp-server-1\");
  SampleServer1::Client client = ezClient.getMain<SampleServer1>();
  auto& waitScope = ezClient.getWaitScope();

 ::SampleServer1::Callback::Client callback = ::SampleServer1::Callback::Client(kj::heap<CallbackImpl>());

  auto request = client.callbackRegisterRequest();      //Register a callback to Server1
  request.setCallback(callback);
  auto promise = request.send();

标签： c++ capnproto

【解决方案1】：

calbackFunc 的实现存在问题。 calbackFunc 本身是从 NEVER_DONE.wait （事件循环）中执行的，因此在此函数中等待 promise 将进行嵌套等待。这在 capnp 中是不允许的。

您可以通过在另一个线程上执行此操作来避免这种情况。前任：

  ::kj::Promise<void> calbackFunc(CalbackFuncContext context){
    
    kj::Thread th([](){
        capnp::EzRpcClient ezClient2("unix:/tmp/capnp-server-2");
        Server2::Client client2 = ezClient2.getMain<SampleServer2>();
        auto& waitScope = ezClient2.getWaitScope();
        {
          auto request = client2.functionSampleRequest();  //Request to SERVER2
          request.setIn(222);
          auto promise = request.send();
         
          promise.wait(waitScope);
          
        }
    });
    return kj::READY_NOW;
  }

【讨论】：

如果你创建一个线程，崩溃将发生异常。请参考：stackoverflow.com/questions/53850808/…
您提到的链接中的用例与此不同。客户端在一个线程中创建，并在另一个线程中发送客户端请求。 Capnp 在客户端初始化时使用线程变量将事件循环附加到线程，因此只有在该线程中才可以使用事件循环。所以在这种情况下，就会发生异常。在我提到的解决方案中，客户端在同一个线程中创建和发送请求，没有问题。
@DuyNguyễn 您已经正确识别了这里的问题，但是使用线程作为解决方案并不是最好的方法。相反，正确的做法是将promise.wait() 替换为promise.then()，并在promise 完成时向其传递一个lambda 以调用。 .then() 将返回另一个承诺，这是 callbackFunc() 应该返回的整体（而不是 READY_NOW）。
@KentonVarda：感谢您的宝贵回答。我会试试。
@KentonVarda 我同意，线程的使用只是说明问题的一个例子。但是，在这种情况下，用then 替换wait 可能不起作用，因为之后EzRpcClient 超出了范围。这可以通过使用kj::heap 分配它并将其移动到then lambda 来避免。

【解决方案2】：

与肯顿在其他答案的 cmets 中的建议类似，以下内容对我有用：

  ::kj::Promise<void> calbackFunc(CalbackFuncContext context){
    auto ezClient2 = kj::heap<capnp::EzRpcClient>("unix:/tmp/capnp-server-2");
    SampleServer2::Client client2 = ezClient2->getMain<SampleServer2>();

    auto request = client2.functionSampleRequest();  //Request to SERVER2
    request.setIn(222);
    return request.send().attach(kj::mv(ezClient2)).ignoreResult();

笔记：

我返回由request.send() 生成的promise，而不是像您使用kj::READY_NOW 那样创建一个新的promise。
我将.attachezClient2，使其保持在request.send() 承诺的范围内。亲自尝试：删除.attach() 并查看它是否崩溃。阅读附件here in the KJ tour。
我.ignoreResult()，只是为了将request.send() 承诺转换为kj::Promise<void>。它仍然会正常完成，但它只是方便地将类型转换为我们可以从calbackFunc 返回的东西。

现在，假设您想在对functionSample() 的请求完成后做一些事情。您可以使用.then 注册一个回调以在完成时调用（请参阅documentation here）：

  ::kj::Promise<void> calbackFunc(CalbackFuncContext context){
    auto ezClient2 = kj::heap<capnp::EzRpcClient>("unix:/tmp/capnp-server-2");
    SampleServer2::Client client2 = ezClient2->getMain<SampleServer2>();

    auto request = client2.functionSampleRequest();  //Request to SERVER2
    request.setIn(222);

    return request.send().ignoreResult()
        .then([]() {
          // DO SOMETHING IN THE CONTINUATION HERE
        }).attach(kj::mv(ezClient2));
  }

【讨论】：