@@ -16,7 +16,6 @@ module Control.Distributed.Process
1616 , send
1717 , getSelfPid
1818 , RemoteTable
19− , initRemoteTable
2019 -- * Matching messages
2120 , Match
2221 , match
@@ -35,99 +34,61 @@ module Control.Distributed.Process
3534 , DiedReason(..)
3635 , DidUnmonitor(..)
3736 , DidUnlink(..)
38− -- * Initialization
39− , newLocalNode
40− , forkProcess
41− , runProcess
37+ -- * Closures
38+ , Closure
39+ , Static
40+ , unClosure
4241 -- * Auxiliary API
43− , closeLocalNode
4442 , catch
4543 , expectTimeout
4644 , spawnAsync
4745 ) where
4846
4947import Prelude hiding (catch)
50−import System.IO (fixIO)
51−import qualified Data.ByteString as BSS (ByteString)
52−import qualified Data.ByteString.Lazy as BSL (fromChunks)
5348import Data.Binary (decode)
54−import Data.Map (Map)
55−import qualified Data.Map as Map (empty, lookup, insert, delete)
56−import qualified Data.List as List (delete)
57−import Data.Set (Set)
58−import qualified Data.Set as Set (empty, insert, delete, member)
59−import Data.Foldable (forM_)
60−import Control.Monad (void)
61−import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
49+import Data.Typeable (Typeable)
50+import Data.Dynamic (fromDyn)
51+import qualified Data.Map as Map ((!))
52+import Control.Monad.Reader (ask)
6253import Control.Monad.IO.Class (MonadIO, liftIO)
54+import qualified Control.Monad.Trans.Class as Trans (lift)
6355import Control.Applicative ((<$>))
64−import Control.Exception (Exception, throwIO, SomeException)
56+import Control.Exception (Exception, throw)
6557import qualified Control.Exception as Exception (catch)
66−import Control.Concurrent (forkIO)
67−import Control.Concurrent.MVar ( newMVar
68− , withMVar
69− , modifyMVar
70− , modifyMVar_
71− , newEmptyMVar
72− , putMVar
73− , takeMVar
74− )
75−import Control.Concurrent.Chan (newChan, writeChan)
76−import Control.Distributed.Process.Internal.CQueue (enqueue, dequeue, newCQueue)
58+import Control.Concurrent.MVar (modifyMVar)
59+import Control.Concurrent.Chan (writeChan)
60+import Control.Distributed.Process.Internal.CQueue (dequeue)
7761import Control.Distributed.Process.Serializable (Serializable, fingerprint)
78−import qualified Network.Transport as NT ( Transport
79− , EndPoint
80− , Connection
81− , Reliability(ReliableOrdered)
82− , defaultConnectHints
83− , send
84− , connect
85− , close
86− , newEndPoint
87− , receive
88− , Event(..)
89− , EventErrorCode(..)
90− , TransportError(..)
91− , address
92− , closeEndPoint
93− , ConnectionId
94− )
95−import Data.Accessor ((^.), (^=), (^:))
96−import System.Random (randomIO)
97−import Control.Distributed.Process.Internal.NodeController (runNodeController)
98−import Control.Distributed.Process.Internal ( RemoteTable
99− , initRemoteTable
100− , NodeId(..)
101− , LocalProcessId(..)
102− , ProcessId(..)
103− , LocalNode(..)
104− , LocalNodeState(..)
105− , LocalProcess(..)
106− , LocalProcessState(..)
107− , Process(..)
108− , Closure
109− , Message(..)
110− , MonitorRef(..)
111− , MonitorNotification(..)
112− , LinkException(..)
113− , DidUnmonitor(..)
114− , DidUnlink(..)
115− , DiedReason(..)
116− , SpawnRef(..)
117− , NCMsg(..)
118− , ProcessSignal(..)
119− , createMessage
120− , messageToPayload
121− , payloadToMessage
122− , payloadToId
123− , idToPayload
124− , localPidCounter
125− , localPidUnique
126− , localProcessWithId
127− , monitorCounter
128− , spawnCounter
129− , connectionTo
130− )
62+import Data.Accessor ((^.), (^:))
63+import Control.Distributed.Process.Internal.Types
64+ ( RemoteTable
65+ , NodeId(..)
66+ , ProcessId(..)
67+ , LocalNode(..)
68+ , LocalProcess(..)
69+ , Process(..)
70+ , runLocalProcess
71+ , Closure(..)
72+ , Static(..)
73+ , Message(..)
74+ , MonitorRef(..)
75+ , MonitorNotification(..)
76+ , LinkException(..)
77+ , DidUnmonitor(..)
78+ , DidUnlink(..)
79+ , DiedReason(..)
80+ , SpawnRef(..)
81+ , NCMsg(..)
82+ , ProcessSignal(..)
83+ , monitorCounter
84+ , spawnCounter
85+ , MessageT
86+ , Closure(..)
87+ )
88+import Control.Distributed.Process.Internal.MessageT
89+ ( sendMessage
90+ , getLocalNode
91+ )
13192
13293-- INTERNAL NOTES
13394--
@@ -198,11 +159,9 @@ expect = receiveWait [match return]
198159
199160-- | Send a message
200161send :: Serializable a => ProcessId -> a -> Process ()
201−send them msg = do
202− -- This requires a lookup on every send. If we want to avoid that we need to
203− -- modify serializable to allow for stateful (IO) deserialization
204− mConn <- getConnectionTo them
205− forM_ mConn $ \conn -> sendMessage them conn (createMessage msg)
162+-- This requires a lookup on every send. If we want to avoid that we need to
163+-- modify serializable to allow for stateful (IO) deserialization
164+send them msg = lift $ sendMessage (Left them) msg
206165
207166-- | Our own process ID
208167getSelfPid :: Process ProcessId
@@ -268,20 +227,26 @@ receiveTimeout t ms = do
268227 Just proc -> Just <$> proc
269228
270229--------------------------------------------------------------------------------
271−-- Auxiliary API --
230+-- Closures --
272231--------------------------------------------------------------------------------
273232
274−-- | Force-close a local node
275−--
276−-- TODO: for now we just close the associated endpoint
277−closeLocalNode :: LocalNode -> IO ()
278−closeLocalNode node =
279− NT.closeEndPoint (localEndPoint node)
233+-- | Deserialize a closure
234+unClosure :: Typeable a => Closure a -> Process a
235+unClosure (Closure (Static label) env) = do
236+ dec <- lookupStatic label
237+ return (dec env)
238+
239+--------------------------------------------------------------------------------
240+-- Auxiliary API --
241+--------------------------------------------------------------------------------
280242
281243-- | Catch exceptions within a process
282244catch :: Exception e => Process a -> (e -> Process a) -> Process a
283245catch p h = do
284− run <- runLocalProcess <$> ask
246+ node <- lift getLocalNode
247+ lproc <- ask
248+ let run :: Process a -> IO a
249+ run proc = runLocalProcess node proc lproc
285250 liftIO $ Exception.catch (run p) (run . h)
286251
287252-- | Like 'expect' but with a timeout
@@ -297,210 +262,10 @@ spawnAsync _nid proc = do
297262 postCtrlMsg (Spawn proc spawnRef)
298263 return spawnRef
299264
300−--------------------------------------------------------------------------------
301−-- Initialization --
302−--------------------------------------------------------------------------------
303−
304−-- | Initialize a new local node.
305−--
306−-- Note that proper Cloud Haskell initialization and configuration is still
307−-- to do.
308−newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
309−newLocalNode transport rtable = do
310− mEndPoint <- NT.newEndPoint transport
311− case mEndPoint of
312− Left ex -> throwIO ex
313− Right endPoint -> do
314− unq <- randomIO
315− state <- newMVar LocalNodeState
316− { _localProcesses = Map.empty
317− , _localPidCounter = 0
318− , _localPidUnique = unq
319− }
320− ctrlChan <- newChan
321− let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
322− , localEndPoint = endPoint
323− , localState = state
324− , localCtrlChan = ctrlChan
325− , remoteTable = rtable
326− }
327− void . forkIO $ runNodeController node
328− void . forkIO $ handleIncomingMessages node
329− return node
330−
331−-- | Run a process on a local node and wait for it to finish
332−runProcess :: LocalNode -> Process () -> IO ()
333−runProcess node proc = do
334− done <- newEmptyMVar
335− void $ forkProcess node (proc >> liftIO (putMVar done ()))
336− takeMVar done
337−
338−-- | Spawn a new process on a local node
339−forkProcess :: LocalNode -> Process () -> IO ProcessId
340−forkProcess node proc = modifyMVar (localState node) $ \st -> do
341− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
342− , lpidUnique = st ^. localPidUnique
343− }
344− let pid = ProcessId { processNodeId = localNodeId node
345− , processLocalId = lpid
346− }
347− pst <- newMVar LocalProcessState { _connections = Map.empty
348− , _monitorCounter = 0
349− , _spawnCounter = 0
350− }
351− queue <- newCQueue
352− (_, lproc) <- fixIO $ \ ~(tid, _) -> do
353− let lproc = LocalProcess { processQueue = queue
354− , processNode = node
355− , processId = pid
356− , processState = pst
357− , processThread = tid
358− }
359− tid' <- forkIO $ do
360− reason <- Exception.catch
361− (runLocalProcess lproc proc >> return DiedNormal)
362− (return . DiedException . (show :: SomeException -> String))
363− -- [Unified: Table 4, rules termination and exiting]
364− modifyMVar_ (localState node) $
365− return . (localProcessWithId lpid ^= Nothing)
366− writeChan (localCtrlChan node) NCMsg
367− { ctrlMsgSender = Left pid
368− , ctrlMsgSignal = Died (Left pid) reason
369− }
370− return (tid', lproc)
371−
372− -- TODO: if the counter overflows we should pick a new unique
373− return ( (localProcessWithId lpid ^= Just lproc)
374− . (localPidCounter ^: (+ 1))
375− $ st
376− , pid
377− )
378−
379−handleIncomingMessages :: LocalNode -> IO ()
380−handleIncomingMessages node = go [] Map.empty Set.empty
381− where
382− go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know
383− -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes
384− -> Set NT.ConnectionId -- ^ Connections to our controller
385− -> IO ()
386− go uninitConns procConns ctrlConns = do
387− event <- NT.receive endpoint
388− case event of
389− NT.ConnectionOpened cid _rel _theirAddr ->
390− go (cid : uninitConns) procConns ctrlConns
391− NT.Received cid payload ->
392− case Map.lookup cid procConns of
393− Just proc -> do
394− let msg = payloadToMessage payload
395− enqueue (processQueue proc) msg
396− go uninitConns procConns ctrlConns
397− Nothing | cid `Set.member` ctrlConns -> do
398− writeChan ctrlChan (decode . BSL.fromChunks $ payload)
399− go uninitConns procConns ctrlConns
400− Nothing | cid `elem` uninitConns ->
401− case payloadToId payload of
402− Just lpid -> do
403− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
404− case mProc of
405− Just proc ->
406− go (List.delete cid uninitConns)
407− (Map.insert cid proc procConns)
408− ctrlConns
409− Nothing ->
410− fail "handleIncomingMessages: TODO 1"
411− Nothing ->
412− go (List.delete cid uninitConns)
413− procConns
414− (Set.insert cid ctrlConns)
415− _ ->
416− fail "handleIncomingMessages: TODO 2"
417− NT.ConnectionClosed cid ->
418− go (List.delete cid uninitConns)
419− (Map.delete cid procConns)
420− (Set.delete cid ctrlConns)
421− NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do
422− -- [Unified table 9, rule node_disconnect]
423− let nid = Right $ NodeId theirAddr
424− writeChan ctrlChan NCMsg
425− { ctrlMsgSender = nid
426− , ctrlMsgSignal = Died nid DiedDisconnect
427− }
428− NT.ErrorEvent _ ->
429− fail "handleIncomingMessages: TODO 4"
430− NT.EndPointClosed ->
431− return ()
432− NT.ReceivedMulticast _ _ ->
433− fail "Unexpected multicast"
434−
435− state = localState node
436− endpoint = localEndPoint node
437− ctrlChan = localCtrlChan node
438−
439265--------------------------------------------------------------------------------
440266-- Auxiliary functions --
441267--------------------------------------------------------------------------------
442268
443−getConnectionTo :: ProcessId -> Process (Maybe NT.Connection)
444−getConnectionTo them = do
445− ourState <- processState <$> ask
446− mConn <- liftIO $ withMVar ourState $ return . (^. connectionTo them)
447− case mConn of
448− Just conn -> return . Just $ conn
449− Nothing -> createConnectionTo them
450−
451−createConnectionTo :: ProcessId -> Process (Maybe NT.Connection)
452−createConnectionTo them = do
453− proc <- ask
454− mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)
455− (nodeAddress . processNodeId $ them)
456− NT.ReliableOrdered
457− NT.defaultConnectHints
458− case mConn of
459− Right conn -> do
460− mConn' <- liftIO $ modifyMVar (processState proc) $ \st ->
461− case st ^. connectionTo them of
462− Just conn' -> return (st, Just conn')
463− Nothing -> return (connectionTo them ^= Just conn $ st, Nothing)
464− case mConn' of
465− Just conn' -> do
466− -- Somebody else already created a connection while we weren't looking
467− -- (We don't want to keep localConnections locked while creating the
468− -- connection because that would limit concurrency too much, and
469− -- since Network.Transport supports lgihtweight connections creating
470− -- an unnecessary connection now and then is cheap anyway)
471− liftIO $ NT.close conn
472− return . Just $ conn'
473− Nothing -> do
474− sendBinary them conn . idToPayload . Just . processLocalId $ them
475− return . Just $ conn
476− Left _err -> do
477− -- TODO: should probably pass this error to remoteProcessFailed
478− node <- processNode <$> ask
479− liftIO $ remoteProcessFailed node them
480− return Nothing
481−
482−sendBinary :: ProcessId -> NT.Connection -> [BSS.ByteString] -> Process ()
483−sendBinary them conn payload = do
484− result <- liftIO $ NT.send conn payload
485− case result of
486− Right () ->
487− return ()
488− Left _ -> do
489− node <- processNode <$> ask
490− liftIO $ remoteProcessFailed node them
491−
492−sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
493−sendMessage them conn = sendBinary them conn . messageToPayload
494−
495−remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
496−remoteProcessFailed node them = do
497− -- [Unified: Table 9 rule node_disconnect]
498− let nid = Right (processNodeId them)
499− writeChan (localCtrlChan node) NCMsg
500− { ctrlMsgSender = nid
501− , ctrlMsgSignal = Died nid DiedDisconnect
502− }
503−
504269getMonitorRefFor :: ProcessId -> Process MonitorRef
505270getMonitorRefFor pid = do
506271 proc <- ask
@@ -519,15 +284,21 @@ getSpawnRef = do
519284 , SpawnRef counter
520285 )
521286
522−-- This is most definitely NOT exported
523−runLocalProcess :: LocalProcess -> Process a -> IO a
524−runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
525−
526287-- | Post a control message on the local node controller
527288postCtrlMsg :: ProcessSignal -> Process ()
528289postCtrlMsg signal = do
529290 us <- getSelfPid
530− ctrlChan <- localCtrlChan . processNode <$> ask
291+ ctrlChan <- localCtrlChan <$> lift getLocalNode
531292 liftIO $ writeChan ctrlChan NCMsg { ctrlMsgSender = Left us
532293 , ctrlMsgSignal = signal
533294 }
295+
296+lookupStatic :: Typeable a => String -> Process a
297+lookupStatic label = do
298+ rtable <- remoteTable <$> lift getLocalNode
299+ return $ fromDyn (rtable Map.! label) (throw typeError)
300+ where
301+ typeError = userError "lookupStatic type error"
302+
303+lift :: MessageT IO a -> Process a
304+lift = Process . Trans.lift