Enabling a Capnp Server to accept parallel requests

[legolas145]

Hello there,
I am pretty new to capnp.I have a Capnp Client and Server that I have written in cpp.I want the capnp server to accept and process parallel requests that the client makes.Heres what my server code looks like

//Api implementation of a interface named getSuggestions which gives some autocomplete suggestions for a word 
::kj::Promise<void> ServerApiImplClass::getSuggestions(SuggestionContext context)
{

//Suggestion logic

//Set the response in context after the suggestion logic

 //Return a fulfilled promise
  return kj::READY_NOW;
}

//Starting the server
int main() {
   capnp::EzRpcServer server(kj::heap<ServerApiImplClass>(), "unix:/tmp/server.sock”);
    auto& wait_scope = server.getWaitScope();
   kj::NEVER_DONE.wait(wait_scope);
}

I also have a capnp client that I am using to test

//Invoking the server to get the results
void DisplaySuggestions(string input) {
  APIResponse api_res;

  capnp::EzRpcClient rpc_client("unix:/tmp/server.sock”);
  auto server_client = rpc_client.getMain<Suggestion>();
  auto &wait_scope = server_client.getWaitScope();

  auto req = server_client.suggestionRequest();
  req.setQuery(input);
  auto reqPromise = req.send();  
  auto res = reqPromise.wait(wait_scope);
  //Print the results
}

//Giving parallel requests to server
int main() {
   thread suggestion_thread_1(&DisplaySuggestions,”Sta”);
  thread suggestion_thread_2(&DisplaySuggestions,”Mic”);
  suggestion_thread_1.join();
  suggestion_thread_2.join();
}

I notice that I am not getting the requests parallely in my capnp server.It seems that only after a request has been processed by the server, I am getting the next request.So I assume the requests are getting somewhere queued in between capnp client and server.

Does capnp internally queues the requests? I want to understand how can I modify my code so that my server can accept parallel requests that the client can make.Any help would be appreciated :)

[kentonv]

Cap'n Proto uses event loop concurrency. You need to use the KJ event loop and Promises to perform any work asynchronously, so that other requests are allowed to arrive in the meantime. See:

https://github.com/capnproto/capnproto/blob/v2/kjdoc/tour.md#asynchronous-event-loop

In your code, you attempt to create two threads that wait on the same socket. The second thread probably fails with an "address in use" error. Only one thread can be listening on a socket at a time.

[legolas145]

Hey @kentonv ,Thanks for the reply.
I modified the capnp api implementation to something like this

// Api implementation of a interface named getSuggestions which gives some autocomplete suggestions
// for a word
::kj::Promise<void> ServerApiImplClass::getSuggestions(SuggestionContext context) {
  // Creating a new promise fulfiller
  auto paf = kj::newPromiseAndFulfiller<vector<string>>();

  // Async Suggestion logic that internally spawn a thread and fulfills the promise fulfiller.This function returns immediately.
  GetAutocompleteSuggestions(context.getParams().getQuery(), kj::mv(paf));

  // Set the response in context after the suggestion logic
  return paf.promise.then([context = kj::mv(context)](vector<string> suggestions){
    //Populate the suggestions into context result	
    })
}

But I am still not understanding as to how should I make this run in an event loop.I am using a EzRPC to spawn my server instance.I assume that internally has its own event looping mechanism.

[kentonv]

GetAutocompleteSuggestions would need to use KJ promises internally for any I/O it performs, in order to yield to the event loop while it waits.

Alternatively, you could use newPromiseAndCrossThreadFulfiller, and then have GetAutocompleteSuggestions run in a separate thread.

[legolas145]

Ok got it @kentonv. As you had pointed out,I was not using the cross thread fulfiller.Rather just a normal fulfiller to fulfill the promise.Now I see that my server is able to process and respond to parallel requests.Thanks for the assistance :)