Mux Improvements #5093

karknu · 2025-03-07T07:53:33Z

Description

This PR consist of a series of commits aimed to make Mux more efficient.
Mainly this is done by attempting to read and write larger chunks of data.

Checklist

Quality

Commit sequence makes sense and have useful messages, see ref.
New tests are added and existing tests are updated.
Self-reviewed the PR.

Maintenance

Linked an issue or added the PR to the current sprint of ouroboros-network project.
Added labels.
Updated changelog files.
The documentation has been properly updated, see ref.

Lower level benchmark for reading and writing over Sockets. Uses local TCP sockets over ephemeral ports.

Enforce at least 1 byte payload data per SDU.

Define msHeaderLength and use it instead of 8 as hardcoded headerlength.

Permit sending of multiple SDUs through a single call to sendMany for Socket bearers Bearers without vector IO support emulate it through multiple calls to write.

This implements an optional read buffer for the Socket bearer. If provided with a read buffer the socket bearer will read as much data available on the socket that can fit in the read buffer. Subsequent read requests will first be served data from the buffer.

Poll the egress queue at most 1000 times per second. This gives us an oppertunity to pack multiple messages into larger SDUs, and write those SDUs with a single sendMany call.

Use a Builder for ingress. Messages shorter than 128 bytes will be copied.

Use RecvLowWater to signal that we want to read at least the length of the SDU.

coot

I did the first pass, the PR looks nice.

coot · 2025-03-07T09:06:01Z

network-mux/bench/socket_read_write/Main.hs

+   doRead :: Int64 -> ByteChannel IO -> Int64 -> IO ()
+   doRead maxData _ cnt | cnt >= maxData = return ()
+   doRead maxData chan !cnt = do


We don't need to pass the maxData, since sndSize is in scope:

doRead :: ByteChannel IO -> Int64 -> IO () doRead _ cnt | cnt >= totalPayloadLen sndSize = return () doRead chan !cnt = do

network-mux/bench/socket_read_write/Main.hs

network-mux/network-mux.cabal

coot · 2025-03-07T09:14:06Z

network-mux/test/Test/Mux.hs

+            realLen <- choose (0, 8) -- Size of mux header is 8
+            len <- if realLen == 8 then return 0
+                                   else arbitrary


Could you explain it?

We are generating an SDU with an invalid length. With the change in this commit an SDU with a zero size payload will throw an exception. The change updates the generator to cover this case.

Could you add a comment in the code?

coot · 2025-03-07T10:45:21Z

network-mux/src/Network/Mux/Bearer/Socket.hs

-          buf <- Win32.Async.recv sd (fromIntegral l)
+              buf <- Win32.Async.recv sd (fromIntegral len)
 #else
-          buf <- Socket.recv sd l
+              buf <- (case readBuffer_m of
+                          Nothing         -> Socket.recv sd len
+                          Just readBuffer -> recvBuf readBuffer len
+                         )
 #endif
-                    `catch` Mx.handleIOException "recv errored"
-          if BL.null buf
-              then do
-                  when waitingOnNxtHeader $
-                      {- This may not be an error, but could be an orderly shutdown.
-                       - We wait 1 seconds to give the mux protocols time to perform
-                       - a clean up and exit.
-                       -}
-                      threadDelay 1
-                  throwIO $ Mx.BearerClosed (show sd ++
-                      " closed when reading data, waiting on next header " ++
-                      show waitingOnNxtHeader)
-              else do
-                  traceWith tracer $ Mx.TraceRecvEnd (fromIntegral $ BL.length buf)
-                  return buf
+                      `catch` Mx.handleIOException "recv errored"
+              if BL.null buf
+                  then do
+                      when waitingOnNxtHeader $
+                          {- This may not be an error, but could be an orderly shutdown.
+                           - We wait 1 seconds to give the mux protocols time to perform
+                           - a clean up and exit.
+                           -}
+                          threadDelay 1
+                      throwIO $ Mx.BearerClosed (show sd ++
+                          " closed when reading data, waiting on next header " ++
+                          show waitingOnNxtHeader)
+                  else return buf


if we keep the indentation in the whole where clause, then the diff will be much simpler.

coot · 2025-03-07T10:47:46Z

network-mux/test/Test/Mux.hs

@@ -406,10 +406,12 @@ prop_mux_snd_recv_bi (DummyRun messages) = ioProperty $ do
                      (-1)
                      clientTracer
                      QueueChannel { writeQueue = client_w, readQueue = client_r }
+                      Nothing


Do we have tests that cover the ReadBuffer?

Currently only prop_mux_close_io, . @crocodile-dentist is working on improving test coverage for Socket bearers and I will make sure thay they cover ReadBuffer too.

coot · 2025-03-07T10:58:07Z

network-mux/src/Network/Mux/Types.hs

@@ -174,7 +175,7 @@ data Status
 -- Mux internal types
 --

-type IngressQueue m = StrictTVar m BL.ByteString
+type IngressQueue m = StrictTVar m (Int64, Builder)


We need to be careful when writing data, (,) are lazy...

coot · 2025-03-07T10:58:39Z

network-mux/src/Network/Mux/Ingress.hs

+          let len' = len + BL.length (msBlob sdu)
+          if len' <= fromIntegral qMax
+              then do
+                let buf' = if len == 0
+                               then -- Don't copy the payload if the queue was empty
+                                 lazyByteStringInsert $ msBlob sdu
+                              else -- Copy payloads smaller than 128 bytes
+                                 buf <> (lazyByteStringThreshold 128 $ msBlob sdu)
+                writeTVar q $ (len', buf')


Let's force len' and buf'.

network-mux/src/Network/Mux/Ingress.hs

Co-authored-by: coot <coot@coot.me>

crocodile-dentist · 2025-03-07T08:52:15Z

network-mux/src/Network/Mux/Egress.hs

+           Just (TLSRDemand mpc md d) -> do
+             sdu <- processSingleWanton egressQueue sduSize mpc md d
+             buildBatch (sdu:sdus) (sdusLength + sduLength sdu)
+           Nothing -> return sdus


I would reverse the sdus here.

You or Niel gave that feedback previously, but this is just one out of 3 cases. I could either do it where I do it now or I will have to do it in 3 different places.

What about reversing when we return the value:

buildBatch sdus _ | length sdus >= maxSDUsPerBatch = return $ reverse sdus buildBatch sdus sdusLength | sdusLength >= batchSize = return $ reverse sdus buildBatch sdus !sdusLength = do demand_m <- atomically $ tryReadTBQueue egressQueue case demand_m of Just (TLSRDemand mpc md d) -> do sdu <- processSingleWanton egressQueue sduSize mpc md d buildBatch (sdu:sdus) (sdusLength + sduLength sdu) Nothing -> return sdus

karknu added 6 commits March 6, 2025 17:22

Benchmark for reading and writing over sockets

4635f0f

Lower level benchmark for reading and writing over Sockets. Uses local TCP sockets over ephemeral ports.

Enforce a min SDU size.

e048995

Enforce at least 1 byte payload data per SDU.

size limits for self connect test

a7de73e

Define msHeaderLength

c567a6d

Define msHeaderLength and use it instead of 8 as hardcoded headerlength.

add a writeMany option for batch sending of SDUs

151c39a

Permit sending of multiple SDUs through a single call to sendMany for Socket bearers Bearers without vector IO support emulate it through multiple calls to write.

Threaded benchmark for muxer and demuxer

d83d82a

karknu force-pushed the karknu/main_read_buffer branch from c094ef5 to aa52d92 Compare March 7, 2025 07:54

karknu added 5 commits March 7, 2025 09:26

poll the egress queue

e4446fb

Poll the egress queue at most 1000 times per second. This gives us an oppertunity to pack multiple messages into larger SDUs, and write those SDUs with a single sendMany call.

Benchmarks for ingress queuing

981c144

Use a ByteString Builder for ingress

3a51995

Use a Builder for ingress. Messages shorter than 128 bytes will be copied.

Require at least a full SDU from the kernel.

a274cc2

Use RecvLowWater to signal that we want to read at least the length of the SDU.

karknu force-pushed the karknu/main_read_buffer branch from aa52d92 to a274cc2 Compare March 7, 2025 08:26

Update CHANGELOG

205b8e4

karknu marked this pull request as ready for review March 7, 2025 08:40

karknu requested a review from a team as a code owner March 7, 2025 08:40

coot added the mux issues related to network-mux label Mar 7, 2025

coot reviewed Mar 7, 2025

View reviewed changes

karknu and others added 2 commits March 7, 2025 12:19

Update network-mux/src/Network/Mux/Ingress.hs

a39096e

Co-authored-by: coot <coot@coot.me>

Update network-mux/network-mux.cabal

811fef3

Co-authored-by: coot <coot@coot.me>

crocodile-dentist reviewed Mar 7, 2025

View reviewed changes

karknu added 4 commits March 7, 2025 13:12

fixup: force len and buf

837e461

fixup benchmark

df9f90e

fixup bench

3cfa20f

fixup writemany

b63018b

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Mux Improvements #5093

Mux Improvements #5093

karknu commented Mar 7, 2025

coot left a comment

coot Mar 7, 2025

karknu Mar 7, 2025

coot Mar 7, 2025

karknu Mar 7, 2025

coot Mar 7, 2025

coot Mar 7, 2025

coot Mar 7, 2025

karknu Mar 7, 2025

coot Mar 7, 2025

karknu Mar 7, 2025

coot Mar 7, 2025

karknu Mar 7, 2025

crocodile-dentist Mar 7, 2025

karknu Mar 7, 2025

coot Mar 7, 2025

Mux Improvements #5093

Are you sure you want to change the base?

Mux Improvements #5093

Conversation

karknu commented Mar 7, 2025

Description

Checklist

Quality

Maintenance

coot left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment