diff --git a/Control/Workflow.hs b/Control/Workflow.hs --- a/Control/Workflow.hs +++ b/Control/Workflow.hs @@ -80,7 +80,8 @@ module Control.Workflow ( - Workflow -- a useful type name + Stat +, Workflow -- a useful type name , WorkflowList , PMonadTrans (..) , MonadCatchIO (..) @@ -100,6 +101,8 @@ , WFErrors(..) -- * Lifting to the Workflow monad , step +, getWFStat +, stepExec --, while --, label --, stepControl @@ -110,7 +113,6 @@ , newWFRef , stepWFRef , readWFRef -, getNRefs , getWFRef -- * State manipulation , writeWFRef @@ -254,8 +256,10 @@ plift= Control.Monad.Trans.lift --- | Handle with care: this instance will force --- the unwanted execution at recovery of every liftted IO procedure +-- the execution at recovery of every liftted IO procedure -- better use 'step . liftIO' instead of 'liftIO' + + instance MonadIO m => MonadIO (WF Stat m) where liftIO= unsafeIOtoWF @@ -403,13 +407,13 @@ exec1nc :: ( Monad m, MonadIO m, CMC.MonadCatchIO m) => String -> Workflow m a -> m a exec1nc str f = - (do + do v <- getState str f () case v of Right (name, f, stat) -> do r <- runWF1 name f stat False return r - Left err -> CMC.throw err) + Left err -> CMC.throw err `CMC.catch` (\(e :: CE.SomeException) -> liftIO $ do let name= keyWF str () @@ -451,8 +455,8 @@ => m a -> Workflow m a -step mx= WF(\s -> do - let stat= state s +step mx= WF(\s -> do + let recovers= recover s versionss= versions s -- !> "vvvvvvvvvvvvvvvvvvv" @@ -466,26 +470,34 @@ else do let ref= self s when (recovers && L.null versionss) $ do - liftIO $ atomically $ do s' <- readDBRef ref `justifyM` error ("step: not found: "++ wfName s) writeDBRef ref s'{recover= False,references= references s} - stepExec ref mx) - -stepExec sref mx= do - x' <- mx + stepExec1 ref mx) - liftIO . atomicallySync $ do - s <- readDBRef sref >>= return . fromMaybe (error $ "step: readDBRef: not found:" ++ keyObjDBRef sref) +getWFStat :: Monad m => Workflow m (DBRef Stat) +getWFStat= WF $ \s -> return (s,self s) - let versionss= versions s - dynx= toIDyn x' - ver= dynx: versionss - s'= s{ recover= False, versions = ver, state= state s+1} +stepExec + :: (Typeable t, Serialize t, MonadIO m) => + DBRef Stat -> m t -> m (DBRef Stat, t) +stepExec ref mx= do + (s,x) <- stepExec1 ref mx + return (self s, x) - writeDBRef sref s' +stepExec1 sref mx= do + x' <- mx + liftIO . atomicallySync $ do + s <- readDBRef sref >>= return . fromMaybe (error $ "step: readDBRef: not found:" ++ keyObjDBRef sref) + let versionss= versions s + dynx= toIDyn x' + ver= dynx: versionss + s'= s{ recover= False, versions = ver, state= state s+1} + writeDBRef sref s' + return (s', x') - return (s', x') +--undoStep :: Monad m => Workflow m () +--undoStep= WF $ \s@Stat{..} -> return(s{state=state-1, versions= L.tail versions},()) isInRecover :: Monad m => Workflow m Bool isInRecover = WF(\s@Stat{..} -> @@ -512,7 +524,7 @@ in case recover s && not(L.null $ versions s) of True -> f >>= \x -> return (s{versions= L.tail $ versions s},x) - False -> stepExec (self s) f) + False -> stepExec1 (self s) f) -- Executes a computation 'f' in a loop while the return value meets the condition 'cond' is met. -- At recovery time, the current state of the loop is restored. @@ -715,8 +727,7 @@ :: ( CMC.MonadCatchIO m, MonadIO m , Serialize a, Serialize b , Typeable a - , Indexable a - , Typeable b) + , Indexable a) => String -- ^ Name of workflow in the workflow list -> a -- ^ Initial value (ever use the initial value even to restart the workflow) -> WorkflowList m a b -- ^ function to execute @@ -730,9 +741,8 @@ -- | Re-start the non finished workflows in the list, for all the initial values that they may have been invoked restartWorkflows - :: (Serialize a, Serialize b, Typeable a - , Indexable b, Typeable b) - => WorkflowList IO a b -- the list of workflows that implement the module + :: (Serialize a, Typeable a) + => 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)) @@ -925,29 +935,29 @@ let (n,flag)= if recover then (state - (L.length versions) -1 ,False) else (state -1 ,True) - let ref = WFRef n self - let s'= s{references= (n,(toIDyn r,flag)):references } + ref = WFRef n self + s'= s{references= (n,(toIDyn r,flag)):references } liftIO $ atomically $ writeDBRef self s' r `seq` return (s',(ref,r)) ) -getNRefs wfname= do - st <- getResource stat0{wfName= wfname} `onNothing` error ("Workflow not found: "++ wfname) - return $ L.length $ references st +getWFRef :: Indexable a => Int -> String -> a -> WFRef b +getWFRef n s v= WFRef n $ getDBRef $ keyResource stat0{wfName= keyWF s v} -getWFRef :: Int -> String -> WFRef a -getWFRef n s= WFRef n $ getDBRef s ---getReference ::(Monad m --- , Serialize a --- , Typeable a) --- => Int -> Workflow m (Maybe a) ---getReference n = WF $ \st -> --- case L.lookup n $! references st of --- Just (r,_) -> return(st, Just $ fromIDyn r) --- Nothing -> do --- let n1= state st - n --- return (st, Just . fromIDyn $ versions st !! n1) +--getNRefs wfname= do +-- st <- getResource stat0{wfName= wfname} `onNothing` error ("Workflow not found: "++ wfname) +-- return $ L.length $ references st +-- |return a reference to the last step result +--getWFRef ::(MonadIO m,Serialize a, Typeable a) => Monad m => a -> Workflow m (WFRef a) +--getWFRef r = WF(\s@Stat{..} -> do +-- let (n,flag)= if recover +-- then (state - (L.length versions) -1 ,False) +-- else (state -1 ,True) +-- ref = WFRef n self +-- s'= s{references= (n,(toIDyn r,flag)):references } +-- liftIO $ atomically $ writeDBRef self s' +-- r `seq` return (s',ref) ) -- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad @@ -963,7 +973,7 @@ case L.lookup n $! references st of Just (r,_) -> return . Just $ fromIDyn r Nothing -> do - let n1= state st - n + let n1= if recover st then n else state st - n return . Just . fromIDyn $ versions st !! n1 -- flushDBRef ref !> "readWFRef" @@ -1035,7 +1045,7 @@ =>String -> a -> a -> m () moveState wf t t'= liftIO $ do atomicallySync $ do - withSTMResources[stat0{wfName= n}] $ change n + mrun <- readDBRef tvRunningWfs case mrun of Nothing -> return() @@ -1044,18 +1054,19 @@ let th= case mr of Nothing -> Nothing; Just(_,mt)-> mt let map'= M.insert n' (wf,th) $ M.delete n map writeDBRef tvRunningWfs $ Running map' + withSTMResources[stat0{wfName= n}] $ change n where n = keyWF wf t n'= keyWF wf t' - change m [Nothing] = error $ "changeState: Workflow not found: "++ show n + change n [Nothing] = error $ "moveState: Workflow not found: "++ show n change n [Just s] = resources{toAdd= [ s{wfName=n',versions = toIDyn t': L.tail( versions s) }] ,toDelete=[s]} - doit n [Nothing]= error $ "moveState: state not found for: " ++ n + -- | Log a message in the workflow history. I can be printed out with 'printWFhistory' -- The message is printed in the standard output too logWF :: MonadIO m => String -> Workflow m () @@ -1127,11 +1138,11 @@ Just mx -> do let Stat{ versions= d:_}= mx case safeFromIDyn d of - Nothing -> retry -- `debug` "waithFor retry Nothing" - Just x -> + Left _ -> retry -- `debug` "waithFor retry Nothing" + Right x -> case filter x of False -> retry -- `debug` "waitFor false filter retry" - True -> return x -- `debug` "waitfor return" + True -> return x -- `debug` "waitfor return" diff --git a/Control/Workflow/Stat.hs b/Control/Workflow/Stat.hs --- a/Control/Workflow/Stat.hs +++ b/Control/Workflow/Stat.hs @@ -129,19 +129,19 @@ keyWF :: Indexable a => String -> a -> String keyWF wn x= wn ++ "/" ++ key x - -data WFRef a= WFRef !Int !(DBRef Stat) deriving (Typeable, Show) +data WFRef a= WFRef !Int !(DBRef Stat) deriving (Typeable, Show, Read) instance Indexable (WFRef a) where - key (WFRef n ref)= keyObjDBRef ref++('#':show n) + key (WFRef n ref)= keyObjDBRef ref++('-':show n) --instance Serialize a => Serializable a where -- serialize = runW . showp -- deserialize = runR readp -pathWFlows= (defPath (1:: Int)) ++ "Workflow/" +pathWFlows= (defPath (1:: Int)) ++ "workflow/" stFName st = pathWFlows ++ keyResource st + Persist fr fw fd = defaultPersist --nheader= "/header" diff --git a/Workflow.cabal b/Workflow.cabal --- a/Workflow.cabal +++ b/Workflow.cabal @@ -1,5 +1,5 @@ name: Workflow -version: 0.8.0.1 +version: 0.8.0.2 cabal-version: >= 1.6 build-type: Simple license: BSD3 @@ -7,17 +7,17 @@ maintainer: agocorona@gmail.com stability: experimental bug-reports: agocorona@gmail.com -synopsis: Monad transformer for thread state persistence and workflow patterns +synopsis: Workflow patterns and a monad for thread state logging & recovery description: Transparent support for interruptible computations. A workflow can be seen as a persistent thread that executes a 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. It also can be used if you like to restart your - program at the point where the user left it last time. - . + program at the point where the user left it last time + . . The computation can be restarted at the interrupted point thanks to its logged state in permanent storage. - Besides that, the package also provides other services associated to workflows - . - This release inprove the logging/recovery process in workflows with with many steps + The thread state is located in files by default. It can be moved and continued in another computer. + Besides that, the package also provides other higher level services associated to workflows: Workflow patters and + and a general configuarion utility. . See "Control.Workflow" for details