Removing writerLock in favor of setting a flag to call flushBuffer (#1208)

* Removing writerLock in favor of setting a flag to call flushBuffer
* Instead of using writerLock around flushBuffer (socket flush), set a flag to tell the sendMessageBatch to flushBuffer after a write, instead of locking around send.

* better place to flush

* fixed api doc

* don't catch flush exception as it's likely a legit issue

Author: scottf

src/main/java/io/nats/client/Connection.java

@@ -539,8 +539,8 @@ enum Status {
     String createInbox();
 
     /**
-     * Immediately flushes the underlying connection buffer if the connection is valid.
-     * @throws IOException the connection flush fails
+     * Flushes the underlying connection buffer the next chance it gets if the connection is valid.
+     * @throws IOException not applicable even though it's part of the signature due to implementation change
      */
     void flushBuffer() throws IOException;
 

src/main/java/io/nats/client/impl/NatsConnectionWriter.java

@@ -37,7 +37,6 @@ class NatsConnectionWriter implements Runnable {
 
     private final NatsConnection connection;
 
-    private final ReentrantLock writerLock;
     private Future<Boolean> stopped;
     private Future<DataPort> dataPortFuture;
     private DataPort dataPort;
@@ -51,10 +50,10 @@ class NatsConnectionWriter implements Runnable {
     private final MessageQueue outgoing;
     private final MessageQueue reconnectOutgoing;
     private final long reconnectBufferSize;
+    private final AtomicBoolean flushBuffer;
 
     NatsConnectionWriter(NatsConnection connection, NatsConnectionWriter sourceWriter) {
         this.connection = connection;
-        writerLock = new ReentrantLock();
 
         this.running = new AtomicBoolean(false);
         this.reconnectMode = new AtomicBoolean(sourceWriter != null);
@@ -77,6 +76,8 @@ class NatsConnectionWriter implements Runnable {
         reconnectOutgoing = new MessageQueue(true, options.getRequestCleanupInterval(),
             sourceWriter == null ? null : sourceWriter.reconnectOutgoing);
         reconnectBufferSize = options.getReconnectBufferSize();
+
+        flushBuffer = new AtomicBoolean(false);
     }
 
     // Should only be called if the current thread has exited.
@@ -122,60 +123,58 @@ boolean isRunning() {
     }
 
     void sendMessageBatch(NatsMessage msg, DataPort dataPort, StatisticsCollector stats) throws IOException {
-        writerLock.lock();
-        try {
-            int sendPosition = 0;
-            int sbl = sendBufferLength.get();
-
-            while (msg != null) {
-                long size = msg.getSizeInBytes();
-
-                if (sendPosition + size > sbl) {
-                    if (sendPosition > 0) {
-                        dataPort.write(sendBuffer, sendPosition);
-                        connection.getNatsStatistics().registerWrite(sendPosition);
-                        sendPosition = 0;
-                    }
-                    if (size > sbl) { // have to resize b/c can't fit 1 message
-                        sbl = bufferAllocSize((int) size, BUFFER_BLOCK_SIZE);
-                        sendBufferLength.set(sbl);
-                        sendBuffer = new byte[sbl];
-                    }
-                }
+        int sendPosition = 0;
+        int sbl = sendBufferLength.get();
 
-                ByteArrayBuilder bab = msg.getProtocolBab();
-                int babLen = bab.length();
-                System.arraycopy(bab.internalArray(), 0, sendBuffer, sendPosition, babLen);
-                sendPosition += babLen;
+        while (msg != null) {
+            long size = msg.getSizeInBytes();
 
-                sendBuffer[sendPosition++] = CR;
-                sendBuffer[sendPosition++] = LF;
+            if (sendPosition + size > sbl) {
+                if (sendPosition > 0) {
+                    dataPort.write(sendBuffer, sendPosition);
+                    connection.getNatsStatistics().registerWrite(sendPosition);
+                    sendPosition = 0;
+                }
+                if (size > sbl) { // have to resize b/c can't fit 1 message
+                    sbl = bufferAllocSize((int) size, BUFFER_BLOCK_SIZE);
+                    sendBufferLength.set(sbl);
+                    sendBuffer = new byte[sbl];
+                }
+            }
 
-                if (!msg.isProtocol()) {
-                    sendPosition += msg.copyNotEmptyHeaders(sendPosition, sendBuffer);
+            ByteArrayBuilder bab = msg.getProtocolBab();
+            int babLen = bab.length();
+            System.arraycopy(bab.internalArray(), 0, sendBuffer, sendPosition, babLen);
+            sendPosition += babLen;
 
-                    byte[] bytes = msg.getData(); // guaranteed to not be null
-                    if (bytes.length > 0) {
-                        System.arraycopy(bytes, 0, sendBuffer, sendPosition, bytes.length);
-                        sendPosition += bytes.length;
-                    }
+            sendBuffer[sendPosition++] = CR;
+            sendBuffer[sendPosition++] = LF;
 
-                    sendBuffer[sendPosition++] = CR;
-                    sendBuffer[sendPosition++] = LF;
-                }
+            if (!msg.isProtocol()) {
+                sendPosition += msg.copyNotEmptyHeaders(sendPosition, sendBuffer);
 
-                stats.incrementOutMsgs();
-                stats.incrementOutBytes(size);
+                byte[] bytes = msg.getData(); // guaranteed to not be null
+                if (bytes.length > 0) {
+                    System.arraycopy(bytes, 0, sendBuffer, sendPosition, bytes.length);
+                    sendPosition += bytes.length;
+                }
 
-                msg = msg.next;
+                sendBuffer[sendPosition++] = CR;
+                sendBuffer[sendPosition++] = LF;
             }
 
-            dataPort.write(sendBuffer, sendPosition);
-            connection.getNatsStatistics().registerWrite(sendPosition);
+            stats.incrementOutMsgs();
+            stats.incrementOutBytes(size);
+
+            msg = msg.next;
         }
-        finally {
-            writerLock.unlock();
+
+        // no need to write if there are no bytes
+        if (sendPosition > 0) {
+            dataPort.write(sendBuffer, sendPosition);
         }
+
+        connection.getNatsStatistics().registerWrite(sendPosition);
     }
 
     @Override
@@ -197,11 +196,14 @@ public void run() {
                     msg = this.outgoing.accumulate(sendBufferLength.get(), maxAccumulate, waitForMessage);
                 }
 
-                if (msg == null) { // Make sure we are still running
-                    continue;
+                if (msg != null) {
+                    sendMessageBatch(msg, dataPort, stats);
                 }
 
-                sendMessageBatch(msg, dataPort, stats);
+                if (flushBuffer.get()) {
+                    flushBuffer.set(false);
+                    dataPort.flush();
+                }
             }
         } catch (IOException | BufferOverflowException io) {
             // if already not running, an IOE is not unreasonable in a transition state
@@ -237,18 +239,8 @@ void queueInternalMessage(NatsMessage msg) {
     }
 
     void flushBuffer() {
-        // Since there is no connection level locking, we rely on synchronization
-        // of the APIs here.
-        writerLock.lock();
-        try {
-            if (this.running.get()) {
-                dataPort.flush();
-            }
-        } catch (Exception e) {
-            // NOOP;
-        }
-        finally {
-            writerLock.unlock();
+        if (running.get()) {
+            flushBuffer.set(true);
         }
     }
 }