IQ.Lvbs/pycapnp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix retransmit bug for large messages causing message corruption

Browse Source 
Logic bug: We are looping over segments sent by the C++ library and sending them
over a python transport. If the last message is larger than the transport pause
threshold, this causes the transport to pause us. In that case, we forget to
increment the current write_index, causing us to retransmit the same message in
an infinite loop.

This is a serious bug, because it causes messages to become corrupted.
This commit is contained in:
Lasse Blaauwbroek
2023-10-16 10:59:34 +02:00
committed by Jacob Alexander
parent 302a96d84d
commit a30fd77a1c
1 changed files with 5 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
capnp/lib/capnp.pyx
View File

@@ -2371,6 +2371,7 @@ cdef class _AsyncIoStream:
elif self.protocol.transport is not None and hasattr(self.protocol.transport, "close"): elif self.protocol.transport is not None and hasattr(self.protocol.transport, "close"):
self.protocol.transport.close() self.protocol.transport.close()
# Call connection_lost immediately, instead of waiting for the transport to do it. # Call connection_lost immediately, instead of waiting for the transport to do it.
# TODO: This might be a questionable thing to do...
self.protocol.connection_lost("Stream is closing") self.protocol.connection_lost("Stream is closing")
async def wait_closed(self): async def wait_closed(self):
@@ -2558,15 +2559,15 @@ cdef class _PyAsyncIoStreamProtocol(DummyBaseClass, asyncio.BufferedProtocol):
self.read_reset() self.read_reset()
cdef write_loop(self): cdef write_loop(self):
if not self.write_in_progress: return if self.write_paused or not self.write_in_progress: return
cdef const ArrayPtr[const uint8_t]* piece cdef const ArrayPtr[const uint8_t]* piece
for i in range(self.write_index, self.write_pieces.size()): for i in range(self.write_index, self.write_pieces.size()):
if self.write_paused:
self.write_index = i
break
piece = &self.write_pieces[i] piece = &self.write_pieces[i]
view = memoryview.PyMemoryView_FromMemory(<char*>piece.begin(), piece.size(), buffer.PyBUF_READ) view = memoryview.PyMemoryView_FromMemory(<char*>piece.begin(), piece.size(), buffer.PyBUF_READ)
self.transport.write(view) self.transport.write(view)
if self.write_paused:
self.write_index = i+1
break
if not self.write_paused: if not self.write_paused:
self.write_fulfiller.fulfill() self.write_fulfiller.fulfill()
self.write_reset() self.write_reset()