diff --git a/Control/Workflow.hs b/Control/Workflow.hs --- a/Control/Workflow.hs +++ b/Control/Workflow.hs @@ -15,6 +15,9 @@ -} module Control.Workflow ( Workflow -- a useful type name + ,WorkflowStep + ,WorkflowList + ,Stat ,step -- :: (Monad m) => (a -> m a) -> ( a -> Workflow m a) -- encapsulates a monadic computation into state monad that brings persistence and @@ -44,8 +47,9 @@ , waitFor -- ::(IResource a, Serialize a) => (a ->Bool) -> a -> IO a -- wait until a object (with a certaing key=keyResource x) meet a certain condition -- (useful for checking external actions, possibly by other workflows or by direct use of TCache primitives ) - - ,syncWrite -- syncWrite:: Monad m => + , waitUntil -- :: Integer -> IO() + -- wait until the absolute time in seconds is reached (as returned by getClockTime) + , syncWrite -- syncWrite:: Monad m => -- Bool -> True means that changes are inmediately saved after each step -- Int -> number of seconds between saves when async -- Int -> max size of the cache when async @@ -66,13 +70,15 @@ import System.IO.Unsafe import Control.Monad(when,liftM) import Unsafe.Coerce -import Control.Concurrent (forkIO) +import Control.Concurrent (forkIO,threadDelay) import Control.Concurrent.STM(atomically, retry, readTVar) import Debug.Trace -import Data.TCache +import Data.TCache.Dynamic import Data.RefSerialize import Data.List((\\),find,elemIndices) +import Data.Typeable +import System.Time debug a b = trace b a @@ -84,15 +90,14 @@ type WorkflowList m a = [(String, WorkflowStep m a)] -data Stat a= Workflows [(String,String)] - | Stat{ wfName :: String, state:: Int, index :: Int, recover:: Bool, sync :: Bool , versions :: [a]} - | I a - +data Stat a= Workflows [(String,String)] + |Stat{ wfName :: String, state:: Int, index :: Int, recover:: Bool, sync :: Bool , versions :: [a]} + + deriving (Typeable) stat0 = Stat{ wfName="", state=0, index=0, recover=False, versions =[], sync= True} - -- serialization of data is done trough RefSerialize because it permits to store -- different versions of the same object with minumum memory. @@ -101,14 +106,12 @@ str <- showp list return $ "StatWorkflows "++ str - showp (I x) = do - xs <- rshowp x - return $ "I " ++ xs + 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, rData, rWorkflows] where + readp = choice [rStat, rWorkflows] where rStat= do symbol "Stat" wfName <- stringLiteral @@ -119,11 +122,8 @@ versions<- rreadp return $ Stat wfName (fromIntegral state) (fromIntegral index) recover sync versions - rData= do - symbol "I" - a <- rreadp - return $ I a + rWorkflows= do symbol "StatWorkflows" list <- readp @@ -131,29 +131,26 @@ --persistence trough TCache , default persistence in files -instance (IResource a, Serialize a)=> IResource (Stat a) where +instance (IResource a, Serialize a,Typeable a)=> IResource (Stat a) where keyResource Stat{wfName=name, versions = []}= prefix ++name - keyResource Stat{wfName=name, versions = (a:_)}= prefix ++name++"."++keyResource a - keyResource (I a)= keyResource a - keyResource (Workflows _)= "StatWorkflows" + keyResource Stat{wfName=name, versions = (a:_)}= prefix ++name++"#"++keyResource a - defPath (I a)= defPath a - defPath _= "Workflows/" -- directory for Workflow data - --serialize (I x)= "I "++ serialize x + keyResource w@(Workflows xs)= "StatWorkflows" + + + defPath x= "Workflows/" ++ show (typeOf x)++"/" -- directory for Workflow data + serialize x= runW $ showp x deserialize str = runR readp str -prefix= "Stat." +prefix= "Stat#" lengthPrefix= length prefix -insertResources xs= withResources [] (\_-> xs) +insertDResources xs= withDResources [] (\_-> xs) --unsafeIOtoWF :: Monad m => IO a -> WF m (Stat b) a unsafeIOtoWF x= let y= unsafePerformIO x in y `seq` return y - - - instance Monad m => Monad (WF m s) where return x = WF (\s -> return (s, x)) WF g >>= f = WF (\s ->do @@ -164,7 +161,7 @@ return (s3, x')) -class (IResource a, Serialize a) => Workflow_ a where +class (IResource a, Serialize a,Typeable a) => Workflow_ a where -- | step lift a monadic computation (a -> m a) in in to the WF monad, provides state loging and automatic resume step :: (Monad m) => (a -> m a) -> ( a -> Workflow m a) step f = \x -> WF(\s -> do @@ -189,24 +186,28 @@ Nothing -> error $ "workflow stat not found: " ++key Just old -> old - in [Insert $ Workflows newlist,Delete s, Insert s', Insert (I x')] + in [Insert $ toIDyn $ (Workflows newlist :: Stat a) + ,Delete $ toIDyn s, Insert $ toIDyn s' + ,Insert (toIDyn x')] -- `debug`("insert "++keyResource s'++","++keyResource x'++" delete: "++ -- keyResource s++","++keyResource x) - | otherwise= [Insert s', Insert (I x')] + | otherwise= [Insert $ toIDyn s', Insert $ toIDyn x'] -- `debug`("insert "++keyResource s'++","++keyResource x') let doit [Nothing] = doit1 [] - doit [Just (Workflows xs)] = doit1 xs + doit [Just d] = let Workflows xs= (fromIDyn d :: Stat a) + in doit1 xs - withResourcesID [Workflows undefined] doit + withDResourcesID [toIDyn $ (Workflows undefined :: Stat a)] doit - when (sync s) $ unsafeIOtoWF $ syncCache (refcache :: Cache (Stat a)) + when (sync s) $ unsafeIOtoWF $ syncCache return (s', x') ) -- | start or continue a workflow. WorkflowList is a assoclist of (name, workflow computation) startWF :: (Monad m) => String -> a -> WorkflowList m a ->m a startWF name v wfs= do + unsafeIOtoWF (registerType :: IO (Control.Workflow.Stat a)) case lookup name wfs of Nothing -> error $ "MonadWF.startWF: workflow not found: "++name; Just f -> do @@ -214,7 +215,7 @@ let stat1= stat0{index=0,wfName= name,versions=[v]} ::Stat a let key= keyResource stat1 (vn, stat, found) <- do - wxs <- unsafeIOtoWF $ readResource $ (Workflows undefined :: Stat a) + wxs <- unsafeIOtoWF $ getResource $ (Workflows undefined :: Stat a) case wxs of Nothing -> return (v,stat1, False) Just (Workflows xs) -> @@ -241,6 +242,7 @@ restartWorkflows :: (IResource a, Serialize a) => WorkflowList IO a -> IO () restartWorkflows map = do + unsafeIOtoWF (registerType :: IO (Control.Workflow.Stat a)) mw <- getResource ((Workflows undefined ) :: Stat a) case mw of Nothing -> return () @@ -248,17 +250,21 @@ where start :: (String, String) -> IO () start (key,_)= do - let name= let [init,end]= elemIndices '.' key + let name= let [init,end]= elemIndices '#' key - start= drop (init + 1) key - in take (end-init -1) start + rest= drop (init + 1) key + in take (end-init -1) rest case lookup name map of - Just f -> do - Just st <- getResource stat0{ wfName=key} - + Just f -> do + let st0= Key $ (defPath (stat0 :: Stat a))++key + mst <- getDResource $ IDynamic st0 + case mst of + Nothing -> error $ "getResource: not found "++ keyResource st0 + Just (idyn) -> do + let st = fromIDyn idyn :: Stat a forkIO $ runWF key f (head $ versions st) st >> return () return () - Nothing -> error $ "workflow not found: "++ key + Nothing -> error $ "workflow not found: "++ name @@ -270,16 +276,17 @@ let key= keyResource s' let delWF [Nothing] = error $ " Workflow list not found: " - delWF [Just (Workflows xs)]= + delWF [Just d]= let Workflows xs= (fromIDyn d :: Stat a) in case lookup key xs of Nothing -> error $"runWF not found state for key: "++ key Just oldkey -> - (map (Delete . I) $ versions s') ++ -- delete all intermediate objects generated - [Insert $ Workflows (xs \\ [(key,oldkey)]),Delete s'] + (map (Delete . toIDyn) $ versions s') ++ -- delete all intermediate objects generated + [Insert . toIDyn $ (Workflows (xs \\ [(key,oldkey)]) ::Stat a) + ,Delete $ toIDyn s'] - unsafeIOtoWF $ withResourcesID [Workflows undefined] delWF - when (sync s) $ unsafeIOtoWF $ syncCache (refcache :: Cache (Stat a)) + unsafeIOtoWF $ withDResourcesID [toIDyn $ (Workflows undefined :: Stat a)] delWF + when (sync s) $ unsafeIOtoWF $ syncCache return v' -- switch on and off syncronous write for each step (default is syncronous) @@ -288,12 +295,12 @@ syncWrite:: (Monad m , IResource a)=> Bool -> Int -> Int -> WF m (Stat a) () syncWrite bool time maxsize= WF(\s ->do when (bool== False) $ do - unsafeIOtoWF $ clearSyncCacheProc (refcache :: Cache (Stat a)) time defaultCheck maxsize + unsafeIOtoWF $ clearSyncCacheProc time defaultCheck maxsize return () return (s{ sync= bool},())) -instance (IResource a,Serialize a) => Workflow_ a +instance (IResource a,Serialize a,Typeable a) => Workflow_ a @@ -324,10 +331,10 @@ --------- event handling-------------- reference x=do - mv <- getTVars [x ] + mv <- getDTVars [x ] case mv of [Nothing] -> do - insertResources [x] + insertDResources [x] reference x [Just cl] -> return cl @@ -338,12 +345,12 @@ --waitFor :: (IResource a, RefSerialize a) => Filter a -> a -> WF IO (Stat a) a --waitFor filter x= (step $ waitFor1 filter) x where --NOTE if you Delete the object from te cache, waitFor will no longuer work -waitFor :: (IResource a, Serialize a) => Filter a -> a -> IO a +waitFor :: (IResource a, Serialize a, Typeable a) => Filter a -> a -> IO a waitFor filter x= do - - tv <- reference (I x) + tv <- reference (toIDyn $ x) atomically $ do - I x <- readTVar tv + dyn <- readTVar tv + let x= fromIDyn dyn case filter x of False -> retry @@ -351,3 +358,26 @@ + +waitUntil:: Integer -> IO() +waitUntil t= do + threadDelayInteger $ delay* 1000000 + + print ("execution at tnow="++show tnow++" ,t="++show t) + + where + tnow= n where TOD n _ = unsafePerformIO getClockTime + + delay | t-tnow >0 = t-tnow + | otherwise = 0 + + threadDelayInteger:: Integer -> IO() + threadDelayInteger time| time < imaxInt = threadDelay $ fromIntegral time + | otherwise=do threadDelay maxInt + threadDelayInteger ( time - imaxInt) + where maxInt = maxBound :: Int + imaxInt= fromIntegral maxInt + + + + diff --git a/Control/Workflow.new.hs b/Control/Workflow.new.hs new file mode 100644 --- /dev/null +++ b/Control/Workflow.new.hs @@ -0,0 +1,381 @@ +{-# OPTIONS -fglasgow-exts -fallow-overlapping-instances -fallow-undecidable-instances -O2 #-} +{-transparent low level support (state logging, resume of the computation state, wait for data condition) for very long time + computations. Workflow give the two first services to any monadic computation of type (a-> m a) + + f x >>=\x'-> g x' >>= \x''->... z by + + prefixing the user with the method step: + + step f x >>= \x'-> step g x' >>= \x''->... + +in this way, a workflow can be described with the familiar "do" notation. In principle, there is no other limitation +on the syntax but the restriction (a -> m a): All computations consume and produce the same type of data. + +Alberto Gomez Corona agocorona@gmail.com 2008 +-} +module Control.Workflow ( + Workflow -- a useful type name + ,WorkflowStep + ,WorkflowList + ,Stat + + ,step -- :: (Monad m) => (a -> m a) -> ( a -> Workflow m a) + -- encapsulates a monadic computation into state monad that brings persistence and + -- recovery services + + ,startWF -- :: (Monad m) => + -- String -> mame of workflow in the workflowlist + -- a -> initial data value + -- WorkflowList m a -> assoc-list of (workflow name string,Workflow methods) + -- m a resulting value + -- start or continue a workflow + + + , restartWorkflows -- :: (IResource a, Serialize a) => WorkflowList IO a -> IO () + -- re-start the non finished workflows. needs the assoclist. + + + , getStep -- Monad m => Int -> Workflow m a return the n-tn intermediate result + -- if Int < 0 count from the current result back + + , getAll -- :: Monad m => Workfow m [a] return all the intermediate results + + , unsafeIOtoWF -- executes a IO operation. this is executed whenever re-started, no matter where is the resume point + -- This is useful for external IO re-initializations not controllable by the State monad. + + + , waitFor -- ::(IResource a, Serialize a) => (a ->Bool) -> a -> IO a + -- wait until a object (with a certaing key=keyResource x) meet a certain condition + -- (useful for checking external actions, possibly by other workflows or by direct use of TCache primitives ) + , waitUntil -- :: Integer -> IO() + -- wait until the absolute time in seconds is reached (as returned by getClockTime) + , syncWrite -- syncWrite:: Monad m => + -- Bool -> True means that changes are inmediately saved after each step + -- Int -> number of seconds between saves when async + -- Int -> max size of the cache when async + -- WF m (Stat a) () in the workflow monad + + -- Turn on and off syncronized writing to disk + -- select async mode only + -- -for very fast workflow steps or + -- -when the cache policies are dictated outside of the workflow + -- trough SyncCacheProc (see TCache module) + +) + + +where + + +import System.IO.Unsafe +import Control.Monad(when,liftM) +import Unsafe.Coerce +import Control.Concurrent (forkIO,threadDelay) +import Control.Concurrent.STM(atomically, retry, readTVar) +import Debug.Trace + +import Data.TCache.Dynamic +import Data.RefSerialize +import Data.List((\\),find,elemIndices) +import Data.Typeable +import System.Time + +debug a b = trace b a + +data WF m s l = WF { st :: s -> m (s,l) } + +type Workflow m l= WF m (Stat l) l -- not so scary + +type WorkflowStep m a= ( a -> Workflow m a) + +type WorkflowList m a = [(String, WorkflowStep m a)] + +data Stat a= Workflows [(String,String)] + |Stat{ wfName :: String, state:: Int, index :: Int, recover:: Bool, sync :: Bool , versions :: [a]} + + deriving (Typeable) + +stat0 = Stat{ wfName="", state=0, index=0, recover=False, versions =[], sync= True} + + +-- serialization of data is done trough RefSerialize because it permits to store +-- different versions of the same object with minumum memory. + +instance Serialize a => Serialize (Stat a) where + showp (Workflows 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 <- stringLiteral + state <- integer + index <- integer + recover <- bool + sync <- bool + versions<- rreadp + return $ Stat wfName (fromIntegral state) (fromIntegral index) recover sync versions + + + + rWorkflows= do + symbol "StatWorkflows" + list <- readp + return $ Workflows list + +--persistence trough TCache , default persistence in files + +instance (IResource a, Serialize a,Typeable a)=> IResource (Stat a) where + keyResource Stat{wfName=name, versions = []}= show (name,"") + keyResource Stat{wfName=name, versions = (a:_)}= show (name,keyResource a) + + keyResource w@(Workflows xs)= "StatWorkflows" + + + defPath x= "Workflows/" ++ show (typeOf x)++"/" -- directory for Workflow data + + serialize x= runW $ showp x + deserialize str = runR readp str + + + +insertDResources xs= withDResources [] (\_-> xs) + +--unsafeIOtoWF :: Monad m => IO a -> WF m (Stat b) a +unsafeIOtoWF x= let y= unsafePerformIO x in y `seq` return y + + + + +instance Monad m => Monad (WF m s) 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 lift a monadic computation (a -> m a) in in to the WF monad, provides state loging and automatic resume + step :: (Monad m) => (a -> m a) -> ( a -> Workflow m a) + step f = \x -> WF(\s -> do + let stat= state s + let ind= index s + if recover s && ind < stat + then return (s{index=ind +1 }, versions s !! (stat - ind-1) ) + else do + x'<- f x + let s'= s{recover= False, versions = x': versions s, state= state s+1} + unsafeIOtoWF $ do + + let + doit1 xs + | keyResource s /= keyResource s' -- `debug` ("keys:"++keyResource s ++", "++keyResource s') + = + let newlist= newpair :(xs \\[oldpair]) + newpair= (keyResource s',original) + oldpair= (key,original) + key= keyResource s + original= case lookup key xs of + Nothing -> error $ "workflow stat not found: " ++key + Just old -> old + + in [Insert $ toIDyn $ (Workflows newlist :: Stat a) + ,Delete $ toIDyn s, Insert $ toIDyn s' + ,Insert (toIDyn x')] + -- `debug`("insert "++keyResource s'++","++keyResource x'++" delete: "++ + -- keyResource s++","++keyResource x) + | otherwise= [Insert $ toIDyn s', Insert $ toIDyn x'] + -- `debug`("insert "++keyResource s'++","++keyResource x') + let + doit [Nothing] = doit1 [] + doit [Just d] = let Workflows xs= (fromIDyn d :: Stat a) + in doit1 xs + + withDResourcesID [toIDyn $ (Workflows undefined :: Stat a)] doit + + when (sync s) $ unsafeIOtoWF $ syncCache + + return (s', x') ) + + -- | start or continue a workflow. WorkflowList is a assoclist of (name, workflow computation) + startWF :: (Monad m) => String -> a -> WorkflowList m a ->m a + startWF name v wfs= do + unsafeIOtoWF (registerType :: IO (Stat a)) + case lookup name wfs of + Nothing -> error $ "MonadWF.startWF: workflow not found: "++name; + Just f -> do + + let stat1= stat0{index=0,wfName= name,versions=[v]} ::Stat a + let key= keyResource stat1 + (vn, stat, found) <- do + wxs <- unsafeIOtoWF $ getResource $ (Workflows undefined :: Stat a) + case wxs of + Nothing -> return (v,stat1, False) + Just (Workflows xs) -> + case find (\(_,s)-> s == key) xs of + + Just (key1, oldkey1) -> do + -- already in course + let (name,_)= read key1 ::(String, String) + mst <- unsafeIOtoWF $ getResource stat0{wfName= name} + case mst of + Nothing -> error $ "no stat for key. " ++ key + Just s@Stat{versions=(a:_)} -> return (a,s{index=0,recover=True}, True) -- the last value + + Nothing -> return (v, stat1, False) + -- insert it in the running workflow list + when (found == False) $ + let addWF [Nothing] = [Workflows [(key,key)],stat] + addWF [Just (Workflows xs)]= [Workflows ((key,key):xs),stat] + + in unsafeIOtoWF $ withResources [(Workflows undefined :: Stat a)] addWF + + runWF name f vn stat -- `debug` (serialize stat) + + + + + restartWorkflows :: (IResource a, Serialize a) => WorkflowList IO a -> IO () + restartWorkflows map = do + unsafeIOtoWF (registerType :: IO (Stat a)) + mw <- getResource ((Workflows undefined ) :: Stat a) + case mw of + Nothing -> return () + Just (Workflows all) -> mapM_ start all + where + start :: (String, String) -> IO () + start (key,_)= do + let (name ,_)= read key :: (String, String) + + case lookup name map of + Just f -> do + Just st <- getResource stat0{ wfName=key} + + forkIO $ runWF key f (head $ versions st) st >> return () + return () + Nothing -> error $ "workflow not found: "++ name + + + + runWF :: (Monad m,IResource a, Serialize a) => String ->( a -> Workflow m a) -> a -> (Stat a) -> m a + runWF name f v s=do + + (s', v') <- st (f v) $ s + + let key= keyResource s' + + let delWF [Nothing] = error $ " Workflow list not found: " + delWF [Just d]= let Workflows xs= (fromIDyn d :: Stat a) in + case lookup key xs of + Nothing -> error $"runWF not found state for key: "++ key + Just oldkey -> + (map (Delete . toIDyn) $ versions s') ++ -- delete all intermediate objects generated + [Insert . toIDyn $ (Workflows (xs \\ [(key,oldkey)]) ::Stat a) + ,Delete $ toIDyn s'] + + + unsafeIOtoWF $ withDResourcesID [toIDyn $ (Workflows undefined :: Stat a)] delWF + when (sync s) $ unsafeIOtoWF $ syncCache + return v' + + -- switch on and off syncronous write for each step (default is syncronous) + -- for very fast steps, asyncronous is better.h + -- when TCache is used for other purposes, is better to define the cache policy direct + syncWrite:: (Monad m , IResource a)=> Bool -> Int -> Int -> WF m (Stat a) () + syncWrite bool time maxsize= WF(\s ->do + when (bool== False) $ do + unsafeIOtoWF $ clearSyncCacheProc time defaultCheck maxsize + return () + return (s{ sync= bool},())) + + +instance (IResource a,Serialize a,Typeable a) => Workflow_ a + + + +-- return the result of the previous step +getStep :: Monad m => Int -> Workflow m a +getStep i= WF(\s ->do + let stat= state s + return (s, if i > 0 && i <= stat then versions s !! (stat - i) + else if i < 0 && i >= -stat then versions s !! (stat +i) + else error "getStep: wrong index") + ) + +-- get all the step results +getAll :: Monad m => WF m (Stat a) [a] +getAll = WF(\s ->return (s, take (state s) $ versions s)) + +{- + +exec :: (Monad m) => WF m s a -> s -> m a +exec (WF f) s = do (s', x) <- f s + return x + +run :: (Monad m,Monad (WF m s)) => (a -> WF m s a) -> s -> a -> m a +run f s a= exec (f a) s + +-} + + +--------- event handling-------------- +reference x=do + mv <- getDTVars [x ] + case mv of + [Nothing] -> do + insertDResources [x] + reference x + + [Just cl] -> return cl + +type Filter a= (a -> Bool) +-- wait until a object (with a certaing key=keyResource x) meet a certain condition (useful to check external actions ) +-- a --a -> WF m (Stat a) a +--waitFor :: (IResource a, RefSerialize a) => Filter a -> a -> WF IO (Stat a) a +--waitFor filter x= (step $ waitFor1 filter) x where +--NOTE if you Delete the object from te cache, waitFor will no longuer work +waitFor :: (IResource a, Serialize a, Typeable a) => Filter a -> a -> IO a +waitFor filter x= do + tv <- reference (toIDyn $ x) + atomically $ do + dyn <- readTVar tv + let x= fromIDyn dyn + case filter x + of + False -> retry + True -> return x + + + + +waitUntil:: Integer -> IO() +waitUntil t= do + threadDelayInteger $ delay* 1000000 + + print ("execution at tnow="++show tnow++" ,t="++show t) + + where + tnow= n where TOD n _ = unsafePerformIO getClockTime + + delay | t-tnow >0 = t-tnow + | otherwise = 0 + + threadDelayInteger:: Integer -> IO() + threadDelayInteger time| time < imaxInt = threadDelay $ fromIntegral time + | otherwise=do threadDelay maxInt + threadDelayInteger ( time - imaxInt) + where maxInt = maxBound :: Int + imaxInt= fromIntegral maxInt + + + + diff --git a/Workflow.cabal b/Workflow.cabal --- a/Workflow.cabal +++ b/Workflow.cabal @@ -1,41 +1,18 @@ name: Workflow -version: 0.1 -synopsis: library for transparent execution of computations across shutdowns and restarts +version: 0.2 +synopsis: FreeChooser description: - transparent low level support (state logging, resume of the computation state, wait for data condition) for very long time - long living event driven processes. Workflow give the two first services to any monadic computation of type (a-> m a) - - f x >>=\x'-> g x' >>= \x''->... z by - - prefixing the user with the method "step": - - step f x >>= \x'-> step g x' >>= \x''->... - - This means that a workflow can be described with the familiar "do" notation. In principle, there is no other limitation - on the syntax but the restriction (a -> m a): All computations consume and produce the same type of data. - - for a monadic computation, Workflow provides: - - transparent checkpointing for each step in permanent storage (using TCache) - - resume of the monadic computation at the last checkpoint after soft or hard interruption - - use of versioning techniques for storing object changes (using RefSerialize) - - retrieval of the object at any previous step - - suspend the computation until the input object meet certain conditions. useful for inter-workflow - comunications. - - For various reasons, this package force the use of TCache for storage and refSerialize for writing to/from strings - at the end of the workflow all the intermediate data is erased. - see demo.hs and the header of Control.TCache for documentation. - + FreeChooser project -category: Control +category: Application license: BSD3 license-file: LICENSE author: Alberto Gómez Corona maintainer: agocorona@gmail.com Tested-With: GHC == 6.8.2 Build-Type: Simple -build-Depends: base, RefSerialize>=0.2.3, TCache>=0.5.4, stm > 2 +build-Depends: base, TCache>=0.5.4, network, RefSerialize>=0.2.3,stm>=2.1,old-time Cabal-Version: >= 1.2 exposed-modules: Control.Workflow diff --git a/Workflows/StatWorkflows b/Workflows/StatWorkflows deleted file mode 100644 --- a/Workflows/StatWorkflows +++ /dev/null @@ -1,1 +0,0 @@ -StatWorkflows [] diff --git a/demos/demo.hs b/demos/demo.hs --- a/demos/demo.hs +++ b/demos/demo.hs @@ -7,9 +7,10 @@ -- For bugs, questions, whatever, please email me: Alberto Gómez Corona agocorona@gmail.com module Main where -import Data.TCache (IResource(..)) +import Data.TCache.Dynamic (IResource(..),registerType) import Control.Workflow import Debug.Trace +import Data.Typeable import Control.Concurrent(forkIO, threadDelay) @@ -19,7 +20,7 @@ data Data = Name String |Try Int String | Finish String - | None deriving (Read,Show) -- for complex data, better to define a RefSerialize interface (see RefSerialize package) + | None deriving (Read,Show,Typeable) -- for complex data, better to define a RefSerialize interface (see RefSerialize package) instance IResource Data where keyResource (Name str) = "Name" -- the filename of the object to be stored @@ -27,8 +28,8 @@ keyResource (Finish _) = "Try-Finish" -- need tp be the same key for "wait" to check these two values. keyResource None = "None" defPath _ = "data/" -- path where temporary problem-related data will be stored - serialize x = undefined -- the WF monad is in charge of the persistence) - deserialize x= undefined -- (the WF monad is in charge of the persistence) + serialize = show -- the WF monad is in charge of the persistence) + deserialize = read -- (the WF monad is in charge of the persistence) @@ -43,6 +44,7 @@ -- RestartWorkflows restar all workflows in the pending workflow list -- if startWF is called when startworflow has already restarted a the same workflow main= do + registerType None forkIO $ do startWF "hello-ask" None wfList return() @@ -60,7 +62,7 @@ --syncWrite True -- this is the default name@(Name str) <- step askName d step printname name - askNumbers str $ Try 0 "" + askNumbers str $ Try 0 "33" askName :: Data-> IO Data askName _ = do @@ -77,7 +79,7 @@ let filter (Try _ "5") = True filter (Finish _)= True filter _= False - r <- step (waitFor filter) (Try 0 "") -- wait the other thread to store an object with the same key than Try 0 "" + r <- step (waitFor filter) (Try 0 "13") -- wait the other thread to store an object with the same key than Try 0 "" -- with the string "5" or termination case r of Try n "5" -> step1 $ return $ Finish $ "done in the step "++ show n++" !" -- the WF monad store the Finish message @@ -112,7 +114,7 @@ anything= \_-> True askNumber:: String -> Data ->IO Data -askNumber name None= askNumber name (Try 0 "") +askNumber name None= askNumber name (Try 0 "22") askNumber name d@(Try i num)= do print $ "Mr "++name++ " this is your try number "++show (i+1) ++". Guess my number" diff --git a/demos/demo3.hs b/demos/demo3.hs --- a/demos/demo3.hs +++ b/demos/demo3.hs @@ -1,3 +1,4 @@ +{-# 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 @@ -6,7 +7,8 @@ module Main where import Control.Workflow -import Data.TCache +import Data.TCache.Dynamic +import Data.Typeable fact 0 =1 fact n= n * fact (n-1) @@ -28,12 +30,12 @@ -- now the workflow versión -data Fact= Fact Integer Integer deriving (Read, Show) +data Fact= Fact Integer Integer deriving (Read, Show, Typeable) instance IResource Fact where keyResource _= "lastfact" - serialize= undefined - deserialize= undefined + serialize= show + deserialize= read factorialsWF _= do all <- getAll @@ -45,7 +47,7 @@ factLoop fct= do nf <- step (\_ -> do - putStrLn "give me a number if you enter a letter, the program will abort. please restart" + putStrLn "give me a number if you enter a letter, 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 @@ -59,4 +61,6 @@ _ -> factLoop nf -main = startWF "factorials" (Fact 0 0) [("factorials",factorialsWF)] +main = do + registerType $ Fact 0 0 -- register this datatype + startWF "factorials" (Fact 0 0) [("factorials",factorialsWF)]