The new implementation of Stream will write redundant frames out it's socket/serial connection if used asynchronously.
Observed on the STM32 UART.
Expected/Actual Behavior & Steps to replicate:
This pseudo-code should write N packets out the uart,
streamWriter = Stream(uart...
async def foo():
streamWriter.write(packet)
await streamWriter.drain()
for i in range(0,N):
asyncio.create_task(foo())but instead will likely write the first packet N times, the second packet N-1 times, ... the last packet one time (not in that order).
Root cause:
In the drain method of Stream, it pauses at the yield and allows the other coroutines to start, initiating N drain operations where the variable 'off' is 0 in each, and thus the entire buffer is written for each operation. Thus the drain method is not "thread safe" or idempotent in the sense that the 'off' variable is not independent of other concurrent operations; it is thus susceptible to race conditions.
Proposed Fix:
The drain operation should be idempotent, either maintaining some sort of stateful buffer reading in the Stream instance, or limiting itself to processing only one write out loop at a time.
Temporary Workaround:
Comment out the yield operation in drain (yield core._io_queue.queue_write(self.s)). This was tested and seems to be working, but wouldn't consider it safe or reliable since I'm not sure how the entire IO queuing process works and if it's safe to disable.
The new implementation of Stream will write redundant frames out it's socket/serial connection if used asynchronously.
Observed on the STM32 UART.
Expected/Actual Behavior & Steps to replicate:
This pseudo-code should write N packets out the uart,
but instead will likely write the first packet N times, the second packet N-1 times, ... the last packet one time (not in that order).
Root cause:
In the drain method of Stream, it pauses at the yield and allows the other coroutines to start, initiating N drain operations where the variable 'off' is 0 in each, and thus the entire buffer is written for each operation. Thus the drain method is not "thread safe" or idempotent in the sense that the 'off' variable is not independent of other concurrent operations; it is thus susceptible to race conditions.
Proposed Fix:
The drain operation should be idempotent, either maintaining some sort of stateful buffer reading in the Stream instance, or limiting itself to processing only one write out loop at a time.
Temporary Workaround:
Comment out the yield operation in drain (
yield core._io_queue.queue_write(self.s)). This was tested and seems to be working, but wouldn't consider it safe or reliable since I'm not sure how the entire IO queuing process works and if it's safe to disable.