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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
