You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "korex-f (via GitHub)" <gi...@apache.org> on 2026/08/16 06:15:05 UTC
[PR] Fix 66158 trigger log buffer [airflow]
korex-f opened a new pull request, #71669:
URL: https://github.com/apache/airflow/pull/71669
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
-->
`make_buffered_socket_reader` re-scanned the entire accumulated buffer from offset 0 on every socket read, making newline-scanning O(N²) in the size of a single record. A 5 MiB single-line trigger log record (as reported in #66158) could cost gigabytes of cumulative prefix scanning before the terminating newline was found, delaying drainage of all supervised sockets under triggerer load.
This PR:
* Adds a persistent scan cursor (`search_from`) so each byte in the buffer is examined once per delivery, not once per chunk. A 20 MiB single-line record that previously took ~2.7s to scan now completes in well under 1s.
* Sends each completed record to the generator as an explicit immutable `bytes` copy rather than a `bytearray` slice, so the generator never holds a view that could alias the buffer across a later resize.
* Changes EOF handling: a non-newline-terminated fragment left in the buffer when the socket closes is no longer forwarded to the JSON decoder as if it were a complete record. It is dropped and logged once with bounded metadata (fragment byte length only, never the fragment content, which may contain secrets or multi-megabyte data).
No wire-protocol change. The newline-delimited framing between supervisor and subprocess is unchanged; this is scoped to the supervisor-side reader only. `make_buffered_socket_reader`'s return shape, generator contract, `data=` pre-seeding behavior, default buffer size, and selector semantics are all preserved.
A separate GCS-handler symptom mentioned later in #66158 (missing files even for ordinary-sized messages) is not addressed here and looks like a different issue, flagging in case a maintainer wants to split it out.
Note: `test_exit_by_signal`, `test_process_log_messages_from_subprocess`, `test_reinit_supervisor_comms` (missing `airflow_shared` module) and `test_remote_logging_conn_caches_connection_not_client` (missing `amazon` provider) fail identically for me on unmodified `main` in my local dev environment, pre-existing environment gaps, unrelated to this change.
closes: #66158
---
-->
---
##### 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)
<!--
--
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 66158 trigger log buffer [airflow]
Posted by "ColtenOuO (via GitHub)" <gi...@apache.org>.
ColtenOuO commented on code in PR #71669:
URL: https://github.com/apache/airflow/pull/71669#discussion_r3792065515
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
+ # bytearray while a memoryview onto it is live raises BufferError.
+ line = bytes(buffer[: newline_pos + 1])
Review Comment:
Would it be better to use `memoryview` here to save an intermediate copy?
```python
line = bytes(memoryview(buffer)[: newline_pos + 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 66158 trigger log buffer [airflow]
Posted by "ColtenOuO (via GitHub)" <gi...@apache.org>.
ColtenOuO commented on code in PR #71669:
URL: https://github.com/apache/airflow/pull/71669#discussion_r3792065515
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
+ # bytearray while a memoryview onto it is live raises BufferError.
+ line = bytes(buffer[: newline_pos + 1])
Review Comment:
Would it be better to use `memoryview` here to save an intermediate copy?
```python
line = bytes(memoryview(buffer)[: newline_pos + 1])
```
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
+ # bytearray while a memoryview onto it is live raises BufferError.
Review Comment:
I think there 4 lines comment feels a bit off.
```python
buffer = buffer[newline_pos + 1 :]
```
In Python, this operation actually creates a new object, so the original buffer isn't being resized.
--
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 slow triggerer log draining for large records [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #71669:
URL: https://github.com/apache/airflow/pull/71669#issuecomment-5943289606
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.
--
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 slow triggerer log draining for large records [airflow]
Posted by "korex-f (via GitHub)" <gi...@apache.org>.
korex-f commented on code in PR #71669:
URL: https://github.com/apache/airflow/pull/71669#discussion_r3793468090
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
+ # bytearray while a memoryview onto it is live raises BufferError.
Review Comment:
Good catch. The slice creates a new bytearray rather than resizing the original, so I removed the inaccurate explanation and replaced it with a concise comment describing the copy avoidance. I also updated the PR title to describe the user-visible problem more clearly.
--
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 slow triggerer log draining for large records [airflow]
Posted by "korex-f (via GitHub)" <gi...@apache.org>.
korex-f commented on code in PR #71669:
URL: https://github.com/apache/airflow/pull/71669#discussion_r3793470572
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
+ # bytearray while a memoryview onto it is live raises BufferError.
+ line = bytes(buffer[: newline_pos + 1])
Review Comment:
Thanks for the suggestion. I adopted the memoryview approach so conversion to immutable bytes no longer creates an intermediate bytearray copy.
--
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 66158 trigger log buffer [airflow]
Posted by "ColtenOuO (via GitHub)" <gi...@apache.org>.
ColtenOuO commented on code in PR #71669:
URL: https://github.com/apache/airflow/pull/71669#discussion_r3792037508
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
Review Comment:
```
so it must not alias the buffer we're about to resize below
```
This comment feels a bit off.
```python
buffer = buffer[newline_pos + 1 :]
```
In Python, this operation actually creates a new object, so the original buffer isn't being resized.
--
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 66158 trigger log buffer [airflow]
Posted by "ColtenOuO (via GitHub)" <gi...@apache.org>.
ColtenOuO commented on code in PR #71669:
URL: https://github.com/apache/airflow/pull/71669#discussion_r3792037508
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2291,19 +2295,34 @@ def make_buffered_socket_reader(
data: bytes = b"",
on_close: Callable[[socket], None],
buffer_size: int = 4096,
+ drop_incomplete_at_eof: bool = False,
):
buffer = bytearray(data) # This will hold our accumulated binary data
read_buffer = bytearray(buffer_size) # Temporary buffer for each read
+ # Position up to which `buffer` has already been searched for a newline. Persisting
+ # this across calls to `process()` (i.e. across selector callbacks) means a large
+ # record spread over many small reads is scanned once per byte instead of once per
+ # byte *per chunk*, which was quadratic in the record size. Reset to 0 whenever a
+ # complete record is consumed and the buffer is sliced, since the new tail is
+ # unsearched. See #66158.
+ search_from = 0
# We need to start up the generator to get it to the point it's at waiting on the yield
next(gen)
def process(buffer: bytearray) -> bytearray:
+ nonlocal search_from
# We could have read multiple lines in one go, yield them all
- while (newline_pos := buffer.find(b"\n")) != -1:
- line = buffer[: newline_pos + 1]
+ while (newline_pos := buffer.find(b"\n", search_from)) != -1:
+ # Copy the record out as immutable bytes before sending it: the generator's
+ # local may still reference this value at its next suspension point, so it
+ # must not alias the buffer we're about to resize below -- resizing a
Review Comment:
```
so it must not alias the buffer we're about to resize below
```
This comment feels a bit off.
```python
buffer = buffer[newline_pos + 1 :]
```
In Python, this operation actually creates a new object, so the original buffer isn't being resized.
--
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 slow triggerer log draining for large records [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #71669: Fix slow triggerer log draining for large records
URL: https://github.com/apache/airflow/pull/71669
--
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