@@ -1,256 +0,0 @@
1−{-# LANGUAGE ConstraintKinds #-}
2−{-# LANGUAGE ExistentialQuantification #-}
3−{-# LANGUAGE FlexibleContexts #-}
4−{-# LANGUAGE RecordWildCards #-}
5−
6−module Asyncly.Context
7− ( ChildEvent(..)
8− , Loggable
9− , Log (..)
10− , CtxLog (..)
11− , LogEntry (..)
12− , Context(..)
13− , initContext
14− , saveContext
15− , restoreContext
16− , composeContext
17− , setContextMailBox
18− , takeContextMailBox
19− , Location(..)
20− , getLocation
21− , setLocation
22− )
23−where
24−
25−import Control.Concurrent (ThreadId)
26−import Control.Concurrent.STM (TChan)
27−import Control.Exception (SomeException)
28−import Control.Monad.State (StateT, MonadPlus(..))
29−import qualified Control.Monad.Trans.State.Lazy as Lazy (get, gets, modify, put)
30−import Data.Dynamic (Typeable)
31−import Data.IORef (IORef)
32−import Unsafe.Coerce (unsafeCoerce)
33−import GHC.Prim (Any)
34−
35−------------------------------------------------------------------------------
36−-- Log types
37−------------------------------------------------------------------------------
38−
39−-- Logging is done using the 'logged' combinator. It remembers the last value
40−-- returned by a given computation and replays it the next time the computation
41−-- is resumed. However this could be problematic if we do not annotate all
42−-- impure computations. The program can take a different path due to a
43−-- non-logged computation returning a different value. In that case we may
44−-- replay a wrong value. To detect this we can use a unique id for each logging
45−-- site and abort if the id does not match on replay.
46−
47−-- | Constraint type synonym for a value that can be logged.
48−type Loggable a = (Show a, Read a)
49−
50−-- XXX remove the Maybe as we never log Nothing
51−data LogEntry =
52− Executing -- we are inside this computation, not yet done
53− | Result (Maybe String) -- computation done, we have the result to replay
54− deriving (Read, Show)
55−
56−data Log = Log [LogEntry] deriving Show
57−
58−-- log entries and replay entries
59−data CtxLog = CtxLog [LogEntry] [LogEntry] deriving (Read, Show)
60−
61−------------------------------------------------------------------------------
62−-- Parent child thread communication types
63−------------------------------------------------------------------------------
64−
65−data ChildEvent a = ChildDone ThreadId (Maybe SomeException)
66−
67−------------------------------------------------------------------------------
68−-- State threaded around the monad
69−------------------------------------------------------------------------------
70−
71−-- | Describes the context of a computation.
72−data Context = Context
73− {
74− ---------------------------------------------------------------------------
75− -- Execution state
76− ---------------------------------------------------------------------------
77−
78− mailBox :: Maybe Any -- untyped,
79− -- ^ mailBox to send event data to the child thread
80−
81− , currentm :: Any -- untyped, the real type is: m a
82− -- ^ The 'm' in an 'm >>= f' operation of the monad. This tells us where we
83− -- are in the monadic computation at this point.
84−
85− , fstack :: [Any] -- untyped, the real type is: [a -> m b]
86− -- ^ The head of the list stores the 'f' corresponding to 'currentm' in an
87− -- 'm >>= f' operation of the monad. This is stack because when a
88− -- computation has nested binds we need to store the next 'f' for each
89− -- layer of nesting.
90−
91− -- Logging is done by the 'logged' primitive in this journal only if logsRef
92− -- exists.
93− , journal :: CtxLog
94−
95− -- When we suspend we save the logs in this IORef and exit.
96− , logsRef :: Maybe (IORef [Log])
97− , location :: Location
98−
99− ---------------------------------------------------------------------------
100− -- User state
101− ---------------------------------------------------------------------------
102−
103− -- , mfData :: M.Map TypeRep Any -- untyped, type coerced
104− -- ^ State data accessed with get or put operations
105−
106− ---------------------------------------------------------------------------
107− -- Thread creation, communication and cleanup
108− ---------------------------------------------------------------------------
109− -- When a new thread is created the parent records it in the
110− -- 'pendingThreads' field and 'threadCredit' is decremented. When a child
111− -- thread is done it sends a done event to the parent on an unbounded
112− -- channel and goes away. Before starting a new thread the parent always
113− -- processes the unbounded channel to clear up the pending zombies. This
114− -- strategy ensures that the zombie queue will never grow more than the
115− -- number of running threads. Before exiting, the parent thread waits on
116− -- the channel until all its children are cleaned up.
117− ---------------------------------------------------------------------------
118−
119− -- XXX setup a cleanup computation to run rather than passing all these
120− -- params.
121−
122− , childChannel :: TChan (ChildEvent Any)
123− -- ^ A channel for the immediate children to communicate to us when they
124− -- die. Each thread has its own dedicated channel for its children
125−
126− -- We always track the child threads, otherwise the programmer needs to
127− -- worry about if some threads may remain hanging because they are stuck in
128− -- an infinite loop or a forever blocked IO. Also, it is not clear whether
129− -- there is a significant gain by disabling tracking. Instead of using
130− -- threadIds we can also keep a count instead, that will allow us to wait
131− -- for all children to drain but we cannot kill them.
132− --
133− -- We need these as IORefs since we need any changes to these to be
134− -- reflected back in the calling code that waits for the children.
135−
136− , pendingThreads :: IORef [ThreadId]
137− -- ^ Active immediate child threads spawned by this thread, modified only
138− -- by this thread, therefore atomic modification is not needed.
139−
140− -- By default this limit is shared by the current tree of threads. But at
141− -- any point the 'threads' primitive can be used to start a new limit for
142− -- the computation under that primitive.
143− --
144− -- Shared across many threads, therefore atomic access is required.
145− -- This is incremented by the child thread itelf before it exits.
146− --
147− , threadCredit :: IORef Int
148− -- ^ How many more threads are allowed to be created?
149− } deriving Typeable
150−
151−initContext
152− :: m a
153− -> TChan (ChildEvent a)
154− -> IORef [ThreadId]
155− -> IORef Int
156− -> (b -> m a)
157− -> Maybe (IORef [Log])
158− -> Context
159−initContext x childChan pending credit finalizer lref =
160− Context { mailBox = Nothing
161− , currentm = unsafeCoerce x
162− , fstack = [unsafeCoerce finalizer]
163− , journal = CtxLog [] []
164− , logsRef = lref
165− , location = Worker
166− -- , mfData = mempty
167− , childChannel = unsafeCoerce childChan
168− , pendingThreads = pending
169− , threadCredit = credit }
170−
171−------------------------------------------------------------------------------
172−-- Where is the computation running?
173−------------------------------------------------------------------------------
174−
175−-- RemoteNode is used in the Alternative instance. We stop the computation when
176−-- the location is RemoteNode, we do not execute any alternatives.
177−--
178−-- WaitingParent is used in logging. When a parent forks a child and waits for
179−-- it, it is marked as WaitingParent and we log a "Wait" entry in the log.
180−--
181−data Location = Worker | WaitingParent | RemoteNode
182− deriving (Typeable, Eq, Show)
183−
184−getLocation :: Monad m => StateT Context m Location
185−getLocation = Lazy.gets location
186−
187−setLocation :: Monad m => Location -> StateT Context m ()
188−setLocation loc = Lazy.modify $ \Context { .. } -> Context { location = loc, .. }
189−
190−------------------------------------------------------------------------------
191−-- Saving and restoring the context of a computation. We perform the monadic
192−-- computation one step at a time storing the context of where we are in the
193−-- computation. At any point the remaining computation can be fully recovered
194−-- from this context.
195−------------------------------------------------------------------------------
196−
197−-- Save the 'm' and 'f', in case we migrate to a new thread before the current
198−-- bind operation completes, we run the operation manually in the new thread
199−-- using this context.
200−saveContext :: Monad m => f a -> (a -> f b) -> StateT Context m ()
201−saveContext m f =
202− Lazy.modify $ \Context { fstack = fs, .. } ->
203− Context { currentm = unsafeCoerce m
204− , fstack = unsafeCoerce f : fs
205− , .. }
206−
207−-- pop the top function from the continuation stack, create the next closure,
208−-- set it as the current closure and return it.
209−restoreContext :: (MonadPlus m1, Monad m) => Maybe a -> StateT Context m (m1 b)
210−restoreContext x = do
211− -- XXX fstack must be non-empty when this is called.
212− ctx@Context { fstack = f:fs } <- Lazy.get
213− let mnext = maybe mzero (unsafeCoerce f) x
214− Lazy.put ctx { currentm = unsafeCoerce mnext, fstack = fs }
215− return mnext
216−
217−-- | We can retrieve a context at any point and resume that context at some
218−-- later point, upon resumption we start executing again from the same point
219−-- where the context was retrieved from.
220−--
221−-- This function composes the stored context to recover the computation.
222−--
223−composeContext :: Monad m => Context -> m a
224−composeContext Context { currentm = m
225− , fstack = fs
226− } =
227− unsafeCoerce m >>= composefStack fs
228−
229− where
230−
231− -- Compose the stack of pending functions in fstack using the bind
232− -- operation of the monad. The types don't match. We just coerce the types
233− -- here, we know that they actually match.
234−
235− -- The last computation is always 'mzero' to stop the whole computation.
236− -- Before returning we save the result of the computation in the context.
237−
238− -- composefStack :: Monad m => [a -> AsyncT m a] -> a -> AsyncT m b
239− composefStack [] _ = error "Bug: this should never be reached"
240− composefStack (f:ff) x = (unsafeCoerce f) x >>= composefStack ff
241−
242−------------------------------------------------------------------------------
243−-- Mailbox
244−------------------------------------------------------------------------------
245−
246−setContextMailBox :: Context -> a -> Context
247−setContextMailBox ctx mbdata = ctx { mailBox = Just $ unsafeCoerce mbdata }
248−
249−takeContextMailBox :: Monad m => StateT Context m (Either Context a)
250−takeContextMailBox = do
251− ctx <- Lazy.get
252− case mailBox ctx of
253− Nothing -> return $ Left ctx
254− Just x -> do
255− Lazy.put ctx { mailBox = Nothing }
256− return $ Right (unsafeCoerce x)