Workflow 0.5.6 → 0.5.8
raw patch · 20 files changed
+3075/−818 lines, 20 filesdep +MonadCatchIO-mtldep +binarydep +bytestringdep ~RefSerializedep ~TCachedep ~base
Dependencies added: MonadCatchIO-mtl, binary, bytestring, extensible-exceptions, monadIO, transformers, vector
Dependency ranges changed: RefSerialize, TCache, base
Files
- Control/Workflow.hs +0/−779
- Control/Workflow/Binary.hs +105/−0
- Control/Workflow/Binary/BinDefs.hs +94/−0
- Control/Workflow/Binary/Patterns.hs +107/−0
- Control/Workflow/GenSerializer.hs +70/−0
- Control/Workflow/IDynamic.hs +207/−0
- Control/Workflow/Patterns.inc.hs +172/−0
- Control/Workflow/Stat.hs +78/−0
- Control/Workflow/Text.hs +115/−0
- Control/Workflow/Text/Patterns.hs +97/−0
- Control/Workflow/Text/TextDefs.hs +151/−0
- Control/Workflow/Workflow.inc.hs +898/−0
- Data/Persistent/Queue/Binary.hs +58/−0
- Data/Persistent/Queue/Queue.inc.hs +205/−0
- Data/Persistent/Queue/Text.hs +60/−0
- Demos/Fact.hs +55/−0
- Demos/Inspect.hs +92/−0
- Demos/docAprobal.hs +386/−0
- Demos/sequence.hs +16/−0
- Workflow.cabal +109/−39
− Control/Workflow.hs
@@ -1,779 +0,0 @@-{-# OPTIONS -fglasgow-exts -XOverlappingInstances -XUndecidableInstances -O2 #-}------------------------------------------------------------------------------------ Module : Control.Workflow--- Copyright : Alberto Gómez Corona--- License : see LICENSE------ Maintainer : agocorona@gmail.com--- Stability : experimental---{- |-Transparent support for interruptable computations. A workflow can be seen as a persistent thread.-The main features are:-- * Transparent state logging trough a monad transformer: step :: m a -> Workflow m a.-- * Resume the computation state after an accidental o planned program shutdown (restartWorkflows).-- *Event handling (waithFor, waitForData).-- * Monitoring of workflows with state change display and other auxiliary features.-- * Communications with other processes including other workflows trough persistent data objects,- inspecttion of intermediate workflow results , Queues so that no data is lost due to shutdowns--In this way, a very long computation that may take more time than the average time between-hardware or software failures, shutdowns etc. The workflow can be defined transparently in a single monadic procedure.-Besides state logging and recovery, there are a number of communication primitives that are aware-of persistence across reinitiations such are persistent queues, persistent timeouts, or wait for events- in the STM monad. These primitives permits inter-woikflow communications and communications with- external threads.-- This package uses TCache for persistence and event handling.-- It also uses the package Refserialize. This package permits to reduce the workflow state load,- since the RefSerialize package permits to serialize and deserialize complex and autoreferenced data structures without- loosing such references, this is critical when big and structured data, such are documents, suffer little- modifications across a set of workflow steps. Therefore, it is also recommended to use Refserialize for- big user-defined objects that have small pieces that suffer little modifications during the workflow. As an- added bonus, the history will show such changes with more detail.--The 'step' primitive is the lift operation that converts a result of type @m a@ to a type @'Workflow' m a@-with automatic state loggin and recovery. To allow such features, Every @a@ must be instance of-'Typeable' and 'IResource' (defined in the @TCache@ package).--In fact, Workflow can be considered as an instance of a partial monad transformed. defined as such:-- @class 'PMonadTrans' t m a where-- 'plift' :: Monad m => m a -> t m a---- instance (Monad m,MonadIO m, IResource a, Typeable a)-- => PMonadTrans (WF Stat) m a where-- 'plift' = 'step'- @---It is partial because the lift operation is not defined for every monad @m@ and data type @a@ , but for monads and data-types that meet certain conditions. In this case, to be instances of @MonadIO@, @IResource@ and @Typeable@ respectively.--to avoid to define the last two interfaces however, 'Read' and 'Show'' can be used to derive instances of 'IResource'- for most of the useful cases. This is the set of automatic derivations:--- @(Read a, Show a) => 'Serialize' a-- Typeable a => 'Indexable' a (a single key for all values. enough for workflows)-- ('Indexable' a, 'Serialize' a) => IResource a@---Therefore deriving to be instance of @Read, Show@ is enough for every intermediate data result along the computation--Because 'Data.TCache.Dynamic' from the package 'TCache' is used for persistence, every data type must be registered-by using 'registerType'--Here is a compkete example: This is a counter that shows a sequence of numbers, one a second:--@module Main where-import Control.Concurrent(threadDelay)-import System.IO (hFlush,stdout)-- count n= do- putStr (show n ++ " ") >> hFlush stdout >> threadDelay 1000000- count (n+1)-- main= count 0@--This is the same program, with the added feature of remembering the last count after interrupted:---@module Main where-import Control.Workflow-import Control.Concurrent(threadDelay)-import System.IO (hFlush,stdout)---mcount n= do- 'step' $ putStr (show n ++ " ") >> hFlush stdout >> threadDelay 1000000- mcount (n+1)----main= do- registerType :: IO ()- registerType :: IO Int- let start= 0 :: Int- startWF "count" start [("count", mcount)] :: IO ()@--This is the execution log:--@Worflow-0.5.5\demos>runghc sequence.hs-0 1 2 3 4 5 6 7 sequence.hs: win32ConsoleHandler-sequence.hs: sequence.hs: interrupted-Worflow-0.5.5\demos>-Worflow-0.5.5\demos>runghc sequence.hs-7 8 9 10 11 ....@---}----------------------------------------------------------------------------------module Control.Workflow- ( Workflow -- a useful type name- , WorkflowList- , IResource(..)- , registerType- , PMonadTrans (..)- , Indexable (key)- , step- , startWF- , restartWorkflows- , getStep- , getAll- , logWF- , getWFKeys- , getWFHistory -- return the list of steps results- , delWFHistory -- delete the workflow history- , printHistory -- print the history- , unsafeIOtoWF- , waitFor- , waitUntil- , waitUntilSTM- , syncWrite- , writeQueue- , writeQueueSTM- , readQueue- , readQueueSTM- , unreadQueue- , unreadQueueSTM- , getTimeSeconds- , getTimeoutFlag- , isEmptyQueue- , isEmptyQueueSTM-)-where---import System.IO.Unsafe-import Control.Monad(when,liftM)-import Control.Exception(Exception, throw)-import Control.Concurrent (forkIO,threadDelay, ThreadId)-import Control.Concurrent.STM-import GHC.Conc(unsafeIOToSTM)-import GHC.Base (maxInt)-import Data.TCache.Dynamic-import Data.RefSerialize-import Data.List((\\),find,elemIndices, isPrefixOf)-import Data.Typeable-import System.Time-import Control.Monad.Trans-import Control.Monad (replicateM)-import System.IO(hPutStrLn, stderr)-import Data.List(elemIndex)-import Data.Maybe(fromJust, isNothing)-import qualified Data.Map as M(Map,fromList,elems, insert, lookup)---import System.Mem.StableName--{---import Debug.Trace-debug a b = trace b a--report :: Exception e => IO a -> String -> e -> IO a-report f text e= catch f (\e -> throw $ userError (text++": "++ show e))--freport text f = report f text--}--data WF s m l = WF { st :: s -> m (s,l) }--type Workflow m l= WF Stat m l -- not so scary--type WorkflowList m a b= [(String, a -> Workflow m b) ]----data Stat = RunningWorkflows [String]-- | Stat{ wfName :: String, state:: Int, index :: Int, recover:: Bool, sync :: Bool- , versions ::[IDynamic], timeout :: Maybe (TVar Bool)}- deriving (Typeable)--stat0 = Stat{ wfName="", state=0, index=0, recover=False, versions = []- , sync= True, timeout= Nothing}--hasht x= (hashStableName . unsafePerformIO . makeStableName) x---- serialization of data is done trough RefSerialize because it permits to store--- different versions of the same object with minumum memory.---instance Serialize IDynamic where- showp= tshowp- readp = treadp--instance Serialize Stat where- showp (RunningWorkflows list)= do- str <- showp list- return $ "StatWorkflows "++ str--- showp (Stat wfName state index recover sync versions _)= do- parsea <- rshowp versions- return $ "Stat "++ show wfName ++" "++ show state++" "++show index++" "++show recover++" "- ++ show sync ++ parsea-- readp = choice [rStat, rWorkflows] where- rStat= do- symbol "Stat"- wfName <- readp -- stringLiteral- state <- readp -- integer- index <- readp --integer- recover <- readp --bool- sync <- readp --bool- versions <- rreadp--- return $ Stat wfName state index recover sync versions Nothing---- rWorkflows= do- symbol "StatWorkflows"- list <- readp- return $ RunningWorkflows list-----persistence trough TCache , default persistence in files-workflowsPath= "Workflows/"--instance IResource Stat where- keyResource s@Stat{wfName=name}= "Stat#" ++ name- keyResource (RunningWorkflows _)= "RunningWorkflows"-- defPath _= workflowsPath -- directory for Workflow data-- serialize x= runW $ showp x- deserialize str = runR readp str--{- | Indexablle can be used to derive instances of IResource- This is the set of automatic derivations:-- *(Read a, Show a) => Serialize a-- *Typeable a => Indexable a (a single key for all values. enough for workflows)-- *(Indexable a, Serialize a) => IResource a--}-class Indexable a where- key:: a -> String--instance Typeable a => Indexable a where- key x= show $ typeOf x--instance (Serialize a, Indexable a) => IResource a where- keyResource x=key x- tshowp= showp- treadp= readp----- | executes a IO 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,IResource a, Typeable a) => PMonadTrans (WF Stat) m a where- plift = step---- | An instance of MonadTrans is an instance of PMonadTrans-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--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- (s3, x') <- fun s1-- return (s3, x'))----class (IResource a, Serialize a,Typeable a) => Workflow_ a where--- | step lifts a monadic computation to the WF monad, and provides transparent state loging and resume of computation-step :: (Monad m,MonadIO m,IResource a, Typeable a) => m a -> Workflow m a-step 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 versionss= versions s- let ver= toIDyn x': versionss- let s'= s{recover= False, versions = ver, state= state s+1}-- liftIO $ do-- withResources ([]::[Stat]) (\_-> [s' ]) --`debug` "update cache"-- when (sync s ) syncCache-- return (s', x') )----- | start or continue a workflow.-startWF- :: (Monad m,MonadIO m,IResource a, Typeable a, IResource b, Typeable b)- => String -- ^ name of workflow in the workflow list- -> a -- ^ initial value (even use the initial value even to restart the workflow)- -> WorkflowList m a b -- ^ workflow list. t is an assoc-list of (workflow name string,Workflow methods)- -> m b -- ^ result of the computation---startWF namewf v wfs= do- liftIO (registerType :: IO Stat)- liftIO (registerType :: IO Control.Workflow.Queue)- liftIO (registerType :: IO String)- liftIO (registerType :: IO Integer)--- case lookup namewf wfs of- Nothing -> error $ "startWF: workflow name not found in workflow list: "++namewf;- Just f -> do- let name= namewf ++ "#" ++ keyResource v- let stat1= stat0{wfName= name , versions= [toIDyn v]}- mst <- liftIO $ getResource stat1-- let (vn, stat,create)= case mst of- Nothing -> (v, stat1, True)- Just s-> (v,s{index=0,recover=True},False) -- the last value---- let- addwf [wf ] = resources{ toAdd=[ toIDyn $ RunningWorkflows (name:xs) ]- ++ [ toIDyn stat]}- where xs= case wf of Nothing -> []; Just dyn -> xs where RunningWorkflows xs = fromIDyn dyn-- when create $ liftIO . atomically $ withDSTMResources [toIDyn $ RunningWorkflows undefined ] addwf-- runWF name f vn stat -- `debug` (serialize stat)----runWF :: (Monad m,MonadIO m,IResource a,Typeable a, IResource b, Typeable b)- => String -> ( a -> Workflow m b) -> a -> Stat -> m b-runWF name f v s=do- when (sync s) $ liftIO $ syncCache- (s', v') <- st (f v) $ s --`debug` "runWF********"-- let delWF Nothing = error $ "runWF: Workflow list not found: "- delWF (Just (RunningWorkflows xs ))=- let name= name++ "#" ++ keyResource v- in case elem name xs of- False -> error $"runWF: not found state for workflow: "++ name- True -> RunningWorkflows (xs \\ [name])-- liftIO $ withResource (RunningWorkflows undefined) delWF- when (sync s) $ liftIO $ syncCache- return v'----- | re-start the non finished workflows started for all initial values that are listed in the workflow list-restartWorkflows- ::(IResource a, Typeable a, IResource b, Typeable b)- => WorkflowList IO a b -- the list of workflows that implement the module- -> IO () -- Only workflows in the IO monad can be restarted with restartWorkflows-restartWorkflows map = do- liftIO (registerType :: IO Stat)- liftIO (registerType :: IO Control.Workflow.Queue)- liftIO (registerType :: IO String)- liftIO (registerType :: IO Integer)-- mw <- liftIO $ getResource ((RunningWorkflows undefined ) ) -- :: IO (Maybe(Stat a))- case mw of- Nothing -> return ()- Just (RunningWorkflows all) -> mapM_ start all- where-- start key1= do-- let key= case elemIndex '#' key1 of- Just n -> take n key1- Nothing -> key1-- let mf= lookup key map- if isNothing mf then return ()- else do- let f= fromJust mf- let st0= stat0{wfName = key1}- mst <- liftIO $ getResource st0- case mst of- Nothing -> error $ "restartWorkflows: not found "++ keyResource st0- Just st-> do- liftIO . forkIO $ runWF key f (fromIDyn . last $ versions st ) st{index=0,recover=True} >> return ()- return ()----- | change the logging policy (default is syncronous)--- Workflow uses the package TCache for logging--- for very fast workflow steps or when TCache is used also for other purposes , asyncronous is a better option-syncWrite:: (Monad m, MonadIO m)- => Bool -- ^True means syncronoys: changes are inmediately saved after each step- -> Int -- ^ number of seconds between saves when asyncronous- -> Int -- ^ size of the cache when async- -> WF Stat m () -- ^ in the workflow monad-syncWrite bool time maxsize= WF(\s -> do- when (bool== False) $ do- liftIO $ clearSyncCacheProc time defaultCheck maxsize- return ()- return (s{ sync= bool},()))-----getStep- :: (IResource 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)- else if i < 0 && i >= -stat then fromIDyn $ versions s !! (stat +i)- else error "getStep: wrong index")- )---- | return all the intermediate results. it is supposed that all the intermediate result have--- the same type.-getAll :: (IResource a, Typeable a, Monad m) => WF Stat m [a]-getAll = WF(\s -> return (s, map fromIDyn . take (state s+1) $ versions s))---- | log a message in the workflow history. I can be printed out with 'printWFhistory'-logWF :: (Monad m, MonadIO m) => String -> Workflow m ()-logWF str= WF (\s -> do- time <- liftIO $ getClockTime >>= toCalendarTime >>= return . calendarTimeToString- let str2 = time ++ ": "++ str- let (state1, index1, versions1)=- let stat= state s ; ind= index s in- if recover s && ind < stat- then (stat,ind +1, versions s)- else (stat +1, ind, toIDyn str2 : versions s)- return (s{versions= versions1, state= state1, index= index1}, ()) )------ | return the list of object keys that are running-getWFKeys :: String -> IO [String]-getWFKeys wfname= do- mwfs <- getResource $ RunningWorkflows undefined- case mwfs of- Nothing -> return []- Just (RunningWorkflows wfs) -> return $ map (tail . dropWhile (/= '#')) $ filter (isPrefixOf wfname) wfs---- | return the current state of the computation, in the IO monad-getWFHistory :: (IResource a) => String -> a -> IO (Maybe Stat)-getWFHistory wfname x= getResource stat0{wfName= wfname ++ "#" ++ keyResource x}---- | delete the workflow. Make sure that the workdlow is not running-delWFHistory :: IResource a => String -> a -> IO ()-delWFHistory wfname1 x=do- let wfname= wfname1 ++ "#" ++ keyResource x- let- doit [Just (RunningWorkflows wfs)] =- resources{ toAdd = [RunningWorkflows (wfs \\ [wfname])]- , toDelete= [stat0{wfName= wfname}] }- doit _ = error "delWFHistory: list of running workflows not found"- atomically $ withSTMResources[RunningWorkflows undefined] doit- syncCache----- | print the state changes along the workflow, that is, all the intermediate results-printHistory :: Stat -> IO ()-printHistory stat= do- putStrLn . runW $ showp $ Pretty stat- putStrLn "-----------------------------------"-- {-- mapM_ f . zip [1..] . reverse $ versions stat where- f :: (Int,IDynamic) -> IO()- f (n, ( IDynamic x))= do- putStr "Step "- putStr $ show n- putStr " "- putStrLn $ serialize x--}---data Pretty = Pretty Stat-instance Serialize Pretty where-- showp (Pretty (Stat wfName state index recover sync versions _))= do- name <- showp wfName- vers <- showElem (zip ( reverse $ take (length versions)[1..] ) versions ) ""- return $ "Workflow name= " ++ name ++ "\n" ++ vers-- where- showElem [] str= return $ str ++ "\n"- showElem ((n, IDynamic e):es) str= do- etext <- tshowp e- showElem es $ "Step " ++ show n ++ ": " ++ etext ++ "\n" ++ str-- readp = undefined---------------- event handling---------------reference :: IDynamic -> STM (TVar IDynamic)-reference x=do- mv <- getTVars [ x ]- case mv of- [Nothing] -> do- insertResources [x]- reference x-- [Just cl] -> return cl- where- insertResources xs= withDSTMResources [] $ const resources{toAdd= [x]}---- |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 longuer work--- inside the wokflow, it can be used by lifting it :--- do--- x <- step $ ..--- y <- step $ waitForData ...--- ..--waitForData :: (IResource a, Typeable a,IResource b, Typeable b)- => (b -> Bool) -- ^ The condition that the retrieved object must meet- -> a -- ^ a partially defined object for which keyResource can be extracted- -> IO b -- ^ return the retrieved object that meet the condition and has the given key-waitForData filter x= atomically $ waitForDataSTM filter x--waitForDataSTM :: (IResource a, Typeable a,IResource b, Typeable b)- => (b -> Bool) -- ^ The condition that the retrieved object must meet- -> a -- ^ a partially defined object for which keyResource can be extracted- -> STM b -- ^ return the retrieved object that meet the condition and has the given key-waitForDataSTM filter x= do- tv <- reference $ toIDyn x- do- dyn <- readTVar tv- case safeFromIDyn dyn of- Nothing -> retry- Just x ->- case filter x of- False -> retry- True -> return x---waitFor- :: (IResource a, Typeable a, IResource 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- :: (IResource a, Typeable a, IResource 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- -> STM b -- ^ The first event that meet the condition-waitForSTM filter wfname x= do- mtv <- getTVars [toIDyn stat0{wfName=wfname ++ "#" ++ keyResource x}] -- `debug` "**waitFor***"- case mtv of- [Nothing] -> error $ "workflow "++ wfname ++" not initialized for data " ++ keyResource x- [Just tv] ->- do- dyn <- readTVar tv- let Stat{ versions= d: _}= fromIDyn dyn- case safeFromIDyn d of- Nothing -> retry -- `debug` "waithFor retry Nothing"- Just x ->- case filter x of- False -> retry -- `debug` "waitFor false filter retry"- True -> return x -- `debug` "waitfor return"------- | start the timeout and return the flag to be monitored by 'waitUntilSTM'-getTimeoutFlag :: (MonadIO m)- => Integer -- ^ wait time in secods- -> Workflow m (TVar Bool) -- ^ the returned flag in the workflow monad-getTimeoutFlag t=do- 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- 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, in the STM monad.- This permits to compose timeouts with locks waiting for data.-- *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 $ readQueueSTM docQueue `orElse` waitUntilSTM flag >> return True- case ap of- False -> 'logWF' "False or timeout" >> correctWF doc- True -> do- @--}--waitUntilSTM :: TVar Bool -> STM()-waitUntilSTM tv = do-- b <- readTVar tv- if b == False then retry else return ()--waitUntil:: Integer -> IO()-waitUntil t= do- tnow <- getTimeSeconds- let delay | t-tnow < 0= 0- | t-tnow > (fromIntegral maxInt) = maxInt- | otherwise = fromIntegral $ t - tnow- threadDelay $ delay * 1000000- if t - tnow <= 0 then return () else waitUntil t---data Queue= Queue {name :: String, imp :: [IDynamic], out :: [IDynamic]} deriving (Typeable)--instance Serialize Queue where- showp (Queue name imp out)= do- sin<- showp imp- sout <- showp out- return $ "Queue " ++ show name ++ " " ++ sin ++ " " ++ sout-- readp = do- symbol "Queue"- name <- readp- sin <- readp- sout <- readp- return $ Queue name sin sout--instance IResource Queue where- keyResource (Queue name _ _)= "Queue#" ++ name- serialize x= runW $ showp x- deserialize str = runR readp str- defPath _= workflowsPath----- | delete elements from the Queue stack and return them in the IO monad-readQueue- :: (IResource a , Typeable a)- => String -- ^ Queue name- -> IO a -- ^ the returned elems-readQueue = atomically . readQueueSTM---- | delete elements from the Queue stack an return them. in the STM monad-readQueueSTM :: (IResource a , Typeable a) => String -> STM a-readQueueSTM queue = do- let qempty= Queue queue [] []- let empty= toIDyn qempty- reference empty -- make sure that the queue has been created- d <- withSTMResources [qempty] doit -- otherwise, it will not retry- releaseTVars [empty]- return $ fromIDyn d- where- doit [ Nothing] = Retry- doit [Just(Queue _ [] [])] = Retry- doit [Just(Queue _ imp [])] = doit [Just (Queue queue [] $ reverse imp)]-- doit [Just (Queue _ imp list)] =- resources { toAdd= [ Queue queue imp (tail list)]- , toReturn= head list }--unreadQueue :: (IResource a , Typeable a) => String -> a -> IO ()-unreadQueue queue x= atomically $ unreadQueueSTM queue x--unreadQueueSTM :: (IResource a , Typeable a) => String -> a -> STM ()-unreadQueueSTM queue x=- withSTMResources [Queue queue undefined undefined] $ \[r]-> resources{ toAdd= doit r}- where- doit Nothing = [Queue queue [] [ toIDyn x] ]- doit (Just(Queue _ imp out)) = [Queue queue imp ( toIDyn x : out) ]---- | insert an element on top of the Queue Stack-writeQueue :: (IResource a, Typeable a) => String -> a -> IO ()-writeQueue queue v = atomically $ writeQueueSTM queue v---- | Like writeQueue, but in the STM monad-writeQueueSTM :: (IResource a, Typeable a) => String -> a -> STM ()-writeQueueSTM queue v=- withSTMResources [Queue queue undefined undefined] $ \[r]-> resources{ toAdd= doit r}- where- doit Nothing = [Queue queue [toIDyn v] []]- doit (Just(Queue _ imp out)) = [Queue queue ( toIDyn v : imp) out]---isEmptyQueue = atomically . isEmptyQueueSTM--isEmptyQueueSTM :: String -> STM Bool-isEmptyQueueSTM queue= do- withDSTMResources [toIDyn $ Queue queue undefined undefined] doit- where- doit [ r]= resources{toReturn= ret} where-- ret=case r of- Nothing -> True- Just x -> case fromIDyn x of- Queue _ [] [] -> True- _ -> False---
+ Control/Workflow/Binary.hs view
@@ -0,0 +1,105 @@+{-# LANGUAGE+ OverlappingInstances+ , UndecidableInstances+ , ExistentialQuantification+ , ScopedTypeVariables+ , MultiParamTypeClasses+ , FlexibleInstances+ , FlexibleContexts+ , TypeSynonymInstances+ , DeriveDataTypeable+ , CPP+ #-}+{-# 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.++Thiis module uses Data.Binary serialization. Here the constraint @DynSerializer w r a@ is equivalent to+@Data.Binary a@++If you need to debug the workflow by reading the log or if you use largue structures that are subject of modifications along the workflow, as is the case+typically of multiuser workflows with documents, then use Text seriialization with "Control.Workflow.Text" instead+++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.Binary+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)@++-}++module Control.Workflow.Binary+ (+ Workflow -- a useful type name+, WorkflowList+, PMonadTrans (..)+, MonadCatchIO (..)++, throw+, Indexable(..)+, MonadIO(..)+-- * Start/restart workflows+, start+, exec+, exec1d+, exec1+, wfExec+, restartWorkflows+, WFErrors(..)+-- * Lifting to the Workflow monad+, step+, stepControl+, unsafeIOtoWF+-- * References to workflow log values+, WFRef+, getWFRef+, newWFRef+, stepWFRef+, readWFRef+, writeWFRef+-- * Workflow inspect+, getAll+, safeFromIDyn+, getWFKeys+, getWFHistory+, waitFor+, waitForSTM+-- * Persistent timeouts+, waitUntilSTM+, getTimeoutFlag+-- * Trace logging+, logWF+-- * Termination of workflows+, killThreadWF+, killWF+, delWF+, killThreadWF1+, killWF1+, delWF1+-- * Log writing policy+, syncWrite+, SyncMode(..)+)+where+import Control.Workflow.Binary.BinDefs++#include "Workflow.inc.hs"
+ Control/Workflow/Binary/BinDefs.hs view
@@ -0,0 +1,94 @@+{-# LANGUAGE++ MultiParamTypeClasses+ , FlexibleInstances+ , ScopedTypeVariables+ #-}+module Control.Workflow.Binary.BinDefs where+import Control.Workflow.GenSerializer+import Control.Workflow.IDynamic+import Control.Workflow.Stat+import Data.TCache.DefaultPersistence(Indexable(..))+import Data.Binary+import Data.Binary.Put+import Data.Binary.Get+import System.IO.Unsafe+import Data.IORef+import Data.ByteString.Lazy.Char8 as B hiding (index)+import Data.Map as M+import Control.Concurrent(ThreadId,forkIO)+import Data.Typeable+import Data.TCache+++instance Binary a => Serializer PutM Get a where+ serial = put+ deserial = get++instance RunSerializer PutM Get where+ runSerial = runPut+ runDeserial = runGet++instance Binary a => DynSerializer PutM Get a++--instance TwoSerializer PutM Get PutM Get () ()++instance Binary IDynamic where+ put (IDyn t) =+ case unsafePerformIO $ readIORef t of+ DRight x -> put . runSerial $ serial x+ DLeft (s, _) -> put s++ get = do+ s <- get+ return $ IDyn . unsafePerformIO . newIORef $ DLeft (s, (undefined, pack ""))++++instance Binary Stat where+ put (Running map)= do+ put (0 :: Word8)+ put $ Prelude.map (\(k,(w,_)) -> (k,w)) $ M.toList map++ put (Stat wfName state index recover versions _) = do+ put (1 :: Word8)+ put wfName+ put state+ put index+ put recover+ put versions++ get = do+ t <- get :: Get Word8+ case t of+ 0 -> do list <- get+ return . Running . M.fromList $ Prelude.map(\(k,w)-> (k,(w,Nothing))) list+ 1 -> do+ wfName <- get+ state <- get+ index <- get+ recover <- get+ versions <- get+ return $ Stat wfName state index recover versions Nothing++instance Binary ThreadId where+ put _= put $ pack "th"+ get = get >>= \(_ :: String) -> return $ unsafePerformIO . forkIO $ return ()+++instance Binary a => Binary (WFRef a) where+ put (WFRef n ref)= do+ put n+ put $ keyObjDBRef ref++ get= do+ n <- get+ k <- get+ return . WFRef n $ getDBRef k+++instance Indexable Stat where+ key s@Stat{wfName=name}= "Stat#" ++ name+ key (Running _)= keyRunning+ defPath= const "WorkflowState/bin/"+
+ Control/Workflow/Binary/Patterns.hs view
@@ -0,0 +1,107 @@+{-# LANGUAGE DeriveDataTypeable, FlexibleContexts, ScopedTypeVariables, CPP #-}+{-# OPTIONS -IControl/Workflow #-}++{- | This module contains monadic combinators that express+some workflow patterns.+see the docAprobal.hs example included in the package++This version uses Data.Binary serialization.+Here the constraint `DynSerializer w r a` is equivalent to `Data.Binary a`.++EXAMPLE:++This fragment below describes the approbal procedure of a document.+First the document reference is sent to a list of bosses trough a queue.+ithey return a boolean in a return queue ( askUser)+the booleans are summed up according with a monoid instance (sumUp)++if the resullt is false, the correctWF workflow is executed+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+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)+++@docApprobal :: Document -> Workflow IO ()+docApprobal doc = `getWFRef` \>>= docApprobal1+++docApprobal1 rdoc=+ return True \>>=+ log \"requesting approbal from bosses\" \>>=+ `sumUp` 0 (map (askUser doc rdoc) bosses) \>>=+ checkValidated \>>=+ log \"requesting approbal from superbosses or timeout\" \>>=+ `sumUp` (7*60*60*24) (map(askUser doc rdoc) superbosses) \>>=+ checkValidated1+++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)+ `step` . `pop` $ qdocApprobal (title doc)++log txt x = `logWF` txt >> return x++checkValidated :: Bool -> `Workflow` IO Bool+checkValidated val =+ case val of+ False -> correctWF (title doc) rdoc >> return False+ _ -> return True+++checkValidated1 :: Bool -> Workflow IO ()+checkValidated1 val = step $ do+ case val of+ False -> `push` qrejected doc+ _ -> `push` qapproved doc+ mapM (\u ->deleteFromQueue (quser u) rdoc) superbosses@+++-}++module Control.Workflow.Binary.Patterns(+-- * Low level combinators+split, merge, select,+-- * High level conbinators+vote, sumUp, Select(..))+ where+import Control.Concurrent.STM+import Data.Monoid+import Control.Concurrent.MonadIO+import qualified Control.Monad.CatchIO as CMC+import Control.Exception(SomeException)+import Control.Workflow.Binary+import Prelude hiding (catch)+import Control.Monad(when)+import Control.Exception.Extensible(Exception)+import Control.Workflow.GenSerializer+--import Debug.Trace++import Control.Workflow.Binary+import Control.Workflow.Stat(keyWF)+import Data.Typeable++import Data.Binary++instance Binary Select where+ put Select= put (1 :: Int)+ put Discard= put (2 :: Int)+ put FinishDiscard = put (3 :: Int)+ put FinishSelect = put (4 :: Int)++ get= do+ n <- get :: Get Int+ case n of+ 1 -> return Select+ 2 -> return Discard+ 3 -> return FinishDiscard+ 4 -> return FinishSelect+++#include "Patterns.inc.hs"
+ Control/Workflow/GenSerializer.hs view
@@ -0,0 +1,70 @@+-----------------------------------------------------------------------------+--+-- Module : Control.Workflow.GenSerializer+-- Copyright :+-- License : BSD3+--+-- Maintainer : agocorona@gmail.com+-- Stability : experimental+-- Portability :+--+-----------------------------------------------------------------------------++{-# OPTIONS+ -XMultiParamTypeClasses+ -XFunctionalDependencies+ -XFlexibleContexts+ -XFlexibleInstances+ -XUndecidableInstances+ -XScopedTypeVariables+ #-}++{- |+ This module includes the definition of a generic (de)serializer. This is used as a class constraints+ for the Workflow methods.++ Data.RefSerialize (defined in "Control.Workflow.Text.TextDefs") and Data.Binary ("Control.Workflow.Binary.BinDefs")+ are particular instances of thiis generic serializer.+-}++module Control.Workflow.GenSerializer where+import Data.ByteString.Lazy.Char8 as B+import Data.RefSerialize(Context)+import Data.Map as M+import Control.Concurrent+import System.IO.Unsafe++class (Monad writerm, Monad readerm)+ => Serializer writerm readerm a | a -> writerm readerm where+ serial :: a -> writerm ()+ deserial :: readerm a+++++class (DynSerializer w r a, DynSerializer w r b) => TwoSerializer w r a b+++instance (DynSerializer w r a, DynSerializer w r b) => TwoSerializer w r a b++class (DynSerializer w r a, DynSerializer w r b,DynSerializer w r c) => ThreeSerializer w r a b c+++instance (DynSerializer w r a, DynSerializer w r b, DynSerializer w r c) => ThreeSerializer w r a b c+++class (Monad writerm+ ,Monad readerm)+ => RunSerializer writerm readerm+ | writerm -> readerm+ , readerm -> writerm+ where+ runSerial :: writerm () -> ByteString+ runDeserial :: Serializer writerm readerm a => readerm a -> ByteString -> a++class (Serializer w r a, RunSerializer w r) => DynSerializer w r a | a -> w r where+ serialM :: a -> w ByteString+ serialM = return . runSerial . serial++ fromDynData :: ByteString ->(Context, ByteString) -> a+ fromDynData s _= runDeserial deserial s
+ Control/Workflow/IDynamic.hs view
@@ -0,0 +1,207 @@+{-# OPTIONS -XExistentialQuantification+ -XOverlappingInstances+ -XUndecidableInstances+ -XScopedTypeVariables+ -XDeriveDataTypeable+ -XTypeSynonymInstances+ -XIncoherentInstances+ -XOverloadedStrings+ -XMultiParamTypeClasses+ -XFunctionalDependencies+ -XFlexibleInstances #-}+{- |+IDynamic is a indexable and serializable version of Dynamic. (See @Data.Dynamic@). It is used as containers of objects+in the cache so any new datatype can be incrementally stored without recompilation.+IDimamic provices methods for safe casting, besides serializaton, deserialirezation and retrieval by key.+-} +module Control.Workflow.IDynamic where +import Data.Typeable +import Unsafe.Coerce +import System.IO.Unsafe+import Data.TCache +import Data.TCache.DefaultPersistence+import Data.RefSerialize+++import Data.ByteString.Lazy.Char8 as B+ +import Data.Word+import Numeric (showHex, readHex)+import Control.Exception(handle, SomeException, ErrorCall)+import Control.Monad(replicateM)+import Data.Word+import Control.Concurrent.MVar+import Data.IORef+import Data.Map as M(empty)+import Control.Workflow.GenSerializer+import Data.HashTable as HT+--import Debug.Trace++--a !> b = trace b a+++ +data IDynamic = IDyn (IORef IDynType)++data IDynType= forall a w r.(Typeable a, DynSerializer w r a)+ => DRight !a+ | DLeft !(ByteString ,(Context, ByteString))+++ 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+++++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"++-}+ +errorfied str str2= error $ str ++ ": IDynamic object not reified: "++ str2++ ++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,""))++-}+ +instance Show IDynamic where + show (IDyn r) =+ let t= unsafePerformIO $ readIORef r+ in case t of+ DRight x -> "IDyn " ++ ( unpack . runSerial $ serial x) ++ ")" + DLeft (s, _) -> "IDyn " ++ unpack s+++++ +toIDyn x= IDyn . unsafePerformIO . newIORef $ DRight x+ + +fromIDyn :: (Typeable a , DynSerializer m n a)=> IDynamic -> a+fromIDyn x=r where+ r = case safeFromIDyn x of+ Nothing -> error $ "fromIDyn: casting failure for data "+ ++ show x ++ " to type: "+ ++ (show $ typeOf r) + Just v -> v+++safeFromIDyn :: (Typeable a, DynSerializer m n a) => IDynamic -> Maybe a +safeFromIDyn (IDyn r)=unsafePerformIO $ do+ t<- readIORef r+ case t of+ DRight x -> return $ cast x++ DLeft (str, c) ->+ handle (\(e :: SomeException) -> return Nothing) $ -- !> ("safeFromIDyn : "++ show e)) $+ do+ let v= fromDynData str c+ 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)++
+ Control/Workflow/Patterns.inc.hs view
@@ -0,0 +1,172 @@++data 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`+split :: ( Typeable b+ , DynSerializer w r (Maybe b)+ , HasFork io+ , CMC.MonadCatchIO io)+ => [a -> Workflow io b] -> a -> Workflow io [ActionWF b]+split actions a = mapM (\ac ->+ do+ mv <- newWFRef Nothing+ fork (ac a >>= step . liftIO . atomically . writeWFRef mv . Just)+ r <- getWFRef+ return $ ActionWF mv r)++ actions++++-- | wait for the results and apply the cond to produce a single output in the Workflow monad+merge :: ( MonadIO io+ , Typeable a+ , Typeable b+ , TwoSerializer w r (Maybe a) b)+ => ([a] -> io b) -> [ActionWF a] -> Workflow io b+merge cond actions= mapM (\(ActionWF mv _) -> readWFRef1 mv ) actions >>= step . cond++readWFRef1 :: ( MonadIO io+ , DynSerializer w r (Maybe 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+++data Select+ = Select+ | Discard+ | FinishDiscard+ | FinishSelect+ deriving(Typeable, Read, Show)++instance Exception Select++-- | 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 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.+--+-- This is necessary for the modelization of real-life institutional cycles such are political elections+-- timeout of 0 means no timeout.+select ::+ ( TwoSerializer w r (Maybe a) [a]+ , Typeable a+ , HasFork io+ , CMC.MonadCatchIO io)+ => Integer+ -> (a -> io Select)+ -> [ActionWF a]+ -> Workflow io [a]+select timeout check actions= do+ res <- newMVar []+ flag <- getTimeoutFlag timeout+ parent <- myThreadId+ checks <- newEmptyMVar+ count <- newMVar 1+ let process = 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+ FinishSelect -> do+ addRes r+ throwTo parent FinishDiscard++ n <- CMC.block $ do+ n <- takeMVar count+ putMVar count (n+1)+ return n++ if ( n == length actions)+ then throwTo parent FinishDiscard+ else return ()++ `CMC.catch` (\(e :: Select) -> throwTo parent e)++ do+ ws <- mapM ( fork . check') actions+ putMVar checks ws++ liftIO $ atomically $ do+ v <- readTVar flag -- wait fo timeout+ case v of+ False -> retry+ True -> return ()+ throw FinishDiscard+ where++ addRes r= 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++ stepControl $ CMC.catch process -- (WF $ \s -> process >>= \ r -> return (s, r))+ (\(e :: Select)-> do+ 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+--+-- @+-- vote timeout actions comp x=+-- split actions x >>= select timeout (const $ return Select) >>= comp+-- @+vote+ :: ( TwoSerializer w r (Maybe b) [b]+ , Typeable b+ , HasFork io+ , CMC.MonadCatchIO io)+ => Integer+ -> [a -> Workflow io b]+ -> ([b] -> Workflow io c)+ -> a+ -> Workflow io c+vote timeout actions comp x=+ split actions x >>= select timeout (const $ return Select) >>= comp+++-- | sum the outputs of a list of workflows according with its monoid definition+--+-- @ sumUp timeout actions = vote timeout actions (return . mconcat) @+sumUp+ :: ( TwoSerializer w r (Maybe b) [b]+ , Typeable b+ , Monoid b+ , HasFork io+ , CMC.MonadCatchIO io)+ => Integer+ -> [a -> Workflow io b]+ -> a+ -> Workflow io b+sumUp timeout actions = vote timeout actions (return . mconcat)+++
+ Control/Workflow/Stat.hs view
@@ -0,0 +1,78 @@+{-# OPTIONS -XUndecidableInstances+ -XDeriveDataTypeable+ -XTypeSynonymInstances+ -XExistentialQuantification+ -XMultiParamTypeClasses+ -XFlexibleInstances+ -XOverloadedStrings++ #-}+module Control.Workflow.Stat where++import Data.TCache++import Data.ByteString.Lazy.Char8(pack, unpack)+import System.IO.Unsafe+import Data.Typeable+import qualified Data.Map as M+import Control.Concurrent(ThreadId)+import Control.Concurrent.STM(TVar, newTVarIO)+import Data.IORef+import Control.Workflow.GenSerializer+import Control.Workflow.IDynamic+import Control.Monad(replicateM)+import Data.TCache.DefaultPersistence+import Data.ByteString.Lazy.Char8 hiding (index)+import Control.Workflow.IDynamic+++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)}+ deriving (Typeable)++stat0 = Stat{ wfName="", state=0, index=0, recover=False, versions = []+ , timeout= Nothing}++-- 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+++data WFRef a= WFRef !Int !(DBRef Stat) deriving (Typeable, Show)++instance Indexable (WFRef a) where+ key (WFRef n ref)= keyObjDBRef ref++('#':show n)+++++++instance (Serializer w r a, RunSerializer w r) => Serializable a where+ serialize = runSerial . serial++ deserialize = runDeserial deserial+++++keyRunning= "Running"
+ Control/Workflow/Text.hs view
@@ -0,0 +1,115 @@+{-# LANGUAGE+ OverlappingInstances+ , UndecidableInstances+ , ExistentialQuantification+ , ScopedTypeVariables+ , MultiParamTypeClasses+ , FlexibleInstances+ , FlexibleContexts+ , TypeSynonymInstances+ , DeriveDataTypeable+ , CPP+ #-}+{-# 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.++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.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)@++-}++module Control.Workflow.Text+(+ Workflow -- a useful type name+, WorkflowList+, PMonadTrans (..)+, MonadCatchIO (..)++, throw+, Indexable(..)+, MonadIO(..)+-- * Start/restart workflows+, start+, exec+, exec1d+, exec1+, wfExec+, restartWorkflows+, WFErrors(..)+-- * Lifting to the Workflow monad+, step+, stepControl+, unsafeIOtoWF+-- * References to intermediate values in the workflow log+, WFRef+, getWFRef+, newWFRef+, stepWFRef+, readWFRef+, writeWFRef+-- * Workflow inspect+, getAll+, safeFromIDyn+, getWFKeys+, getWFHistory+, waitFor+, waitForSTM+-- * Persistent timeouts+, waitUntilSTM+, getTimeoutFlag+-- * Trace logging+, logWF+-- * Termination of workflows+, killThreadWF+, killWF+, delWF+, killThreadWF1+, killWF1+, delWF1+-- * Log writing policy+, syncWrite+, SyncMode(..)+-- * Print log history+, printHistory+)+where++import Control.Workflow.Text.TextDefs++#include "Workflow.inc.hs"+
+ Control/Workflow/Text/Patterns.hs view
@@ -0,0 +1,97 @@+{-# LANGUAGE DeriveDataTypeable+ , FlexibleContexts+ , ScopedTypeVariables+ , CPP+ #-}+{-# 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:++This fragment below describes the approbal procedure of a document.+First the document reference is sent to a list of bosses trough a queue.+ithey return a boolean in a return queue ( askUser)+the booleans are summed up according with a monoid instance (sumUp)++if the resullt is false, the correctWF workflow is executed+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+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)+++@docApprobal :: Document -> Workflow IO ()+docApprobal doc = `getWFRef` \>>= docApprobal1+++docApprobal1 rdoc=+ return True \>>=+ log \"requesting approbal from bosses\" \>>=+ `sumUp` 0 (map (askUser doc rdoc) bosses) \>>=+ checkValidated \>>=+ log \"requesting approbal from superbosses or timeout\" \>>=+ `sumUp` (7*60*60*24) (map(askUser doc rdoc) superbosses) \>>=+ checkValidated1+++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)+ `step` . `pop` $ qdocApprobal (title doc)++log txt x = `logWF` txt >> return x++checkValidated :: Bool -> `Workflow` IO Bool+checkValidated val =+ case val of+ False -> correctWF (title doc) rdoc >> return False+ _ -> return True+++checkValidated1 :: Bool -> Workflow IO ()+checkValidated1 val = step $ do+ case val of+ False -> `push` qrejected doc+ _ -> `push` qapproved doc+ mapM (\u ->deleteFromQueue (quser u) rdoc) superbosses@++-}++module Control.Workflow.Text.Patterns(+-- * Low level combinators+split, merge, select,+-- * High level conbinators+vote, sumUp, Select(..)+) 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.Text+import Data.Typeable+import Prelude hiding (catch)+import Control.Monad(when)+import Control.Exception.Extensible (Exception)+import Control.Workflow.GenSerializer+import Control.Workflow.Stat+import Debug.Trace+import Data.TCache++a !> b = trace b a++#include "Patterns.inc.hs"+
+ Control/Workflow/Text/TextDefs.hs view
@@ -0,0 +1,151 @@+{-# LANGUAGE++ MultiParamTypeClasses+ , FlexibleInstances+ , UndecidableInstances+++ #-}+module Control.Workflow.Text.TextDefs where+import Control.Workflow.IDynamic+import Control.Workflow.GenSerializer+import Data.RefSerialize+import System.IO.Unsafe+import Data.TCache.DefaultPersistence(Indexable(..))+import Data.IORef+import Unsafe.Coerce+import Data.ByteString.Lazy.Char8 as B hiding (index)+import Control.Workflow.Stat+import Data.Map as M+import Control.Concurrent+import Data.TCache++instance Serialize a => Serializer ST ST a where+ serial = showp+ deserial = readp++instance RunSerializer ST ST where+ runSerial = runW+ runDeserial = runR++instance Serialize a => DynSerializer ST ST a where+ serialM = showps+ fromDynData s c= runRC c readp s+++instance Serialize IDynamic where+++ showp (IDyn t)=+ case unsafePerformIO $ readIORef t of+ DRight x -> do+ insertString $ pack dynPrefix+ showpx <- unsafeCoerce $ serialM x+ showpText . fromIntegral $ B.length showpx+ insertString showpx+++ DLeft (showpx,_) -> -- error $ "IDynamic not reified :: "++ unpack showpx+ do+ insertString $ pack dynPrefix+ showpText 0++++ readp = do+ symbol dynPrefix+ n <- readpText+ s <- takep n+ c <- getContext+ return . IDyn . unsafePerformIO . newIORef $ DLeft ( s, c)+ <?> "IDynamic"+++instance Serialize Stat where+ showp (Running map)= do+ insertString $ 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+++ readp = choice [rStat, rWorkflows] where+ rStat= do+ symbol "Stat"+ wfName <- stringLiteral+ state <- integer >>= return . fromIntegral+ index <- integer >>= return . fromIntegral+ recover <- bool+ versions <- readp+ return $ Stat wfName state index recover versions Nothing+ <?> "Stat"++ rWorkflows= do+ symbol "Running"+ list <- readp+ return $ Running $ M.fromList $ Prelude.map(\(k,w)-> (k,(w,Nothing))) list+ <?> "RunningWoorkflows"+++instance Serialize ThreadId where+ showp th= insertString . pack $ show th+ readp = (readp :: ST ByteString) >> (return . unsafePerformIO . forkIO $ return ())+++newtype Pretty = Pretty Stat++instance Show Pretty where+ show= unpack . runW . sp+ where+ sp (Pretty (Stat wfName state index recover versions _ ))= do+ insertString $ pack "Workflow name= "+ showp wfName+ insertString $ pack "\n"+ showElem $ Prelude.reverse $ (Prelude.zip ( Prelude.reverse [1..] ) versions )+++ showElem :: [(Int,IDynamic)] -> ST ()+ showElem [] = insertChar '\n'+ showElem ((n, dyn):es) = do+ showp $ pack "Step "+ showp n+ showp $ pack ": "+ showp dyn+ insertChar '\n'+ showElem es+++statPrefix= "Stat#"+instance Indexable Stat where+ key s@Stat{wfName=name}= statPrefix ++ name+ key (Running _)= keyRunning+ defPath= const "WorkflowState/Text/"++wFRefStr = "WFRef"++instance Serialize (WFRef a) where+ showp (WFRef n ref)= do+ insertString $ pack wFRefStr+ showp n+ showp $ keyObjDBRef ref++ readp= do+ symbol wFRefStr+ n <- readp+ 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 "-----------------------------------"++
+ Control/Workflow/Workflow.inc.hs view
@@ -0,0 +1,898 @@+++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 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)+++import System.IO(hPutStrLn, stderr)+import Data.List(elemIndex)+import Data.Maybe(fromJust, isNothing, isJust, mapMaybe)+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.DefaultPersistence+import Control.Workflow.GenSerializer+import Control.Workflow.IDynamic+import Unsafe.Coerce+import Control.Workflow.Stat++--import Debug.Trace+--a !> b= trace b a+++type Workflow m l= WF Stat m l -- not so scary++type WorkflowList m a b= [(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+ (s3, x') <- fun s1+ return (s3, 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+ , DynSerializer w r a+ , Typeable a)+ => PMonadTrans (WF Stat) m a+ where+ plift = step++-- | An instance of MonadTrans is an instance of PMonadTrans+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+++{- | adapted from MonadCatchIO-mtl. Workflow need 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++{-++{-+-- | 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 (TwoSerializer w r () 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)++++instance (HasFork 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))+++++-- | 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, TwoSerializer w r () 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 :: (TwoSerializer w r () b, Typeable b+ ,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 ()+ liftIO syncIt -- !> str++++-- | a version of exec with no seed parameter.+exec1 :: ( TwoSerializer w r () a, Typeable a+ , 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, TwoSerializer w r a b, Typeable a+ , Typeable b+ , 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)+ syncIt+ CMC.throw e )+++++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+++++instance Indexable String where+ key= id++instance Indexable Int where+ key= show++instance Indexable Integer where+ key= show+++instance Indexable () where+ key= show++-- | lifts a monadic computation to the WF monad, and provides transparent state loging and resuming of computation+step :: ( Monad m+ , MonadIO m+ , DynSerializer w r 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+ , DynSerializer w r a+ , Typeable a)+ => m a+ -> Workflow m a+stepControl= stepControl1 True++stepControl1 :: ( Monad m+ , MonadIO m+ , DynSerializer w r a+ , Typeable a)+ => Bool -> 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 = getDBRef $ key s''+ s'<- liftIO . atomically $ do+ s <- if isControl+ then readDBRef sref >>= justify ("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}++ writeDBRef sref s'+ return s'+ liftIO syncIt+ return (s', x') )++justify str Nothing = error str+justify _ (Just x) = return x+++++-- | 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+ :: (Monad m+ , MonadIO m+ , Indexable a+ , TwoSerializer w r a b+ , Typeable a+ , Indexable b+ , Typeable b)+ => String -- ^ name thar 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+ Right (name, f, stat) ->+ --to reify the first Dynamic obhect+ runWF name (f . fromIDyn . Prelude.head $ versions stat) stat >>= return . Right++ 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++instance Show WFErrors where+ show NotFound= "Not Found"+ show AlreadyRunning= "Already Running"+ show Timeout= "Timeout"+ show (Exception e)= "Exception: "++ show e++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+ if is not runing, restart it+else start anew.+-}+++getState :: (Monad m, MonadIO m, Indexable a, DynSerializer w r 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+ let stat1= stat0{wfName= key , versions= [toIDyn v]}+ case M.lookup key map of+ Nothing -> do -- no workflow started for this object+ mythread <- unsafeIOToSTM $ myThreadId+ writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map+ withSTMResources ([] :: [Stat]) $+ \_-> resources{toAdd=[stat1],toReturn= 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+ withSTMResources[stat1] $+ \mst->+ let stat'= case mst of+ [Nothing] -> error $ "Workflow not found: "++ key+ [Just s] -> s{index=0,recover= True}+ in resources{toAdd=[stat'],toReturn = Right (key, f, stat') }++syncIt= do+ (sync,_) <- atomically $ readTVar tvSyncWrite+ when (sync ==Synchronous) syncCache++runWF :: (Monad m,MonadIO m+ , DynSerializer w r 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+ 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+ writeDBRef tvRunningWfs . Running $ M.delete n map -- `debug` "clearFromRunningList"+++-- | re-start the non finished workflows in the list, for all the initial values that they may have been called+restartWorkflows+ :: (TwoSerializer w r a b, Typeable a+ , Indexable b, Typeable b)+ => WorkflowList IO a 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 key1= key ++ "#" ++ kv+ let mf= Prelude.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 . Prelude.last $ versions st )) st{index=0,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)+++-- | 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))+++-- | return the list of object keys that are running for a workflow+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, DynSerializer w r a) => String -> a -> IO (Maybe Stat)+getWFHistory wfname x= getResource stat0{wfName= keyWF wfname x}++++-- | kill the executing thread if not killed, but not its state.+-- `exec` `start` or `restartWorkflows` will continue the workflow+killThreadWF :: ( Indexable a+ , DynSerializer w r a++ , Typeable a+ , MonadIO m)+ => String -> a -> m()+killThreadWF wfname x= do+ let name= keyWF wfname x+ killThreadWFm name >> return ()++-- | a version of `KillThreadWF` for workflows started wit no parameter by `exec1`+killThreadWF1 :: MonadIO m => String -> m()+killThreadWF1 name= killThreadWF name ()++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+killWF :: (Indexable a,MonadIO m) => String -> a -> m ()+killWF name1 x= do+ let name= keyWF name1 x+ map <- killThreadWFm name+ liftIO . atomically . writeDBRef tvRunningWfs . Running $ M.delete name map+ return ()++-- | a version of `KillWF` for workflows started wit no parameter by `exec1`+killWF1 :: MonadIO m => String -> m ()+killWF1 name = killWF name ()++-- | 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+ , Typeable a)+ => String -> a -> m()+delWF name1 x= liftIO $ do+ let name= keyWF name1 x+ mrun <- atomically $ readDBRef tvRunningWfs+ case mrun of+ Nothing -> return()+ Just (Running map) -> do+ atomically . writeDBRef tvRunningWfs . Running $! M.delete name map+ atomically . withSTMResources [] $ const resources{ toDelete= [stat0{wfName= name}] }++-- | a version of `delWF` for workflows started wit no parameter by `exec1`+delWF1 :: MonadIO m=> String -> m()+delWF1 name= delWF name ()+++clearRunningFlag name= liftIO $ atomically $ do+ mrun <- readDBRef tvRunningWfs+ case mrun of+ 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+ return (map,Just th)+ Nothing ->+ return (map, Nothing)++-- | Return the the workflow reference to the last logged result , usually, the last result stored by `step`.+-- wiorkflow references can be can be accessed (to its content) 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 :: ( DynSerializer w r a+ , Typeable a+ , MonadIO m)+ => Workflow m (WFRef a)+getWFRef = WF (\s -> do+ let n= if recover s then index s else state s+ let ref = WFRef n (getDBRef $ key s)+++ return (s,ref))++-- | Execute an step but return a reference to the result instead of the result itself+--+-- @stepWFRef exp= `step` exp >>= `getWFRef`@+stepWFRef :: ( DynSerializer w r a+ , Typeable a+ , MonadIO m)+ => m a -> Workflow m (WFRef a)+stepWFRef exp= step exp >> getWFRef++-- | Log a value and return a reference to it.+--+-- @newWFRef x= `step` $ return x >>= `getWFRef`@+newWFRef :: ( DynSerializer w r a+ , Typeable a+ , MonadIO m)+ => a -> Workflow m (WFRef a)+newWFRef x= step (return x) >> getWFRef++-- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad+readWFRef :: ( DynSerializer w r a+ , Typeable a)+ => WFRef a+ -> STM (Maybe a)+readWFRef (WFRef n ref)= do+ 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+++-- | 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:+--+-- @worflow= exec1 "wf" do+-- r <- `stepWFRef` expr+-- `push` \"queue\" r+-- back <- `pop` \"queueback\"+-- ...+-- @++writeWFRef :: ( DynSerializer w r 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: " ++ 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++ writeDBRef ref s{ versions= elems'}+++++-- | 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 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 longuer 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 kwaitForData filter x= atomically $ waitForDataSTM filter x+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 untiil a condition is met.+waitFor+ :: ( Indexable a, TwoSerializer w r a 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, TwoSerializer w r a 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+ -> 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***"++ 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+ Nothing -> retry -- `debug` "waithFor retry Nothing"+ Just x ->+ case filter x of+ False -> retry -- `debug` "waitFor false filter retry"+ True -> return x -- `debug` "waitfor return"+++++-- | 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+-- no matter if the workflow is restarted. The time that the worlkflow has been stopped count also.+-- 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+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+ (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))++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 ...+ @+-}+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)+++wait :: Integer -> IO()+wait delta= do+ let delay | delta < 0= 0+ | delta > (fromIntegral maxInt) = maxInt+ | otherwise = fromIntegral $ delta+ threadDelay $ delay * 1000000+ if delta <= 0 then return () else wait $ delta - (fromIntegral delay )++
+ Data/Persistent/Queue/Binary.hs view
@@ -0,0 +1,58 @@+{-# OPTIONS -XDeriveDataTypeable+ -XTypeSynonymInstances+ -XMultiParamTypeClasses+ -XExistentialQuantification+ -XOverloadedStrings+ -XFlexibleInstances+ -XUndecidableInstances+ -XFunctionalDependencies+ -XFlexibleContexts+ -XIncoherentInstances+ -IControl/Workflow+ -XCPP #-}+{-# OPTIONS -IData/Persistent/QUeue #-}+{- |+This module implements a persistent, transactional collection with Queue interface as well as indexed access by key+This module uses `Data.Binary` for serialization.++Here @QueueConstraints w r a@ means @Data.Binary.Binary a@++For optimal (de)serialization if you store in the queue different versions of largue structures , for+example, documents you better use "Data.RefSerialize" and "Data.Persistent.Queue.Text" Instead.++-}+module Data.Persistent.Queue.Binary(+RefQueue(..), getQRef,+pop,popSTM,pick,Data.Persistent.Queue.Binary.flush, flushSTM,+pickAll, pickAllSTM, push,pushSTM,+pickElem, pickElemSTM, readAll, readAllSTM,+deleteElem, deleteElemSTM,+unreadSTM,Data.Persistent.Queue.Binary.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.IORef++import Data.ByteString.Lazy.Char8++import Control.Workflow.GenSerializer+ +import Data.Binary+import Data.Binary.Put+import Data.Binary.Get++instance SerialiserString PutM Get where+ serialString= put+ deserialString= get+++instance Indexable (Queue a) where+ key (Queue k _ _)= queuePrefix ++ k+ defPath _= "WorkflowState/Binary/"++#include "Queue.inc.hs"
+ Data/Persistent/Queue/Queue.inc.hs view
@@ -0,0 +1,205 @@+++data Queue a= Queue {name :: String, imp :: [a], out :: [a]} deriving (Typeable)++class (Monad writerm+ ,Monad readerm)+ => SerialiserString writerm readerm+ | writerm -> readerm+ , readerm -> writerm+ where+ serialString :: String -> writerm ()+ deserialString :: readerm String++class ( Serializer w r [a]+ , SerialiserString w r+ , RunSerializer w r)+ => QueueConstraints w r a++instance ( Serializer w r [a]+ , SerialiserString w r+ , RunSerializer w r)+ => QueueConstraints w r a++instance (Serializer w r [a]+ , SerialiserString w r)+ => Serializer w r (Queue a) where+ serial (Queue n i o)= serialString n >> serial i >> serial o+ deserial= do+ n <- deserialString+ i <- deserial+ o <- deserial+ return $ Queue n i o+++++queuePrefix= "Queue#"+lenQPrefix= Prelude.length queuePrefix++++instance QueueConstraints w r a => Serializable (Queue a ) where+ serialize = runSerial . serial+ deserialize = runDeserial deserial++-- | a queue reference+type RefQueue a= DBRef (Queue a)++unreadSTM :: (Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a -> IO Bool+isEmpty = atomically . isEmptySTM++isEmptySTM :: (Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => String -> RefQueue a+getQRef n = getDBRef . key $ Queue n undefined undefined+++-- | empty the queue (factually, it is deleted)+flush :: (Typeable a, QueueConstraints w r a) => RefQueue a -> IO ()+flush = atomically . flushSTM++-- | version in the STM monad+flushSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> STM ()+flushSTM tv= delDBRef tv++-- | read the first element in the queue and delete it (pop)+pop+ :: (Typeable a, QueueConstraints w r a) => RefQueue a -- ^ Queue name+ -> IO a -- ^ the returned elems+pop tv = atomically $ popSTM tv+++readQRef :: (Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a+ -> STM a+popSTM tv=do+ dx <- readQRef tv+ doit dx++ where+ --doit :: (Typeable 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, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a -> a -> IO ()+push tv v = atomically $ pushSTM tv v++-- | version in the STM monad+pushSTM :: (Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a -> IO [a]+pickAll= atomically . pickAllSTM++-- | version in the STM monad+pickAllSTM :: (Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a -> String -> IO(Maybe a)+pickElem tv key= atomically $ pickElemSTM tv key++-- | version in the STM monad+pickElemSTM :: (Indexable a,Typeable a, QueueConstraints w r 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, QueueConstraints w r a)+ => RefQueue a -> a -> IO()+updateElem tv x = atomically $ updateElemSTM tv x++-- | version in the STM monad+updateElemSTM :: (Indexable a,Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a -> IO [a]+readAll= atomically . readAllSTM++-- | a version in the STM monad+readAllSTM :: (Typeable a, QueueConstraints w r 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, QueueConstraints w r a) => RefQueue a-> a -> IO ()+deleteElem tv x= atomically $ deleteElemSTM tv x++-- | verison in the STM monad+deleteElemSTM :: (Typeable a,Indexable a,QueueConstraints w r 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
+ Data/Persistent/Queue/Text.hs view
@@ -0,0 +1,60 @@+{-# OPTIONS -XDeriveDataTypeable+ -XTypeSynonymInstances+ -XMultiParamTypeClasses+ -XExistentialQuantification+ -XOverloadedStrings+ -XFlexibleInstances+ -XUndecidableInstances+ -XFunctionalDependencies+ -XFlexibleContexts+ -XIncoherentInstances+ -IControl/Workflow+ -XCPP+ #-}+{-# 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.Text(+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 Control.Workflow.GenSerializer++import Debug.Trace++a !> b= trace b a++instance SerialiserString ST ST where+ serialString = showp+ deserialString = readp++instance Indexable (Queue a) where+ key (Queue k _ _)= queuePrefix ++ k+ defPath _= "WorkflowState/Text/"+++#include "Queue.inc.hs"++
+ Demos/Fact.hs view
@@ -0,0 +1,55 @@+{-# OPTIONS -XDeriveDataTypeable #-}+-- this program imput numbers and calculate their factorials. The workflow control a record all the inputs and outputs+-- so that when the program restart, all the previous results are shown.+-- if the program abort by a runtime error or a power failure, the program will still work+-- enter 0 for exit and finalize the workflow (all the intermediate data will be erased)+-- enter any alphanumeric character for aborting and then re-start.++module Main where+import Control.Workflow.Binary+import Data.Typeable+import Data.Binary+import Data.Maybe+++fact 0 =1+fact n= n * fact (n-1)+++-- now the workflow versión+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+ n <- get+ v <- get+ return $ Fact n v+++factorials = do+ all <- getAll+ let lfacts = mapMaybe safeFromIDyn all :: [Fact]+ unsafeIOtoWF $ putStrLn "Factorials calculated so far:"+ unsafeIOtoWF $ mapM (\fct -> print fct) lfacts+ factLoop (Fact 0 1)+ where+ factLoop fct= do+ nf <- plift $ do -- plift == step+ putStrLn "give me a number if you enter a letter or 0, the program will abort. Then, please restart to see how the program continues"+ str<- getLine+ let n= read str :: Integer -- if you enter alphanumeric characters the program will abort. please restart+ let fct= fact n+ print fct+ return $ Fact n fct++ case nf of+ Fact 0 _ -> do+ unsafeIOtoWF $ print "bye"+ return (Fact 0 0)+ _ -> factLoop nf+++main = exec1 "factorials" factorials
+ Demos/Inspect.hs view
@@ -0,0 +1,92 @@+{-# LANGUAGE ScopedTypeVariables #-}+-- example of the Workflow package .+-- This demo shows inter-workflow communications.+-- two workflows that interact by inspecting its other's state. One ask the user for a numbers.+-- When the total number of tries+-- is exhausted, update Data with termination and ends.+---It can end when termination comes from the other workflow. The other wait for "5", print+-- a message , update Data with termination to the other and finish. t+-- you can break the code at any moment. The flow will re-start in the last interrupted point+-- For bugs, questions, whatever, please email me: Alberto Gómez Corona agocorona@gmail.com++module Main where+import Control.Workflow.Text+--import Debug.Trace+--import Data.Typeable+import Control.Concurrent+import Control.Exception+import System.Exit+import Control.Concurrent.STM+++--debug a b = trace b a++++-- start_ if the WF was in a intermediate state, restart it+-- A workflow state is identified by:+-- the name of the workflow+-- the key of the object whose worflow was called++main= do+ forkIO $ exec1 "wait" wait >> return ()+ exec1 "hello-ask" hello+ threadDelay 1000000+ delWF1 "wait"+ delWF1 "hello-ask"++++-- ask for numbers or "end". Inspect the step values of the other workflow+-- and exit if the value is "good"+hello :: Workflow IO ()+hello = do+ unsafeIOtoWF $ do+ putStrLn ""+ putStrLn "At any step you can break and re-start the program"+ putStrLn "The program will restart at the interrupted step."+ putStrLn ""+ --syncWrite Synchronous -- this is the default+ name <- step $ do+ print "what is your name?"+ getLine+ step $ putStrLn $ "hello Mr "++ name+ loop 0 name+ where+ loop i name=do+ unsafeIOtoWF $ threadDelay 100000+ str <- step $ do+ putStrLn $ "Mr "++name++ " this is your try number "++show (i+1) ++". Guess my number, press \"end\" to finish"+ getLine+ flag <- getTimeoutFlag 1+ -- waith for any character in any step on the "wait" workflow within one second span+ let anyString :: String -> Bool+ anyString = const True+ s <- step . atomically $ (waitForSTM anyString "wait" () >>= return . Just)+ `orElse`+ (waitUntilSTM flag >> return Nothing )+ case s of+ Just "good" -> return ()+ _ -> loop (i+1) name+++-- wait for a "5" or "end" in any step of the "hello" workflow, put an "good" and exit+wait :: Workflow IO ()+wait = do+ let+ filter "5" = True+ filter "end" = True+ filter _ = False+ step $ do+ r <- waitFor filter "hello-ask" () -- wait the other thread to store an object with the same key than Try 0 ""+ -- with the string "5" or termination+ case r of+ "5" -> print "done ! " >> return "good" -- put exit in the WF state+ "end" -> print "end received. Bye" >> return "good" -- put exit in the WF state++ -- wait for the inspection of the other workflow+ unsafeIOtoWF $ threadDelay 500000++++
+ Demos/docAprobal.hs view
@@ -0,0 +1,386 @@+{-# LANGUAGE DeriveDataTypeable, ScopedTypeVariables #-}+{-+ This program is an example of simple workflow management. Once a document+ is created by the user, a workflow controls two levels of approbal (boss and superboss) trough+ messages to the presentation layer of the three different user roles.++ A document is created by the user "user", then is validated by two bosses and thwo super bosses.+ If any of the two dissapprobe, the document is sent to the user to modify it.++ This program can handle as many document workflows as you like simultaneously.++ Workflow patterns and queue communication primitives are used.++ The second level of approbal has a timeout . The seralization of the document is+ trough the Serialize class of the RefSerialize package.++ approbed and dissapprobed documents are stored in their respective queues++ When te document title is modified, the workflow launches a new workflow with the new+ document and stops.++++-}+import Control.Workflow.Text++import Data.Persistent.Queue.Text+import Control.Workflow.Text.Patterns++import Data.Typeable+import System.Exit+import Data.List (find)+import Data.Maybe(fromJust)+import Control.Monad (when)+import Control.Concurrent ( forkIO)+import GHC.Conc ( atomically)+import Data.RefSerialize+import Data.TCache(syncCache)+import Data.ByteString.Lazy.Char8(pack)++import Data.Monoid+++import Debug.Trace+++(!>) a b= trace b a++data Document=Document{title :: String , text :: [String]} deriving (Read, Show,Eq,Typeable)++instance Indexable Document where+ key (Document t _)= "Doc#"++ t++instance Serialize Document where+ showp (Document title text)= do+ insertString $ pack "Document"+ showp title+ rshowp text+++ readp= do+ symbol "Document"+ title <- readp+ text <- rreadp+ return $ Document title text++--instance Binary Document where+-- put (Document title text)= do+-- put title+-- put text+-- get= do+-- title <- get+-- text <- get+-- return $ Document title text++++user= "user"++approved = "approved"+rejected = "rejected"++quser :: String -> RefQueue (WFRef Document)+quser user= getQRef user++qdoc :: String -> RefQueue Document+qdoc doc = getQRef doc++qdocApprobal :: String -> RefQueue Bool+qdocApprobal doc = getQRef doc+++qapproved :: RefQueue Document+qapproved = getQRef approved++qrejected :: RefQueue Document+qrejected = getQRef rejected+++++main = do+ -- restart the interrupted document approbal workflows (if necessary)++ restartWorkflows [("docApprobal",docApprobal)]++ putStrLn "\nThis program is an example of simple workflow management; once a document is created a workflow thread controls the flow o mail messages to three different users that approbe or disapprobe and modify the document"+ putStrLn "A document is created by the user, then is validated by the boss and the super boss. If any of the two dissapprobe, the document is sent to the user to modify it."+ putStrLn "\n please login as:\n 1- user\n 2- boss1\n 3- boos2\n 4- super boss1\n 5- super boss2\n\n Enter the number"++ n <- getLine+ case n of+ "1" -> userMenu+ "2" -> aprobal "boss1"+ "3" -> aprobal "boss2"+ "4" -> aprobal "superboss1"+ "5" -> aprobal "superboss2"++++bosses= ["boss1", "boss2"]+superbosses= ["superboss1", "superboss2"]++-- used by sumUp to sum the boolean "votes"+-- in this case OR is used+instance Monoid Bool where+ mappend = (||)+ mempty= False++{-+ the approbal procedure of a document below.+First the document reference is sent to a list of bosses trough a queue.+they return a boolean trough a return queue ( askUser)+the booleans are summed up according with a monoid instance by sumUp++in checkValidated, if the resullt is false, the correctWF workflow is executed+If the result is True, the pipeline continues to the next stage++the next stage is the same process with a new list of users (superbosses).+This time, there is a timeout of one day That time counts even if the program is+not running. the result of the users that voted is summedup according with the+same monoid instance++in chechValidated1, 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)++-}++docApprobal :: Document -> Workflow IO ()+docApprobal doc = getWFRef >>= docApprobal1+ where+ -- using a reference instead of the doc itself+ docApprobal1 rdoc=+ return True >>=+ log "requesting approbal from bosses" >>=+ sumUp 0 (map(askUser (title doc) rdoc) bosses ) >>=+ checkValidated >>=+ log "requesting approbal from superbosses or timeout" >>=+ sumUp (1*day) (map(askUser (title doc) rdoc) superbosses) >>=+ checkValidated1++ where+ sec= 1+ min= 60* sec+ hour= 60* min+ day= 24*hour+ askUser _ _ user False = return False+ askUser title rdoc user True = do+ step $ push (quser user) rdoc+ logWF ("wait for any response from the user: " ++ user)+ step . pop $ qdocApprobal title+++ log txt x = logWF txt >> return x++ checkValidated :: Bool -> Workflow IO Bool+ checkValidated val =+ case val of+ False -> correctWF (title doc) rdoc >> return False+ !> "not validated. re-sent to the user for correction"+ _ -> return True+++ checkValidated1 :: Bool -> Workflow IO ()+ checkValidated1 val = step $ do+ case val of+ False -> push qrejected doc+ _ -> push qapproved doc++ -- because there may have been a timeout,+ -- the doc references may remain in the queue+ mapM_ (\u ->deleteElem (quser u) rdoc) superbosses+++++{- old code of docAprobal with no sumUp pattern+docApprobal :: Document -> Workflow IO ()+docApprobal doc= do+ logWF "message sent to the boss requesting approbal"+ step $ writeTQueue qboss doc++ -- wait for any response from the boss+ ap <- step $ readTQueue $ qdoc doc+ case ap of+ False -> do logWF "not approved, sent to the user for correction"+ correctWF doc+ True -> do+ logWF " approved, send a message to the superboss requesting approbal"+ step $ writeTQueue qsuperboss doc++ -- wait for any response from the superboss+ -- if no response from the superboss in 5 minutes, it is validated+ flag <- getTimeoutFlag $ 5 * 60+ ap <- step . atomically $ readTQueueSTM (qdoc doc)+ `orElse`+ waitUntilSTM flag >> return True+ case ap of+ False -> do logWF "not approved, sent to the user for correction"+ correctWF doc+ True -> do+ logWF " approved, sent to the list of approved documents"+ step $ writeTQueue qapproved doc++-}++correctWF :: String -> WFRef Document -> Workflow IO ()+correctWF title1 rdoc= do+ -- send a message to the user to correct the document+ step $ push (quser user) rdoc+ -- wait for document edition+ doc' <- step $ pop (qdoc title1)+ if title1 /= title doc'+ -- if doc and new doc edited hace different document title, then start a new workflow for this new document+ -- since a workflow is identified by the workflow name and the key of the starting data, this is a convenient thing.+ then step $ exec "docApprobal" docApprobal doc'+ -- else continue the current workflow by retryng the approbal process+ else docApprobal doc'+++create = do+ separator+ doc <- readDoc+ putStrLn "The document has been sent to the boss.\nPlease wait for the approbal"+ forkIO $ exec "docApprobal" docApprobal doc+ userMenu++userMenu= do+ separator+ putStrLn"\n\n1- Create document\n2- Documents to modify\n3- Approbed documents\n4- manage workflows\nany other- exit"+ n <- getLine+ case n of+ "1" -> create+ "2" -> modify+ "3" -> view+ "4" -> history+ _ -> syncCache >> exitSuccess++ userMenu++++history= do+ separator+ putStr "MANAGE WORKFLOWS\n"+ ks <- getWFKeys "docApprobal"+ mapM (\(n,d) -> putStr (show n) >> putStr "- " >> putStrLn d) $ zip [1..] ks+ putStr $ show $ length ks + 1+ putStrLn "- back"+ putStrLn ""+ putStrLn " select v[space] <number> to view the history or d[space]<number> to delete it"+ l <- getLine+ if length l /= 3 || (head l /= 'v' && head l /= 'd') then history else do+ let n= read $ drop 2 l+ let docproto= Document{title= ks !! (n-1), text=undefined}+ case head l of+ 'v' -> do+ getWFHistory "docApprobal" docproto >>= printHistory . fromJust+ history+ 'd' -> do+ delWF "docApprobal" docproto+ history++ _ -> history++separator= putStrLn "------------------------------------------------"+++modify :: IO ()+modify= do+ separator+ empty <- isEmpty (quser user) :: IO Bool+ if empty then putStrLn "no more documents to modify\nthanks, enter as Boss for the approbal"+ else do+ rdoc <- pick (quser user)+ putStrLn "Please correct this doc"+ Just doc <- atomically $ readWFRef rdoc+ print doc+ doc1 <- readDoc++ -- return $ diff doc1 doc+ atomically $ do+ popSTM (quser user)+ pushSTM (qdoc $ title doc) doc1+ modify++diff (Document t xs) (Document _ ys)=+ Document t $ map (search ys) xs+ where+ search xs x= case find (==x) xs of+ Just x' -> x'+ Nothing -> x+++readDoc :: IO Document+readDoc = do+ putStrLn "please enter the title of the document"+ title1 <- getLine+ h <- getWFHistory "docApprobal" $ Document title1 undefined+ case h of+ Just _ -> putStrLn "sorry document title already existent, try other" >> readDoc+ Nothing -> do+ putStrLn "please enter the text. "+ putStrLn "the edition will end wth a empty line "+ text <- readDoc1 [title1]+ return $ Document title1 text+ where+ readDoc1 text= do+ line <- getLine+ if line == "" then return text else readDoc1 $ text ++ [line]+++++view= do+ separator+ putStrLn "LIST OF APPROVED DOCUMENTS:"+ view1+ where+ view1= do+ empty <- isEmpty qapproved+ if empty then return () else do+ doc <- pop qapproved :: IO Document+ print doc+ view1++++aprobal who= do+ separator+ aprobalList+ putStrLn $ "thanks , press any key to exit, "++ who+ getLine+ syncCache+ return ()+ where+ aprobalList= do+ empty <- isEmpty (quser who)+ if empty+ then do+ putStrLn "No more document to validate. Bye"+ return ()+ else do+ rdoc <- pick (quser who)+ syncCache+ approbal1 rdoc+ aprobalList+ approbal1 :: WFRef Document -> IO ()+ approbal1 rdoc= do+ putStrLn $ "hi " ++ who ++", a new request for aprobal has arrived:"+ Just doc <- atomically $ readWFRef rdoc+ print doc+ putStrLn $ "Would you approbe this document? s/n"+ l <- getLine+ if l/= "s" && l /= "n" then approbal1 rdoc else do+ let b= head l+ let res= if b == 's' then True else False+ -- send the message to the workflow+ atomically $ do+ popSTM (quser who)+ pushSTM (qdocApprobal $ title doc) res+ syncCache++++
+ Demos/sequence.hs view
@@ -0,0 +1,16 @@++module Main where+import Control.Workflow.Binary+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 specify the return type++main= exec1 "count" $ mcount (0 :: Int)+
Workflow.cabal view
@@ -1,44 +1,114 @@-name: Workflow-version: 0.5.6-synopsis: library for transparent execution of interruptible computations-description:- Transparent support for interruptable 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.- .- The computantion can be restarted at the interrupted point because such monad is encapsulated inside- a state monad transformer that transparently checkpoint the computation state. Besides that, the package also provides- other services associated to workflows- The main features are:- .- * logging of each intermediate action results in disk.- .- * resume the monadic computation at the last checkpoint after soft or hard interruption.- .- * suspend a computation until the input object meet certain conditions. useful for inter-workflow comunications.-- .- * Communications with other processes including other workflows trough persistent data objects,- inspection of intermediate workflow results , persistent queues, persistent timeouts so that no data is lost due- to shutdowns- .- * A workflow can initiate anoter workflow and wait for the resutl- .- * workflow management and monitoriing, view workflow history and intermediate results.+name: Workflow+version: 0.5.8 +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 -category: Control, Workflow, Concurrent, Middleware-Stability: experimental-license: BSD3-license-file: LICENSE-author: Alberto Gómez Corona-maintainer: agocorona@gmail.com-Tested-With:-Build-Type: Simple-build-Depends: base >=3 && <4, containers, RefSerialize>=0.2.4, TCache>=0.6.4, stm > 2, old-time, mtl-Cabal-Version: >= 1.2 -exposed-modules: Control.Workflow-ghc-options: -O2 +stability: experimental+homepage:+package-url:+bug-reports: agocorona@gmail.com+synopsis: library for transparent execution of interruptible computations+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.+ .+ The computantion 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,+ .+ * 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 monitoriing, view workflow history and intermediate results. +++category: Control, Workflow, Concurrent+author: Alberto Gómez Corona+data-files:+data-dir: ""++extra-tmp-files:+exposed-modules:+ Control.Workflow.Binary+ Control.Workflow.Binary.Patterns+ Control.Workflow.Text+ Control.Workflow.Text.Patterns++ Data.Persistent.Queue.Text+ Data.Persistent.Queue.Binary++other-modules:+ Control.Workflow.Text.TextDefs+ Control.Workflow.Binary.BinDefs+ Control.Workflow.IDynamic+ Control.Workflow.Stat+ Control.Workflow.GenSerializer++extra-source-files:+ Control/Workflow/Workflow.inc.hs+ Control/Workflow/Patterns.inc.hs+ Data/Persistent/Queue/Queue.inc.hs+ Demos/docAprobal.hs+ Demos/sequence.hs+ Demos/Fact.hs+ Demos/Inspect.hs++exposed: True+buildable: True+build-tools:+cpp-options:+cc-options:+ld-options:+pkgconfig-depends:+frameworks:+c-sources:++extensions:+extra-libraries:+extra-lib-dirs:+includes:+install-includes:+include-dirs:+hs-source-dirs: .++ghc-prof-options:+ghc-shared-options:+ghc-options:+hugs-options:+nhc98-options:+jhc-options: