🤱🏻 🧑‍🤝‍🧑 🖕🏽 So, you want to optimize gRPC. Part 2 🛶 🙆 😺

For future students on the course "Java Developer. Basic" we prepared a translation of the article.

We also invite everyone to an open webinar for this course, where the participants, together with the teacher, will analyze how the Internet works.

gRPC? , , . , 60%. , 10 000%.

, "-" Java. , , , . , .

KvService. RPC, , , . synchronized, , RPC:

private final Map<ByteBuffer, ByteBuffer> store = new HashMap<>();

@Override
public synchronized void create(
    CreateRequest request, StreamObserver<CreateResponse> responseObserver) {
  ByteBuffer key = request.getKey().asReadOnlyByteBuffer();
  ByteBuffer value = request.getValue().asReadOnlyByteBuffer();
  simulateWork(WRITE_DELAY_MILLIS);
  if (store.putIfAbsent(key, value) == null) {
    responseObserver.onNext(CreateResponse.getDefaultInstance());
    responseObserver.onCompleted();
    return;
  }
  responseObserver.onError(Status.ALREADY_EXISTS.asRuntimeException());
}

, : RPC! , . .

, RPC. , RPC, . «-» , . "foo", "bar". , . , .

, . synchronized

Java , KvService. create

. Map. HashMap , , HashMap . synchronized

.

Java : ConcurrentHashMap

. Map. , . , , . putIfAbsent

, , , , .

ConcurrentMap

putIfAbsent

, HashMap

ConcurrentHashMap

synchronized

:

private final ConcurrentMap<ByteBuffer, ByteBuffer> store = new ConcurrentHashMap<>();

@Override
public void create(
    CreateRequest request, StreamObserver<CreateResponse> responseObserver) {
  ByteBuffer key = request.getKey().asReadOnlyByteBuffer();
  ByteBuffer value = request.getValue().asReadOnlyByteBuffer();
  simulateWork(WRITE_DELAY_MILLIS);
  if (store.putIfAbsent(key, value) == null) {
    responseObserver.onNext(CreateResponse.getDefaultInstance());
    responseObserver.onCompleted();
    return;
  }
  responseObserver.onError(Status.ALREADY_EXISTS.asRuntimeException());
}

create

. retrieve

delete

. update

. , :

@Override
public synchronized void update(
    UpdateRequest request, StreamObserver<UpdateResponse> responseObserver) {
  ByteBuffer key = request.getKey().asReadOnlyByteBuffer();
  ByteBuffer newValue = request.getValue().asReadOnlyByteBuffer();
  simulateWork(WRITE_DELAY_MILLIS);
  ByteBuffer oldValue = store.get(key);
  if (oldValue == null) {
    responseObserver.onError(Status.NOT_FOUND.asRuntimeException());
    return;
  }
  store.replace(key, oldValue, newValue);
  responseObserver.onNext(UpdateResponse.getDefaultInstance());
  responseObserver.onCompleted();
}

store

:

, ConcurrentMap

. , Map, . , , , .

, , replace

. replace

true

, . (ConcurrentMap

, , , !) do-while

:

@Override
public void update(
    UpdateRequest request, StreamObserver<UpdateResponse> responseObserver) {
  // ...
  ByteBuffer oldValue;
  do {
    oldValue = store.get(key);
    if (oldValue == null) {
      responseObserver.onError(Status.NOT_FOUND.asRuntimeException());
      return;
    }
  } while (!store.replace(key, oldValue, newValue));
  responseObserver.onNext(UpdateResponse.getDefaultInstance());
  responseObserver.onCompleted();
}

, RPC store.get()

store.replace()

, . , . .

, : RPC . , API, . , , .

, ListenableFuture

. 100 RPC. , — . , . RPC , , . , :

$ ./gradlew installDist
$ time ./build/install/kvstore/bin/kvstore
Apr 16, 2018 10:38:42 AM io.grpc.examples.KvRunner runClient
INFO: Did 24.067 RPCs/s

real	1m0.886s
user	0m9.340s
sys	0m1.660s

Apr 16, 2018 10:36:48 AM io.grpc.examples.KvRunner runClient
INFO: Did 2,449.8 RPCs/s

real	1m0.968s
user	0m52.184s
sys	0m20.692s

! 24 RPC , 2400. API, . API. «-», , .

user

. 9 , 9 60 . 5 , 52 . , . KvServer , . , - . RPC, . , 12 , 12- . !

, :

Apr 16, 2018 10:38:40 AM io.grpc.examples.KvClient$3 onFailure
INFO: Key not found
io.grpc.StatusRuntimeException: NOT_FOUND

, . 100 RPC, . , API. , , .

gRPC-. , . , . .

"Java Developer. Basic".

« ».

So, you want to optimize gRPC. Part 2

More articles: