JVMCBC-1657 Better backpressure for row-based HTTP services

Motivation
----------
`autoRead` was activated every time an item was requested,
even if there was an existing surplus of buffered rows.
This could lead to reading and buffering more rows than the
consumer requested, potentially without limit.

The problem was most pronounced when result rows are small
enough that several fit in a single network receive buffer.

Modifications
-------------
Rename `requested` to `demand`, because the value
decreases as demand is met (when rows are supplied).

Instead of enabling autoRead whenever an item is requested,
enable it only if demand is positive.

Prevent a race condition by synchronizing access
to the `demand` counter and the autoRead setting.

Change-Id: Ia82ed52a7182b380efaec81d3a70812d21d340df
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/229015
Tested-by: Build Bot <[email protected]>
Reviewed-by: Michael Reiche <[email protected]>

Author: dnault

core-io/src/main/java/com/couchbase/client/core/io/netty/chunk/BaseChunkResponseParser.java

@@ -33,7 +33,6 @@
 
 import java.nio.ByteBuffer;
 import java.util.Optional;
-import java.util.concurrent.atomic.AtomicLong;
 
 import static com.couchbase.client.core.Reactor.emitFailureHandler;
 
@@ -79,9 +78,17 @@ public abstract class BaseChunkResponseParser<H extends ChunkHeader, ROW extends
   private Sinks.Many<ROW> rowSink;
 
   /**
-   * Holds the currently user-requested rows for backpressure handling.
+   * For backpressure, a running total of the number of rows requested by the subscriber,
+   * minus the number of rows emitted. A negative value indicates a surplus of available rows.
+   * <p>
+   * A value of {@code Long.MAX_VALUE} indicates backpressure is disabled.
    */
-  private final AtomicLong requested = new AtomicLong(0);
+  private long demand = 0;
+
+  /**
+   * Guards access to `demand` and channel autoRead setting.
+   */
+  private final Object autoReadLock = new Object();
 
   /**
    * Subclass implements this to return the "meat" of the decoding, the chunk parser.
@@ -174,23 +181,58 @@ public void initialize(final ChannelConfig channelConfig) {
     parser = parserBuilder().build();
     this.channelConfig = channelConfig;
     this.trailer = Sinks.one();
-    this.requested.set(0);
+    synchronized (autoReadLock) {
+      this.demand = 0;
+    }
 
     this.rowSink = Sinks.many().unicast().onBackpressureBuffer();
     this.rows = rowSink
       .asFlux()
       .doOnRequest(v -> {
-        requested.addAndGet(v);
-        if (!channelConfig.isAutoRead()) {
-          channelConfig.setAutoRead(true);
+        synchronized (autoReadLock) {
+          if (demand == Long.MAX_VALUE) {
+            return;
+          }
+
+          try {
+            demand = Math.addExact(demand, v);
+
+          } catch (ArithmeticException e) {
+            // They asked for it, so open the floodgates.
+            demand = Long.MAX_VALUE;
+            channelConfig.setAutoRead(true);
+            return;
+          }
+
+          if (demand > 0 && !channelConfig.isAutoRead()) {
+            channelConfig.setAutoRead(true);
+          }
         }
       })
-      .doOnTerminate(() -> channelConfig.setAutoRead(true))
-      .doOnCancel(() -> channelConfig.setAutoRead(true))
+      .doOnTerminate(this::readRemainingRowsWithoutBackpressure)
+      .doOnCancel(this::readRemainingRowsWithoutBackpressure)
       .publish()
       .refCount();
   }
 
+  /**
+   * Allow the remaining result rows to be consumed, even in the absence of demand.
+   * Must not be called before the row subscription is terminated or cancelled.
+   * <p>
+   * From a memory usage perspective, this is safe because future attempts to emit
+   * rows into the sink fail instead of buffering the items.
+   * <p>
+   * Do this because the current API contract requires publishing metadata
+   * even if the row sink terminates or is cancelled. Otherwise, we would
+   * probably be better off simply closing the HTTP connection.
+   */
+  private void readRemainingRowsWithoutBackpressure() {
+    synchronized (autoReadLock) {
+      demand = Long.MAX_VALUE;
+      channelConfig.setAutoRead(true);
+    }
+  }
+
   @Override
   public Flux<ROW> rows() {
     return rows;
@@ -241,9 +283,17 @@ public Optional<CouchbaseException> decodingFailure() {
    */
   protected void emitRow(final ROW row) {
     rowSink.emitNext(row, emitFailureHandler());
-    requested.decrementAndGet();
-    if (requested.get() <= 0 && channelConfig.isAutoRead() && rowSink.currentSubscriberCount() > 0) {
-      channelConfig.setAutoRead(false);
+
+    synchronized (autoReadLock) {
+      if (demand == Long.MAX_VALUE) {
+        return;
+      }
+
+      demand--;
+
+      if (demand <= 0 && channelConfig.isAutoRead()) {
+        channelConfig.setAutoRead(false);
+      }
     }
   }