wip: less copying

Author: jerrinot

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

@@ -31,6 +31,7 @@
 #include <unistd.h>
 #include <sys/errno.h>
 #include <stdlib.h>
+#include <stdint.h>
 #include <string.h>
 #include "net.h"
 #include <netdb.h>
@@ -356,4 +357,36 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_sendTo
         (JNIEnv *e, jclass cl, jint fd, jlong ptr, jint len, jlong sockaddr) {
     return (jint) sendto((int) fd, (const void *) ptr, (size_t) len, 0, (const struct sockaddr *) sockaddr,
                          sizeof(struct sockaddr_in));
-}
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_sendToScatter
+        (JNIEnv *e, jclass cl, jint fd, jlong segmentsPtr, jint segmentCount, jlong sockaddr) {
+    if (segmentCount <= 0) {
+        return 0;
+    }
+
+    struct iovec *iov = calloc((size_t) segmentCount, sizeof(struct iovec));
+    if (iov == NULL) {
+        errno = ENOMEM;
+        return -1;
+    }
+
+    const char *segment = (const char *) segmentsPtr;
+    for (int i = 0; i < segmentCount; i++) {
+        iov[i].iov_base = (void *) (uintptr_t) (*(const jlong *) segment);
+        iov[i].iov_len = (size_t) (*(const jlong *) (segment + 8));
+        segment += 16;
+    }
+
+    struct msghdr msg;
+    memset(&msg, 0, sizeof(msg));
+    msg.msg_name = (void *) sockaddr;
+    msg.msg_namelen = sizeof(struct sockaddr_in);
+    msg.msg_iov = iov;
+    msg.msg_iovlen = (size_t) segmentCount;
+
+    ssize_t sent;
+    RESTARTABLE(sendmsg((int) fd, &msg, 0), sent);
+    free(iov);
+    return sent < 0 ? -1 : (jint) sent;
+}

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

@@ -25,6 +25,7 @@
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #include <mstcpip.h>
+#include <stdint.h>
 #include "../share/net.h"
 #include "errno.h"
 
@@ -287,4 +288,45 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_sendTo
         SaveLastError();
     }
     return result;
-}
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_sendToScatter
+        (JNIEnv *e, jclass cl, jint fd, jlong segmentsPtr, jint segmentCount, jlong sockaddr) {
+    if (segmentCount <= 0) {
+        return 0;
+    }
+
+    WSABUF *buffers = calloc((size_t) segmentCount, sizeof(WSABUF));
+    if (buffers == NULL) {
+        WSASetLastError(WSA_NOT_ENOUGH_MEMORY);
+        SaveLastError();
+        return -1;
+    }
+
+    const char *segment = (const char *) segmentsPtr;
+    for (int i = 0; i < segmentCount; i++) {
+        buffers[i].buf = (CHAR *) (uintptr_t) (*(const jlong *) segment);
+        buffers[i].len = (ULONG) (*(const jlong *) (segment + 8));
+        segment += 16;
+    }
+
+    DWORD bytesSent = 0;
+    int result = WSASendTo(
+            (SOCKET) fd,
+            buffers,
+            (DWORD) segmentCount,
+            &bytesSent,
+            0,
+            (const struct sockaddr *) sockaddr,
+            sizeof(struct sockaddr_in),
+            NULL,
+            NULL
+    );
+    free(buffers);
+
+    if (result == SOCKET_ERROR) {
+        SaveLastError();
+        return -1;
+    }
+    return (jint) bytesSent;
+}

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

@@ -85,4 +85,10 @@ public void send(long ptr, int len) {
             throw new LineSenderException("send error").errno(nf.errno());
         }
     }
+
+    public void sendSegments(long segmentsPtr, int segmentCount, int totalLen) {
+        if (nf.sendToRawScatter(fd, segmentsPtr, segmentCount, sockaddr) != totalLen) {
+            throw new LineSenderException("send error").errno(nf.errno());
+        }
+    }
 }

core/src/main/java/io/questdb/client/cutlass/line/udp/UdpLineChannel.java

@@ -0,0 +1,119 @@
+/*
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package io.questdb.client.cutlass.qwp.client;
+
+import io.questdb.client.std.MemoryTag;
+import io.questdb.client.std.QuietCloseable;
+import io.questdb.client.std.Unsafe;
+
+final class NativeSegmentList implements QuietCloseable {
+    static final int SEGMENT_SIZE = 16;
+
+    private int capacity;
+    private long ptr;
+    private int size;
+    private long totalLength;
+
+    NativeSegmentList() {
+        this(16);
+    }
+
+    NativeSegmentList(int initialCapacity) {
+        this.capacity = Math.max(initialCapacity, 4);
+        this.ptr = Unsafe.malloc((long) capacity * SEGMENT_SIZE, MemoryTag.NATIVE_DEFAULT);
+    }
+
+    void add(long address, long length) {
+        if (length <= 0) {
+            return;
+        }
+        ensureCapacity(size + 1);
+        long segmentPtr = ptr + (long) size * SEGMENT_SIZE;
+        Unsafe.getUnsafe().putLong(segmentPtr, address);
+        Unsafe.getUnsafe().putLong(segmentPtr + 8, length);
+        size++;
+        totalLength += length;
+    }
+
+    void appendFrom(NativeSegmentList other) {
+        if (other.size == 0) {
+            return;
+        }
+        ensureCapacity(size + other.size);
+        Unsafe.getUnsafe().copyMemory(
+                other.ptr,
+                ptr + (long) size * SEGMENT_SIZE,
+                (long) other.size * SEGMENT_SIZE
+        );
+        size += other.size;
+        totalLength += other.totalLength;
+    }
+
+    long getAddress() {
+        return ptr;
+    }
+
+    int getSegmentCount() {
+        return size;
+    }
+
+    long getTotalLength() {
+        return totalLength;
+    }
+
+    @Override
+    public void close() {
+        if (ptr != 0) {
+            Unsafe.free(ptr, (long) capacity * SEGMENT_SIZE, MemoryTag.NATIVE_DEFAULT);
+            ptr = 0;
+            capacity = 0;
+            size = 0;
+            totalLength = 0;
+        }
+    }
+
+    void reset() {
+        size = 0;
+        totalLength = 0;
+    }
+
+    private void ensureCapacity(int required) {
+        if (required <= capacity) {
+            return;
+        }
+
+        int newCapacity = capacity;
+        while (newCapacity < required) {
+            newCapacity *= 2;
+        }
+
+        ptr = Unsafe.realloc(
+                ptr,
+                (long) capacity * SEGMENT_SIZE,
+                (long) newCapacity * SEGMENT_SIZE,
+                MemoryTag.NATIVE_DEFAULT
+        );
+        capacity = newCapacity;
+    }
+}