class InputMergeActor(AbstractActor):
def __init__(self, parent=None, name="Input merger", **kw):
...
self._lock = threading.Lock() # <-- plain, non-reentrant
def _execute(self, inData, _scope_id=None, source=None) -> None:
with self._lock: # <-- acquired here
self.setStarted()
try:
self._cache_inputs(source, inData)
finally:
self.setFinished()
if not self._has_all_required_triggers():
return
self._propagate_cached_inputs() # <-- still holding the lock
# -> _trigger_downstream()
# -> actor.trigger(...)File ".../concurrent/futures/_base.py", line 550, in set_result
self._invoke_callbacks()
File ".../concurrent/futures/_base.py", line 340, in _invoke_callbacks
callback(self)
File ".../pypushflow/concurrent/thread.py", line 55, in cb
callback(result)
File ".../pypushflow/concurrent/scaling.py", line 106, in _callback
return callback(return_value)
File ".../pypushflow/ThreadCountingActor.py", line 126, in callback_wrapper
return cb(*args, **kw)
File ".../pypushflow/PythonActor.py", line 106, in resultHandler
self._triggerDownStreamActors(downstreamData)
...
File ".../ewoksppf/bindings.py", line 265, in _execute
with self._lock:
