You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "renat-sagut (via GitHub)" <gi...@apache.org> on 2026/04/21 18:39:09 UTC
[PR] fix frame mismatch in triggerer [airflow]
renat-sagut opened a new pull request, #65622:
URL: https://github.com/apache/airflow/pull/65622
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3170949467
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
@uranusjr OK for you then? For me sounds reasonable.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3139776333
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
i'd expect that normal deferred tasks also used TriggerCommsDecoder as they are executed inside the triggerer
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3119769461
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
Review Comment:
this can probably be deleted?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3120262158
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
Review Comment:
I think this is OK as this is a in-memory operation only (there should not be a send/receive happening when converting to bytes) to having this in critical section looks OK for me. In AsyncIo there would not be a parallel operation anyway as thread is blocked for conversion.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3139776333
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
yes, we saw frame mismatch errors even without assets. but i'd expect that normal deferred tasks also used TriggerCommsDecoder as they are handled inside the triggerer.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #65622:
URL: https://github.com/apache/airflow/pull/65622#issuecomment-4291217309
Cool! Thanks for the PR! Is there a chance to add a pytest in order to prevent a regression for the bug?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3158888013
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
Why do we need to acquire the sync lock for async operations? Shouldn’t the async lock be enough?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3143820889
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
added _lock_sync and _lock_async context managers
added tests
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #65622:
URL: https://github.com/apache/airflow/pull/65622#issuecomment-4299429558
@renat-sagut This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Unresolved review comments** (3 threads (1 from a maintainer, 2 from other reviewers)): please walk through each unresolved review thread. **Even if a suggestion looks incorrect or irrelevant** — and some of them will be, especially any comments left by automated reviewers like GitHub Copilot — it is still the author's responsibility to respond: apply the fix, reply in-thread with a brief explanation of why the suggestion does not apply, or resolve the thread if the feedback is no longer relevant. Leaving threads unaddressed for weeks blocks the PR from moving forward.
**What to do next:**
- Walk through each unresolved review thread and respond as described above.
- Make sure static checks pass locally: `prek run --from-ref main --stage pre-commit`.
- Mark the PR as "Ready for review" when you're done.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3126928130
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
re-opening, would be good to have an expert opinion. (I am not confident in this code area and fear I mis-understand the context of fix)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3119815281
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
Review Comment:
It overloads the implementation in task-sdk/src/airflow/sdk/execution_time/comms.py:261 ... I assume this has a reason and needs to be kept. But would leave the judgement to the experts.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3140247718
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
Yes something like this, should be tested though if the behaviour is kept.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3160098569
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
in #64869 it is prooposed to only use the threading.Lock instead of asyncio.Lock + threading.Lock
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3143829769
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
+ try:
+ yield
+ finally:
+ self._thread_lock.release()
+
def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""Send a request to the parent and block until the response is received."""
- frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
Review Comment:
moved frame generation inside critical section in order to protect id_counter
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3139761129
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
As far as I understood @renat-sagut we also saw frame mismatch errors the last days on "normal deferred tasks" - so I assume it is happening for all Async calls on the socker to supervisor if not properly locked. @renat-sagut correct me if I am wrong...
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3139874612
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
What does Claude say? To me the code looks good imho. It would be good to add logging errors and possibly stacktrace from logging, so it’s documented in the PR. When I did the fix reported by TP, Claude helped me write a unit test to cover for regression in the future, as always its not easy to reproduce concurrenncy issues.
Still I think the CommDecoder should have like some kind of generic (protected) send method so the locking is done at one central place, now its duplicated at 2 different classes for the same method. Or we write a decorator which handles the locking for sync/async, dunno if its possible though.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #65622:
URL: https://github.com/apache/airflow/pull/65622#issuecomment-4300311339
Quick follow-up to the triage comment above — one clarification on the "Unresolved review comments" item:
Once you believe a thread has been addressed — whether by pushing a fix, or by replying in-thread with an explanation of why the suggestion doesn't apply — **please mark the thread as resolved yourself** by clicking the "Resolve conversation" button at the bottom of each thread. Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as "still waiting on the author" and keeps the PR from moving forward. The author doing the resolve-click is the expected convention on this project.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3169534750
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
> When would a sync task be in another thread?
when using sync_to_async in KubernetesPodTrigger
https://github.com/apache/airflow/blob/3e5c2c59f1184ca7d38a030bd3045d665f5505fb/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py#L395
in Kafka AwaitMessageTrigger
https://github.com/apache/airflow/blob/3e5c2c59f1184ca7d38a030bd3045d665f5505fb/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py#L113
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3167536016
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
That’s why I proposed to generalize this solution instead of duplicating it in TriggerCommsDecoder, as this was the original solution, and which apparently solved this issue, but was not DRY.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3169534750
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
> When would a sync task be in another thread?
when using sync_to_async in KubernetesPodTrigger
https://github.com/apache/airflow/blob/3e5c2c59f1184ca7d38a030bd3045d665f5505fb/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py#L395
in Kafka AwaitMessageTrigger
https://github.com/apache/airflow/blob/3e5c2c59f1184ca7d38a030bd3045d665f5505fb/providers/apache/kafka/src/airflow/providers/apache/kafka/triggers/await_message.py#L116
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3119848575
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
original solution is also overriding the same method
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3139925676
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
Or instead of decorator a context manager (or support both). A context manager can be used in a more fine grained manner as decorator would apply to whole method. But at least we would have dedicated method which can be easily reused but also tested.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3160098569
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
in #64869 it is proposed to only use the threading.Lock instead of asyncio.Lock + threading.Lock
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3159976108
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
@dabla is this explanation correct?
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
The comms socket can be accessed at the same time by async tasks (in the async thread) and sync tasks (in other threads). For example, sync_to_async runs sync functions in a shared thread. That's why we need the sync lock as well.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3165763244
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
When would a sync task be in another thread?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3135508622
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
Those issues won't be related to the CommsDecoder but the TriggerCommsDecoder right?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3139776333
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
yes, we saw frame mismatch errors even without assets. but i'd expect that normal deferred tasks also used TriggerCommsDecoder as they are executed inside the triggerer.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3119835164
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
With this change you are masking the implementation in CommdDecoder.asend() (in task-sdk/src/airflow/sdk/execution_time/comms.py:221) - but the problem of multi-threading and locking is a generic problem.
Have you considered applying the same change in CommdDecoder base class?
(I assume this is also used for the case of async tasks in worker with the new task run interface in Airflow 3.2, if we have async problems in triggerer while sending/receiving, will it not be the same for any other async task using the same in the base class?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #65622:
URL: https://github.com/apache/airflow/pull/65622#issuecomment-4290954190
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our [Contributors' Guide](https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [prek-hooks]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-prek-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/airflow-core/docs/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3120273316
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
@ashb / @dabla Probably you are the experts for this. Might be OK but not sure if this has a reason not being consolidated with base class
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3131628890
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
We had the same issue with CommsDecoder, I also had to apply a fix lately reported by @uranusjr.
The solution brought by @renat-sagut is similar to the asend method in CommsDecoder which looks like this:
`
async def asend(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""
Send a request to the parent without blocking.
Uses async lock for coroutine safety and thread lock for socket safety.
"""
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
frame_bytes = frame.as_bytes()
async with self._async_lock:
# Acquire the threading lock without blocking the event loop
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._thread_lock.acquire)
try:
# Async write to socket
await loop.sock_sendall(self.socket, frame_bytes)
if isinstance(msg, ResendLoggingFD):
if recv_fds is None:
return None
# Blocking read in a thread
frame, fds = await asyncio.to_thread(self._read_frame, maxfds=1)
resp = self._from_frame(frame)
if TYPE_CHECKING:
assert isinstance(resp, SentFDs)
resp.fds = fds
return resp # type: ignore[return-value]
# Normal blocking read in a thread
frame = await asyncio.to_thread(self._read_frame)
return self._from_frame(frame)
finally:
self._thread_lock.release()
`
Maybe we should try to make the critical part generic in CommDecoder (dunno if possible), so we don't encounter this regression at different places.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3167415849
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
> The comms socket can be accessed at the same time by async tasks (in the async thread) and sync tasks (in other threads). For example, sync_to_async runs sync functions in a shared thread. That's why we need the sync lock as well.
Indeed that is correct.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3167536016
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
That’s why I proposed to generalize this solution instead of duplicating it in TriggerCommsDecoder, as this was the original solution, and which apparently solved this issue, bit was not DRY.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3119982218
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
Review Comment:
`frame` and `bytes` will be inside the critical section both in sync and async versions of send. does it need to be reworked?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3119982218
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
Review Comment:
frame and bytes will be inside the critical section both in sync and async versions of send. does it need to be reworked?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3131628890
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
We had the same issue with CommsDecoder, I also had to apply a fix lately reported by @uranusjr.
The solution brought by @renat-sagut is similar to the asend method in CommsDecoder which looks like this:
```
async def asend(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""
Send a request to the parent without blocking.
Uses async lock for coroutine safety and thread lock for socket safety.
"""
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
frame_bytes = frame.as_bytes()
async with self._async_lock:
# Acquire the threading lock without blocking the event loop
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._thread_lock.acquire)
try:
# Async write to socket
await loop.sock_sendall(self.socket, frame_bytes)
if isinstance(msg, ResendLoggingFD):
if recv_fds is None:
return None
# Blocking read in a thread
frame, fds = await asyncio.to_thread(self._read_frame, maxfds=1)
resp = self._from_frame(frame)
if TYPE_CHECKING:
assert isinstance(resp, SentFDs)
resp.fds = fds
return resp # type: ignore[return-value]
# Normal blocking read in a thread
frame = await asyncio.to_thread(self._read_frame)
return self._from_frame(frame)
finally:
self._thread_lock.release()
```
Maybe we should try to make the critical part generic in CommDecoder (dunno if possible), so we don't encounter this regression at different places.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3132072903
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
@uranusjr Is there a bug ticket or PR? It is hitting us in PROD and we patched this locally - Happy to contribute (me or Renat) the best fix - with guidance happy to adjust!
I also have the feeling that in the base class CommsDecoder some sync is needed. We also saw today that multiepl deferrd tasks w/o Assets were generating this same erro on Airflow 3.2.1
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3140176627
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=}, {expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
@dabla do you mean something like this?
```
@asynccontextmanager
async def _socket_protection(self):
async with self._async_lock:
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._thread_lock.acquire)
try:
yield
finally:
self._thread_lock.release()
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3167435664
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
> When would a sync task be in another thread?
When you use multi threading in a Python operator using Airflow hooks?
I introduced this in CommsDecoder together with the async PythonOperator, both lock types are needed otherwise I experienced frame mismatch errors.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham closed pull request #65622: fix frame mismatch in triggerer
URL: https://github.com/apache/airflow/pull/65622
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix frame mismatch in triggerer [airflow]
Posted by "renat-sagut (via GitHub)" <gi...@apache.org>.
renat-sagut commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3159965882
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -193,14 +194,36 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ @contextmanager
+ def _lock_sync(self):
+ """Acquire the thread lock for synchronous operations."""
+ with self._thread_lock:
+ yield
+
+ @asynccontextmanager
+ async def _lock_async(self):
+ """
+ Acquire both the async lock and the thread lock for asynchronous operations.
+
+ The thread lock is acquired via a thread executor so the event loop
+ is not blocked while waiting for other holders.
+ """
+ async with self._async_lock:
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
The socket can be accessed at the same time by async tasks (in the async thread) and sync tasks (in other threads). For example, sync_to_async runs sync functions in a shared thread. That's why we need the sync lock as well.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org