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