Skip to content

Commit 02ae1f6

Browse files
tae898claude
andcommitted
JVM async executor: define each JPype proxy class once per process, not once per callback (#238)
Every _create_*_callback helper declared a @jpype.JImplements class inside the call, so each callback built a Python type, a Java proxy class and a method table that stayed for the life of the process. callbacks.py already caches its proxy classes for the same reason. The classes are now built once per kind under a lock and the Python callables are state of the instance; the callbacks, their arguments and their order are unchanged. Tests count the class definitions and the type objects, and check that callbacks sharing a class keep their own arguments. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 1b315c1 commit 02ae1f6

3 files changed

Lines changed: 491 additions & 104 deletions

File tree

‎src/arcadedb/async_executor.py‎

Lines changed: 168 additions & 104 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
100000
3737
"""
3838

39+
import threading
3940
from collections.abc import Mapping
4041
from typing import TYPE_CHECKING, Any, Callable, Optional, Sequence, Union
4142

@@ -49,6 +50,163 @@
4950

5051
_LOGGER = get_logger(__name__)
5152

53+
# The proxy classes are defined once per process, on first use (the JVM must be running, so not at import).
54+
# A @jpype.JImplements class defined on every call costs a class build per callback (JPype creates a Python
55+
# type, a Java proxy class and its method table) and keeps that memory for the life of the process:
56+
# callbacks.py found about 6 KB of Python heap per definition (the same cache is used there).
57+
# The callbacks of one call are state of the instance, not of a closure: each call builds an instance
58+
# of the shared class around its own Python callables.
59+
_PROXY_CLASSES = {}
60+
# defined under a lock: on a free-threaded CPython (no GIL) two threads that submit their first callback of a kind together
61+
# would each define a class, and the second definition leaks like the ones this cache exists to avoid
62+
_PROXY_LOCK = threading.Lock()
63+
64+
65+
def _proxy_class(kind, build):
66+
"""The proxy class of ``kind``: ``build()`` defines it once and every later call gets that class."""
67+
cls = _PROXY_CLASSES.get(kind)
68+
if cls is None:
69+
with _PROXY_LOCK:
70+
cls = _PROXY_CLASSES.get(kind)
71+
if cls is None:
72+
cls = _PROXY_CLASSES[kind] = build()
73+
return cls
74+
75+
76+
def _build_new_record_proxy():
77+
from .graph import Document
78+
79+
@jpype.JImplements("com.arcadedb.database.async.NewRecordCallback")
80+
class _NewRecordCallback:
81+
def __init__(self, python_callback):
82+
self._callback = python_callback
83+
84+
@jpype.JOverride
85+
def call(self, record):
86+
self._callback(Document.wrap(record))
87+
88+
return _NewRecordCallback
89+
90+
91+
def _build_ok_proxy():
92+
@jpype.JImplements(jpype.JClass("com.arcadedb.database.async.OkCallback"))
93+
class PythonOkCallback:
94+
def __init__(self, python_callback):
95+
self._callback = python_callback
96+
97+
@jpype.JOverride
98+
def call(self):
99+
if self._callback:
100+
self._callback()
101+
102+
return PythonOkCallback
103+
104+
105+
def _build_new_edge_proxy():
106+
@jpype.JImplements(jpype.JClass("com.arcadedb.database.async.NewEdgeCallback"))
107+
class PythonNewEdgeCallback:
108+
def __init__(self, python_callback):
109+
self._callback = python_callback
110+
111+
@jpype.JOverride
112+
def call(
113+
self,
114+
edge,
115+
created_source_vertex,
116+
created_dest_vertex,
117+
):
118+
if self._callback:
119+
self._callback(edge, created_source_vertex, created_dest_vertex)
120+
121+
return PythonNewEdgeCallback
122+
123+
124+
def _build_error_proxy():
125+
@jpype.JImplements(jpype.JClass("com.arcadedb.database.async.ErrorCallback"))
126+
class PythonErrorCallback:
127+
def __init__(self, python_callback):
128+
self._callback = python_callback
129+
130+
@jpype.JOverride
131+
def call(self, exception):
132+
if self._callback:
133+
self._callback(exception)
134+
135+
return PythonErrorCallback
136+
137+
138+
def _build_result_proxy():
139+
@jpype.JImplements(
140+
jpype.JClass("com.arcadedb.database.async.AsyncResultsetCallback")
141+
)
142+
class PythonResultCallback:
143+
def __init__(self, python_callback, error_callback):
144+
self._callback = python_callback
145+
self._error_callback = error_callback
146+
147+
@jpype.JOverride
148+
def onComplete(self, resultset):
149+
if not self._callback:
150+
return
151+
from .results import Result
152+
153+
while resultset.hasNext():
154+
self._callback(Result(resultset.next()))
155+
156+
@jpype.JOverride
157+
def onError(self, exception):
158+
if self._error_callback:
159+
self._error_callback(exception)
160+
161+
return PythonResultCallback
162+
163+
164+
def _build_document_proxy():
165+
@jpype.JImplements(jpype.JClass("com.arcadedb.database.DocumentCallback"))
166+
class PythonDocumentCallback:
167+
def __init__(self, python_callback):
168+
self._callback = python_callback
169+
170+
@jpype.JOverride
171+
def onRecord(self, record):
172+
if self._callback is None:
173+
return True
174+
result = self._callback(record)
175+
return True if result is None else bool(result)
176+
177+
return PythonDocumentCallback
178+
179+
180+
def _build_error_record_proxy():
181+
@jpype.JImplements(jpype.JClass("com.arcadedb.engine.ErrorRecordCallback"))
182+
class PythonErrorRecordCallback:
183+
def __init__(self, python_callback):
184+
self._callback = python_callback
185+
186+
@jpype.JOverride
187+
def onErrorLoading(self, rid, exception):
188+
if self._callback is None:
189+
return True
190+
result = self._callback(rid, exception)
191+
return True if result is None else bool(result)
192+
193+
return PythonErrorRecordCallback
194+
195+
196+
def _build_transaction_scope_proxy():
197+
@jpype.JImplements(
198+
jpype.JClass("com.arcadedb.database.BasicDatabase$TransactionScope")
199+
)
200+
class PythonTransactionScope:
201+
def __init__(self, python_callback):
202+
self._callback = python_callback
203+
204+
@jpype.JOverride
205+
def execute(self):
206+
self._callback()
207+
208+
return PythonTransactionScope
209+
52210

53211
def _convert_column(values):
54212
"""Convert one non-numeric column, reusing the Java object per distinct str.
@@ -346,15 +504,7 @@ def create_record(
346504
)
347505

348506
def _create_new_record_callback(self, python_callback):
349-
from .graph import Document
350-
351-
@jpype.JImplements("com.arcadedb.database.async.NewRecordCallback")
352-
class _NewRecordCallback:
353-
@jpype.JOverride
354-
def call(self, record):
355-
python_callback(Document.wrap(record))
356-
357-
return _NewRecordCallback()
507+
return _proxy_class("new_record", _build_new_record_proxy)(python_callback)
358508

359509
def new_edge(
360510
self,
@@ -963,19 +1113,7 @@ def on_error(self, callback: Callable[[Exception], None]) -> "AsyncExecutor":
9631113

9641114
def _create_ok_callback(self, python_callback):
9651115
"""Create Java OkCallback from Python function."""
966-
OkCallback = jpype.JClass("com.arcadedb.database.async.OkCallback")
967-
968-
# Capture callback in closure
969-
callback = python_callback
970-
971-
@jpype.JImplements(OkCallback)
972-
class PythonOkCallback:
973-
@jpype.JOverride
974-
def call(self):
975-
if callback:
976-
callback()
977-
978-
return PythonOkCallback()
1116+
return _proxy_class("ok", _build_ok_proxy)(python_callback)
9791117

9801118
def _create_new_edge_callback(self, python_callback):
9811119
"""Create Java NewEdgeCallback from Python function.
@@ -985,103 +1123,29 @@ def _create_new_edge_callback(self, python_callback):
9851123
- created_source_vertex: True if source vertex was created
9861124
- created_dest_vertex: True if destination vertex was created
9871125
"""
988-
NewEdgeCallback = jpype.JClass("com.arcadedb.database.async.NewEdgeCallback")
989-
990-
# Capture callback in closure
991-
callback = python_callback
992-
993-
@jpype.JImplements(NewEdgeCallback)
994-
class PythonNewEdgeCallback:
995-
@jpype.JOverride
996-
def call(
997-
self,
998-
edge,
999-
created_source_vertex,
1000-
created_dest_vertex,
1001-
):
1002-
if callback:
1003-
callback(edge, created_source_vertex, created_dest_vertex)
1004-
1005-
return PythonNewEdgeCallback()
1126+
return _proxy_class("new_edge", _build_new_edge_proxy)(python_callback)
10061127

10071128
def _create_error_callback(self, python_callback):
10081129
"""Create Java ErrorCallback from Python function."""
1009-
ErrorCallback = jpype.JClass("com.arcadedb.database.async.ErrorCallback")
1010-
1011-
@jpype.JImplements(ErrorCallback)
1012-
class PythonErrorCallback:
1013-
@jpype.JOverride
1014-
def call(self, exception):
1015-
if python_callback:
1016-
python_callback(exception)
1017-
1018-
return PythonErrorCallback()
1130+
return _proxy_class("error", _build_error_proxy)(python_callback)
10191131

10201132
def _create_result_callback(self, python_callback, error_callback=None):
10211133
"""Create Java AsyncResultsetCallback from Python functions."""
1022-
AsyncResultsetCallback = jpype.JClass(
1023-
"com.arcadedb.database.async.AsyncResultsetCallback"
1134+
return _proxy_class("result", _build_result_proxy)(
1135+
python_callback, error_callback
10241136
)
10251137

1026-
@jpype.JImplements(AsyncResultsetCallback)
1027-
class PythonResultCallback:
1028-
@jpype.JOverride
1029-
def onComplete(self, resultset):
1030-
if not python_callback:
1031-
return
1032-
from .results import Result
1033-
1034-
while resultset.hasNext():
1035-
python_callback(Result(resultset.next()))
1036-
1037-
@jpype.JOverride
1038-
def onError(self, exception):
1039-
if error_callback:
1040-
error_callback(exception)
1041-
1042-
return PythonResultCallback()
1043-
10441138
def _create_document_callback(self, python_callback):
1045-
DocumentCallback = jpype.JClass("com.arcadedb.database.DocumentCallback")
1046-
1047-
@jpype.JImplements(DocumentCallback)
1048-
class PythonDocumentCallback:
1049-
@jpype.JOverride
1050-
def onRecord(self, record):
1051-
if python_callback is None:
1052-
return True
1053-
result = python_callback(record)
1054-
return True if result is None else bool(result)
1055-
1056-
return PythonDocumentCallback()
1139+
return _proxy_class("document", _build_document_proxy)(python_callback)
10571140

10581141
def _create_error_record_callback(self, python_callback):
1059-
ErrorRecordCallback = jpype.JClass("com.arcadedb.engine.ErrorRecordCallback")
1060-
1061-
@jpype.JImplements(ErrorRecordCallback)
1062-
class PythonErrorRecordCallback:
1063-
@jpype.JOverride
1064-
def onErrorLoading(self, rid, exception):
1065-
if python_callback is None:
1066-
return True
1067-
result = python_callback(rid, exception)
1068-
return True if result is None else bool(result)
1069-
1070-
return PythonErrorRecordCallback()
1142+
return _proxy_class("error_record", _build_error_record_proxy)(python_callback)
10711143

10721144
def _create_transaction_scope(self, python_callback):
1073-
TransactionScope = jpype.JClass(
1074-
"com.arcadedb.database.BasicDatabase$TransactionScope"
1145+
return _proxy_class("transaction_scope", _build_transaction_scope_proxy)(
1146+
python_callback
10751147
)
10761148

1077-
@jpype.JImplements(TransactionScope)
1078-
class PythonTransactionScope:
1079-
@jpype.JOverride
1080-
def execute(self):
1081-
python_callback()
1082-
1083-
return PythonTransactionScope()
1084-
10851149
def _unwrap_record(self, record):
10861150
java_document = getattr(record, "_java_document", None)
10871151
return java_document if java_document is not None else record

‎tests/conftest.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@
9797
"test_build_no_uv_refresh.py": "build scripts of the JVM wheel",
9898
"test_build_wheel_check.py": "build scripts of the JVM wheel",
9999
"test_query_vector_params_jvm.py": "the java.util objects the JVM build's parameter conversion builds",
100+
"test_async_proxy_cache.py": "the JPype proxy classes of the JVM build's async executor",
100101
}
101102
if BACKEND == "native" and not os.environ.get("NATIVE_PARITY_ALL"):
102103
collect_ignore.extend(NATIVE_NOT_APPLICABLE)

0 commit comments

Comments
 (0)