Is it possible to Run a MultiThread Rpc Server? about activej HOT 2 OPEN

Hi, @PanagiotisDrakatos

It is indeed possible to launch a multithreaded RPC server. Here is a demo launcher that does just that.

However, looking at your project, I can see that the issue is that your code is blocking the eventloop thread.

The code in RpcRequestHandler is executed inside the eventloop thread. So any blocking operations will make eventloop unavailable until that operation is completed. This is why it appears that requests are processed sequentially. You should avoid blocking code, such as I/O or Thread.sleep(), inside the eventloop.

For example, let’s rewrite the RpcErasureServer#downloadErasureChunks method. It has blocking code inside the while loop (Thread.sleep()). The same logic could be achieved by using Promises. For example, the Promises#loop method, which can be used to write asynchronous loops. Your code would look like this:

private RpcRequestHandler<ErasureRequest, ErasureResponse> downloadErasureChunks(IChunksService<T> service) {
   return request -> Promises.loop(0,
               rc -> rc < ConsensusConfiguration.CYCLES &&
                    (serializable_length == 0 ||
                     CachedSerializableErasureObject.getInstance().getSerializableErasureObject() == null),
               rc -> Promises.delay(ConsensusConfiguration.HEARTBEAT_INTERVAL, rc + 1)
         )
         .map(rc -> {
            if (rc == ConsensusConfiguration.CYCLES)
               return new ErasureResponse(null);
            T result = service.downloadErasureChunks();
            if (result == null)
               return new ErasureResponse(null);
            return new ErasureResponse(this.valueMapper.encode(result, serializable_length));
         });
}

Look at the Promises#loop call in place of the while cycle. Promises#loop takes an initial seed value 0 (rc=0). It also takes a loop break condition as its second argument. Finally, we pass a function that converts the current rc value to the next rc value (rc + 1), but it also returns the next value after a delay of HEARTBEAT_INTERVAL milliseconds. A delay does not cause a thread to block (like with Thread.sleep()), but rather schedules a task to the eventloop. So, other requests/tasks can be executed while the eventloop is waiting for the delay to pass.

Notice that we call service.downloadErasureChunks() inside the eventloop. I looked at this method and did not find blocking operations, so it’s fine to call it from the eventloop thread. However, if downloadErasure() would access DB, file system, or perform any other I/O operations, you would need to execute it using an external Executor. I noticed there is some blocking code in RPC handlers located in the RpcAdrestusServer class. You might consider rewriting them, and wrapping blocking code using Promise#ofBlocking methods.

You can look at this example for reference. Here we wrap blocking calls to JDBC API using Promise#ofBlocking with an external Executor.

from activej.

@eduard-vasinskyi I have no words to say excellent as always you gave solutions to the point. Thank you very much. I appreciate it. God bless you

from activej.