weiqingy commented on code in PR #944:
URL: https://github.com/apache/flink-agents/pull/944#discussion_r3696783616
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String
pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL,
pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
Review Comment:
Combining both failures and rethrowing is the right call. What I keep
looking at is what happens to this exception one frame up:
```java
// PythonBridgeManager.close(), lines 292-302
if (pythonActionExecutor != null) { pythonActionExecutor.close(); }
if (pythonInterpreter != null) { pythonInterpreter.close(); }
if (pythonEnvironmentManager != null) { pythonEnvironmentManager.close(); }
```
That is a plain sequence, so on exactly the failure path this PR is built
for, `pythonInterpreter.close()` never runs, and that is the release that tears
down the interpreter owning every handle still outstanding.
`ActionExecutionOperator.close()` (lines 469-489) has the same shape across its
five closes.
To be clear, this is pre-existing. The old `close()` threw on a failed
`interpreter.invoke` too, so nothing has regressed here. But given the stated
goal is "releases both handles even if one cleanup operation fails", how do you
see that goal holding one frame up? Carrying the same `firstOrSuppressed`
pattern into `PythonBridgeManager.close()` would make the guarantee end-to-end,
though I may be missing a reason the interpreter is fine to leak on that path.
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String
pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL,
pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
Review Comment:
nit: The copy-then-null reads as defensive style, but it looks load-bearing.
`PyObject.close()` in pemja 0.5.7 is an unguarded `decRef(tState, pyobject)`
with no null check and no double-close flag, so clearing the fields first is
the only thing stopping a repeated `close()` from decrementing a second time on
an already-released handle. Someone later tidying this into
`closePythonObject(CLOSE_ASYNC_THREAD_POOL, pythonAsyncThreadPool)` would drop
that quietly, and the only assertion that would notice is the three-line tail
of `releasesBothPythonObjectsWhenLogicalCleanupFails`.
Would a short comment here save the next reader that trip? There is
precedent right next door at `ActionExecutionOperator.java:471` (`// Must close
before pythonInterpreter since cached resources may hold Python references.`).
Something like this, if it helps:
```java
// Clear the fields before releasing: PyObject.close() is an unguarded
native decRef,
// so a repeated close() must not reach the same handle twice.
```
##########
runtime/src/main/java/org/apache/flink/agents/runtime/python/utils/PythonActionExecutor.java:
##########
@@ -188,17 +189,32 @@ public boolean callPythonAwaitable(String
pythonAwaitableRef) {
}
public void close() throws Exception {
- if (interpreter != null) {
- if (pythonAsyncThreadPool != null) {
- interpreter.invoke(CLOSE_ASYNC_THREAD_POOL,
pythonAsyncThreadPool);
- }
+ PyObject asyncThreadPool = pythonAsyncThreadPool;
+ PyObject runnerContext = pythonRunnerContext;
+ pythonAsyncThreadPool = null;
+ pythonRunnerContext = null;
+
+ Exception exception = null;
+ try {
+ closePythonObject(CLOSE_ASYNC_THREAD_POOL, asyncThreadPool);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+ try {
+ closePythonObject(CLOSE_FLINK_RUNNER_CONTEXT, runnerContext);
+ } catch (Exception e) {
+ exception = ExceptionUtils.firstOrSuppressed(e, exception);
+ }
+
+ if (exception != null) {
+ throw exception;
+ }
+ }
- if (pythonRunnerContext != null) {
- try {
- interpreter.invoke(CLOSE_FLINK_RUNNER_CONTEXT,
pythonRunnerContext);
- } finally {
- pythonRunnerContext = null;
- }
+ private void closePythonObject(String closeFunction, PyObject
pythonObject) throws Exception {
+ if (pythonObject != null) {
+ try (pythonObject) {
Review Comment:
This is the shape the whole fix turns on: logical cleanup inside, native
release on the way out. Two other Pemja handles in the same lifecycle still
have the pre-PR shape.
`Mem0LongTermMemory.close()` (`Mem0LongTermMemory.java:137-140`) calls
`adapter.callMethod(pyMem0, "close", Map.of())` and never `pyMem0.close()`,
which is the old `PythonActionExecutor.close()` exactly. It is live rather than
dead code: `RunnerContextImpl.java:339-345` calls `ltm.close()`, and the handle
comes from `PythonBridgeManager.java:237`.
`PythonResourceAdapterImpl.pythonResourceContext`
(`PythonResourceAdapterImpl.java:86,99`) is built as
`interpreter.invoke(GET_RESOURCE_CONTEXT, this)`, so it is a Java object handed
into Python, the same JNI-global-ref pattern #942 describes. That class has no
`close()` at all, and `PythonBridgeManager.close()` never touches the adapter.
I read the PR and #942 as deliberately scoped to the action executor, so I
am not suggesting you widen this one. Is a follow-up issue the plan for the
sibling handles, or is there something that already releases those two that I
have missed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
