@@ -116,27 +116,20 @@ runProcess node proc = do
116116 --TODO: should use linking for waiting for the end
117117 takeMVar waitVar
118118
119+-- | `forkProcess` forks and executes process on a given node. This
120+-- returns the ProcessId when the new process has been created.
119121forkProcess :: LocalNode -> Process () -> IO ProcessId
120122forkProcess node (Process action) = do
121−
122123 (sendAddr, chan) <- Trans.newConnection (ndTransport node)
123124 sendEnd <- Trans.connect sendAddr
124125 processTable@(ProcessTable lpid _) <- takeMVar (ndProcessTable node)
125126 let pid = ProcessId sendEnd NodeId lpid
126− forkIO $ do
127+ _ <- forkIO $ do
127128 tid <- myThreadId
128129 putMVar (ndProcessTable node) (insertProcess tid processTable)
129130 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
131+ _ <- forkIO $ receiverPump chan queue
132+ action $ ProcessState pid chan queue node
140133 return pid
141134
142135 where
@@ -145,11 +138,11 @@ forkProcess node (Process action) = do
145138 let pte = ProcessTableEntry tid
146139 in ProcessTable (nextPid+1) (IntMap.insert nextPid pte table)
147140
148− receiverPump chan queue = do
141+ receiverPump :: Trans.ReceiveEnd -> CQueue Message -> IO ()
142+ receiverPump chan queue = forever $ do
149143 msgBlobs <- Trans.receive chan
150144 let (typerep, body) = read (concatMap BS.unpack msgBlobs)
151145 enqueue queue (Message typerep body)
152− receiverPump chan queue
153146
154147send :: (Typeable a, Show a) => ProcessId -> a -> Process ()
155148send (ProcessId chan _ _) msg =