diff --git a/src/arcadedb/async_executor.py b/src/arcadedb/async_executor.py index 8377b95..66b18af 100644 --- a/src/arcadedb/async_executor.py +++ b/src/arcadedb/async_executor.py @@ -36,6 +36,7 @@ 100000 """ +import threading from collections.abc import Mapping from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence, Union @@ -49,6 +50,163 @@ _LOGGER = get_logger(__name__) +# The proxy classes are defined once per process, on first use (the JVM must be running, so not at import). +# A @jpype.JImplements class defined on every call costs a class build per callback (JPype creates a Python +# type, a Java proxy class and its method table) and keeps that memory for the life of the process: +# callbacks.py found about 6 KB of Python heap per definition (the same cache is used there). +# The callbacks of one call are state of the instance, not of a closure: each call builds an instance +# of the shared class around its own Python callables. +_PROXY_CLASSES = {} +# defined under a lock: on a free-threaded CPython (no GIL) two threads that submit their first callback of a kind together +# would each define a class, and the second definition leaks like the ones this cache exists to avoid +_PROXY_LOCK = threading.Lock() + + +def _proxy_class(kind, build): + """The proxy class of ``kind``: ``build()`` defines it once and every later call gets that class.""" + cls = _PROXY_CLASSES.get(kind) + if cls is None: + with _PROXY_LOCK: + cls = _PROXY_CLASSES.get(kind) + if cls is None: + cls = _PROXY_CLASSES[kind] = build() + return cls + + +def _build_new_record_proxy(): + from .graph import Document + + @jpype.JImplements("com.arcadedb.database.async.NewRecordCallback") + class _NewRecordCallback: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def call(self, record): + self._callback(Document.wrap(record)) + + return _NewRecordCallback + + +def _build_ok_proxy(): + @jpype.JImplements(jpype.JClass("com.arcadedb.database.async.OkCallback")) + class PythonOkCallback: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def call(self): + if self._callback: + self._callback() + + return PythonOkCallback + + +def _build_new_edge_proxy(): + @jpype.JImplements(jpype.JClass("com.arcadedb.database.async.NewEdgeCallback")) + class PythonNewEdgeCallback: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def call( + self, + edge, + created_source_vertex, + created_dest_vertex, + ): + if self._callback: + self._callback(edge, created_source_vertex, created_dest_vertex) + + return PythonNewEdgeCallback + + +def _build_error_proxy(): + @jpype.JImplements(jpype.JClass("com.arcadedb.database.async.ErrorCallback")) + class PythonErrorCallback: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def call(self, exception): + if self._callback: + self._callback(exception) + + return PythonErrorCallback + + +def _build_result_proxy(): + @jpype.JImplements( + jpype.JClass("com.arcadedb.database.async.AsyncResultsetCallback") + ) + class PythonResultCallback: + def __init__(self, python_callback, error_callback): + self._callback = python_callback + self._error_callback = error_callback + + @jpype.JOverride + def onComplete(self, resultset): + if not self._callback: + return + from .results import Result + + while resultset.hasNext(): + self._callback(Result(resultset.next())) + + @jpype.JOverride + def onError(self, exception): + if self._error_callback: + self._error_callback(exception) + + return PythonResultCallback + + +def _build_document_proxy(): + @jpype.JImplements(jpype.JClass("com.arcadedb.database.DocumentCallback")) + class PythonDocumentCallback: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def onRecord(self, record): + if self._callback is None: + return True + result = self._callback(record) + return True if result is None else bool(result) + + return PythonDocumentCallback + + +def _build_error_record_proxy(): + @jpype.JImplements(jpype.JClass("com.arcadedb.engine.ErrorRecordCallback")) + class PythonErrorRecordCallback: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def onErrorLoading(self, rid, exception): + if self._callback is None: + return True + result = self._callback(rid, exception) + return True if result is None else bool(result) + + return PythonErrorRecordCallback + + +def _build_transaction_scope_proxy(): + @jpype.JImplements( + jpype.JClass("com.arcadedb.database.BasicDatabase$TransactionScope") + ) + class PythonTransactionScope: + def __init__(self, python_callback): + self._callback = python_callback + + @jpype.JOverride + def execute(self): + self._callback() + + return PythonTransactionScope + def _convert_column(values): """Convert one non-numeric column, reusing the Java object per distinct str. @@ -346,15 +504,7 @@ def create_record( ) def _create_new_record_callback(self, python_callback): - from .graph import Document - - @jpype.JImplements("com.arcadedb.database.async.NewRecordCallback") - class _NewRecordCallback: - @jpype.JOverride - def call(self, record): - python_callback(Document.wrap(record)) - - return _NewRecordCallback() + return _proxy_class("new_record", _build_new_record_proxy)(python_callback) def new_edge( self, @@ -963,19 +1113,7 @@ def on_error(self, callback: Callable[[Exception], None]) -> "AsyncExecutor": def _create_ok_callback(self, python_callback): """Create Java OkCallback from Python function.""" - OkCallback = jpype.JClass("com.arcadedb.database.async.OkCallback") - - # Capture callback in closure - callback = python_callback - - @jpype.JImplements(OkCallback) - class PythonOkCallback: - @jpype.JOverride - def call(self): - if callback: - callback() - - return PythonOkCallback() + return _proxy_class("ok", _build_ok_proxy)(python_callback) def _create_new_edge_callback(self, python_callback): """Create Java NewEdgeCallback from Python function. @@ -985,103 +1123,29 @@ def _create_new_edge_callback(self, python_callback): - created_source_vertex: True if source vertex was created - created_dest_vertex: True if destination vertex was created """ - NewEdgeCallback = jpype.JClass("com.arcadedb.database.async.NewEdgeCallback") - - # Capture callback in closure - callback = python_callback - - @jpype.JImplements(NewEdgeCallback) - class PythonNewEdgeCallback: - @jpype.JOverride - def call( - self, - edge, - created_source_vertex, - created_dest_vertex, - ): - if callback: - callback(edge, created_source_vertex, created_dest_vertex) - - return PythonNewEdgeCallback() + return _proxy_class("new_edge", _build_new_edge_proxy)(python_callback) def _create_error_callback(self, python_callback): """Create Java ErrorCallback from Python function.""" - ErrorCallback = jpype.JClass("com.arcadedb.database.async.ErrorCallback") - - @jpype.JImplements(ErrorCallback) - class PythonErrorCallback: - @jpype.JOverride - def call(self, exception): - if python_callback: - python_callback(exception) - - return PythonErrorCallback() + return _proxy_class("error", _build_error_proxy)(python_callback) def _create_result_callback(self, python_callback, error_callback=None): """Create Java AsyncResultsetCallback from Python functions.""" - AsyncResultsetCallback = jpype.JClass( - "com.arcadedb.database.async.AsyncResultsetCallback" + return _proxy_class("result", _build_result_proxy)( + python_callback, error_callback ) - @jpype.JImplements(AsyncResultsetCallback) - class PythonResultCallback: - @jpype.JOverride - def onComplete(self, resultset): - if not python_callback: - return - from .results import Result - - while resultset.hasNext(): - python_callback(Result(resultset.next())) - - @jpype.JOverride - def onError(self, exception): - if error_callback: - error_callback(exception) - - return PythonResultCallback() - def _create_document_callback(self, python_callback): - DocumentCallback = jpype.JClass("com.arcadedb.database.DocumentCallback") - - @jpype.JImplements(DocumentCallback) - class PythonDocumentCallback: - @jpype.JOverride - def onRecord(self, record): - if python_callback is None: - return True - result = python_callback(record) - return True if result is None else bool(result) - - return PythonDocumentCallback() + return _proxy_class("document", _build_document_proxy)(python_callback) def _create_error_record_callback(self, python_callback): - ErrorRecordCallback = jpype.JClass("com.arcadedb.engine.ErrorRecordCallback") - - @jpype.JImplements(ErrorRecordCallback) - class PythonErrorRecordCallback: - @jpype.JOverride - def onErrorLoading(self, rid, exception): - if python_callback is None: - return True - result = python_callback(rid, exception) - return True if result is None else bool(result) - - return PythonErrorRecordCallback() + return _proxy_class("error_record", _build_error_record_proxy)(python_callback) def _create_transaction_scope(self, python_callback): - TransactionScope = jpype.JClass( - "com.arcadedb.database.BasicDatabase$TransactionScope" + return _proxy_class("transaction_scope", _build_transaction_scope_proxy)( + python_callback ) - @jpype.JImplements(TransactionScope) - class PythonTransactionScope: - @jpype.JOverride - def execute(self): - python_callback() - - return PythonTransactionScope() - def _unwrap_record(self, record): java_document = getattr(record, "_java_document", None) return java_document if java_document is not None else record diff --git a/tests/conftest.py b/tests/conftest.py index 01ab332..4488d12 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -97,6 +97,7 @@ "test_build_no_uv_refresh.py": "build scripts of the JVM wheel", "test_build_wheel_check.py": "build scripts of the JVM wheel", "test_query_vector_params_jvm.py": "the java.util objects the JVM build's parameter conversion builds", + "test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor", } if BACKEND == "native" and not os.environ.get("NATIVE_PARITY_ALL"): collect_ignore.extend(NATIVE_NOT_APPLICABLE) diff --git a/tests/test_async_proxy_cache.py b/tests/test_async_proxy_cache.py new file mode 100644 index 0000000..8152a38 --- /dev/null +++ b/tests/test_async_proxy_cache.py @@ -0,0 +1,322 @@ +"""The JVM build's async executor defines each JPype proxy class once, not once per callback. + +Every callback of an async call (``create_record``, ``command``, ``query``, ``new_edge``, ``scan_type``, +``transaction``, ``on_ok``, ``on_error``) is a Java object that calls back into Python. The executor used to define a new +``@jpype.JImplements`` class for each one, which is a Python type, a Java proxy class and a method table per callback, +kept for the life of the process (the same cost that ``callbacks.py`` removed for Python functions in SQL). +The classes are now defined once per kind, and the Python callables of one call are state of its instance. + +These tests count the class definitions, check that no type objects pile up, and check that what a callback sees is +unchanged: each callback gets the arguments of its own call (no crosstalk between callbacks that share a class), and +the delivery order of a one-writer executor is the submission order. +""" + +import gc +import threading + +import jpype +import pytest + +from arcadedb import async_executor + +# the proxy kinds of the executor: new_record, ok, new_edge, error, result, document, error_record, transaction_scope +KINDS = 8 + + +@pytest.fixture +def items_db(temp_db): + temp_db.command("sql", "CREATE DOCUMENT TYPE Item") + temp_db.command("sql", "CREATE PROPERTY Item.n INTEGER") + temp_db.command("sql", "CREATE VERTEX TYPE P") + temp_db.command("sql", "CREATE PROPERTY P.n INTEGER") + temp_db.command("sql", "CREATE EDGE TYPE K") + return temp_db + + +@pytest.fixture +def definitions(monkeypatch): + """Count the ``jpype.JImplements`` class definitions, starting from an empty class cache.""" + count = {"n": 0} + real = jpype.JImplements + + def counting(*args, **kwargs): + count["n"] += 1 + return real(*args, **kwargs) + + monkeypatch.setattr(jpype, "JImplements", counting) + # raising=False: a tree without the cache has no such name, and then every registration defines a class + monkeypatch.setattr(async_executor, "_PROXY_CLASSES", {}, raising=False) + return count + + +def _register_every_kind(executor, db, rounds): + """Build the Java proxy of every kind ``rounds`` times, without submitting anything.""" + noop = lambda *_a: None # noqa: E731 + for _ in range(rounds): + executor._create_new_record_callback(noop) + executor._create_ok_callback(noop) + executor._create_new_edge_callback(noop) + executor._create_error_callback(noop) + executor._create_result_callback(noop, noop) + executor._create_document_callback(noop) + executor._create_error_record_callback(noop) + executor._create_transaction_scope(noop) + + +def test_proxy_classes_are_defined_once_per_kind_across_executors( + items_db, definitions +): + for _ in range(3): + executor = items_db.async_executor() + _register_every_kind(executor, items_db, rounds=50) + executor.close() + + assert 0 < definitions["n"] <= KINDS, definitions["n"] + + +def test_a_callback_heavy_load_defines_no_class_after_the_first_round( + items_db, definitions +): + executor = items_db.async_executor().set_parallel_level(1).set_commit_every(50) + seen = [] + # the first round defines the classes + executor.command( + "sql", "INSERT INTO Item SET n = ?", callback=seen.append, args=(-1,) + ) + doc = items_db.new_document("Item") + doc.set("n", -2) + executor.create_record(doc, callback=seen.append, error_callback=seen.append) + executor.query("sql", "SELECT FROM Item", seen.append) + executor.wait_completion() + after_first = definitions["n"] + + for i in range(200): + executor.command( + "sql", "INSERT INTO Item SET n = ?", callback=seen.append, args=(i,) + ) + doc = items_db.new_document("Item") + doc.set("n", i) + executor.create_record(doc, callback=seen.append, error_callback=seen.append) + executor.query("sql", "SELECT FROM Item WHERE n = ?", seen.append, args=(i,)) + executor.wait_completion() + executor.close() + + assert after_first <= KINDS + assert definitions["n"] == after_first + assert items_db.count_type("Item") == 2 + 400 + + +def test_registrations_do_not_pile_up_type_objects(items_db): + executor = items_db.async_executor() + _register_every_kind(executor, items_db, rounds=2) # warm: the classes exist + + def types(): + gc.collect() + return sum(1 for o in gc.get_objects() if isinstance(o, type)) + + before = types() + _register_every_kind(executor, items_db, rounds=100) # 800 proxies + grown = types() - before + executor.close() + + assert grown < 50, f"{grown} new type objects after 800 registrations" + + +def test_concurrent_first_use_defines_the_class_once(items_db, definitions): + executor = items_db.async_executor() + barrier = threading.Barrier(8) + errors = [] + + def first_use(): + try: + barrier.wait(30) + executor._create_ok_callback(lambda: None) + except Exception as exc: # noqa: BLE001 + errors.append(exc) + + threads = [threading.Thread(target=first_use) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join(30) + executor.close() + + assert not errors, errors + assert definitions["n"] == 1, definitions["n"] + + +def test_each_command_callback_gets_its_own_rows(items_db): + """Callbacks share a proxy class; none of them may see another call's result or error.""" + got = {tag: [] for tag in "abc"} + failed = {"x": [], "y": []} + executor = items_db.async_executor().set_parallel_level(1).set_commit_every(1) + + for i, tag in enumerate("abc"): + executor.command( + "sql", + "INSERT INTO Item SET n = ?", + callback=lambda r, tag=tag: got[tag].append(r.get("n")), + error_callback=lambda e: failed["y"].append(e), + args=(i,), + ) + executor.command( + "sql", + "INSERT INTO NoSuchType SET n = 1", + callback=lambda r: failed["y"].append(r), + error_callback=lambda e: failed["x"].append(str(e)), + ) + executor.wait_completion() + executor.close() + + assert got == {"a": [0], "b": [1], "c": [2]} + assert len(failed["x"]) == 1 and "NoSuchType" in failed["x"][0] + assert failed["y"] == [] + + +def test_query_callbacks_keep_row_order_and_do_not_mix(items_db): + with items_db.transaction(): + for i in range(30): + items_db.command("sql", "INSERT INTO Item SET n = ?", i) + low, high = [], [] + executor = items_db.async_executor().set_parallel_level(1) + executor.query( + "sql", + "SELECT n FROM Item WHERE n < 10 ORDER BY n", + lambda r: low.append(r.get("n")), + ) + executor.query( + "sql", + "SELECT n FROM Item WHERE n >= 20 ORDER BY n", + lambda r: high.append(r.get("n")), + ) + executor.wait_completion() + executor.close() + + assert low == list(range(10)) + assert high == list(range(20, 30)) + + +def test_create_record_callbacks_run_in_submission_order_on_one_writer(items_db): + seen, errors = [], [] + executor = items_db.async_executor().set_parallel_level(1).set_commit_every(25) + for i in range(200): + doc = items_db.new_document("Item") + doc.set("n", i) + executor.create_record( + doc, + callback=lambda d: seen.append(d.get("n")), + error_callback=errors.append, + ) + executor.wait_completion() + executor.close() + + assert errors == [] + assert seen == list(range(200)) + assert items_db.count_type("Item") == 200 + + +def test_create_record_callback_delivers_every_record_at_parallel_four(items_db): + seen = [] + lock = threading.Lock() + + def on_record(d): + with lock: + seen.append(d.get("n")) + + executor = items_db.async_executor().set_parallel_level(4).set_commit_every(25) + for i in range(200): + doc = items_db.new_document("Item") + doc.set("n", i) + executor.create_record(doc, callback=on_record) + executor.wait_completion() + executor.close() + + assert sorted(seen) == list(range(200)) + + +def test_new_edge_callback_gets_the_edge_and_the_two_flags(items_db): + with items_db.transaction(): + a = items_db.new_vertex("P").set("n", 1).save() + b = items_db.new_vertex("P").set("n", 2).save() + calls = [] + executor = items_db.async_executor().set_parallel_level(1) + executor.new_edge(a, "K", b, callback=lambda *args: calls.append(args)) + executor.wait_completion() + executor.close() + + assert len(calls) == 1 + edge, created_source, created_dest = calls[0] + assert str(edge.getTypeName()) == "K" + assert created_source in (True, False) and created_dest in (True, False) + + +def test_scan_type_and_error_callbacks_still_see_every_record(items_db): + with items_db.transaction(): + for i in range(10): + items_db.command("sql", "INSERT INTO Item SET n = ?", i) + seen, lock = [], threading.Lock() + + def on_record(record): + with lock: + seen.append(int(record.get("n"))) + # None means "keep going" + + executor = items_db.async_executor() + executor.scan_type("Item", on_record, error_callback=lambda rid, exc: True) + executor.wait_completion() + executor.close() + + assert sorted(seen) == list(range(10)) + + +def test_transactions_run_their_own_block_and_report_to_their_own_handlers(items_db): + ran, ok, failed = [], [], [] + executor = items_db.async_executor().set_parallel_level(1) + + def good(): + ran.append("good") + items_db.command("sql", "INSERT INTO Item SET n = 1") + + def bad(): + ran.append("bad") + raise ValueError("boom") + + executor.transaction( + good, + retries=1, + ok_callback=lambda: ok.append("good"), + error_callback=lambda e: failed.append(("good", e)), + ) + executor.transaction( + bad, + retries=1, + ok_callback=lambda: ok.append("bad"), + error_callback=lambda e: failed.append(("bad", e)), + ) + executor.wait_completion(30_000) + executor.close() + + assert ran == ["good", "bad"] + assert ok == ["good"] + assert [tag for tag, _ in failed] == ["bad"] + assert items_db.count_type("Item") == 1 + + +def test_executor_wide_handlers_still_fire(items_db): + ok, errors = [], [] + executor = ( + items_db.async_executor() + .set_parallel_level(1) + .set_commit_every(1) + .on_ok(lambda: ok.append(1)) + .on_error(lambda e: errors.append(str(e))) + ) + for i in range(5): + doc = items_db.new_document("Item") + doc.set("n", i) + executor.create_record(doc) + executor.wait_completion() + executor.close() + + assert len(ok) >= 1 + assert errors == []