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

Send buffers in chunks <2GB #5134

Closed
wants to merge 7 commits into from
+17 −9
Closed

Send buffers in chunks <2GB #5134

Changes from all commits
Commits
Show all changes
7 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
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
26 changes: 17 additions & 9 deletions distributed/comm/tcp.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,7 +202,7 @@ async def read(self, deserializers=None):
2, range(0, frames_nbytes + C_INT_MAX, C_INT_MAX)
):
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 @@ -268,16 +268,24 @@ 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()
each_frame = memoryview(each_frame)

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

stream._write_buffer.append(each_frame)
stream._total_write_index += each_frame_nbytes
# Workaround for OpenSSL 1.0.2 (can drop with OpenSSL 1.1.1)
for i, j in sliding_window(
2, range(0, each_frame_nbytes + C_INT_MAX, C_INT_MAX)
):
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