@@ -0,0 +1,545 @@
1+{-# LANGUAGE ExistentialQuantification #-}
2+{-# LANGUAGE FlexibleContexts #-}
3+{-# LANGUAGE FlexibleInstances #-}
4+{-# LANGUAGE InstanceSigs #-}
5+{-# LANGUAGE MultiParamTypeClasses #-}
6+{-# LANGUAGE Rank2Types #-}
7+{-# LANGUAGE RecordWildCards #-}
8+{-# LANGUAGE ScopedTypeVariables #-}
9+{-# LANGUAGE TypeFamilies #-}
10+{-# LANGUAGE UndecidableInstances #-}
11+
12+module Duct.AsyncT
13+ ( AsyncT (..)
14+ , waitAsync
15+ , compose
16+ , (<**)
17+ , onNothing
18+ , RemoteStatus (..)
19+ , back
20+ , waitForChildren
21+ , dbg
22+ )
23+where
24+
25+import Control.Applicative (Alternative (..))
26+import Control.Concurrent (ThreadId, myThreadId)
27+import Control.Concurrent.STM (TChan, atomically, newTChan,
28+ readTChan)
29+import Control.Exception (SomeException, catch)
30+import Control.Monad.Base (MonadBase (..), liftBaseDefault)
31+import Control.Monad.State (MonadIO (..), MonadPlus (..),
32+ MonadState (..), StateT (..),
33+ liftM, modify, runStateT, when)
34+import Control.Monad.Trans.Class (MonadTrans (lift))
35+import Control.Monad.Trans.Control (ComposeSt, MonadBaseControl (..),
36+ MonadTransControl (..),
37+ defaultLiftBaseWith,
38+ defaultRestoreM)
39+import Data.Dynamic (Typeable)
40+import Data.IORef (IORef, newIORef, readIORef,
41+ writeIORef)
42+import Data.Maybe (isJust, isNothing)
43+import Unsafe.Coerce (unsafeCoerce)
44+
45+import Duct.Event
46+import Duct.Log
47+
48+-- import Debug.Trace
49+
50+type StateM m = StateT EventF m
51+
52+newtype AsyncT m a = AsyncT { runAsyncT :: StateM m (Maybe a) }
53+
54+------------------------------------------------------------------------------
55+-- Remote data types
56+------------------------------------------------------------------------------
57+
58+data RemoteStatus = WasRemote | WasParallel | NoRemote
59+ deriving (Typeable, Eq, Show)
60+
61+------------------------------------------------------------------------------
62+-- Utilities
63+------------------------------------------------------------------------------
64+
65+-- dbg x = trace x (return ())
66+dbg :: Monad m => a -> m ()
67+dbg _ = return ()
68+
69+-- | If the first parameter is 'Nothing' return the second parameter otherwise
70+-- return the first parameter..
71+onNothing :: Monad m => m (Maybe b) -> m b -> m b
72+onNothing iox iox'= do
73+ mx <- iox
74+ case mx of
75+ Just x -> return x
76+ Nothing -> iox'
77+
78+------------------------------------------------------------------------------
79+-- Functor
80+------------------------------------------------------------------------------
81+
82+instance Monad (AsyncT m) => Functor (AsyncT m) where
83+ fmap f mx = do
84+ x <- mx
85+ return $ f x
86+
87+------------------------------------------------------------------------------
88+-- Applicative
89+------------------------------------------------------------------------------
90+
91+-- | Warning: Radically untyped stuff. handle with care
92+getContinuations :: Monad m => StateM m [a -> AsyncT m b]
93+getContinuations = do
94+ EventF { fcomp = fs } <- get
95+ return $ unsafeCoerce fs
96+
97+-- | Save a closure and a continuation ('x' and 'f' in 'x >>= f').
98+setContinuation :: Monad m
99+ => AsyncT m a -> (a -> AsyncT m b) -> [c -> AsyncT m c] -> StateM m ()
100+setContinuation b c fs = do
101+ modify $ \EventF{..} -> EventF { xcomp = b
102+ , fcomp = unsafeCoerce c : fs
103+ , .. }
104+
105+-- | Restore the continuations to the provided ones.
106+-- | NOTE: Events are also cleared out.
107+restoreStack :: MonadState EventF m => t -> m ()
108+restoreStack fs = modify $ \EventF {..} ->
109+ EventF { event = Nothing, fcomp = (unsafeCoerce fs), .. }
110+
111+instance MonadIO m => Applicative (AsyncT m) where
112+ pure a = AsyncT . return $ Just a
113+
114+ f <*> g = AsyncT $ do
115+ rf <- liftIO $ newIORef (Nothing,[])
116+ rg <- liftIO $ newIORef (Nothing,[])
117+ fs <- getContinuations
118+
119+ let hasWait (_:Wait:_) = True
120+ hasWait _ = False
121+
122+ appf k = AsyncT $ do
123+ Log rec _ full <- getData `onNothing` return (Log False [] [])
124+ (liftIO $ writeIORef rf (Just k,full))
125+ (x, full2)<- liftIO $ readIORef rg
126+ when (hasWait full ) $
127+ let full'= head full: full2
128+ in (setData $ Log rec full' full')
129+ return $ Just (unsafeCoerce k) <*> x
130+
131+ appg x = AsyncT $ do
132+ Log rec _ full <- getData `onNothing` return (Log False [] [])
133+ liftIO $ writeIORef rg (Just x, full)
134+ (k,full1) <- liftIO $ readIORef rf
135+ when (hasWait full) $
136+ let full'= head full: full1
137+ in (setData $ Log rec full' full')
138+ return $ unsafeCoerce k <*> Just x
139+
140+ setContinuation f appf fs
141+
142+ k <- runAsyncT f
143+ was <- getData `onNothing` return NoRemote
144+ when (was == WasParallel) $ setData NoRemote
145+
146+ Log recovery _ full <- getData `onNothing` return (Log False [] [])
147+
148+
149+
150+ if was== WasRemote || (not recovery && was == NoRemote && isNothing k )
151+ -- if the first operand was a remote request
152+ -- (so this node is not master and hasn't to execute the whole expression)
153+ -- or it was not an asyncronous term (a normal term without async or parallel
154+ -- like primitives) and is nothing
155+ then do
156+ restoreStack fs
157+ return Nothing
158+ else do
159+ when (isJust k) $ liftIO $ writeIORef rf (k,full)
160+ -- when necessary since it maybe WasParallel and Nothing
161+
162+ setContinuation g appg fs
163+
164+ x <- runAsyncT g
165+ Log _ _ full' <- getData `onNothing` return (Log False [] [])
166+ liftIO $ writeIORef rg (x,full')
167+ restoreStack fs
168+ k'' <- if was== WasParallel
169+ then do
170+ (k',_) <- liftIO $ readIORef rf -- since k may have been updated by a parallel f
171+ return k'
172+ else return k
173+ return $ k'' <*> x
174+
175+------------------------------------------------------------------------------
176+-- Alternative
177+------------------------------------------------------------------------------
178+
179+instance MonadIO m => Alternative (AsyncT m) where
180+ empty = AsyncT $ return Nothing
181+ (<|>) x y = AsyncT $ do
182+ mx <- runAsyncT x
183+ was <- getData `onNothing` return NoRemote
184+ if was == WasRemote
185+ then return Nothing
186+ else case mx of
187+ Nothing -> runAsyncT y
188+ justx -> return justx
189+
190+-- | A synonym of 'empty' that can be used in a monadic expression. It stops
191+-- the computation, which allows the next computation in an 'Alternative'
192+-- ('<|>') composition to run.
193+stop :: Alternative m => m stopped
194+stop = empty
195+
196+------------------------------------------------------------------------------
197+-- Monad
198+------------------------------------------------------------------------------
199+
200+-- | Total variant of `tail` that returns an empty list when given an empty list.
201+tailsafe :: [a] -> [a]
202+tailsafe [] = []
203+tailsafe (_:xs) = xs
204+
205+-- | Set the current closure and continuation for the current statement
206+setEventCont :: Monad m => AsyncT m a -> (a -> AsyncT m b) -> StateM m EventF
207+setEventCont x f = do
208+ EventF { fcomp = fs, .. } <- get
209+ let cont = EventF { xcomp = x
210+ , fcomp = unsafeCoerce f : unsafeCoerce fs
211+ , .. }
212+ put cont
213+ return cont
214+
215+-- | Reset the closure and continuation. Remove inner binds that the previous
216+-- computations may have stacked in the list of continuations.
217+resetEventCont :: forall m a b. MonadIO m
218+ => Maybe a -> EventF -> StateM m (AsyncT m b -> AsyncT m b)
219+resetEventCont mx _ = do
220+ EventF { fcomp = fs, .. } <- get
221+ let f my = case my of
222+ Nothing -> (empty :: AsyncT m a)
223+ Just x -> unsafeCoerce (head fs) x
224+ put $ EventF { xcomp = f mx
225+ , fcomp = tailsafe (unsafeCoerce fs)
226+ , .. }
227+ return id
228+
229+instance MonadIO m => Monad (AsyncT m) where
230+ return = pure
231+ x >>= f = AsyncT $ do
232+ c <- setEventCont x f
233+ mk <- runAsyncT x
234+ _ <- resetEventCont mk c
235+ case mk of
236+ Just k -> runAsyncT (f k)
237+ Nothing -> return Nothing
238+
239+instance MonadIO m => MonadPlus (AsyncT m) where
240+ mzero = empty
241+ mplus = (<|>)
242+
243+instance (Monoid a, Monad (AsyncT m)) => Monoid (AsyncT m a) where
244+ mappend x y = mappend <$> x <*> y
245+ mempty = return mempty
246+
247+------------------------------------------------------------------------------
248+-- Backtracking
249+------------------------------------------------------------------------------
250+
251+-- | Run the closure (the 'x' in 'x >>= f') of the current bind operation.
252+runClosure :: EventF -> StateM m (Maybe a)
253+runClosure EventF { xcomp = x } = runAsyncT (unsafeCoerce x)
254+
255+-- | Compose a list of continuations.
256+compose :: MonadIO m => [a -> AsyncT m a] -> (a -> AsyncT m b)
257+compose [] = const empty
258+compose (f:fs) = \x -> f x >>= compose fs
259+
260+-- | Run the continuation (the 'f' in 'x >>= f') of the current bind operation with the current state.
261+runContinuation :: MonadIO m => EventF -> a -> StateM m (Maybe b)
262+runContinuation EventF { fcomp = fs } =
263+ runAsyncT . (compose $ (unsafeCoerce fs))
264+
265+data Backtrack b = Show b => Backtrack
266+ { backtracking :: Maybe b
267+ , backStack :: [EventF]
268+ } deriving Typeable
269+
270+backStateOf :: (Monad m, Show a) => a -> m (Backtrack a)
271+backStateOf reason = return $ Backtrack (Nothing `asTypeOf` (Just reason)) []
272+
273+-- | Start the undo process for the given undo track id. Performs all the undo
274+-- actions registered till now in reverse order. An undo action can use
275+-- 'forward' to stop the undo process and resume forward execution. If there
276+-- are no more undo actions registered execution stops and a 'stop' action is
277+-- returned.
278+--
279+back :: (MonadIO m, Typeable b, Show b) => b -> AsyncT m a
280+back reason = AsyncT $ do
281+ bs <- getData `onNothing` backStateOf reason
282+ goBackt bs
283+
284+ where
285+
286+ goBackt (Backtrack _ [] )= return Nothing
287+ goBackt (Backtrack _ (stack@(first : bs)) )= do
288+ (setData $ Backtrack (Just reason) stack)
289+
290+ mr <- runClosure first
291+
292+ Backtrack b _ <- getData `onNothing` backStateOf reason
293+ case mr of
294+ Nothing -> return empty
295+ Just x -> case b of
296+ Nothing -> runContinuation first x
297+ justreason -> goBackt $ Backtrack justreason bs
298+
299+------------------------------------------------------------------------------
300+-- MonadIO
301+------------------------------------------------------------------------------
302+
303+instance MonadIO m => MonadIO (AsyncT m) where
304+ liftIO mx = do
305+ ex <- liftIO' $ (mx >>= return . Right) `catch`
306+ (\(e :: SomeException) -> return $ Left e)
307+ case ex of
308+ Left e -> back e
309+ Right x -> return x
310+ where liftIO' x = AsyncT $ liftIO x >>= return . Just
311+
312+-------------------------------------------------------------------------------
313+-- AsyncT transformer
314+-------------------------------------------------------------------------------
315+
316+instance MonadTrans AsyncT where
317+ lift mx = AsyncT $ lift mx >>= return . Just
318+
319+instance MonadTransControl AsyncT where
320+ type StT AsyncT a = (Maybe a, EventF)
321+ liftWith f = AsyncT $ StateT $ \s ->
322+ liftM (\x -> (Just x, s))
323+ (f $ \t -> runStateT (runAsyncT t) s)
324+ restoreT = AsyncT . StateT . const
325+ {-# INLINABLE liftWith #-}
326+ {-# INLINABLE restoreT #-}
327+
328+instance (MonadBase b m, MonadIO m) => MonadBase b (AsyncT m) where
329+ liftBase = liftBaseDefault
330+
331+instance (MonadBaseControl b m, MonadIO m) => MonadBaseControl b (AsyncT m) where
332+ type StM (AsyncT m) a = ComposeSt AsyncT m a
333+ liftBaseWith = defaultLiftBaseWith
334+ restoreM = defaultRestoreM
335+ {-# INLINABLE liftBaseWith #-}
336+ {-# INLINABLE restoreM #-}
337+
338+------------------------------------------------------------------------------
339+-- Thread management
340+------------------------------------------------------------------------------
341+
342+drainChildren :: MonadIO m => TChan ThreadId -> [EventF] -> m ()
343+drainChildren chan pending =
344+ if length pending == 0
345+ then return ()
346+ else do
347+ tid <- liftIO $ atomically $ readTChan chan
348+ dbg $ "Reaping child: " ++ show tid
349+ drainChildren chan $ filter (\x -> threadId x /= tid) pending
350+
351+waitForChildren :: MonadIO m => TChan ThreadId -> IORef [EventF] -> m ()
352+waitForChildren chan pendingRef = do
353+ pending <- liftIO $ readIORef pendingRef
354+ tid <- liftIO myThreadId
355+ dbg $ "Waiting for children: " ++ show tid
356+ drainChildren chan pending
357+ dbg $ "Done: " ++ show tid
358+ liftIO $ writeIORef pendingRef []
359+
360+------------------------------------------------------------------------------
361+-- Running the monad
362+------------------------------------------------------------------------------
363+
364+-- | Run a transient computation with a default initial state
365+runTransient :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, EventF)
366+runTransient t = do
367+ zombieChan <- liftIO $ atomically newTChan
368+ pendingRef <- liftIO $ newIORef []
369+ credit <- liftIO $ newIORef maxBound
370+ th <- liftIO $ myThreadId
371+
372+ r <- runStateT (runAsyncT t) $ initEventF
373+ (empty :: AsyncT m a) th zombieChan pendingRef credit
374+
375+ waitForChildren zombieChan pendingRef
376+ return r
377+
378+-- | Run an 'AsyncT m' computation. Returns the result of the computation or
379+-- may throw an exception. The computation finishes with a 'Nothing' result
380+-- when the input sources are exhausted.
381+-- XXX pass a collector function and return a Traversable.
382+-- Ideally it should be a non-empty list instead.
383+--waitAsync :: MonadIO m => AsyncT m a -> m [a]
384+waitAsync :: MonadIO m => AsyncT m a -> m (Maybe a)
385+waitAsync m = do
386+ (r, _) <- runTransient m
387+ return r
388+
389+------------------------------------------------------------------------------
390+-- * Extensible State: Session Data Management
391+------------------------------------------------------------------------------
392+
393+-- | Retrieve a previously stored data item of the given data type from the
394+-- monad state. The data type to retrieve is implicitly determined from the
395+-- requested type context.
396+-- If the data item is not found, an 'empty' value (a void event) is returned.
397+-- Remember that an empty value stops the monad computation. If you want to
398+-- print an error message or a default value in that case, you can use an
399+-- 'Alternative' composition. For example:
400+--
401+-- > getSData <|> error "no data"
402+-- > getInt = getSData <|> return (0 :: Int)
403+getSData :: (Monad m, Typeable a) => AsyncT m a
404+getSData = AsyncT getData
405+
406+-- | 'setSData' stores a data item in the monad state which can be retrieved
407+-- later using 'getData' or 'getSData'. Stored data items are keyed by their
408+-- data type, and therefore the data type must be 'Typeable' and only one item
409+-- of a given type can be stored. A newtype wrapper can be used to distinguish
410+-- two data items of the same type when required.
411+--
412+-- @
413+-- import Control.Monad.IO.Class (liftIO)
414+-- import Transient.Base
415+-- import Data.Typeable
416+--
417+-- data Person = Person
418+-- { name :: String
419+-- , age :: Int
420+-- } deriving Typeable
421+--
422+-- main = keep $ do
423+-- setSData $ Person "Alberto" 55
424+-- Person name age <- getSData
425+-- liftIO $ print (name, age)
426+-- @
427+setSData :: (Monad m, Typeable a) => a -> AsyncT m ()
428+setSData x = AsyncT $ setData x >> return (Just ())
429+
430+-- | Accepts a function that takes the current value of the stored data type
431+-- and returns the modified value. If the function returns 'Nothing' the value
432+-- is deleted otherwise updated.
433+modifySData :: (Monad m, Typeable a) => (Maybe a -> Maybe a) -> AsyncT m ()
434+modifySData x = AsyncT $ modifyData x >> return (Just ())
435+
436+-- | Delete the data item of the given type from the monad state.
437+delSData :: (Monad m, Typeable a) => a -> AsyncT m ()
438+delSData x = AsyncT $ delData x >> return (Just ())
439+
440+------------------------------------------------------------------------------
441+-- MonadState for AsyncT m
442+------------------------------------------------------------------------------
443+
444+instance (Monad m, Monad (AsyncT m)) => MonadState EventF (AsyncT m) where
445+ get = AsyncT $ get >>= return . Just
446+ put x = AsyncT $ put x >> return (Just ())
447+ state f = AsyncT $ do
448+ s <- get
449+ let ~(a, s') = f s
450+ put s'
451+ return $ Just a
452+
453+{-
454+-- | Run an action, if the result is a void action undo any state changes
455+-- that it might have caused.
456+try :: MonadIO m => AsyncT m a -> AsyncT m a
457+try mx = do
458+ sd <- gets mfData
459+ mx <|> (modify (\s -> s { mfData = sd }) >> empty)
460+
461+-- | Executes the computation and reset the state either if it fails or not
462+sandbox :: MonadIO m => AsyncT m a -> AsyncT m a
463+sandbox mx = do
464+ sd <- gets mfData
465+ mx <*** modify (\s ->s { mfData = sd})
466+ -}
467+
468+------------------------------------------------------------------------------
469+-- More operators, instances
470+------------------------------------------------------------------------------
471+
472+instance (Num a, Monad (AsyncT m)) => Num (AsyncT m a) where
473+ fromInteger = return . fromInteger
474+ mf + mg = (+) <$> mf <*> mg
475+ mf * mg = (*) <$> mf <*> mg
476+ negate f = f >>= return . negate
477+ abs f = f >>= return . abs
478+ signum f = f >>= return . signum
479+
480+class AdditionalOperators m where
481+
482+ -- | Run @m a@ discarding its result before running @m b@.
483+ (**>) :: m a -> m b -> m b
484+
485+ -- | Run @m b@ discarding its result, after the whole task set @m a@ is
486+ -- done.
487+ (<**) :: m a -> m b -> m a
488+
489+ atEnd' :: m a -> m b -> m a
490+ atEnd' = (<**)
491+
492+ -- | Run @m b@ discarding its result, once after each task in @m a@, and
493+ -- once again after the whole task set is done.
494+ (<***) :: m a -> m b -> m a
495+
496+ atEnd :: m a -> m b -> m a
497+ atEnd = (<***)
498+
499+instance (Monad m, Monad (AsyncT m)) => AdditionalOperators (AsyncT m) where
500+
501+ (**>) :: AsyncT m a -> AsyncT m b -> AsyncT m b
502+ (**>) x y =
503+ AsyncT $ do
504+ runAsyncT x
505+ runAsyncT y
506+
507+ (<***) :: AsyncT m a -> AsyncT m b -> AsyncT m a
508+ (<***) ma mb =
509+ AsyncT $ do
510+ fs <- getContinuations
511+ setContinuation ma (\x -> mb >> return x) fs
512+ a <- runAsyncT ma
513+ runAsyncT mb
514+ restoreStack fs
515+ return a
516+
517+ (<**) :: AsyncT m a -> AsyncT m b -> AsyncT m a
518+ (<**) ma mb =
519+ AsyncT $ do
520+ a <- runAsyncT ma
521+
522+ runAsyncT mb
523+ return a
524+
525+infixr 1 <***, <**, **>
526+
527+-- | Run @b@ once, discarding its result when the first task in task set @a@
528+-- has finished. Useful to start a singleton task after the first task has been
529+-- setup.
530+(<|) :: MonadIO m => AsyncT m a -> AsyncT m b -> AsyncT m a
531+(<|) ma mb = AsyncT $ do
532+ fs <- getContinuations
533+ ref <- liftIO $ newIORef False
534+ setContinuation ma (cont ref) fs
535+ r <- runAsyncT ma
536+ restoreStack fs
537+ return r
538+ where cont ref x = AsyncT $ do
539+ n <- liftIO $ readIORef ref
540+ if n == True
541+ then return $ Just x
542+ else do liftIO $ writeIORef ref True
543+ runAsyncT mb
544+ return $ Just x
545+