Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Limit TCP writes w/Tornado to 2GB #6557

Merged
merged 10 commits into from
Jun 24, 2022
Merged

Limit TCP writes w/Tornado to 2GB #6557

Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0205fd4
Use `nbytes` attribute for `memoryview`s
jakirkham Jul 28, 2021
bf83431
Drop extra `memoryview` cast
jakirkham Jul 28, 2021
1b974d0
Always cast frames to `memoryview`s
jakirkham Jul 28, 2021
63b0c79
Write frames in chunks smaller than 2GB
jakirkham Jul 28, 2021
ae9df18
Only send non-empty chunks
jakirkham Jul 28, 2021
4bc2eb1
Check stream is still open before each send
jakirkham Jul 28, 2021
445a38b
Drop unneeded check
jakirkham Jul 28, 2021
af20253
Merge branch 'old_send' into send_lt_2gb_bufs
hhuuggoo Jun 10, 2022
c39339e
Apply suggestions from code review
jakirkham Jun 10, 2022
4950d33
Lint
jsignell Jun 22, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 20 additions & 11 deletions distributed/comm/tcp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,7 @@ async def read(self, deserializers=None):
range(0, frames_nbytes + OPENSSL_MAX_CHUNKSIZE, OPENSSL_MAX_CHUNKSIZE),
):
chunk = frames[i:j]
chunk_nbytes = len(chunk)
chunk_nbytes = chunk.nbytes
n = await stream.read_into(chunk)
assert n == chunk_nbytes, (n, chunk_nbytes)
except StreamClosedError as e:
Expand Down Expand Up @@ -299,16 +299,25 @@ async def write(self, msg, serializers=None, on_error="message"):
# trick to enque all frames for writing beforehand
for each_frame_nbytes, each_frame in zip(frames_nbytes, frames):
if each_frame_nbytes:
if stream._write_buffer is None:
raise StreamClosedError()

if isinstance(each_frame, memoryview):
# Make sure that `len(data) == data.nbytes`
# See <https://github.com/tornadoweb/tornado/pull/2996>
each_frame = ensure_memoryview(each_frame)

stream._write_buffer.append(each_frame)
stream._total_write_index += each_frame_nbytes
# Make sure that `len(data) == data.nbytes`
# See <https://github.com/tornadoweb/tornado/pull/2996>
each_frame = ensure_memoryview(each_frame)
for i, j in sliding_window(
2,
range(
0,
each_frame_nbytes + OPENSSL_MAX_CHUNKSIZE,
OPENSSL_MAX_CHUNKSIZE,
),
):
chunk = each_frame[i:j]
chunk_nbytes = chunk.nbytes

if stream._write_buffer is None:
raise StreamClosedError()

stream._write_buffer.append(chunk)
stream._total_write_index += chunk_nbytes

# start writing frames
stream.write(b"")
Expand Down