Workflow (empty) → 0.1
raw patch · 7 files changed
+611/−0 lines, 7 filesdep +RefSerializedep +TCachedep +basesetup-changed
Dependencies added: RefSerialize, TCache, base, stm
Files
- Control/Workflow.hs +353/−0
- LICENSE +27/−0
- Setup.lhs +5/−0
- Workflow.cabal +42/−0
- Workflows/StatWorkflows +1/−0
- demos/demo.hs +121/−0
- demos/demo3.hs +62/−0
+ Control/Workflow.hs view
@@ -0,0 +1,353 @@+{-# 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++ ,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 ) ++ ,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)+import Control.Concurrent.STM(atomically, retry, readTVar)+import Debug.Trace++import Data.TCache+import Data.RefSerialize+import Data.List((\\),find,elemIndices)++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]} + | I a+++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 (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+ 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 + + rData= do+ symbol "I"+ a <- rreadp+ return $ I a ++ rWorkflows= do+ symbol "StatWorkflows"+ list <- readp+ return $ Workflows list+ +--persistence trough TCache , default persistence in files + +instance (IResource a, Serialize 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"++ defPath (I a)= defPath a+ defPath _= "Workflows/" -- directory for Workflow data+ --serialize (I x)= "I "++ serialize x + serialize x= runW $ showp x+ deserialize str = runR readp str++prefix= "Stat."+lengthPrefix= length prefix++insertResources xs= withResources [] (\_-> 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) => 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 $ Workflows newlist,Delete s, Insert s', Insert (I x')] + -- `debug`("insert "++keyResource s'++","++keyResource x'++" delete: "+++ -- keyResource s++","++keyResource x)+ | otherwise= [Insert s', Insert (I x')] + -- `debug`("insert "++keyResource s'++","++keyResource x')+ let+ doit [Nothing] = doit1 []+ doit [Just (Workflows xs)] = doit1 xs++ withResourcesID [Workflows undefined] doit+ + when (sync s) $ unsafeIOtoWF $ syncCache (refcache :: Cache (Stat a))+ + 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+ 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 $ readResource $ (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+ mst <- unsafeIOtoWF $ getResource stat0{wfName= drop lengthPrefix key1}+ 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+ 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= let [init,end]= elemIndices '.' key + + start= drop (init + 1) key+ in take (end-init -1) start + 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: "++ key++ + + 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 (Workflows xs)]= + 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'] ++ + unsafeIOtoWF $ withResourcesID [Workflows undefined] delWF + when (sync s) $ unsafeIOtoWF $ syncCache (refcache :: Cache (Stat a)) + 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 (refcache :: Cache (Stat a)) time defaultCheck maxsize+ return ()+ return (s{ sync= bool},()))+++instance (IResource a,Serialize 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 <- getTVars [x ]+ case mv of+ [Nothing] -> do + insertResources [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) => Filter a -> a -> IO a+waitFor filter x= do+ + tv <- reference (I x) + atomically $ do + I x <- readTVar tv+ case filter x + of + False -> retry+ True -> return x + ++
+ LICENSE view
@@ -0,0 +1,27 @@+Copyright (c) Alberto Gómez Corona 2008++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:+1. Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.+2. Redistributions in binary form must reproduce the above copyright+ notice, this list of conditions and the following disclaimer in the+ documentation and/or other materials provided with the distribution.+3. Neither the name of the author nor the names of his contributors+ may be used to endorse or promote products derived from this software+ without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE+ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS+OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT+LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY+OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF+SUCH DAMAGE.
+ Setup.lhs view
@@ -0,0 +1,5 @@+#! /usr/bin/runghc + +> import Distribution.Simple +> +> main = defaultMain
+ Workflow.cabal view
@@ -0,0 +1,42 @@+name: Workflow+version: 0.1+synopsis: library for transparent execution of computations across shutdowns and restarts+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.++ ++category: Control+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+Cabal-Version: >= 1.2++exposed-modules: Control.Workflow+ghc-options:
+ Workflows/StatWorkflows view
@@ -0,0 +1,1 @@+StatWorkflows []
+ demos/demo.hs view
@@ -0,0 +1,121 @@+{-# OPTIONS -fglasgow-exts #-}+-- example of the Workflow package .+-- two workflows that interact by modifing the structure "Data". 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 with 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 Data.TCache (IResource(..))+import Control.Workflow+import Debug.Trace+import Control.Concurrent(forkIO, threadDelay)+++debug a b = trace b a++wfList= [("hello-ask",hello),("wait",wait)]++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)++instance IResource Data where+ keyResource (Name str) = "Name" -- the filename of the object to be stored+ keyResource (Try n s) = "Try-Finish" -- need to be the same key for "wait" to check these two values.+ 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) + + +++-- startWF insert a workflow in the workflow list and start it. 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 last versión of the object++-- In general the last can change, therefore a the workflow identifier change. In this case, startWF will not detect that+-- condition and will restart a new workflow with the starting object+-- RestartWorkflows restar all workflows in the pending workflow list+-- if startWF is called when startworflow has already restarted a the same workflow +main= do + forkIO $ do+ startWF "hello-ask" None wfList + return()++ startWF "wait" None wfList+ +++hello d= 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 True -- this is the default+ name@(Name str) <- step askName d + step printname name + askNumbers str $ Try 0 "" ++askName :: Data-> IO Data+askName _ = do + print "what is your name?"+ str <- getLine+ return $ Name str++printname :: Data -> IO Data+printname (Name name)= do+ print $ "hello Mr "++name+ return None+ +wait _ = do+ 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 ""+ -- 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+ + Finish str -> step2 $ print str -- finish message sent by the other thread (max count reached)+ + step2 $ threadDelay 1000000 -- wait the finalization of the other thead + ++step1 f= step (\_->f) None ++step2 f= step(\_->f >> return None) None + + + +askNumbers :: String -> Data -> Workflow IO Data +askNumbers name d = do + step2 $ threadDelay 5000 -- wait for the other tread to process.+ + r <- step (waitFor anything) d -- get the last value of the object with key "try-finish", to look for the other thread actions+ case r of+ + Finish msg -> step2 $ print msg --the other thread sent a finalize response+ -- `debug` "hello-ask: finish detected"+ Try 9 num -> step1 $ return $ Finish "sorry, no more guesses" --send finalization to the wait thread+ -- `debug` "hello-ask: Finish returned"+ _ -> do + nd <- step (askNumber name) d++ askNumbers name nd + where+ anything= \_-> True + +askNumber:: String -> Data ->IO Data+askNumber name None= askNumber name (Try 0 "")+askNumber name d@(Try i num)= do ++ print $ "Mr "++name++ " this is your try number "++show (i+1) ++". Guess my number"+ str <- getLine+ return $ Try (i+1) str+
+ demos/demo3.hs view
@@ -0,0 +1,62 @@+-- 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+import Data.TCache++fact 0 =1+fact n= n * fact (n-1)++{-+this is the origina program without workflow++factorials= do+ print "give me a number"+ str<- getLine+ let n= read str :: Integer+ let fct= fact n+ print fct+ factorials++ +main1= factorials+-}+++-- now the workflow versión+data Fact= Fact Integer Integer deriving (Read, Show)++instance IResource Fact where+ keyResource _= "lastfact"+ serialize= undefined+ deserialize= undefined++factorialsWF _= do+ all <- getAll+ unsafeIOtoWF $ putStrLn "Factorials calculated so far:"+ unsafeIOtoWF $ putStrLn $ concatMap (\(Fact n fct)-> "number "++ show n ++ " factorial is "++ show fct++ "\n") all+ factLoop (Fact 0 0)+ where++ factLoop fct= do++ nf <- step (\_ -> do + putStrLn "give me a number if you enter a letter, the program will abort. please restart"+ 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) fct+ + case nf of+ Fact 0 _ -> do+ unsafeIOtoWF $ print "bye" + return (Fact 0 0)+ _ -> factLoop nf+++main = startWF "factorials" (Fact 0 0) [("factorials",factorialsWF)]