@@ -9,7 +9,6 @@ import Network.Transport
10import Control.Applicative
10import Control.Applicative
11import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
11import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
12−import Control.Concurrent.Chan
13import Control.Concurrent.MVar
12import Control.Concurrent.MVar
14import Control.Exception (SomeException, IOException, AsyncException(ThreadKilled),
13import Control.Exception (SomeException, IOException, AsyncException(ThreadKilled),
15 fromException, throwTo, throw, catch, handle)
14 fromException, throwTo, throw, catch, handle)
@@ -67,7 +66,7 @@ foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
67foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
66foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
69type ChanId = Int
68type ChanId = Int
70−type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
69+type Chans = MVar (ChanId, IntMap (MVar [ByteString], [(ThreadId, Socket)]))
72-- | This deals with several different configuration properties:
71-- | This deals with several different configuration properties:
73-- * Buffer size, specified in Hints
72-- * Buffer size, specified in Hints
@@ -101,7 +100,7 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
101 return Transport
100 return Transport
102 { newConnectionWith = {-# SCC "newConnectionWith" #-}\_ -> do
101 { newConnectionWith = {-# SCC "newConnectionWith" #-}\_ -> do
103 (chanId, chanMap) <- takeMVar chans
102 (chanId, chanMap) <- takeMVar chans
104− chan <- newChan
103+ chan <- newEmptyMVar
105 putMVar chans (chanId + 1, IntMap.insert chanId (chan, []) chanMap)
104 putMVar chans (chanId + 1, IntMap.insert chanId (chan, []) chanMap)
106 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
105 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
107 , newMulticastWith = error "newMulticastWith: not defined"
106 , newMulticastWith = error "newMulticastWith: not defined"
@@ -149,10 +148,10 @@ mkSourceEnd host service chanId = withSocketsDo $ do
149 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
148 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
152−mkTargetEnd :: Chans -> ChanId -> Chan [ByteString] -> TargetEnd
151+mkTargetEnd :: Chans -> ChanId -> MVar [ByteString] -> TargetEnd
153mkTargetEnd chans chanId chan = TargetEnd
152mkTargetEnd chans chanId chan = TargetEnd
154− { -- for now we will implement this as a Chan
153+ { -- for now we will implement this as an MVar
155− receive = {-# SCC "receive" #-} readChan chan
154+ receive = {-# SCC "receive" #-} takeMVar chan
156 , closeTargetEnd = {-# SCC "closeTargetEnd" #-} do
155 , closeTargetEnd = {-# SCC "closeTargetEnd" #-} do
157 (chanId', chanMap) <- takeMVar chans
156 (chanId', chanMap) <- takeMVar chans
158 case IntMap.lookup chanId chanMap of
157 case IntMap.lookup chanId chanMap of
@@ -195,7 +194,7 @@ procConnections chans sock = forever $ do
195-- extracted from the socket, and then written to the Chan only when
194-- extracted from the socket, and then written to the Chan only when
196-- complete. If either of the first header size is null this indicates the
195-- complete. If either of the first header size is null this indicates the
197-- socket has closed.
196-- socket has closed.
198−procMessages :: Chans -> ChanId -> Chan [ByteString] -> Socket -> IO ()
197+procMessages :: Chans -> ChanId -> MVar [ByteString] -> Socket -> IO ()
199procMessages chans chanId chan sock = do
198procMessages chans chanId chan sock = do
200 esizeBS <- recvExact sock 1
199 esizeBS <- recvExact sock 1
201 case esizeBS of
200 case esizeBS of
@@ -229,7 +228,7 @@ procMessages chans chanId chan sock = do
229 case ebs of
228 case ebs of
230 Left _ -> closeSocket
229 Left _ -> closeSocket
231 Right bs -> do
230 Right bs -> do
232− writeChan chan [bs]
231+ putMVar chan [bs]
233 procMessages chans chanId chan sock
232 procMessages chans chanId chan sock
235-- | The normal result of `recvExact sock n` is `Right ByteString`
234-- | The normal result of `recvExact sock n` is `Right ByteString`