diff --git a/Control/Workflow.hs b/Control/Workflow.hs --- a/Control/Workflow.hs +++ b/Control/Workflow.hs @@ -1,1342 +1,1346 @@ -{-# LANGUAGE OverlappingInstances - , UndecidableInstances - , ExistentialQuantification - , ScopedTypeVariables - , MultiParamTypeClasses - , FlexibleInstances - , FlexibleContexts - , TypeSynonymInstances - , DeriveDataTypeable - , RecordWildCards - , BangPatterns - #-} -{-# OPTIONS -IControl/Workflow #-} - - -{- | A workflow can be seen as a persistent thread. -The workflow monad writes a log that permit to restore the thread -at the interrupted point. `step` is the (partial) monad transformer for -the Workflow monad. A workflow is defined by its name and, optionally -by the key of the single parameter passed. There primitives for starting workflows -also restart the interrupted workflow when it has been in execution previously. - - -A small example that print the sequence of integers in te console -if you interrupt the progam, when restarted again, it will -start from the last printed number - -@module Main where -import Control.Workflow -import Control.Concurrent(threadDelay) -import System.IO (hFlush,stdout) - -mcount n= do `step` $ do - putStr (show n ++ \" \") - hFlush stdout - threadDelay 1000000 - mcount (n+1) - return () -- to disambiguate the return type - -main= `exec1` \"count\" $ mcount (0 :: Int)@ - ->>>runghc demos\sequence.hs ->0 1 2 3 ->CTRL-C Pressed ->>>runghc demos\sequence.hs ->3 4 5 6 7 ->CTRL-C Pressed ->>>runghc demos\sequence.hs ->7 8 9 10 11 -... - -The program restart at the last saved step. - -As you can see, some side effect can be re-executed after recovery if -the log is not complete. This may happen after an unexpected shutdown (in this case) -or due to an asynchronous log writing policy. (see `syncWrite`) - -When the step results are big and complex, use the "Data.RefSerialize" package to define the (de)serialization instances -The log size will be reduced. printWFHistory` method will print the structure changes -in each step. - -If instead of `RefSerialize`, you use read and show instances, there will - be no reduction. but still it will work, and the log will be readable for debugging purposes. - The RefSerialize istance is automatically derived from Read, Show instances. - -Data.Binary instances are also fine for serialization. To use Binary, just define a binary instance -of your data by using `showpBinary` and `readpBinary`. - -Within the RefSerialize instance of a structure, you can freely mix -Show,Read RefSerialize and Data Binary instances. - - - -"Control.Workflow.Patterns" contains higher level workflow patterns for handling multiple workflows - -"Control.Workflow.Configuration" permits the use of workflows for configuration purposes - --} - -module Control.Workflow - -( - Stat -, Workflow -- a useful type name -, WorkflowList -, PMonadTrans (..) -, MonadCatchIO (..) -, HasFork(..) -, throw -, Indexable(..) -, keyWF --- * Start/restart workflows -, start -, exec -, exec1d -, exec1 -, exec1nc -, wfExec -, startWF -, restartWorkflows -, WFErrors(..) --- * Lifting to the Workflow monad -, step -, getWFStat -, stepExec ---, while ---, label ---, stepControl ---, stepDebug -, unsafeIOtoWF --- * References to intermediate values in the workflow log -, WFRef -, newWFRef -, stepWFRef -, readWFRef --- * State manipulation -, writeWFRef -, moveState --- * Workflow inspection -, waitWFActive -, getAll ---, getStep -, safeFromIDyn -, getWFKeys -, getWFHistory -, waitFor -, waitForSTM --- * Persistent timeouts -, waitUntilSTM -, getTimeoutFlag -, withTimeout -, withKillTimeout --- * Trace logging -, logWF --- * Termination of workflows -, clearRunningFlag -, killThreadWF -, killWF -, delWF -, killThreadWF1 -, delWFHistory -, delWFHistory1 --- * Log writing policy -, syncWrite -, SyncMode(..) --- * Print log history -, showHistory -, isInRecover --- * Low leve, internal use -, runWF1 -, getState - -) - -where - -import Prelude hiding (catch) -import System.IO.Unsafe -import Control.Monad(when,liftM) -import Control.Applicative -import qualified Control.Exception as CE (Exception,AsyncException(ThreadKilled), SomeException, ErrorCall, throwIO, handle,finally,catch,block,unblock) -import Control.Concurrent -- (forkIO,threadDelay, ThreadId, myThreadId, killThread) -import Control.Concurrent.STM -import GHC.Conc(unsafeIOToSTM) -import GHC.Base (maxInt) - - -import Data.ByteString.Lazy.Char8 as B hiding (index) -import Data.ByteString.Lazy as BL(putStrLn) -import Data.List as L -import Data.Typeable -import System.Time -import Control.Monad.Trans ---import Control.Concurrent.MonadIO(HasFork(..),MVar,newMVar,takeMVar,putMVar,readMVar) - - -import System.IO(hPutStrLn, stderr) -import Data.List(elemIndex) -import Data.Maybe -import Data.IORef -import System.IO.Unsafe(unsafePerformIO) -import Data.Map as M(Map,fromList,elems, insert, delete, lookup,toList, fromList,keys) -import qualified Control.Monad.CatchIO as CMC -import qualified Control.Exception.Extensible as E - -import Data.TCache -import Data.TCache.Defs -import Data.RefSerialize -import Data.Persistent.IDynamic -import Unsafe.Coerce -import System.Mem.StableName -import Control.Workflow.Stat - ---import Debug.Trace ---(!>)= flip trace - -type Workflow m = WF Stat m -- not so scary - -type WorkflowList m a b= M.Map String (a -> Workflow m b) - -instance Monad m => Monad (WF s m) where - return x = WF (\s -> return (s, x)) - WF g >>= f = WF (\s -> do - (s1, x) <- g s - let WF fun= f x - fun s1) - - - -instance (Monad m,Functor m) => Functor (Workflow m ) where - fmap f (WF g)= WF (\s -> do - (s1, x) <- g s - return (s1, f x)) - -tvRunningWfs = getDBRef $ keyRunning :: DBRef Stat - - - --- | Executes a computation inside of the workflow monad whatever the monad encapsulated in the workflow. --- Warning: this computation is executed whenever --- the workflow restarts, no matter if it has been already executed previously. This is useful for intializations or debugging. --- To avoid re-execution when restarting use: @'step' $ unsafeIOtoWF...@ --- --- To perform IO actions in a workflow that encapsulates an IO monad, use step over the IO action directly: --- --- @ 'step' $ action @ --- --- instead of --- --- @ 'step' $ unsafeIOtoWF $ action @ -unsafeIOtoWF :: (Monad m) => IO a -> Workflow m a -unsafeIOtoWF x= let y= unsafePerformIO ( x >>= return) in y `seq` return y - - -{- | @PMonadTrans@ permits |to define a partial monad transformer. They are not defined for all kinds of data -but the ones that have instances of certain classes.That is because in the lift instance code there are some -hidden use of these classes. This also may permit an accurate control of effects. -An instance of MonadTrans is an instance of PMonadTrans --} -class PMonadTrans t m a where - plift :: Monad m => m a -> t m a - - - --- | @plift= step@ -instance (Monad m - , MonadIO m - , Serialize a - , Typeable a) - => PMonadTrans (WF Stat) m a - where - plift = step - -instance (Monad m, Functor m) => Applicative (Workflow m) where - pure x= return x - WF f <*> WF g= WF $ \s -> do - (s1, k) <- f s - (s2, x) <- g s1 - return (s2,k x) - - - --- | An instance of MonadTrans is an instance of PMonadTrans -instance (MonadTrans t, Monad m) => PMonadTrans t m a where - plift= Control.Monad.Trans.lift - ---- | Handle with care: this instance will force --- the execution at recovery of every liftted IO procedure --- better use 'step . liftIO' instead of 'liftIO' - - -instance MonadIO m => MonadIO (WF Stat m) where - liftIO= unsafeIOtoWF - - -{- | Adapted from the @MonadCatchIO-mtl@ package. However, in this case it is needed to express serializable constraints about the returned values, -so the usual class definitions for lifting IO functions are not suitable. --} - -class MonadCatchIO m a where - -- | Generalized version of 'E.catch' - catch :: E.Exception e => m a -> (e -> m a) -> m a - - -- | Generalized version of 'E.block' - block :: m a -> m a - - -- | Generalized version of 'E.unblock' - unblock :: m a -> m a - - - --- | Generalized version of 'E.throwIO' -throw :: (MonadIO m, E.Exception e) => e -> m a -throw = liftIO . E.throwIO - - - - - -instance (Serialize a - , Typeable a,MonadIO m, CMC.MonadCatchIO m) - => MonadCatchIO (WF Stat m) a where - catch exp exc = do - expwf <- step $ getTempName - excwf <- step $ getTempName - step $ do - ex <- CMC.catch (exec1d expwf exp >>= return . Right ) $ \e-> return $ Left e - - case ex of - Right r -> return r -- All right - Left e ->exec1d excwf (exc e) - -- An exception occured in the main workflow - -- the exception workflow is executed - - - - - block exp=WF $ \s -> CMC.block (st exp $ s) - - unblock exp= WF $ \s -> CMC.unblock (st exp $ s) - -data WFInfo= WFInfo{ name :: String - , finished :: Bool - , haserror :: Maybe WFErrors } - deriving (Typeable,Read, Show) - -class MonadIO io => HasFork io where - fork :: io () -> io ThreadId - -instance HasFork IO where - fork= forkIO - -instance (HasFork io, MonadIO io - , CMC.MonadCatchIO io) - => HasFork (WF Stat io) where - fork f = do - (r,info@(WFInfo str finished status)) <- stepWFRef $ getTempName >>= \n -> return(WFInfo n False Nothing) - - WF $ \s -> do - th <- if finished then fork $ return() - else - fork $ - exec1 str f >> labelFinish r str Nothing - `CMC.catch` \(E.ErrorCall str) -> do - liftIO . atomicallySync $ writeWFRef r (WFInfo str True (Just $ WFException str)) -- !> ("ERROR *****"++show e) - killWF1 $ keyWF str () - `CMC.catch` \(e :: E.SomeException) -> do - liftIO . atomicallySync $ writeWFRef r (WFInfo str True (Just . WFException $ show e)) -- !> ("ERROR *****"++show e) - killWF1 $ keyWF str () - - - return (s,th) - where - labelFinish r str err= liftIO . atomicallySync $ writeWFRef r (WFInfo str True err) -- !> "finished" - - --- | Start or restart an anonymous workflow inside another workflow. --- Its state is deleted when finished and the result is stored in --- the parent's WF state. -wfExec - :: (Serialize a, Typeable a - , CMC.MonadCatchIO m, MonadIO m) - => Workflow m a -> Workflow m a -wfExec f= do - str <- step $ getTempName - step $ exec1 str f - --- | A version of exec1 that deletes its state after complete execution or thread killed -exec1d :: (MonadIO m, CMC.MonadCatchIO m) - => String -> (Workflow m b) -> m b -exec1d str f= do - r <- exec1 str f - delit - return r - `CMC.catch` (\e@CE.ThreadKilled -> delit >> throw e) - - where - delit= do - delWF str () - - - - --- | A version of exec with no seed parameter. -exec1 :: ( Monad m, MonadIO m, CMC.MonadCatchIO m) - => String -> Workflow m a -> m a - -exec1 str f= exec str (const f) () - - - - --- | Start or continue a workflow with exception handling --- the workflow flags are updated even in case of exception --- `WFerrors` are raised as exceptions -exec :: ( Indexable a, Serialize a, Typeable a - , Monad m, MonadIO m, CMC.MonadCatchIO m) - => String -> (a -> Workflow m b) -> a -> m b -exec str f x = - (do - v <- getState str f x - case v of - Right (name, f, stat) -> do - r <- runWF name (f x) stat - return r - Left err -> CMC.throw err) - `CMC.catch` - (\(e :: CE.SomeException) -> liftIO $ do - let name= keyWF str x - clearRunningFlag name --`debug` ("exception"++ show e) - - CMC.throw e ) - --- | executes a workflow, but does not mark it as finished even if --- the process ended. --- It this case, the workflow just will return the last result. --- If the workflow was gathering data from user questions for a configuration, then this --- primitive will store them in the log the first time, and can be retrieve it the next time. -exec1nc :: ( Monad m, MonadIO m, CMC.MonadCatchIO m) - => String -> Workflow m a -> m a -exec1nc str f =do - v <- getState str f () - case v of - Left err -> CMC.throw err - Right (name, f, stat) -> do - runWF1 name f stat False - - `CMC.catch` - (\(e :: CE.SomeException) -> liftIO $ do - let name= keyWF str () - clearRunningFlag name --`debug` ("exception"++ show e) - CMC.throw e ) - `CMC.finally` - (liftIO . atomically . - when(recover stat) $ do - let ref= self stat - s <- readDBRef ref `justifyM` error ("step: not found: "++ wfName stat) - writeDBRef ref s{recover= False,versions=L.reverse $ versions s}) - -mv :: MVar Int -mv= unsafePerformIO $ newMVar 0 - -getTempName :: MonadIO m => m String -getTempName= liftIO $ do - seq <- takeMVar mv - putMVar mv (seq + 1) - TOD t _ <- getClockTime - return $ "anon"++ show t ++ show seq - - - - --- Permits the modification of the workflow state by the procedure being lifted --- if the boolean value is True. This is used internally for control purposes ---stepControl :: ( Monad m --- , MonadIO m --- , Serialize a --- , Typeable a) --- => m a --- -> Workflow m a ---stepControl= stepControl1 True - - --- | Lifts a monadic computation to the WF monad, and provides transparent state loging and resuming the computation --- Note: Side effect can be repeated at recovery time if the log was not complete before shut down --- see the integer sequence example, above. -step :: ( MonadIO m - , Serialize a - , Typeable a) - => m a - -> Workflow m a - -step mx= WF(\s -> do - let - recovers= recover s - versionss= versions s --- !> "vvvvvvvvvvvvvvvvvvv" --- !> (unpack $ runW $ showp $ versions s) --- !> (show $ references s) --- !> (show $ "recover="++ show( recover s)) --- !> "^^^^^^^^^^^^^^^^^^^" - if recovers && not (L.null versionss) - then - return (s{versions=L.tail versionss }, fromIDyn $ L.head versionss ) - else do - let ref= self s - when (recovers && L.null versionss) $ do - liftIO $ atomically $ do - s' <- readDBRef ref `justifyM` error ("step: not found: "++ wfName s) - writeDBRef ref s'{recover= False,references= references s} - stepExec1 ref mx) - -getWFStat :: Monad m => Workflow m (DBRef Stat) -getWFStat= WF $ \s -> return (s,self s) - -stepExec - :: (Typeable t, Serialize t, MonadIO m) => - DBRef Stat -> m t -> m (DBRef Stat, t) -stepExec ref mx= do - (s,x) <- stepExec1 ref mx - return (self s, x) - -stepExec1 sref mx= do - x' <- mx - liftIO . atomicallySync $ do - s <- readDBRef sref >>= return . fromMaybe (error $ "step: readDBRef: not found:" ++ keyObjDBRef sref) - let versionss= versions s - dynx= toIDyn x' - ver= dynx: versionss - s'= s{ recover= False, versions = ver, state= state s+1} - writeDBRef sref s' - return (s', x') - ---undoStep :: Monad m => Workflow m () ---undoStep= WF $ \s@Stat{..} -> return(s{state=state-1, versions= L.tail versions},()) - --- | True if the workflow in recovery mode, reading the log to recover the process state -isInRecover :: Monad m => Workflow m Bool -isInRecover = WF(\s@Stat{..} -> - if recover && not (L.null versions ) then return(s,True ) - else if recover== True then return(s{recover=False}, False) - else return (s,False)) - --- | For debugging purposes. --- At recovery time, instead of returning the stored value from the log --- , stepDebug executes the computation 'f' as normally. --- . It permits the exact re-execution of a workflow process -stepDebug :: ( Monad m - , MonadIO m - , Serialize a - , Typeable a) - => m a - -> Workflow m a -stepDebug f = r - where - r= do - WF(\s -> - let stat= state s - - - in case recover s && not(L.null $ versions s) of - True -> f >>= \x -> return (s{versions= L.tail $ versions s},x) - False -> stepExec1 (self s) f) - --- Executes a computation 'f' in a loop while the return value meets the condition 'cond' is met. --- At recovery time, the current state of the loop is restored. --- The loop restart at the last internal state that was (saved) before shutdown. --- --- The use of 'while' permits a faster recovery when the loop has many steps and the log is very long, as is the case in --- MFlow applications, ---while --- :: MonadIO m => --- (b -> Bool) -> Workflow m b -> Workflow m b ---while cond f= do --- n <- WF $ \s -> return (s,state s - L.length (versions s)) ----- do ----- let versionss= versions s ----- if recover s && not (L.null versionss) ----- then return (s{versions=L.tail versionss }, fromIDyn $ L.head versionss ) ----- ----- else return(s{recover= False, state=state s + 1 ----- ,versions= (toIDyn $ state s):versionss} ----- ,state s) --- while1 n --- where --- while1 n =do --- label n --- x <- f --- if cond x --- then while1 n --- else return x --- ---data Label= Label Int deriving (Eq,Typeable,Read,Show) ---label n = do --- let !label= Label n --- r <- isInRecover --- if r --- then WF(\s@Stat{..} -> --- let !label@(Label n) = fromIDyn $ L.head versions --- !vers = filterMax (\d -> Just label /= safeFromIDyn d) versions -- !> (show label) --- in return (s{versions= L.tail vers}, fromIDyn . L.head $ vers )) --- else do --- step $ return label --- where --- filterMax f xs= --- case L.dropWhile f (L.tail xs) of --- [] -> xs --- [_] -> xs --- xs' -> filterMax f xs' --- - - - --- | Start or continue a workflow . --- 'WFErrors' and exceptions are returned as @Left err@ (even if they were triggered as exceptions). --- Other exceptions are returned as @Left (Exception e)@ --- use `killWF` or `delWF` in case of error to clear the log. -start - :: ( CMC.MonadCatchIO m - , MonadIO m - , Indexable a - , Serialize a - , Typeable a) - => String -- ^ name that identifies the workflow. - -> (a -> Workflow m b) -- ^ workflow to execute - -> a -- ^ initial value (ever use the initial value for restarting the workflow) - -> m (Either WFErrors b) -- ^ result of the computation -start namewf f1 v = do - ei <- getState namewf f1 v - case ei of - Left error -> return $ Left error - Right (name, f, stat) -> - runWF name (f v) stat >>= return . Right - `CMC.catch` - (\(e :: WFErrors) -> do - let name= keyWF namewf v - clearRunningFlag name - return $ Left e ) - `CMC.catch` - (\(E.ErrorCall msg) ->do - let name= keyWF namewf v - clearRunningFlag name - return . Left $ WFException msg ) - `CMC.catch` - (\(e :: CE.SomeException) -> liftIO $ do - let name= keyWF namewf v - clearRunningFlag name - return . Left $ WFException $ show e ) - - - - --- | Return conditions from the invocation of start/restart primitives -data WFErrors = NotFound | AlreadyRunning | Timeout | WFException String deriving (Typeable, Read, Show) - ---instance Show WFErrors where --- show NotFound= "Not Found" --- show AlreadyRunning= "Already Running" --- show Timeout= "Timeout" --- show (Exception e)= "Exception: "++ show e - ---instance Serialize WFErrors where --- showp NotFound= insertString "NotFound" --- showp AlreadyRunning= insertString "AlreadyRunning" --- showp Timeout= insertString "Timeout" --- showp (Exception e)= insertString "Exception: ">> showp e --- --- readp= choice[notfound,already,timeout, exc] --- where --- notfound= symbol "NotFound" >> return NotFound --- already= symbol "AlreadyRunning" >> return AlreadyRunning --- timeout= symbol "Timeout" >> return Timeout --- exc= symbol "Exception" >> readp >>= \s -> return (Exception s) - -instance CE.Exception WFErrors - - - -{- -lookup for any workflow for the entry value v -if namewf is found and is running, return arlready running - if is not runing, restart it -else start anew. --} - - -getState :: (Monad m, MonadIO m, Indexable a, Serialize a, Typeable a) - => String -> x -> a - -> m (Either WFErrors (String, x, Stat)) -getState namewf f v= liftIO . atomically $ getStateSTM - where - getStateSTM = do - mrunning <- readDBRef tvRunningWfs - case mrunning of - Nothing -> do - writeDBRef tvRunningWfs (Running $ fromList []) - getStateSTM - Just(Running map) -> do - let key= keyWF namewf v - dynv= toIDyn v - stat1= stat0{wfName= key,versions=[dynv],state=1, self= sref`seq`sref} - sref= getDBRef $ keyResource stat1 - case M.lookup key map of - Nothing -> do -- no workflow started for this object - mythread <- unsafeIOToSTM $ myThreadId - safeIOToSTM $ delResource stat1 >> writeResource stat1 - writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map - writeDBRef sref stat1 - return $ Right (key, f, stat1) -- !> "NEW WF" - - Just (wf, started) -> -- a workflow has been initiated for this object - if isJust started - then return $ Left AlreadyRunning -- !> "already running" - else do - mst <- readDBRef sref -- !> "has been running but not running now" - stat' <- case mst of - Nothing -> return stat1 -- error $ "getState: Workflow not found: "++ key - Just s' -> do - -- the thread may have been killed by an exception when running - s <- case recover s' of - True -> return s' - False -> do - s'' <- safeIOToSTM $ readResource s' `onNothing` return stat1 - let i= state s'' - j= state s' - return s'{versions= versions s'' ++ L.reverse ( L.take ( j - i) $ versions s')} - if isJust (timeout s) - then do - tnow <- unsafeIOToSTM getTimeSeconds - if lastActive s+ fromJust(timeout s) > tnow -- !>("lastActive="++show (lastActive s) ++ "tnow="++show tnow) - then - return s{recover= True,timeout=Nothing} - else - -- has been inactive for too much time, clean it - return stat1 - - else return s{recover= True} - - - writeDBRef sref stat' - mythread <- unsafeIOToSTM myThreadId - writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map - - return $ Right (key, f, stat') - - - -runWF :: ( Monad m, MonadIO m) - => String -> Workflow m b -> Stat -> m b -runWF n f s = runWF1 n f s True - - - -runWF1 n f s clear= do - (s', v') <- st f s{versions= L.tail $ versions s} - liftIO $ if clear then clearFromRunningList n - else clearRunningFlag n >> return () - return v' - where - - -- eliminate the thread from the list of running workflows but leave the state - clearFromRunningList n = atomicallySync $ do - Just(Running map) <- readDBRef tvRunningWfs -- !> "clearFormRunning" - writeDBRef tvRunningWfs . Running $ M.delete n map --- flushDBRef (getDBRef n :: DBRef Stat) - --- | Start or continue a workflow from a list of workflows with exception handling. --- see 'start' for details about exception and error handling -startWF - :: ( CMC.MonadCatchIO m, MonadIO m - , Serialize a, Serialize b - , Typeable a - , Indexable a) - => String -- ^ Name of workflow in the workflow list - -> a -- ^ Initial value (ever use the initial value even to restart the workflow) - -> WorkflowList m a b -- ^ function to execute - -> m (Either WFErrors b) -- ^ Result of the computation -startWF namewf v wfs= - case M.lookup namewf wfs of - Nothing -> return $ Left NotFound - Just f -> start namewf f v - - - --- | Re-start the non finished workflows in the list, for all the initial values that they may have been invoked -restartWorkflows - :: (Serialize a, Typeable a) - => M.Map String (a -> Workflow IO b) -- the list of workflows that implement the module - -> IO () -- Only workflows in the IO monad can be restarted with restartWorkflows -restartWorkflows map = do - mw <- liftIO $ getResource ((Running undefined ) ) -- :: IO (Maybe(Stat a)) - case mw of - Nothing -> return () - Just (Running all) -> mapM_ start . mapMaybe filter . toList $ all - where - filter (a, (b,Nothing)) = Just (b, a) - filter _ = Nothing - - start (key, kv)= do - let mf= M.lookup key map - case mf of - Nothing -> return () - Just f -> do - let st0= stat0{wfName = kv} - mst <- liftIO $ getResource st0 - case mst of - Nothing -> error $ "restartWorkflows: workflow not found "++ keyResource st0 - Just st-> do - liftIO . forkIO $ runWF key (f (fromIDyn . L.head $ versions st )) st{recover=True} >> return () - return () --- ei <- getState namewf f1 v --- case ei of --- Left error -> return $ Left error --- Right (name, f, stat) -> - - --- | Return all the steps of the workflow log. The values are dynamic --- --- to get all the steps with result of type Int: --- @all <- `getAll` --- let lfacts = mapMaybe `safeFromIDyn` all :: [Int]@ -getAll :: Monad m => Workflow m [IDynamic] -getAll= WF(\s -> return (s, versions s)) - ---getStep --- :: (Serialize a, Typeable a, Monad m) --- => Int -- ^ the step number. If negative, count from the current state backwards --- -> Workflow m a -- ^ return the n-tn intermediate step result ---getStep i= WF(\s -> do --- --- let stat= state s --- --- return (s, if i > 0 && i < stat then fromIDyn $ versions s !! (stat -i-1) --- else if i <= 0 && i > -stat then fromIDyn $ versions s !! (stat - ind +i-1) --- else error "getStep: wrong index") --- ) - --- | Return the keys of the workflows that are running with a given prefix -getWFKeys :: String -> IO [String] -getWFKeys wfname= do - mwfs <- atomically $ readDBRef tvRunningWfs - case mwfs of - Nothing -> return [] - Just (Running wfs) -> return $ Prelude.filter (L.isPrefixOf wfname) $ M.keys wfs - --- | Return the current state of the computation, in the IO monad -getWFHistory :: (Indexable a, Serialize a) => String -> a -> IO (Maybe Stat) -getWFHistory wfname x= getResource stat0{wfName= keyWF wfname x} - --- | Delete the history of a workflow. --- Be sure that this WF has finished. - ---{-# DEPRECATED delWFHistory, delWFHistory1 "use delWF instead" #-} - -delWFHistory name1 x = do - let name= keyWF name1 x - delWFHistory1 name - -delWFHistory1 name = do - let proto= stat0{wfName= name} --- when (isJust mdir) $ --- moveFile (defPath proto ++ key proto) (defPath proto ++ fromJust mdir) - atomically . withSTMResources [] $ const resources{ toDelete= [proto] } - --- | wait until the workflow is restarted -waitWFActive wf= do - r <- threadWF wf - case r of -- wait for change in the wofkflow state - Just (_, Nothing) -> retry - _ -> return () - where - threadWF wf= do - Just(Running map) <- readDBRef tvRunningWfs - return $ M.lookup wf map - - --- | Kill the executing thread if not killed, but not its state. --- `exec` `start` or `restartWorkflows` will continue the workflow -killThreadWF :: ( Indexable a - , Serialize a - - , Typeable a - , MonadIO m) - => String -> a -> m() -killThreadWF wfname x= do - let name= keyWF wfname x - killThreadWF1 name - --- | A version of `KillThreadWF` for workflows started wit no parameter by `exec1` -killThreadWF1 :: MonadIO m => String -> m() -killThreadWF1 name= killThreadWFm name >> return () - -killThreadWFm name= do - (map,f) <- clearRunningFlag name - case f of - Just th -> liftIO $ killThread th - Nothing -> return() - return map - - - --- | Kill the process (if running) and drop it from the list of --- restart-able workflows. Its state history remains , so it can be inspected with --- `getWfHistory` `printWFHistory` and so on. --- --- When the workflow has been called with no parameter, use: () --- -killWF :: (Indexable a,MonadIO m) => String -> a -> m () -killWF name1 x= do - let name= keyWF name1 x - killWF1 name - - -killWF1 :: MonadIO m => String -> m () -killWF1 name = do - map <- killThreadWFm name - liftIO . atomically . writeDBRef tvRunningWfs . Running $ M.delete name map - return () - --- | Delete the WF from the running list and delete the workflow state from persistent storage. --- Use it to perform cleanup if the process has been killed. --- --- When the workflow has been called with no parameter, use: () -delWF :: ( Indexable a - , MonadIO m - , Typeable a) - => String -> a -> m() -delWF name1 x= do - let name= keyWF name1 x - delWF1 name - - - -delWF1 :: MonadIO m => String -> m() -delWF1 name= liftIO $ atomicallySync $ do - mrun <- readDBRef tvRunningWfs - case mrun of - Nothing -> return() - Just (Running map) -> do - writeDBRef tvRunningWfs . Running $! M.delete name map - delDBRef (getDBRef $ keyResource $ stat0{wfName= name} :: DBRef Stat) - - -clearRunningFlag name= liftIO $ atomically $ do - Running map <- readDBRef tvRunningWfs `onNothing` error ( "clearRunningFLag: no workflow list" ++ name) - case M.lookup name map of - Just(_, Nothing) -> return (map,Nothing) - Just(v, Just th) -> do - writeDBRef tvRunningWfs . Running $ M.insert name (v, Nothing) map --- flushDBRef (getDBRef $ keyResource stat0{wfName=name} :: DBRef Stat) - return (map,Just th) - Nothing -> - return (map, Nothing) - - - - - --- | Log a value in the workflow log and return a reference to it. --- --- @newWFRef x= `stepWFRef` (return x) >>= return . fst@ -newWFRef :: ( Serialize a - , Typeable a - , MonadIO m - , CMC.MonadCatchIO m) - => a -> Workflow m (WFRef a) -newWFRef x= stepWFRef (return x) >>= return . fst - --- | Execute an step and return a reference to the result besides the result itself --- -stepWFRef :: ( Serialize a - , Typeable a - , MonadIO m) - => m a -> Workflow m (WFRef a,a) -stepWFRef exp= do - r <- step exp -- !> "stepWFRef" - WF(\s@Stat{..} -> do - let (n,flag)= if recover - then (state - (L.length versions) -1 ,False) - else (state -1 ,True) - ref = WFRef n self - s'= s{references= (n,(toIDyn r,flag)):references } - liftIO $ atomically $ writeDBRef self s' - r `seq` return (s',(ref,r)) ) - --- | return a reference to the last logged entry in the workflow --- In case the type of the reference is not of the type expected, it return an error string. ---getWFRef ::(Typeable b, Serialize b,MonadIO m) => Workflow m (Either String (WFRef b)) ---getWFRef= WF $ \s -> liftIO $ doit s --- where --- doit s@Stat{..}= do --- let (n,flag)= if recover --- then (state - (L.length versions ) -1 ,False) --- else (state - 1 ,True) --- --- mr= (safeFromIDyn $ versions !! n !> show n !> show state) `asTypeOf` typeOfRef (doit s) --- case mr `seq` mr of --- Left r -> return (s,Left r) --- Right r -> do --- let s'= s{references= (n,(toIDyn r,flag)):references } --- atomically $ writeDBRef self s' --- let ref = WFRef n self --- return (s,Right ref) --- where --- typeOfRef :: IO (Stat,Either String (WFRef a)) -> Either String a --- typeOfRef= undefined - ---getNRefs wfname= do --- st <- getResource stat0{wfName= wfname} `onNothing` error ("Workflow not found: "++ wfname) --- return $ L.length $ references st - --- |return a reference to the last step result ---getWFRef ::(MonadIO m,Serialize a, Typeable a) => Monad m => a -> Workflow m (WFRef a) ---getWFRef r = WF(\s@Stat{..} -> do --- let (n,flag)= if recover --- then (state - (L.length versions) -1 ,False) --- else (state -1 ,True) --- ref = WFRef n self --- s'= s{references= (n,(toIDyn r,flag)):references } --- liftIO $ atomically $ writeDBRef self s' --- r `seq` return (s',ref) ) - - --- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad -readWFRef :: ( Serialize a - , Typeable a) - => WFRef a - -> STM (Maybe a) -readWFRef (WFRef n ref)= do - mst <- readDBRef ref - case mst of - Nothing -> return Nothing - Just st -> do - case L.lookup n $! references st of - Just (r,_) -> return . Just $ fromIDyn r - Nothing -> do - let n1= if recover st then n else state st - n - return . Just . fromIDyn $ versions st !! n1 -- !> (show (L.length $ versions st) ++ " "++ show n1) - --- flushDBRef ref !> "readWFRef" --- st <- readDBRef ref `justifyM` (error $ "readWFRef: reference has been deleted from storaga: "++ show ref) - --- let elems= case ms of --- Just s -> versions s ++ (L.reverse $ L.take (state s' - state s) (versions s')) --- Nothing -> L.reverse $ versions s' --- x = elems !! n --- writeDBRef ref s' - --- return . Just $! fromIDyn x - - -justifyM io y= io >>= return . fromMaybe y - --- | Writes a new value en in the workflow reference, that is, in the workflow log. --- Why would you use this?. Don't do that!. modifiying the content of the workflow log would --- change the excution flow when the workflow restarts. This metod is used internally in the package. --- The best way to communicate with a workflow is trough a persistent queue, using "Data.Persistent.Collection": --- --- @worflow= exec1 "wf" do --- r <- `stepWFRef` expr --- `push` \"queue\" r --- back <- `pop` \"queueback\" --- ... --- @ - -writeWFRef :: ( Serialize a - , Typeable a) - => WFRef a - -> a - -> STM () -writeWFRef r@(WFRef n ref) x= do - mr <- readDBRef ref - case mr of - Nothing -> error $ "writeWFRef: workflow does not exist: " ++ show ref - Just st@Stat{..} -> - writeDBRef ref st{references= add x references} -- !> ("writeWFREF"++ show r) - - where - add x xs= (n,(toIDyn x,False)) : L.filter (\(n',_) -> n/=n') xs --- flushDBRef ref !> "writeWFRef" --- s <- safeIOToSTM $ readResourceByKey (keyObjDBRef ref) `justifyM` (error $ "writeWFRef: reference has been deleted from storaga: "++ show ref) --- let elems= versions s ++ (L.reverse $ L.take (state s' - state s) (versions s')) --- --- (h,t)= L.splitAt n elems --- elems'= h ++ (toIDyn x:tail' t) --- --- tail' []= [] --- tail' t = L.tail t - - - --- elems `seq` writeDBRef ref s{ versions= elems'} --- safeIOToSTM $ delResource s >> writeResource s{ versions= L.map tosave $ L.reverse elems'} --- writeDBRef ref s' - - --- | Moves the state of workflow with a seed value to become the state of other seed value --- This may be of interest when the entry value --- changes its key value but should not initiate a new workflow --- but continues with the current one - -moveState :: (MonadIO m - , Indexable a - , Serialize a - , Typeable a) - =>String -> a -> a -> m () -moveState wf t t'= liftIO $ do - atomicallySync $ do - mrun <- readDBRef tvRunningWfs - case mrun of - Nothing -> return() - Just (Running map) -> do - let mr= M.lookup n map - let th= case mr of Nothing -> Nothing; Just(_,mt)-> mt - let map'= M.insert n' (wf,th) $ M.delete n map - writeDBRef tvRunningWfs $ Running map' - withSTMResources[stat0{wfName= n}] $ change n - - where - n = keyWF wf t - n'= keyWF wf t' - change n [Nothing]= error $ "moveState: Workflow not found: "++ show n - change n [Just s] = resources{toAdd= [s{wfName=n' - ,versions = toIDyn t': L.tail( versions s) }] - ,toDelete=[s]} - - - - - --- | Log a message in the workflow history. I can be printed out with 'printWFhistory' --- The message is printed in the standard output too -logWF :: MonadIO m => String -> Workflow m () -logWF str=do - str <- step . liftIO $ do - time <- getClockTime >>= toCalendarTime >>= return . calendarTimeToString - Prelude.putStrLn str - return $ time ++ ": "++ str - WF $ \s -> str `seq` return (s, ()) - - - ---------- event handling-------------- - - --- | Wait until a TCache object (with a certaing key) meet a certain condition (useful to check external actions ) --- NOTE if anoter process delete the object from te cache, then waitForData will no longer work --- inside the wokflow, it can be used by lifting it : --- do --- x <- step $ .. --- y <- step $ waitForData ... --- .. - -waitForData :: (IResource a, Typeable a) - => (a -> Bool) -- ^ The condition that the retrieved object must meet - -> a -- ^ a partially defined object for which keyResource can be extracted - -> IO a -- ^ return the retrieved object that meet the condition and has the given key -waitForData f x = atomically $ waitForDataSTM f x - -waitForDataSTM :: (IResource a, Typeable a) - => (a -> Bool) -- ^ The condition that the retrieved object must meet - -> a -- ^ a partially defined object for which keyResource can be extracted - -> STM a -- ^ return the retrieved object that meet the condition and has the given key -waitForDataSTM filter x= do - tv <- newDBRef x - do - mx <- readDBRef tv >>= \v -> return $ cast v - case mx of - Nothing -> retry - Just x -> - case filter x of - False -> retry - True -> return x - --- | Observe the workflow log until a condition is met. -waitFor - :: ( Indexable a, Serialize a, Serialize b, Typeable a - , Indexable b, Typeable b) - => (b -> Bool) -- ^ The condition that the retrieved object must meet - -> String -- ^ The workflow name - -> a -- ^ the INITIAL value used in the workflow to start it - -> IO b -- ^ The first event that meet the condition -waitFor filter wfname x= atomically $ waitForSTM filter wfname x - -waitForSTM - :: ( Indexable a, Serialize a, Serialize b, Typeable a - , Indexable b, Typeable b) - => (b -> Bool) -- ^ The condition that the retrieved object must meet - -> String -- ^ The workflow name - -> a -- ^ The INITIAL value used in the workflow - -> STM b -- ^ The first event that meet the condition -waitForSTM filter wfname x= do - let name= keyWF wfname x - let tv= getDBRef . keyResource $ stat0{wfName= name} -- `debug` "**waitFor***" - - mmx <- readDBRef tv - case mmx of - Nothing -> error ("waitForSTM: Workflow does not exist: "++ name) - Just mx -> do - let Stat{ versions= d:_}= mx - case safeFromIDyn d of - Left _ -> retry -- `debug` "waithFor retry Nothing" - Right x -> - case filter x of - False -> retry -- `debug` "waitFor false filter retry" - True -> return x -- `debug` "waitfor return" - - - ---{-# DEPRECATED waitUntilSTM, getTimeoutFlag "use withTimeout instead" #-} - --- | Start the timeout and return the flag to be monitored by 'waitUntilSTM' --- This timeout is persistent. This means that the counter is initialized in the first call to getTimeoutFlag --- no matter if the workflow is restarted. The time during which the worlkflow has been stopped count also. --- Thus, the wait time can exceed the time between failures. --- when timeout is 0 means no timeout. -getTimeoutFlag - :: MonadIO m - => Integer -- ^ wait time in secods. This timing start from the first time that the timeout was started on. Sucessive restarts of the workflow will respect this timing - -> Workflow m (TVar Bool) -- ^ the returned flag in the workflow monad -getTimeoutFlag 0 = WF $ \s -> liftIO $ newTVarIO False >>= \tv -> return (s, tv) -getTimeoutFlag t = do - tnow <- step $ liftIO getTimeSeconds - flag tnow t - where - flag tnow delta = WF $ \s -> do - tv <- liftIO $ newTVarIO False - - liftIO $ do - let t = tnow + delta - atomically $ writeTVar tv False - forkIO $ do waitUntil t ; atomically $ writeTVar tv True - return (s, tv) - - - - -getTimeSeconds :: IO Integer -getTimeSeconds= do - TOD n _ <- getClockTime - return n - -{- | Wait until a certain clock time has passed by monitoring its flag, in the STM monad. - This permits to compose timeouts with locks waiting for data using `orElse` - - *example: wait for any respoinse from a Queue if no response is given in 5 minutes, it is returned True. - - @ - flag \<- 'getTimeoutFlag' $ 5 * 60 - ap \<- `step` . atomically $ readSomewhere >>= return . Just `orElse` 'waitUntilSTM' flag >> return Nothing - case ap of - Nothing -> do 'logWF' "timeout" ... - Just x -> do 'logWF' $ "received" ++ show x ... - @ --} - ---longWait :: Integer -> Workflow m a -> Workflow m a ---longWait time wf= --- WF $ \s -> do --- flag <- getTimeoutFlag time --- forkIO $ do --- atomically $ do --- b <- readTVar flag --- if b == False then retry else return () --- start (wfName s) wf "" --- myThreadId >>= killThread - - -waitUntilSTM :: TVar Bool -> STM() -waitUntilSTM tv = do - b <- readTVar tv - if b == False then retry else return () - --- | Wait until a certain clock time has passed by monitoring its flag, in the IO monad. --- See `waitUntilSTM` - -waitUntil:: Integer -> IO() -waitUntil t= getTimeSeconds >>= \tnow -> wait ((t-tnow)*1000000) - - -wait :: Integer -> IO() -wait delta= do - let delay | delta < 0= 0 - | delta > (fromIntegral maxInt) = maxInt - | otherwise = fromIntegral $ delta - threadDelay $ delay - if delta <= 0 then return () else wait $ delta - (fromIntegral delay ) - --- | Return either the result of the STM conputation or Nothing in case of timeout. --- The computation can retry --- This timeout is persistent. This means that the counter is initialized in the first call to getTimeoutFlag --- no matter if the workflow is restarted. The time during which the worlkflow has been stopped count also. --- Thus, the wait time can exceed the time between failures. --- when timeout is 0 it means no timeout. -withTimeout :: ( MonadIO m, Typeable a, Serialize a)=> Integer -> STM a -> Workflow m (Maybe a) -withTimeout time f = do - flag <- getTimeoutFlag time - step . liftIO . atomically $ (f >>= return . Just ) - `orElse` - (waitUntilSTM flag >> return Nothing) - - --- | Executes a computation understanding that it is inside the --- workflow identified by 'id'. If 'f' finish after 'time' --- it genetates a 'Timeout' exception which may result in the end of the workflow if the --- programmer does not catch it. --- If the workflow is restarted after 'time2' has elapsed, the workflow --- will restart from the beginning. If not, it will restart after the last logged step. --- --- Usually @time2> time@ --- --- @time2=0@ means @time2@ is infinite ---withKillTimeout :: CMC.MonadCatchIO m => String -> Int -> Integer -> m a -> m a ---withKillTimeout id time time2 f = do --- tid <- liftIO myThreadId --- tstart <- liftIO getTimeSeconds --- let final= liftIO $ do --- tnow <- getTimeSeconds --- let ref = getDBRef $ keyResource $ stat0{wfName=id} -- !> (keyResource $ stat0{wfName=id} ) --- when (time2 /=0) . atomically $ do --- s <- readDBRef ref `onNothing` error ( "withKillTimeout: Workflow not found: "++ id) --- writeDBRef ref s{lastActive= tnow,timeout= Just (time2 - fromIntegral (tnow - tstart))} --- clearRunningFlag id --- let proc= do --- twatchdog <- liftIO $ case time of --- 0 -> return tid --- _ -> forkIO $ threadDelay (time * 1000000) >> throwTo tid Timeout --- r <- f --- liftIO $ killThread twatchdog --- return r --- --- proc `CMC.finally` final - -withKillTimeout :: CMC.MonadCatchIO m => String -> Int -> Integer -> m a -> m a -withKillTimeout id time time2 f = do - tid <- liftIO myThreadId - twatchdog <- liftIO $ forkIO $ threadDelay (time * 1000000) >> throwTo tid Timeout - r <- f - liftIO $ killThread twatchdog - return r - `CMC.catch` \(e :: WFErrors) -> - case e of - Timeout -> liftIO $ do - - tnow <- getTimeSeconds - let ref = getDBRef $ keyResource $ stat0{wfName=id} -- !> (keyResource $ stat0{wfName=id} ) - when (time2 /=0) $ atomically $ do - s <- readDBRef ref `onNothing` error ( "withKillTimeout: Workflow not found: "++ id) - writeDBRef ref s{lastActive= tnow,timeout= Just (time2-fromIntegral time)} - syncCache - clearRunningFlag id - - throw Timeout -- !> "Timeout 2" - _ -> throw e - - - -transientTimeout 0= atomically $ newTVar False -transientTimeout t= do - flag <- atomically $ newTVar False - forkIO $ threadDelay (t * 1000000) >> atomically (writeTVar flag True) - return flag +{-# LANGUAGE OverlappingInstances + , UndecidableInstances + , ExistentialQuantification + , ScopedTypeVariables + , MultiParamTypeClasses + , FlexibleInstances + , FlexibleContexts + , TypeSynonymInstances + , DeriveDataTypeable + , RecordWildCards + , BangPatterns + #-} +{-# OPTIONS -IControl/Workflow #-} + + +{- | A workflow can be seen as a persistent thread. +The workflow monad writes a log that permit to restore the thread +at the interrupted point. `step` is the (partial) monad transformer for +the Workflow monad. A workflow is defined by its name and, optionally +by the key of the single parameter passed. There primitives for starting workflows +also restart the interrupted workflow when it has been in execution previously. + + +A small example that print the sequence of integers in te console +if you interrupt the progam, when restarted again, it will +start from the last printed number + +@module Main where +import Control.Workflow +import Control.Concurrent(threadDelay) +import System.IO (hFlush,stdout) + +mcount n= do `step` $ do + putStr (show n ++ \" \") + hFlush stdout + threadDelay 1000000 + mcount (n+1) + return () -- to disambiguate the return type + +main= `exec1` \"count\" $ mcount (0 :: Int)@ + +>>>runghc demos\sequence.hs +>0 1 2 3 +>CTRL-C Pressed +>>>runghc demos\sequence.hs +>3 4 5 6 7 +>CTRL-C Pressed +>>>runghc demos\sequence.hs +>7 8 9 10 11 +... + +The program restart at the last saved step. + +As you can see, some side effect can be re-executed after recovery if +the log is not complete. This may happen after an unexpected shutdown (in this case) +or due to an asynchronous log writing policy. (see `syncWrite`) + +When the step results are big and complex, use the "Data.RefSerialize" package to define the (de)serialization instances +The log size will be reduced. showHistory` method will print the structure changes +in each step. + +If instead of `RefSerialize`, you use read and show instances, there will + be no reduction. but still it will work, and the log will be readable for debugging purposes. + The RefSerialize istance is automatically derived from Read, Show instances. + +Data.Binary instances are also fine for serialization. To use Binary, just define a binary instance +of your data by using `showpBinary` and `readpBinary`. + +Within the RefSerialize instance of a structure, you can freely mix +Show,Read RefSerialize and Data Binary instances. + + + +"Control.Workflow.Patterns" contains higher level workflow patterns for handling multiple workflows + +"Control.Workflow.Configuration" permits the use of workflows for configuration purposes + +-} + +module Control.Workflow + +( + Stat +, Workflow -- a useful type name +, WorkflowList +, PMonadTrans (..) +, MonadCatchIO (..) +, HasFork(..) +, throw +, Indexable(..) +, keyWF +-- * Start/restart workflows +, start +, exec +, exec1d +, exec1 +, exec1nc +, wfExec +, startWF +, restartWorkflows +, WFErrors(..) +-- * Lifting to the Workflow monad +, step +, getWFStat +, stepExec +--, while +--, label +--, stepControl +--, stepDebug +, unsafeIOtoWF +-- * References to intermediate values in the workflow log +, WFRef +, newWFRef +, stepWFRef +, readWFRef +-- * State manipulation +, writeWFRef +, moveState +-- * Workflow inspection +, waitWFActive +, getAll +--, getStep +, safeFromIDyn +, getWFKeys +, getWFHistory +, waitFor +, waitForSTM +-- * Persistent timeouts +, waitUntilSTM +, getTimeoutFlag +, withTimeout +, withKillTimeout +-- * Trace logging +, logWF +-- * Termination of workflows +, clearRunningFlag +, killThreadWF +, killWF +, delWF +, killThreadWF1 +, delWFHistory +, delWFHistory1 +-- * Log writing policy +, syncWrite +, SyncMode(..) +-- * Print log history +, showHistory +, isInRecover +-- * Low leve, internal use +, runWF1 +, getState + +) + +where + +import Prelude hiding (catch) +import System.IO.Unsafe +import Control.Monad(when,liftM) +import Control.Applicative +import qualified Control.Exception as CE (Exception,AsyncException(ThreadKilled), SomeException, ErrorCall, throwIO, handle,finally,catch) +import Control.Concurrent -- (forkIO,threadDelay, ThreadId, myThreadId, killThread) +import Control.Concurrent.STM +import GHC.Conc(unsafeIOToSTM) +import GHC.Base (maxInt) + + +import Data.ByteString.Lazy.Char8 as B hiding (index) +import Data.ByteString.Lazy as BL(putStrLn) +import Data.List as L +import Data.Typeable +import System.Time +import Control.Monad.Trans +--import Control.Concurrent.MonadIO(HasFork(..),MVar,newMVar,takeMVar,putMVar,readMVar) + + +import System.IO(hPutStrLn, stderr) +import Data.List(elemIndex) +import Data.Maybe +import Data.IORef +import System.IO.Unsafe(unsafePerformIO) +import Data.Map as M(Map,fromList,elems, insert, delete, lookup,toList, fromList,keys) +import qualified Control.Monad.Catch as CMC +import qualified Control.Exception.Extensible as E + +import Data.TCache +import Data.TCache.Defs +import Data.RefSerialize +import Data.Persistent.IDynamic +import Unsafe.Coerce +import System.Mem.StableName +import Control.Workflow.Stat + +--import Debug.Trace +--(!>)= flip trace + +type Workflow m = WF Stat m -- not so scary + +type WorkflowList m a b= M.Map String (a -> Workflow m b) + +instance Monad m => Monad (WF s m) where + return x = WF (\s -> return (s, x)) + WF g >>= f = WF (\s -> do + (s1, x) <- g s + let WF fun= f x + fun s1) + + + +instance (Monad m,Functor m) => Functor (Workflow m ) where + fmap f (WF g)= WF (\s -> do + (s1, x) <- g s + return (s1, f x)) + +tvRunningWfs = getDBRef $ keyRunning :: DBRef Stat + + + +-- | Executes a computation inside of the workflow monad whatever the monad encapsulated in the workflow. +-- Warning: this computation is executed whenever +-- the workflow restarts, no matter if it has been already executed previously. This is useful for intializations or debugging. +-- To avoid re-execution when restarting use: @'step' $ unsafeIOtoWF...@ +-- +-- To perform IO actions in a workflow that encapsulates an IO monad, use step over the IO action directly: +-- +-- @ 'step' $ action @ +-- +-- instead of +-- +-- @ 'step' $ unsafeIOtoWF $ action @ +unsafeIOtoWF :: (Monad m) => IO a -> Workflow m a +unsafeIOtoWF x= let y= unsafePerformIO ( x >>= return) in y `seq` return y + + +{- | @PMonadTrans@ permits |to define a partial monad transformer. They are not defined for all kinds of data +but the ones that have instances of certain classes.That is because in the lift instance code there are some +hidden use of these classes. This also may permit an accurate control of effects. +An instance of MonadTrans is an instance of PMonadTrans +-} +class PMonadTrans t m a where + plift :: Monad m => m a -> t m a + + + +-- | @plift= step@ +instance (Monad m + , MonadIO m + , Serialize a + , Typeable a) + => PMonadTrans (WF Stat) m a + where + plift = step + +instance (Monad m, Functor m) => Applicative (Workflow m) where + pure x= return x + WF f <*> WF g= WF $ \s -> do + (s1, k) <- f s + (s2, x) <- g s1 + return (s2,k x) + + + +-- | An instance of MonadTrans is an instance of PMonadTrans +instance (MonadTrans t, Monad m) => PMonadTrans t m a where + plift= Control.Monad.Trans.lift + +--- | Handle with care: this instance will force +-- the execution at recovery of every liftted IO procedure +-- better use 'step . liftIO' instead of 'liftIO' + + +instance MonadIO m => MonadIO (WF Stat m) where + liftIO= unsafeIOtoWF + + +{- | Adapted from the @MonadCatchIO-mtl@ package. However, in this case it is needed to express serializable constraints about the returned values, +so the usual class definitions for lifting IO functions are not suitable. +-} + +class MonadCatchIO m a where + -- | Generalized version of 'E.catch' + catch :: E.Exception e => m a -> (e -> m a) -> m a + + -- | Generalized version of 'E.block' + block :: m a -> m a + + -- | Generalized version of 'E.unblock' + unblock :: m a -> m a + + + +-- | Generalized version of 'E.throwIO' +throw :: (MonadIO m, E.Exception e) => e -> m a +throw = liftIO . E.throwIO + + + + + +--instance (Serialize a +-- , Typeable a,MonadIO m, CMC.MonadCatch m) +-- => MonadCatchIO (WF Stat m) a where +-- catch exp exc = do +-- expwf <- step $ getTempName +-- excwf <- step $ getTempName +-- step $ do +-- ex <- CMC.catch (exec1d expwf exp >>= return . Right ) $ \e-> return $ Left e +-- +-- case ex of +-- Right r -> return r -- All right +-- Left e ->exec1d excwf (exc e) +-- -- An exception occured in the main workflow +-- -- the exception workflow is executed +-- +-- +-- +-- +-- block exp=WF $ \s -> CMC.block (st exp $ s) +---- +-- unblock exp= WF $ \s -> CMC.unblock (st exp $ s) + +data WFInfo= WFInfo{ name :: String + , finished :: Bool + , haserror :: Maybe WFErrors } + deriving (Typeable,Read, Show) + +class MonadIO io => HasFork io where + fork :: io () -> io ThreadId + +instance HasFork IO where + fork= forkIO + +instance (HasFork io, MonadIO io + , CMC.MonadCatch io) + => HasFork (WF Stat io) where + fork f = do + (r,info@(WFInfo str finished status)) <- stepWFRef $ getTempName >>= \n -> return(WFInfo n False Nothing) + + WF $ \s -> do + th <- if finished then fork $ return() + else + fork $ + exec1 str f >> labelFinish r str Nothing + `CMC.catch` \(E.ErrorCall str) -> do + liftIO . atomicallySync $ writeWFRef r (WFInfo str True (Just $ WFException str)) -- !> ("ERROR *****"++show e) + killWF1 $ keyWF str () + `CMC.catch` \(e :: E.SomeException) -> do + liftIO . atomicallySync $ writeWFRef r (WFInfo str True (Just . WFException $ show e)) -- !> ("ERROR *****"++show e) + killWF1 $ keyWF str () + + + return (s,th) + where + labelFinish r str err= liftIO . atomicallySync $ writeWFRef r (WFInfo str True err) -- !> "finished" + + +-- | Start or restart an anonymous workflow inside another workflow. +-- Its state is deleted when finished and the result is stored in +-- the parent's WF state. +wfExec + :: (Serialize a, Typeable a + , CMC.MonadCatch m, MonadIO m) + => Workflow m a -> Workflow m a +wfExec f= do + str <- step $ getTempName + step $ exec1 str f + +-- | A version of exec1 that deletes its state after complete execution or thread killed +exec1d :: (MonadIO m, CMC.MonadCatch m) + => String -> (Workflow m b) -> m b +exec1d str f= do + r <- exec1 str f + delit + return r + `CMC.catch` (\e@CE.ThreadKilled -> delit >> throw e) + + where + delit= do + delWF str () + + + + +-- | A version of exec with no seed parameter. +exec1 :: ( Monad m, MonadIO m, CMC.MonadCatch m) + => String -> Workflow m a -> m a + +exec1 str f= exec str (const f) () + + + + +-- | Start or continue a workflow with exception handling +-- the workflow flags are updated even in case of exception +-- `WFerrors` are raised as exceptions +exec :: ( Indexable a, Serialize a, Typeable a + , Monad m, MonadIO m, CMC.MonadCatch m) + => String -> (a -> Workflow m b) -> a -> m b +exec str f x = + (do + v <- getState str f x + case v of + Right (name, f, stat) -> do + r <- runWF name (f x) stat + return r + Left err -> CMC.throwM err) + `CMC.catch` + (\(e :: CE.SomeException) -> liftIO $ do + let name= keyWF str x + clearRunningFlag name --`debug` ("exception"++ show e) + + CMC.throwM e ) + +-- | executes a workflow, but does not mark it as finished even if +-- the process ended. +-- It this case, the workflow just will return the last result. +-- If the workflow was gathering data from user questions for a configuration, then this +-- primitive will store them in the log the first time, and can be retrieve it the next time. +exec1nc :: ( Monad m, MonadIO m, CMC.MonadCatch m) + => String -> Workflow m a -> m a +exec1nc str f =do + v <- getState str f () + case v of + Left err -> CMC.throwM err + Right (name, f, stat) -> do + runWF1 name f stat False + + `CMC.catch` + (\(e :: CE.SomeException) -> liftIO $ do + let name= keyWF str () + clearRunningFlag name --`debug` ("exception"++ show e) + CMC.throwM e ) + `CMC.finally` + (liftIO . atomically . + when(recover stat) $ do + let ref= self stat + s <- readDBRef ref `justifyM` error ("step: not found: "++ wfName stat) + writeDBRef ref s{recover= False,versions=L.reverse $ versions s}) + +mv :: MVar Int +mv= unsafePerformIO $ newMVar 0 + +getTempName :: MonadIO m => m String +getTempName= liftIO $ do + seq <- takeMVar mv + putMVar mv (seq + 1) + TOD t _ <- getClockTime + return $ "anon"++ show t ++ show seq + + + + +-- Permits the modification of the workflow state by the procedure being lifted +-- if the boolean value is True. This is used internally for control purposes +--stepControl :: ( Monad m +-- , MonadIO m +-- , Serialize a +-- , Typeable a) +-- => m a +-- -> Workflow m a +--stepControl= stepControl1 True + + +-- | Lifts a monadic computation to the WF monad, and provides transparent state loging and resuming the computation +-- Note: Side effect can be repeated at recovery time if the log was not complete before shut down +-- see the integer sequence example, above. +step :: ( MonadIO m + , Serialize a + , Typeable a) + => m a + -> Workflow m a + +step mx= WF(\s -> do + let + recovers= recover s + versionss= versions s +-- !> "vvvvvvvvvvvvvvvvvvv" +-- !> (unpack $ runW $ showp $ versions s) +-- !> (show $ references s) +-- !> (show $ "recover="++ show( recover s)) +-- !> "^^^^^^^^^^^^^^^^^^^" + if recovers && not (L.null versionss) + then + return (s{versions=L.tail versionss }, fromIDyn $ L.head versionss ) + else do + let ref= self s + when (recovers && L.null versionss) $ do + liftIO $ atomically $ do + s' <- readDBRef ref `justifyM` error ("step: not found: "++ wfName s) + writeDBRef ref s'{recover= False,references= references s} + stepExec1 ref mx) + +getWFStat :: Monad m => Workflow m (DBRef Stat) +getWFStat= WF $ \s -> return (s,self s) + +stepExec + :: (Typeable t, Serialize t, MonadIO m) => + DBRef Stat -> m t -> m (DBRef Stat, t) +stepExec ref mx= do + (s,x) <- stepExec1 ref mx + return (self s, x) + +stepExec1 sref mx= do + x' <- mx + liftIO . atomicallySync $ do + s <- readDBRef sref >>= return . fromMaybe (error $ "step: readDBRef: not found:" ++ keyObjDBRef sref) + let versionss= versions s + dynx= toIDyn x' + ver= dynx: versionss + s'= s{ recover= False, versions = ver, state= state s+1} + writeDBRef sref s' + return (s', x') + +--undoStep :: Monad m => Workflow m () +--undoStep= WF $ \s@Stat{..} -> return(s{state=state-1, versions= L.tail versions},()) + +-- | True if the workflow in recovery mode, reading the log to recover the process state +isInRecover :: Monad m => Workflow m Bool +isInRecover = WF(\s@Stat{..} -> + if recover && not (L.null versions ) then return(s,True ) + else if recover== True then return(s{recover=False}, False) + else return (s,False)) + +-- | For debugging purposes. +-- At recovery time, instead of returning the stored value from the log +-- , stepDebug executes the computation 'f' as normally. +-- . It permits the exact re-execution of a workflow process +stepDebug :: ( Monad m + , MonadIO m + , Serialize a + , Typeable a) + => m a + -> Workflow m a +stepDebug f = r + where + r= do + WF(\s -> + let stat= state s + + + in case recover s && not(L.null $ versions s) of + True -> f >>= \x -> return (s{versions= L.tail $ versions s},x) + False -> stepExec1 (self s) f) + +-- Executes a computation 'f' in a loop while the return value meets the condition 'cond' is met. +-- At recovery time, the current state of the loop is restored. +-- The loop restart at the last internal state that was (saved) before shutdown. +-- +-- The use of 'while' permits a faster recovery when the loop has many steps and the log is very long, as is the case in +-- MFlow applications, +--while +-- :: MonadIO m => +-- (b -> Bool) -> Workflow m b -> Workflow m b +--while cond f= do +-- n <- WF $ \s -> return (s,state s - L.length (versions s)) +---- do +---- let versionss= versions s +---- if recover s && not (L.null versionss) +---- then return (s{versions=L.tail versionss }, fromIDyn $ L.head versionss ) +---- +---- else return(s{recover= False, state=state s + 1 +---- ,versions= (toIDyn $ state s):versionss} +---- ,state s) +-- while1 n +-- where +-- while1 n =do +-- label n +-- x <- f +-- if cond x +-- then while1 n +-- else return x +-- +--data Label= Label Int deriving (Eq,Typeable,Read,Show) +--label n = do +-- let !label= Label n +-- r <- isInRecover +-- if r +-- then WF(\s@Stat{..} -> +-- let !label@(Label n) = fromIDyn $ L.head versions +-- !vers = filterMax (\d -> Just label /= safeFromIDyn d) versions -- !> (show label) +-- in return (s{versions= L.tail vers}, fromIDyn . L.head $ vers )) +-- else do +-- step $ return label +-- where +-- filterMax f xs= +-- case L.dropWhile f (L.tail xs) of +-- [] -> xs +-- [_] -> xs +-- xs' -> filterMax f xs' +-- + + + +-- | Start or continue a workflow . +-- 'WFErrors' and exceptions are returned as @Left err@ (even if they were triggered as exceptions). +-- Other exceptions are returned as @Left (Exception e)@ +-- use `killWF` or `delWF` in case of error to clear the log. +start + :: ( CMC.MonadCatch m + , MonadIO m + , Indexable a + , Serialize a + , Typeable a) + => String -- ^ name that identifies the workflow. + -> (a -> Workflow m b) -- ^ workflow to execute + -> a -- ^ initial value (ever use the initial value for restarting the workflow) + -> m (Either WFErrors b) -- ^ result of the computation +start namewf f1 v = do + ei <- getState namewf f1 v + case ei of + Left error -> return $ Left error + Right (name, f, stat) -> + runWF name (f v) stat >>= return . Right + `CMC.catch` + (\(e :: WFErrors) -> do + let name= keyWF namewf v + clearRunningFlag name + return $ Left e ) + `CMC.catch` + (\(E.ErrorCall msg) ->do + let name= keyWF namewf v + clearRunningFlag name + return . Left $ WFException msg ) + `CMC.catch` + (\(e :: CE.SomeException) -> liftIO $ do + let name= keyWF namewf v + clearRunningFlag name + return . Left $ WFException $ show e ) + + + + +-- | Return conditions from the invocation of start/restart primitives +data WFErrors = NotFound | AlreadyRunning | Timeout | WFException String deriving (Typeable, Read, Show) + +--instance Show WFErrors where +-- show NotFound= "Not Found" +-- show AlreadyRunning= "Already Running" +-- show Timeout= "Timeout" +-- show (Exception e)= "Exception: "++ show e + +--instance Serialize WFErrors where +-- showp NotFound= insertString "NotFound" +-- showp AlreadyRunning= insertString "AlreadyRunning" +-- showp Timeout= insertString "Timeout" +-- showp (Exception e)= insertString "Exception: ">> showp e +-- +-- readp= choice[notfound,already,timeout, exc] +-- where +-- notfound= symbol "NotFound" >> return NotFound +-- already= symbol "AlreadyRunning" >> return AlreadyRunning +-- timeout= symbol "Timeout" >> return Timeout +-- exc= symbol "Exception" >> readp >>= \s -> return (Exception s) + +instance CE.Exception WFErrors + + + +{- +lookup for any workflow for the entry value v +if namewf is found and is running, return arlready running + if is not runing, restart it +else start anew. +-} + + +getState :: (Monad m, MonadIO m, Indexable a, Serialize a, Typeable a) + => String -> x -> a + -> m (Either WFErrors (String, x, Stat)) +getState namewf f v= liftIO . atomically $ getStateSTM + where + getStateSTM = do + mrunning <- readDBRef tvRunningWfs + case mrunning of + Nothing -> do + writeDBRef tvRunningWfs (Running $ fromList []) + getStateSTM + Just(Running map) -> do + let key= keyWF namewf v + dynv= toIDyn v + stat1= stat0{wfName= key,versions=[dynv],state=1, self= sref`seq`sref} + sref= getDBRef $ keyResource stat1 + case M.lookup key map of + Nothing -> do -- no workflow started for this object + mythread <- unsafeIOToSTM $ myThreadId + safeIOToSTM $ delResource stat1 >> writeResource stat1 + writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map + writeDBRef sref stat1 + return $ Right (key, f, stat1) -- !> "NEW WF" + + Just (wf, started) -> -- a workflow has been initiated for this object + if isJust started + then return $ Left AlreadyRunning -- !> "already running" + else do + mst <- readDBRef sref -- !> "has been running but not running now" + stat' <- case mst of + Nothing -> return stat1 -- error $ "getState: Workflow not found: "++ key + Just s' -> do + -- the thread may have been killed by an exception when running + s <- case recover s' of + True -> return s' + False -> do + s'' <- safeIOToSTM $ readResource s' `onNothing` return stat1 + let i= state s'' + j= state s' + return s'{versions= versions s'' ++ L.reverse ( L.take ( j - i) $ versions s')} + if isJust (timeout s) + then do + tnow <- unsafeIOToSTM getTimeSeconds + if lastActive s+ fromJust(timeout s) > tnow -- !>("lastActive="++show (lastActive s) ++ "tnow="++show tnow) + then + return s{recover= True,timeout=Nothing} + else + -- has been inactive for too much time, clean it + return stat1 + + else return s{recover= True} + + + writeDBRef sref stat' + mythread <- unsafeIOToSTM myThreadId + writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map + + return $ Right (key, f, stat') + + + +runWF :: ( Monad m, MonadIO m) + => String -> Workflow m b -> Stat -> m b +runWF n f s = runWF1 n f s True + + + +runWF1 n f s clear= do + (s', v') <- st f s{versions= L.tail $ versions s} + liftIO $ if clear then clearFromRunningList n + else clearRunningFlag n >> return () + return v' + where + + -- eliminate the thread from the list of running workflows but leave the state + clearFromRunningList n = atomicallySync $ do + Just(Running map) <- readDBRef tvRunningWfs -- !> "clearFormRunning" + writeDBRef tvRunningWfs . Running $ M.delete n map +-- flushDBRef (getDBRef n :: DBRef Stat) + +-- | Start or continue a workflow from a list of workflows with exception handling. +-- see 'start' for details about exception and error handling +startWF + :: ( CMC.MonadCatch m, MonadIO m + , Serialize a, Serialize b + , Typeable a + , Indexable a) + => String -- ^ Name of workflow in the workflow list + -> a -- ^ Initial value (ever use the initial value even to restart the workflow) + -> WorkflowList m a b -- ^ function to execute + -> m (Either WFErrors b) -- ^ Result of the computation +startWF namewf v wfs= + case M.lookup namewf wfs of + Nothing -> return $ Left NotFound + Just f -> start namewf f v + + + +-- | Re-start the non finished workflows in the list, for all the initial values +-- that they may have been invoked. The list contain he identifiers of the workflows and +-- the procedures to be called. All the workflows initiated with exec* or start* will be +-- restarted with all possible seed values. + +restartWorkflows + :: (Serialize a, Typeable a) + => M.Map String (a -> Workflow IO b) -- the list of workflows that implement the module + -> IO () -- Only workflows in the IO monad can be restarted with restartWorkflows +restartWorkflows map = do + mw <- atomically $ readDBRef tvRunningWfs -- :: IO (Maybe(Stat a)) + case mw of + Nothing -> return () + Just (Running all) -> mapM_ start . mapMaybe filter . toList $ all + where + filter (a, (b,Nothing)) = Just (b, a) + filter _ = Nothing + + start (key, kv)= do + let mf= M.lookup key map + case mf of + Nothing -> return () + Just f -> do + let st0 = stat0{wfName = kv} + mst <- liftIO $ getResource st0 + case mst of + Nothing -> error $ "restartWorkflows: workflow not found "++ keyResource st0 + Just st-> do + liftIO . forkIO $ runWF key (f (fromIDyn . L.head $ versions st )) st{recover=True} >> return () + return () +-- ei <- getState namewf f1 v +-- case ei of +-- Left error -> return $ Left error +-- Right (name, f, stat) -> + + +-- | Return all the steps of the workflow log. The values are dynamic +-- +-- to get all the steps with result of type Int: +-- @all <- `getAll` +-- let lfacts = mapMaybe `safeFromIDyn` all :: [Int]@ +getAll :: Monad m => Workflow m [IDynamic] +getAll= WF(\s -> return (s, versions s)) + +--getStep +-- :: (Serialize a, Typeable a, Monad m) +-- => Int -- ^ the step number. If negative, count from the current state backwards +-- -> Workflow m a -- ^ return the n-tn intermediate step result +--getStep i= WF(\s -> do +-- +-- let stat= state s +-- +-- return (s, if i > 0 && i < stat then fromIDyn $ versions s !! (stat -i-1) +-- else if i <= 0 && i > -stat then fromIDyn $ versions s !! (stat - ind +i-1) +-- else error "getStep: wrong index") +-- ) + +-- | Return the keys of the workflows that are running with a given prefix +getWFKeys :: String -> IO [String] +getWFKeys wfname= do + mwfs <- atomically $ readDBRef tvRunningWfs + case mwfs of + Nothing -> return [] + Just (Running wfs) -> return $ Prelude.filter (L.isPrefixOf wfname) $ M.keys wfs + +-- | Return the current state of the computation, in the IO monad +getWFHistory :: (Indexable a, Serialize a) => String -> a -> IO (Maybe Stat) +getWFHistory wfname x= getResource stat0{wfName= keyWF wfname x} + +-- | Delete the history of a workflow. +-- Be sure that this WF has finished. + +--{-# DEPRECATED delWFHistory, delWFHistory1 "use delWF instead" #-} + +delWFHistory name1 x = do + let name= keyWF name1 x + delWFHistory1 name + +delWFHistory1 name = do + let proto= stat0{wfName= name} +-- when (isJust mdir) $ +-- moveFile (defPath proto ++ key proto) (defPath proto ++ fromJust mdir) + atomically . withSTMResources [] $ const resources{ toDelete= [proto] } + +-- | wait until the workflow is restarted +waitWFActive wf= do + r <- threadWF wf + case r of -- wait for change in the wofkflow state + Just (_, Nothing) -> retry + _ -> return () + where + threadWF wf= do + Just(Running map) <- readDBRef tvRunningWfs + return $ M.lookup wf map + + +-- | Kill the executing thread if not killed, but not its state. +-- `exec` `start` or `restartWorkflows` will continue the workflow +killThreadWF :: ( Indexable a + , Serialize a + + , Typeable a + , MonadIO m) + => String -> a -> m() +killThreadWF wfname x= do + let name= keyWF wfname x + killThreadWF1 name + +-- | A version of `KillThreadWF` for workflows started wit no parameter by `exec1` +killThreadWF1 :: MonadIO m => String -> m() +killThreadWF1 name= killThreadWFm name >> return () + +killThreadWFm name= do + (map,f) <- clearRunningFlag name + case f of + Just th -> liftIO $ killThread th + Nothing -> return() + return map + + + +-- | Kill the process (if running) and drop it from the list of +-- restart-able workflows. Its state history remains , so it can be inspected with +-- `getWfHistory` `showHistory` and so on. +-- +-- When the workflow has been called with no parameter, use: () +-- +killWF :: (Indexable a,MonadIO m) => String -> a -> m () +killWF name1 x= do + let name= keyWF name1 x + killWF1 name + + +killWF1 :: MonadIO m => String -> m () +killWF1 name = do + map <- killThreadWFm name + liftIO . atomically . writeDBRef tvRunningWfs . Running $ M.delete name map + return () + +-- | Delete the WF from the running list and delete the workflow state from persistent storage. +-- Use it to perform cleanup if the process has been killed. +-- +-- When the workflow has been called with no parameter, use: () +delWF :: ( Indexable a + , MonadIO m + , Typeable a) + => String -> a -> m() +delWF name1 x= do + let name= keyWF name1 x + delWF1 name + + + +delWF1 :: MonadIO m => String -> m() +delWF1 name= liftIO $ atomicallySync $ do + mrun <- readDBRef tvRunningWfs + case mrun of + Nothing -> return() + Just (Running map) -> do + writeDBRef tvRunningWfs . Running $! M.delete name map + delDBRef (getDBRef $ keyResource $ stat0{wfName= name} :: DBRef Stat) + + +clearRunningFlag name= liftIO $ atomically $ do + Running map <- readDBRef tvRunningWfs `onNothing` error ( "clearRunningFLag: no workflow list" ++ name) + case M.lookup name map of + Just(_, Nothing) -> return (map,Nothing) + Just(v, Just th) -> do + writeDBRef tvRunningWfs . Running $ M.insert name (v, Nothing) map +-- flushDBRef (getDBRef $ keyResource stat0{wfName=name} :: DBRef Stat) + return (map,Just th) + Nothing -> + return (map, Nothing) + + + + + +-- | Log a value in the workflow log and return a reference to it. +-- +-- @newWFRef x= `stepWFRef` (return x) >>= return . fst@ +newWFRef :: ( Serialize a + , Typeable a + , MonadIO m + , CMC.MonadCatch m) + => a -> Workflow m (WFRef a) +newWFRef x= stepWFRef (return x) >>= return . fst + +-- | Execute an step and return a reference to the result besides the result itself +-- +stepWFRef :: ( Serialize a + , Typeable a + , MonadIO m) + => m a -> Workflow m (WFRef a,a) +stepWFRef exp= do + r <- step exp -- !> "stepWFRef" + WF(\s@Stat{..} -> do + let (n,flag)= if recover + then (state - (L.length versions) -1 ,False) + else (state -1 ,True) + ref = WFRef n self + s'= s{references= (n,(toIDyn r,flag)):references } + liftIO $ atomically $ writeDBRef self s' + r `seq` return (s',(ref,r)) ) + +-- | return a reference to the last logged entry in the workflow +-- In case the type of the reference is not of the type expected, it return an error string. +--getWFRef ::(Typeable b, Serialize b,MonadIO m) => Workflow m (Either String (WFRef b)) +--getWFRef= WF $ \s -> liftIO $ doit s +-- where +-- doit s@Stat{..}= do +-- let (n,flag)= if recover +-- then (state - (L.length versions ) -1 ,False) +-- else (state - 1 ,True) +-- +-- mr= (safeFromIDyn $ versions !! n !> show n !> show state) `asTypeOf` typeOfRef (doit s) +-- case mr `seq` mr of +-- Left r -> return (s,Left r) +-- Right r -> do +-- let s'= s{references= (n,(toIDyn r,flag)):references } +-- atomically $ writeDBRef self s' +-- let ref = WFRef n self +-- return (s,Right ref) +-- where +-- typeOfRef :: IO (Stat,Either String (WFRef a)) -> Either String a +-- typeOfRef= undefined + +--getNRefs wfname= do +-- st <- getResource stat0{wfName= wfname} `onNothing` error ("Workflow not found: "++ wfname) +-- return $ L.length $ references st + +-- |return a reference to the last step result +--getWFRef ::(MonadIO m,Serialize a, Typeable a) => Monad m => a -> Workflow m (WFRef a) +--getWFRef r = WF(\s@Stat{..} -> do +-- let (n,flag)= if recover +-- then (state - (L.length versions) -1 ,False) +-- else (state -1 ,True) +-- ref = WFRef n self +-- s'= s{references= (n,(toIDyn r,flag)):references } +-- liftIO $ atomically $ writeDBRef self s' +-- r `seq` return (s',ref) ) + + +-- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad +readWFRef :: ( Serialize a + , Typeable a) + => WFRef a + -> STM (Maybe a) +readWFRef (WFRef n ref)= do + mst <- readDBRef ref + case mst of + Nothing -> return Nothing + Just st -> do + case L.lookup n $! references st of + Just (r,_) -> return . Just $ fromIDyn r + Nothing -> do + let n1= if recover st then n else state st - n + return . Just . fromIDyn $ versions st !! n1 -- !> (show (L.length $ versions st) ++ " "++ show n1) + +-- flushDBRef ref !> "readWFRef" +-- st <- readDBRef ref `justifyM` (error $ "readWFRef: reference has been deleted from storaga: "++ show ref) + +-- let elems= case ms of +-- Just s -> versions s ++ (L.reverse $ L.take (state s' - state s) (versions s')) +-- Nothing -> L.reverse $ versions s' +-- x = elems !! n +-- writeDBRef ref s' + +-- return . Just $! fromIDyn x + + +justifyM io y= io >>= return . fromMaybe y + +-- | Writes a new value en in the workflow reference, that is, in the workflow log. +-- Why would you use this?. Don't do that!. modifiying the content of the workflow log would +-- change the excution flow when the workflow restarts. This metod is used internally in the package. +-- The best way to communicate with a workflow is trough a persistent queue, using "Data.Persistent.Collection": +-- +-- @worflow= exec1 "wf" do +-- r <- `stepWFRef` expr +-- `push` \"queue\" r +-- back <- `pop` \"queueback\" +-- ... +-- @ + +writeWFRef :: ( Serialize a + , Typeable a) + => WFRef a + -> a + -> STM () +writeWFRef r@(WFRef n ref) x= do + mr <- readDBRef ref + case mr of + Nothing -> error $ "writeWFRef: workflow does not exist: " ++ show ref + Just st@Stat{..} -> + writeDBRef ref st{references= add x references} -- !> ("writeWFREF"++ show r) + + where + add x xs= (n,(toIDyn x,False)) : L.filter (\(n',_) -> n/=n') xs +-- flushDBRef ref !> "writeWFRef" +-- s <- safeIOToSTM $ readResourceByKey (keyObjDBRef ref) `justifyM` (error $ "writeWFRef: reference has been deleted from storaga: "++ show ref) +-- let elems= versions s ++ (L.reverse $ L.take (state s' - state s) (versions s')) +-- +-- (h,t)= L.splitAt n elems +-- elems'= h ++ (toIDyn x:tail' t) +-- +-- tail' []= [] +-- tail' t = L.tail t + + + +-- elems `seq` writeDBRef ref s{ versions= elems'} +-- safeIOToSTM $ delResource s >> writeResource s{ versions= L.map tosave $ L.reverse elems'} +-- writeDBRef ref s' + + +-- | Moves the state of workflow with a seed value to become the state of other seed value +-- This may be of interest when the entry value +-- changes its key value but should not initiate a new workflow +-- but continues with the current one + +moveState :: (MonadIO m + , Indexable a + , Serialize a + , Typeable a) + =>String -> a -> a -> m () +moveState wf t t'= liftIO $ do + atomicallySync $ do + mrun <- readDBRef tvRunningWfs + case mrun of + Nothing -> return() + Just (Running map) -> do + let mr= M.lookup n map + let th= case mr of Nothing -> Nothing; Just(_,mt)-> mt + let map'= M.insert n' (wf,th) $ M.delete n map + writeDBRef tvRunningWfs $ Running map' + withSTMResources[stat0{wfName= n}] $ change n + + where + n = keyWF wf t + n'= keyWF wf t' + change n [Nothing]= error $ "moveState: Workflow not found: "++ show n + change n [Just s] = resources{toAdd= [s{wfName=n' + ,versions = toIDyn t': L.tail( versions s) }] + ,toDelete=[s]} + + + + + +-- | Log a message in the workflow history. I can be printed out with 'showHistory' +-- The message is printed in the standard output too +logWF :: MonadIO m => String -> Workflow m () +logWF str= do + str <- step . liftIO $ do + time <- getClockTime >>= toCalendarTime >>= return . calendarTimeToString + Prelude.putStrLn str + return $ time ++ ": "++ str + WF $ \s -> str `seq` return (s, ()) + + + +--------- event handling-------------- + + +-- | Wait until a TCache object (with a certaing key) meet a certain condition (useful to check external actions ) +-- NOTE if anoter process delete the object from te cache, then waitForData will no longer work +-- inside the wokflow, it can be used by lifting it : +-- do +-- x <- step $ .. +-- y <- step $ waitForData ... +-- .. + +waitForData :: (IResource a, Typeable a) + => (a -> Bool) -- ^ The condition that the retrieved object must meet + -> a -- ^ a partially defined object for which keyResource can be extracted + -> IO a -- ^ return the retrieved object that meet the condition and has the given key +waitForData f x = atomically $ waitForDataSTM f x + +waitForDataSTM :: (IResource a, Typeable a) + => (a -> Bool) -- ^ The condition that the retrieved object must meet + -> a -- ^ a partially defined object for which keyResource can be extracted + -> STM a -- ^ return the retrieved object that meet the condition and has the given key +waitForDataSTM filter x= do + tv <- newDBRef x + do + mx <- readDBRef tv >>= \v -> return $ cast v + case mx of + Nothing -> retry + Just x -> + case filter x of + False -> retry + True -> return x + +-- | Observe the workflow log until a condition is met. +waitFor + :: ( Indexable a, Serialize a, Serialize b, Typeable a + , Indexable b, Typeable b) + => (b -> Bool) -- ^ The condition that the retrieved object must meet + -> String -- ^ The workflow name + -> a -- ^ the INITIAL value used in the workflow to start it + -> IO b -- ^ The first event that meet the condition +waitFor filter wfname x= atomically $ waitForSTM filter wfname x + +waitForSTM + :: ( Indexable a, Serialize a, Serialize b, Typeable a + , Indexable b, Typeable b) + => (b -> Bool) -- ^ The condition that the retrieved object must meet + -> String -- ^ The workflow name + -> a -- ^ The INITIAL value used in the workflow + -> STM b -- ^ The first event that meet the condition +waitForSTM filter wfname x= do + let name= keyWF wfname x + let tv= getDBRef . keyResource $ stat0{wfName= name} -- `debug` "**waitFor***" + + mmx <- readDBRef tv + case mmx of + Nothing -> error ("waitForSTM: Workflow does not exist: "++ name) + Just mx -> do + let Stat{ versions= d:_}= mx + case safeFromIDyn d of + Left _ -> retry -- `debug` "waithFor retry Nothing" + Right x -> + case filter x of + False -> retry -- `debug` "waitFor false filter retry" + True -> return x -- `debug` "waitfor return" + + + +--{-# DEPRECATED waitUntilSTM, getTimeoutFlag "use withTimeout instead" #-} + +-- | Start the timeout and return the flag to be monitored by 'waitUntilSTM' +-- This timeout is persistent. This means that the counter is initialized in the first call to getTimeoutFlag +-- no matter if the workflow is restarted. The time during which the worlkflow has been stopped count also. +-- Thus, the wait time can exceed the time between failures. +-- when timeout is 0 means no timeout. +getTimeoutFlag + :: MonadIO m + => Integer -- ^ wait time in secods. This timing start from the first time that the timeout was started on. Sucessive restarts of the workflow will respect this timing + -> Workflow m (TVar Bool) -- ^ the returned flag in the workflow monad +getTimeoutFlag 0 = WF $ \s -> liftIO $ newTVarIO False >>= \tv -> return (s, tv) +getTimeoutFlag t = do + tnow <- step $ liftIO getTimeSeconds + flag tnow t + where + flag tnow delta = WF $ \s -> do + tv <- liftIO $ newTVarIO False + + liftIO $ do + let t = tnow + delta + atomically $ writeTVar tv False + forkIO $ do waitUntil t ; atomically $ writeTVar tv True + return (s, tv) + + + + +getTimeSeconds :: IO Integer +getTimeSeconds= do + TOD n _ <- getClockTime + return n + +{- | Wait until a certain clock time has passed by monitoring its flag, in the STM monad. + This permits to compose timeouts with locks waiting for data using `orElse` + + *example: wait for any respoinse from a Queue if no response is given in 5 minutes, it is returned True. + + @ + flag \<- 'getTimeoutFlag' $ 5 * 60 + ap \<- `step` . atomically $ readSomewhere >>= return . Just `orElse` 'waitUntilSTM' flag >> return Nothing + case ap of + Nothing -> do 'logWF' "timeout" ... + Just x -> do 'logWF' $ "received" ++ show x ... + @ +-} + +--longWait :: Integer -> Workflow m a -> Workflow m a +--longWait time wf= +-- WF $ \s -> do +-- flag <- getTimeoutFlag time +-- forkIO $ do +-- atomically $ do +-- b <- readTVar flag +-- if b == False then retry else return () +-- start (wfName s) wf "" +-- myThreadId >>= killThread + + +waitUntilSTM :: TVar Bool -> STM() +waitUntilSTM tv = do + b <- readTVar tv + if b == False then retry else return () + +-- | Wait until a certain clock time has passed by monitoring its flag, in the IO monad. +-- See `waitUntilSTM` + +waitUntil:: Integer -> IO() +waitUntil t= getTimeSeconds >>= \tnow -> wait ((t-tnow)*1000000) + + +wait :: Integer -> IO() +wait delta= do + let delay | delta < 0= 0 + | delta > (fromIntegral maxInt) = maxInt + | otherwise = fromIntegral $ delta + threadDelay $ delay + if delta <= 0 then return () else wait $ delta - (fromIntegral delay ) + +-- | Return either the result of the STM conputation or Nothing in case of timeout. +-- The computation can retry +-- This timeout is persistent. This means that the counter is initialized in the first call to getTimeoutFlag +-- no matter if the workflow is restarted. The time during which the worlkflow has been stopped count also. +-- Thus, the wait time can exceed the time between failures. +-- when timeout is 0 it means no timeout. +withTimeout :: ( MonadIO m, Typeable a, Serialize a)=> Integer -> STM a -> Workflow m (Maybe a) +withTimeout time f = do + flag <- getTimeoutFlag time + step . liftIO . atomically $ (f >>= return . Just ) + `orElse` + (waitUntilSTM flag >> return Nothing) + + +-- | Executes a computation understanding that it is inside the +-- workflow identified by 'id'. If 'f' finish after 'time' +-- it genetates a 'Timeout' exception which may result in the end of the workflow if the +-- programmer does not catch it. +-- If the workflow is restarted after 'time2' has elapsed, the workflow +-- will restart from the beginning. If not, it will restart after the last logged step. +-- +-- Usually @time2> time@ +-- +-- @time2=0@ means @time2@ is infinite +--withKillTimeout :: CMC.MonadCatchIO m => String -> Int -> Integer -> m a -> m a +--withKillTimeout id time time2 f = do +-- tid <- liftIO myThreadId +-- tstart <- liftIO getTimeSeconds +-- let final= liftIO $ do +-- tnow <- getTimeSeconds +-- let ref = getDBRef $ keyResource $ stat0{wfName=id} -- !> (keyResource $ stat0{wfName=id} ) +-- when (time2 /=0) . atomically $ do +-- s <- readDBRef ref `onNothing` error ( "withKillTimeout: Workflow not found: "++ id) +-- writeDBRef ref s{lastActive= tnow,timeout= Just (time2 - fromIntegral (tnow - tstart))} +-- clearRunningFlag id +-- let proc= do +-- twatchdog <- liftIO $ case time of +-- 0 -> return tid +-- _ -> forkIO $ threadDelay (time * 1000000) >> throwTo tid Timeout +-- r <- f +-- liftIO $ killThread twatchdog +-- return r +-- +-- proc `CMC.finally` final + +withKillTimeout :: (MonadIO m,CMC.MonadCatch m) => String -> Int -> Integer -> m a -> m a +withKillTimeout id time time2 f = do + tid <- liftIO myThreadId + twatchdog <- liftIO $ forkIO $ threadDelay (time * 1000000) >> throwTo tid Timeout + r <- f + liftIO $ killThread twatchdog + return r + `CMC.catch` \(e :: WFErrors) -> + case e of + Timeout -> liftIO $ do + + tnow <- getTimeSeconds + let ref = getDBRef $ keyResource $ stat0{wfName=id} -- !> (keyResource $ stat0{wfName=id} ) + when (time2 /=0) $ atomically $ do + s <- readDBRef ref `onNothing` error ( "withKillTimeout: Workflow not found: "++ id) + writeDBRef ref s{lastActive= tnow,timeout= Just (time2-fromIntegral time)} + syncCache + clearRunningFlag id + + throw Timeout -- !> "Timeout 2" + _ -> throw e + + + +transientTimeout 0= atomically $ newTVar False +transientTimeout t= do + flag <- atomically $ newTVar False + forkIO $ threadDelay (t * 1000000) >> atomically (writeTVar flag True) + return flag diff --git a/Control/Workflow/Configuration.hs b/Control/Workflow/Configuration.hs --- a/Control/Workflow/Configuration.hs +++ b/Control/Workflow/Configuration.hs @@ -16,7 +16,7 @@ import Data.RefSerialize import Control.Monad.Trans import Control.Exception -import Control.Monad.CatchIO as CMC +import Control.Monad.Catch as CMC -------------- configuation -- | to execute a computation every time it is invoked. A synonimous of `unsafeIOtoWF` @@ -29,7 +29,7 @@ -- | executes a computation with `once` and `ever` statements -- a synonym of `exec1nc` -runConfiguration :: ( Monad m, MonadIO m, CMC.MonadCatchIO m) +runConfiguration :: ( Monad m, MonadIO m, CMC.MonadCatch m) => String -> Workflow m a -> m a runConfiguration = exec1nc diff --git a/Control/Workflow/Patterns.hs b/Control/Workflow/Patterns.hs --- a/Control/Workflow/Patterns.hs +++ b/Control/Workflow/Patterns.hs @@ -78,7 +78,7 @@ import Control.Concurrent.STM import Data.Monoid -import qualified Control.Monad.CatchIO as CMC +import qualified Control.Monad.Catch as CMC import Control.Workflow.Stat import Control.Workflow @@ -102,7 +102,7 @@ split :: ( Typeable b , Serialize b , HasFork io - , CMC.MonadCatchIO io) + , CMC.MonadCatch io) => [a -> Workflow io b] -> a -> Workflow io [ActionWF b] split actions a = mapM (\ac -> do @@ -161,7 +161,7 @@ -- , Serialize [a] , Typeable a , HasFork io - , CMC.MonadCatchIO io) + , CMC.MonadCatch io) => Integer -> (a -> STM Select) -> [ActionWF a] @@ -187,7 +187,7 @@ addRes i r unsafeIOToSTM $ throwTo parent FinishDiscard - n <- liftIO $ CMC.block $ do + n <- liftIO $ do -- liftIO $ CMC.block $ do n <- takeMVar count putMVar count (n+1) return n -- !> ("SELECT" ++ show n) @@ -241,7 +241,7 @@ -- , Serialize [b] , Typeable b , HasFork io - , CMC.MonadCatchIO io) + , CMC.MonadCatch io) => Integer -> [a -> Workflow io b] -> ([b] -> Workflow io c) @@ -260,7 +260,7 @@ , Typeable b , Monoid b , HasFork io - , CMC.MonadCatchIO io) + , CMC.MonadCatch io) => Integer -> [a -> Workflow io b] -> a @@ -275,7 +275,7 @@ r <- exec1 "sumup" $ sumUp 0 [f 1, f 2] "0" print r - `CMC.catch` \(e::SomeException) -> syncCache -- !> "syncCache" + `CMC.catch` \(e:: SomeException) -> syncCache -- !> "syncCache" f :: Int -> String -> Workflow IO String diff --git a/Workflow.cabal b/Workflow.cabal --- a/Workflow.cabal +++ b/Workflow.cabal @@ -1,5 +1,5 @@ name: Workflow -version: 0.8.0.8 +version: 0.8.0.9 cabal-version: >= 1.6 build-type: Simple license: BSD3 @@ -38,7 +38,7 @@ library - build-depends: MonadCatchIO-transformers -any, RefSerialize ==0.3.1.* , + build-depends: exceptions -any, RefSerialize ==0.3.1.* , TCache -any && <1.0, base >=4 && <5, binary -any, bytestring -any, containers -any, directory -any, extensible-exceptions -any, mtl -any, old-time -any, stm >2, vector -any