Addressing Concurrency Issues in MplexStream and YamuxStream

[paschal533]

Description:

We’re working on improving the thread-safety and data integrity of MplexStream and YamuxStream in PR #639 ([link to PR]). This discussion thread is to gather feedback, clarify technical details, and explore the best solutions for handling concurrency issues in these stream multiplexers. Thanks to @seetadev for suggesting we move the detailed conversation here to benefit related initiatives.

Overview of PR #639

The PR addresses interleaving issues during concurrent reads and writes in MplexStream and YamuxStream. Here’s a summary:

What Was Wrong? (Issue #639)

• Concurrent reads and writes in MplexStream and YamuxStream could lead to data interleaving or race conditions, potentially causing data corruption.
• MplexStream lacks robust synchronization, making it prone to issues with trio.MemoryReceiveChannel.
• YamuxStream has existing synchronization (window_lock and streams_lock), but a flow control bug was identified when sending large messages (> DEFAULT_WINDOW_SIZE).

How Was It Fixed?

• MplexStream:
  • Added read_lock and write_lock attributes to ensure serialized access during reads and writes.
  • Introduced tests to verify proper serialization of concurrent operations.
• YamuxStream:
  • Initially added locks, but testing revealed they may not be necessary due to existing synchronization.
  • Fixed a flow control bug in send_window_update where recv_window wasn’t updating correctly, causing deadlocks with large messages.
  • Updated the read method to handle window updates properly and removed incorrect window updates from the write method.
  • Added a timeout in the write method to avoid lock release errors.
• Documentation:
  • Added notes on the new locks in MplexStream and their impact on thread-safety.
  • Planned updates to release notes and additional documentation.
• Testing:
  • New tests (test_yamux_read_write_lock.py) demonstrate race conditions without proper synchronization.
  • Tests now pass for large message sizes (up to 512 KiB), confirming the flow control fix.

Current Status

• The PR is in draft mode as we finalize the YamuxStream changes and evaluate whether locks are needed.
• CI/CD failures (possibly due to strict type checks post-merge) are being resolved.
• The commit history has been squashed for easier review.

Next Steps

• @kaneki003 is implementing the latest feedback from me, particularly around recv_window handling in the read method.
• We’re resolving CI/CD issues and plan to re-run tests with varying message sizes (1 KiB, 256 KiB, 512 KiB) to measure throughput.
• Benchmarking and additional stress tests are planned to validate performance.

Get Involved

Please share your thoughts, test cases, or benchmarking results If you’ve worked on similar concurrency issues in py-libp2p or other networking libraries, your insights would be invaluable. Let’s collaborate to make these multiplexers robust and performant.

Pinging @kaneki003, @seetadev, and @acul71 for their ongoing contributions. Thanks for driving this forward

[paschal533]

Hi @kaneki003, Thanks for the detailed explanation and for iterating on this! I totally get where you’re coming from with the workflow you initially envisioned, it makes sense to think of recv_window tracking the buffer’s state to prevent overflow, kind of like a TCP-style window. Your updated approach, moving window updates to read, is a big step in the right direction, and I appreciate you digging into the Yamux spec to align with it.

However, there’s a small but important misunderstanding about recv_window that we need to clear up. You mentioned decrementing recv_window in read to track the “current limit” of received data, triggering a window update when it’s low. I see why you did this, it feels intuitive to mirror the sender’s send_window reduction, but in Yamux, the receiver’s recv_window isn’t supposed to be decremented at all. Let me explain why and answer your question about where the decrement should happen.

In Yamux, The sender decrements its send_window when it sends data (which your write method does correctly with self.send_window -= to_send).
The receiver’s recv_window is just an advertised limit to tell the sender how much it can send. It’s not decremented when data is received or read. Instead, it’s increased when the receiver sends a window update (via self.recv_window += increment in send_window_update) to say it can accept more data.
The receiver’s buffer (self.conn.stream_buffers[stream_id]) tracks how much data is actually received and stored. When you read data in read, you free up buffer space, so you send a window update to let the sender know it can send more.
So, to your question, “If we don’t decrement recv_window in read or when data is received, where does it take place?” it doesn’t happen. The receiver doesn’t decrement recv_window. The decrementing happens only on the sender’s side with send_window, and the receiver just manages its buffer and sends window updates based on how much data it’s processed. Your current code’s decrement of recv_window in read makes it act like a buffer counter, but that’s not needed since the buffer itself (self.conn.stream_buffers[stream_id])` already tracks this.

The problem with decrementing recv_window in read is that it can mess up flow control. For example, if recv_window gets too low or negative, the window update (DEFAULT_WINDOW_SIZE - self.recv_window) might send the wrong increment, causing stalls or buffer issues. Plus, you’re only doing updates for specific-size reads (n > 0), but we also need them for n == -1 (read until EOF) since that clears the buffer too.

[kaneki003]

Thanks @Paschal for the detailed explanation.Regarding the n==-1(or EOF) case, i specifically left it ,since i first needed review on my implementation so that we can know the pitfalls in the code before doing the same in n==-1 case.

In your idea above,if i understand it correctly is that the stream_buffers keep track of how much data is received,and the recv_window would only increase,not decrease.However, i couldn't get the fact about how you are calculating the incrementing to be send to sender.If possible could you highlight it using workflow ,it would helpful to understand it clearly.

Also,your concern regarding recv_window will get too low,this will never happen due to this check.

if self.recv_window <= DEFAULT_WINDOW_SIZE // 2:
   logging.debug(
       f"Stream {self.stream_id}: "
       f"Low recv_window ({self.recv_window}), sending update"
   )
   await self.send_window_update(None, skip_lock=True)

This can be understood like this,suppose the recv_window,goes negative.That would mean that the sender sent more than the current recv_window.Now sender can send at max DEFAULT_WINDOW_SIZE,which would only happen when it hadn't send anything till now or the recv_window is available to it's default size.So ,at max the recv_window would be reached zero,not below that which contradicts to the fact that it's negative.

Also,taking the worst case as zero recv_window size,as soon as this happens,the above check would detect it ,and send an increment which is equal to DEFAULT_WINDOW_SIZE,thus,replenishing both send_window & recv_window

increment_value = DEFAULT_WINDOW_SIZE - self.recv_window

If possible,could you also highlight the case/example ,where wrong increment value is being calculated.

[paschal533]

Hi @kaneki003, Thanks for the detailed reply and for explaining why you held off on the n == -1 case, that’s a smart move to get feedback on the core changes first. I’m stoked that you’re diving deep into the Yamux flow control, and I totally get your reasoning about decrementing recv_window to track the buffer and prevent overflow. Your check for recv_window <= DEFAULT_WINDOW_SIZE // 2 is a good safety net, but I think there’s still a misunderstanding about recv_window that’s causing some issues. Let me clarify how the increment calculation works with a workflow, show where your current increment can go wrong, and address why recv_window shouldn’t be decremented.

On the n == -1 Case:

Cool, let’s extend the window update logic to n == -1 now that we’re refining the approach. Since this case clears the buffer, it’s a perfect spot to send a window update, just like for specific-size reads.

On recv_window and Increment Calculation:

You’re right that stream_buffers tracks the received data, and in my proposed fix, recv_window only increases (no decrementing). You asked how we calculate the increment sent to the sender, great question! The idea is to base the increment on the amount of data read (len(data)), which reflects how much buffer space was freed. This tells the sender exactly how much more it can send without overwhelming the receiver’s buffer.

Why Not Decrement recv_window?

You’re decrementing recv_window in read to track received data, but this isn’t how Yamux works. The Yamux spec says the receiver’s recv_window is an advertised limit that only increases with window updates. The sender decrements its send_window when sending data, and the receiver’s buffer (stream_buffers) tracks what’s received. Decrementing recv_window duplicates the buffer’s job and can lead to errors. Your check (recv_window <= DEFAULT_WINDOW_SIZE // 2) helps, but it doesn’t fully prevent issues.

Can recv_window Go Negative?

You said recv_window can’t go negative because the sender’s send_window is capped at DEFAULT_WINDOW_SIZE. But in concurrent scenarios, multiple reads can decrement recv_window before a window update is sent. For example:

• Peer1 sends 256 KB, send_window = 0 KB, Peer2’s buffer = 256 KB.
• Peer2 reads 100 KB, recv_window = 156 KB (256 - 100).
• Another coroutine reads 100 KB concurrently, recv_window = 56 KB.
• A third read of 100 KB makes recv_window = -44 KB (56 - 100).
• Your check catches this eventually, but recv_window can briefly go negative, causing a bad increment

Example of Wrong Increment in Your Code:

You asked where your increment (DEFAULT_WINDOW_SIZE - self.recv_window) goes wrong. Here’s an example:

• Initial: recv_window = 256 KB, send_window = 256 KB.
• Peer1 sends 100 KB, send_window = 156 KB, buffer = 100 KB.
• Peer2 reads 80 KB, recv_window = 176 KB (256 - 80), buffer = 20 KB. No update (recv_window > 128 KB).
• Peer2 reads 20 KB, recv_window = 156 KB. No update.
• Peer1 sends 50 KB, send_window = 106 KB, buffer = 50 KB.
• Peer2 reads 60 KB (all 50 KB + 10 KB new), recv_window = 96 KB (156 - 60).
• Check triggers (recv_window < 128 KB), increment = 256 - 96 = 160 KB.
• Peer2: recv_window = 256 KB, Peer1: send_window = 266 KB.
• Problem: The 160 KB increment is too large. Peer2 only freed 60 KB, but advertised 160 KB, allowing Peer1 to send up to 256 KB, which could overflow the buffer.
• Fix: Use increment = len(data) = 60 KB, so recv_window = 216 KB, send_window = 166 KB, matching the buffer’s state.

Proposed Fix:

Let’s remove the recv_window decrement and add window updates for both n == -1 and n > 0 based on len(data)