Safe Haskell | None |
---|
A workflow can be seen as a persistent thread.
The workflow monad writes a log that permit to restore the thread
at the interrupted point. step
is the (partial) monad transformer for
the Workflow monad. A workflow is defined by its name and, optionally
by the key of the single parameter passed. There primitives for starting workflows
also restart the interrupted workflow when it has been in execution previously.
A small example that print the sequence of integers in te console if you interrupt the progam, when restarted again, it will start from the last printed number
module Main where import Control.Workflow import Control.Concurrent(threadDelay) import System.IO (hFlush,stdout) mcount n= dostep
$ do putStr (show n ++ " ") hFlush stdout threadDelay 1000000 mcount (n+1) return () -- to disambiguate the return type main=exec1
"count" $ mcount (0 :: Int)
>>>
runghc demos\sequence.hs
>0 1 2 3 >CTRL-C Pressed>>>
runghc demos\sequence.hs
>3 4 5 6 7 >CTRL-C Pressed>>>
runghc demos\sequence.hs
>7 8 9 10 11 ...
The program restart at the last saved step.
As you can see, some side effect can be re-executed after recovery if
the log is not complete. This may happen after an unexpected shutdown (in this case)
or due to an asynchronous log writing policy. (see syncWrite
)
When the step results are big and complex, use the Data.RefSerialize package to define the (de)serialization instances The log size will be reduced. printWFHistory` method will print the structure changes in each step.
If instead of RefSerialize
, you use read and show instances, there will
be no reduction. but still it will work, and the log will be readable for debugging purposes.
The RefSerialize istance is automatically derived from Read, Show instances.
Data.Binary instances are also fine for serialization. To use Binary, just define a binary instance
of your data by using showpBinary
and readpBinary
.
Within the RefSerialize instance of a structure, you can freely mix Show,Read RefSerialize and Data Binary instances.
Control.Workflow.Patterns contains higuer level workflow patters of multiuser workflows
Control.Workflow.Configuration permits the use of workflows for configuration purposes
- type Workflow m = WF Stat m
- type WorkflowList m a b = Map String (a -> Workflow m b)
- class PMonadTrans t m a where
- class MonadCatchIO m a where
- class MonadIO io => HasFork io where
- throw :: (MonadIO m, Exception e) => e -> m a
- class Indexable a where
- keyWF :: Indexable a => String -> a -> String
- start :: (MonadCatchIO m, MonadIO m, Indexable a, Serialize a, Serialize b, Typeable a, Typeable b) => String -> (a -> Workflow m b) -> a -> m (Either WFErrors b)
- exec :: (Indexable a, Serialize a, Serialize b, Typeable a, Typeable b, Monad m, MonadIO m, MonadCatchIO m) => String -> (a -> Workflow m b) -> a -> m b
- exec1d :: (Serialize b, Typeable b, MonadIO m, MonadCatchIO m) => String -> Workflow m b -> m b
- exec1 :: (Serialize a, Typeable a, Monad m, MonadIO m, MonadCatchIO m) => String -> Workflow m a -> m a
- wfExec :: (Indexable a, Serialize a, Typeable a, MonadCatchIO m, MonadIO m) => Workflow m a -> Workflow m a
- startWF :: (MonadCatchIO m, MonadIO m, Serialize a, Serialize b, Typeable a, Indexable a, Typeable b) => String -> a -> WorkflowList m a b -> m (Either WFErrors b)
- restartWorkflows :: (Serialize a, Serialize b, Typeable a, Indexable b, Typeable b) => WorkflowList IO a b -> IO ()
- data WFErrors
- = NotFound
- | AlreadyRunning
- | Timeout
- | WFException String
- step :: (Monad m, MonadIO m, Serialize a, Typeable a) => m a -> Workflow m a
- unsafeIOtoWF :: Monad m => IO a -> Workflow m a
- data WFRef a
- newWFRef :: (Serialize a, Typeable a, MonadIO m, MonadCatchIO m) => a -> Workflow m (WFRef a)
- stepWFRef :: (Serialize a, Typeable a, MonadIO m) => m a -> Workflow m (WFRef a, a)
- readWFRef :: (Serialize a, Typeable a) => WFRef a -> STM (Maybe a)
- writeWFRef :: (Serialize a, Typeable a) => WFRef a -> a -> STM ()
- moveState :: (MonadIO m, Indexable a, Serialize a, Typeable a) => String -> a -> a -> m ()
- waitWFActive :: String -> STM ()
- getAll :: Monad m => Workflow m [IDynamic]
- safeFromIDyn :: (Typeable a, Serialize a) => IDynamic -> Maybe a
- getWFKeys :: String -> IO [String]
- getWFHistory :: (Indexable a, Serialize a) => String -> a -> IO (Maybe Stat)
- waitFor :: (Indexable a, Serialize a, Serialize b, Typeable a, Indexable b, Typeable b) => (b -> Bool) -> String -> a -> IO b
- waitForSTM :: (Indexable a, Serialize a, Serialize b, Typeable a, Indexable b, Typeable b) => (b -> Bool) -> String -> a -> STM b
- waitUntilSTM :: TVar Bool -> STM ()
- getTimeoutFlag :: MonadIO m => Integer -> Workflow m (TVar Bool)
- withTimeout :: (MonadIO m, Typeable a, Serialize a) => Integer -> STM a -> Workflow m (Maybe a)
- withKillTimeout :: MonadCatchIO m => String -> Int -> Integer -> m a -> m a
- logWF :: MonadIO m => String -> Workflow m ()
- clearRunningFlag :: MonadIO m => [Char] -> m (Map String (String, Maybe ThreadId), Maybe ThreadId)
- killThreadWF :: (Indexable a, Serialize a, Typeable a, MonadIO m) => String -> a -> m ()
- killWF :: (Indexable a, MonadIO m) => String -> a -> m ()
- delWF :: (Indexable a, MonadIO m, Typeable a) => String -> a -> m ()
- killThreadWF1 :: MonadIO m => String -> m ()
- killWF1 :: MonadIO m => String -> m ()
- delWF1 :: MonadIO m => String -> m ()
- delWFHistory :: Indexable a => [Char] -> a -> IO ()
- delWFHistory1 :: [Char] -> IO ()
- syncWrite :: SyncMode -> IO ()
- data SyncMode
- = Synchronous
- | Asyncronous { }
- | SyncManual
- showHistory :: Stat -> ByteString
- isInRecover :: Monad m => Workflow m Bool
Documentation
type WorkflowList m a b = Map String (a -> Workflow m b)Source
class PMonadTrans t m a whereSource
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
(MonadTrans t, Monad m) => PMonadTrans t m a | An instance of MonadTrans is an instance of PMonadTrans |
(Monad m, MonadIO m, Serialize a, Typeable a) => PMonadTrans (WF Stat) m a | plift= step |
class MonadCatchIO m a whereSource
adapted from the MonadCatchIO-mtl
package. However, in tis case is needed to express serializable constraints about the returned values,
so the usual class definitions for lifting IO functions are not suitable.
catch :: Exception e => m a -> (e -> m a) -> m aSource
Generalized version of catch
Generalized version of block
Generalized version of unblock
(Serialize a, Typeable a, MonadIO m, MonadCatchIO m) => MonadCatchIO (WF Stat m) a |
class Indexable a where
Indexable is an utility class used to derive instances of IResource
Example:
data Person= Person{ pname :: String, cars :: [DBRef Car]} deriving (Show, Read, Typeable) data Car= Car{owner :: DBRef Person , cname:: String} deriving (Show, Read, Eq, Typeable)
Since Person and Car are instances of Read
ans Show
, by defining the Indexable
instance
will implicitly define the IResource instance for file persistence:
instance Indexable Person where key Person{pname=n} = "Person " ++ n instance Indexable Car where key Car{cname= n} = "Car " ++ n
:: a | |
-> String | additional extension for default file paths. IMPORTANT: defPath must depend on the datatype, not the value (must be constant). Default is TCacheData/ |
keyWF :: Indexable a => String -> a -> StringSource
Return the unique name of a workflow with a parameter (executed with exec or start)
Start/restart workflows
:: (MonadCatchIO m, MonadIO m, Indexable a, Serialize a, Serialize b, Typeable a, 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 |
exec :: (Indexable a, Serialize a, Serialize b, Typeable a, Typeable b, Monad m, MonadIO m, MonadCatchIO m) => String -> (a -> Workflow m b) -> a -> m bSource
Start or continue a workflow with exception handling
the workflow flags are updated even in case of exception
WFerrors
are raised as exceptions
exec1d :: (Serialize b, Typeable b, MonadIO m, MonadCatchIO m) => String -> Workflow m b -> m bSource
A version of exec1 that deletes its state after complete execution or thread killed
exec1 :: (Serialize a, Typeable a, Monad m, MonadIO m, MonadCatchIO m) => String -> Workflow m a -> m aSource
A version of exec with no seed parameter.
wfExec :: (Indexable a, Serialize a, Typeable a, MonadCatchIO m, MonadIO m) => Workflow m a -> Workflow m aSource
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.
:: (MonadCatchIO m, MonadIO m, Serialize a, Serialize b, Typeable a, Indexable a, Typeable b) | |
=> String | Name of workflow in the workflow list |
-> a | Initial value (ever use the initial value even to restart the workflow) |
-> WorkflowList m a b | function to execute |
-> m (Either WFErrors b) | Result of the computation |
Start or continue a workflow from a list of workflows with exception handling.
see start
for details about exception and error handling
restartWorkflows :: (Serialize a, Serialize b, Typeable a, Indexable b, Typeable b) => WorkflowList IO a b -> IO ()Source
Re-start the non finished workflows in the list, for all the initial values that they may have been invoked
Return conditions from the invocation of start/restart primitives
Lifting to the Workflow monad
step :: (Monad m, MonadIO m, Serialize a, Typeable a) => m a -> Workflow m aSource
Lifts a monadic computation to the WF monad, and provides transparent state loging and resuming the computation Note: Side effect can be repeated at recovery time if the log was not complete before shut down see the integer sequence example, above.
unsafeIOtoWF :: Monad m => IO a -> Workflow m aSource
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
References to intermediate values in the workflow log
newWFRef :: (Serialize a, Typeable a, MonadIO m, MonadCatchIO m) => a -> Workflow m (WFRef a)Source
Log a value in the workflow log and return a reference to it.
newWFRef x= stepWFRef
(return x) >>= return . fst
stepWFRef :: (Serialize a, Typeable a, MonadIO m) => m a -> Workflow m (WFRef a, a)Source
Execute an step but return a reference to the result besides the result itself
readWFRef :: (Serialize a, Typeable a) => WFRef a -> STM (Maybe a)Source
Read the content of a Workflow reference. Note that its result is not in the Workflow monad
State manipulation
writeWFRef :: (Serialize a, Typeable a) => WFRef a -> a -> STM ()Source
Writes a new value en in the workflow reference, that is, in the workflow log. Why would you use this?. Don't do that!. modifiying the content of the workflow log would change the excution flow when the workflow restarts. This metod is used internally in the package. The best way to communicate with a workflow is trough a persistent queue, using Data.Persistent.Collection:
worflow= exec1 wf do r <-stepWFRef
exprpush
"queue" r back <-pop
"queueback" ...
moveState :: (MonadIO m, Indexable a, Serialize a, Typeable a) => String -> a -> a -> m ()Source
Moves the state of workflow with a seed value to become the state of other seed value This may be of interest when the entry value changes its key value but should not initiate a new workflow but continues with the current one
Workflow inspect
waitWFActive :: String -> STM ()Source
getAll :: Monad m => Workflow m [IDynamic]Source
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]
safeFromIDyn :: (Typeable a, Serialize a) => IDynamic -> Maybe aSource
getWFKeys :: String -> IO [String]Source
Return the list of object keys that are running for a workflow
getWFHistory :: (Indexable a, Serialize a) => String -> a -> IO (Maybe Stat)Source
Return the current state of the computation, in the IO monad
:: (Indexable a, Serialize a, Serialize b, Typeable a, Indexable b, Typeable b) | |
=> (b -> Bool) | The condition that the retrieved object must meet |
-> String | The workflow name |
-> a | the INITIAL value used in the workflow to start it |
-> IO b | The first event that meet the condition |
Observe the workflow log untiil a condition is met.
Persistent timeouts
waitUntilSTM :: TVar Bool -> STM ()Source
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 . JustorElse
waitUntilSTM
flag >> return Nothing case ap of Nothing -> dologWF
timeout ... Just x -> dologWF
$ received ++ show x ...
getTimeoutFlag :: MonadIO m => Integer -> Workflow m (TVar Bool)Source
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.
withTimeout :: (MonadIO m, Typeable a, Serialize a) => Integer -> STM a -> Workflow m (Maybe a)Source
Return either the result of the STM conputation or Nothing in case of timeout This timeout is persistent. This means that the time start to count from the first call to getTimeoutFlag on no matter if the workflow is restarted. The time that the worlkflow has been stopped count also. Thus, the wait time can exceed the time between failures. when timeout is 0 means no timeout.
withKillTimeout :: MonadCatchIO m => String -> Int -> Integer -> m a -> m aSource
Executes a computation understanding that it is inside the
workflow identified by id
. If f
finish after time
it genetates a Timeout
exception which must result in the end of the workflow.
If the workflow is restarted after time2
has elapsed, the workflow
will restart from the beginning. If not, it will restart at the last checkpoint.
Usually time2> time
time2=0
means time2
is infinite
Trace logging
logWF :: MonadIO m => String -> Workflow m ()Source
Log a message in the workflow history. I can be printed out with printWFhistory
The message is printed in the standard output too
Termination of workflows
clearRunningFlag :: MonadIO m => [Char] -> m (Map String (String, Maybe ThreadId), Maybe ThreadId)Source
killThreadWF :: (Indexable a, Serialize a, Typeable a, MonadIO m) => String -> a -> m ()Source
Kill the executing thread if not killed, but not its state.
exec
start
or restartWorkflows
will continue the workflow
killWF :: (Indexable a, MonadIO m) => String -> a -> m ()Source
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
delWF :: (Indexable a, MonadIO m, Typeable a) => String -> a -> m ()Source
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.
killThreadWF1 :: MonadIO m => String -> m ()Source
A version of KillThreadWF
for workflows started wit no parameter by exec1
killWF1 :: MonadIO m => String -> m ()Source
A version of KillWF
for workflows started wit no parameter by exec1
delWFHistory :: Indexable a => [Char] -> a -> IO ()Source
Delete the history of a workflow. Be sure that this WF has finished.
delWFHistory1 :: [Char] -> IO ()Source
Log writing policy
syncWrite :: SyncMode -> IO ()
Specify the cache synchronization policy with permanent storage. See SyncMode
for details
data SyncMode
Synchronous | sync state to permanent storage when |
Asyncronous | |
SyncManual | use |
Print log history
showHistory :: Stat -> ByteStringSource
show the state changes along the workflow, that is, all the intermediate results
isInRecover :: Monad m => Workflow m BoolSource