@@ -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 #-}
25+import Control.Applicative (Alternative (..))
26+import Control.Concurrent (ThreadId, myThreadId)
27+import Control.Concurrent.STM (TChan, atomically, newTChan,
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 (..),
39+import Data.Dynamic (Typeable)
40+import Data.IORef (IORef, newIORef, readIORef,
42+import Data.Maybe (isJust, isNothing)
43+import Unsafe.Coerce (unsafeCoerce)
50+type StateM m = StateT EventF m
52+newtype AsyncT m a = AsyncT { runAsyncT :: StateM m (Maybe a) }
54+------------------------------------------------------------------------------
56+------------------------------------------------------------------------------
58+data RemoteStatus = WasRemote | WasParallel | NoRemote
59+ deriving (Typeable, Eq, Show)
61+------------------------------------------------------------------------------
63+------------------------------------------------------------------------------
65+-- dbg x = trace x (return ())
66+dbg :: Monad m => a -> m ()
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
78+------------------------------------------------------------------------------
80+------------------------------------------------------------------------------
82+instance Monad (AsyncT m) => Functor (AsyncT m) where
87+------------------------------------------------------------------------------
89+------------------------------------------------------------------------------
91+-- | Warning: Radically untyped stuff. handle with care
92+getContinuations :: Monad m => StateM m [a -> AsyncT m b]
94+ EventF { fcomp = fs } <- get
95+ return $ unsafeCoerce fs
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
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), .. }
111+instance MonadIO m => Applicative (AsyncT m) where
112+ pure a = AsyncT . return $ Just a
114+ f <*> g = AsyncT $ do
115+ rf <- liftIO $ newIORef (Nothing,[])
116+ rg <- liftIO $ newIORef (Nothing,[])
117+ fs <- getContinuations
119+ let hasWait (_:Wait:_) = True
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
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
140+ setContinuation f appf fs
143+ was <- getData `onNothing` return NoRemote
144+ when (was == WasParallel) $ setData NoRemote
146+ Log recovery _ full <- getData `onNothing` return (Log False [] [])
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
159+ when (isJust k) $ liftIO $ writeIORef rf (k,full)
160+ -- when necessary since it maybe WasParallel and Nothing
162+ setContinuation g appg fs
165+ Log _ _ full' <- getData `onNothing` return (Log False [] [])
166+ liftIO $ writeIORef rg (x,full')
168+ k'' <- if was== WasParallel
170+ (k',_) <- liftIO $ readIORef rf -- since k may have been updated by a parallel f
175+------------------------------------------------------------------------------
177+------------------------------------------------------------------------------
179+instance MonadIO m => Alternative (AsyncT m) where
180+ empty = AsyncT $ return Nothing
181+ (<|>) x y = AsyncT $ do
183+ was <- getData `onNothing` return NoRemote
187+ Nothing -> runAsyncT y
188+ justx -> return justx
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
196+------------------------------------------------------------------------------
198+------------------------------------------------------------------------------
200+-- | Total variant of `tail` that returns an empty list when given an empty list.
201+tailsafe :: [a] -> [a]
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
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)
229+instance MonadIO m => Monad (AsyncT m) where
231+ x >>= f = AsyncT $ do
232+ c <- setEventCont x f
234+ _ <- resetEventCont mk c
236+ Just k -> runAsyncT (f k)
237+ Nothing -> return Nothing
239+instance MonadIO m => MonadPlus (AsyncT m) where
243+instance (Monoid a, Monad (AsyncT m)) => Monoid (AsyncT m a) where
244+ mappend x y = mappend <$> x <*> y
245+ mempty = return mempty
247+------------------------------------------------------------------------------
249+------------------------------------------------------------------------------
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)
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
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))
265+data Backtrack b = Show b => Backtrack
266+ { backtracking :: Maybe b
267+ , backStack :: [EventF]
270+backStateOf :: (Monad m, Show a) => a -> m (Backtrack a)
271+backStateOf reason = return $ Backtrack (Nothing `asTypeOf` (Just reason)) []
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
279+back :: (MonadIO m, Typeable b, Show b) => b -> AsyncT m a
280+back reason = AsyncT $ do
281+ bs <- getData `onNothing` backStateOf reason
286+ goBackt (Backtrack _ [] )= return Nothing
287+ goBackt (Backtrack _ (stack@(first : bs)) )= do
288+ (setData $ Backtrack (Just reason) stack)
290+ mr <- runClosure first
292+ Backtrack b _ <- getData `onNothing` backStateOf reason
294+ Nothing -> return empty
296+ Nothing -> runContinuation first x
297+ justreason -> goBackt $ Backtrack justreason bs
299+------------------------------------------------------------------------------
301+------------------------------------------------------------------------------
303+instance MonadIO m => MonadIO (AsyncT m) where
305+ ex <- liftIO' $ (mx >>= return . Right) `catch`
306+ (\(e :: SomeException) -> return $ Left e)
310+ where liftIO' x = AsyncT $ liftIO x >>= return . Just
312+-------------------------------------------------------------------------------
313+-- AsyncT transformer
314+-------------------------------------------------------------------------------
316+instance MonadTrans AsyncT where
317+ lift mx = AsyncT $ lift mx >>= return . Just
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 #-}
328+instance (MonadBase b m, MonadIO m) => MonadBase b (AsyncT m) where
329+ liftBase = liftBaseDefault
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 #-}
338+------------------------------------------------------------------------------
340+------------------------------------------------------------------------------
342+drainChildren :: MonadIO m => TChan ThreadId -> [EventF] -> m ()
343+drainChildren chan pending =
344+ if length pending == 0
347+ tid <- liftIO $ atomically $ readTChan chan
348+ dbg $ "Reaping child: " ++ show tid
349+ drainChildren chan $ filter (\x -> threadId x /= tid) pending
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 []
360+------------------------------------------------------------------------------
362+------------------------------------------------------------------------------
364+-- | Run a transient computation with a default initial state
365+runTransient :: forall m a. MonadIO m => AsyncT m a -> m (Maybe a, EventF)
367+ zombieChan <- liftIO $ atomically newTChan
368+ pendingRef <- liftIO $ newIORef []
369+ credit <- liftIO $ newIORef maxBound
370+ th <- liftIO $ myThreadId
372+ r <- runStateT (runAsyncT t) $ initEventF
373+ (empty :: AsyncT m a) th zombieChan pendingRef credit
375+ waitForChildren zombieChan pendingRef
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)
386+ (r, _) <- runTransient m
389+------------------------------------------------------------------------------
390+-- * Extensible State: Session Data Management
391+------------------------------------------------------------------------------
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:
401+-- > getSData <|> error "no data"
402+-- > getInt = getSData <|> return (0 :: Int)
403+getSData :: (Monad m, Typeable a) => AsyncT m a
404+getSData = AsyncT getData
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.
413+-- import Control.Monad.IO.Class (liftIO)
414+-- import Transient.Base
415+-- import Data.Typeable
417+-- data Person = Person
420+-- } deriving Typeable
423+-- setSData $ Person "Alberto" 55
424+-- Person name age <- getSData
425+-- liftIO $ print (name, age)
427+setSData :: (Monad m, Typeable a) => a -> AsyncT m ()
428+setSData x = AsyncT $ setData x >> return (Just ())
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 ())
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 ())
440+------------------------------------------------------------------------------
441+-- MonadState for AsyncT m
442+------------------------------------------------------------------------------
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
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
459+ mx <|> (modify (\s -> s { mfData = sd }) >> empty)
461+-- | Executes the computation and reset the state either if it fails or not
462+sandbox :: MonadIO m => AsyncT m a -> AsyncT m a
465+ mx <*** modify (\s ->s { mfData = sd})
468+------------------------------------------------------------------------------
469+-- More operators, instances
470+------------------------------------------------------------------------------
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
480+class AdditionalOperators m where
482+ -- | Run @m a@ discarding its result before running @m b@.
483+ (**>) :: m a -> m b -> m b
485+ -- | Run @m b@ discarding its result, after the whole task set @m a@ is
487+ (<**) :: m a -> m b -> m a
489+ atEnd' :: m a -> m b -> m a
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
496+ atEnd :: m a -> m b -> m a
499+instance (Monad m, Monad (AsyncT m)) => AdditionalOperators (AsyncT m) where
501+ (**>) :: AsyncT m a -> AsyncT m b -> AsyncT m b
507+ (<***) :: AsyncT m a -> AsyncT m b -> AsyncT m a
510+ fs <- getContinuations
511+ setContinuation ma (\x -> mb >> return x) fs
517+ (<**) :: AsyncT m a -> AsyncT m b -> AsyncT m a
525+infixr 1 <***, <**, **>
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
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
538+ where cont ref x = AsyncT $ do
539+ n <- liftIO $ readIORef ref
541+ then return $ Just x
542+ else do liftIO $ writeIORef ref True