RATIS-2509. Introduce local read API to reduce serde cost

[ivandika3]

What changes were proposed in this pull request?

See https://issues.apache.org/jira/browse/RATIS-2509

OM code integration

    public OMResponse submitReadRequest(OMRequest omRequest, Function<OMRequest, OMResponse> handler)
      throws ServiceException {
    try {
      return serverDivision.readOnlyAsync(() -> CompletableFuture.completedFuture(handler.apply(omRequest))).get();
    } catch (ExecutionException | IOException ex) {
      throw new ServiceException(ex.getMessage(), ex);
    } catch (InterruptedException ex) {
      Thread.currentThread().interrupt();
      throw new ServiceException(ex.getMessage(), ex);
    }

  }

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2509

How was this patch tested?

OM Leader read performance testing, the Ratis read performance is on par with the direct read.

• Before patch: 235020 QPS -> 180433 QPS (24% QPS reduction)
• After patch: 235020 QPS -> 228362 (3% QPS reduction)

Follower read throughput remains unchanged, so might not be the main overhead.

[szetszwo]

@ivandika3 , thanks for working on this! Please see the comments inlined and also https://issues.apache.org/jira/secure/attachment/13082106/1448_review.patch

[szetszwo]

• Since the query method in StateMachine does not throw IOException, let's just use Supplier here. It will force the StateMachine to handle IOException, if there is any.

//StateMachine
  CompletableFuture<Message> query(Message request);

• Also, we need clientId and readRequestType to support read-after-write-consistency.

    default <T> CompletableFuture<T> readOnlyAsync(ClientId clientId, ReadRequestTypeProto readRequestType,
        Supplier<CompletableFuture<T>> query) throws IOException {

[ivandika3]

Good point, updated.

[szetszwo]

+1 the change looks good.

[ivandika3]

@szetszwo Thanks for the review. Currently I'm trying to wait until my GH actions are enabled before merging it since I haven't seen any green CI for this patch. If we can find another way, that's good. Otherwise, we can wait until my GH actions situation is resolved.

[ivandika3]

I wonder whether it is better to use thenComposeAsync to improve performance.

return getReadIndexForReadOnly(clientId, readRequestType)
          .thenComposeAsync(readIndex -> getReadRequests().waitToAdvance(readIndex), clientExecutor)
          .thenComposeAsync(readIndex -> supplyReadOnly(query), clientExecutor);

or just the supplyReadOnly

return getReadIndexForReadOnly(clientId, readRequestType)
          .thenCompose(readIndex -> getReadRequests().waitToAdvance(readIndex))
          .thenComposeAsync(readIndex -> supplyReadOnly(query), clientExecutor);

Seems currently with thenCompose, the actual read is going to be executed sequentially in gRPC SerializingExecutor.

Let me benchmark it. Can also be applied to readAsync.

[szetszwo]

We should use clientExecutor as earliest as possible. It will keep using the executor for the later calls; see below:

//output
                                    main: main
                             supplyAsync: pool-A-thread1
        thenCompose after pool-A-thread1: pool-A-thread1
   thenComposeAsync after pool-A-thread1: pool-B-thread1
  thenCompose again after pool-B-thread1: pool-B-thread1

  public static void main(String[] args) {
    final ExecutorService poolA = Executors.newCachedThreadPool(ConcurrentUtils.newThreadFactory("pool-A"));
    final ExecutorService poolB = Executors.newCachedThreadPool(ConcurrentUtils.newThreadFactory("pool-B"));

    printCurrentThread("main");
    CompletableFuture.supplyAsync(() -> printCurrentThread("supplyAsync"), poolA)
        .thenCompose(s -> CompletableFuture.completedFuture(printCurrentThread("thenCompose after " + s)))
        .thenComposeAsync(s -> CompletableFuture.completedFuture(printCurrentThread("thenComposeAsync after " + s)), poolB)
        .thenCompose(s -> CompletableFuture.completedFuture(printCurrentThread("thenCompose again after " + s)))
        ;
  }

  static String printCurrentThread(String label) {
    final String name = Thread.currentThread().getName();
    System.out.printf("%40s: %s%n", label, name);
    return name;
  }

[spacemonkd]

Isn't this going to bypass the writeIndexCache?
Say if a caller constructs a ReadRequestTypeProto with readAfterWriteConsistent as true and passes it to readOnlyAsync, isn't the flag getting ignored when the node is the leader?

Maybe we can add a Javadoc saying readAfterWriteConsistent is not supported here, or we can throw an exception like IllegalArgumentException in case readAfterWriteConsistent flag is set so that we get some visibility into this behaviour?

[spacemonkd]

Just one suggestion.
As far as I understand from the code ClientId.randomId() is not used at all for the default path. It goes straight to checkLeaderStateForReadOnly and then supplyReadOnly.

Maybe we can overload the Division interface with something like:

 default <T> CompletableFuture<T> readOnlyAsync(Supplier<CompletableFuture<T>> query) throws IOException {
     return readOnlyAsync(ClientId.randomId(), ReadRequestTypeProto.getDefaultInstance(), query);
   }

so that callers can avoid importing ClientId and RaftClientRequest?
Right now it is only one call site so it is fine I guess, but if we have more callers this might be a better way.

[spacemonkd]

Thanks @ivandika3 for picking this up.
I had a few questions / suggestions and your inputs would be great, also it seems the checkstyle is failing. We should be good to merge once that is addressed.

Also on the test coverage I asked Claude to analyse the scenarios and it gave the following cases not covered yet.

Untested paths for readOnlyAsync:

• Follower in LINEARIZABLE mode when leader is unknown (the ReadIndexException at line 1143)
• Follower in LINEARIZABLE mode during snapshot installation (the ReadException at line 1138)
• Supplier that returns null (the Objects.requireNonNull at line 1161)
• Supplier that throws (the catch (Throwable t) at line 1163)
• Server not in RUNNING state (the assertLifeCycleState at line 1173)

Maybe it might be good to add these cases as well?