remove ResumptionState to reduce allocations

mostroverkhov · mostroverkhov · commit 8a15df05676c · 2019-04-06T02:00:28.000+03:00
Signed-off-by: Maksym Ostroverkhov &lt;m.ostroverkhov@gmail.com&gt;
diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java b/rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java
@@ -81,19 +81,20 @@ public ClientRSocketSession(
             multiplexer -> {
               /*reconnect resumable connection*/
               reconnect(multiplexer.asClientServerConnection());
-
-              ResumptionState state = resumableConnection.state();
+              long impliedPosition = resumableConnection.impliedPosition();
+              long position = resumableConnection.position();
               logger.debug(
-                  "Client ResumableConnection reconnected. Sending RESUME frame with state: {}",
-                  state);
+                  "Client ResumableConnection reconnected. Sending RESUME frame with state: [impliedPos: {}, pos: {}]",
+                  impliedPosition,
+                  position);
               /*Connection is established again: send RESUME frame to server, listen for RESUME_OK*/
               sendFrame(
                       ResumeFrameFlyweight.encode(
                           allocator,
                           /*retain so token is not released once sent as part of resume frame*/
                           resumeToken.retain(),
-                          state.impliedPosition(),
-                          state.position()))
+                          impliedPosition,
+                          position))
                   .then(multiplexer.asSetupConnection().receive().next())
                   .subscribe(this::resumeWith);
             },
@@ -112,19 +113,21 @@ public ClientRSocketSession continueWith(Mono<? extends ResumeAwareConnection> n
   @Override
   public ClientRSocketSession resumeWith(ByteBuf resumeOkFrame) {
     logger.debug("ResumeOK FRAME received");
-    ResumptionState resumptionState = stateFromFrame(resumeOkFrame);
+    long remotePos = remotePos(resumeOkFrame);
+    long remoteImpliedPos = remoteImpliedPos(resumeOkFrame);
+    resumeOkFrame.release();
+
     resumableConnection.resume(
-        resumptionState,
+        remotePos,
+        remoteImpliedPos,
         pos ->
             pos.then()
                 /*Resumption is impossible: send CONNECTION_ERROR*/
                 .onErrorResume(
                     err ->
                         sendFrame(
                                 ErrorFrameFlyweight.encode(
-                                    allocator,
-                                    0,
-                                    errorFrameThrowable(resumptionState.impliedPosition())))
+                                    allocator, 0, errorFrameThrowable(remoteImpliedPos)))
                             .then(Mono.fromRunnable(resumableConnection::dispose))
                             /*Resumption is impossible: no need to return control to ResumableConnection*/
                             .then(Mono.never())));
@@ -156,10 +159,12 @@ private Mono<Void> sendFrame(ByteBuf frame) {
     return resumableConnection.sendOne(frame).onErrorResume(err -> Mono.empty());
   }
 
-  private static ResumptionState stateFromFrame(ByteBuf resumeOkFrame) {
-    long impliedPos = ResumeOkFrameFlyweight.lastReceivedClientPos(resumeOkFrame);
-    resumeOkFrame.release();
-    return ResumptionState.fromServer(impliedPos);
+  private static long remoteImpliedPos(ByteBuf resumeOkFrame) {
+    return ResumeOkFrameFlyweight.lastReceivedClientPos(resumeOkFrame);
+  }
+
+  private static long remotePos(ByteBuf resumeOkFrame) {
+    return -1;
   }
 
   private static ConnectionErrorException errorFrameThrowable(long impliedPos) {
diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java
@@ -119,9 +119,9 @@ public void reconnect(ResumeAwareConnection connection) {
   /*after receiving RESUME (Server) or RESUME_OK (Client)
   calculate and send resume frames */
   public void resume(
-      ResumptionState peerResumptionState, Function<Mono<Long>, Mono<Void>> resumeFrameSent) {
+      long remotePos, long remoteImpliedPos, Function<Mono<Long>, Mono<Void>> resumeFrameSent) {
     /*race between sendFrame and doResume may lead to duplicate frames on resume store*/
-    actions.onNext(new Resume(peerResumptionState, resumeFrameSent));
+    actions.onNext(new Resume(remotePos, remoteImpliedPos, resumeFrameSent));
   }
 
   @Override
@@ -152,6 +152,10 @@ public Flux<ByteBuf> receive() {
                 .onErrorResume(err -> Mono.never()));
   }
 
+  public long position() {
+    return resumableFramesStore.framePosition();
+  }
+
   @Override
   public long impliedPosition() {
     return resumableFramesStore.frameImpliedPosition();
@@ -209,11 +213,6 @@ private void sendFrame(ByteBuf f) {
     }
   }
 
-  ResumptionState state() {
-    return new ResumptionState(
-        resumableFramesStore.framePosition(), resumableFramesStore.frameImpliedPosition());
-  }
-
   Flux<Throwable> connectionErrors() {
     return connectionErrors;
   }
@@ -237,20 +236,30 @@ private void doResumeStart(ResumeAwareConnection connection) {
   }
 
   private void doResume(
-      ResumptionState peerResumptionState, Function<Mono<Long>, Mono<Void>> sendResumeFrame) {
-    ResumptionState localResumptionState = state();
+      long remotePosition,
+      long remoteImpliedPosition,
+      Function<Mono<Long>, Mono<Void>> sendResumeFrame) {
+    long localPosition = position();
+    long localImpliedPosition = impliedPosition();
 
     logger.debug(
         "Resumption start. Calculating implied pos using: {}",
         resumedFramesCalculator.getClass().getSimpleName());
     logger.debug(
-        "Resumption states. local: {}, remote: {}", localResumptionState, peerResumptionState);
-
-    Mono<Long> res = resumedFramesCalculator.calculate(localResumptionState, peerResumptionState);
+        "Resumption states. local: [pos: {}, impliedPos: {}], remote: [pos: {}, impliedPos: {}]",
+        localPosition,
+        localImpliedPosition,
+        remotePosition,
+        remoteImpliedPosition);
+
+    Mono<Long> res =
+        resumedFramesCalculator.calculate(
+            localPosition, localImpliedPosition,
+            remotePosition, remoteImpliedPosition);
     Mono<Long> localImpliedPos =
         res.doOnSuccess(notUsed -> state = State.RESUME)
             .doOnSuccess(this::releaseFramesToPosition)
-            .map(remoteImpliedPos -> localResumptionState.impliedPosition());
+            .map(remoteImpliedPos -> localImpliedPosition);
 
     sendResumeFrame
         .apply(localImpliedPos)
@@ -355,18 +364,20 @@ public void run() {
   }
 
   class Resume implements Runnable {
-    private final ResumptionState peerResumptionState;
+    private final long remotePos;
+    private final long remoteImpliedPos;
     private final Function<Mono<Long>, Mono<Void>> resumeFrameSent;
 
     public Resume(
-        ResumptionState peerResumptionState, Function<Mono<Long>, Mono<Void>> resumeFrameSent) {
-      this.peerResumptionState = peerResumptionState;
+        long remotePos, long remoteImpliedPos, Function<Mono<Long>, Mono<Void>> resumeFrameSent) {
+      this.remotePos = remotePos;
+      this.remoteImpliedPos = remoteImpliedPos;
       this.resumeFrameSent = resumeFrameSent;
     }
 
     @Override
     public void run() {
-      doResume(peerResumptionState, resumeFrameSent);
+      doResume(remotePos, remoteImpliedPos, resumeFrameSent);
     }
   }
 
diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumeStateException.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumeStateException.java
@@ -18,19 +18,32 @@
 
 class ResumeStateException extends RuntimeException {
   private static final long serialVersionUID = -5393753463377588732L;
-  private final ResumptionState local;
-  private final ResumptionState remote;
+  private final long localPos;
+  private final long localImpliedPos;
+  private final long remotePos;
+  private final long remoteImpliedPos;
 
-  public ResumeStateException(ResumptionState local, ResumptionState remote) {
-    this.local = local;
-    this.remote = remote;
+  public ResumeStateException(
+      long localPos, long localImpliedPos, long remotePos, long remoteImpliedPos) {
+    this.localPos = localPos;
+    this.localImpliedPos = localImpliedPos;
+    this.remotePos = remotePos;
+    this.remoteImpliedPos = remoteImpliedPos;
   }
 
-  public ResumptionState localState() {
-    return local;
+  public long getLocalPos() {
+    return localPos;
   }
 
-  public ResumptionState remoteState() {
-    return remote;
+  public long getLocalImpliedPos() {
+    return localImpliedPos;
+  }
+
+  public long getRemotePos() {
+    return remotePos;
+  }
+
+  public long getRemoteImpliedPos() {
+    return remoteImpliedPos;
   }
 }
diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumedFramesCalculator.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumedFramesCalculator.java
@@ -23,30 +23,39 @@ interface ResumedFramesCalculator {
   ResumedFramesCalculator ofClient = new ClientResumedFramesCalculator();
   ResumedFramesCalculator ofServer = new ServerResumedFramesCalculator();
 
-  Mono<Long> calculate(ResumptionState local, ResumptionState remote);
+  Mono<Long> calculate(long localPos, long localImpliedPos, long remotePos, long remoteImpliedPos);
 
   class ClientResumedFramesCalculator implements ResumedFramesCalculator {
 
+    /*ResumptionState clientState, ResumptionState serverState*/
     @Override
-    public Mono<Long> calculate(ResumptionState clientState, ResumptionState serverState) {
-      long serverImplied = serverState.impliedPosition();
-      if (serverImplied >= clientState.position()) {
-        return Mono.just(serverImplied);
+    public Mono<Long> calculate(
+        long clientPos, long clientImpliedPos, long serverPos, long serverImpliedPos) {
+      if (serverImpliedPos >= clientPos) {
+        return Mono.just(serverImpliedPos);
       } else {
-        return Mono.error(new ResumeStateException(clientState, serverState));
+        return Mono.error(
+            new ResumeStateException(
+                clientPos, clientImpliedPos,
+                serverPos, serverImpliedPos));
       }
     }
   }
 
   class ServerResumedFramesCalculator implements ResumedFramesCalculator {
 
+    /*ResumptionState serverState, ResumptionState clientState*/
     @Override
-    public Mono<Long> calculate(ResumptionState serverState, ResumptionState clientState) {
-      boolean clientStateValid = clientState.position() <= serverState.impliedPosition();
-      boolean serverStateValid = serverState.position() <= clientState.impliedPosition();
+    public Mono<Long> calculate(
+        long serverPos, long serverImpliedPos, long clientPos, long clientImpliedPos) {
+      boolean clientStateValid = clientPos <= serverImpliedPos;
+      boolean serverStateValid = serverPos <= clientImpliedPos;
       return clientStateValid && serverStateValid
-          ? Mono.just(clientState.impliedPosition())
-          : Mono.error(new ResumeStateException(serverState, clientState));
+          ? Mono.just(clientImpliedPos)
+          : Mono.error(
+              new ResumeStateException(
+                  serverPos, serverImpliedPos,
+                  clientPos, clientImpliedPos));
     }
   }
 }
diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java b/rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java
@@ -96,8 +96,13 @@ public ServerRSocketSession continueWith(ResumeAwareConnection newConnection) {
   @Override
   public ServerRSocketSession resumeWith(ByteBuf resumeFrame) {
     logger.debug("Resume FRAME received");
+    long remotePos = remotePos(resumeFrame);
+    long remoteImpliedPos = remoteImpliedPos(resumeFrame);
+    resumeFrame.release();
+
     resumableConnection.resume(
-        stateFromFrame(resumeFrame),
+        remotePos,
+        remoteImpliedPos,
         pos ->
             pos.flatMap(
                     impliedPos -> sendFrame(ResumeOkFrameFlyweight.encode(allocator, impliedPos)))
@@ -135,11 +140,12 @@ private Mono<Void> sendFrame(ByteBuf frame) {
     return resumableConnection.sendOne(frame).onErrorResume(e -> Mono.empty());
   }
 
-  private static ResumptionState stateFromFrame(ByteBuf resumeFrame) {
-    long peerPos = ResumeFrameFlyweight.firstAvailableClientPos(resumeFrame);
-    long peerImpliedPos = ResumeFrameFlyweight.lastReceivedServerPos(resumeFrame);
-    resumeFrame.release();
-    return ResumptionState.fromClient(peerPos, peerImpliedPos);
+  private static long remotePos(ByteBuf resumeFrame) {
+    return ResumeFrameFlyweight.firstAvailableClientPos(resumeFrame);
+  }
+
+  private static long remoteImpliedPos(ByteBuf resumeFrame) {
+    return ResumeFrameFlyweight.lastReceivedServerPos(resumeFrame);
   }
 
   private static RejectedResumeException errorFrameThrowable(Throwable err) {
@@ -148,8 +154,11 @@ private static RejectedResumeException errorFrameThrowable(Throwable err) {
       ResumeStateException resumeException = ((ResumeStateException) err);
       msg =
           String.format(
-              "resumption_pos=[ remote: %s, local: %s]",
-              resumeException.remoteState(), resumeException.localState());
+              "resumption_pos=[ remote: { pos: %d, impliedPos: %d }, local: { pos: %d, impliedPos: %d }]",
+              resumeException.getRemotePos(),
+              resumeException.getRemoteImpliedPos(),
+              resumeException.getLocalPos(),
+              resumeException.getLocalImpliedPos());
     } else {
       msg = String.format("resume_internal_error: %s", err.getMessage());
     }
diff --git a/rsocket-core/src/test/java/io/rsocket/resume/ResumeCalculatorTest.java b/rsocket-core/src/test/java/io/rsocket/resume/ResumeCalculatorTest.java
@@ -34,47 +34,37 @@ void setUp() {
 
   @Test
   void clientResumeSuccess() {
-    ResumptionState local = ResumptionState.fromClient(1, 42);
-    ResumptionState remote = ResumptionState.fromServer(3);
-    StepVerifier.create(clientResumeCalculator.calculate(local, remote))
+    StepVerifier.create(clientResumeCalculator.calculate(1, 42, -1, 3))
         .expectNext(3L)
         .expectComplete()
         .verify(Duration.ofSeconds(1));
   }
 
   @Test
   void clientResumeError() {
-    ResumptionState local = ResumptionState.fromClient(4, 42);
-    ResumptionState remote = ResumptionState.fromServer(3);
-    StepVerifier.create(clientResumeCalculator.calculate(local, remote))
+    StepVerifier.create(clientResumeCalculator.calculate(4, 42, -1, 3))
         .expectError(ResumeStateException.class)
         .verify(Duration.ofSeconds(1));
   }
 
   @Test
   void serverResumeSuccess() {
-    ResumptionState local = ResumptionState.fromClient(1, 42);
-    ResumptionState remote = ResumptionState.fromClient(4, 23);
-    StepVerifier.create(serverResumeCalculator.calculate(local, remote))
+    StepVerifier.create(serverResumeCalculator.calculate(1, 42, 4, 23))
         .expectNext(23L)
         .expectComplete()
         .verify(Duration.ofSeconds(1));
   }
 
   @Test
   void serverResumeErrorClientState() {
-    ResumptionState local = ResumptionState.fromClient(1, 3);
-    ResumptionState remote = ResumptionState.fromClient(4, 23);
-    StepVerifier.create(serverResumeCalculator.calculate(local, remote))
+    StepVerifier.create(serverResumeCalculator.calculate(1, 3, 4, 23))
         .expectError(ResumeStateException.class)
         .verify(Duration.ofSeconds(1));
   }
 
   @Test
   void serverResumeErrorServerState() {
-    ResumptionState local = ResumptionState.fromClient(4, 42);
-    ResumptionState remote = ResumptionState.fromClient(4, 1);
-    StepVerifier.create(serverResumeCalculator.calculate(local, remote))
+    StepVerifier.create(serverResumeCalculator.calculate(4, 42, 4, 1))
         .expectError(ResumeStateException.class)
         .verify(Duration.ofSeconds(1));
   }
