perf(qwp): batch WebSocket sends and cut per-frame syscalls

core/src/main/java/io/questdb/client/Sender.java

@@ -1118,6 +1118,10 @@ final class LineSenderBuilder {
         private final IntList ports = new IntList();
         private long authTimeoutMillis = QwpWebSocketSender.DEFAULT_AUTH_TIMEOUT_MS;
         private int autoFlushBytes = PARAMETER_NOT_SET_EXPLICITLY;
+        // Set by disableAutoFlush(). auto_flush_interval=off also stores the interval as
+        // Integer.MAX_VALUE, so the WebSocket validation keys on this flag to reject only the
+        // full disable and accept the interval-only opt-out.
+        private boolean autoFlushDisabled;
         private int autoFlushIntervalMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private int autoFlushRows = PARAMETER_NOT_SET_EXPLICITLY;
         private int bufferCapacity = PARAMETER_NOT_SET_EXPLICITLY;
@@ -1865,7 +1869,8 @@ public LineSenderBuilder connectionListenerInboxCapacity(int capacity) {
          * It is essential to ensure the maximum buffer capacity is sufficient to accommodate all locally buffered data.
          * <p>
          * This method should only be used when communicating via the HTTP protocol. Calling this method is illegal when
-         * communicating over the TCP protocol.
+         * communicating over the TCP or WebSocket protocol. On WebSocket, {@code auto_flush_interval=off} turns off
+         * only the time-based trigger.
          *
          * @return this instance for method chaining
          * @see #autoFlushRows(int)
@@ -1883,6 +1888,7 @@ public LineSenderBuilder disableAutoFlush() {
 
             this.autoFlushRows = AUTO_FLUSH_DISABLED;
             this.autoFlushIntervalMillis = Integer.MAX_VALUE;
+            this.autoFlushDisabled = true;
             return this;
         }
 
@@ -3561,9 +3567,14 @@ private QwpWebSocketSender buildWebSocket(HttpTokenProvider tokenProvider) {
 
             int actualAutoFlushRows = autoFlushRows == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_ROWS : autoFlushRows;
             int actualAutoFlushBytes = autoFlushBytes == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_WS_AUTO_FLUSH_BYTES : autoFlushBytes;
-            long actualAutoFlushIntervalNanos = autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY
-                    ? DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS
-                    : TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis);
+            long actualAutoFlushIntervalNanos;
+            if (autoFlushIntervalMillis == PARAMETER_NOT_SET_EXPLICITLY) {
+                actualAutoFlushIntervalNanos = DEFAULT_WS_AUTO_FLUSH_INTERVAL_NANOS;
+            } else if (autoFlushIntervalMillis == Integer.MAX_VALUE) {
+                actualAutoFlushIntervalNanos = 0; // explicit off: QwpWebSocketSender treats 0 as disabled
+            } else {
+                actualAutoFlushIntervalNanos = TimeUnit.MILLISECONDS.toNanos(autoFlushIntervalMillis);
+            }
 
             Supplier<String> wsAuthHeader = buildWebSocketAuthHeader(tokenProvider);
 
@@ -5057,7 +5068,7 @@ private void validateParameters() {
                 if (protocolVersion != PARAMETER_NOT_SET_EXPLICITLY) {
                     throw new LineSenderException("protocol version is not supported for WebSocket protocol");
                 }
-                if (autoFlushIntervalMillis == Integer.MAX_VALUE) {
+                if (autoFlushDisabled) {
                     throw new LineSenderException("disabling auto-flush is not supported for WebSocket protocol");
                 }
                 if ((sfDurability == SfDurability.FLUSH || sfDurability == SfDurability.APPEND)) {

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -241,6 +241,42 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock
         }
     }
 
+    /**
+     * Appends one binary frame to the send batch, back to back with any frame
+     * already appended. Nothing reaches the wire until {@link #batchFlush()}.
+     * Grows the send buffer if the frame does not fit; a caller bounding the
+     * batch checks {@link #batchByteCap()} first. The batch shares the send
+     * buffer with {@link #sendBinary}, so do not interleave the two.
+     */
+    public void batchAppendBinary(long dataPtr, int length) {
+        checkConnected();
+        sendBuffer.appendMaskedBinaryFrame(dataPtr, length);
+    }
+
+    /**
+     * Byte budget for one send batch: the default send buffer size, fixed for
+     * the connection. It does not follow the buffer's capacity, which only
+     * grows, so one large frame cannot widen every later batch.
+     */
+    public int batchByteCap() {
+        return DEFAULT_SEND_BUFFER_SIZE;
+    }
+
+    /**
+     * Writes every frame appended since the last flush with one send. No-op
+     * when nothing was appended. The batch is empty afterwards, even on failure.
+     */
+    public void batchFlush() {
+        int len = sendBuffer.getWritePos();
+        if (len > 0) {
+            try {
+                doSend(sendBuffer.getBufferPtr(), len, defaultTimeout);
+            } finally {
+                sendBuffer.reset();
+            }
+        }
+    }
+
     @Override
     public void close() {
         // CAS gate: exactly one closer runs the teardown below. Closers can be
@@ -483,6 +519,7 @@ public boolean receiveFrame(WebSocketFrameHandler handler, int timeout) {
             }
 
             // Ensure buffer has space
+            compactRecvBuffer();
             if (recvPos >= recvBufSize - 1024) {
                 growRecvBuffer();
             }
@@ -668,6 +705,7 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
         }
 
         // Try one non-blocking recv
+        compactRecvBuffer();
         if (recvPos >= recvBufSize - 1024) {
             growRecvBuffer();
         }
@@ -1145,18 +1183,23 @@ private void doConnect(CharSequence host, int port) {
     private void doSend(long ptr, int len, int timeout) {
         long startTime = System.nanoTime();
         recvDrained = false;
+        // Send first, poll only on backpressure (mirrors recvOrDie): the
+        // socket is non-blocking, and a pre-send ioWait would cost an
+        // epoll_ctl + epoll_wait per frame even when it is writable.
         while (len > 0) {
-            int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime);
-            ioWait(remainingTimeout, IOOperation.WRITE);
             int sent = dieIfNegative(socket.send(ptr, len));
             while (socket.wantsTlsWrite()) {
-                remainingTimeout = getRemainingTimeOrThrow(timeout, startTime);
+                int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime);
                 ioWait(remainingTimeout, IOOperation.WRITE);
                 dieIfNegative(socket.tlsIO(Socket.WRITE_FLAG));
             }
             if (sent > 0) {
                 ptr += sent;
                 len -= sent;
+            } else {
+                // send returned ERETRY (0): socket full, park until writable
+                int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime);
+                ioWait(remainingTimeout, IOOperation.WRITE);
             }
         }
     }
@@ -1414,8 +1457,13 @@ private int tryParseFrame(WebSocketFrameHandler handler) {
             // Advance read position
             recvReadPos += consumed;
 
-            // Compact buffer if needed
-            compactRecvBuffer();
+            // Rewind once drained; a partial tail stays put until the next
+            // recv compacts it. Compacting here, per frame, memmoved the rest
+            // of a burst once per frame: quadratic in the burst.
+            if (recvReadPos == recvPos) {
+                recvReadPos = 0;
+                recvPos = 0;
+            }
 
             return PARSE_OK;
         }

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketSendBuffer.java

@@ -110,6 +110,32 @@ public WebSocketSendBuffer(int initialCapacity, int maxBufferSize) {
         this.payloadStartOffset = 0;
     }
 
+    /**
+     * Appends one complete masked binary frame at the write position. Unlike
+     * {@link #beginFrame()}/{@link #endFrame(int)}, which reserve
+     * {@code MAX_HEADER_SIZE} and leave a gap before a shorter header, the
+     * header here is exact-size, so consecutive frames are contiguous and the
+     * whole buffer can go out in one write.
+     *
+     * @param payloadPtr pointer to the payload
+     * @param payloadLen payload length
+     */
+    public void appendMaskedBinaryFrame(long payloadPtr, int payloadLen) {
+        int headerSize = WebSocketFrameWriter.headerSize(payloadLen, true);
+        long requiredCapacity = (long) writePos + headerSize + payloadLen;
+        if (requiredCapacity > bufCapacity) {
+            grow(requiredCapacity);
+        }
+        int maskKey = rnd.nextInt();
+        long payload = bufPtr + writePos + WebSocketFrameWriter.writeHeader(
+                bufPtr + writePos, true, WebSocketOpcode.BINARY, payloadLen, maskKey);
+        if (payloadLen > 0) {
+            Vect.memcpy(payload, payloadPtr, payloadLen);
+            WebSocketFrameWriter.maskPayload(payload, payloadLen, maskKey);
+        }
+        writePos += headerSize + payloadLen;
+    }
+
     /**
      * Begins a new WebSocket frame. Reserves space for the maximum header size.
      * The opcode is specified later when ending the frame via {@link #endFrame(int)}.

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpEgressIoThread.java

@@ -638,11 +638,11 @@ private void handleResultBatch(long payloadPtr, int payloadLen) {
         // releaseBuffer offers the token that unblocks this take.
         //
         // Wait uninterruptibly: if close() interrupts us here we MUST NOT
-        // unwind back to tryParseFrame, because tryParseFrame's tail
-        // advances recvReadPos and runs compactRecvBuffer (Vect.memmove)
-        // over the bytes the user's column pointers still alias inside
-        // recv-buf. A premature unwind on interrupt is the close()-during-
-        // onBatch UAF: the user thread's onBatch reads garbage mid-call.
+        // unwind back to tryParseFrame, because the next receive's
+        // compactRecvBuffer (Vect.memmove) and recv write over the bytes the
+        // user's column pointers still alias inside recv-buf. A premature
+        // unwind on interrupt is the close()-during-onBatch UAF: the user
+        // thread's onBatch reads garbage mid-call.
         // Stay parked; close()'s 5-second join timeout will abandon the
         // I/O thread (the documented timeout-leak path), but only AFTER
         // the user's onBatch has finished -- no UAF.

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -86,6 +86,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.LongSupplier;
 import java.util.function.Supplier;
 
 /**
@@ -374,6 +375,9 @@ public class QwpWebSocketSender implements Sender {
     private SenderErrorHandler errorHandler = DefaultSenderErrorHandler.INSTANCE;
     private int errorInboxCapacity = SenderErrorDispatcher.DEFAULT_CAPACITY;
     private long firstPendingRowTimeNanos;
+    // Clock for the auto-flush age check only (sendRow / shouldAutoFlush).
+    // Test seam: setNanoClockForTesting swaps it to count or steer the reads.
+    private LongSupplier nanoClock = System::nanoTime;
     // Additive offset applied to every user-visible FSN this sender reports
     // (flushAndGetSequence, awaitAckedFsn's target, getAckedFsn, drain's
     // watermark, and every FSN the I/O loop surfaces through the progress
@@ -2453,7 +2457,7 @@ public int getAutoFlushBytes() {
     }
 
     /**
-     * Returns the auto-flush interval in nanoseconds.
+     * Returns the auto-flush interval in nanoseconds (0 = interval-based auto-flush disabled).
      */
     public long getAutoFlushIntervalNanos() {
         return autoFlushIntervalNanos;
@@ -2527,6 +2531,11 @@ public void setChunkPublishFaultForTesting(Runnable fault) {
         this.chunkPublishFaultForTesting = fault;
     }
 
+    @TestOnly
+    public void setNanoClockForTesting(LongSupplier nanoClock) {
+        this.nanoClock = nanoClock;
+    }
+
     /**
      * Background orphan-drainer pool, or {@code null} when
      * {@code drain_orphans} is off or no orphan slot was adopted.
@@ -2908,6 +2917,8 @@ public long getTotalErrorNotificationsDelivered() {
      * Count of frames re-sent during post-reconnect catch-up windows since
      * this sender started. Zero in steady state; a sustained nonzero rate
      * signals flapping where every reconnect replays meaningful work.
+     * A frame counts as replayed only if an earlier send completed successfully.
+     * Failed batch flushes do not count, even if some bytes were written.
      * Sender-lifetime: carried across symbol-dictionary recycles and retained
      * after {@link #close()}.
      */
@@ -2916,8 +2927,10 @@ public long getTotalFramesReplayed() {
     }
 
     /**
-     * Binary frames the cursor I/O loop has issued to the wire since this
-     * sender started, replays included. Sender-lifetime: carried across
+     * Binary frames the cursor I/O loop has successfully sent since this
+     * sender started, replays included. A batch contributes only when its
+     * entire flush succeeds; failed flushes contribute no frames, even if
+     * some bytes were written. Sender-lifetime: carried across
      * symbol-dictionary recycles and retained after {@link #close()}.
      */
     public long getTotalFramesSent() {
@@ -7132,8 +7145,9 @@ private void sendRow() {
         long bufferedNow = currentTableBuffer.nextRow(
                 currentTableBufferSnapshotBytes, cap > 0 ? cap : Long.MAX_VALUE);
 
-        if (pendingRowCount == 0) {
-            firstPendingRowTimeNanos = System.nanoTime();
+        // Interval off (0): no clock read on the row path at all.
+        if (pendingRowCount == 0 && autoFlushIntervalNanos > 0) {
+            firstPendingRowTimeNanos = nanoClock.getAsLong();
         }
         pendingRowCount++;
 
@@ -7164,8 +7178,11 @@ private boolean shouldAutoFlush() {
         if (effectiveAutoFlushBytes > 0 && getPendingBytes() >= effectiveAutoFlushBytes) {
             return true;
         }
-        if (autoFlushIntervalNanos > 0) {
-            long ageNanos = System.nanoTime() - firstPendingRowTimeNanos;
+        // pendingRowCount == 1: sendRow() just timestamped this very row, so its
+        // age is ~0 and can never reach the (>= 1 ms) interval. Skipping the
+        // clock read there saves one nanoTime() per flush for per-row flushers.
+        if (autoFlushIntervalNanos > 0 && pendingRowCount > 1) {
+            long ageNanos = nanoClock.getAsLong() - firstPendingRowTimeNanos;
             return ageNanos >= autoFlushIntervalNanos;
         }
         return false;