From 0e60ead85b46c481dfc14e82f5e5975f9d5c695b Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Mon, 26 Oct 2026 06:14:25 +0900 Subject: [PATCH] Native: release the callback ids of every async executor call Only AsyncExecutor.transaction gave its callback ids back. create_record, query, command, new_edge, new_edge_by_keys and scan_type registered an id per callback in extras._callbacks and never released it, so the table grew without bound over a long-lived executor, and on_ok / on_error added two more each time they were set. All calls now go through one _queue helper. A queued call puts its ids on the executor's list after the library accepted it; wait_completion releases the ids that were queued before it started (a call queued by another thread during the wait keeps its callbacks, and a timed-out or interrupted wait puts the ids back); close releases everything. A call the library refuses gives its ids back at once. on_ok and on_error keep one id each until close, and a new handler replaces the old id. Python only: no change to the library. Co-Authored-By: Claude Sonnet 5.5 --- src/arcadedb/_native/extras.py | 121 ++++++++++---- tests/conftest.py | 2 + tests/test_native_async_callbacks.py | 238 +++++++++++++++++++++++++++ 3 files changed, 325 insertions(+), 36 deletions(-) create mode 100644 tests/test_native_async_callbacks.py diff --git a/src/arcadedb/_native/extras.py b/src/arcadedb/_native/extras.py index c374396..03a68d7 100644 --- a/src/arcadedb/_native/extras.py +++ b/src/arcadedb/_native/extras.py @@ -415,7 +415,14 @@ def __init__(self, database): self._db = database self._closed = False self._owner = database + # the ids (extras._callbacks) of the calls that are queued: wait_completion releases the ones that were queued before it + # started, close releases all of them. on_ok and on_error are executor-wide, so they keep one id each until close. self._cb_ids: list = [] + self._on_ok_id = 0 + self._on_error_id = 0 + self._cb_lock = ( + threading.Lock() + ) # guards the three above; never held across a call into the library def __enter__(self): return self @@ -497,6 +504,24 @@ def kill(self): self._op("kill") # work + def _queue(self, op: str, **args): + """Queue an async call whose callbacks (``cb``, ``ok``, ``err``) are already registered. + + The library keeps those ids until the writers have run the call, which only ``wait_completion`` and ``close`` know. So a + call that was queued puts its ids on the list that they release, and a call that the library refuses gives them back + at once, as no callback of it will ever run. The ids go on the list after the call is queued, not before, so the list + holds only calls that ``wait_completion`` can wait for. + """ + ids = [args[k] for k in ("cb", "ok", "err") if args.get(k)] + try: + self._op(op, **args) + except BaseException: + for cid in ids: + _forget(cid) + raise + with self._cb_lock: + self._cb_ids.extend(ids) + def _record_callback(self, callback, kind="d"): from .records import _wrap_dict @@ -512,7 +537,7 @@ def create_record(self, document, callback=None, error_callback=None): kind = "v" if hasattr(document, "get_out_edges") else "d" cb = self._record_callback(callback, kind) if callback is not None else 0 err = self._error_callback(error_callback) if error_callback is not None else 0 - self._op( + self._queue( "create_record", type=document._type, kind=kind, @@ -527,16 +552,6 @@ def _statement(self, op, language, text, callback, args, params, error_callback) args = tuple(args or ()) if args and params: raise ValueError("Use either positional args or named params, not both") - cb = 0 - if callback is not None: - - def on_result(message): - # the JPype wrapper calls the callback once per row of the completed statement - for row in NativeResultSet(self._db, message["reply"]): - callback(row) - - cb = _register(on_result) - err = self._error_callback(error_callback) if error_callback is not None else 0 if params: values = dict(params) elif args: @@ -547,7 +562,17 @@ def on_result(message): ) else: values = None - self._op(op, lang=language, text=text, params=values, cb=cb, err=err) + cb = 0 + if callback is not None: + + def on_result(message): + # the JPype wrapper calls the callback once per row of the completed statement + for row in NativeResultSet(self._db, message["reply"]): + callback(row) + + cb = _register(on_result) + err = self._error_callback(error_callback) if error_callback is not None else 0 + self._queue(op, lang=language, text=text, params=values, cb=cb, err=err) def query( self, language, query_text, callback, args=None, error_callback=None, **params @@ -618,12 +643,13 @@ def new_edge( callback=None, **properties, ): + src, dst = _rid_of(source_vertex), _rid_of(destination_vertex_or_rid) cb = self._edge_callback(callback) if callback is not None else 0 - self._op( + self._queue( "new_edge", - src=_rid_of(source_vertex), + src=src, type=edge_type, - dst=_rid_of(destination_vertex_or_rid), + dst=dst, light=bool(light), props=properties, cb=cb, @@ -660,7 +686,7 @@ def listed(names, values, what): destination_key_names, destination_key_values, "destination" ) cb = self._edge_callback(callback) if callback is not None else 0 - self._op( + self._queue( "new_edge_by_keys", src_type=source_vertex_type, src_names=src_names, @@ -701,7 +727,7 @@ def on_error(message): return {"v": True if result is None else bool(result)} err = _register(on_error) - self._op( + self._queue( "scan_type", type=type_name, polymorphic=bool(polymorphic), cb=cb, err=err ) @@ -726,37 +752,60 @@ def run(message): if error_callback is not None: args["err"] = self._error_callback(error_callback) # a retry calls the block again, so the callbacks live until the executor has finished its work (wait_completion, close) - self._cb_ids.extend(v for k, v in args.items() if k in ("cb", "ok", "err")) if slot is not None: args["slot"] = int(slot) - self._op("transaction", **args) - - def _release_callbacks(self): - """Forget the callbacks of the finished ``transaction`` calls.""" - ids, self._cb_ids = self._cb_ids, [] + self._queue("transaction", **args) + + def _release_callbacks(self, ids=None): + """Forget the callbacks of calls that have finished (``ids``), or all that the executor holds (on close).""" + if ids is None: + with self._cb_lock: + ids, self._cb_ids = self._cb_ids, [] + ids += [self._on_ok_id, self._on_error_id] + self._on_ok_id = self._on_error_id = 0 for cid in ids: _forget(cid) def wait_completion(self, timeout_ms: Optional[int] = None): - if timeout_ms is None: - self._op("wait") - elif timeout_ms < 0: + if timeout_ms is not None and timeout_ms < 0: raise ValueError(f"timeout_ms must be None or >= 0, got {timeout_ms}") - elif timeout_ms == 0: - if self.is_processing(): - raise TimeoutError("Async operations have not completed (0ms poll)") - elif not self._op("wait", timeout_ms=timeout_ms)["done"]: - raise TimeoutError( - f"Async operations did not complete within {timeout_ms}ms" - ) - self._release_callbacks() # nothing is pending any more + # the calls queued so far: a call that another thread queues while this one waits keeps its callbacks + with self._cb_lock: + finished, self._cb_ids = self._cb_ids, [] + try: + if timeout_ms is None: + self._op("wait") + elif timeout_ms == 0: + if self.is_processing(): + raise TimeoutError("Async operations have not completed (0ms poll)") + elif not self._op("wait", timeout_ms=timeout_ms)["done"]: + raise TimeoutError( + f"Async operations did not complete within {timeout_ms}ms" + ) + except BaseException: + with self._cb_lock: # still pending: the callbacks may still run + self._cb_ids[:0] = finished + raise + self._release_callbacks(finished) + + def _set_handler(self, op: str, slot: str, cid: int): + """Install an executor-wide handler; it replaces the earlier one, whose id goes back.""" + try: + self._op(op, cb=cid) + except BaseException: + _forget(cid) + raise + with self._cb_lock: + replaced = getattr(self, slot) + setattr(self, slot, cid) + _forget(replaced) def on_ok(self, callback: Callable[[], None]): - self._op("on_ok", cb=_register(lambda message: callback())) + self._set_handler("on_ok", "_on_ok_id", _register(lambda message: callback())) return self def on_error(self, callback: Callable[[Exception], None]): - self._op("on_error", cb=self._error_callback(callback)) + self._set_handler("on_error", "_on_error_id", self._error_callback(callback)) return self def close(self): diff --git a/tests/conftest.py b/tests/conftest.py index eaaeed1..c3eb374 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -51,6 +51,8 @@ collect_ignore.append("test_native_log_dir.py") # the heap limit that the native glue puts on the isolate (a GraalVM isolate argument; the JVM build has heap_size) collect_ignore.append("test_native_heap.py") + # the native build's callback table (extras._callbacks), which the async executor releases its ids from + collect_ignore.append("test_native_async_callbacks.py") if not os.path.isfile( os.path.join(os.path.dirname(os.path.dirname(__file__)), "docs", "index.md") ): diff --git a/tests/test_native_async_callbacks.py b/tests/test_native_async_callbacks.py new file mode 100644 index 0000000..4919ef3 --- /dev/null +++ b/tests/test_native_async_callbacks.py @@ -0,0 +1,238 @@ +"""Native build: the async executor releases the callback ids of its operations. + +Every callback of an async call is a Python function that the library calls back by number (``extras._callbacks``). Only +``transaction`` used to give its numbers back, so a long-lived executor grew the table with every ``create_record``, +``query``, ``command``, ``new_edge``, ``new_edge_by_keys`` and ``scan_type`` call that carried a callback. The numbers of +the calls that were queued are released when ``wait_completion`` returns, or when the executor is closed. The two +executor-wide handlers, ``on_ok`` and ``on_error``, last until the executor is closed (a new handler replaces the old one). +""" + +import pytest + +import arcadedb +from arcadedb._native import extras + +ROUNDS = 25 + + +def _size(): + with extras._callbacks_lock: + return len(extras._callbacks) + + +@pytest.fixture +def graph_db(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Item") + temp_db.command("sql", "CREATE VERTEX TYPE P") + temp_db.command("sql", "CREATE PROPERTY P.n INTEGER") + temp_db.command("sql", "CREATE INDEX ON P (n) UNIQUE") + temp_db.command("sql", "CREATE EDGE TYPE K") + with temp_db.transaction(): + temp_db.new_vertex("P").set("n", 1).save() + temp_db.new_vertex("P").set("n", 2).save() + return temp_db + + +def _submit_every_kind(db, executor, seen): + """One call of each kind that registers a callback, each with its own handlers.""" + a = db.query("sql", "SELECT FROM P WHERE n = 1").first().get_rid() + b = db.query("sql", "SELECT FROM P WHERE n = 2").first().get_rid() + doc = db.new_document("Item") + doc.set("n", 1) + executor.create_record( + doc, callback=lambda r: seen.append("create_record"), error_callback=seen.append + ) + executor.query( + "sql", + "SELECT FROM Item", + lambda row: seen.append("query"), + error_callback=seen.append, + ) + executor.command( + "sql", + "INSERT INTO Item SET n = ?", + callback=lambda row: seen.append("command"), + args=(2,), + error_callback=seen.append, + ) + executor.new_edge(a, "K", b, callback=lambda *edge: seen.append("new_edge")) + executor.new_edge_by_keys( + "P", + "n", + 1, + "P", + "n", + 2, + False, + "K", + False, + False, + callback=lambda *edge: seen.append("new_edge_by_keys"), + ) + executor.scan_type( + "Item", + lambda record: seen.append("scan_type"), + error_callback=lambda rid, exc: seen.append(exc), + ) + executor.transaction( + lambda: db.command("sql", "INSERT INTO Item SET n = 3"), + ok_callback=lambda: seen.append("transaction"), + error_callback=seen.append, + ) + + +def test_wait_completion_gives_back_the_ids_of_every_kind_of_call(graph_db): + baseline = _size() + executor = graph_db.async_executor() + seen = [] + for _ in range(ROUNDS): + _submit_every_kind(graph_db, executor, seen) + assert _size() > baseline # the calls hold their ids while they are queued + executor.wait_completion(30_000) + assert _size() == baseline + executor.close() + + assert _size() == baseline + # every kind ran its callback, so the ids were released after use and not instead of it + assert { + "create_record", + "query", + "command", + "new_edge", + "new_edge_by_keys", + "scan_type", + "transaction", + } <= set(seen) + assert not [s for s in seen if isinstance(s, Exception)] + + +def test_a_poll_that_finds_the_executor_busy_keeps_the_ids(graph_db): + baseline = _size() + executor = graph_db.async_executor() + seen = [] + _submit_every_kind(graph_db, executor, seen) + try: + executor.wait_completion( + 0 + ) # raises while the writers are busy, passes once they are not + except TimeoutError: + assert _size() > baseline + executor.wait_completion(30_000) + assert _size() == baseline + executor.close() + + +def test_a_call_the_engine_refuses_gives_back_its_ids(graph_db): + baseline = _size() + executor = graph_db.async_executor() + with pytest.raises(arcadedb.ArcadeDBError): + executor.new_edge( + "#99:99", # no such record: the engine refuses the call itself, no callback will ever run + "K", + "#99:98", + callback=lambda *edge: None, + ) + assert _size() == baseline + executor.close() + + +def test_close_gives_back_the_ids_of_calls_that_were_never_waited_for(graph_db): + baseline = _size() + executor = graph_db.async_executor() + seen = [] + for _ in range(5): + _submit_every_kind(graph_db, executor, seen) + assert _size() > baseline + executor.close() # no wait_completion: closing is the other release point + + assert _size() == baseline + + +def test_closing_the_database_gives_back_the_ids_of_its_executors(graph_db): + baseline = _size() + executor = graph_db.async_executor() + _submit_every_kind(graph_db, executor, []) + assert _size() > baseline + graph_db.close() + + assert executor.is_closed() + assert _size() == baseline + + +def test_on_ok_and_on_error_last_until_close_and_a_new_one_replaces_the_old(graph_db): + baseline = _size() + executor = graph_db.async_executor() + executor.on_ok(lambda: None).on_error(lambda exc: None) + assert _size() == baseline + 2 + for _ in range(ROUNDS): # the same two slots every time, not two more each time + executor.on_ok(lambda: None).on_error(lambda exc: None) + assert _size() == baseline + 2 + + executor.command("sql", "INSERT INTO Item SET n = 9") + executor.wait_completion(30_000) + assert _size() == baseline + 2 # a wait does not unhook the executor-wide handlers + + executor.close() + assert _size() == baseline + + +def test_the_executor_wide_handlers_still_fire_after_many_waits(graph_db): + ok = [] + executor = graph_db.async_executor().on_ok(lambda: ok.append(1)) + for n in range(5): + executor.command("sql", "INSERT INTO Item SET n = ?", n) + executor.wait_completion(30_000) + executor.close() + + assert ok + + +def test_a_new_on_ok_replaces_the_old_one_in_the_engine(graph_db): + # the id of a replaced handler goes back, so the engine must not call the replaced handler any more + first, second = [], [] + executor = graph_db.async_executor().on_ok(lambda: first.append(1)) + executor.on_ok(lambda: second.append(1)) + executor.command("sql", "INSERT INTO Item SET n = 1") + executor.wait_completion(30_000) + executor.close() + + assert second + assert not first + + +def test_calls_queued_while_another_thread_waits_keep_their_callbacks(graph_db): + import threading + + baseline = _size() + executor = graph_db.async_executor() + ran, errors = [], [] + producers = 4 + per_producer = 50 + + def produce(tag): + try: + for n in range(per_producer): + executor.command( + "sql", + "INSERT INTO Item SET n = ?", + callback=lambda row: ran.append(1), + args=(tag * 1000 + n,), + ) + except Exception as exc: # pragma: no cover - reported by the assert below + errors.append(exc) + + threads = [threading.Thread(target=produce, args=(t,)) for t in range(producers)] + for thread in threads: + thread.start() + while any(thread.is_alive() for thread in threads): + executor.wait_completion( + 30_000 + ) # releases the calls queued so far, not the later ones + for thread in threads: + thread.join() + executor.wait_completion(30_000) + executor.close() + + assert not errors + assert len(ran) == producers * per_producer + assert _size() == baseline