@@ -9,7 +9,6 @@ import Network.Transport
99
1010import Control.Applicative
1111import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
12−import Control.Concurrent.Chan
1312import Control.Concurrent.MVar
1413import Control.Exception (SomeException, IOException, AsyncException(ThreadKilled),
1514 fromException, throwTo, throw, catch, handle)
@@ -67,7 +66,7 @@ foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
6766foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
6867
6968type ChanId = Int
70−type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
69+type Chans = MVar (ChanId, IntMap (MVar [ByteString], [(ThreadId, Socket)]))
7170
7271-- | This deals with several different configuration properties:
7372-- * Buffer size, specified in Hints
@@ -101,7 +100,7 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
101100 return Transport
102101 { newConnectionWith = {-# SCC "newConnectionWith" #-}\_ -> do
103102 (chanId, chanMap) <- takeMVar chans
104− chan <- newChan
103+ chan <- newEmptyMVar
105104 putMVar chans (chanId + 1, IntMap.insert chanId (chan, []) chanMap)
106105 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
107106 , newMulticastWith = error "newMulticastWith: not defined"
@@ -149,10 +148,10 @@ mkSourceEnd host service chanId = withSocketsDo $ do
149148 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
150149 }
151150
152−mkTargetEnd :: Chans -> ChanId -> Chan [ByteString] -> TargetEnd
151+mkTargetEnd :: Chans -> ChanId -> MVar [ByteString] -> TargetEnd
153152mkTargetEnd chans chanId chan = TargetEnd
154− { -- for now we will implement this as a Chan
155− receive = {-# SCC "receive" #-} readChan chan
153+ { -- for now we will implement this as an MVar
154+ receive = {-# SCC "receive" #-} takeMVar chan
156155 , closeTargetEnd = {-# SCC "closeTargetEnd" #-} do
157156 (chanId', chanMap) <- takeMVar chans
158157 case IntMap.lookup chanId chanMap of
@@ -195,7 +194,7 @@ procConnections chans sock = forever $ do
195194-- extracted from the socket, and then written to the Chan only when
196195-- complete. If either of the first header size is null this indicates the
197196-- socket has closed.
198−procMessages :: Chans -> ChanId -> Chan [ByteString] -> Socket -> IO ()
197+procMessages :: Chans -> ChanId -> MVar [ByteString] -> Socket -> IO ()
199198procMessages chans chanId chan sock = do
200199 esizeBS <- recvExact sock 1
201200 case esizeBS of
@@ -229,7 +228,7 @@ procMessages chans chanId chan sock = do
229228 case ebs of
230229 Left _ -> closeSocket
231230 Right bs -> do
232− writeChan chan [bs]
231+ putMVar chan [bs]
233232 procMessages chans chanId chan sock
234233
235234-- | The normal result of `recvExact sock n` is `Right ByteString`