uranusjr commented on code in PR #75184:
URL: https://github.com/apache/airflow/pull/75184#discussion_r4397312950
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -425,23 +474,38 @@ def _on_child_started(
# Language runtimes do not send StartExecutionTimeout, so the clock
starts once they have the task.
self._arm_execution_timeout()
+ _xcom_backend: XComBackendProcess | None = attrs.field(default=None,
init=False)
+ _pending_xcom_calls: deque[Callable[[XComBackendResult], None]] =
attrs.field(factory=deque, init=False)
Review Comment:
Matching answers to requests is still positional. It works while the helper
runs one call at a time, but _queued_calls and _unanswered_calls in
XComBackendProcess must stay in sync with this deque.
Mayne make `XComBackendProcess.send_call(call, finish)` own the `(call,
finish)` pairs instead, so the supervisor doesn’t keep its own deque? Or put an
id on the XCom request/response and look the callback up by id instead (Similar
to how the supervisor does duplex).
##########
task-sdk/src/airflow/sdk/coordinators/_subprocess.py:
##########
@@ -425,23 +474,38 @@ def _on_child_started(
# Language runtimes do not send StartExecutionTimeout, so the clock
starts once they have the task.
self._arm_execution_timeout()
+ _xcom_backend: XComBackendProcess | None = attrs.field(default=None,
init=False)
+ _pending_xcom_calls: deque[Callable[[XComBackendResult], None]] =
attrs.field(factory=deque, init=False)
Review Comment:
Matching answers to requests is still positional. It works while the helper
runs one call at a time, but _queued_calls and _unanswered_calls in
XComBackendProcess must stay in sync with this deque.
Mayne make `XComBackendProcess.send_call(call, finish)` own the `(call,
finish)` pairs instead, so the supervisor doesn’t keep its own deque? Or put an
id on the XCom request/response and look the callback up by id instead (similar
to how the supervisor does duplex).
--
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]
