diff --git a/Control/Workflow.hs b/Control/Workflow.hs new file mode 100644 --- /dev/null +++ b/Control/Workflow.hs @@ -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 + + + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,5 @@ +#! /usr/bin/runghc + +> import Distribution.Simple +> +> main = defaultMain diff --git a/Workflow.cabal b/Workflow.cabal new file mode 100644 --- /dev/null +++ b/Workflow.cabal @@ -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: diff --git a/Workflows/StatWorkflows b/Workflows/StatWorkflows new file mode 100644 --- /dev/null +++ b/Workflows/StatWorkflows @@ -0,0 +1,1 @@ +StatWorkflows [] diff --git a/demos/demo.hs b/demos/demo.hs new file mode 100644 --- /dev/null +++ b/demos/demo.hs @@ -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 + diff --git a/demos/demo3.hs b/demos/demo3.hs new file mode 100644 --- /dev/null +++ b/demos/demo3.hs @@ -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)]