Capnproto - Make client request in server callback

Question

I need to do some client-requests inside a server-callback and are not sure where to store the created capnp::EzRpcClient and CompareNetwork::Client comparer objects. That's because the clients go out of scope (I think - I just get a SEGFAULT, but that seems to be the reason). It's basically a master who forwards a loading request to it's slaves (slaves can register via a REG-request and their address is stored).
So - where/how should I store the client objects? Is there any "best-practice" with this? I think temporarily storing them in some class member variable is a little dirty, isn't it?

GroupMaster2.cpp:

kj::Promise<void> GroupMaster2::CompareMasterImpl::load(LoadContext context) {
    auto loadData = context.getParams().getIds();

    slaveListLock.lock();

    auto promList = kj::Vector<kj::Promise<std::pair<std::string, CompareNetwork::Status>>>();

    for(auto& slave : slaveList) {
        try {
            capnp::EzRpcClient client(slave.second->address);
            CompareNetwork::Client comparer = client.getMain<CompareNetwork>();
            auto request = comparer.loadRequest();
            std::string addrCopy(slave.first);
            request.setIds(loadData);
            auto loadPromise = request.send();
            promList.add(loadPromise.then([addrCopy](auto response) {
                return std::make_pair(addrCopy, response.getStatus());
            },
            [&, addrCopy](kj::Exception && exception) {
                slaveListLock.lock();
                slaveList.erase(addrCopy);//something failed, remove this slave!
                slaveListLock.unlock();
                std::cout << "ErrLoad0: " << std::string(exception.getDescription()) << std::endl;
                return std::make_pair(addrCopy, CompareNetwork::Status::ERROR);
            }));
        }
        catch(...) {
            std::cout << "Error sending load to: " << slave.first << std::endl;
        }
    }

    slaveListLock.unlock();

    auto retProm = kj::joinPromises(promList.releaseAsArray()).then([&, KJ_CPCAP(context)](kj::Array<std::pair<std::string, CompareNetwork::Status>> res) mutable {
        bool error = false;

        for(auto& loadRes : res) {
            switch(loadRes.second) {
                case CompareNetwork::Status::OK: {
                        std::cout << "LOAD OK: " << loadRes.first << std::endl;
                        break;
                    }

                case CompareNetwork::Status::ERROR: {
                        std::cout << "LOAD ERROR: " << loadRes.first << std::endl;
                        error = true;
                        break;
                    }
            }
        }
        if(!error)
            context.getResults().setStatus(CompareNetwork::Status::OK);
        else
            context.getResults().setStatus(CompareNetwork::Status::ERROR);
    }, [](kj::Exception && exception) {
        std::cout << __FILE__ << ":" << __LINE__ << std::string(exception.getDescription()) << std::endl;
    }).eagerlyEvaluate([](kj::Exception && exception) {
        std::cout << __FILE__ << ":" << __LINE__ << std::string(exception.getDescription()) << std::endl;
    });
    std::cout << "ReturnedLoad" << std::endl;
    return retProm;
}

GroupNetworkData.capnp:

interface CompareNetwork {
  compare @0 (jobs :JobPartList) -> (res :JobResList);
  load @1 (ids :WIDList) -> (status :Status);

  struct JobPartList {
    jobParts @0 :List(JobPart);

    struct JobPart {
      jobs @0 :List(Job);
      startID @1 :UInt32;

      struct Job {
        wid1 @0 :UInt32;
        wid2 @1 :UInt32;
      }
    }
  }

  struct JobResList {
    jobResults @0 :List(JobRes);
    struct JobRes {
      jobIndex @0 :UInt32;
      result @1 :Float64;
    }
  }

  struct WIDList {
    ids @0 :List(WID);
    struct WID {
      id @0 :UInt32;
    }
  }

  enum Status {
    ok @0;
    error @1;
  }
}

interface CompareMaster extends(CompareNetwork) {
  reg @0 (data :SlaveData) -> (status :CompareNetwork.Status);
  struct SlaveData {
    perfInd @0 :Float64;
    maxMem @1 :UInt32;
    address @2 :Text;
  }
}

Thanks in advance! dvs23

score 0 · Answer 1 · answered Apr 07 '18 at 22:48

C++14 seems to have the answer:
How to capture a unique_ptr into a lambda expression?
Allocate both objects on heap with kj::heap and move the kj::Own to the callback lambda:

for(auto& slave : slaveList) {
    try {
        auto client = kj::heap<capnp::EzRpcClient>(slave.second->address);
        auto comparer = kj::heap<CompareNetwork::Client>(client->getMain<CompareNetwork>());
        auto request = comparer->loadRequest();
        std::string addrCopy(slave.first);
        request.setIds(loadData);
        auto loadPromise = request.send();
        promList.add(loadPromise.then([addrCopy, client = std::move(client), comparer = std::move(comparer)](auto response) {
            return std::make_pair(addrCopy, response.getStatus());
        },
        [&, addrCopy](kj::Exception && exception) {
            slaveListLock.lock();
            slaveList.erase(addrCopy);//something failed, remove this slave!
            slaveListLock.unlock();
            std::cout << "ErrLoad0: " << std::string(exception.getDescription()) << std::endl;
            return std::make_pair(addrCopy, CompareNetwork::Status::ERROR);
        }));
    }
    catch(...) {
        std::cout << "Error sending load to: " << slave.first << std::endl;
    }
}

Better ideas are welcome, but at the moment it seems to work pretty well :)

Capnproto - Make client request in server callback

1 Answers1