diff --git a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java b/native/src/main/java/org/humemai/nativelib/ArcadeNative.java index 2cbd978..37eb836 100644 --- a/native/src/main/java/org/humemai/nativelib/ArcadeNative.java +++ b/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 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 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; } } diff --git a/src/arcadedb/_native/database.py b/src/arcadedb/_native/database.py index d20cb82..5463d08 100644 --- a/src/arcadedb/_native/database.py +++ b/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"): diff --git a/src/arcadedb/_native/lib.py b/src/arcadedb/_native/lib.py index 85fea6e..984e799 100644 --- a/src/arcadedb/_native/lib.py +++ b/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) diff --git a/tests/test_native_heap.py b/tests/test_native_heap.py index b0bfcd9..05150de 100644 --- a/tests/test_native_heap.py +++ b/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:: line of /proc/self/cgroup) is read first: //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 diff --git a/tests/test_native_result_transport.py b/tests/test_native_result_transport.py new file mode 100644 index 0000000..33224fb --- /dev/null +++ b/tests/test_native_result_transport.py @@ -0,0 +1,175 @@ +"""Query results across the native boundary. The reply text (and the float32 tail of a reply with vectors) the image hands to Python is +copied out of the image in bulk, not one byte at a time, and the batches after the first carry float arrays of 8 or more elements in a +float32 tail (arcadedb_n_next_bin), not as number text. Every value has to come back exactly as it went in, for replies of every size: +tiny, several batches (arcadedb_n_exec returns the first 256 rows, arcadedb_n_next or arcadedb_n_next_bin the following 2048 at a time) +and single rows of about a megabyte of multi-byte text. The tests use the public API only: ARCADEDB_NATIVE_NEXT_BIN=0 runs them over the +number-text path of the later batches. +""" + +import struct + +DIM = 16 # 8 or more floats per vector: they travel in a float32 tail (as JSON numbers in the later batches with ARCADEDB_NATIVE_NEXT_BIN=0) + +TEXTS = [ + "", + "plain", + "café 日本語 \U0001f600 end", # 2-, 3- and 4-byte UTF-8 sequences + 'quote " backslash \\ slash / newline \n tab \t nul-free \u0001 control', + "x" * 300, +] + + +def f32(x): + """The double a float32 value widens to (what the engine returns for an ARRAY_OF_FLOATS element).""" + return struct.unpack("f", struct.pack("f", x))[0] + + +def _rows(n): + rows = [] + for i in range(n): + rows.append( + { + "id": i, + "big": (1 << 62) - i if i % 2 else -(1 << 62) + i, + "val": (i + 0.5) / 7.0, + "txt": TEXTS[i % len(TEXTS)], + "vec": [f32((i * DIM + k) / 1000.0 - 3.0) for k in range(DIM)], + } + ) + return rows + + +def _typed_db(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE R") + for p, t in ( + ("id", "INTEGER"), + ("big", "LONG"), + ("val", "DOUBLE"), + ("txt", "STRING"), + ("vec", "ARRAY_OF_FLOATS"), + ): + temp_db.command("sql", f"CREATE PROPERTY R.{p} {t}") + return temp_db + + +def test_a_reply_of_several_batches_round_trips_exactly(temp_db): + db = _typed_db(temp_db) + rows = _rows( + 5000 + ) # 256 in the first reply, then 2048 + 2048 + 648, vectors in the float tail of each + db.insert_many("R", rows) + got = db.query("sql", "SELECT id, big, val, txt, vec FROM R ORDER BY id").to_list() + assert len(got) == len(rows) + for want, have in zip(rows, got): + assert have["id"] == want["id"] + assert have["big"] == want["big"] + assert have["val"] == want["val"] + assert have["txt"] == want["txt"] + assert list(have["vec"]) == want["vec"] + + +def test_a_small_reply_with_a_vector_round_trips_exactly(temp_db): + db = _typed_db(temp_db) + rows = _rows(3) + db.insert_many("R", rows) + got = db.query("sql", "SELECT id, vec, txt FROM R ORDER BY id").to_list() + assert [list(r["vec"]) for r in got] == [r["vec"] for r in rows] + assert [r["txt"] for r in got] == [r["txt"] for r in rows] + + +def test_rows_of_about_a_megabyte_of_multibyte_text_round_trip(temp_db): + db = _typed_db(temp_db) + unit = "aé日\U0001f600" # 1 + 2 + 3 + 4 bytes + rows = [ + {"id": i, "txt": unit * (250_000 + i)} for i in range(4) + ] # ~1.2 MB of UTF-8 per row + db.insert_many("R", rows) + got = db.query("sql", "SELECT id, txt FROM R ORDER BY id").to_list() + assert [(r["id"], r["txt"]) for r in got] == [(r["id"], r["txt"]) for r in rows] + + +def test_empty_and_one_row_replies(temp_db): + db = _typed_db(temp_db) + assert db.query("sql", "SELECT FROM R").to_list() == [] + assert db.query("sql", "SELECT '' AS s").to_list() == [{"s": ""}] + assert db.query("sql", "SELECT 1 AS x").first().get("x") == 1 + + +# ---- vectors on every batch (first reply and arcadedb_n_next batches), bit-exact, with short, empty and missing vectors + +SPECIAL = [ + float("inf"), + float("-inf"), + -0.0, + 0.0, + 1e-45, + 3.4028235e38, + 0.1, + -123456.789, +] # 8 values: a float32 tail on the binary path + + +def _bits(x): + """The 8 bytes of a double as hex (any NaN is 'nan'): bit-exact comparison, so -0.0 differs from 0.0.""" + return "nan" if x != x else struct.pack("