Native: later result batches carry vectors in the float32 tail (arcadedb_n_next_bin)

native/src/main/java/org/humemai/nativelib/ArcadeNative.java

@@ -33,11 +33,14 @@
 import org.graalvm.word.WordFactory;
 import org.graalvm.nativeimage.c.type.CTypeConversion;
 
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
 
 public final class ArcadeNative {
   static final Map<Long, Database> DBS = new ConcurrentHashMap<>();
@@ -291,6 +294,27 @@ public static CCharPointer next(final IsolateThread thread, final long rsId, fin
     }
   }
 
+  /**
+   * arcadedb_n_next with the float32 tail of arcadedb_n_exec flag 4: float[] values of 8 or more elements of the batch come back in a binary tail
+   * ({"\u0001":["B",offset,count]} in the rows, ,"bin":total at the end of the JSON, the floats after its NUL), not as number text. Offsets are
+   * per reply. A reply without such a value is the same text arcadedb_n_next returns.
+   */
+  @CEntryPoint(name = "arcadedb_n_next_bin")
+  public static CCharPointer nextBin(final IsolateThread thread, final long rsId, final int max) {
+    ResultSet rs = null;
+    try {
+      rs = RESULT_SETS.get(rsId);
+      if (rs == null)
+        throw new IllegalArgumentException("unknown result set " + rsId);
+      final ResultSet cursor = rs;
+      return withFloatTail(() -> batchOf(rsId, cursor, max));
+    } catch (final Throwable t) {
+      RESULT_SETS.remove(rsId);
+      closeQuietly(rs);
+      return err(t);
+    }
+  }
+
   @CEntryPoint(name = "arcadedb_n_rs_close")
   public static CCharPointer rsClose(final IsolateThread thread, final long rsId) {
     closeQuietly(RESULT_SETS.remove(rsId));
@@ -488,11 +512,19 @@ private static CCharPointer replyOf(final ResultSet rs, final int max, final int
     }
     if ((flags & 4) == 0)
       return out(batch(rs, max));
+    return withFloatTail(() -> batch(rs, max));
+  }
+
+  /**
+   * Runs encode with the float32 sink set (float[] values of 8 or more elements go to the tail of the reply as {"\u0001":["B",off,n]}) and returns
+   * the reply: out(text) when no vector was met, else the text with ,"bin":total, a NUL, padding to 4 bytes and the floats.
+   */
+  private static CCharPointer withFloatTail(final Supplier<String> encode) {
     final NativeCodec.FloatSink sink = new NativeCodec.FloatSink();
     NativeCodec.SINK.set(sink);
     final String reply;
     try {
-      reply = batch(rs, max);
+      reply = encode.get();
     } finally {
       NativeCodec.SINK.remove();
     }
@@ -501,13 +533,12 @@ private static CCharPointer replyOf(final ResultSet rs, final int max, final int
     final byte[] b = (reply.substring(0, reply.length() - 1) + ",\"bin\":" + sink.n + "}").getBytes(StandardCharsets.UTF_8);
     final int off = (b.length + 1 + 3) & ~3;
     final CCharPointer p = UnmanagedMemory.malloc(off + 4 * sink.n);
-    for (int i = 0; i < b.length; i++)
-      p.write(i, b[i]);
-    for (int i = b.length; i < off; i++)
-      p.write(i, (byte) 0);
-    final CFloatPointer fp = (CFloatPointer) ((org.graalvm.word.Pointer) p).add(off);
-    for (int i = 0; i < sink.n; i++)
-      fp.write(i, sink.a[i]);
+    // bulk copies (Unsafe.copyMemory), not one p.write per byte: the text, the zero padding up to the 4-byte boundary, then the float32 tail in native order
+    final ByteBuffer bb = CTypeConversion.asByteBuffer(p, (int) (off + 4L * sink.n));
+    bb.put(b, 0, b.length);
+    while (bb.position() < off)
+      bb.put((byte) 0);
+    bb.order(ByteOrder.nativeOrder()).asFloatBuffer().put(sink.a, 0, sink.n);
     return p;
   }
 
@@ -704,9 +735,8 @@ static Chars chars(final CCharPointer p) {
   static CCharPointer out(final String s) {
     final byte[] b = s.getBytes(StandardCharsets.UTF_8);
     final CCharPointer p = UnmanagedMemory.malloc(b.length + 1);
-    for (int i = 0; i < b.length; i++)
-      p.write(i, b[i]);
-    p.write(b.length, (byte) 0);
+    // one bulk copy (Unsafe.copyMemory) and the terminating zero; this was a Java loop of p.write(i, b[i]), one byte per iteration
+    CTypeConversion.asByteBuffer(p, b.length + 1).put(b, 0, b.length).put((byte) 0);
     return p;
   }
 }

src/arcadedb/_native/database.py

@@ -214,7 +214,12 @@ def _fill(self) -> bool:
         if not self._more:
             return False
         try:
-            reply = loads(lib.call("arcadedb_n_next", self._rs, _NEXT_BATCH))
+            if lib.NEXT_BIN and _BINARY_REPLY:
+                reply = loads(
+                    *lib.call_bin("arcadedb_n_next_bin", self._rs, _NEXT_BATCH)
+                )
+            else:
+                reply = loads(lib.call("arcadedb_n_next", self._rs, _NEXT_BATCH))
         except ArcadeDBError:
             raise
         if not reply.get("ok"):

src/arcadedb/_native/lib.py

@@ -29,6 +29,8 @@
 INSERT_ROWS = os.environ.get("ARCADEDB_NATIVE_INSERT_ROWS", "1") != "0"
 # ARCADEDB_NATIVE_EXEC_MANY: the rows of an insert_many chunk that JSON text cannot carry unchanged go in one crossing (arcadedb_n_exec_many), not one call each
 EXEC_MANY = os.environ.get("ARCADEDB_NATIVE_EXEC_MANY", "1") != "0"
+# ARCADEDB_NATIVE_NEXT_BIN: the batches after the first read through arcadedb_n_next_bin, float arrays of 8 or more elements in a float32 tail (as flag 4 of arcadedb_n_exec does for the first batch), not as number text
+NEXT_BIN = os.environ.get("ARCADEDB_NATIVE_NEXT_BIN", "1") != "0"
 
 # the directory the engine log, logging.properties and the crash files go to (None until the library is loaded)
 LOG_DIR_USED = None
@@ -213,6 +215,7 @@ class _Params(ctypes.Structure):
             "arcadedb_n_insert_cols": [P, L, S, S, S, L],
             "arcadedb_n_insert_colp": [P, L, S, S, P, I],
             "arcadedb_n_next": [P, L, I],
+            "arcadedb_n_next_bin": [P, L, I],
             "arcadedb_n_rs_close": [P, L],
             "arcadedb_n_cmd": [P, L, I, S, S, S, S, L, I],
             "arcadedb_n_last_error": [P],
@@ -251,7 +254,7 @@ class _Params(ctypes.Structure):
                 )
                 else ctypes.c_longlong if name == "arcadedb_n_cmd" else ctypes.c_void_p
             )
-        global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY
+        global BINARY_PARAMS, CMD_PATH, CLOSE_BATCH, INSERT_ROWS, EXEC_MANY, NEXT_BIN
         if not hasattr(
             lib, "arcadedb_n_exec_bin"
         ):  # a library built before the binary path
@@ -264,6 +267,10 @@ class _Params(ctypes.Structure):
             INSERT_ROWS = False
         if not hasattr(lib, "arcadedb_n_exec_many"):
             EXEC_MANY = False
+        if not hasattr(
+            lib, "arcadedb_n_next_bin"
+        ):  # a library built before the binary tail of later batches
+            NEXT_BIN = False
         _isolate = isolate
         _creator = threading.get_ident()
         _local.thread = _Thread(thread, owned=False)

tests/test_native_heap.py

@@ -64,25 +64,121 @@ def _diag(tmp_path, isolate_args=None) -> dict:
     return json.loads(result.stdout.strip().splitlines()[-1])
 
 
-def _memory_limit() -> int:
-    """Bytes the isolate sees on Linux: the smaller of the host memory and a cgroup limit."""
-    limits = [os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE")]
-    for name in (
-        "/sys/fs/cgroup/memory.max",
-        "/sys/fs/cgroup/memory/memory.limit_in_bytes",
-    ):
+def _cgroup_memory_limit(proc_cgroup="/proc/self/cgroup", root="/sys/fs/cgroup"):
+    """The cgroup v2 memory limit that applies to this process, or None when there is none.
+
+    The process's own cgroup (the 0::<path> line of /proc/self/cgroup) is read first: <root>/<path>/memory.max. A value of
+    max means no limit there, so the walk goes up one directory at a time to root and returns the first number it finds.
+    Files that are absent or unreadable count as max.
+    """
+    try:
+        lines = Path(proc_cgroup).read_text().splitlines()
+    except OSError:
+        return None
+    path = next(
+        (
+            parts[2]
+            for parts in (line.split(":", 2) for line in lines)
+            if len(parts) == 3 and parts[0] == "0" and parts[1] == ""
+        ),
+        None,
+    )
+    if path is None:  # cgroup v1 only
+        return None
+    top = Path(root)
+    here = top / path.lstrip("/")
+    while True:
         try:
-            limits.append(int(Path(name).read_text()))
-        except (OSError, ValueError):  # absent, or "max" (no limit)
+            return int((here / "memory.max").read_text())
+        except (OSError, ValueError):  # absent, unreadable, or "max" (no limit here)
             pass
+        if here == top or top not in here.parents:
+            return None
+        here = here.parent
+
+
+def _memory_limit() -> int:
+    """Bytes the isolate sees on Linux: the smaller of the host memory and the cgroup limit of this process."""
+    limits = [os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE")]
+    limit = _cgroup_memory_limit()
+    if limit is not None:
+        limits.append(limit)
+    try:  # cgroup v1
+        limits.append(
+            int(Path("/sys/fs/cgroup/memory/memory.limit_in_bytes").read_text())
+        )
+    except (OSError, ValueError):
+        pass
     return min(limits)
 
 
+def _fake_cgroup(tmp_path, own, limits):
+    """(proc_cgroup, root) of a fake tree: this process is in cgroup own; limits maps a cgroup path to its memory.max text."""
+    root = tmp_path / "cgroup"
+    for name, text in limits.items():
+        d = root / name.strip("/")
+        d.mkdir(parents=True, exist_ok=True)
+        (d / "memory.max").write_text(text + "\n")
+    proc = tmp_path / "proc_cgroup"
+    proc.write_text(f"0::{own}\n")
+    return proc, root
+
+
+def test_the_cgroup_limit_of_the_process_own_cgroup_wins(tmp_path):
+    proc, root = _fake_cgroup(
+        tmp_path,
+        "/a/b/c",
+        {"/": "max", "/a": "2097152", "/a/b": "1048576", "/a/b/c": "4096"},
+    )
+    assert _cgroup_memory_limit(proc, root) == 4096
+
+
+def test_the_cgroup_limit_is_taken_from_the_first_parent_that_has_one(tmp_path):
+    proc, root = _fake_cgroup(
+        tmp_path,
+        "/a/b/c",
+        {"/": "max", "/a": "2097152", "/a/b": "1048576", "/a/b/c": "max"},
+    )
+    assert _cgroup_memory_limit(proc, root) == 1048576
+
+
+def test_a_directory_without_memory_max_counts_as_unlimited(tmp_path):
+    proc, root = _fake_cgroup(tmp_path, "/a/b/c", {"/a": "8192"})
+    (root / "a" / "b" / "c").mkdir(parents=True)  # no memory.max file in c or b
+    assert _cgroup_memory_limit(proc, root) == 8192
+
+
+def test_the_root_cgroup_limit_counts_too(tmp_path):
+    proc, root = _fake_cgroup(tmp_path, "/a", {"/": "65536", "/a": "max"})
+    assert _cgroup_memory_limit(proc, root) == 65536
+
+
+def test_no_cgroup_limit_at_all_is_none(tmp_path):
+    proc, root = _fake_cgroup(
+        tmp_path, "/a/b", {"/": "max", "/a": "max", "/a/b": "max"}
+    )
+    assert _cgroup_memory_limit(proc, root) is None
+
+
+def test_the_cgroup_line_is_found_among_cgroup_v1_lines(tmp_path):
+    proc, root = _fake_cgroup(tmp_path, "/x", {"/x": "512"})
+    proc.write_text("12:memory:/legacy\n1:name=systemd:/legacy\n0::/x\n")
+    assert _cgroup_memory_limit(proc, root) == 512
+
+
+def test_a_process_without_a_v2_cgroup_or_without_proc_has_no_limit(tmp_path):
+    proc, root = _fake_cgroup(tmp_path, "/x", {"/x": "512"})
+    proc.write_text("12:memory:/legacy\n")
+    assert _cgroup_memory_limit(proc, root) is None
+    assert _cgroup_memory_limit(tmp_path / "missing", root) is None
+
+
 def test_default_max_heap_is_half_of_the_memory(tmp_path):
     default = _diag(tmp_path)["max_heap"]
     full = _diag(tmp_path, "-XX:MaximumHeapSizePercent=100")["max_heap"]
     assert 0.45 <= default / full <= 0.55, (default, full)
-    if sys.platform == "linux":  # follows the container limit, not the host RAM
+    # on Linux the heap follows the cgroup limit (a container, a systemd scope), not the host RAM
+    if sys.platform == "linux":
         assert 0.40 <= default / _memory_limit() <= 0.60, default