git.iko.soy/distributed-process
on master help sign in
Make sure that concurrent sends don't interleave
b44da5aEdsko de Vries
network-transport-tcp/src/Network/Transport/TCP.hs+10−3
compare with master
unifiedsplit
@@ -372,7 +372,7 @@ data ValidRemoteEndPointState = ValidRemoteEndPointState
372 { _remoteOutgoing :: !Int
372 { _remoteOutgoing :: !Int
373 , _remoteIncoming :: !IntSet
373 , _remoteIncoming :: !IntSet
374 , remoteSocket :: !N.Socket
374 , remoteSocket :: !N.Socket
375− , sendOn :: [ByteString] -> IO ()
375+ , remoteSendLock :: !(MVar ())
376 , _pendingCtrlRequests :: !(IntMap (MVar (Either IOException [ByteString])))
376 , _pendingCtrlRequests :: !(IntMap (MVar (Either IOException [ByteString])))
377 , _nextCtrlRequestId :: !ControlRequestId
377 , _nextCtrlRequestId :: !ControlRequestId
378 }
378 }
@@ -727,11 +727,12 @@ handleConnectionRequest transport sock = handle handleException $ do
727 tryCloseSocket sock
727 tryCloseSocket sock
728 return Nothing
728 return Nothing
729 else do
729 else do
730+ sendLock <- newMVar ()
730 let vst = ValidRemoteEndPointState
731 let vst = ValidRemoteEndPointState
731 { remoteSocket = sock
732 { remoteSocket = sock
733+ , remoteSendLock = sendLock
732 , _remoteOutgoing = 0
734 , _remoteOutgoing = 0
733 , _remoteIncoming = IntSet.empty
735 , _remoteIncoming = IntSet.empty
734− , sendOn = sendMany sock
735 , _pendingCtrlRequests = IntMap.empty
736 , _pendingCtrlRequests = IntMap.empty
736 , _nextCtrlRequestId = 0
737 , _nextCtrlRequestId = 0
737 }
738 }
@@ -1039,11 +1040,12 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1039 (connectTimeout hints)
1040 (connectTimeout hints)
1040 didAccept <- case result of
1041 didAccept <- case result of
1041 Right (sock, ConnectionRequestAccepted) -> do
1042 Right (sock, ConnectionRequestAccepted) -> do
1043+ sendLock <- newMVar ()
1042 let vst = ValidRemoteEndPointState
1044 let vst = ValidRemoteEndPointState
1043 { remoteSocket = sock
1045 { remoteSocket = sock
1046+ , remoteSendLock = sendLock
1044 , _remoteOutgoing = 0
1047 , _remoteOutgoing = 0
1045 , _remoteIncoming = IntSet.empty
1048 , _remoteIncoming = IntSet.empty
1046− , sendOn = sendMany sock
1047 , _pendingCtrlRequests = IntMap.empty
1049 , _pendingCtrlRequests = IntMap.empty
1048 , _nextCtrlRequestId = 0
1050 , _nextCtrlRequestId = 0
1049 }
1051 }
@@ -1386,6 +1388,11 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
1386 ourState = localState ourEndPoint
1388 ourState = localState ourEndPoint
1387 ourAddress = localAddress ourEndPoint
1389 ourAddress = localAddress ourEndPoint
1388
1390
1391+-- | Send a payload over a heavyweight connection (thread safe)
1392+sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO ()
1393+sendOn vst bs = withMVar (remoteSendLock vst) $ \() ->
1394+ sendMany (remoteSocket vst) bs
1395+
1389--------------------------------------------------------------------------------
1396--------------------------------------------------------------------------------
1390-- Scheduling actions --
1397-- Scheduling actions --
1391--------------------------------------------------------------------------------
1398--------------------------------------------------------------------------------