@@ -0,0 +1,220 @@
1+{-# LANGUAGE ExplicitForAll, ScopedTypeVariables #-}
2+module Control.Distributed.Process (
3+ -- * Processes
4+ Process,
5+ liftIO,
6+ ProcessId,
7+-- NodeId,
8+
9+ -- * Basic messaging
10+ send,
11+ expect,
12+
13+ -- * Process management
14+ spawnLocal,
15+ getSelfPid,
16+-- getSelfNode,
17+
18+ -- * Initialisation
19+ Transport,
20+ LocalNode,
21+ newLocalNode,
22+ runProcess,
23+
24+ ) where
25+
26+import qualified Network.Transport as Trans
27+import Network.Transport (Transport)
28+
29+import qualified Data.ByteString.Char8 as BS
30+import Data.ByteString.Char8 (ByteString)
31+import qualified Data.IntMap as IntMap
32+import Data.IntMap (IntMap)
33+import Data.IORef
34+import Control.Applicative
35+import Control.Monad
36+import Control.Monad.Trans
37+import Control.Concurrent.MVar
38+import Control.Concurrent.Chan
39+import Control.Concurrent
40+import Control.Exception
41+import Data.Typeable
42+
43+
44+------------------------
45+-- Cloud Haskell layer
46+--
47+
48+data NodeId = NodeId -- !Trans.SendEnd !Trans.SendEnd
49+
50+data ProcessId = ProcessId !Trans.SendEnd !NodeId !LocalProcessId
51+type LocalProcessId = Int
52+
53+newtype SendPort a = SendPort Trans.SendEnd
54+
55+newtype Process a = Process { unProcess :: ProcessState -> IO a }
56+
57+instance Functor Process where
58+ fmap f m = Process (\ps -> unProcess m ps >>= \x -> return (f x))
59+
60+instance Applicative Process where
61+ pure = return
62+ (<*>) = ap
63+
64+instance Monad Process where
65+ m >>= k = Process (\ps -> unProcess m ps >>= \x -> unProcess (k x) ps)
66+ return x = Process (\_ -> return x)
67+
68+instance MonadIO Process where
69+ liftIO io = Process (\_ -> io)
70+
71+getProcessState :: Process ProcessState
72+getProcessState = Process (\ps -> return ps)
73+
74+getLocalNode :: Process LocalNode
75+getLocalNode = prNode <$> getProcessState
76+
77+getSelfPid :: Process ProcessId
78+getSelfPid = prPid <$> getProcessState
79+
80+-- State a process carries around and has local access to
81+--
82+data ProcessState = ProcessState {
83+ prPid :: !ProcessId,
84+ prChan :: !Trans.ReceiveEnd,
85+ prQueue :: !(CQueue Message),
86+ prNode :: !LocalNode
87+ }
88+
89+data Message = Message String --string rep of TypeRep, sigh.
90+ String
91+
92+-- Context for a local node, all processes running on a node have direct access to this
93+--
94+data LocalNode = LocalNode {
95+ ndProcessTable :: !(MVar ProcessTable),
96+ ndTransport :: !Transport
97+ }
98+
99+data ProcessTable = ProcessTable !LocalProcessId !(IntMap ProcessTableEntry)
100+data ProcessTableEntry = ProcessTableEntry {
101+ pteThread :: !ThreadId
102+ }
103+
104+newLocalNode :: Transport -> IO LocalNode
105+newLocalNode trans = do
106+ processTableVar <- newMVar (ProcessTable 0 IntMap.empty)
107+ return LocalNode {
108+ ndProcessTable = processTableVar,
109+ ndTransport = trans
110+ }
111+
112+runProcess :: LocalNode -> Process () -> IO ()
113+runProcess node proc = do
114+ waitVar <- newEmptyMVar
115+ _ <- forkProcess node (proc >> liftIO (putMVar waitVar ()))
116+ --TODO: should use linking for waiting for the end
117+ takeMVar waitVar
118+
119+forkProcess :: LocalNode -> Process () -> IO ProcessId
120+forkProcess node (Process action) = do
121+
122+ (sendAddr, chan) <- Trans.newConnection (ndTransport node)
123+ sendEnd <- Trans.connectWith sendAddr Trans.SendHints
124+ processTable@(ProcessTable lpid _) <- takeMVar (ndProcessTable node)
125+ let pid = ProcessId sendEnd NodeId lpid
126+ forkIO $ do
127+ tid <- myThreadId
128+ lpid <- putMVar (ndProcessTable node) (insertProcess tid processTable)
129+ queue <- newCQueue
130+ forkIO $ receiverPump chan queue
131+
132+ let pstate = ProcessState {
133+ prPid = pid,
134+ prChan = chan,
135+ prQueue= queue,
136+ prNode = node
137+ }
138+
139+ action pstate
140+ return pid
141+
142+ where
143+ insertProcess :: ThreadId -> ProcessTable -> ProcessTable
144+ insertProcess tid (ProcessTable nextPid table) =
145+ let pte = ProcessTableEntry tid
146+ in ProcessTable (nextPid+1) (IntMap.insert nextPid pte table)
147+
148+ receiverPump chan queue = do
149+ msgBlobs <- Trans.receive chan
150+ let (typerep, body) = read (concatMap BS.unpack msgBlobs)
151+-- print ("receiverPump", typerep, body)
152+ enqueue queue (Message typerep body)
153+
154+send :: (Typeable a, Show a) => ProcessId -> a -> Process ()
155+send (ProcessId chan _ _) msg =
156+ liftIO (Trans.send chan msgBlobs)
157+ where
158+ msgBlobs :: [ByteString]
159+ msgBlobs = [BS.pack (show (show (typeOf msg), show msg))]
160+
161+expect :: forall a. (Typeable a, Read a) => Process a
162+expect = do
163+ ProcessState { prQueue = queue } <- getProcessState
164+ let typerepstr = show (typeOf (undefined :: a))
165+ Message _ body <- liftIO $
166+ dequeueMatching queue (\(Message typerepstr' _) -> typerepstr' == typerepstr)
167+ return (read body)
168+
169+spawnLocal :: Process () -> Process ProcessId
170+spawnLocal proc = do
171+ node <- getLocalNode
172+ liftIO $ forkProcess node proc
173+
174+{-
175+send :: Serializable a -> ProcessId -> a -> Process ()
176+expect :: Serializable a -> Process a
177+
178+newChan :: Serializable a => Process (SendPort a, ReceivePort a)
179+sendChan :: Serializable a => SendPort a -> a -> Process ()
180+receiveChan :: Serializable a => ReceivePort a -> Process a
181+
182+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
183+terminate :: ProcessM a
184+getSelfPid :: ProcessM ProcessId
185+getSelfNode :: ProcessM NodeId
186+
187+monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
188+
189+-}
190+
191+-- Concurrent queue for single reader, single writer
192+--
193+data CQueue a = CQueue (MVar [a]) -- arrived
194+ (Chan a) -- incomming
195+
196+newCQueue :: IO (CQueue a)
197+newCQueue = do
198+ arrived <- newMVar []
199+ incomming <- newChan
200+ return (CQueue arrived incomming)
201+
202+enqueue :: CQueue a -> a -> IO ()
203+enqueue (CQueue _arrived incomming) a = writeChan incomming a
204+
205+dequeueMatching :: forall a. CQueue a -> (a -> Bool) -> IO a
206+dequeueMatching (CQueue arrived incomming) matches = do
207+ modifyMVar arrived (checkArrived [])
208+ where
209+ checkArrived :: [a] -> [a] -> IO ([a], a)
210+ checkArrived xs' [] = checkIncomming xs'
211+ checkArrived xs' (x:xs)
212+ | matches x = return (reverse xs' ++ xs, x)
213+ | otherwise = checkArrived (x:xs') xs
214+
215+ checkIncomming :: [a] -> IO ([a], a)
216+ checkIncomming xs' = do
217+ x <- readChan incomming
218+ if matches x
219+ then return (reverse xs', x)
220+ else checkIncomming (x:xs')