wenjin272 commented on code in PR #1147:
URL: https://github.com/apache/flink-agents/pull/1147#discussion_r4096402406
##########
runtime/src/main/java/org/apache/flink/agents/runtime/operator/ActionExecutionOperator.java:
##########
@@ -236,10 +347,23 @@ public void open() throws Exception {
getContainingTask().getEnvironment().getTaskManagerInfo().getTmpDirectories(),
getRuntimeContext().getJobInfo().getJobId(),
metricGroup,
- this::checkMailboxThread,
+ parallelExecutionWithoutCoroutineEnabled
+ ? parallelExecutionLock::checkReentrant
+ : this::checkMailboxThread,
jobIdentifier,
getRuntimeContext().getUserCodeClassLoader());
+ if (parallelExecutionWithoutCoroutineEnabled) {
+ executionCoordinator =
+ new ParallelExecutionCoordinator(
+ parallelExecutionLock,
+ mailboxExecutor::execute,
+ Work::new,
+ pythonBridge::releaseCurrentThreadInterpreter,
Review Comment:
Thanks for addressing the earlier findings! The managed-worker cleanup
introduces a resource-lifetime issue.
On the parallel path, `ctx.getResource()` checks lock ownership rather than
mailbox-thread identity, so a Python resource can be lazily created on an
action worker. Its cached `PyObject` retains that worker interpreter's native
state.
During operator shutdown, the coordinator closes first, and worker cleanup
closes the interpreter. `ResourceCache.close()` then closes the cached
resource, whose `PyObject.close()` accesses the already-freed native state.
I reproduced this with the actual coordinator, resource cache, adapters, and
native Pemja: the current shutdown order crashes in `PyObject.decRef →
PyEval_AcquireThread`. Closing the cache while the worker interpreter is still
alive succeeds, even from the mailbox thread, so this is a lifetime issue
rather than cross-thread access alone.
Could we ensure cached resource handles cannot outlive their interpreter
state? Reordering shutdown alone would not cover surplus workers retiring after
the idle timeout. The current mocked tests do not exercise cached resource
cleanup after interpreter teardown.
--
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]
