@@ -1,35 +1,308 @@
1−module Control.Distributed.Process ( expect ) where
1+module Control.Distributed.Process
2+ ( -- * Cloud Haskell API
3+ ProcessId
4+ , Process
5+ , expect
6+ , send
7+ , getSelfPid
8+ -- * Initialization
9+ , newLocalNode
10+ , forkProcess
11+ , runProcess
12+ ) where
213
3−import Data.ByteString.Lazy (ByteString)
4−import Data.Typeable (Typeable(typeOf))
5−import Data.Binary (decode)
6−import Control.Monad.Reader (MonadReader(..), ReaderT, liftIO)
14+import qualified Data.ByteString as BSS (ByteString, concat)
15+import qualified Data.ByteString.Lazy as BSL ( ByteString
16+ , toChunks
17+ , fromChunks
18+ , splitAt
19+ )
20+import Data.Binary (Binary, decode, encode, put, get)
21+import Data.Map (Map)
22+import qualified Data.Map as Map (empty)
23+import Data.IntMap (IntMap)
24+import qualified Data.IntMap as IntMap (empty, lookup, insert, delete)
25+import qualified Data.List as List (delete)
26+import Data.Typeable (Typeable)
27+import Control.Monad (void, liftM2)
28+import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
29+import Control.Monad.IO.Class (MonadIO, liftIO)
730import Control.Applicative ((<$>))
31+import Control.Category ((>>>))
32+import Control.Exception (throwIO)
33+import Control.Concurrent (forkIO)
34+import Control.Concurrent.MVar ( MVar
35+ , newMVar
36+ , withMVar
37+ , modifyMVar
38+ , modifyMVar_
39+ , newEmptyMVar
40+ , putMVar
41+ , takeMVar
42+ )
843import Control.Distributed.Process.Internal.CQueue ( CQueue
944 , dequeueMatching
45+ , enqueue
46+ , newCQueue
1047 )
1148import Control.Distributed.Process.Serializable ( Serializable
1249 , Fingerprint
1350 , encodeFingerprint
51+ , decodeFingerprint
1452 , fingerprint
53+ , sizeOfFingerprint
1554 )
55+import qualified Network.Transport as NT ( Transport
56+ , EndPoint
57+ , EndPointAddress
58+ , Connection
59+ , Reliability(ReliableOrdered)
60+ , defaultConnectHints
61+ , send
62+ , connect
63+ , close
64+ , newEndPoint
65+ , receive
66+ , Event(..)
67+ , address
68+ )
69+import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
70+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
71+import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
72+
73+type LocalProcessId = Int
74+
75+data ProcessId = ProcessId
76+ { processAddress :: NT.EndPointAddress
77+ , processLocalId :: LocalProcessId
78+ }
79+ deriving (Eq, Ord, Typeable, Show)
80+
81+instance Binary ProcessId where
82+ put pid = do put (processAddress pid)
83+ put (processLocalId pid)
84+ get = liftM2 ProcessId get get
85+
86+data Message = Message
87+ { messageFingerprint :: Fingerprint
88+ , messageEncoding :: BSL.ByteString
89+ }
90+
91+data LocalNode = LocalNode
92+ { localEndPoint :: NT.EndPoint
93+ , localState :: MVar LocalNodeState
94+ }
95+
96+data LocalNodeState = LocalNodeState
97+ { _localConnections :: Map ProcessId NT.Connection
98+ , _localProcesses :: IntMap ProcessState
99+ , _localNextProcId :: LocalProcessId
100+ }
16101
17−data Message = Message { messageFingerprint :: Fingerprint
18− , messageEncoding :: ByteString
19− }
102+data ProcessState = ProcessState
103+ { processQueue :: CQueue Message
104+ , processNode :: LocalNode
105+ , processId :: ProcessId
106+ }
20107
21−data ProcessState = ProcessState { processQueue :: CQueue Message }
108+newtype Process a = Process { unProcess :: ReaderT ProcessState IO a }
109+ deriving (Functor, Monad, MonadIO, MonadReader ProcessState)
22110
23−newtype Process a = Process (ReaderT ProcessState IO a)
24− deriving (Monad)
111+--------------------------------------------------------------------------------
112+-- Cloud Haskell API --
113+--------------------------------------------------------------------------------
25114
26115expect :: forall a. Serializable a => Process a
27−expect = Process $ do
116+expect = do
28117 queue <- processQueue <$> ask
29118 let fp = fingerprint (undefined :: a)
30119 msg <- liftIO $ dequeueMatching queue ((== fp) . messageFingerprint)
31120 return (decode . messageEncoding $ msg)
32121
122+send :: Serializable a => ProcessId -> a -> Process ()
123+send pid msg = do
124+ -- This requires a lookup on every send. If we want to avoid that we need to
125+ -- modify serializable to allow for stateful (IO) deserialization
126+ node <- processNode <$> ask
127+ liftIO $ do
128+ conn <- connectionTo node pid
129+ sendBinary node pid conn ( encodeFingerprint (fingerprint msg)
130+ : BSL.toChunks (encode msg)
131+ )
132+
133+getSelfPid :: Process ProcessId
134+getSelfPid = processId <$> ask
135+
136+--------------------------------------------------------------------------------
137+-- Initialization --
138+--------------------------------------------------------------------------------
139+
140+newLocalNode :: NT.Transport -> IO LocalNode
141+newLocalNode transport = do
142+ mEndPoint <- NT.newEndPoint transport
143+ case mEndPoint of
144+ Left ex -> throwIO ex
145+ Right endPoint -> do
146+ state <- newMVar LocalNodeState
147+ { _localConnections = Map.empty
148+ , _localProcesses = IntMap.empty
149+ , _localNextProcId = firstNonReservedProcessId
150+ }
151+ let node = LocalNode { localEndPoint = endPoint
152+ , localState = state
153+ }
154+ void . forkIO $ handleIncomingMessages node
155+ return node
156+
157+runProcess :: LocalNode -> Process () -> IO ()
158+runProcess node proc = do
159+ done <- newEmptyMVar
160+ void $ forkProcess node (proc >> liftIO (putMVar done ()))
161+ takeMVar done
162+
163+forkProcess :: LocalNode -> Process () -> IO ProcessId
164+forkProcess node proc = do
165+ queue <- newCQueue
166+ state <- modifyMVar (localState node) $ \st -> do
167+ let lpid = st ^. localNextProcId
168+ let pid = ProcessId { processAddress = NT.address (localEndPoint node)
169+ , processLocalId = lpid
170+ }
171+ let state = ProcessState { processQueue = queue
172+ , processNode = node
173+ , processId = pid
174+ }
175+ return ( (localProcessWithId lpid ^= Just state)
176+ . (localNextProcId ^: (+ 1))
177+ $ st
178+ , state
179+ )
180+ void . forkIO $ runReaderT (unProcess proc) state
181+ return (processId state)
182+
183+handleIncomingMessages :: LocalNode -> IO ()
184+handleIncomingMessages node = go [] IntMap.empty
185+ where
186+ go halfOpenConns openConns = do
187+ event <- NT.receive endpoint
188+ case event of
189+ NT.ConnectionOpened cid _rel _theirAddr ->
190+ go (cid : halfOpenConns) openConns
191+ NT.Received cid payload ->
192+ case IntMap.lookup cid openConns of
193+ Just proc -> do
194+ let msg = payloadToMessage payload
195+ enqueue (processQueue proc) msg
196+ Nothing -> if cid `elem` halfOpenConns
197+ then do
198+ let pid = NTI.decodeInt32 . BSS.concat $ payload
199+ mProc <- withMVar state $ return . (^. localProcessWithId pid)
200+ case mProc of
201+ Just proc ->
202+ go (List.delete pid halfOpenConns)
203+ (IntMap.insert cid proc openConns)
204+ Nothing ->
205+ fail "TODO"
206+ else
207+ fail "TODO"
208+ NT.ConnectionClosed cid ->
209+ go (List.delete cid halfOpenConns) (IntMap.delete cid openConns)
210+ NT.ErrorEvent _ ->
211+ fail "TODO"
212+ NT.EndPointClosed ->
213+ return ()
214+ NT.ReceivedMulticast _ _ ->
215+ fail "Unexpected multicast"
216+
217+ state = localState node
218+ endpoint = localEndPoint node
219+
220+--------------------------------------------------------------------------------
221+-- Auxiliary functions --
222+--------------------------------------------------------------------------------
223+
224+connectionTo :: LocalNode -> ProcessId -> IO NT.Connection
225+connectionTo node pid = do
226+ mConn <- withMVar (localState node) $ return . (^. localConnectionTo pid)
227+ case mConn of
228+ Just conn -> return conn
229+ Nothing -> createConnectionTo node pid
230+
231+createConnectionTo :: LocalNode -> ProcessId -> IO NT.Connection
232+createConnectionTo node pid = do
233+ mConn <- NT.connect (localEndPoint node)
234+ (processAddress pid)
235+ NT.ReliableOrdered
236+ NT.defaultConnectHints
237+ case mConn of
238+ Right conn -> do
239+ mConn' <- modifyMVar (localState node) $ \st ->
240+ case st ^. localConnectionTo pid of
241+ Just conn' -> return (st, Just conn')
242+ Nothing -> return ( localConnectionTo pid ^= Just conn $ st
243+ , Nothing
244+ )
245+ case mConn' of
246+ Just conn' -> do
247+ -- Somebody else already created a connection while we weren't looking
248+ -- (We don't want to keep localConnections locked while creating the
249+ -- connection because that would limit concurrency too much, and
250+ -- since Network.Transport supports lgihtweight connections creating
251+ -- an unnecessary connection now and then is cheap anyway)
252+ NT.close conn
253+ return conn'
254+ Nothing -> do
255+ sendBinary node pid conn [NTI.encodeInt32 . processLocalId $ pid]
256+ return conn
257+ Left err ->
258+ throwIO err
259+
260+sendBinary :: LocalNode
261+ -> ProcessId
262+ -> NT.Connection
263+ -> [BSS.ByteString]
264+ -> IO ()
265+sendBinary node pid conn msg = do
266+ result <- NT.send conn msg
267+ case result of
268+ Right () -> return ()
269+ Left err -> do
270+ modifyMVar_ (localState node) $
271+ return . (localConnectionTo pid ^= Nothing)
272+ throwIO err
273+
274+payloadToMessage :: [BSS.ByteString] -> Message
275+payloadToMessage payload = Message fp msg
276+ where
277+ (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)
278+ $ BSL.fromChunks payload
279+ fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
280+
281+--------------------------------------------------------------------------------
282+-- Constants --
283+--------------------------------------------------------------------------------
284+
285+firstNonReservedProcessId :: Int
286+firstNonReservedProcessId = 1024
287+
288+--------------------------------------------------------------------------------
289+-- Accessors --
290+--------------------------------------------------------------------------------
291+
292+localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)
293+localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
294+
295+localProcesses :: Accessor LocalNodeState (IntMap ProcessState)
296+localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
297+
298+localNextProcId :: Accessor LocalNodeState Int
299+localNextProcId = accessor _localNextProcId (\pid st -> st { _localNextProcId = pid })
300+
301+localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)
302+localConnectionTo pid = localConnections >>> DAC.mapMaybe pid
303+
304+localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe ProcessState)
305+localProcessWithId lpid = localProcesses >>> DAC.intMapMaybe lpid
33306
34307{-
35308{-# LANGUAGE CPP, ExplicitForAll, ScopedTypeVariables #-}