Workflow 0.6.0.0 → 0.7.0.0
raw patch · 11 files changed
+1029/−989 lines, 11 filesdep +directorydep −monadIOdep −transformersdep −vectordep ~TCache
Dependencies added: directory
Dependencies removed: monadIO, transformers, vector
Dependency ranges changed: TCache
Files
- Control/Workflow.hs +534/−354
- Control/Workflow/Configuration.hs +34/−0
- Control/Workflow/IDynamic.hs +38/−127
- Control/Workflow/Patterns.hs +85/−58
- Control/Workflow/Stat.hs +241/−76
- Data/Persistent/Queue.hs +0/−240
- Demos/Fact.hs +1/−1
- Demos/docAprobal.hs +25/−18
- Demos/pr.hs +30/−13
- Demos/sequence.hs +9/−6
- Workflow.cabal +32/−96
Control/Workflow.hs view
@@ -7,40 +7,29 @@ , 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. The primitives for starting workflows-also restart the workflow when it has been in execution previously.+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. -This is the main module that uses the `RefSerialize` paclkage for serialization. Here the constraint @DynSerializer w r a@ is equivalent to-@Data.RefSerialize a@ -For workflows that uses big structures, for example, documents-use this module in combination with the 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 define read and show instances, there will- be no reduction. but still the log will be readable for debugging purposes.--for workflows that does not care about this, use the binary alternative: "Control.Workflow.Binary"- 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.Text+import Control.Workflow import Control.Concurrent(threadDelay) import System.IO (hFlush,stdout) - mcount n= do `step` $ do putStr (show n ++ \" \") hFlush stdout@@ -50,16 +39,55 @@ 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 higuer level workflow patters of multiuser workflows++"Control.Workflow.Configuration" permits the use of workflows for configuration purposes+ -} module Control.Workflow+ ( Workflow -- a useful type name , WorkflowList , PMonadTrans (..) , MonadCatchIO (..)+, HasFork(..) , throw , Indexable(..)+, keyWF -- * Start/restart workflows , start , exec@@ -71,18 +99,23 @@ , WFErrors(..) -- * Lifting to the Workflow monad , step-, stepControl+--, while+--, label+--, stepControl+--, stepDebug , unsafeIOtoWF -- * References to intermediate values in the workflow log , WFRef-, getWFRef , newWFRef , stepWFRef , readWFRef+-- * State manipulation , writeWFRef+, moveState -- * Workflow inspect , waitWFActive , getAll+--, getStep , safeFromIDyn , getWFKeys , getWFHistory@@ -91,6 +124,8 @@ -- * Persistent timeouts , waitUntilSTM , getTimeoutFlag+, withTimeout+, withKillTimeout -- * Trace logging , logWF -- * Termination of workflows@@ -107,15 +142,17 @@ , syncWrite , SyncMode(..) -- * Print log history-, printHistory+, showHistory+, isInRecover )+ where import Prelude hiding (catch) import System.IO.Unsafe import Control.Monad(when,liftM)-import qualified Control.Exception as CE (Exception,AsyncException(ThreadKilled), SomeException, throwIO, handle,finally,catch,block,unblock)-import Control.Concurrent (forkIO,threadDelay, ThreadId, myThreadId, killThread)+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)@@ -127,12 +164,12 @@ import Data.Typeable import System.Time import Control.Monad.Trans-import Control.Concurrent.MonadIO(HasFork(..),MVar,newMVar,takeMVar,putMVar)+--import Control.Concurrent.MonadIO(HasFork(..),MVar,newMVar,takeMVar,putMVar,readMVar) import System.IO(hPutStrLn, stderr) import Data.List(elemIndex)-import Data.Maybe(fromJust, isNothing, isJust, mapMaybe)+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)@@ -140,19 +177,20 @@ import qualified Control.Exception.Extensible as E import Data.TCache-import Data.TCache.DefaultPersistence+import Data.TCache.Defs import Data.RefSerialize import Control.Workflow.IDynamic import Unsafe.Coerce+import System.Mem.StableName import Control.Workflow.Stat---+ --import Debug.Trace --a !> b= trace b a type Workflow m = WF Stat m -- not so scary -type WorkflowList m a b= [(String, a -> Workflow m b) ]+type WorkflowList m a b= M.Map String (a -> Workflow m b) instance Monad m => Monad (WF s m) where@@ -160,8 +198,7 @@ WF g >>= f = WF (\s -> do (s1, x) <- g s let WF fun= f x- (s3, x') <- fun s1- return (s3, x'))+ fun s1) @@ -174,7 +211,7 @@ --- | executes a computation inside of the workflow monad whatever the monad encapsulated in the workflow.+-- | 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...@@@ -190,7 +227,7 @@ 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+{- | @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@@ -200,7 +237,7 @@ --- | plift= step+-- | @plift= step@ instance (Monad m , MonadIO m , Serialize a@@ -213,11 +250,14 @@ instance (MonadTrans t, Monad m) => PMonadTrans t m a where plift= Control.Monad.Trans.lift -instance Monad m => MonadIO (WF Stat m) where- liftIO=unsafeIOtoWF+--- | Handle with care: this instance will force+-- the unwanted 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 MonadCatchIO-mtl. Workflow need to express serializable constraints about the returned values,+{- | adapted from the @MonadCatchIO-mtl@ package. However, in tis case is needed to express serializable constraints about the returned values, so the usual class definitions for lifting IO functions are not suitable. -} @@ -237,98 +277,10 @@ throw :: (MonadIO m, E.Exception e) => e -> m a throw = liftIO . E.throwIO -{- -{---- | Generalized version of 'E.try'-try :: (MonadCatchIO m a, E.Exception e) => m a -> m (Either e a) --- | Generalized version of 'E.tryJust'-tryJust :: (MonadCatchIO m a, E.Exception e)- => (e -> Maybe b) -> m a -> m (Either b a) --}--- | Generalized version of 'E.Handler'-data Handler m a = forall e . E.Exception e => Handler (e -> m a) --{--instance (MonadCatchIO m a, Error e) => MonadCatchIO (ErrorT e m) a where- m `catch` f = mapErrorT (\m' -> m' `catch` (\e -> runErrorT $ f e)) m- block = mapErrorT block- unblock = mapErrorT unblock--}----try a = catch (a >>= \ v -> return (Right v)) (\e -> return (Left e))--tryJust p a = do- r <- try a- case r of- Right v -> return (Right v)- Left e -> case p e of- Nothing -> throw e `asTypeOf` (return $ Left undefined)- Just b -> return (Left b)---- | Generalized version of 'E.bracket'-bracket :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c) => m a -> (a -> m b) -> (a -> m c) -> m c-bracket before after thing =- block (do a <- before- r <- unblock (thing a) `onException` after a- _void $ after a- return r)---- | A variant of 'bracket' where the return value from the first computation--- is not required.-bracket_ :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c)- => m a -- ^ computation to run first (\"acquire resource\")- -> m b -- ^ computation to run last (\"release resource\")- -> m c -- ^ computation to run in-between- -> m c -- returns the value from the in-between computation-bracket_ before after thing =- block $ do _void before- r <- unblock thing `onException` after- _void after- return r---- | A specialised variant of 'bracket' with just a computation to run--- afterward.-finally :: (Monad m, MonadIO m, MonadCatchIO m a)- => m a -- ^ computation to run first- -> m b -- ^ computation to run afterward (even if an exception was- -- raised)- -> m a -- returns the value from the first computation-thing `finally` after =- block $ do r <- unblock thing `onException` after- _void after- return r-{---- | Like 'bracket', but only performs the final action if there was an--- exception raised by the in-between computation.-bracketOnError :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c)- => m a -- ^ computation to run first (\"acqexeuire resource\")- -> (a -> m b)-- ^ computation to run last (\"release resource\")- -> (a -> m c)-- ^ computation to run in-between- -> m c -- returns the value from the in-between- -- computation-bracketOnError before after thing =- block $ do a <- before- unblock (thing a) `onException` after a--}--- | Generalized version of 'E.onException'-onException :: (MonadIO m, MonadCatchIO m a) => m a -> m b -> m a-onException a onEx = a `catch` (\e -> onEx >> throw (e:: E.SomeException))--_void :: Monad m => m a -> m ()-_void a = a >> return ()----}---- instance (Serialize a , Typeable a,MonadIO m, CMC.MonadCatchIO m) => MonadCatchIO (WF Stat m) a where@@ -352,28 +304,41 @@ 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+instance HasFork IO where+ fork= forkIO++instance (HasFork io, MonadIO io , CMC.MonadCatchIO io)+ => HasFork (WF Stat io) where fork f = do- (str, finished) <- step $ getTempName >>= \n -> return(n, False)- r <- getWFRef- WF (\s ->- do th <- if finished- then fork $ return ()- else fork $ do- exec1 str f- liftIO $ do atomically $ writeWFRef r (str, True)- syncIt- return(s,th))+ (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 :: 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++-- | 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 :: (Indexable a, Serialize a, Typeable a@@ -383,9 +348,9 @@ str <- step $ getTempName step $ exec1 str f --- | a version of exec1 that deletes its state after complete execution or thread killed+-- | A version of exec1 that deletes its state after complete execution or thread killed exec1d :: (Serialize b, Typeable b- ,CMC.MonadCatchIO m)+ ,MonadIO m, CMC.MonadCatchIO m) => String -> (Workflow m b) -> m b exec1d str f= do r <- exec1 str f@@ -396,11 +361,11 @@ where delit= do delWF str ()- liftIO syncIt -- !> str --- | a version of exec with no seed parameter.++-- | A version of exec with no seed parameter. exec1 :: ( Serialize a, Typeable a , Monad m, MonadIO m, CMC.MonadCatchIO m) => String -> Workflow m a -> m a@@ -410,9 +375,9 @@ --- | start or continue a workflow with exception handling--- | the workflow flags are updated even in case of exception--- | `WFerrors` are raised as exceptions+-- | 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, Serialize b, Typeable a , Typeable b , Monad m, MonadIO m, CMC.MonadCatchIO m)@@ -429,7 +394,7 @@ (\(e :: CE.SomeException) -> liftIO $ do let name= keyWF str x clearRunningFlag name --`debug` ("exception"++ show e)- syncIt+ CMC.throw e ) @@ -448,69 +413,145 @@ +-- 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 -instance Indexable () where- key= show---- | lifts a monadic computation to the WF monad, and provides transparent state loging and resuming of computation+-- | 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 :: ( Monad m , MonadIO m , Serialize a , Typeable a) => m a -> Workflow m a-step= stepControl1 False --- | permits 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+step mx= WF(\s -> do+ let stat= state s+ 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 -stepControl1 :: ( Monad m+ liftIO $ atomically $ do+ s' <- readDBRef ref `justifyM` error ("step: not found: "++ wfName s)+ writeDBRef ref s'{recover= False,references= references s}+ stepExec ref mx)++stepExec 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')++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)- => Bool -> m a+ => m a -> Workflow m a-stepControl1 isControl mx= WF(\s'' -> do- let stat= state s''- let ind= index s''- if recover s'' && ind < stat- then return (s''{index=ind +1 }, fromIDyn $ versions s'' !! (stat - ind-1) )- else do- x' <- mx- let sref = self s''- s'<- liftIO . atomically $ do- s <- if isControl- then readDBRef sref >>= unjustify ("step: readDBRef: not found:" ++ keyObjDBRef sref)- else return s''- let versionss= versions s- let dynx= toIDyn x'- let ver= dynx: versionss- let s'= s{ recover= False, versions = ver, state= state s+1}+stepDebug f = r+ where+ r= do+ WF(\s ->+ let stat= state s - writeDBRef sref s'- return s'- liftIO syncIt- return (s', x') ) -unjustify str Nothing = error str-unjustify _ (Just x) = return x+ in case recover s && not(L.null $ versions s) of+ True -> f >>= \x -> return (s{versions= L.tail $ versions s},x)+ False -> stepExec (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 with no exception handling.--- | the programmer has to handle inconsistencies in the workflow state--- | using `killWF` or `delWF` in case of exception.+-- | 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 erro to clear the log. start- :: ( Monad m+ :: ( CMC.MonadCatchIO m , MonadIO m , Indexable a , Serialize a, Serialize b@@ -521,26 +562,49 @@ -> 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- Right (name, f, stat) ->- runWF name (f v) stat >>= return . Right+ 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 :: CE.SomeException) -> liftIO $ do+ let name= keyWF namewf v+ clearRunningFlag name+ return . Left $ WFException $ show e ) - Left error -> return $ Left error --- | return conditions from the invocation of start/restart primitives-data WFErrors = NotFound | AlreadyRunning | Timeout | forall e.CE.Exception e => Exception e deriving Typeable+-- | 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 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 ---tvRunningWfs = unsafePerformIO . refDBRefIO $ Running (M.fromList [] :: Map String (String, (Maybe ThreadId))) + {- lookup for any workflow for the entry value v if namewf is found and is running, return arlready running@@ -558,73 +622,86 @@ mrunning <- readDBRef tvRunningWfs case mrunning of Nothing -> do- writeDBRef tvRunningWfs (Running $ fromList [])- getStateSTM+ writeDBRef tvRunningWfs (Running $ fromList [])+ getStateSTM Just(Running map) -> do let key= keyWF namewf v- stat1= stat0{wfName= key,versions=[toIDyn v],self= sref}+ 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) Just (wf, started) -> -- a workflow has been initiated for this object if isJust started- then return $ Left AlreadyRunning -- `debug` "already running"- else do -- has been started but not running now- mythread <- unsafeIOToSTM $ myThreadId- writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map+ then return $ Left AlreadyRunning -- !> "already running"+ else do -- has been running but not running now mst <- readDBRef sref- let stat' = case mst of- Nothing -> error $ "Workflow not found: "++ key- Just s -> s{index=0,recover= True}+ stat' <- case mst of+ Nothing -> error $ "getState: Workflow not found: "++ key+ Just s -> do+ -- the thread may have been killed by an exception when running+ when(not $ recover s) $ error ("flow "++key++ "found in a wrong state: report it")+ 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') -syncIt= do- (sync,_) <- atomically $ readTVar tvSyncWrite- when (sync ==Synchronous) syncCache + runWF :: (Monad m,MonadIO m , Serialize b, Typeable b) => String -> Workflow m b -> Stat -> m b runWF n f s= do- sync <- liftIO $! do- (sync,_) <- atomically $ readTVar tvSyncWrite- when (sync ==Synchronous) syncCache- return sync- (s', v') <- st f $ s- liftIO $! do- clearFromRunningList n- when (sync ==Synchronous) syncCache+ (s', v') <- st f s{versions= L.tail $ versions s} -- !> (show $ versions s)+ liftIO $! clearFromRunningList n return v' where -- eliminate the thread from the list of running workflows but leave the state- clearFromRunningList n = atomically $ do- Just(Running map) <- readDBRef tvRunningWfs+ clearFromRunningList n = atomicallySync $ do+ Just(Running map) <- readDBRef tvRunningWfs -- !> "clearFormRunning" writeDBRef tvRunningWfs . Running $ M.delete n map -- `debug` "clearFromRunningList"---- | start or continue a workflow from a list of workflows in the IO monad with exception handling. The excepton is returned as a Left value+ 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- :: ( MonadIO m+ :: ( CMC.MonadCatchIO m, MonadIO m , Serialize a, Serialize b , Typeable a , Indexable a , Typeable b)- => 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+ => 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 Prelude.lookup namewf wfs of+ 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 called+++-- | Re-start the non finished workflows in the list, for all the initial values that they may have been invoked restartWorkflows :: (Serialize a, Serialize b, Typeable a , Indexable b, Typeable b)@@ -640,9 +717,7 @@ filter _ = Nothing start (key, kv)= do-- --let key1= key ++ "#" ++ kv- let mf= Prelude.lookup key map+ let mf= M.lookup key map case mf of Nothing -> return () Just f -> do@@ -651,48 +726,15 @@ case mst of Nothing -> error $ "restartWorkflows: workflow not found "++ keyResource st0 Just st-> do- liftIO . forkIO $ runWF key (f (fromIDyn . Prelude.last $ versions st )) st{index=0,recover=True} >> return ()+ liftIO . forkIO $ runWF key (f (fromIDyn . L.head $ versions st )) st{recover=True} >> return () return ()---- | choose between text and binary persistence for the workflow state--- text persistence is used for---- *(1)debugging purposes---- * (2)when step returns largue structures that share common contents between steps,--- for example, when a workflow edit and ammend a document among many users---- * (3) When tracking the modifications made in the object trough `getWFHistory` or--- `printWFHistory`------ |--- The execution log is cached in memory using the package `TCache`. This procedure defines the polcy for writing the cache into permanent storage.------ For fast workflows, or when TCache` is used also for other purposes , `Asynchronous` is the best option------ `Asynchronous` mode invokes `clearSyncCache`. For more complex use of the syncronization--- please use this `clearSyncCache`.------ When interruptions are controlled, use `SyncManual` mode and include a call to `syncCache` in the finalizaton code--syncWrite:: (Monad m, MonadIO m) => SyncMode -> m ()-syncWrite mode= do- (_,thread) <- liftIO . atomically $ readTVar tvSyncWrite- when (isJust thread ) $ liftIO . killThread . fromJust $ thread- case mode of- Synchronous -> modeWrite- SyncManual -> modeWrite- Asyncronous time maxsize -> do- th <- liftIO $ clearSyncCacheProc time defaultCheck maxsize >> return()- liftIO . atomically $ writeTVar tvSyncWrite (mode,Just th)- where- modeWrite=- liftIO . atomically $ writeTVar tvSyncWrite (mode, Nothing)+-- 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+-- | Return all the steps of the workflow log. The values are dynamic -- -- to get all the steps with result of type Int: -- @all <- `getAll`@@ -700,8 +742,20 @@ 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 ind= index s+-- 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 list of object keys that are running for a workflow+-- | Return the list of object keys that are running for a workflow getWFKeys :: String -> IO [String] getWFKeys wfname= do mwfs <- atomically $ readDBRef tvRunningWfs@@ -709,17 +763,24 @@ 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+-- | 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. -delWFHistory name1 x= do+--{-# DEPRECATED delWFHistory, delWFHistory1 "use delWF and delWF1 instead" #-}++delWFHistory name1 x = do let name= keyWF name1 x delWFHistory1 name -delWFHistory1 name =- atomically . withSTMResources [] $ const resources{ toDelete= [stat0{wfName= 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] } waitWFActive wf= do@@ -733,7 +794,7 @@ return $ M.lookup wf map --- | kill the executing thread if not killed, but not its state.+-- | Kill the executing thread if not killed, but not its state. -- `exec` `start` or `restartWorkflows` will continue the workflow killThreadWF :: ( Indexable a , Serialize a@@ -745,7 +806,7 @@ let name= keyWF wfname x killThreadWF1 name --- | a version of `KillThreadWF` for workflows started wit no parameter by `exec1`+-- | A version of `KillThreadWF` for workflows started wit no parameter by `exec1` killThreadWF1 :: MonadIO m => String -> m() killThreadWF1 name= killThreadWFm name >> return () @@ -758,7 +819,7 @@ --- | kill the process (if running) and drop it from the list of+-- | 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 killWF :: (Indexable a,MonadIO m) => String -> a -> m ()@@ -766,14 +827,14 @@ let name= keyWF name1 x killWF1 name --- | a version of `KillWF` for workflows started wit no parameter by `exec1`+-- | A version of `KillWF` for workflows started wit no parameter by `exec1` 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.+-- | 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. delWF :: ( Indexable a , MonadIO m@@ -784,93 +845,126 @@ delWF1 name --- | a version of `delWF` for workflows started wit no parameter by `exec1`+-- | A version of `delWF` for workflows started wit no parameter by `exec1` delWF1 :: MonadIO m=> String -> m() delWF1 name= liftIO $ do mrun <- atomically $ readDBRef tvRunningWfs case mrun of Nothing -> return() Just (Running map) -> do- atomically . writeDBRef tvRunningWfs . Running $! M.delete name map+ atomicallySync . writeDBRef tvRunningWfs . Running $! M.delete name map delWFHistory1 name- syncIt + clearRunningFlag name= liftIO $ atomically $ do mrun <- readDBRef tvRunningWfs case mrun of- Nothing -> error $ "clearRunningFLag non existing workflows" ++ name+ Nothing -> error $ "clearRunningFLag: non existing workflows" ++ name Just(Running map) -> do 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) + -- | Return the reference to the last logged result , usually, the last result stored by `step`. -- wiorkflow references can be accessed outside of the workflow -- . They also can be (de)serialized. -- -- WARNING getWFRef can produce casting errors when the type demanded -- do not match the serialized data. Instead, `newDBRef` and `stepWFRef` are type safe at runtuime.-getWFRef :: ( Monad m,- MonadIO m,- Serialize a- , Typeable a)- => Workflow m (WFRef a)-getWFRef =ret- where- ret= WF (\s -> do- let n= if recover s then index s else state s- let ref = WFRef n (self s)- -- to reify the object being accessed- -- if not reified, the serializer will write a null object- let r= fromIDyn (versions s !! (state s - n)) `asTypeOf` typeofRef ret- r `seq` return (s,ref))- where- typeofRef :: Workflow m (WFRef a) -> a- typeofRef= undefined -- never will be executed--- | Execute an step but return a reference to the result instead of the result itself------ @stepWFRef exp= `step` exp >>= `getWFRef`@-stepWFRef :: ( Serialize a- , Typeable a- , MonadIO m)- => m a -> Workflow m (WFRef a)-stepWFRef exp= step exp >> getWFRef+--getWFRef :: ( Monad m,+-- MonadIO m,+-- Serialize a+-- , Typeable a)+-- => Workflow m (WFRef a)+--getWFRef =ret !> "geWFRef"+-- where+-- ret= WF (\s -> do+-- let n= if recover s then state s - (L.length $ versions s)+-- else (state s -1)+-- let ref = WFRef n (self s)+-- -- to reify the object being accessed+-- -- if not reified, the serializer will write a null object+-- let versionss= versions s+-- when ( L.null versionss) $ error "getWFRef: empty log, no step to point to"+-- let r= fromIDyn (L.head $ versionss) `asTypeOf` typeofRef ret+-- r `seq` return (s,ref))+-- where+-- typeofRef :: Workflow m (WFRef a) -> a+-- typeofRef= undefined -- never will be executed + -- | Log a value and return a reference to it. -- -- @newWFRef x= `step` $ return x >>= `getWFRef`@ newWFRef :: ( Serialize a , Typeable a- , MonadIO m)+ , MonadIO m+ , CMC.MonadCatchIO m) => a -> Workflow m (WFRef a)-newWFRef x= step (return x) >> getWFRef+newWFRef x= stepWFRef (return x) >>= return . fst +-- | Execute an step but 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)+ let ref = WFRef n self+ let s'= s{references= (n,(toIDyn r,flag)):references }+ liftIO $ atomically $ writeDBRef self s'+ r `seq` return (s',(ref,r)) )++ -- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad-readWFRef :: ( Serialize a- , Typeable a)+readWFRef :: ( Serialize a+ , Typeable a) => WFRef a -> STM (Maybe a) readWFRef (WFRef n ref)= do- mr <- readDBRef ref- case mr of+ mr <- readDBRef ref+ case mr of Nothing -> return Nothing- Just s -> do- let elems= versions s- l = state s -- L.length elems- x = elems !! (l - n)- return . Just $! fromIDyn x+ Just st ->+ case L.lookup n $! references st of+ Just (r,_) -> return . Just $ fromIDyn r+ Nothing -> do+ let n1= state st - n+ return . Just . fromIDyn $ versions st !! 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 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:+-- 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@@ -887,24 +981,65 @@ writeWFRef r@(WFRef n ref) x= do mr <- readDBRef ref case mr of- Nothing -> error $ "writeWFRef: workflow does not exist: " ++ keyObjDBRef ref- Just s -> do- let elems= versions s- l = state s -- L.length elems- p = l - n- (h,t)= L.splitAt p elems- elems'= h ++ (toIDyn x:tail' t)- tail' []= []- tail' t= L.tail t+ Nothing -> error $ "writeWFRef: workflow does not exist: " ++ show ref+ Just st@Stat{..} ->+ writeDBRef ref st{references= add x references} -- !> ("writeWFREF"++ show r) - writeDBRef ref s{ versions= elems'}+ 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+ withSTMResources[stat0{wfName= n}] $ change n+ 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'++ where+ n = keyWF wf t+ n'= keyWF wf t'+ change m [Nothing] = error $ "changeState: Workflow not found: "++ show n+ change n [Just s] = resources{toAdd= [ s{wfName=n',versions = toIDyn t': L.tail( versions s) }]+ ,toDelete=[s]}++ doit n [Nothing]= error $ "moveState: state not found for: " ++ n+++ -- | Log a message in the workflow history. I can be printed out with 'printWFhistory' -- The message is printed in the standard output too-logWF :: (Monad m, MonadIO m) => String -> Workflow m ()+logWF :: MonadIO m => String -> Workflow m () logWF str=do str <- step . liftIO $ do time <- getClockTime >>= toCalendarTime >>= return . calendarTimeToString@@ -946,7 +1081,7 @@ False -> retry True -> return x --- | observe the workflow log untiil a condition is met.+-- | Observe the workflow log untiil a condition is met. waitFor :: ( Indexable a, Serialize a, Serialize b, Typeable a , Indexable b, Typeable b)@@ -965,7 +1100,7 @@ -> STM b -- ^ The first event that meet the condition waitForSTM filter wfname x= do let name= keyWF wfname x- let tv= getDBRef . key $ stat0{wfName= name} -- `debug` "**waitFor***"+ let tv= getDBRef . keyResource $ stat0{wfName= name} -- `debug` "**waitFor***" mmx <- readDBRef tv case mmx of@@ -981,6 +1116,7 @@ +--{-# 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 time start to count from the first call to getTimeoutFlag on@@ -988,26 +1124,23 @@ -- 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 is understood to start from the first time that the timeout was started. Sucessive restarts of the workflow will respect this timing- -> Workflow m (TVar Bool) -- ^ the returned flag in the workflow monad+ :: 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+ tnow <- step $ liftIO getTimeSeconds flag tnow t where- flag tnow delta = WF(\s -> do- (s', tv) <- case timeout s of- Nothing -> do- tv <- liftIO $ newTVarIO False- return (s{timeout= Just tv}, tv)- Just tv -> return (s, tv)- liftIO $ do- let t = tnow + delta- atomically $ writeTVar tv False- forkIO $ do waitUntil t ; atomically $ writeTVar tv True- return (s', tv))+ 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@@ -1046,5 +1179,52 @@ threadDelay $ delay * 1000000 if delta <= 0 then return () else wait $ delta - (fromIntegral delay ) +-- | Return either the result of the STM conputation or Nothing in case of timeout+-- This timeout is persistent. This means that the time start to count from the first call to getTimeoutFlag on+-- no matter if the workflow is restarted. The time that the worlkflow has been stopped count also.+-- Thus, the wait time can exceed the time between failures.+-- when timeout is 0 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 must result in the end of the workflow.+-- If the workflow is restarted after 'time2' has elapsed, the workflow+-- will restart from the beginning. If not, it will restart at the last checkpoint.+--+-- 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+ 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
+ Control/Workflow/Configuration.hs view
@@ -0,0 +1,34 @@++++{-# OPTIONS+ -XScopedTypeVariables+#-}++{- | Helpers for application initialization -}++module Control.Workflow.Configuration (once, ever, runConfiguration++) where++import Control.Workflow+import Data.Typeable+import Data.RefSerialize+import Control.Monad.Trans+import Control.Exception++-------------- configuation+-- | to execute a computation every time it is invoked. A synonimous of `unsafeIOtoWF`+ever:: (Typeable a,Serialize a, MonadIO m) => IO a -> Workflow m a+ever= unsafeIOtoWF++-- | to execute one computation once . It executes at the first run only+once :: (Typeable a,Serialize a, MonadIO m) => m a -> Workflow m a+once= step++-- | executes a computation with `once` and `ever` statements+runConfiguration confname confProc = handle (\(e :: SomeException) -> return ())+ $ exec1 confname $ do+ confProc+ error ""+ step $ return ()
Control/Workflow/IDynamic.hs view
@@ -1,4 +1,4 @@-{-# OPTIONS -XExistentialQuantification+ {-# OPTIONS -XExistentialQuantification -XOverlappingInstances -XUndecidableInstances -XScopedTypeVariables@@ -19,9 +19,9 @@ import Unsafe.Coerce import System.IO.Unsafe import Data.TCache -import Data.TCache.DefaultPersistence+import Data.TCache.Defs import Data.RefSerialize-+import Data.Char (showLitChar) import Data.ByteString.Lazy.Char8 as B @@ -47,59 +47,20 @@ deriving Typeable- -{--class (Monad writerm , Monad readerm)- => Serializer writerm readerm a- | a -> writerm readerm where- serial :: a -> writerm ()- deserial :: readerm a- fromString :: ByteString -> a- toString :: a -> ByteString - serialM :: a -> writerm ByteString- serialM = return . toString-- fromDynData :: ByteString ->(Context, ByteString) -> a- fromDynData s _= fromString s-- dGetContext :: a -> readerm (Context, ByteString)- dGetContext _ = return (M.empty,"")---class (Serializer w r a, Serializer w' r' b) => TwoSerializer w r w' r' a b---instance (Serializer w r a, Serializer w' r' b) => TwoSerializer w r w' r' a b--}-{-- symbols :: (Show symbolType, Eq symbolType)=> [symbolType]- symbols= []---instance (Serializer m n s Int, Serializer m n s a) => Serializer m n s [a] where- serial xs= serial (Prelude.length xs) >> mapM_ serial xs- deserial = do- n <- deserial- replicateM n deserial--instance Binary a => Serializer PutM Get a where- serial = put- deserial = get--+newtype Save= Save ByteString deriving Typeable+ +tosave d@(IDyn r)= unsafePerformIO $ do+ mr<- readIORef r+ case mr of+ DRight _ -> return d+ DLeft (s,_) -> writeIORef r (DRight $ Save s) >> return d -choices :: (Serializer writerm readerm c symbolType- , Serializer writerm readerm symbolType a)- =>[(symbolType, readerm a)] -> readerm a-choices l= do- n <- deserial- case lookup n l of- Just f -> f- Nothing -> error $ "case "++ show n ++ "not in choice"+instance Serialize Save where+ showp (Save s)= insertString s+ readp = error "readp not impremented for Save" --} errorfied str str2= error $ str ++ ": IDynamic object not reified: "++ str2 @@ -107,61 +68,7 @@ dynPrefix= "Dyn" dynPrefixSp= append (pack dynPrefix) " "--{--instance Serialize IDynamic where- showp (IDyn x)= do- showpx <- showp x- len <- showpText . fromIntegral $ B.length showpx- return $ dynPrefixSp `append` len `append` " " `append` showpx-- showp (IDyns showpx)= do- len <- showpText $ B.length showpx- return $ dynPrefixSp `append` len `append` " " `append` showpx- readp = do- symbol dynPrefix- n <- readpText- s <- takep n- c <- getContext- return $ IDyns $ s `append` "*where*" `append` c---instance Binary IDynamic where- put (IDyn x) = put $! toString x- put (IDyns s) = put s-- get = do- s <- get- return $ IDyns s----instance (Serializer w r ByteString) => Serializer w r IDynamic where- serial (IDyn r)=- let t= unsafePerformIO $ readIORef r- in case t of-- DRight x -> do- s <- unsafeCoerce $ serialM x -- thatś why Binary and Text versions fo workflow can not be mixed- serial $! (s :: ByteString)- DLeft (s , _) -> serial s-- deserial = do- s <- deserial- c <- dGetContext (undefined :: IDynamic)- return $ IDyn $! ref s c- where- ref s c= unsafePerformIO . newIORef $ DLeft (s,c)-- toString (IDyn r)=- let t= unsafePerformIO $ readIORef r- in case t of- DRight x -> toString x- DLeft (s, _) -> s-- fromString s= IDyn . unsafePerformIO . newIORef $ DLeft (s,(M.empty,""))---}+notreified = pack $ dynPrefix ++" 0" @@ -170,26 +77,27 @@ showp (IDyn t)= case unsafePerformIO $ readIORef t of DRight x -> do- insertString $ pack dynPrefix- showpx <- showps x- showpText . fromIntegral $ B.length showpx- insertString showpx---+-- insertString $ pack dynPrefix+ c <- getWContext+ showpx <- rshowps x+-- showpText . fromIntegral $ B.length showpx+ showp $ unpack showpx - DLeft (showpx,_) -> do -- error $ "IDynamic not reified :: "++ unpack showpx- insertString $ pack dynPrefix- showpText 0+ DLeft (showpx,_) -> -- error $ "IDynamic not reified :: "++ unpack showpx+-- insertString notreified+ insertString $ encode showpx+ where+ encode = pack . show . unpack - readp = do- symbol dynPrefix- n <- readpText- s <- takep n+ readp = lexeme (do+-- symbol dynPrefix+-- n <- readpText+-- s <- takep n + s <- rreadp :: STR String - c <- getContext- return . IDyn . unsafePerformIO . newIORef $ DLeft ( s, c)+ c <- getRContext+ return . IDyn . unsafePerformIO . newIORef $ DLeft ( pack s, c)) <?> "IDynamic" @@ -198,8 +106,8 @@ show (IDyn r) = let t= unsafePerformIO $ readIORef r in case t of- DRight x -> "IDyn " ++ ( unpack . runW $ showp x) ++ ")" - DLeft (s, _) -> "IDyn " ++ unpack s+ DRight x -> "IDyn (" ++ ( unpack . runW $ showp x) ++ ")" + DLeft (s, _) -> "IDyns " ++ unpack s @@ -226,11 +134,14 @@ DLeft (str, c) -> handle (\(e :: SomeException) -> return Nothing) $ -- !> ("safeFromIDyn : "++ show e)) $ do- let v= runRC c readp str+ let v= runRC c rreadp str writeIORef r $! DRight v -- !> ("***reified "++ unpack str) return $! Just v -- !> ("*** end reified " ++ unpack str) --main= print (safeFromIDyn $ IDyn $ unsafePerformIO $ newIORef $ DLeft $ (pack "1", (unsafePerformIO $ HT.new (==) HT.hashInt, pack "")) :: Maybe Int) -+reifyM :: (Typeable a,Serialize a) => IDynamic -> a -> IO a+reifyM dyn v = do+ let v'= fromIDyn dyn+ return $ v' `seq` v'
Control/Workflow/Patterns.hs view
@@ -1,18 +1,13 @@ {-# LANGUAGE DeriveDataTypeable , ScopedTypeVariables+ , FlexibleInstances , FlexibleContexts- #-} {-# OPTIONS -IControl/Workflow #-} {- | This module contains monadic combinators that express some workflow patterns. see the docAprobal.hs example included in the package -Here the constraint `DynSerializer w r a` is equivalent to `Data.Refserialize a`-This version permits optimal (de)serialization if you store in the queue different versions of largue structures, for-example, documents. You must define the right RefSerialize instance however.-See an example in docAprobal.hs incuded in the paclkage.-Alternatively you can use Data.Binary serlializatiion with Control.Workflow.Binary.Patterns EXAMPLE: @@ -25,13 +20,17 @@ If the result is True, the pipeline continues to the next stage (checkValidated) the next stage is the same process with a new list of users (superbosses).-This time, there is a timeout of 7 days. the result of the users that voted is summed+There is a timeout of seven days. The result of the users that voted is summed up according with the same monoid instance if the result is true the document is added to the persistent list of approbed documents-if the result is false, the document is added to the persistent list of rejectec documents (checlkValidated1)+if the result is false, the document is added to the persistent list of rejectec documents (@checlkValidated1@) +The program can be interrupted at any moment. The Workflow monad will restartWorkflows+it at the point where it was interrupted. +This example uses queues from "Data.Persistent.Queue"+ @docApprobal :: Document -> Workflow IO () docApprobal doc = `getWFRef` \>>= docApprobal1 @@ -49,7 +48,7 @@ askUser _ _ user False = return False askUser doc rdoc user True = do `step` $ `push` (quser user) rdoc- `logWF` ("wait for any response from the user: " ++ user)+ `logWF` (\"wait for any response from the user: \" ++ user) `step` . `pop` $ qdocApprobal (title doc) log txt x = `logWF` txt >> return x@@ -78,25 +77,28 @@ ) where import Control.Concurrent.STM import Data.Monoid-import Control.Concurrent.MonadIO+ import qualified Control.Monad.CatchIO as CMC+ import Control.Workflow.Stat import Control.Workflow import Data.Typeable import Prelude hiding (catch)-import Control.Monad(when)-import Control.Exception.Extensible (Exception)+import Control.Monad+import Control.Monad.Trans+import Control.Concurrent+import Control.Exception.Extensible (Exception,SomeException) import Data.RefSerialize import Control.Workflow.Stat-import Debug.Trace+ import Data.TCache+import Debug.Trace -a !> b = trace b a -data ActionWF a= ActionWF (WFRef(Maybe a)) (WFRef (String, Bool))+newtype ActionWF a= ActionWF (WFRef(Maybe a)) -- (WFRef (String, Bool)) --- | spawn a list of independent workflows (the first argument) with a seed value (the second argument).--- Their results are reduced by `merge` or `select`+-- | spawn a list of independent workflow 'actions' with a seed value 'a'+-- The results are reduced by `merge` or `select` split :: ( Typeable b , Serialize b , HasFork io@@ -105,10 +107,10 @@ split actions a = mapM (\ac -> do mv <- newWFRef Nothing- fork (ac a >>= step . liftIO . atomically . writeWFRef mv . Just)- r <- getWFRef- return $ ActionWF mv r)+ fork (ac a >>= \v -> (step . liftIO . atomicallySync . writeWFRef mv . Just) v ) + return $ ActionWF mv )+ actions @@ -119,20 +121,22 @@ , Typeable b , Serialize a, Serialize b) => ([a] -> io b) -> [ActionWF a] -> Workflow io b-merge cond actions= mapM (\(ActionWF mv _) -> readWFRef1 mv ) actions >>= step . cond+merge cond results= step $ mapM (\(ActionWF mv ) -> readWFRef1 mv ) results >>= cond -- !> "cond" readWFRef1 :: ( MonadIO io , Serialize a , Typeable a) => WFRef (Maybe a) -> io a-readWFRef1 mv = liftIO . atomically $ do- v <- readWFRef mv- case v of- Just(Just v) -> return v- Just Nothing -> retry- Nothing -> error $ "readWFRef1: workflow not found "++ show mv+readWFRef1 r = liftIO . atomically $ do + mv <- readWFRef r + case mv of+ Just(Just v) -> return v -- !> "return v"+ Just Nothing -> retry -- !> "retry"+ Nothing -> error $ "readWFRef1: workflow not found "++ show r++ data Select = Select | Discard@@ -144,12 +148,12 @@ -- | select the outputs of the workflows produced by `split` constrained within a timeout. -- The check filter, can select , discard or finish the entire computation before--- the timeout is reached. When the computation finalizes, it stop all+-- the timeout is reached. When the computation finalizes, it kill all -- the pending workflows and return the list of selected outputs--- the timeout is in seconds and is no limited to Int values, so it can last for years.+-- the timeout is in seconds and is no limited to a single execution, so it can proceed for years. -- -- This is necessary for the modelization of real-life institutional cycles such are political elections--- timeout of 0 means no timeout.+-- A timeout of 0 means no timeout. select :: ( Serialize a , Serialize [a]@@ -161,39 +165,39 @@ -> [ActionWF a] -> Workflow io [a] select timeout check actions= do- res <- newMVar []+ res <- liftIO $ newMVar [] flag <- getTimeoutFlag timeout- parent <- myThreadId- checks <- newEmptyMVar- count <- newMVar 1+ parent <- liftIO myThreadId+ checThreads <- liftIO $ newEmptyMVar+ count <- liftIO $ newMVar 1 let process = do- let check' (ActionWF ac _) = do+ let check' (ActionWF ac ) = do r <- readWFRef1 ac b <- check r case b of Discard -> return () Select -> addRes r FinishDiscard -> do- throwTo parent FinishDiscard+ liftIO $ throwTo parent FinishDiscard FinishSelect -> do addRes r- throwTo parent FinishDiscard+ liftIO $ throwTo parent FinishDiscard - n <- CMC.block $ do+ n <- liftIO $ CMC.block $ do n <- takeMVar count putMVar count (n+1)- return n+ return n -- !> ("SELECT" ++ show n) if ( n == length actions)- then throwTo parent FinishDiscard+ then liftIO $ throwTo parent FinishDiscard else return () - `CMC.catch` (\(e :: Select) -> throwTo parent e)+ `CMC.catch` (\(e :: Select) -> liftIO $ throwTo parent e) - do- ws <- mapM ( fork . check') actions- putMVar checks ws + ws <- mapM ( fork . check') actions+ liftIO $ putMVar checThreads ws+ liftIO $ atomically $ do v <- readTVar flag -- wait fo timeout case v of@@ -202,33 +206,26 @@ throw FinishDiscard where - addRes r= CMC.block $ do+ addRes r= liftIO $ CMC.block $ do l <- takeMVar res putMVar res $ r : l - let killall = do- mapM_ (\(ActionWF _ th) -> killWFP th) actions- ws <- readMVar checks- liftIO $ mapM_ killThread ws+ let killall = liftIO $ do+ ws <- readMVar checThreads+ liftIO $ mapM_ killThread ws -- !> "KILLALL" - stepControl $ CMC.catch process -- (WF $ \s -> process >>= \ r -> return (s, r))+ step $ CMC.catch process -- (WF $ \s -> process >>= \ r -> return (s, r)) (\(e :: Select)-> do- readMVar res+ liftIO $ readMVar res ) `CMC.finally` killall -killWFP r= liftIO $ do- s <- atomically $ do- (s,_)<- readWFRef r >>= justify ("wfSelect " ++ show r)- writeWFRef r (s, True)- return s - killWF s () justify str Nothing = error str justify _ (Just x) = return x --- | spawn a list of workflows and reduces the results according with the comp parameter within a given timeout+-- | spawn a list of workflows and reduces the results according with the 'comp' parameter within a given timeout -- -- @ -- vote timeout actions comp x=@@ -265,6 +262,36 @@ -> Workflow io b sumUp timeout actions = vote timeout actions (return . mconcat) ++++main= do+ syncWrite SyncManual+ r <- exec1 "sumup" $ sumUp 0 [f 1, f 2] "0"+ print r++ `CMC.catch` \(e::SomeException) -> syncCache -- !> "syncCache"+++f :: Int -> String -> Workflow IO String+f n s= step ( threadDelay ( 5000000 * n)) >> return ( s ++"1")+++main2=do+ syncWrite SyncManual+ exec1 "split" $ split (take 10 $ repeat (step . print)) "hi" >>= merge (const $ return True)+++main3=do+-- syncWrite SyncManual+ refs <- exec1 "WFRef" $ do++ refs <- replicateM 20 $ newWFRef Nothing --"bye initial valoe"+ mapM (\r -> fork $ unsafeIOtoWF $ atomically $ writeWFRef r $ Just "hi final value") refs+++ return refs+ mapM (\r -> readWFRef1 r >>= print) refs
Control/Workflow/Stat.hs view
@@ -5,13 +5,15 @@ -XMultiParamTypeClasses -XFlexibleInstances -XOverloadedStrings-+ -XRecordWildCards+ -XScopedTypeVariables #-} module Control.Workflow.Stat where import Data.TCache+import Data.TCache.Defs -import Data.ByteString.Lazy.Char8(pack, unpack)+import System.IO import System.IO.Unsafe import Data.Typeable import qualified Data.Map as M@@ -21,85 +23,110 @@ import Data.RefSerialize import Control.Workflow.IDynamic import Control.Monad(replicateM)-import Data.TCache.DefaultPersistence-import Data.ByteString.Lazy.Char8 hiding (index)++import qualified Data.ByteString.Lazy.Char8 as B hiding (index)+import Data.ByteString.Char8(findSubstring) import Control.Workflow.IDynamic-import Control.Concurrent(forkIO)+import Control.Concurrent+import Control.Exception(bracket,SomeException)+import System.IO.Error +import System.Directory+import Data.List+import Control.Monad -data WF s m l = WF { st :: s -> m (s,l) }+--import Debug.Trace+--+--(!>) = flip trace +data WF s m l = WF { st :: s -> m (s,l) } -data SyncMode= Synchronous -- ^ write state after every step- | Asyncronous- {frecuency :: Int -- ^ number of seconds between saves when asyncronous- ,cacheSize :: Int -- ^ size of the cache when async- }- | SyncManual -- ^ use Data.TCache.syncCache to write the state- deriving Eq--tvSyncWrite= unsafePerformIO $ newTVarIO (Synchronous, Nothing)- data Stat = Running (M.Map String (String, (Maybe ThreadId)))- | Stat{ wfName :: String- , state:: Int- , index :: Int- , recover:: Bool- , versions ::[IDynamic]- , timeout :: Maybe (TVar Bool)- , self :: DBRef Stat+ | Stat{ self :: DBRef Stat+ , wfName :: String+ , state :: Int+ , recover :: Bool+ , timeout :: Maybe Integer+ , lastActive:: Integer+ , context :: (Context, B.ByteString)+ , references:: [(Int,(IDynamic,Bool))]+ , versions :: [IDynamic] } deriving (Typeable) -stat0 = Stat{ wfName="", state=0, index=0, recover=False, versions = []- , timeout= Nothing, self=getDBRef ""}+stat0 = Stat{ wfName="", state=0, recover=False, versions = []+ , lastActive=0, timeout= Nothing+ , context = (unsafePerformIO newContext,"")+ , references= []+ , self=getDBRef ""} +statPrefix1= "Stat"+statPrefix= statPrefix1 ++"/" -statPrefix= "Stat#"-instance Indexable Stat where- key s@Stat{wfName=name}= statPrefix ++ name- key (Running _)= keyRunning- defPath _= (defPath (1::Int)) ++ "Workflow/"+header Stat{..}= do+ insertString "\r\n"+ insertString $ B.pack statPrefix1+ showpText wfName+ showpText state+ insertChar('(')+ showp timeout+ insertChar(')')+ showp lastActive+-- showp $ markAsWritten references+-- where+-- markAsWritten = map (\(n,(r,_)) -> (n,(r,True))) +getHeader= do+ symbol statPrefix1+ wfName <- readp+ state <- readp+ timeout <- parens readp+ lastActive <- readp+-- references <- readp+ c <- getRContext+ return (wfName, state, timeout, lastActive,[],c) +lenLen= 50++ instance Serialize Stat where showp (Running map)= do- insertString $ pack "Running"+ insertString $ B.pack "Running" showp $ Prelude.map (\(k,(w,_)) -> (k,w)) $ M.toList map - showp stat@( Stat wfName state index recover versions _ _ )=do- insertString $ pack "Stat"- showpText wfName- showpText state- showpText index- showpText recover- showp versions-+ showp stat@Stat{..} = do+ s <- showps $ Prelude.reverse versions+ let l= show (B.length s + lenLen) ++" "++ show state+ insertString . B.pack $ l ++ take (fromIntegral lenLen - length l - 2) (repeat ' ')++ "\r\n"+ insertString s+ header stat - readp = choice [rStat, rWorkflows] where+ readp = choice [rStat, rWorkflows] <?> "on reading Workflow State" where rStat= do- symbol "Stat"- wfName <- stringLiteral- state <- integer >>= return . fromIntegral- index <- integer >>= return . fromIntegral- recover <- bool+ integer+ integer versions <- readp- let self= getDBRef $ key stat0{wfName= wfName}- return $ Stat wfName state index recover versions Nothing self- <?> "Stat"+ (wfName, state, timeout, lastActive,references,cont) <- getHeader ++ let self= getDBRef $ keyResource stat0{wfName= wfName}+ return $ Stat self wfName state True timeout lastActive+ cont references versions++ rWorkflows= do symbol "Running" list <- readp return $ Running $ M.fromList $ Prelude.map(\(k,w)-> (k,(w,Nothing))) list- <?> "RunningWoorkflows" --- return the unique name of a workflow with a parameter (executed with exec or start)+++-- | Return the unique name of a workflow with a parameter (executed with exec or start) keyWF :: Indexable a => String -> a -> String-keyWF wn x= wn ++ "#" ++ key x+keyWF wn x= wn ++ "/" ++ key x data WFRef a= WFRef !Int !(DBRef Stat) deriving (Typeable, Show)@@ -108,51 +135,190 @@ key (WFRef n ref)= keyObjDBRef ref++('#':show n) +--instance Serialize a => Serializable a where+-- serialize = runW . showp+-- deserialize = runR readp +pathWFlows= (defPath (1:: Int)) ++ "Workflow/"+stFName st = pathWFlows ++ keyResource st+Persist fr fw fd = defaultPersist +--nheader= "/header"+--nlog= "/log"+--ncontext= "/context" -instance Serialize a => Serializable a where- serialize = runW . showp+instance IResource Stat where - deserialize = runR readp+ keyResource s@Stat{wfName=name}= statPrefix ++ name+ keyResource (Running _)= keyRunning + readResourceByKey k = fr (pathWFlows ++ k)+ >>= return . fmap ( runR readp) + delResource st= fd (stFName st) -- removeFile (stFName st) `catch`\(e :: IOError) -> return () + writeResource runn@(Running _)= B.writeFile (stFName runn) . runW $ showp runn++--+ writeResource stat@Stat{..}+ | recover = return ()++ | refs <- filter (\(n,(_,written))-> not written) references,+ not $ null refs= do+ let n= stFName stat+ st <- readResource stat -- !> ("WRITING references " ++ wfName )+ safe n $ \h -> do+ let elems= case st of+ Just s@Stat{state=states,versions= verss} -> verss ++ (reverse $ take (state - states) versions )+ Nothing -> reverse versions++ let versions'= substs elems refs+ hSeek h AbsoluteSeek 0+ B.hPut h $ runWC context $ showp $ stat{versions=reverse versions'}++ writeContext h+ hTell h >>= hSetFileSize h++ | otherwise= do+ let n= stFName stat+ safe n $ \h -> do+ (seek,written) <- getWritten h+ writeLog seek written h+++ where++ writeHeader h= B.hPut h $ runWC context $ header stat++ writeLog seek written h++ | written==0=do+ hSeek h AbsoluteSeek 0 -- !> ("WRITING complete " ++ wfName )+ B.hPut h . runWC context . showp $ stat++ writeContext h+ hTell h >>= hSetFileSize h++ | otherwise= do+ hSeek h AbsoluteSeek 0 -- !> ("WRITING partial " ++ wfName )+ let s = runWC context $ insertString "\r\n" >> showpe written ( reverse $ take (state - written) versions)+ let l= show (seek -3 + B.length s) ++" "++ show state+ B.hPut h . B.pack $ l ++ take (fromIntegral lenLen - length l - 2) (repeat ' ') ++ "\r\n"+ hSeek h AbsoluteSeek (fromIntegral seek - 3)+ B.hPut h s+ writeHeader h+ writeContext h+ hTell h >>= hSetFileSize h++ subst elems (n,( x,_))=+ let+ tail' []= []+ tail' t = tail t+ (h,t)= splitAt n elems+ in h ++ ( x:tail' t)++ substs elems xs= foldl subst elems xs++ writeContext h= B.hPut h $ showContext (fst context) True++ getWritten h= do+ size <- hFileSize h+ if size == 0 then return (0,0)+ else do+ s <- B.hGetNonBlocking h (fromIntegral lenLen)+ return $ runR ( return (,) `ap` readp `ap` readp) s+-- seek <- readp+-- written <- readp+-- ) s+++++ showpe _ [] = insertChar ']'+ showpe 0 (x:xs) = do+ rshowp x+ showpe 1 xs+ showpe v (x:l) = insertString "," >> rshowp x >> showpe v l++++safe name f= bracket+ (openFile name ReadWriteMode)+ hClose+ f+ `catch` (handler name (safe name f))+ where+ handler name doagain e + | isDoesNotExistError e=do + createDirectoryIfMissing True $ Prelude.take (1+(Prelude.last $ Data.List.elemIndices '/' name)) name --maybe the path does not exist + doagain +++ | otherwise= if ("invalid" `isInfixOf` ioeGetErrorString e)+ then+ error $ "writeResource: " ++ show e ++ " defPath and/or keyResource are not suitable for a file path" + else do+ hPutStrLn stderr $ "defaultWriteResource: " ++ show e ++ " in file: " ++ name ++ " retrying"+ doagain+++hReadFile h = do+ s <- hFileSize h+ if s == 0 then return ""+ else B.hGetNonBlocking h (fromIntegral s)+++readHeader scont h= do+ size <- hFileSize h+ if size==0 then return Nothing else do+ s <- B.hGetNonBlocking h (fromIntegral size)+ return . Just $ runR getHeader $ s `B.append` scont++++ keyRunning= "Running" instance Serialize ThreadId where- showp th= insertString . pack $ show th- readp = (readp :: ST ByteString) >> (return . unsafePerformIO . forkIO $ return ())+ showp th= return () -- insertString . pack $ show th+ readp = {-(readp `asTypeOf` return ByteString) >>-} (return . unsafePerformIO . forkIO $ return ()) -newtype Pretty = Pretty Stat -instance Show Pretty where- show= unpack . runW . sp+-- | show the state changes along the workflow, that is, all the intermediate results+showHistory :: Stat -> B.ByteString+showHistory Stat {..}= runW sp where- sp (Pretty (Stat wfName state index recover versions _ _))= do- insertString $ pack "Workflow name= "+ sp = do+ insertString $ B.pack "Workflow name= " showp wfName- insertString $ pack "\n"- showElem $ Prelude.reverse $ (Prelude.zip ( Prelude.reverse [1..] ) versions )-+ insertString $ B.pack "\n"+ showElem $ zip [1..] versions+ c <- getWContext+ insertString $ showContext (fst c) True - showElem :: [(Int,IDynamic)] -> ST ()+-- showElem :: [(Int,IDynamic)] -> STW () showElem [] = insertChar '\n'- showElem ((n, dyn):es) = do- showp $ pack "Step "- showp n- showp $ pack ": "- showp dyn+ showElem ((n , dyn):es) = do+ insertString $ B.pack "Step "+ showp (n :: Int)+ insertString $ B.pack ": "+ showp1 dyn insertChar '\n' showElem es +showp1 (IDyn r)=+ case unsafePerformIO $ readIORef r of+ DRight x -> showp x+ DLeft (s, _) -> insertString s ++ instance Indexable String where key= id @@ -163,11 +329,14 @@ key= show +instance Indexable () where+ key _= "void"+ wFRefStr = "WFRef" instance Serialize (WFRef a) where showp (WFRef n ref)= do- insertString $ pack wFRefStr+ insertString $ B.pack wFRefStr showp n showp $ keyObjDBRef ref @@ -177,9 +346,5 @@ k <- readp return . WFRef n $ getDBRef k --- | print the state changes along the workflow, that is, all the intermediate results-printHistory :: Stat -> IO ()-printHistory stat= do- Prelude.putStrLn . show $ Pretty stat- Prelude.putStrLn "-----------------------------------"+
− Data/Persistent/Queue.hs
@@ -1,240 +0,0 @@-{-# OPTIONS -XDeriveDataTypeable- -XTypeSynonymInstances- -XMultiParamTypeClasses- -XExistentialQuantification- -XOverloadedStrings- -XFlexibleInstances- -XUndecidableInstances- -XFunctionalDependencies--- -IControl/Workflow-- #-}-{-# OPTIONS -IData/Persistent/Queue #-}-{- |-This module implements a persistent, transactional collection with Queue interface as well as- indexed access by key--use this version if you store in the queue different versions of largue structures, for-example, documents and define a "Data.RefSerialize" instance. If not, use "Data.Persistent.Queue.Binary" Instead.--Here @QueueConstraints w r a@ means @Data.RefSerlialize.Serialize a@---}--module Data.Persistent.Queue(-RefQueue(..), getQRef,-pop,popSTM,pick, flush, flushSTM,-pickAll, pickAllSTM, push,pushSTM,-pickElem, pickElemSTM, readAll, readAllSTM,-deleteElem, deleteElemSTM,-unreadSTM,isEmpty,isEmptySTM-) where-import Data.Typeable-import Control.Concurrent.STM(STM,atomically, retry)-import Control.Monad(when)-import Data.TCache.DefaultPersistence--import Data.TCache-import System.IO.Unsafe-import Data.RefSerialize-import Data.ByteString.Lazy.Char8-import Data.RefSerialize--import Debug.Trace--a !> b= trace b a---instance Indexable (Queue a) where- key (Queue k _ _)= queuePrefix ++ k-----data Queue a= Queue {name :: String, imp :: [a], out :: [a]} deriving (Typeable)----instance Serialize a => Serialize (Queue a) where- showp (Queue n i o)= showp n >> showp i >> showp o- readp = do- n <- readp- i <- readp- o <- readp- return $ Queue n i o-----queuePrefix= "Queue#"-lenQPrefix= Prelude.length queuePrefix----instance Serialize a => Serializable (Queue a ) where- serialize = runW . showp- deserialize = runR readp---- | a queue reference-type RefQueue a= DBRef (Queue a)--unreadSTM :: (Typeable a, Serialize a) => RefQueue a -> a -> STM ()-unreadSTM queue x= do- r <- readQRef queue- writeDBRef queue $ doit r- where- doit (Queue n imp out) = Queue n imp ( x : out)----- | check if the queue is empty-isEmpty :: (Typeable a, Serialize a) => RefQueue a -> IO Bool-isEmpty = atomically . isEmptySTM--isEmptySTM :: (Typeable a, Serialize a) => RefQueue a -> STM Bool-isEmptySTM queue= do- r <- readDBRef queue- return $ case r of- Nothing -> True- Just (Queue _ [] []) -> True- _ -> False------ | get the reference to new or existing queue trough its name-getQRef :: (Typeable a, Serialize a) => String -> RefQueue a-getQRef n = getDBRef . key $ Queue n undefined undefined----- | empty the queue (factually, it is deleted)-flush :: (Typeable a, Serialize a) => RefQueue a -> IO ()-flush = atomically . flushSTM---- | version in the STM monad-flushSTM :: (Typeable a, Serialize a) => RefQueue a -> STM ()-flushSTM tv= delDBRef tv---- | read the first element in the queue and delete it (pop)-pop- :: (Typeable a, Serialize a) => RefQueue a -- ^ Queue name- -> IO a -- ^ the returned elems-pop tv = atomically $ popSTM tv---readQRef :: (Typeable a, Serialize a) => RefQueue a -> STM(Queue a)-readQRef tv= do- mdx <- readDBRef tv- case mdx of- Nothing -> do- let q= Queue ( Prelude.drop lenQPrefix $ keyObjDBRef tv) [] []- writeDBRef tv q- return q- Just dx ->- return dx---- | version in the STM monad-popSTM :: (Typeable a, Serialize a) => RefQueue a- -> STM a-popSTM tv=do- dx <- readQRef tv- doit dx-- where- --doit :: (Typeable a, Serialize a, Serializer m n [a]) => Queue a -> STM a- doit (Queue n [x] [])= do- writeDBRef tv $ (Queue n [] [])- return x- doit (Queue _ [] []) = retry- doit (Queue n imp []) = doit (Queue n [] $ Prelude.reverse imp)- doit (Queue n imp list ) = do- writeDBRef tv (Queue n imp (Prelude.tail list ))- return $ Prelude.head list---- | read the first element in the queue but it does not delete it-pick- :: (Typeable a, Serialize a) => RefQueue a -- ^ Queue name- -> IO a -- ^ the returned elems-pick tv = atomically $ do- dx <- readQRef tv- doit dx- where- doit (Queue _ [x] [])= return x- doit (Queue _ [] []) = retry- doit (Queue n imp []) = doit (Queue n [] $ Prelude.reverse imp)- doit (Queue n imp list ) = return $ Prelude.head list---- | push an element in the queue-push :: (Typeable a, Serialize a) => RefQueue a -> a -> IO ()-push tv v = atomically $ pushSTM tv v---- | version in the STM monad-pushSTM :: (Typeable a, Serialize a) => RefQueue a -> a -> STM ()-pushSTM tv v=- readQRef tv >>= \ ((Queue n imp out)) -> writeDBRef tv $ Queue n (v : imp) out---- | return the list of all elements in the queue. The queue remains unchanged-pickAll :: (Typeable a, Serialize a) => RefQueue a -> IO [a]-pickAll= atomically . pickAllSTM---- | version in the STM monad-pickAllSTM :: (Typeable a, Serialize a) => RefQueue a -> STM [a]-pickAllSTM tv= do- (Queue name imp out) <- readQRef tv- return $ out ++ Prelude.reverse imp---- | return the first element in the queue that has the given key-pickElem ::(Indexable a,Typeable a, Serialize a) => RefQueue a -> String -> IO(Maybe a)-pickElem tv key= atomically $ pickElemSTM tv key---- | version in the STM monad-pickElemSTM :: (Indexable a,Typeable a, Serialize a)- => RefQueue a -> String -> STM(Maybe a)-pickElemSTM tv key1= do- Queue name imp out <- readQRef tv- let xs= out ++ Prelude.reverse imp- when (not $ Prelude.null imp) $ writeDBRef tv $ Queue name [] xs- case Prelude.filter (\x-> key x == key1) xs of- [] -> return $ Nothing- (x:_) -> return $ Just x---- | update the first element of the queue with a new element with the same key-updateElem :: (Indexable a,Typeable a, Serialize a)- => RefQueue a -> a -> IO()-updateElem tv x = atomically $ updateElemSTM tv x---- | version in the STM monad-updateElemSTM :: (Indexable a,Typeable a, Serialize a)- => RefQueue a -> a -> STM()-updateElemSTM tv v= do- Queue name imp out <- readQRef tv- let xs= out ++ Prelude.reverse imp- let xs'= Prelude.map (\x -> if key x == n then v else x) xs- writeDBRef tv $ Queue name [] xs'- where- n= key v---- | return the list of all elements in the queue and empty it-readAll :: (Typeable a, Serialize a) => RefQueue a -> IO [a]-readAll= atomically . readAllSTM---- | a version in the STM monad-readAllSTM :: (Typeable a, Serialize a) => RefQueue a -> STM [a]-readAllSTM tv= do- Queue name imp out <- readQRef tv- writeDBRef tv $ Queue name [] []- return $ out ++ Prelude.reverse imp---- | delete all the elements of the queue that has the key of the parameter passed-deleteElem :: (Indexable a,Typeable a, Serialize a) => RefQueue a-> a -> IO ()-deleteElem tv x= atomically $ deleteElemSTM tv x---- | verison in the STM monad-deleteElemSTM :: (Typeable a, Serialize a,Indexable a) => RefQueue a-> a -> STM ()-deleteElemSTM tv x= do- Queue name imp out <- readQRef tv- let xs= out ++ Prelude.reverse imp- writeDBRef tv $ Queue name [] $ Prelude.filter (\x-> key x /= k) xs- where- k=key x-
Demos/Fact.hs view
@@ -21,7 +21,7 @@ data Fact= Fact Integer Integer deriving (Typeable, Read, Show) --- binary interface is not used+ instance Binary Fact where put (Fact n v)= put n >> put v get= do
Demos/docAprobal.hs view
@@ -22,9 +22,10 @@ -}+module Main where import Control.Workflow -import Data.Persistent.Queue+import Data.Persistent.Collection import Control.Workflow.Patterns import Data.Typeable@@ -32,11 +33,11 @@ import Data.List (find) import Data.Maybe(fromJust) import Control.Monad (when)-import Control.Concurrent ( forkIO)+import Control.Concurrent import GHC.Conc ( atomically) import Data.RefSerialize import Data.TCache(syncCache)-import Data.ByteString.Lazy.Char8(pack)+import qualified Data.ByteString.Lazy.Char8 as B import Data.Monoid @@ -53,7 +54,7 @@ instance Serialize Document where showp (Document title text)= do- insertString $ pack "Document"+ insertString $ B.pack "Document" showp title rshowp text @@ -98,9 +99,11 @@ +loop= loop main = do -- restart the interrupted document approbal workflows (if necessary)+ syncWrite SyncManual restartWorkflows [("docApprobal",docApprobal)] @@ -115,7 +118,7 @@ "3" -> aprobal "boss2" "4" -> aprobal "superboss1" "5" -> aprobal "superboss2"-+ _ -> exitWith ExitSuccess bosses= ["boss1", "boss2"]@@ -147,7 +150,7 @@ -} docApprobal :: Document -> Workflow IO ()-docApprobal doc = getWFRef >>= docApprobal1+docApprobal doc = newWFRef doc >>= docApprobal1 where -- using a reference instead of the doc itself docApprobal1 rdoc=@@ -168,10 +171,10 @@ askUser title rdoc user True = do step $ push (quser user) rdoc logWF ("wait for any response from the user: " ++ user)- step . pop $ qdocApprobal title+ step . pop $ qdocApprobal title - log txt x = logWF txt >> return x+-- log txt x = logWF txt >> return x checkValidated :: Bool -> Workflow IO Bool checkValidated val =@@ -254,7 +257,7 @@ "2" -> modify "3" -> view "4" -> history- _ -> syncCache >> exitSuccess+ _ -> syncCache >> exitSuccess !> "syncCache" userMenu @@ -275,7 +278,7 @@ let docproto= Document{title= ks !! (n-1), text=undefined} case head l of 'v' -> do- getWFHistory "docApprobal" docproto >>= printHistory . fromJust+ getWFHistory "docApprobal" docproto >>= B.putStrLn . showHistory . fromJust history 'd' -> do delWF "docApprobal" docproto@@ -289,7 +292,8 @@ modify :: IO () modify= do separator- empty <- isEmpty (quser user) :: IO Bool+ let quseruser= quser user+ empty <- isEmpty (quseruser) :: IO Bool if empty then putStrLn "no more documents to modify\nthanks, enter as Boss for the approbal" else do rdoc <- pick (quser user)@@ -300,7 +304,7 @@ -- return $ diff doc1 doc atomically $ do- popSTM (quser user)+ popSTM (quseruser) pushSTM (qdoc $ title doc) doc1 modify @@ -350,19 +354,22 @@ separator aprobalList putStrLn $ "thanks , press any key to exit, "++ who- getLine+ threadDelay 10000000 syncCache+ threadDelay 1000000 return () where+ quserwho= quser who aprobalList= do- empty <- isEmpty (quser who)+ empty <- isEmpty (quserwho) if empty then do putStrLn "No more document to validate. Bye"+ return () else do- rdoc <- pick (quser who)- syncCache+ rdoc <- pick (quserwho)+ approbal1 rdoc aprobalList approbal1 :: WFRef Document -> IO ()@@ -377,9 +384,9 @@ let res= if b == 's' then True else False -- send the message to the workflow atomically $ do- popSTM (quser who)+ popSTM (quserwho) pushSTM (qdocApprobal $ title doc) res- syncCache+
Demos/pr.hs view
@@ -1,26 +1,43 @@ module Main where import Control.Workflow+import Control.Workflow.Stat++import Data.TCache import Control.Concurrent(threadDelay) import System.IO (hFlush,stdout)-import Data.Vector as V-import System.IO.Unsafe-import Control.Concurrent.MVar+import Control.Concurrent+import qualified Data.ByteString.Lazy.Char8 as B -count= unsafePerformIO $ newMVar 0 --- delayed evaluation of logged step values -instance Indexable (Vector a) where- key= const "Vector"+main2= do+ Just stat <- getWFHistory "docApprobal" "Doc#title"+ B.putStrLn $ showHistory stat+ withResource stat $ \(Just stat) -> stat{recover= False}+ syncCache -mcount cart= do- i <- step $ modifyMVar count (\n-> threadDelay 1000000>> print cart >> return (n+1, n `mod` 3)) - let newCart= cart V.// [(i, cart V.! i + 1 )] - mcount newCart- return ()+main= do+ refs <- exec1 "WFRef" $ do+ step $ return (1 :: Int)+ (ref,s) <- stepWFRef $ return "bye initial valoe"+ step $ return (3 :: Int) -main= start "count" mcount (V.fromList [0,0,0 :: Int])+ return ref++ atomically $ writeWFRef refs "hi final value"+ s <- atomically $ readWFRef refs+ print s+ Just stat <- getWFHistory "WFRef" ()+ B.putStrLn $ showHistory stat+ syncCache+ atomically flushAll+ Just stat <- getWFHistory "WFRef" ()+ B.putStrLn $ showHistory stat+ s <- atomically $ readWFRef refs+ print s++
Demos/sequence.hs view
@@ -4,12 +4,15 @@ import Control.Concurrent(threadDelay) import System.IO (hFlush,stdout) +printLine x= do+ putStr (show x ++ " ")+ hFlush stdout+ threadDelay 1000000 -mcount n= do step $ do- putStr (show n ++ " ")- hFlush stdout- threadDelay 1000000++mcount :: Int -> Workflow IO ()+mcount n= do step $ printLine n mcount (n+1)- return () -- to specify the return type -main= exec1 "count" $ mcount (0 :: Int)++main= exec1 "count" $ mcount 0
Workflow.cabal view
@@ -1,119 +1,55 @@ name: Workflow-version: 0.6.0.0-+version: 0.7.0.0+cabal-version: >= 1.6 build-type: Simple license: BSD3 license-file: LICENSE-copyright: maintainer: agocorona@gmail.com-build-depends: vector -any, transformers,- extensible-exceptions -any, MonadCatchIO-mtl -any,- monadIO -any,- base >=4 && < 5,- binary -any,- bytestring -any,- containers -any,- RefSerialize >=0.2.8 && < 0.3,- TCache >=0.9 && <1.0,- stm >2, old-time,- mtl -any--- stability: experimental-homepage:-package-url: bug-reports: agocorona@gmail.com-synopsis: library for transparent execution of interruptible computations+synopsis: Monadic transformer for persistence in threads. and workflow patterns description: Transparent support for interruptible computations. A workflow can be seen as a persistent thread that executes any monadic computation. Therefore, it can be used in very time consuming computations such are CPU intensive calculations or procedures that are most of the time waiting for the action of a process or an user, that are prone to comunication- failures, timeouts or shutdowns.+ failures, timeouts or shutdowns. It also can be used if you like to restart your+ program at the point where the user left it last time. .- The computantion can be restarted at the interrupted point thanks to its logged state in permanent storage.+ The computation can be restarted at the interrupted point thanks to its logged state in permanent storage. Besides that, the package also provides other services associated to workflows .- New in this release,- .- * 0.6.0.0 Changes in ghc 7.4 forces to drop the abstract serializer- now Text and binary versions are the same.- For binary serialization, use 'showBinary' and 'readpBinary'- .- * 0.5.8.2 minor changes for the MFlow package- .- * 0.5.8.1 solved a bug that caused a "casting failure"- .- * 0.5.8:- .- *registerType is no longer needed- .- * Configurable state persistence (for example, in databases) . Default in files- .- * Added Data.Binary serialization (Use `Control.Workflow.Binary`) besides Text serialization (`Control.Wokflow.Text`)- .- * New simpler and more intuitive workflow primitives with excepion handling- .- * Instances of classes in Control.Concurrent.MonadIO, MonadCatchIO- .- * An EDSL of workflow patterns. To express bifurcations, confluences, timeouts etc- .- * References to workflow objects, that can be accessed externally: WFRef's- .- Previous versions:- .- * Persisten communications facilities trough persistent data objects, inspection of workflow states , persistent queues, persistent timeouts and, new in this release, References to internal workflow objects WFRef's- .- * workflow management and monitoring, view workflow history and intermediate results.-+ This release inprove the logging/recovery process in workflows with with many steps+ .+ See "Control.Workflow" for details +category: Control, Workflow -category: Control, Workflow, Concurrent author: Alberto Gómez Corona-data-files:+ data-dir: "" -extra-tmp-files:-exposed-modules:+extra-source-files: Demos/Fact.hs Demos/Inspect.hs+ Demos/docAprobal.hs Demos/pr.hs Demos/sequence.hs - Control.Workflow- Control.Workflow.Patterns - Data.Persistent.Queue --other-modules:- Control.Workflow.IDynamic- Control.Workflow.Stat---extra-source-files:- Demos/docAprobal.hs- Demos/sequence.hs- Demos/Fact.hs- Demos/Inspect.hs- Demos/pr.hs--exposed: True-buildable: True-build-tools:-cpp-options:-cc-options:-ld-options:-pkgconfig-depends:-frameworks:-c-sources:+library+ build-depends: MonadCatchIO-mtl -any, RefSerialize >=0.2.8 && <0.3,+ 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 -extensions:-extra-libraries:-extra-lib-dirs:-includes:-install-includes:-include-dirs:-hs-source-dirs: .+ exposed-modules: Control.Workflow+ Control.Workflow.Configuration+ Control.Workflow.Patterns+ exposed: True+ buildable: True+ extensions: OverlappingInstances UndecidableInstances+ MultiParamTypeClasses ExistentialQuantification+ TypeSynonymInstances RecordWildCards DeriveDataTypeable+ hs-source-dirs: .+ other-modules: Control.Workflow.Stat Control.Workflow.Stat+ Control.Workflow.IDynamic -ghc-prof-options:-ghc-shared-options:-ghc-options:-hugs-options:-nhc98-options:-jhc-options:+source-repository head+ type : git+ location: https://github.com/agocorona/Workflow