perf(qwp): event-driven I/O loop, zero CPU when idle

core/src/main/c/share/net.c

@@ -34,6 +34,11 @@
 #include <stdint.h>
 #include <string.h>
 #include <poll.h>
+#ifdef __linux__
+#include <sys/eventfd.h>
+#else
+#include <sys/event.h>
+#endif
 #include <time.h>
 #include "glibc_compat.h"
 #include "net.h"
@@ -439,6 +444,107 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_socketUdp
     return fd;
 }
 
+// Doorbell + idle wait for the QWP I/O thread.
+//   Linux:         eventfd; the wait is poll(2) on socket + eventfd.
+//   macOS/FreeBSD: a kqueue holding an EVFILT_USER; the kqueue is both the
+//                  doorbell and the wait, the socket is (re)armed per wait.
+// A ring is one syscall on the producer and touches no network stack. EINTR
+// in a wait returns 0: the caller re-checks its inputs and waits again.
+#ifdef __linux__
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_doorbellCreate
+        (JNIEnv *e, jclass cl) {
+    return eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_doorbellRing
+        (JNIEnv *e, jclass cl, jint fd) {
+    uint64_t one = 1;
+    ssize_t n;
+    RESTARTABLE(write((int) fd, &one, sizeof(one)), n);
+    // EAGAIN = counter saturated = already ringing.
+    return (n == (ssize_t) sizeof(one) || (n < 0 && errno == EAGAIN)) ? 0 : -1;
+}
+
+JNIEXPORT void JNICALL Java_io_questdb_client_network_Net_doorbellClose
+        (JNIEnv *e, jclass cl, jint fd) {
+    close((int) fd);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_awaitReadableOrRing
+        (JNIEnv *e, jclass cl, jint socketFd, jint doorbell, jint timeoutMillis) {
+    struct pollfd fds[2];
+    fds[0].fd = (int) socketFd;
+    fds[0].events = POLLIN;
+    fds[0].revents = 0;
+    fds[1].fd = (int) doorbell;
+    fds[1].events = POLLIN;
+    fds[1].revents = 0;
+    int n = poll(fds, 2, (int) timeoutMillis);
+    if (n > 0 && (fds[1].revents & POLLIN)) {
+        // Consume the ring here: a wait that ended on the socket alone skips
+        // the (would-be EAGAIN) read entirely.
+        uint64_t v;
+        ssize_t r;
+        RESTARTABLE(read((int) doorbell, &v, sizeof(v)), r);
+    }
+    return (n < 0 && errno == EINTR) ? 0 : n;
+}
+
+#else // kqueue platforms
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_doorbellCreate
+        (JNIEnv *e, jclass cl) {
+    int kq = kqueue();
+    if (kq < 0) {
+        return -1;
+    }
+    struct kevent ev;
+    // EV_CLEAR: a delivered ring resets itself, an undelivered one stays pending.
+    EV_SET(&ev, 0, EVFILT_USER, EV_ADD | EV_CLEAR, 0, 0, NULL);
+    if (kevent(kq, &ev, 1, NULL, 0, NULL) < 0) {
+        int saved = errno;
+        close(kq);
+        errno = saved;
+        return -1;
+    }
+    return kq;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_doorbellRing
+        (JNIEnv *e, jclass cl, jint kq) {
+    struct kevent ev;
+    EV_SET(&ev, 0, EVFILT_USER, 0, NOTE_TRIGGER, 0, NULL);
+    int n;
+    RESTARTABLE(kevent((int) kq, &ev, 1, NULL, 0, NULL), n);
+    return n < 0 ? -1 : 0;
+}
+
+JNIEXPORT void JNICALL Java_io_questdb_client_network_Net_doorbellClose
+        (JNIEnv *e, jclass cl, jint kq) {
+    close((int) kq);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_awaitReadableOrRing
+        (JNIEnv *e, jclass cl, jint socketFd, jint kq, jint timeoutMillis) {
+    // Re-armed on every wait; a closed (reconnected) socket drops out of the
+    // kqueue by itself, so there is nothing to unregister.
+    struct kevent change;
+    EV_SET(&change, (uintptr_t) socketFd, EVFILT_READ, EV_ADD | EV_ONESHOT, 0, 0, NULL);
+    struct timespec ts;
+    struct timespec *pts = NULL;
+    if (timeoutMillis >= 0) {
+        ts.tv_sec = timeoutMillis / 1000;
+        ts.tv_nsec = (long) (timeoutMillis % 1000) * 1000000L;
+        pts = &ts;
+    }
+    struct kevent out[2];
+    int n = kevent((int) kq, &change, 1, out, 2, pts);
+    return (n < 0 && errno == EINTR) ? 0 : n;
+}
+
+#endif
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_setMulticastInterface
         (JNIEnv *e, jclass cl, jint fd, jint ipv4address) {
     struct in_addr address;

core/src/main/c/windows/net.c

@@ -325,6 +325,66 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_setTcpNoDelay
     return set_int_sockopt((SOCKET) fd, IPPROTO_TCP, TCP_NODELAY, noDelay);
 }
 
+// The doorbell is a manual-reset kernel Event (kernel handles fit in 32 bits
+// on 64-bit Windows). The wait is WSAEventSelect + WaitForMultipleObjects,
+// because WSAPoll cannot wait on anything that is not a socket.
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_doorbellCreate
+        (JNIEnv *e, jclass cl) {
+    HANDLE h = CreateEvent(NULL, TRUE, FALSE, NULL);
+    if (h == NULL) {
+        SaveLastError();
+        return -1;
+    }
+    return (jint) (uintptr_t) h;
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_doorbellRing
+        (JNIEnv *e, jclass cl, jint doorbell) {
+    if (!SetEvent((HANDLE) (uintptr_t) (uint32_t) doorbell)) {
+        SaveLastError();
+        return -1;
+    }
+    return 0;
+}
+
+JNIEXPORT void JNICALL Java_io_questdb_client_network_Net_doorbellClose
+        (JNIEnv *e, jclass cl, jint doorbell) {
+    CloseHandle((HANDLE) (uintptr_t) (uint32_t) doorbell);
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_awaitReadableOrRing
+        (JNIEnv *e, jclass cl, jint socketFd, jint doorbell, jint timeoutMillis) {
+    WSAEVENT sockEvent = WSACreateEvent();
+    if (sockEvent == WSA_INVALID_EVENT) {
+        SaveLastError();
+        return -1;
+    }
+    // FD_READ is recorded immediately when data is already pending, so this is
+    // a level-triggered readability wait.
+    if (WSAEventSelect((SOCKET) socketFd, sockEvent, FD_READ | FD_CLOSE) == SOCKET_ERROR) {
+        SaveLastError();
+        WSACloseEvent(sockEvent);
+        return -1;
+    }
+    HANDLE handles[2] = {sockEvent, (HANDLE) (uintptr_t) (uint32_t) doorbell};
+    DWORD r = WaitForMultipleObjects(2, handles, FALSE, timeoutMillis < 0 ? INFINITE : (DWORD) timeoutMillis);
+    // Detach the event; the socket stays non-blocking, which it already was.
+    WSAEventSelect((SOCKET) socketFd, NULL, 0);
+    WSACloseEvent(sockEvent);
+    if (r == WAIT_TIMEOUT) {
+        return 0;
+    }
+    if (r == WAIT_OBJECT_0 + 1) {
+        // Consume the ring (manual-reset event).
+        ResetEvent((HANDLE) (uintptr_t) (uint32_t) doorbell);
+    }
+    if (r == WAIT_FAILED) {
+        SaveLastError();
+        return -1;
+    }
+    return 1;
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_socketUdp
         (JNIEnv *e, jclass cl) {
     SOCKET s = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);

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

@@ -1047,7 +1047,7 @@ final class LineSenderBuilder {
         private static final int DEFAULT_MAXIMUM_BUFFER_CAPACITY = 100 * 1024 * 1024;
         private static final int DEFAULT_MAX_BACKGROUND_DRAINERS = 4;
         private static final int DEFAULT_MAX_BACKOFF_MILLIS = 1_000;
-        // Default ceiling on cursor-allocated bytes (active + spare + sealed).
+        // Default ceiling on cursor-allocated bytes (active + sealed segments).
         // RAM is precious; if you're not persisting to disk, you don't get
         // to balloon. Memory mode = 128 MiB (32 segments at default size).
         private static final long DEFAULT_MAX_BYTES_MEMORY = 128L * 1024 * 1024;
@@ -2962,8 +2962,8 @@ public LineSenderBuilder storeAndForwardMaxSegmentBytes(long maxSegmentBytes) {
         }
 
         /**
-         * Hard cap on cursor-allocated bytes: active + spare + sealed
-         * segments, plus the symbol-dictionary side-file with {@code sf_dir}.
+         * Hard cap on cursor-allocated bytes: active + sealed segments, plus
+         * the symbol-dictionary side-file with {@code sf_dir}.
          * When the cap is reached, the producer's {@code Sender.flush()}
          * blocks until ACK-driven trim frees space; if the cap is exhausted
          * past the configured deadline (default 30 s), {@code flush()} throws.
@@ -2992,7 +2992,8 @@ public LineSenderBuilder storeAndForwardMaxTotalBytes(long maxTotalBytes) {
          * every other sender built on it, instead of to a private budget of
          * {@code sf_max_total_bytes}. Together those senders never hold more
          * than the budget's capacity, except that each is always granted its
-         * minimum working set: the active segment plus one spare. This
+         * minimum working set: the active segment plus the one it rotates
+         * into. This
          * sender's own {@code sf_max_total_bytes} then caps only the
          * background orphan drainers it starts. The connection pool behind
          * {@link QuestDB} builds every sender this way. WebSocket transport

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

@@ -30,6 +30,7 @@
 import io.questdb.client.cutlass.qwp.websocket.WebSocketFrameParser;
 import io.questdb.client.cutlass.qwp.websocket.WebSocketOpcode;
 import io.questdb.client.network.IOOperation;
+import io.questdb.client.network.Net;
 import io.questdb.client.network.NetworkFacade;
 import io.questdb.client.network.Socket;
 import io.questdb.client.network.SocketFactory;
@@ -105,6 +106,14 @@ public abstract class WebSocketClient implements QuietCloseable {
     // (not a bare volatile check-then-act) so concurrent closers cannot both
     // enter close() and double-run disconnect()/Unsafe.free.
     private final AtomicBoolean closed = new AtomicBoolean();
+    // See awaitReadableOrWake/wake. wakeLock serialises wake() with close().
+    private final Object wakeLock = new Object();
+    private int doorbell = -1;
+    // True after a recv() found nothing. Until the next send or
+    // awaitReadableOrWake, tryReceiveFrame answers false without a syscall:
+    // the I/O loop's pre-park receive pass otherwise repeats the EAGAIN it got
+    // on the previous iteration. I/O-thread-only, like the recv buffer.
+    private boolean recvDrained;
     // Upper bound (ms) on the TCP connect. <= 0 disables the application-level
     // timeout and falls back to the OS connect timeout. Seeded from the
     // configuration; the QWP sender may override it via setConnectTimeout().
@@ -207,7 +216,13 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock
             this.rnd = new SecureRnd();
             this.upgraded = false;
             this.closed.set(false);
+
+            this.doorbell = Net.doorbellCreate();
+            if (doorbell < 0) {
+                throw new HttpClientException("could not create doorbell [errno=").put(nf.errno()).put(']');
+            }
         } catch (Throwable t) {
+            closeDoorbell();
             if (recvBufPtr != 0) {
                 Unsafe.free(recvBufPtr, recvBufSize, MemoryTag.NATIVE_DEFAULT);
                 recvBufPtr = 0;
@@ -250,6 +265,7 @@ public void close() {
                 Unsafe.free(recvBufPtr, recvBufSize, MemoryTag.NATIVE_DEFAULT);
                 recvBufPtr = 0;
             }
+            closeDoorbell();
         }
     }
 
@@ -624,6 +640,13 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
             return result == PARSE_OK;
         }
 
+        if (recvDrained) {
+            // The last recv found nothing and nothing has been sent or waited
+            // for since: a repeat recv would be another EAGAIN. Bytes that land
+            // meanwhile end the next awaitReadableOrWake at once (level-triggered).
+            return false;
+        }
+
         // Try one non-blocking recv
         if (recvPos >= recvBufSize - 1024) {
             growRecvBuffer();
@@ -634,6 +657,7 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
             throw new HttpClientException("peer disconnect [errno=").errno(nf.errno()).put(']');
         }
         if (n == 0) {
+            recvDrained = true;
             return false; // No data available
         }
         recvPos += n;
@@ -1099,6 +1123,7 @@ private void doConnect(CharSequence host, int port) {
 
     private void doSend(long ptr, int len, int timeout) {
         long startTime = System.nanoTime();
+        recvDrained = false;
         while (len > 0) {
             int remainingTimeout = getRemainingTimeOrThrow(timeout, startTime);
             ioWait(remainingTimeout, IOOperation.WRITE);
@@ -1472,4 +1497,49 @@ protected void dieWaiting(int n) {
      * Sets up platform-specific I/O wait mechanism after connection.
      */
     protected abstract void setupIoWait();
+
+    /**
+     * Blocks until the socket becomes readable, another thread calls
+     * {@link #wake()}, or {@code timeoutMillis} elapses ({@code -1} = no
+     * timeout). Returns {@code true} unless it timed out; spurious returns are
+     * possible and callers re-check their inputs. The doorbell is stateful: a
+     * wake that lands before the wait begins ends it at once. Independent of
+     * the readiness queue {@link #ioWait} uses, so a pending wake never leaks
+     * into a handshake or send wait.
+     */
+    public boolean awaitReadableOrWake(int timeoutMillis) {
+        recvDrained = false;
+        if (socket.hasBufferedInput()) {
+            // TLS: decrypted or undecrypted input already sits in user space,
+            // invisible to poll(2). Report readable without blocking.
+            return true;
+        }
+        int n = Net.awaitReadableOrRing(socket.getFd(), doorbell, timeoutMillis);
+        if (n < 0) {
+            throw new HttpClientException("internal error: poll failure [errno=").put(nf.errno()).put(']');
+        }
+        return n > 0;
+    }
+
+    /**
+     * Wakes a thread blocked in {@link #awaitReadableOrWake}; with nobody
+     * waiting the wake is remembered until the next wait. Safe from any thread,
+     * including concurrently with {@link #close()}.
+     */
+    public void wake() {
+        synchronized (wakeLock) {
+            if (doorbell >= 0) {
+                Net.doorbellRing(doorbell);
+            }
+        }
+    }
+
+    private void closeDoorbell() {
+        synchronized (wakeLock) {
+            if (doorbell >= 0) {
+                Net.doorbellClose(doorbell);
+                doorbell = -1;
+            }
+        }
+    }
 }