Workflow 0.1 → 0.2

raw patch · 6 files changed

+491/−98 lines, 6 filesdep +networkdep +old-timedep ~stmPVP ok

version bump matches the API change (PVP)

Dependencies added: network, old-time

Dependency ranges changed: stm

API changes (from Hackage documentation)

Files

• Control/Workflow.hs +86/−56
• Control/Workflow.new.hs +381/−0
• Workflow.cabal +5/−28
• Workflows/StatWorkflows +0/−1
• demos/demo.hs +9/−7
• demos/demo3.hs +10/−6

@@ -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+++           +           

@@ -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+++           +           

@@ -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

@@ -1,1 +0,0 @@-StatWorkflows []

@@ -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"

@@ -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)]