Workflow 0.5.8.2 → 0.6.0.0
raw patch · 23 files changed
+1759/−2218 lines, 23 files
Files
- Control/Workflow.hs +1050/−0
- Control/Workflow/Binary.hs +0/−105
- Control/Workflow/Binary/BinDefs.hs +0/−104
- Control/Workflow/Binary/Patterns.hs +0/−107
- Control/Workflow/GenSerializer.hs +0/−70
- Control/Workflow/IDynamic.hs +37/−6
- Control/Workflow/Patterns.hs +271/−0
- Control/Workflow/Patterns.inc.hs +0/−172
- Control/Workflow/Stat.hs +113/−6
- Control/Workflow/Text.hs +0/−118
- Control/Workflow/Text/Patterns.hs +0/−97
- Control/Workflow/Text/TextDefs.hs +0/−161
- Control/Workflow/Workflow.inc.hs +0/−930
- Data/Persistent/Queue.hs +240/−0
- Data/Persistent/Queue/Binary.hs +0/−58
- Data/Persistent/Queue/Queue.inc.hs +0/−205
- Data/Persistent/Queue/Text.hs +0/−60
- Demos/Fact.hs +5/−1
- Demos/Inspect.hs +1/−1
- Demos/docAprobal.hs +3/−3
- Demos/pr.hs +26/−0
- Demos/sequence.hs +1/−1
- Workflow.cabal +12/−13
+ Control/Workflow.hs view
@@ -0,0 +1,1050 @@+{-# LANGUAGE OverlappingInstances+ , UndecidableInstances+ , ExistentialQuantification+ , ScopedTypeVariables+ , MultiParamTypeClasses+ , FlexibleInstances+ , FlexibleContexts+ , TypeSynonymInstances+ , DeriveDataTypeable++ #-}+{-# OPTIONS -IControl/Workflow #-}++{- | A workflow can be seen as a persistent thread.+The workflow monad writes a log that permit to restore the thread+at the interrupted point. `step` is the (partial) monad transformer for+the Workflow monad. A workflow is defined by its name and, optionally+by the key of the single parameter passed. The primitives for starting workflows+also restart the workflow when it has been in execution previously.++This is the main module that uses the `RefSerialize` paclkage for serialization. Here the constraint @DynSerializer w r a@ is equivalent to+@Data.RefSerialize a@++For workflows that uses big structures, for example, documents+use this module in combination with the RefSerialize package to define the (de)serialization instances+The log size will be reduced. printWFHistory` method will print the structure changes+in each step.++If instead of RefSerialize, you define read and show instances, there will+ be no reduction. but still the log will be readable for debugging purposes.++for workflows that does not care about this, use the binary alternative: "Control.Workflow.Binary"++A small example that print the sequence of integers in te console+if you interrupt the progam, when restarted again, it will+start from the last printed number++@module Main where+import Control.Workflow.Text+import Control.Concurrent(threadDelay)+import System.IO (hFlush,stdout)+++mcount n= do `step` $ do+ putStr (show n ++ \" \")+ hFlush stdout+ threadDelay 1000000+ mcount (n+1)+ return () -- to disambiguate the return type++main= `exec1` \"count\" $ mcount (0 :: Int)@++-}++module Control.Workflow+(+ Workflow -- a useful type name+, WorkflowList+, PMonadTrans (..)+, MonadCatchIO (..)+, throw+, Indexable(..)+-- * Start/restart workflows+, start+, exec+, exec1d+, exec1+, wfExec+, startWF+, restartWorkflows+, WFErrors(..)+-- * Lifting to the Workflow monad+, step+, stepControl+, unsafeIOtoWF+-- * References to intermediate values in the workflow log+, WFRef+, getWFRef+, newWFRef+, stepWFRef+, readWFRef+, writeWFRef+-- * Workflow inspect+, waitWFActive+, getAll+, safeFromIDyn+, getWFKeys+, getWFHistory+, waitFor+, waitForSTM+-- * Persistent timeouts+, waitUntilSTM+, getTimeoutFlag+-- * Trace logging+, logWF+-- * Termination of workflows+, clearRunningFlag+, killThreadWF+, killWF+, delWF+, killThreadWF1+, killWF1+, delWF1+, delWFHistory+, delWFHistory1+-- * Log writing policy+, syncWrite+, SyncMode(..)+-- * Print log history+, printHistory+)+where++import Prelude hiding (catch)+import System.IO.Unsafe+import Control.Monad(when,liftM)+import qualified Control.Exception as CE (Exception,AsyncException(ThreadKilled), SomeException, throwIO, handle,finally,catch,block,unblock)+import Control.Concurrent (forkIO,threadDelay, ThreadId, myThreadId, killThread)+import Control.Concurrent.STM+import GHC.Conc(unsafeIOToSTM)+import GHC.Base (maxInt)+++import Data.ByteString.Lazy.Char8 as B hiding (index)+import Data.ByteString.Lazy as BL(putStrLn)+import Data.List as L+import Data.Typeable+import System.Time+import Control.Monad.Trans+import Control.Concurrent.MonadIO(HasFork(..),MVar,newMVar,takeMVar,putMVar)+++import System.IO(hPutStrLn, stderr)+import Data.List(elemIndex)+import Data.Maybe(fromJust, isNothing, isJust, mapMaybe)+import Data.IORef+import System.IO.Unsafe(unsafePerformIO)+import Data.Map as M(Map,fromList,elems, insert, delete, lookup,toList, fromList,keys)+import qualified Control.Monad.CatchIO as CMC+import qualified Control.Exception.Extensible as E++import Data.TCache+import Data.TCache.DefaultPersistence+import Data.RefSerialize+import Control.Workflow.IDynamic+import Unsafe.Coerce+import Control.Workflow.Stat+--+--import Debug.Trace+--a !> b= trace b a+++type Workflow m = WF Stat m -- not so scary++type WorkflowList m a b= [(String, a -> Workflow m b) ]+++instance Monad m => Monad (WF s m) 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'))++++instance (Monad m,Functor m) => Functor (Workflow m ) where+ fmap f (WF g)= WF (\s -> do+ (s1, x) <- g s+ return (s1, f x))++tvRunningWfs = getDBRef $ keyRunning :: DBRef Stat++++-- | executes a computation inside of the workflow monad whatever the monad encapsulated in the workflow.+-- Warning: this computation is executed whenever+-- the workflow restarts, no matter if it has been already executed previously. This is useful for intializations or debugging.+-- To avoid re-execution when restarting use: @'step' $ unsafeIOtoWF...@+--+-- To perform IO actions in a workflow that encapsulates an IO monad, use step over the IO action directly:+--+-- @ 'step' $ action @+--+-- instead of+--+-- @ 'step' $ unsafeIOtoWF $ action @+unsafeIOtoWF :: (Monad m) => IO a -> Workflow m a+unsafeIOtoWF x= let y= unsafePerformIO ( x >>= return) in y `seq` return y+++{- | PMonadTrans permits |to define a partial monad transformer. They are not defined for all kinds of data+but the ones that have instances of certain classes.That is because in the lift instance code there are some+hidden use of these classes. This also may permit an accurate control of effects.+An instance of MonadTrans is an instance of PMonadTrans+-}+class PMonadTrans t m a where+ plift :: Monad m => m a -> t m a++++-- | plift= step+instance (Monad m+ , MonadIO m+ , Serialize a+ , Typeable a)+ => PMonadTrans (WF Stat) m a+ where+ plift = step++-- | An instance of MonadTrans is an instance of PMonadTrans+instance (MonadTrans t, Monad m) => PMonadTrans t m a where+ plift= Control.Monad.Trans.lift++instance Monad m => MonadIO (WF Stat m) where+ liftIO=unsafeIOtoWF+++{- | adapted from MonadCatchIO-mtl. Workflow need to express serializable constraints about the returned values,+so the usual class definitions for lifting IO functions are not suitable.+-}++class MonadCatchIO m a where+ -- | Generalized version of 'E.catch'+ catch :: E.Exception e => m a -> (e -> m a) -> m a++ -- | Generalized version of 'E.block'+ block :: m a -> m a++ -- | Generalized version of 'E.unblock'+ unblock :: m a -> m a++++-- | Generalized version of 'E.throwIO'+throw :: (MonadIO m, E.Exception e) => e -> m a+throw = liftIO . E.throwIO++{-++{-+-- | Generalized version of 'E.try'+try :: (MonadCatchIO m a, E.Exception e) => m a -> m (Either e a)++-- | Generalized version of 'E.tryJust'+tryJust :: (MonadCatchIO m a, E.Exception e)+ => (e -> Maybe b) -> m a -> m (Either b a)++-}+-- | Generalized version of 'E.Handler'+data Handler m a = forall e . E.Exception e => Handler (e -> m a)+++{-+instance (MonadCatchIO m a, Error e) => MonadCatchIO (ErrorT e m) a where+ m `catch` f = mapErrorT (\m' -> m' `catch` (\e -> runErrorT $ f e)) m+ block = mapErrorT block+ unblock = mapErrorT unblock+-}++++try a = catch (a >>= \ v -> return (Right v)) (\e -> return (Left e))++tryJust p a = do+ r <- try a+ case r of+ Right v -> return (Right v)+ Left e -> case p e of+ Nothing -> throw e `asTypeOf` (return $ Left undefined)+ Just b -> return (Left b)++-- | Generalized version of 'E.bracket'+bracket :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c) => m a -> (a -> m b) -> (a -> m c) -> m c+bracket before after thing =+ block (do a <- before+ r <- unblock (thing a) `onException` after a+ _void $ after a+ return r)++-- | A variant of 'bracket' where the return value from the first computation+-- is not required.+bracket_ :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c)+ => m a -- ^ computation to run first (\"acquire resource\")+ -> m b -- ^ computation to run last (\"release resource\")+ -> m c -- ^ computation to run in-between+ -> m c -- returns the value from the in-between computation+bracket_ before after thing =+ block $ do _void before+ r <- unblock thing `onException` after+ _void after+ return r++-- | A specialised variant of 'bracket' with just a computation to run+-- afterward.+finally :: (Monad m, MonadIO m, MonadCatchIO m a)+ => m a -- ^ computation to run first+ -> m b -- ^ computation to run afterward (even if an exception was+ -- raised)+ -> m a -- returns the value from the first computation+thing `finally` after =+ block $ do r <- unblock thing `onException` after+ _void after+ return r+{-+-- | Like 'bracket', but only performs the final action if there was an+-- exception raised by the in-between computation.+bracketOnError :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c)+ => m a -- ^ computation to run first (\"acqexeuire resource\")+ -> (a -> m b)-- ^ computation to run last (\"release resource\")+ -> (a -> m c)-- ^ computation to run in-between+ -> m c -- returns the value from the in-between+ -- computation+bracketOnError before after thing =+ block $ do a <- before+ unblock (thing a) `onException` after a+-}+-- | Generalized version of 'E.onException'+onException :: (MonadIO m, MonadCatchIO m a) => m a -> m b -> m a+onException a onEx = a `catch` (\e -> onEx >> throw (e:: E.SomeException))++_void :: Monad m => m a -> m ()+_void a = a >> return ()+++-}+++++instance (Serialize a+ , Typeable a,MonadIO m, CMC.MonadCatchIO m)+ => MonadCatchIO (WF Stat m) a where+ catch exp exc = do+ expwf <- step $ getTempName+ excwf <- step $ getTempName+ step $ do+ ex <- CMC.catch (exec1d expwf exp >>= return . Right+ ) $ \e-> return $ Left e++ case ex of+ Right r -> return r -- All right+ Left e ->exec1d excwf (exc e)+ -- An exception occured in the main workflow+ -- the exception workflow is executed+++++ block exp=WF $ \s -> CMC.block (st exp $ s)++ unblock exp= WF $ \s -> CMC.unblock (st exp $ s)++++instance (HasFork io+ , CMC.MonadCatchIO io)+ => HasFork (WF Stat io) where+ fork f = do+ (str, finished) <- step $ getTempName >>= \n -> return(n, False)+ r <- getWFRef+ WF (\s ->+ do th <- if finished+ then fork $ return ()+ else fork $ do+ exec1 str f+ liftIO $ do atomically $ writeWFRef r (str, True)+ syncIt+ return(s,th))+++++-- | start or restart an anonymous workflow inside another workflow+-- its state is deleted when finished and the result is stored in+-- the parent's WF state.+wfExec+ :: (Indexable a, Serialize a, Typeable a+ , CMC.MonadCatchIO m, MonadIO m)+ => Workflow m a -> Workflow m a+wfExec f= do+ str <- step $ getTempName+ step $ exec1 str f++-- | a version of exec1 that deletes its state after complete execution or thread killed+exec1d :: (Serialize b, Typeable b+ ,CMC.MonadCatchIO m)+ => String -> (Workflow m b) -> m b+exec1d str f= do+ r <- exec1 str f+ delit+ return r+ `CMC.catch` (\e@CE.ThreadKilled -> delit >> throw e)++ where+ delit= do+ delWF str ()+ liftIO syncIt -- !> str++++-- | a version of exec with no seed parameter.+exec1 :: ( Serialize a, Typeable a+ , Monad m, MonadIO m, CMC.MonadCatchIO m)+ => String -> Workflow m a -> m a++exec1 str f= exec str (const f) ()+++++-- | start or continue a workflow with exception handling+-- | the workflow flags are updated even in case of exception+-- | `WFerrors` are raised as exceptions+exec :: ( Indexable a, Serialize a, Serialize b, Typeable a+ , Typeable b+ , Monad m, MonadIO m, CMC.MonadCatchIO m)+ => String -> (a -> Workflow m b) -> a -> m b+exec str f x =+ (do+ v <- getState str f x+ case v of+ Right (name, f, stat) -> do+ r <- runWF name (f x) stat+ return r+ Left err -> CMC.throw err)+ `CMC.catch`+ (\(e :: CE.SomeException) -> liftIO $ do+ let name= keyWF str x+ clearRunningFlag name --`debug` ("exception"++ show e)+ syncIt+ CMC.throw e )+++++mv :: MVar Int+mv= unsafePerformIO $ newMVar 0++getTempName :: MonadIO m => m String+getTempName= liftIO $ do+ seq <- takeMVar mv+ putMVar mv (seq + 1)+ TOD t _ <- getClockTime+ return $ "anon"++ show t ++ show seq+++++++instance Indexable () where+ key= show++-- | lifts a monadic computation to the WF monad, and provides transparent state loging and resuming of computation+step :: ( Monad m+ , MonadIO m+ , Serialize a+ , Typeable a)+ => m a+ -> Workflow m a+step= stepControl1 False++-- | permits modification of the workflow state by the procedure being lifted+-- if the boolean value is True. This is used internally for control purposes+stepControl :: ( Monad m+ , MonadIO m+ , Serialize a+ , Typeable a)+ => m a+ -> Workflow m a+stepControl= stepControl1 True++stepControl1 :: ( Monad m+ , MonadIO m+ , Serialize a+ , Typeable a)+ => Bool -> m a+ -> Workflow m a+stepControl1 isControl mx= WF(\s'' -> do+ let stat= state s''+ let ind= index s''+ if recover s'' && ind < stat+ then return (s''{index=ind +1 }, fromIDyn $ versions s'' !! (stat - ind-1) )+ else do+ x' <- mx+ let sref = self s''+ s'<- liftIO . atomically $ do+ s <- if isControl+ then readDBRef sref >>= unjustify ("step: readDBRef: not found:" ++ keyObjDBRef sref)+ else return s''+ let versionss= versions s+ let dynx= toIDyn x'+ let ver= dynx: versionss+ let s'= s{ recover= False, versions = ver, state= state s+1}++ writeDBRef sref s'+ return s'+ liftIO syncIt+ return (s', x') )++unjustify str Nothing = error str+unjustify _ (Just x) = return x+++++-- | start or continue a workflow with no exception handling.+-- | the programmer has to handle inconsistencies in the workflow state+-- | using `killWF` or `delWF` in case of exception.+start+ :: ( Monad m+ , MonadIO m+ , Indexable a+ , Serialize a, Serialize b+ , Typeable a+ , Typeable b)+ => String -- ^ name thar identifies the workflow.+ -> (a -> Workflow m b) -- ^ workflow to execute+ -> a -- ^ initial value (ever use the initial value for restarting the workflow)+ -> m (Either WFErrors b) -- ^ result of the computation+start namewf f1 v = do+ ei <- getState namewf f1 v+ case ei of+ Right (name, f, stat) ->+ runWF name (f v) stat >>= return . Right++ Left error -> return $ Left error++-- | return conditions from the invocation of start/restart primitives+data WFErrors = NotFound | AlreadyRunning | Timeout | forall e.CE.Exception e => Exception e deriving Typeable++instance Show WFErrors where+ show NotFound= "Not Found"+ show AlreadyRunning= "Already Running"+ show Timeout= "Timeout"+ show (Exception e)= "Exception: "++ show e++instance CE.Exception WFErrors++--tvRunningWfs = unsafePerformIO . refDBRefIO $ Running (M.fromList [] :: Map String (String, (Maybe ThreadId)))++{-+lookup for any workflow for the entry value v+if namewf is found and is running, return arlready running+ if is not runing, restart it+else start anew.+-}+++getState :: (Monad m, MonadIO m, Indexable a, Serialize a, Typeable a)+ => String -> x -> a+ -> m (Either WFErrors (String, x, Stat))+getState namewf f v= liftIO . atomically $ getStateSTM+ where+ getStateSTM = do+ mrunning <- readDBRef tvRunningWfs+ case mrunning of+ Nothing -> do+ writeDBRef tvRunningWfs (Running $ fromList [])+ getStateSTM+ Just(Running map) -> do+ let key= keyWF namewf v+ stat1= stat0{wfName= key,versions=[toIDyn v],self= sref}+ sref= getDBRef $ keyResource stat1+ case M.lookup key map of+ Nothing -> do -- no workflow started for this object+ mythread <- unsafeIOToSTM $ myThreadId+ writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map+ writeDBRef sref stat1+ return $ Right (key, f, stat1)++ Just (wf, started) -> -- a workflow has been initiated for this object+ if isJust started+ then return $ Left AlreadyRunning -- `debug` "already running"+ else do -- has been started but not running now+ mythread <- unsafeIOToSTM $ myThreadId+ writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map+ mst <- readDBRef sref+ let stat' = case mst of+ Nothing -> error $ "Workflow not found: "++ key+ Just s -> s{index=0,recover= True}+ writeDBRef sref stat'+ return $ Right (key, f, stat')++syncIt= do+ (sync,_) <- atomically $ readTVar tvSyncWrite+ when (sync ==Synchronous) syncCache++runWF :: (Monad m,MonadIO m+ , Serialize b, Typeable b)+ => String -> Workflow m b -> Stat -> m b+runWF n f s= do+ sync <- liftIO $! do+ (sync,_) <- atomically $ readTVar tvSyncWrite+ when (sync ==Synchronous) syncCache+ return sync+ (s', v') <- st f $ s+ liftIO $! do+ clearFromRunningList n+ when (sync ==Synchronous) syncCache+ return v'+ where++ -- eliminate the thread from the list of running workflows but leave the state+ clearFromRunningList n = atomically $ do+ Just(Running map) <- readDBRef tvRunningWfs+ writeDBRef tvRunningWfs . Running $ M.delete n map -- `debug` "clearFromRunningList"++-- | start or continue a workflow from a list of workflows in the IO monad with exception handling. The excepton is returned as a Left value+startWF+ :: ( MonadIO m+ , Serialize a, Serialize b+ , Typeable a+ , Indexable a+ , Typeable b)+ => String -- ^ name of workflow in the workflow list+ -> a -- ^ initial value (ever use the initial value even to restart the workflow)+ -> WorkflowList m a b -- ^ function to execute+ -> m (Either WFErrors b) -- ^ result of the computation+startWF namewf v wfs=+ case Prelude.lookup namewf wfs of+ Nothing -> return $ Left NotFound+ Just f -> start namewf f v++-- | re-start the non finished workflows in the list, for all the initial values that they may have been called+restartWorkflows+ :: (Serialize a, Serialize b, Typeable a+ , Indexable b, Typeable b)+ => WorkflowList IO a b -- the list of workflows that implement the module+ -> IO () -- Only workflows in the IO monad can be restarted with restartWorkflows+restartWorkflows map = do+ mw <- liftIO $ getResource ((Running undefined ) ) -- :: IO (Maybe(Stat a))+ case mw of+ Nothing -> return ()+ Just (Running all) -> mapM_ start . mapMaybe filter . toList $ all+ where+ filter (a, (b,Nothing)) = Just (b, a)+ filter _ = Nothing++ start (key, kv)= do++ --let key1= key ++ "#" ++ kv+ let mf= Prelude.lookup key map+ case mf of+ Nothing -> return ()+ Just f -> do+ let st0= stat0{wfName = kv}+ mst <- liftIO $ getResource st0+ case mst of+ Nothing -> error $ "restartWorkflows: workflow not found "++ keyResource st0+ Just st-> do+ liftIO . forkIO $ runWF key (f (fromIDyn . Prelude.last $ versions st )) st{index=0,recover=True} >> return ()+ return ()++-- | choose between text and binary persistence for the workflow state+-- text persistence is used for++-- *(1)debugging purposes++-- * (2)when step returns largue structures that share common contents between steps,+-- for example, when a workflow edit and ammend a document among many users++-- * (3) When tracking the modifications made in the object trough `getWFHistory` or+-- `printWFHistory`++++-- |+-- The execution log is cached in memory using the package `TCache`. This procedure defines the polcy for writing the cache into permanent storage.+--+-- For fast workflows, or when TCache` is used also for other purposes , `Asynchronous` is the best option+--+-- `Asynchronous` mode invokes `clearSyncCache`. For more complex use of the syncronization+-- please use this `clearSyncCache`.+--+-- When interruptions are controlled, use `SyncManual` mode and include a call to `syncCache` in the finalizaton code++syncWrite:: (Monad m, MonadIO m) => SyncMode -> m ()+syncWrite mode= do+ (_,thread) <- liftIO . atomically $ readTVar tvSyncWrite+ when (isJust thread ) $ liftIO . killThread . fromJust $ thread+ case mode of+ Synchronous -> modeWrite+ SyncManual -> modeWrite+ Asyncronous time maxsize -> do+ th <- liftIO $ clearSyncCacheProc time defaultCheck maxsize >> return()+ liftIO . atomically $ writeTVar tvSyncWrite (mode,Just th)+ where+ modeWrite=+ liftIO . atomically $ writeTVar tvSyncWrite (mode, Nothing)+++-- | return all the steps of the workflow log. The values are dynamic+--+-- to get all the steps with result of type Int:+-- @all <- `getAll`+-- let lfacts = mapMaybe `safeFromIDyn` all :: [Int]@+getAll :: Monad m => Workflow m [IDynamic]+getAll= WF(\s -> return (s, versions s))+++-- | return the list of object keys that are running for a workflow+getWFKeys :: String -> IO [String]+getWFKeys wfname= do+ mwfs <- atomically $ readDBRef tvRunningWfs+ case mwfs of+ Nothing -> return []+ Just (Running wfs) -> return $ Prelude.filter (L.isPrefixOf wfname) $ M.keys wfs++-- | return the current state of the computation, in the IO monad+getWFHistory :: (Indexable a, Serialize a) => String -> a -> IO (Maybe Stat)+getWFHistory wfname x= getResource stat0{wfName= keyWF wfname x}+++delWFHistory name1 x= do+ let name= keyWF name1 x+ delWFHistory1 name++delWFHistory1 name =+ atomically . withSTMResources [] $ const resources{ toDelete= [stat0{wfName= name}] }+++waitWFActive wf= do+ r <- threadWF wf+ case r of -- wait for change in the wofkflow state+ Just (_, Nothing) -> retry+ _ -> return ()+ where+ threadWF wf= do+ Just(Running map) <- readDBRef tvRunningWfs + return $ M.lookup wf map +++-- | kill the executing thread if not killed, but not its state.+-- `exec` `start` or `restartWorkflows` will continue the workflow+killThreadWF :: ( Indexable a+ , Serialize a++ , Typeable a+ , MonadIO m)+ => String -> a -> m()+killThreadWF wfname x= do+ let name= keyWF wfname x+ killThreadWF1 name++-- | a version of `KillThreadWF` for workflows started wit no parameter by `exec1`+killThreadWF1 :: MonadIO m => String -> m()+killThreadWF1 name= killThreadWFm name >> return ()++killThreadWFm name= do+ (map,f) <- clearRunningFlag name+ case f of+ Just th -> liftIO $ killThread th+ Nothing -> return()+ return map++++-- | kill the process (if running) and drop it from the list of+-- restart-able workflows. Its state history remains , so it can be inspected with+-- `getWfHistory` `printWFHistory` and so on+killWF :: (Indexable a,MonadIO m) => String -> a -> m ()+killWF name1 x= do+ let name= keyWF name1 x+ killWF1 name++-- | a version of `KillWF` for workflows started wit no parameter by `exec1`+killWF1 :: MonadIO m => String -> m ()+killWF1 name = do+ map <- killThreadWFm name+ liftIO . atomically . writeDBRef tvRunningWfs . Running $ M.delete name map+ return ()++-- | delete the WF from the running list and delete the workflow state from persistent storage.+-- Use it to perform cleanup if the process has been killed.+delWF :: ( Indexable a+ , MonadIO m+ , Typeable a)+ => String -> a -> m()+delWF name1 x= do+ let name= keyWF name1 x+ delWF1 name+++-- | a version of `delWF` for workflows started wit no parameter by `exec1`+delWF1 :: MonadIO m=> String -> m()+delWF1 name= liftIO $ do+ mrun <- atomically $ readDBRef tvRunningWfs+ case mrun of+ Nothing -> return()+ Just (Running map) -> do+ atomically . writeDBRef tvRunningWfs . Running $! M.delete name map+ delWFHistory1 name+ syncIt++++clearRunningFlag name= liftIO $ atomically $ do+ mrun <- readDBRef tvRunningWfs+ case mrun of+ Nothing -> error $ "clearRunningFLag non existing workflows" ++ name+ Just(Running map) -> do+ case M.lookup name map of+ Just(_, Nothing) -> return (map,Nothing)+ Just(v, Just th) -> do+ writeDBRef tvRunningWfs . Running $ M.insert name (v, Nothing) map+ return (map,Just th)+ Nothing ->+ return (map, Nothing)++-- | Return the reference to the last logged result , usually, the last result stored by `step`.+-- wiorkflow references can be accessed outside of the workflow+-- . They also can be (de)serialized.+--+-- WARNING getWFRef can produce casting errors when the type demanded+-- do not match the serialized data. Instead, `newDBRef` and `stepWFRef` are type safe at runtuime.+getWFRef :: ( Monad m,+ MonadIO m,+ Serialize a+ , Typeable a)+ => Workflow m (WFRef a)+getWFRef =ret+ where+ ret= WF (\s -> do+ let n= if recover s then index s else state s+ let ref = WFRef n (self s)+ -- to reify the object being accessed+ -- if not reified, the serializer will write a null object+ let r= fromIDyn (versions s !! (state s - n)) `asTypeOf` typeofRef ret+ r `seq` return (s,ref))+ where+ typeofRef :: Workflow m (WFRef a) -> a+ typeofRef= undefined -- never will be executed+-- | Execute an step but return a reference to the result instead of the result itself+--+-- @stepWFRef exp= `step` exp >>= `getWFRef`@+stepWFRef :: ( Serialize a+ , Typeable a+ , MonadIO m)+ => m a -> Workflow m (WFRef a)+stepWFRef exp= step exp >> getWFRef++-- | Log a value and return a reference to it.+--+-- @newWFRef x= `step` $ return x >>= `getWFRef`@+newWFRef :: ( Serialize a+ , Typeable a+ , MonadIO m)+ => a -> Workflow m (WFRef a)+newWFRef x= step (return x) >> getWFRef++-- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad+readWFRef :: ( Serialize a+ , Typeable a)+ => WFRef a+ -> STM (Maybe a)+readWFRef (WFRef n ref)= do+ mr <- readDBRef ref+ case mr of+ Nothing -> return Nothing+ Just s -> do+ let elems= versions s+ l = state s -- L.length elems+ x = elems !! (l - n)+ return . Just $! fromIDyn x+++-- | Writes a new value en in the workflow reference, that is, in the workflow log.+-- Why would you use this?. Don do that!. modifiying the content of the workflow log would+-- change the excution flow when the workflow restarts. This metod is used internally in the package+-- the best way to communicate with a workflow is trough a persistent queue:+--+-- @worflow= exec1 "wf" do+-- r <- `stepWFRef` expr+-- `push` \"queue\" r+-- back <- `pop` \"queueback\"+-- ...+-- @++writeWFRef :: ( Serialize a+ , Typeable a)+ => WFRef a+ -> a+ -> STM ()+writeWFRef r@(WFRef n ref) x= do+ mr <- readDBRef ref+ case mr of+ Nothing -> error $ "writeWFRef: workflow does not exist: " ++ keyObjDBRef ref+ Just s -> do+ let elems= versions s+ l = state s -- L.length elems+ p = l - n+ (h,t)= L.splitAt p elems+ elems'= h ++ (toIDyn x:tail' t)+ tail' []= []+ tail' t= L.tail t++ writeDBRef ref s{ versions= elems'}+++++-- | Log a message in the workflow history. I can be printed out with 'printWFhistory'+-- The message is printed in the standard output too+logWF :: (Monad m, MonadIO m) => String -> Workflow m ()+logWF str=do+ str <- step . liftIO $ do+ time <- getClockTime >>= toCalendarTime >>= return . calendarTimeToString+ Prelude.putStrLn str+ return $ time ++ ": "++ str+ WF $ \s -> str `seq` return (s, ())++++--------- event handling--------------+++-- | Wait until a TCache object (with a certaing key) meet a certain condition (useful to check external actions )+-- NOTE if anoter process delete the object from te cache, then waitForData will no longuer work+-- inside the wokflow, it can be used by lifting it :+-- do+-- x <- step $ ..+-- y <- step $ waitForData ...+-- ..++waitForData :: (IResource a, Typeable a)+ => (a -> Bool) -- ^ The condition that the retrieved object must meet+ -> a -- ^ a partially defined object for which keyResource can be extracted+ -> IO a -- ^ return the retrieved object that meet the condition and has the given kwaitForData filter x= atomically $ waitForDataSTM filter x+waitForData f x = atomically $ waitForDataSTM f x++waitForDataSTM :: (IResource a, Typeable a)+ => (a -> Bool) -- ^ The condition that the retrieved object must meet+ -> a -- ^ a partially defined object for which keyResource can be extracted+ -> STM a -- ^ return the retrieved object that meet the condition and has the given key+waitForDataSTM filter x= do+ tv <- newDBRef x+ do+ mx <- readDBRef tv >>= \v -> return $ cast v+ case mx of+ Nothing -> retry+ Just x ->+ case filter x of+ False -> retry+ True -> return x++-- | observe the workflow log untiil a condition is met.+waitFor+ :: ( Indexable a, Serialize a, Serialize b, Typeable a+ , Indexable b, Typeable b)+ => (b -> Bool) -- ^ The condition that the retrieved object must meet+ -> String -- ^ The workflow name+ -> a -- ^ the INITIAL value used in the workflow to start it+ -> IO b -- ^ The first event that meet the condition+waitFor filter wfname x= atomically $ waitForSTM filter wfname x++waitForSTM+ :: ( Indexable a, Serialize a, Serialize b, Typeable a+ , Indexable b, Typeable b)+ => (b -> Bool) -- ^ The condition that the retrieved object must meet+ -> String -- ^ The workflow name+ -> a -- ^ The INITIAL value used in the workflow to start it+ -> STM b -- ^ The first event that meet the condition+waitForSTM filter wfname x= do+ let name= keyWF wfname x+ let tv= getDBRef . key $ stat0{wfName= name} -- `debug` "**waitFor***"++ mmx <- readDBRef tv+ case mmx of+ Nothing -> error ("waitForSTM: Workflow does not exist: "++ name)+ Just mx -> do+ let Stat{ versions= d:_}= mx+ case safeFromIDyn d of+ Nothing -> retry -- `debug` "waithFor retry Nothing"+ Just x ->+ case filter x of+ False -> retry -- `debug` "waitFor false filter retry"+ True -> return x -- `debug` "waitfor return"+++++-- | Start the timeout and return the flag to be monitored by 'waitUntilSTM'+-- This timeout is persistent. This means that the time start to count from the first call to getTimeoutFlag on+-- no matter if the workflow is restarted. The time that the worlkflow has been stopped count also.+-- the wait time can exceed the time between failures.+-- when timeout is 0 means no timeout.+getTimeoutFlag+ :: MonadIO m+ => Integer -- ^ wait time in secods. This timing is understood to start from the first time that the timeout was started. Sucessive restarts of the workflow will respect this timing+ -> Workflow m (TVar Bool) -- ^ the returned flag in the workflow monad+getTimeoutFlag 0 = WF $ \s -> liftIO $ newTVarIO False >>= \tv -> return (s, tv)+getTimeoutFlag t = do+ tnow<- step $ liftIO getTimeSeconds+ flag tnow t+ where+ flag tnow delta = WF(\s -> do+ (s', tv) <- case timeout s of+ Nothing -> do+ tv <- liftIO $ newTVarIO False+ return (s{timeout= Just tv}, tv)+ Just tv -> return (s, tv)+ liftIO $ do+ let t = tnow + delta+ atomically $ writeTVar tv False+ forkIO $ do waitUntil t ; atomically $ writeTVar tv True+ return (s', tv))++getTimeSeconds :: IO Integer+getTimeSeconds= do+ TOD n _ <- getClockTime+ return n++{- | Wait until a certain clock time has passed by monitoring its flag, in the STM monad.+ This permits to compose timeouts with locks waiting for data using `orElse`++ *example: wait for any respoinse from a Queue if no response is given in 5 minutes, it is returned True.++ @+ flag <- 'getTimeoutFlag' $ 5 * 60+ ap <- 'step' . atomically $ readSomewhere >>= return . Just `orElse` 'waitUntilSTM' flag >> return Nothing+ case ap of+ Nothing -> do 'logWF' "timeout" ...+ Just x -> do 'logWF' $ "received" ++ show x ...+ @+-}+waitUntilSTM :: TVar Bool -> STM()+waitUntilSTM tv = do+ b <- readTVar tv+ if b == False then retry else return ()++-- | Wait until a certain clock time has passed by monitoring its flag, in the IO monad.+-- See `waitUntilSTM`++waitUntil:: Integer -> IO()+waitUntil t= getTimeSeconds >>= \tnow -> wait (t-tnow)+++wait :: Integer -> IO()+wait delta= do+ let delay | delta < 0= 0+ | delta > (fromIntegral maxInt) = maxInt+ | otherwise = fromIntegral $ delta+ threadDelay $ delay * 1000000+ if delta <= 0 then return () else wait $ delta - (fromIntegral delay )+++
− Control/Workflow/Binary.hs
@@ -1,105 +0,0 @@-{-# LANGUAGE- OverlappingInstances- , UndecidableInstances- , ExistentialQuantification- , ScopedTypeVariables- , MultiParamTypeClasses- , FlexibleInstances- , FlexibleContexts- , TypeSynonymInstances- , DeriveDataTypeable- , CPP- #-}-{-# OPTIONS -IControl/Workflow #-}--{- |A workflow can be seen as a persistent thread.-The workflow monad writes a log that permit to restore the thread-at the interrupted point. `step` is the (partial) monad transformer for-the Workflow monad. A workflow is defined by its name and, optionally-by the key of the single parameter passed. The primitives for starting workflows-also restart the workflow when it has been in execution previously.--Thiis module uses Data.Binary serialization. Here the constraint @DynSerializer w r a@ is equivalent to-@Data.Binary a@--If you need to debug the workflow by reading the log or if you use largue structures that are subject of modifications along the workflow, as is the case-typically of multiuser workflows with documents, then use Text seriialization with "Control.Workflow.Text" instead---A small example that print the sequence of integers in te console-if you interrupt the progam, when restarted again, it will-start from the last printed number--@module Main where-import Control.Workflow.Binary-import Control.Concurrent(threadDelay)-import System.IO (hFlush,stdout)---mcount n= do `step` $ do- putStr (show n ++ \" \")- hFlush stdout- threadDelay 1000000- mcount (n+1)- return () -- to disambiguate the return type--main= `exec1` \"count\" $ mcount (0 :: Int)@---}--module Control.Workflow.Binary- (- Workflow -- a useful type name-, WorkflowList-, PMonadTrans (..)-, MonadCatchIO (..)--, throw-, Indexable(..)-, MonadIO(..)--- * Start/restart workflows-, start-, exec-, exec1d-, exec1-, wfExec-, restartWorkflows-, WFErrors(..)--- * Lifting to the Workflow monad-, step-, stepControl-, unsafeIOtoWF--- * References to workflow log values-, WFRef-, getWFRef-, newWFRef-, stepWFRef-, readWFRef-, writeWFRef--- * Workflow inspect-, getAll-, safeFromIDyn-, getWFKeys-, getWFHistory-, waitFor-, waitForSTM--- * Persistent timeouts-, waitUntilSTM-, getTimeoutFlag--- * Trace logging-, logWF--- * Termination of workflows-, killThreadWF-, killWF-, delWF-, killThreadWF1-, killWF1-, delWF1--- * Log writing policy-, syncWrite-, SyncMode(..)-)-where-import Control.Workflow.Binary.BinDefs--#include "Workflow.inc.hs"
− Control/Workflow/Binary/BinDefs.hs
@@ -1,104 +0,0 @@-{-# LANGUAGE-- MultiParamTypeClasses- , FlexibleInstances- , ScopedTypeVariables- , TypeSynonymInstances- #-}-module Control.Workflow.Binary.BinDefs where-import Control.Workflow.GenSerializer-import Control.Workflow.IDynamic-import Control.Workflow.Stat-import Data.TCache.DefaultPersistence(Indexable(..))-import Data.Binary-import Data.Binary.Put-import Data.Binary.Get-import System.IO.Unsafe-import Data.IORef-import Data.ByteString.Lazy.Char8 as B hiding (index)-import Data.Map as M-import Control.Concurrent(ThreadId,forkIO)-import Data.Typeable-import Data.TCache---instance Binary a => Serializer PutM Get a where- serial = put- deserial = get--instance RunSerializer PutM Get where- runSerial = runPut- runDeserial = runGet--instance Binary a => DynSerializer PutM Get a----instance TwoSerializer PutM Get PutM Get () ()--instance Binary IDynamic where- put (IDyn t) =- case unsafePerformIO $ readIORef t of- DRight x -> put . runSerial $ serial x- DLeft (s, _) -> put s-- get = do- s <- get- return $ IDyn . unsafePerformIO . newIORef $ DLeft (s, (undefined, pack ""))----instance Binary Stat where- put (Running map)= do- put (0 :: Word8)- put $ Prelude.map (\(k,(w,_)) -> (k,w)) $ M.toList map-- put (Stat wfName state index recover versions _) = do- put (1 :: Word8)- put wfName- put state- put index- put recover- put versions-- get = do- t <- get :: Get Word8- case t of- 0 -> do list <- get- return . Running . M.fromList $ Prelude.map(\(k,w)-> (k,(w,Nothing))) list- 1 -> do- wfName <- get- state <- get- index <- get- recover <- get- versions <- get- return $ Stat wfName state index recover versions Nothing--instance Binary ThreadId where- put _= put $ pack "th"- get = get >>= \(_ :: String) -> return $ unsafePerformIO . forkIO $ return ()---instance Binary (WFRef a) where- put (WFRef n ref)= do- put n- put $ keyObjDBRef ref-- get= do- n <- get- k <- get- return . WFRef n $ getDBRef k---instance Indexable String where- key= id--instance Indexable Int where- key= show--instance Indexable Integer where- key= show--instance Indexable Stat where- key s@Stat{wfName=name}= "Stat#" ++ name- key (Running _)= keyRunning- defPath= const $ defPath "" ++ "WorkflowState/bin/"-
− Control/Workflow/Binary/Patterns.hs
@@ -1,107 +0,0 @@-{-# LANGUAGE DeriveDataTypeable, FlexibleContexts, ScopedTypeVariables, CPP #-}-{-# OPTIONS -IControl/Workflow #-}--{- | This module contains monadic combinators that express-some workflow patterns.-see the docAprobal.hs example included in the package--This version uses Data.Binary serialization.-Here the constraint `DynSerializer w r a` is equivalent to `Data.Binary a`.--EXAMPLE:--This fragment below describes the approbal procedure of a document.-First the document reference is sent to a list of bosses trough a queue.-ithey return a boolean in a return queue ( askUser)-the booleans are summed up according with a monoid instance (sumUp)--if the resullt is false, the correctWF workflow is executed-If the result is True, the pipeline continues to the next stage (checkValidated)--the next stage is the same process with a new list of users (superbosses).-This time, there is a timeout of 7 days. the result of the users that voted is summed-up according with the same monoid instance--if the result is true the document is added to the persistent list of approbed documents-if the result is false, the document is added to the persistent list of rejectec documents (checlkValidated1)---@docApprobal :: Document -> Workflow IO ()-docApprobal doc = `getWFRef` \>>= docApprobal1---docApprobal1 rdoc=- return True \>>=- log \"requesting approbal from bosses\" \>>=- `sumUp` 0 (map (askUser doc rdoc) bosses) \>>=- checkValidated \>>=- log \"requesting approbal from superbosses or timeout\" \>>=- `sumUp` (7*60*60*24) (map(askUser doc rdoc) superbosses) \>>=- checkValidated1---askUser _ _ user False = return False-askUser doc rdoc user True = do- `step` $ `push` (quser user) rdoc- `logWF` (\"wait for any response from the user: \" ++ user)- `step` . `pop` $ qdocApprobal (title doc)--log txt x = `logWF` txt >> return x--checkValidated :: Bool -> `Workflow` IO Bool-checkValidated val =- case val of- False -> correctWF (title doc) rdoc >> return False- _ -> return True---checkValidated1 :: Bool -> Workflow IO ()-checkValidated1 val = step $ do- case val of- False -> `push` qrejected doc- _ -> `push` qapproved doc- mapM (\u ->deleteFromQueue (quser u) rdoc) superbosses@----}--module Control.Workflow.Binary.Patterns(--- * Low level combinators-split, merge, select,--- * High level conbinators-vote, sumUp, Select(..))- where-import Control.Concurrent.STM-import Data.Monoid-import Control.Concurrent.MonadIO-import qualified Control.Monad.CatchIO as CMC-import Control.Exception(SomeException)-import Control.Workflow.Binary-import Prelude hiding (catch)-import Control.Monad(when)-import Control.Exception.Extensible(Exception)-import Control.Workflow.GenSerializer---import Debug.Trace--import Control.Workflow.Binary-import Control.Workflow.Stat(keyWF)-import Data.Typeable--import Data.Binary--instance Binary Select where- put Select= put (1 :: Int)- put Discard= put (2 :: Int)- put FinishDiscard = put (3 :: Int)- put FinishSelect = put (4 :: Int)-- get= do- n <- get :: Get Int- case n of- 1 -> return Select- 2 -> return Discard- 3 -> return FinishDiscard- 4 -> return FinishSelect---#include "Patterns.inc.hs"
− Control/Workflow/GenSerializer.hs
@@ -1,70 +0,0 @@------------------------------------------------------------------------------------ Module : Control.Workflow.GenSerializer--- Copyright :--- License : BSD3------ Maintainer : agocorona@gmail.com--- Stability : experimental--- Portability :-----------------------------------------------------------------------------------{-# OPTIONS- -XMultiParamTypeClasses- -XFunctionalDependencies- -XFlexibleContexts- -XFlexibleInstances- -XUndecidableInstances- -XScopedTypeVariables- #-}--{- |- This module includes the definition of a generic (de)serializer. This is used as a class constraints- for the Workflow methods.-- Data.RefSerialize (defined in "Control.Workflow.Text.TextDefs") and Data.Binary ("Control.Workflow.Binary.BinDefs")- are particular instances of thiis generic serializer.--}--module Control.Workflow.GenSerializer where-import Data.ByteString.Lazy.Char8 as B-import Data.RefSerialize(Context)-import Data.Map as M-import Control.Concurrent-import System.IO.Unsafe--class (Monad writerm, Monad readerm)- => Serializer writerm readerm a | a -> writerm readerm where- serial :: a -> writerm ()- deserial :: readerm a-----class (DynSerializer w r a, DynSerializer w r b) => TwoSerializer w r a b---instance (DynSerializer w r a, DynSerializer w r b) => TwoSerializer w r a b--class (DynSerializer w r a, DynSerializer w r b,DynSerializer w r c) => ThreeSerializer w r a b c---instance (DynSerializer w r a, DynSerializer w r b, DynSerializer w r c) => ThreeSerializer w r a b c---class (Monad writerm- ,Monad readerm)- => RunSerializer writerm readerm- | writerm -> readerm- , readerm -> writerm- where- runSerial :: writerm () -> ByteString- runDeserial :: Serializer writerm readerm a => readerm a -> ByteString -> a--class (Serializer w r a, RunSerializer w r) => DynSerializer w r a | a -> w r where- serialM :: a -> w ByteString- serialM = return . runSerial . serial-- fromDynData :: ByteString ->(Context, ByteString) -> a- fromDynData s _= runDeserial deserial s
Control/Workflow/IDynamic.hs view
@@ -33,7 +33,7 @@ import Control.Concurrent.MVar import Data.IORef import Data.Map as M(empty)-import Control.Workflow.GenSerializer+import Data.RefSerialize import Data.HashTable as HT @@ -41,7 +41,7 @@ data IDynamic = IDyn (IORef IDynType) -data IDynType= forall a w r.(Typeable a, DynSerializer w r a)+data IDynType= forall a w r.(Typeable a, Serialize a) => DRight !a | DLeft !(ByteString ,(Context, ByteString)) @@ -162,12 +162,43 @@ fromString s= IDyn . unsafePerformIO . newIORef $ DLeft (s,(M.empty,"")) -}++++instance Serialize IDynamic where++ showp (IDyn t)=+ case unsafePerformIO $ readIORef t of+ DRight x -> do+ insertString $ pack dynPrefix+ showpx <- showps x+ showpText . fromIntegral $ B.length showpx+ insertString showpx+++++ DLeft (showpx,_) -> do -- error $ "IDynamic not reified :: "++ unpack showpx+ insertString $ pack dynPrefix+ showpText 0++ readp = do+ symbol dynPrefix+ n <- readpText+ s <- takep n+++ c <- getContext+ return . IDyn . unsafePerformIO . newIORef $ DLeft ( s, c)+ <?> "IDynamic"++ instance Show IDynamic where show (IDyn r) = let t= unsafePerformIO $ readIORef r in case t of- DRight x -> "IDyn " ++ ( unpack . runSerial $ serial x) ++ ")" + DRight x -> "IDyn " ++ ( unpack . runW $ showp x) ++ ")" DLeft (s, _) -> "IDyn " ++ unpack s @@ -177,7 +208,7 @@ toIDyn x= IDyn . unsafePerformIO . newIORef $ DRight x -fromIDyn :: (Typeable a , DynSerializer m n a)=> IDynamic -> a+fromIDyn :: (Typeable a , Serialize a)=> IDynamic -> a fromIDyn x=r where r = case safeFromIDyn x of Nothing -> error $ "fromIDyn: casting failure for data "@@ -186,7 +217,7 @@ Just v -> v -safeFromIDyn :: (Typeable a, DynSerializer m n a) => IDynamic -> Maybe a +safeFromIDyn :: (Typeable a, Serialize a) => IDynamic -> Maybe a safeFromIDyn (IDyn r)=unsafePerformIO $ do t<- readIORef r case t of@@ -195,7 +226,7 @@ DLeft (str, c) -> handle (\(e :: SomeException) -> return Nothing) $ -- !> ("safeFromIDyn : "++ show e)) $ do- let v= fromDynData str c+ let v= runRC c readp str writeIORef r $! DRight v -- !> ("***reified "++ unpack str) return $! Just v -- !> ("*** end reified " ++ unpack str)
+ Control/Workflow/Patterns.hs view
@@ -0,0 +1,271 @@+{-# LANGUAGE DeriveDataTypeable+ , ScopedTypeVariables+ , FlexibleContexts++ #-}+{-# OPTIONS -IControl/Workflow #-}++{- | This module contains monadic combinators that express some workflow patterns.+see the docAprobal.hs example included in the package++Here the constraint `DynSerializer w r a` is equivalent to `Data.Refserialize a`+This version permits optimal (de)serialization if you store in the queue different versions of largue structures, for+example, documents. You must define the right RefSerialize instance however.+See an example in docAprobal.hs incuded in the paclkage.+Alternatively you can use Data.Binary serlializatiion with Control.Workflow.Binary.Patterns++EXAMPLE:++This fragment below describes the approbal procedure of a document.+First the document reference is sent to a list of bosses trough a queue.+ithey return a boolean in a return queue ( askUser)+the booleans are summed up according with a monoid instance (sumUp)++if the resullt is false, the correctWF workflow is executed+If the result is True, the pipeline continues to the next stage (checkValidated)++the next stage is the same process with a new list of users (superbosses).+This time, there is a timeout of 7 days. the result of the users that voted is summed+up according with the same monoid instance++if the result is true the document is added to the persistent list of approbed documents+if the result is false, the document is added to the persistent list of rejectec documents (checlkValidated1)+++@docApprobal :: Document -> Workflow IO ()+docApprobal doc = `getWFRef` \>>= docApprobal1+++docApprobal1 rdoc=+ return True \>>=+ log \"requesting approbal from bosses\" \>>=+ `sumUp` 0 (map (askUser doc rdoc) bosses) \>>=+ checkValidated \>>=+ log \"requesting approbal from superbosses or timeout\" \>>=+ `sumUp` (7*60*60*24) (map(askUser doc rdoc) superbosses) \>>=+ checkValidated1+++askUser _ _ user False = return False+askUser doc rdoc user True = do+ `step` $ `push` (quser user) rdoc+ `logWF` ("wait for any response from the user: " ++ user)+ `step` . `pop` $ qdocApprobal (title doc)++log txt x = `logWF` txt >> return x++checkValidated :: Bool -> `Workflow` IO Bool+checkValidated val =+ case val of+ False -> correctWF (title doc) rdoc >> return False+ _ -> return True+++checkValidated1 :: Bool -> Workflow IO ()+checkValidated1 val = step $ do+ case val of+ False -> `push` qrejected doc+ _ -> `push` qapproved doc+ mapM (\u ->deleteFromQueue (quser u) rdoc) superbosses@++-}++module Control.Workflow.Patterns(+-- * Low level combinators+split, merge, select,+-- * High level conbinators+vote, sumUp, Select(..)+) where+import Control.Concurrent.STM+import Data.Monoid+import Control.Concurrent.MonadIO+import qualified Control.Monad.CatchIO as CMC+import Control.Workflow.Stat+import Control.Workflow+import Data.Typeable+import Prelude hiding (catch)+import Control.Monad(when)+import Control.Exception.Extensible (Exception)+import Data.RefSerialize+import Control.Workflow.Stat+import Debug.Trace+import Data.TCache++a !> b = trace b a++data ActionWF a= ActionWF (WFRef(Maybe a)) (WFRef (String, Bool))++-- | spawn a list of independent workflows (the first argument) with a seed value (the second argument).+-- Their results are reduced by `merge` or `select`+split :: ( Typeable b+ , Serialize b+ , HasFork io+ , CMC.MonadCatchIO io)+ => [a -> Workflow io b] -> a -> Workflow io [ActionWF b]+split actions a = mapM (\ac ->+ do+ mv <- newWFRef Nothing+ fork (ac a >>= step . liftIO . atomically . writeWFRef mv . Just)+ r <- getWFRef+ return $ ActionWF mv r)++ actions++++-- | wait for the results and apply the cond to produce a single output in the Workflow monad+merge :: ( MonadIO io+ , Typeable a+ , Typeable b+ , Serialize a, Serialize b)+ => ([a] -> io b) -> [ActionWF a] -> Workflow io b+merge cond actions= mapM (\(ActionWF mv _) -> readWFRef1 mv ) actions >>= step . cond++readWFRef1 :: ( MonadIO io+ , Serialize a+ , Typeable a)+ => WFRef (Maybe a) -> io a+readWFRef1 mv = liftIO . atomically $ do+ v <- readWFRef mv+ case v of+ Just(Just v) -> return v+ Just Nothing -> retry+ Nothing -> error $ "readWFRef1: workflow not found "++ show mv+++data Select+ = Select+ | Discard+ | FinishDiscard+ | FinishSelect+ deriving(Typeable, Read, Show)++instance Exception Select++-- | select the outputs of the workflows produced by `split` constrained within a timeout.+-- The check filter, can select , discard or finish the entire computation before+-- the timeout is reached. When the computation finalizes, it stop all+-- the pending workflows and return the list of selected outputs+-- the timeout is in seconds and is no limited to Int values, so it can last for years.+--+-- This is necessary for the modelization of real-life institutional cycles such are political elections+-- timeout of 0 means no timeout.+select ::+ ( Serialize a+ , Serialize [a]+ , Typeable a+ , HasFork io+ , CMC.MonadCatchIO io)+ => Integer+ -> (a -> io Select)+ -> [ActionWF a]+ -> Workflow io [a]+select timeout check actions= do+ res <- newMVar []+ flag <- getTimeoutFlag timeout+ parent <- myThreadId+ checks <- newEmptyMVar+ count <- newMVar 1+ let process = do+ let check' (ActionWF ac _) = do+ r <- readWFRef1 ac+ b <- check r+ case b of+ Discard -> return ()+ Select -> addRes r+ FinishDiscard -> do+ throwTo parent FinishDiscard+ FinishSelect -> do+ addRes r+ throwTo parent FinishDiscard++ n <- CMC.block $ do+ n <- takeMVar count+ putMVar count (n+1)+ return n++ if ( n == length actions)+ then throwTo parent FinishDiscard+ else return ()++ `CMC.catch` (\(e :: Select) -> throwTo parent e)++ do+ ws <- mapM ( fork . check') actions+ putMVar checks ws++ liftIO $ atomically $ do+ v <- readTVar flag -- wait fo timeout+ case v of+ False -> retry+ True -> return ()+ throw FinishDiscard+ where++ addRes r= CMC.block $ do+ l <- takeMVar res+ putMVar res $ r : l++ let killall = do+ mapM_ (\(ActionWF _ th) -> killWFP th) actions+ ws <- readMVar checks+ liftIO $ mapM_ killThread ws++ stepControl $ CMC.catch process -- (WF $ \s -> process >>= \ r -> return (s, r))+ (\(e :: Select)-> do+ readMVar res+ )+ `CMC.finally` killall++killWFP r= liftIO $ do+ s <- atomically $ do+ (s,_)<- readWFRef r >>= justify ("wfSelect " ++ show r)+ writeWFRef r (s, True)+ return s++ killWF s ()++justify str Nothing = error str+justify _ (Just x) = return x++-- | spawn a list of workflows and reduces the results according with the comp parameter within a given timeout+--+-- @+-- vote timeout actions comp x=+-- split actions x >>= select timeout (const $ return Select) >>= comp+-- @+vote+ :: ( Serialize b+ , Serialize [b]+ , Typeable b+ , HasFork io+ , CMC.MonadCatchIO io)+ => Integer+ -> [a -> Workflow io b]+ -> ([b] -> Workflow io c)+ -> a+ -> Workflow io c+vote timeout actions comp x=+ split actions x >>= select timeout (const $ return Select) >>= comp+++-- | sum the outputs of a list of workflows according with its monoid definition+--+-- @ sumUp timeout actions = vote timeout actions (return . mconcat) @+sumUp+ :: ( Serialize b+ , Serialize [b]+ , Typeable b+ , Monoid b+ , HasFork io+ , CMC.MonadCatchIO io)+ => Integer+ -> [a -> Workflow io b]+ -> a+ -> Workflow io b+sumUp timeout actions = vote timeout actions (return . mconcat)+++++
− Control/Workflow/Patterns.inc.hs
@@ -1,172 +0,0 @@--data ActionWF a= ActionWF (WFRef(Maybe a)) (WFRef (String, Bool))---- | spawn a list of independent workflows (the first argument) with a seed value (the second argument).--- Their results are reduced by `merge` or `select`-split :: ( Typeable b- , DynSerializer w r (Maybe b)- , HasFork io- , CMC.MonadCatchIO io)- => [a -> Workflow io b] -> a -> Workflow io [ActionWF b]-split actions a = mapM (\ac ->- do- mv <- newWFRef Nothing- fork (ac a >>= step . liftIO . atomically . writeWFRef mv . Just)- r <- getWFRef- return $ ActionWF mv r)-- actions------ | wait for the results and apply the cond to produce a single output in the Workflow monad-merge :: ( MonadIO io- , Typeable a- , Typeable b- , TwoSerializer w r (Maybe a) b)- => ([a] -> io b) -> [ActionWF a] -> Workflow io b-merge cond actions= mapM (\(ActionWF mv _) -> readWFRef1 mv ) actions >>= step . cond--readWFRef1 :: ( MonadIO io- , DynSerializer w r (Maybe a)- , Typeable a)- => WFRef (Maybe a) -> io a-readWFRef1 mv = liftIO . atomically $ do- v <- readWFRef mv- case v of- Just(Just v) -> return v- Just Nothing -> retry- Nothing -> error $ "readWFRef1: workflow not found "++ show mv---data Select- = Select- | Discard- | FinishDiscard- | FinishSelect- deriving(Typeable, Read, Show)--instance Exception Select---- | select the outputs of the workflows produced by `split` constrained within a timeout.--- The check filter, can select , discard or finish the entire computation before--- the timeout is reached. When the computation finalizes, it stop all--- the pending workflows and return the list of selected outputs--- the timeout is in seconds and is no limited to Int values, so it can last for years.------ This is necessary for the modelization of real-life institutional cycles such are political elections--- timeout of 0 means no timeout.-select ::- ( TwoSerializer w r (Maybe a) [a]- , Typeable a- , HasFork io- , CMC.MonadCatchIO io)- => Integer- -> (a -> io Select)- -> [ActionWF a]- -> Workflow io [a]-select timeout check actions= do- res <- newMVar []- flag <- getTimeoutFlag timeout- parent <- myThreadId- checks <- newEmptyMVar- count <- newMVar 1- let process = do- let check' (ActionWF ac _) = do- r <- readWFRef1 ac- b <- check r- case b of- Discard -> return ()- Select -> addRes r- FinishDiscard -> do- throwTo parent FinishDiscard- FinishSelect -> do- addRes r- throwTo parent FinishDiscard-- n <- CMC.block $ do- n <- takeMVar count- putMVar count (n+1)- return n-- if ( n == length actions)- then throwTo parent FinishDiscard- else return ()-- `CMC.catch` (\(e :: Select) -> throwTo parent e)-- do- ws <- mapM ( fork . check') actions- putMVar checks ws-- liftIO $ atomically $ do- v <- readTVar flag -- wait fo timeout- case v of- False -> retry- True -> return ()- throw FinishDiscard- where-- addRes r= CMC.block $ do- l <- takeMVar res- putMVar res $ r : l-- let killall = do- mapM_ (\(ActionWF _ th) -> killWFP th) actions- ws <- readMVar checks- liftIO $ mapM_ killThread ws-- stepControl $ CMC.catch process -- (WF $ \s -> process >>= \ r -> return (s, r))- (\(e :: Select)-> do- readMVar res- )- `CMC.finally` killall--killWFP r= liftIO $ do- s <- atomically $ do- (s,_)<- readWFRef r >>= justify ("wfSelect " ++ show r)- writeWFRef r (s, True)- return s-- killWF s ()--justify str Nothing = error str-justify _ (Just x) = return x---- | spawn a list of workflows and reduces the results according with the comp parameter within a given timeout------ @--- vote timeout actions comp x=--- split actions x >>= select timeout (const $ return Select) >>= comp--- @-vote- :: ( TwoSerializer w r (Maybe b) [b]- , Typeable b- , HasFork io- , CMC.MonadCatchIO io)- => Integer- -> [a -> Workflow io b]- -> ([b] -> Workflow io c)- -> a- -> Workflow io c-vote timeout actions comp x=- split actions x >>= select timeout (const $ return Select) >>= comp----- | sum the outputs of a list of workflows according with its monoid definition------ @ sumUp timeout actions = vote timeout actions (return . mconcat) @-sumUp- :: ( TwoSerializer w r (Maybe b) [b]- , Typeable b- , Monoid b- , HasFork io- , CMC.MonadCatchIO io)- => Integer- -> [a -> Workflow io b]- -> a- -> Workflow io b-sumUp timeout actions = vote timeout actions (return . mconcat)---
Control/Workflow/Stat.hs view
@@ -18,12 +18,13 @@ import Control.Concurrent(ThreadId) import Control.Concurrent.STM(TVar, newTVarIO) import Data.IORef-import Control.Workflow.GenSerializer+import Data.RefSerialize import Control.Workflow.IDynamic import Control.Monad(replicateM) import Data.TCache.DefaultPersistence import Data.ByteString.Lazy.Char8 hiding (index) import Control.Workflow.IDynamic+import Control.Concurrent(forkIO) data WF s m l = WF { st :: s -> m (s,l) }@@ -46,12 +47,56 @@ , index :: Int , recover:: Bool , versions ::[IDynamic]- , timeout :: Maybe (TVar Bool)}+ , timeout :: Maybe (TVar Bool)+ , self :: DBRef Stat+ } deriving (Typeable) stat0 = Stat{ wfName="", state=0, index=0, recover=False, versions = []- , timeout= Nothing}+ , timeout= Nothing, self=getDBRef ""} ++statPrefix= "Stat#"+instance Indexable Stat where+ key s@Stat{wfName=name}= statPrefix ++ name+ key (Running _)= keyRunning+ defPath _= (defPath (1::Int)) ++ "Workflow/"+++instance Serialize Stat where+ showp (Running map)= do+ insertString $ pack "Running"+ showp $ Prelude.map (\(k,(w,_)) -> (k,w)) $ M.toList map+++ showp stat@( Stat wfName state index recover versions _ _ )=do+ insertString $ pack "Stat"+ showpText wfName+ showpText state+ showpText index+ showpText recover+ showp versions+++ readp = choice [rStat, rWorkflows] where+ rStat= do+ symbol "Stat"+ wfName <- stringLiteral+ state <- integer >>= return . fromIntegral+ index <- integer >>= return . fromIntegral+ recover <- bool+ versions <- readp+ let self= getDBRef $ key stat0{wfName= wfName}+ return $ Stat wfName state index recover versions Nothing self+ <?> "Stat"++ rWorkflows= do+ symbol "Running"+ list <- readp+ return $ Running $ M.fromList $ Prelude.map(\(k,w)-> (k,(w,Nothing))) list+ <?> "RunningWoorkflows"++ -- return the unique name of a workflow with a parameter (executed with exec or start) keyWF :: Indexable a => String -> a -> String keyWF wn x= wn ++ "#" ++ key x@@ -67,12 +112,74 @@ -instance (Serializer w r a, RunSerializer w r) => Serializable a where- serialize = runSerial . serial+instance Serialize a => Serializable a where+ serialize = runW . showp - deserialize = runDeserial deserial+ deserialize = runR readp keyRunning= "Running"+++++instance Serialize ThreadId where+ showp th= insertString . pack $ show th+ readp = (readp :: ST ByteString) >> (return . unsafePerformIO . forkIO $ return ())+++newtype Pretty = Pretty Stat++instance Show Pretty where+ show= unpack . runW . sp+ where+ sp (Pretty (Stat wfName state index recover versions _ _))= do+ insertString $ pack "Workflow name= "+ showp wfName+ insertString $ pack "\n"+ showElem $ Prelude.reverse $ (Prelude.zip ( Prelude.reverse [1..] ) versions )+++ showElem :: [(Int,IDynamic)] -> ST ()+ showElem [] = insertChar '\n'+ showElem ((n, dyn):es) = do+ showp $ pack "Step "+ showp n+ showp $ pack ": "+ showp dyn+ insertChar '\n'+ showElem es+++instance Indexable String where+ key= id++instance Indexable Int where+ key= show++instance Indexable Integer where+ key= show+++wFRefStr = "WFRef"++instance Serialize (WFRef a) where+ showp (WFRef n ref)= do+ insertString $ pack wFRefStr+ showp n+ showp $ keyObjDBRef ref++ readp= do+ symbol wFRefStr+ n <- readp+ k <- readp+ return . WFRef n $ getDBRef k++-- | print the state changes along the workflow, that is, all the intermediate results+printHistory :: Stat -> IO ()+printHistory stat= do+ Prelude.putStrLn . show $ Pretty stat+ Prelude.putStrLn "-----------------------------------"+
− Control/Workflow/Text.hs
@@ -1,118 +0,0 @@-{-# LANGUAGE OverlappingInstances- , UndecidableInstances- , ExistentialQuantification- , ScopedTypeVariables- , MultiParamTypeClasses- , FlexibleInstances- , FlexibleContexts- , TypeSynonymInstances- , DeriveDataTypeable- , CPP- #-}-{-# OPTIONS -IControl/Workflow #-}--{- | A workflow can be seen as a persistent thread.-The workflow monad writes a log that permit to restore the thread-at the interrupted point. `step` is the (partial) monad transformer for-the Workflow monad. A workflow is defined by its name and, optionally-by the key of the single parameter passed. The primitives for starting workflows-also restart the workflow when it has been in execution previously.--This is the main module that uses the `RefSerialize` paclkage for serialization. Here the constraint @DynSerializer w r a@ is equivalent to-@Data.RefSerialize a@--For workflows that uses big structures, for example, documents-use this module in combination with the RefSerialize package to define the (de)serialization instances-The log size will be reduced. printWFHistory` method will print the structure changes-in each step.--If instead of RefSerialize, you define read and show instances, there will- be no reduction. but still the log will be readable for debugging purposes.--for workflows that does not care about this, use the binary alternative: "Control.Workflow.Binary"--A small example that print the sequence of integers in te console-if you interrupt the progam, when restarted again, it will-start from the last printed number--@module Main where-import Control.Workflow.Text-import Control.Concurrent(threadDelay)-import System.IO (hFlush,stdout)---mcount n= do `step` $ do- putStr (show n ++ \" \")- hFlush stdout- threadDelay 1000000- mcount (n+1)- return () -- to disambiguate the return type--main= `exec1` \"count\" $ mcount (0 :: Int)@---}--module Control.Workflow.Text-(- Workflow -- a useful type name-, WorkflowList-, PMonadTrans (..)-, MonadCatchIO (..)-, throw-, Indexable(..)--- * Start/restart workflows-, start-, exec-, exec1d-, exec1-, wfExec-, startWF-, restartWorkflows-, WFErrors(..)--- * Lifting to the Workflow monad-, step-, stepControl-, unsafeIOtoWF--- * References to intermediate values in the workflow log-, WFRef-, getWFRef-, newWFRef-, stepWFRef-, readWFRef-, writeWFRef--- * Workflow inspect-, waitWFActive-, getAll-, safeFromIDyn-, getWFKeys-, getWFHistory-, waitFor-, waitForSTM--- * Persistent timeouts-, waitUntilSTM-, getTimeoutFlag--- * Trace logging-, logWF--- * Termination of workflows-, clearRunningFlag-, killThreadWF-, killWF-, delWF-, killThreadWF1-, killWF1-, delWF1-, delWFHistory-, delWFHistory1--- * Log writing policy-, syncWrite-, SyncMode(..)--- * Print log history-, printHistory-)-where--import Control.Workflow.Text.TextDefs--#include "Workflow.inc.hs"--
− Control/Workflow/Text/Patterns.hs
@@ -1,97 +0,0 @@-{-# LANGUAGE DeriveDataTypeable- , FlexibleContexts- , ScopedTypeVariables- , CPP- #-}-{-# OPTIONS -IControl/Workflow #-}--{- | This module contains monadic combinators that express some workflow patterns.-see the docAprobal.hs example included in the package--Here the constraint `DynSerializer w r a` is equivalent to `Data.Refserialize a`-This version permits optimal (de)serialization if you store in the queue different versions of largue structures, for-example, documents. You must define the right RefSerialize instance however.-See an example in docAprobal.hs incuded in the paclkage.-Alternatively you can use Data.Binary serlializatiion with Control.Workflow.Binary.Patterns--EXAMPLE:--This fragment below describes the approbal procedure of a document.-First the document reference is sent to a list of bosses trough a queue.-ithey return a boolean in a return queue ( askUser)-the booleans are summed up according with a monoid instance (sumUp)--if the resullt is false, the correctWF workflow is executed-If the result is True, the pipeline continues to the next stage (checkValidated)--the next stage is the same process with a new list of users (superbosses).-This time, there is a timeout of 7 days. the result of the users that voted is summed-up according with the same monoid instance--if the result is true the document is added to the persistent list of approbed documents-if the result is false, the document is added to the persistent list of rejectec documents (checlkValidated1)---@docApprobal :: Document -> Workflow IO ()-docApprobal doc = `getWFRef` \>>= docApprobal1---docApprobal1 rdoc=- return True \>>=- log \"requesting approbal from bosses\" \>>=- `sumUp` 0 (map (askUser doc rdoc) bosses) \>>=- checkValidated \>>=- log \"requesting approbal from superbosses or timeout\" \>>=- `sumUp` (7*60*60*24) (map(askUser doc rdoc) superbosses) \>>=- checkValidated1---askUser _ _ user False = return False-askUser doc rdoc user True = do- `step` $ `push` (quser user) rdoc- `logWF` ("wait for any response from the user: " ++ user)- `step` . `pop` $ qdocApprobal (title doc)--log txt x = `logWF` txt >> return x--checkValidated :: Bool -> `Workflow` IO Bool-checkValidated val =- case val of- False -> correctWF (title doc) rdoc >> return False- _ -> return True---checkValidated1 :: Bool -> Workflow IO ()-checkValidated1 val = step $ do- case val of- False -> `push` qrejected doc- _ -> `push` qapproved doc- mapM (\u ->deleteFromQueue (quser u) rdoc) superbosses@---}--module Control.Workflow.Text.Patterns(--- * Low level combinators-split, merge, select,--- * High level conbinators-vote, sumUp, Select(..)-) where-import Control.Concurrent.STM-import Data.Monoid-import Control.Concurrent.MonadIO-import qualified Control.Monad.CatchIO as CMC-import Control.Workflow.Stat-import Control.Workflow.Text-import Data.Typeable-import Prelude hiding (catch)-import Control.Monad(when)-import Control.Exception.Extensible (Exception)-import Control.Workflow.GenSerializer-import Control.Workflow.Stat-import Debug.Trace-import Data.TCache--a !> b = trace b a--#include "Patterns.inc.hs"-
− Control/Workflow/Text/TextDefs.hs
@@ -1,161 +0,0 @@-{-# LANGUAGE-- MultiParamTypeClasses- , FlexibleInstances- , UndecidableInstances- , TypeSynonymInstances--- #-}-module Control.Workflow.Text.TextDefs where-import Control.Workflow.IDynamic-import Control.Workflow.GenSerializer-import Data.RefSerialize-import System.IO.Unsafe-import Data.TCache.DefaultPersistence(Indexable(..))-import Data.IORef-import Unsafe.Coerce-import Data.ByteString.Lazy.Char8 as B hiding (index)-import Control.Workflow.Stat-import Data.Map as M-import Control.Concurrent-import Data.TCache---instance Serialize a => Serializer ST ST a where- serial = showp- deserial = readp--instance RunSerializer ST ST where- runSerial = runW- runDeserial = runR--instance Serialize a => DynSerializer ST ST a where- serialM = showps- fromDynData s c= runRC c readp s---instance Serialize IDynamic where-- showp (IDyn t)=- case unsafePerformIO $ readIORef t of- DRight x -> do- insertString $ pack dynPrefix- showpx <- unsafeCoerce $ serialM x- showpText . fromIntegral $ B.length showpx- insertString showpx--- DLeft (showpx,_) -> -- error $ "IDynamic not reified :: "++ unpack showpx- do- insertString $ pack dynPrefix- showpText 0---- readp = do- symbol dynPrefix- n <- readpText- s <- takep n- c <- getContext- return . IDyn . unsafePerformIO . newIORef $ DLeft ( s, c)- <?> "IDynamic"---instance Serialize Stat where- showp (Running map)= do- insertString $ pack "Running"- showp $ Prelude.map (\(k,(w,_)) -> (k,w)) $ M.toList map--- showp stat@( Stat wfName state index recover versions _ )=do- insertString $ pack "Stat"- showpText wfName- showpText state- showpText index- showpText recover- showp versions--- readp = choice [rStat, rWorkflows] where- rStat= do- symbol "Stat"- wfName <- stringLiteral- state <- integer >>= return . fromIntegral- index <- integer >>= return . fromIntegral- recover <- bool- versions <- readp- return $ Stat wfName state index recover versions Nothing- <?> "Stat"-- rWorkflows= do- symbol "Running"- list <- readp- return $ Running $ M.fromList $ Prelude.map(\(k,w)-> (k,(w,Nothing))) list- <?> "RunningWoorkflows"---instance Serialize ThreadId where- showp th= insertString . pack $ show th- readp = (readp :: ST ByteString) >> (return . unsafePerformIO . forkIO $ return ())---newtype Pretty = Pretty Stat--instance Show Pretty where- show= unpack . runW . sp- where- sp (Pretty (Stat wfName state index recover versions _ ))= do- insertString $ pack "Workflow name= "- showp wfName- insertString $ pack "\n"- showElem $ Prelude.reverse $ (Prelude.zip ( Prelude.reverse [1..] ) versions )--- showElem :: [(Int,IDynamic)] -> ST ()- showElem [] = insertChar '\n'- showElem ((n, dyn):es) = do- showp $ pack "Step "- showp n- showp $ pack ": "- showp dyn- insertChar '\n'- showElem es---instance Indexable String where- key= id--instance Indexable Int where- key= show--instance Indexable Integer where- key= show--statPrefix= "Stat#"-instance Indexable Stat where- key s@Stat{wfName=name}= statPrefix ++ name- key (Running _)= keyRunning- defPath _= (defPath "") ++ "WorkflowState/Text/"--wFRefStr = "WFRef"--instance Serialize (WFRef a) where- showp (WFRef n ref)= do- insertString $ pack wFRefStr- showp n- showp $ keyObjDBRef ref-- readp= do- symbol wFRefStr- n <- readp- k <- readp- return . WFRef n $ getDBRef k---- | print the state changes along the workflow, that is, all the intermediate results-printHistory :: Stat -> IO ()-printHistory stat= do- Prelude.putStrLn . show $ Pretty stat- Prelude.putStrLn "-----------------------------------"--
− Control/Workflow/Workflow.inc.hs
@@ -1,930 +0,0 @@---import Prelude hiding (catch)-import System.IO.Unsafe-import Control.Monad(when,liftM)-import qualified Control.Exception as CE (Exception,AsyncException(ThreadKilled), SomeException, throwIO, handle,finally,catch,block,unblock)-import Control.Concurrent (forkIO,threadDelay, ThreadId, myThreadId, killThread)-import Control.Concurrent.STM-import GHC.Conc(unsafeIOToSTM)-import GHC.Base (maxInt)---import Data.ByteString.Lazy.Char8 as B hiding (index)-import Data.ByteString.Lazy as BL(putStrLn)-import Data.List as L-import Data.Typeable-import System.Time-import Control.Monad.Trans-import Control.Concurrent.MonadIO(HasFork(..),MVar,newMVar,takeMVar,putMVar)---import System.IO(hPutStrLn, stderr)-import Data.List(elemIndex)-import Data.Maybe(fromJust, isNothing, isJust, mapMaybe)-import Data.IORef-import System.IO.Unsafe(unsafePerformIO)-import Data.Map as M(Map,fromList,elems, insert, delete, lookup,toList, fromList,keys)-import qualified Control.Monad.CatchIO as CMC-import qualified Control.Exception.Extensible as E--import Data.TCache-import Data.TCache.DefaultPersistence-import Control.Workflow.GenSerializer-import Control.Workflow.IDynamic-import Unsafe.Coerce-import Control.Workflow.Stat------import Debug.Trace---a !> b= trace b a---type Workflow m = WF Stat m -- not so scary--type WorkflowList m a b= [(String, a -> Workflow m b) ]---instance Monad m => Monad (WF s m) 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'))----instance (Monad m,Functor m) => Functor (Workflow m ) where- fmap f (WF g)= WF (\s -> do- (s1, x) <- g s- return (s1, f x))--tvRunningWfs = getDBRef $ keyRunning :: DBRef Stat------ | executes a computation inside of the workflow monad whatever the monad encapsulated in the workflow.--- Warning: this computation is executed whenever--- the workflow restarts, no matter if it has been already executed previously. This is useful for intializations or debugging.--- To avoid re-execution when restarting use: @'step' $ unsafeIOtoWF...@------ To perform IO actions in a workflow that encapsulates an IO monad, use step over the IO action directly:------ @ 'step' $ action @------ instead of------ @ 'step' $ unsafeIOtoWF $ action @-unsafeIOtoWF :: (Monad m) => IO a -> Workflow m a-unsafeIOtoWF x= let y= unsafePerformIO ( x >>= return) in y `seq` return y---{- | PMonadTrans permits |to define a partial monad transformer. They are not defined for all kinds of data-but the ones that have instances of certain classes.That is because in the lift instance code there are some-hidden use of these classes. This also may permit an accurate control of effects.-An instance of MonadTrans is an instance of PMonadTrans--}-class PMonadTrans t m a where- plift :: Monad m => m a -> t m a------ | plift= step-instance (Monad m- , MonadIO m- , DynSerializer w r a- , Typeable a)- => PMonadTrans (WF Stat) m a- where- plift = step---- | An instance of MonadTrans is an instance of PMonadTrans-instance (MonadTrans t, Monad m) => PMonadTrans t m a where- plift= Control.Monad.Trans.lift--instance Monad m => MonadIO (WF Stat m) where- liftIO=unsafeIOtoWF---{- | adapted from MonadCatchIO-mtl. Workflow need to express serializable constraints about the returned values,-so the usual class definitions for lifting IO functions are not suitable.--}--class MonadCatchIO m a where- -- | Generalized version of 'E.catch'- catch :: E.Exception e => m a -> (e -> m a) -> m a-- -- | Generalized version of 'E.block'- block :: m a -> m a-- -- | Generalized version of 'E.unblock'- unblock :: m a -> m a------ | Generalized version of 'E.throwIO'-throw :: (MonadIO m, E.Exception e) => e -> m a-throw = liftIO . E.throwIO--{---{---- | Generalized version of 'E.try'-try :: (MonadCatchIO m a, E.Exception e) => m a -> m (Either e a)---- | Generalized version of 'E.tryJust'-tryJust :: (MonadCatchIO m a, E.Exception e)- => (e -> Maybe b) -> m a -> m (Either b a)---}--- | Generalized version of 'E.Handler'-data Handler m a = forall e . E.Exception e => Handler (e -> m a)---{--instance (MonadCatchIO m a, Error e) => MonadCatchIO (ErrorT e m) a where- m `catch` f = mapErrorT (\m' -> m' `catch` (\e -> runErrorT $ f e)) m- block = mapErrorT block- unblock = mapErrorT unblock--}----try a = catch (a >>= \ v -> return (Right v)) (\e -> return (Left e))--tryJust p a = do- r <- try a- case r of- Right v -> return (Right v)- Left e -> case p e of- Nothing -> throw e `asTypeOf` (return $ Left undefined)- Just b -> return (Left b)---- | Generalized version of 'E.bracket'-bracket :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c) => m a -> (a -> m b) -> (a -> m c) -> m c-bracket before after thing =- block (do a <- before- r <- unblock (thing a) `onException` after a- _void $ after a- return r)---- | A variant of 'bracket' where the return value from the first computation--- is not required.-bracket_ :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c)- => m a -- ^ computation to run first (\"acquire resource\")- -> m b -- ^ computation to run last (\"release resource\")- -> m c -- ^ computation to run in-between- -> m c -- returns the value from the in-between computation-bracket_ before after thing =- block $ do _void before- r <- unblock thing `onException` after- _void after- return r---- | A specialised variant of 'bracket' with just a computation to run--- afterward.-finally :: (Monad m, MonadIO m, MonadCatchIO m a)- => m a -- ^ computation to run first- -> m b -- ^ computation to run afterward (even if an exception was- -- raised)- -> m a -- returns the value from the first computation-thing `finally` after =- block $ do r <- unblock thing `onException` after- _void after- return r-{---- | Like 'bracket', but only performs the final action if there was an--- exception raised by the in-between computation.-bracketOnError :: (Monad m, MonadIO m, MonadCatchIO m a, MonadCatchIO m c)- => m a -- ^ computation to run first (\"acqexeuire resource\")- -> (a -> m b)-- ^ computation to run last (\"release resource\")- -> (a -> m c)-- ^ computation to run in-between- -> m c -- returns the value from the in-between- -- computation-bracketOnError before after thing =- block $ do a <- before- unblock (thing a) `onException` after a--}--- | Generalized version of 'E.onException'-onException :: (MonadIO m, MonadCatchIO m a) => m a -> m b -> m a-onException a onEx = a `catch` (\e -> onEx >> throw (e:: E.SomeException))--_void :: Monad m => m a -> m ()-_void a = a >> return ()----}-----instance (TwoSerializer w r () a- , Typeable a,MonadIO m, CMC.MonadCatchIO m)- => MonadCatchIO (WF Stat m) a where- catch exp exc = do- expwf <- step $ getTempName- excwf <- step $ getTempName- step $ do- ex <- CMC.catch (exec1d expwf exp >>= return . Right- ) $ \e-> return $ Left e-- case ex of- Right r -> return r -- All right- Left e ->exec1d excwf (exc e)- -- An exception occured in the main workflow- -- the exception workflow is executed----- block exp=WF $ \s -> CMC.block (st exp $ s)-- unblock exp= WF $ \s -> CMC.unblock (st exp $ s)----instance (HasFork io- , CMC.MonadCatchIO io)- => HasFork (WF Stat io) where- fork f = do- (str, finished) <- step $ getTempName >>= \n -> return(n, False)- r <- getWFRef- WF (\s ->- do th <- if finished- then fork $ return ()- else fork $ do- exec1 str f- liftIO $ do atomically $ writeWFRef r (str, True)- syncIt- return(s,th))------- | start or restart an anonymous workflow inside another workflow--- its state is deleted when finished and the result is stored in--- the parent's WF state.-wfExec- :: (Indexable a, TwoSerializer w r () a, Typeable a- , CMC.MonadCatchIO m, MonadIO m)- => Workflow m a -> Workflow m a-wfExec f= do- str <- step $ getTempName- step $ exec1 str f---- | a version of exec1 that deletes its state after complete execution or thread killed-exec1d :: (TwoSerializer w r () b, Typeable b- ,CMC.MonadCatchIO m)- => String -> (Workflow m b) -> m b-exec1d str f= do- r <- exec1 str f- delit- return r- `CMC.catch` (\e@CE.ThreadKilled -> delit >> throw e)-- where- delit= do- delWF str ()- liftIO syncIt -- !> str------ | a version of exec with no seed parameter.-exec1 :: ( TwoSerializer w r () a, Typeable a- , Monad m, MonadIO m, CMC.MonadCatchIO m)- => String -> Workflow m a -> m a--exec1 str f= exec str (const f) ()------- | start or continue a workflow with exception handling--- | the workflow flags are updated even in case of exception--- | `WFerrors` are raised as exceptions-exec :: ( Indexable a, TwoSerializer w r a b, Typeable a- , Typeable b- , Monad m, MonadIO m, CMC.MonadCatchIO m)- => String -> (a -> Workflow m b) -> a -> m b-exec str f x =- (do- v <- getState str f x- case v of- Right (name, f, stat) -> do- r <- runWF name (f x) stat- return r- Left err -> CMC.throw err)- `CMC.catch`- (\(e :: CE.SomeException) -> liftIO $ do- let name= keyWF str x- clearRunningFlag name --`debug` ("exception"++ show e)- syncIt- CMC.throw e )-----mv :: MVar Int-mv= unsafePerformIO $ newMVar 0--getTempName :: MonadIO m => m String-getTempName= liftIO $ do- seq <- takeMVar mv- putMVar mv (seq + 1)- TOD t _ <- getClockTime- return $ "anon"++ show t ++ show seq-------instance Indexable () where- key= show---- | lifts a monadic computation to the WF monad, and provides transparent state loging and resuming of computation-step :: ( Monad m- , MonadIO m- , DynSerializer w r a- , Typeable a)- => m a- -> Workflow m a-step= stepControl1 False---- | permits modification of the workflow state by the procedure being lifted--- if the boolean value is True. This is used internally for control purposes-stepControl :: ( Monad m- , MonadIO m- , DynSerializer w r a- , Typeable a)- => m a- -> Workflow m a-stepControl= stepControl1 True--stepControl1 :: ( Monad m- , MonadIO m- , DynSerializer w r a- , Typeable a)- => Bool -> m a- -> Workflow m a-stepControl1 isControl mx= WF(\s'' -> do- let stat= state s''- let ind= index s''- if recover s'' && ind < stat- then return (s''{index=ind +1 }, fromIDyn $ versions s'' !! (stat - ind-1) )- else do- x' <- mx- let sref = getDBRef $ key s''- s'<- liftIO . atomically $ do- s <- if isControl- then readDBRef sref >>= unjustify ("step: readDBRef: not found:" ++ keyObjDBRef sref)- else return s''- let versionss= versions s- let dynx= toIDyn x'- let ver= dynx: versionss- let s'= s{ recover= False, versions = ver, state= state s+1}-- writeDBRef sref s'- return s'- liftIO syncIt- return (s', x') )--unjustify str Nothing = error str-unjustify _ (Just x) = return x------- | start or continue a workflow with no exception handling.--- | the programmer has to handle inconsistencies in the workflow state--- | using `killWF` or `delWF` in case of exception.-start- :: ( Monad m- , MonadIO m- , Indexable a- , TwoSerializer w r a b- , Typeable a- , Typeable b)- => String -- ^ name thar identifies the workflow.- -> (a -> Workflow m b) -- ^ workflow to execute- -> a -- ^ initial value (ever use the initial value for restarting the workflow)- -> m (Either WFErrors b) -- ^ result of the computation-start namewf f1 v = do- ei <- getState namewf f1 v- case ei of- Right (name, f, stat) ->- runWF name (f v) stat >>= return . Right-- Left error -> return $ Left error---- | return conditions from the invocation of start/restart primitives-data WFErrors = NotFound | AlreadyRunning | Timeout | forall e.CE.Exception e => Exception e deriving Typeable--instance Show WFErrors where- show NotFound= "Not Found"- show AlreadyRunning= "Already Running"- show Timeout= "Timeout"- show (Exception e)= "Exception: "++ show e--instance CE.Exception WFErrors----tvRunningWfs = unsafePerformIO . refDBRefIO $ Running (M.fromList [] :: Map String (String, (Maybe ThreadId)))--{--lookup for any workflow for the entry value v-if namewf is found and is running, return arlready running- if is not runing, restart it-else start anew.--}---getState :: (Monad m, MonadIO m, Indexable a, DynSerializer w r a, Typeable a)- => String -> x -> a- -> m (Either WFErrors (String, x, Stat))-getState namewf f v= liftIO . atomically $ getStateSTM- where- getStateSTM = do- mrunning <- readDBRef tvRunningWfs- case mrunning of- Nothing -> do- writeDBRef tvRunningWfs (Running $ fromList [])- getStateSTM- Just(Running map) -> do- let key= keyWF namewf v- let stat1= stat0{wfName= key,versions=[toIDyn v]}- case M.lookup key map of- Nothing -> do -- no workflow started for this object- mythread <- unsafeIOToSTM $ myThreadId- writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map- withSTMResources ([] :: [Stat]) $- \_-> resources{toAdd=[stat1],toReturn= Right (key, f, stat1) }-- Just (wf, started) -> -- a workflow has been initiated for this object- if isJust started- then return $ Left AlreadyRunning -- `debug` "already running"- else do -- has been started but not running now- mythread <- unsafeIOToSTM $ myThreadId- writeDBRef tvRunningWfs . Running $ M.insert key (namewf,Just mythread) map- withSTMResources[stat1] $- \mst->- let stat'= case mst of- [Nothing] -> error $ "Workflow not found: "++ key- [Just s] -> s{index=0,recover= True}- in resources{toAdd=[stat'],toReturn = Right (key, f, stat') }--syncIt= do- (sync,_) <- atomically $ readTVar tvSyncWrite- when (sync ==Synchronous) syncCache--runWF :: (Monad m,MonadIO m- , DynSerializer w r b, Typeable b)- => String -> Workflow m b -> Stat -> m b-runWF n f s= do- sync <- liftIO $! do- (sync,_) <- atomically $ readTVar tvSyncWrite- when (sync ==Synchronous) syncCache- return sync- (s', v') <- st f $ s- liftIO $! do- clearFromRunningList n- when (sync ==Synchronous) syncCache- return v'- where-- -- eliminate the thread from the list of running workflows but leave the state- clearFromRunningList n = atomically $ do- Just(Running map) <- readDBRef tvRunningWfs- writeDBRef tvRunningWfs . Running $ M.delete n map -- `debug` "clearFromRunningList"---- | start or continue a workflow from a list of workflows in the IO monad with exception handling. The excepton is returned as a Left value-startWF- :: ( MonadIO m- , TwoSerializer w r a b- , Typeable a- , Indexable a- , Typeable b)- => String -- ^ name of workflow in the workflow list- -> a -- ^ initial value (ever use the initial value even to restart the workflow)- -> WorkflowList m a b -- ^ function to execute- -> m (Either WFErrors b) -- ^ result of the computation-startWF namewf v wfs=- case Prelude.lookup namewf wfs of- Nothing -> return $ Left NotFound- Just f -> start namewf f v---- | re-start the non finished workflows in the list, for all the initial values that they may have been called-restartWorkflows- :: (TwoSerializer w r a b, Typeable a- , Indexable b, Typeable b)- => WorkflowList IO a b -- the list of workflows that implement the module- -> IO () -- Only workflows in the IO monad can be restarted with restartWorkflows-restartWorkflows map = do- mw <- liftIO $ getResource ((Running undefined ) ) -- :: IO (Maybe(Stat a))- case mw of- Nothing -> return ()- Just (Running all) -> mapM_ start . mapMaybe filter . toList $ all- where- filter (a, (b,Nothing)) = Just (b, a)- filter _ = Nothing-- start (key, kv)= do-- --let key1= key ++ "#" ++ kv- let mf= Prelude.lookup key map- case mf of- Nothing -> return ()- Just f -> do- let st0= stat0{wfName = kv}- mst <- liftIO $ getResource st0- case mst of- Nothing -> error $ "restartWorkflows: workflow not found "++ keyResource st0- Just st-> do- liftIO . forkIO $ runWF key (f (fromIDyn . Prelude.last $ versions st )) st{index=0,recover=True} >> return ()- return ()---- | choose between text and binary persistence for the workflow state--- text persistence is used for---- *(1)debugging purposes---- * (2)when step returns largue structures that share common contents between steps,--- for example, when a workflow edit and ammend a document among many users---- * (3) When tracking the modifications made in the object trough `getWFHistory` or--- `printWFHistory`------ |--- The execution log is cached in memory using the package `TCache`. This procedure defines the polcy for writing the cache into permanent storage.------ For fast workflows, or when TCache` is used also for other purposes , `Asynchronous` is the best option------ `Asynchronous` mode invokes `clearSyncCache`. For more complex use of the syncronization--- please use this `clearSyncCache`.------ When interruptions are controlled, use `SyncManual` mode and include a call to `syncCache` in the finalizaton code--syncWrite:: (Monad m, MonadIO m) => SyncMode -> m ()-syncWrite mode= do- (_,thread) <- liftIO . atomically $ readTVar tvSyncWrite- when (isJust thread ) $ liftIO . killThread . fromJust $ thread- case mode of- Synchronous -> modeWrite- SyncManual -> modeWrite- Asyncronous time maxsize -> do- th <- liftIO $ clearSyncCacheProc time defaultCheck maxsize >> return()- liftIO . atomically $ writeTVar tvSyncWrite (mode,Just th)- where- modeWrite=- liftIO . atomically $ writeTVar tvSyncWrite (mode, Nothing)----- | return all the steps of the workflow log. The values are dynamic------ to get all the steps with result of type Int:--- @all <- `getAll`--- let lfacts = mapMaybe `safeFromIDyn` all :: [Int]@-getAll :: Monad m => Workflow m [IDynamic]-getAll= WF(\s -> return (s, versions s))----- | return the list of object keys that are running for a workflow-getWFKeys :: String -> IO [String]-getWFKeys wfname= do- mwfs <- atomically $ readDBRef tvRunningWfs- case mwfs of- Nothing -> return []- Just (Running wfs) -> return $ Prelude.filter (L.isPrefixOf wfname) $ M.keys wfs---- | return the current state of the computation, in the IO monad-getWFHistory :: (Indexable a, DynSerializer w r a) => String -> a -> IO (Maybe Stat)-getWFHistory wfname x= getResource stat0{wfName= keyWF wfname x}---delWFHistory name1 x= do- let name= keyWF name1 x- delWFHistory1 name--delWFHistory1 name =- atomically . withSTMResources [] $ const resources{ toDelete= [stat0{wfName= name}] }---waitWFActive wf= do- r <- threadWF wf- case r of -- wait for change in the wofkflow state- Just (_, Nothing) -> retry- _ -> return ()- where- threadWF wf= do- Just(Running map) <- readDBRef tvRunningWfs - return $ M.lookup wf map ----- | kill the executing thread if not killed, but not its state.--- `exec` `start` or `restartWorkflows` will continue the workflow-killThreadWF :: ( Indexable a- , DynSerializer w r a-- , Typeable a- , MonadIO m)- => String -> a -> m()-killThreadWF wfname x= do- let name= keyWF wfname x- killThreadWF1 name---- | a version of `KillThreadWF` for workflows started wit no parameter by `exec1`-killThreadWF1 :: MonadIO m => String -> m()-killThreadWF1 name= killThreadWFm name >> return ()--killThreadWFm name= do- (map,f) <- clearRunningFlag name- case f of- Just th -> liftIO $ killThread th- Nothing -> return()- return map------ | kill the process (if running) and drop it from the list of--- restart-able workflows. Its state history remains , so it can be inspected with--- `getWfHistory` `printWFHistory` and so on-killWF :: (Indexable a,MonadIO m) => String -> a -> m ()-killWF name1 x= do- let name= keyWF name1 x- killWF1 name---- | a version of `KillWF` for workflows started wit no parameter by `exec1`-killWF1 :: MonadIO m => String -> m ()-killWF1 name = do- map <- killThreadWFm name- liftIO . atomically . writeDBRef tvRunningWfs . Running $ M.delete name map- return ()---- | delete the WF from the running list and delete the workflow state from persistent storage.--- Use it to perform cleanup if the process has been killed.-delWF :: ( Indexable a- , MonadIO m- , Typeable a)- => String -> a -> m()-delWF name1 x= do- let name= keyWF name1 x- delWF1 name----- | a version of `delWF` for workflows started wit no parameter by `exec1`-delWF1 :: MonadIO m=> String -> m()-delWF1 name= liftIO $ do- mrun <- atomically $ readDBRef tvRunningWfs- case mrun of- Nothing -> return()- Just (Running map) -> do- atomically . writeDBRef tvRunningWfs . Running $! M.delete name map- delWFHistory1 name- syncIt----clearRunningFlag name= liftIO $ atomically $ do- mrun <- readDBRef tvRunningWfs- case mrun of- Nothing -> error $ "clearRunningFLag non existing workflows" ++ name- Just(Running map) -> do- case M.lookup name map of- Just(_, Nothing) -> return (map,Nothing)- Just(v, Just th) -> do- writeDBRef tvRunningWfs . Running $ M.insert name (v, Nothing) map- return (map,Just th)- Nothing ->- return (map, Nothing)---- | Return the reference to the last logged result , usually, the last result stored by `step`.--- wiorkflow references can be accessed outside of the workflow--- . They also can be (de)serialized.------ WARNING getWFRef can produce casting errors when the type demanded--- do not match the serialized data. Instead, `newDBRef` and `stepWFRef` are type safe at runtuime.-getWFRef :: ( DynSerializer w r a- , Typeable a- , MonadIO m)- => Workflow m (WFRef a)-getWFRef = WF (\s -> do- let n= if recover s then index s else state s- let ref = WFRef n (getDBRef $ key s)--- return (s,ref))---- | Execute an step but return a reference to the result instead of the result itself------ @stepWFRef exp= `step` exp >>= `getWFRef`@-stepWFRef :: ( DynSerializer w r a- , Typeable a- , MonadIO m)- => m a -> Workflow m (WFRef a)-stepWFRef exp= step exp >> getWFRef---- | Log a value and return a reference to it.------ @newWFRef x= `step` $ return x >>= `getWFRef`@-newWFRef :: ( DynSerializer w r a- , Typeable a- , MonadIO m)- => a -> Workflow m (WFRef a)-newWFRef x= step (return x) >> getWFRef---- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad-readWFRef :: ( DynSerializer w r a- , Typeable a)- => WFRef a- -> STM (Maybe a)-readWFRef (WFRef n ref)= do- mr <- readDBRef ref- case mr of- Nothing -> return Nothing- Just s -> do- let elems= versions s- l = state s -- L.length elems- x = elems !! (l - n)- return . Just $! fromIDyn x----- | Writes a new value en in the workflow reference, that is, in the workflow log.--- Why would you use this?. Don do that!. modifiying the content of the workflow log would--- change the excution flow when the workflow restarts. This metod is used internally in the package--- the best way to communicate with a workflow is trough a persistent queue:------ @worflow= exec1 "wf" do--- r <- `stepWFRef` expr--- `push` \"queue\" r--- back <- `pop` \"queueback\"--- ...--- @--writeWFRef :: ( DynSerializer w r a- , Typeable a)- => WFRef a- -> a- -> STM ()-writeWFRef r@(WFRef n ref) x= do- mr <- readDBRef ref- case mr of- Nothing -> error $ "writeWFRef: workflow does not exist: " ++ keyObjDBRef ref- Just s -> do- let elems= versions s- l = state s -- L.length elems- p = l - n- (h,t)= L.splitAt p elems- elems'= h ++ (toIDyn x:tail' t)- tail' []= []- tail' t= L.tail t-- writeDBRef ref s{ versions= elems'}------- | Log a message in the workflow history. I can be printed out with 'printWFhistory'--- The message is printed in the standard output too-logWF :: (Monad m, MonadIO m) => String -> Workflow m ()-logWF str=do- str <- step . liftIO $ do- time <- getClockTime >>= toCalendarTime >>= return . calendarTimeToString- Prelude.putStrLn str- return $ time ++ ": "++ str- WF $ \s -> str `seq` return (s, ())------------- event handling------------------- | Wait until a TCache object (with a certaing key) meet a certain condition (useful to check external actions )--- NOTE if anoter process delete the object from te cache, then waitForData will no longuer work--- inside the wokflow, it can be used by lifting it :--- do--- x <- step $ ..--- y <- step $ waitForData ...--- ..--waitForData :: (IResource a, Typeable a)- => (a -> Bool) -- ^ The condition that the retrieved object must meet- -> a -- ^ a partially defined object for which keyResource can be extracted- -> IO a -- ^ return the retrieved object that meet the condition and has the given kwaitForData filter x= atomically $ waitForDataSTM filter x-waitForData f x = atomically $ waitForDataSTM f x--waitForDataSTM :: (IResource a, Typeable a)- => (a -> Bool) -- ^ The condition that the retrieved object must meet- -> a -- ^ a partially defined object for which keyResource can be extracted- -> STM a -- ^ return the retrieved object that meet the condition and has the given key-waitForDataSTM filter x= do- tv <- newDBRef x- do- mx <- readDBRef tv >>= \v -> return $ cast v- case mx of- Nothing -> retry- Just x ->- case filter x of- False -> retry- True -> return x---- | observe the workflow log untiil a condition is met.-waitFor- :: ( Indexable a, TwoSerializer w r a b, Typeable a- , Indexable b, Typeable b)- => (b -> Bool) -- ^ The condition that the retrieved object must meet- -> String -- ^ The workflow name- -> a -- ^ the INITIAL value used in the workflow to start it- -> IO b -- ^ The first event that meet the condition-waitFor filter wfname x= atomically $ waitForSTM filter wfname x--waitForSTM- :: ( Indexable a, TwoSerializer w r a b, Typeable a- , Indexable b, Typeable b)- => (b -> Bool) -- ^ The condition that the retrieved object must meet- -> String -- ^ The workflow name- -> a -- ^ The INITIAL value used in the workflow to start it- -> STM b -- ^ The first event that meet the condition-waitForSTM filter wfname x= do- let name= keyWF wfname x- let tv= getDBRef . key $ stat0{wfName= name} -- `debug` "**waitFor***"-- mmx <- readDBRef tv- case mmx of- Nothing -> error ("waitForSTM: Workflow does not exist: "++ name)- Just mx -> do- let Stat{ versions= d:_}= mx- case safeFromIDyn d of- Nothing -> retry -- `debug` "waithFor retry Nothing"- Just x ->- case filter x of- False -> retry -- `debug` "waitFor false filter retry"- True -> return x -- `debug` "waitfor return"------- | Start the timeout and return the flag to be monitored by 'waitUntilSTM'--- This timeout is persistent. This means that the time start to count from the first call to getTimeoutFlag on--- no matter if the workflow is restarted. The time that the worlkflow has been stopped count also.--- the wait time can exceed the time between failures.--- when timeout is 0 means no timeout.-getTimeoutFlag- :: MonadIO m- => Integer -- ^ wait time in secods. This timing is understood to start from the first time that the timeout was started. Sucessive restarts of the workflow will respect this timing- -> Workflow m (TVar Bool) -- ^ the returned flag in the workflow monad-getTimeoutFlag 0 = WF $ \s -> liftIO $ newTVarIO False >>= \tv -> return (s, tv)-getTimeoutFlag t = do- tnow<- step $ liftIO getTimeSeconds- flag tnow t- where- flag tnow delta = WF(\s -> do- (s', tv) <- case timeout s of- Nothing -> do- tv <- liftIO $ newTVarIO False- return (s{timeout= Just tv}, tv)- Just tv -> return (s, tv)- liftIO $ do- let t = tnow + delta- atomically $ writeTVar tv False- forkIO $ do waitUntil t ; atomically $ writeTVar tv True- return (s', tv))--getTimeSeconds :: IO Integer-getTimeSeconds= do- TOD n _ <- getClockTime- return n--{- | Wait until a certain clock time has passed by monitoring its flag, in the STM monad.- This permits to compose timeouts with locks waiting for data using `orElse`-- *example: wait for any respoinse from a Queue if no response is given in 5 minutes, it is returned True.-- @- flag <- 'getTimeoutFlag' $ 5 * 60- ap <- 'step' . atomically $ readSomewhere >>= return . Just `orElse` 'waitUntilSTM' flag >> return Nothing- case ap of- Nothing -> do 'logWF' "timeout" ...- Just x -> do 'logWF' $ "received" ++ show x ...- @--}-waitUntilSTM :: TVar Bool -> STM()-waitUntilSTM tv = do- b <- readTVar tv- if b == False then retry else return ()---- | Wait until a certain clock time has passed by monitoring its flag, in the IO monad.--- See `waitUntilSTM`--waitUntil:: Integer -> IO()-waitUntil t= getTimeSeconds >>= \tnow -> wait (t-tnow)---wait :: Integer -> IO()-wait delta= do- let delay | delta < 0= 0- | delta > (fromIntegral maxInt) = maxInt- | otherwise = fromIntegral $ delta- threadDelay $ delay * 1000000- if delta <= 0 then return () else wait $ delta - (fromIntegral delay )-
+ Data/Persistent/Queue.hs view
@@ -0,0 +1,240 @@+{-# OPTIONS -XDeriveDataTypeable+ -XTypeSynonymInstances+ -XMultiParamTypeClasses+ -XExistentialQuantification+ -XOverloadedStrings+ -XFlexibleInstances+ -XUndecidableInstances+ -XFunctionalDependencies+++ -IControl/Workflow++ #-}+{-# OPTIONS -IData/Persistent/Queue #-}+{- |+This module implements a persistent, transactional collection with Queue interface as well as+ indexed access by key++use this version if you store in the queue different versions of largue structures, for+example, documents and define a "Data.RefSerialize" instance. If not, use "Data.Persistent.Queue.Binary" Instead.++Here @QueueConstraints w r a@ means @Data.RefSerlialize.Serialize a@++-}++module Data.Persistent.Queue(+RefQueue(..), getQRef,+pop,popSTM,pick, flush, flushSTM,+pickAll, pickAllSTM, push,pushSTM,+pickElem, pickElemSTM, readAll, readAllSTM,+deleteElem, deleteElemSTM,+unreadSTM,isEmpty,isEmptySTM+) where+import Data.Typeable+import Control.Concurrent.STM(STM,atomically, retry)+import Control.Monad(when)+import Data.TCache.DefaultPersistence++import Data.TCache+import System.IO.Unsafe+import Data.RefSerialize+import Data.ByteString.Lazy.Char8+import Data.RefSerialize++import Debug.Trace++a !> b= trace b a+++instance Indexable (Queue a) where+ key (Queue k _ _)= queuePrefix ++ k+++++data Queue a= Queue {name :: String, imp :: [a], out :: [a]} deriving (Typeable)++++instance Serialize a => Serialize (Queue a) where+ showp (Queue n i o)= showp n >> showp i >> showp o+ readp = do+ n <- readp+ i <- readp+ o <- readp+ return $ Queue n i o+++++queuePrefix= "Queue#"+lenQPrefix= Prelude.length queuePrefix++++instance Serialize a => Serializable (Queue a ) where+ serialize = runW . showp+ deserialize = runR readp++-- | a queue reference+type RefQueue a= DBRef (Queue a)++unreadSTM :: (Typeable a, Serialize a) => RefQueue a -> a -> STM ()+unreadSTM queue x= do+ r <- readQRef queue+ writeDBRef queue $ doit r+ where+ doit (Queue n imp out) = Queue n imp ( x : out)+++-- | check if the queue is empty+isEmpty :: (Typeable a, Serialize a) => RefQueue a -> IO Bool+isEmpty = atomically . isEmptySTM++isEmptySTM :: (Typeable a, Serialize a) => RefQueue a -> STM Bool+isEmptySTM queue= do+ r <- readDBRef queue+ return $ case r of+ Nothing -> True+ Just (Queue _ [] []) -> True+ _ -> False++++-- | get the reference to new or existing queue trough its name+getQRef :: (Typeable a, Serialize a) => String -> RefQueue a+getQRef n = getDBRef . key $ Queue n undefined undefined+++-- | empty the queue (factually, it is deleted)+flush :: (Typeable a, Serialize a) => RefQueue a -> IO ()+flush = atomically . flushSTM++-- | version in the STM monad+flushSTM :: (Typeable a, Serialize a) => RefQueue a -> STM ()+flushSTM tv= delDBRef tv++-- | read the first element in the queue and delete it (pop)+pop+ :: (Typeable a, Serialize a) => RefQueue a -- ^ Queue name+ -> IO a -- ^ the returned elems+pop tv = atomically $ popSTM tv+++readQRef :: (Typeable a, Serialize a) => RefQueue a -> STM(Queue a)+readQRef tv= do+ mdx <- readDBRef tv+ case mdx of+ Nothing -> do+ let q= Queue ( Prelude.drop lenQPrefix $ keyObjDBRef tv) [] []+ writeDBRef tv q+ return q+ Just dx ->+ return dx++-- | version in the STM monad+popSTM :: (Typeable a, Serialize a) => RefQueue a+ -> STM a+popSTM tv=do+ dx <- readQRef tv+ doit dx++ where+ --doit :: (Typeable a, Serialize a, Serializer m n [a]) => Queue a -> STM a+ doit (Queue n [x] [])= do+ writeDBRef tv $ (Queue n [] [])+ return x+ doit (Queue _ [] []) = retry+ doit (Queue n imp []) = doit (Queue n [] $ Prelude.reverse imp)+ doit (Queue n imp list ) = do+ writeDBRef tv (Queue n imp (Prelude.tail list ))+ return $ Prelude.head list++-- | read the first element in the queue but it does not delete it+pick+ :: (Typeable a, Serialize a) => RefQueue a -- ^ Queue name+ -> IO a -- ^ the returned elems+pick tv = atomically $ do+ dx <- readQRef tv+ doit dx+ where+ doit (Queue _ [x] [])= return x+ doit (Queue _ [] []) = retry+ doit (Queue n imp []) = doit (Queue n [] $ Prelude.reverse imp)+ doit (Queue n imp list ) = return $ Prelude.head list++-- | push an element in the queue+push :: (Typeable a, Serialize a) => RefQueue a -> a -> IO ()+push tv v = atomically $ pushSTM tv v++-- | version in the STM monad+pushSTM :: (Typeable a, Serialize a) => RefQueue a -> a -> STM ()+pushSTM tv v=+ readQRef tv >>= \ ((Queue n imp out)) -> writeDBRef tv $ Queue n (v : imp) out++-- | return the list of all elements in the queue. The queue remains unchanged+pickAll :: (Typeable a, Serialize a) => RefQueue a -> IO [a]+pickAll= atomically . pickAllSTM++-- | version in the STM monad+pickAllSTM :: (Typeable a, Serialize a) => RefQueue a -> STM [a]+pickAllSTM tv= do+ (Queue name imp out) <- readQRef tv+ return $ out ++ Prelude.reverse imp++-- | return the first element in the queue that has the given key+pickElem ::(Indexable a,Typeable a, Serialize a) => RefQueue a -> String -> IO(Maybe a)+pickElem tv key= atomically $ pickElemSTM tv key++-- | version in the STM monad+pickElemSTM :: (Indexable a,Typeable a, Serialize a)+ => RefQueue a -> String -> STM(Maybe a)+pickElemSTM tv key1= do+ Queue name imp out <- readQRef tv+ let xs= out ++ Prelude.reverse imp+ when (not $ Prelude.null imp) $ writeDBRef tv $ Queue name [] xs+ case Prelude.filter (\x-> key x == key1) xs of+ [] -> return $ Nothing+ (x:_) -> return $ Just x++-- | update the first element of the queue with a new element with the same key+updateElem :: (Indexable a,Typeable a, Serialize a)+ => RefQueue a -> a -> IO()+updateElem tv x = atomically $ updateElemSTM tv x++-- | version in the STM monad+updateElemSTM :: (Indexable a,Typeable a, Serialize a)+ => RefQueue a -> a -> STM()+updateElemSTM tv v= do+ Queue name imp out <- readQRef tv+ let xs= out ++ Prelude.reverse imp+ let xs'= Prelude.map (\x -> if key x == n then v else x) xs+ writeDBRef tv $ Queue name [] xs'+ where+ n= key v++-- | return the list of all elements in the queue and empty it+readAll :: (Typeable a, Serialize a) => RefQueue a -> IO [a]+readAll= atomically . readAllSTM++-- | a version in the STM monad+readAllSTM :: (Typeable a, Serialize a) => RefQueue a -> STM [a]+readAllSTM tv= do+ Queue name imp out <- readQRef tv+ writeDBRef tv $ Queue name [] []+ return $ out ++ Prelude.reverse imp++-- | delete all the elements of the queue that has the key of the parameter passed+deleteElem :: (Indexable a,Typeable a, Serialize a) => RefQueue a-> a -> IO ()+deleteElem tv x= atomically $ deleteElemSTM tv x++-- | verison in the STM monad+deleteElemSTM :: (Typeable a, Serialize a,Indexable a) => RefQueue a-> a -> STM ()+deleteElemSTM tv x= do+ Queue name imp out <- readQRef tv+ let xs= out ++ Prelude.reverse imp+ writeDBRef tv $ Queue name [] $ Prelude.filter (\x-> key x /= k) xs+ where+ k=key x+
− Data/Persistent/Queue/Binary.hs
@@ -1,58 +0,0 @@-{-# OPTIONS -XDeriveDataTypeable- -XTypeSynonymInstances- -XMultiParamTypeClasses- -XExistentialQuantification- -XOverloadedStrings- -XFlexibleInstances- -XUndecidableInstances- -XFunctionalDependencies- -XFlexibleContexts- -XIncoherentInstances- -IControl/Workflow- -XCPP #-}-{-# OPTIONS -IData/Persistent/Queue #-}-{- |-This module implements a persistent, transactional collection with Queue interface as well as indexed access by key-This module uses `Data.Binary` for serialization.--Here @QueueConstraints w r a@ means @Data.Binary.Binary a@--For optimal (de)serialization if you store in the queue different versions of largue structures , for-example, documents you better use "Data.RefSerialize" and "Data.Persistent.Queue.Text" Instead.---}-module Data.Persistent.Queue.Binary(-RefQueue(..), getQRef,-pop,popSTM,pick,Data.Persistent.Queue.Binary.flush, flushSTM,-pickAll, pickAllSTM, push,pushSTM,-pickElem, pickElemSTM, readAll, readAllSTM,-deleteElem, deleteElemSTM,-unreadSTM,Data.Persistent.Queue.Binary.isEmpty,isEmptySTM-) where-import Data.Typeable-import Control.Concurrent.STM(STM,atomically, retry)-import Control.Monad(when)-import Data.TCache.DefaultPersistence--import Data.TCache-import System.IO.Unsafe-import Data.IORef--import Data.ByteString.Lazy.Char8--import Control.Workflow.GenSerializer- -import Data.Binary-import Data.Binary.Put-import Data.Binary.Get--instance SerialiserString PutM Get where- serialString= put- deserialString= get---instance Indexable (Queue a) where- key (Queue k _ _)= queuePrefix ++ k- defPath _= "WorkflowState/Binary/"--#include "Queue.inc.hs"
− Data/Persistent/Queue/Queue.inc.hs
@@ -1,205 +0,0 @@---data Queue a= Queue {name :: String, imp :: [a], out :: [a]} deriving (Typeable)--class (Monad writerm- ,Monad readerm)- => SerialiserString writerm readerm- | writerm -> readerm- , readerm -> writerm- where- serialString :: String -> writerm ()- deserialString :: readerm String--class ( Serializer w r [a]- , SerialiserString w r- , RunSerializer w r)- => QueueConstraints w r a--instance ( Serializer w r [a]- , SerialiserString w r- , RunSerializer w r)- => QueueConstraints w r a--instance (Serializer w r [a]- , SerialiserString w r)- => Serializer w r (Queue a) where- serial (Queue n i o)= serialString n >> serial i >> serial o- deserial= do- n <- deserialString- i <- deserial- o <- deserial- return $ Queue n i o-----queuePrefix= "Queue#"-lenQPrefix= Prelude.length queuePrefix----instance QueueConstraints w r a => Serializable (Queue a ) where- serialize = runSerial . serial- deserialize = runDeserial deserial---- | a queue reference-type RefQueue a= DBRef (Queue a)--unreadSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> a -> STM ()-unreadSTM queue x= do- r <- readQRef queue- writeDBRef queue $ doit r- where- doit (Queue n imp out) = Queue n imp ( x : out)----- | check if the queue is empty-isEmpty :: (Typeable a, QueueConstraints w r a) => RefQueue a -> IO Bool-isEmpty = atomically . isEmptySTM--isEmptySTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> STM Bool-isEmptySTM queue= do- r <- readDBRef queue- return $ case r of- Nothing -> True- Just (Queue _ [] []) -> True- _ -> False------ | get the reference to new or existing queue trough its name-getQRef :: (Typeable a, QueueConstraints w r a) => String -> RefQueue a-getQRef n = getDBRef . key $ Queue n undefined undefined----- | empty the queue (factually, it is deleted)-flush :: (Typeable a, QueueConstraints w r a) => RefQueue a -> IO ()-flush = atomically . flushSTM---- | version in the STM monad-flushSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> STM ()-flushSTM tv= delDBRef tv---- | read the first element in the queue and delete it (pop)-pop- :: (Typeable a, QueueConstraints w r a) => RefQueue a -- ^ Queue name- -> IO a -- ^ the returned elems-pop tv = atomically $ popSTM tv---readQRef :: (Typeable a, QueueConstraints w r a) => RefQueue a -> STM(Queue a)-readQRef tv= do- mdx <- readDBRef tv- case mdx of- Nothing -> do- let q= Queue ( Prelude.drop lenQPrefix $ keyObjDBRef tv) [] []- writeDBRef tv q- return q- Just dx ->- return dx---- | version in the STM monad-popSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a- -> STM a-popSTM tv=do- dx <- readQRef tv- doit dx-- where- --doit :: (Typeable a, Serializer m n [a]) => Queue a -> STM a- doit (Queue n [x] [])= do- writeDBRef tv $ (Queue n [] [])- return x- doit (Queue _ [] []) = retry- doit (Queue n imp []) = doit (Queue n [] $ Prelude.reverse imp)- doit (Queue n imp list ) = do- writeDBRef tv (Queue n imp (Prelude.tail list ))- return $ Prelude.head list---- | read the first element in the queue but it does not delete it-pick- :: (Typeable a, QueueConstraints w r a) => RefQueue a -- ^ Queue name- -> IO a -- ^ the returned elems-pick tv = atomically $ do- dx <- readQRef tv- doit dx- where- doit (Queue _ [x] [])= return x- doit (Queue _ [] []) = retry- doit (Queue n imp []) = doit (Queue n [] $ Prelude.reverse imp)- doit (Queue n imp list ) = return $ Prelude.head list---- | push an element in the queue-push :: (Typeable a, QueueConstraints w r a) => RefQueue a -> a -> IO ()-push tv v = atomically $ pushSTM tv v---- | version in the STM monad-pushSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> a -> STM ()-pushSTM tv v=- readQRef tv >>= \ ((Queue n imp out)) -> writeDBRef tv $ Queue n (v : imp) out---- | return the list of all elements in the queue. The queue remains unchanged-pickAll :: (Typeable a, QueueConstraints w r a) => RefQueue a -> IO [a]-pickAll= atomically . pickAllSTM---- | version in the STM monad-pickAllSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> STM [a]-pickAllSTM tv= do- (Queue name imp out) <- readQRef tv- return $ out ++ Prelude.reverse imp---- | return the first element in the queue that has the given key-pickElem ::(Indexable a,Typeable a, QueueConstraints w r a) => RefQueue a -> String -> IO(Maybe a)-pickElem tv key= atomically $ pickElemSTM tv key---- | version in the STM monad-pickElemSTM :: (Indexable a,Typeable a, QueueConstraints w r a)- => RefQueue a -> String -> STM(Maybe a)-pickElemSTM tv key1= do- Queue name imp out <- readQRef tv- let xs= out ++ Prelude.reverse imp- when (not $ Prelude.null imp) $ writeDBRef tv $ Queue name [] xs- case Prelude.filter (\x-> key x == key1) xs of- [] -> return $ Nothing- (x:_) -> return $ Just x---- | update the first element of the queue with a new element with the same key-updateElem :: (Indexable a,Typeable a, QueueConstraints w r a)- => RefQueue a -> a -> IO()-updateElem tv x = atomically $ updateElemSTM tv x---- | version in the STM monad-updateElemSTM :: (Indexable a,Typeable a, QueueConstraints w r a)- => RefQueue a -> a -> STM()-updateElemSTM tv v= do- Queue name imp out <- readQRef tv- let xs= out ++ Prelude.reverse imp- let xs'= Prelude.map (\x -> if key x == n then v else x) xs- writeDBRef tv $ Queue name [] xs'- where- n= key v---- | return the list of all elements in the queue and empty it-readAll :: (Typeable a, QueueConstraints w r a) => RefQueue a -> IO [a]-readAll= atomically . readAllSTM---- | a version in the STM monad-readAllSTM :: (Typeable a, QueueConstraints w r a) => RefQueue a -> STM [a]-readAllSTM tv= do- Queue name imp out <- readQRef tv- writeDBRef tv $ Queue name [] []- return $ out ++ Prelude.reverse imp---- | delete all the elements of the queue that has the key of the parameter passed-deleteElem :: (Indexable a,Typeable a, QueueConstraints w r a) => RefQueue a-> a -> IO ()-deleteElem tv x= atomically $ deleteElemSTM tv x---- | verison in the STM monad-deleteElemSTM :: (Typeable a,Indexable a,QueueConstraints w r a) => RefQueue a-> a -> STM ()-deleteElemSTM tv x= do- Queue name imp out <- readQRef tv- let xs= out ++ Prelude.reverse imp- writeDBRef tv $ Queue name [] $ Prelude.filter (\x-> key x /= k) xs- where- k=key x
− Data/Persistent/Queue/Text.hs
@@ -1,60 +0,0 @@-{-# OPTIONS -XDeriveDataTypeable- -XTypeSynonymInstances- -XMultiParamTypeClasses- -XExistentialQuantification- -XOverloadedStrings- -XFlexibleInstances- -XUndecidableInstances- -XFunctionalDependencies- -XFlexibleContexts- -XIncoherentInstances- -IControl/Workflow- -XCPP- #-}-{-# OPTIONS -IData/Persistent/Queue #-}-{- |-This module implements a persistent, transactional collection with Queue interface as well as- indexed access by key--use this version if you store in the queue different versions of largue structures, for-example, documents and define a "Data.RefSerialize" instance. If not, use "Data.Persistent.Queue.Binary" Instead.--Here @QueueConstraints w r a@ means @Data.RefSerlialize.Serialize a@---}--module Data.Persistent.Queue.Text(-RefQueue(..), getQRef,-pop,popSTM,pick, flush, flushSTM,-pickAll, pickAllSTM, push,pushSTM,-pickElem, pickElemSTM, readAll, readAllSTM,-deleteElem, deleteElemSTM,-unreadSTM,isEmpty,isEmptySTM-) where-import Data.Typeable-import Control.Concurrent.STM(STM,atomically, retry)-import Control.Monad(when)-import Data.TCache.DefaultPersistence--import Data.TCache-import System.IO.Unsafe-import Data.RefSerialize-import Data.ByteString.Lazy.Char8-import Control.Workflow.GenSerializer--import Debug.Trace--a !> b= trace b a--instance SerialiserString ST ST where- serialString = showp- deserialString = readp--instance Indexable (Queue a) where- key (Queue k _ _)= queuePrefix ++ k- defPath _= "WorkflowState/Text/"---#include "Queue.inc.hs"--
Demos/Fact.hs view
@@ -6,9 +6,10 @@ -- enter any alphanumeric character for aborting and then re-start. module Main where-import Control.Workflow.Binary+import Control.Workflow import Data.Typeable import Data.Binary+import Data.RefSerialize import Data.Maybe @@ -28,6 +29,9 @@ v <- get return $ Fact n v +instance Serialize Fact where+ showp= showpBinary+ readp= readpBinary factorials = do all <- getAll
Demos/Inspect.hs view
@@ -10,7 +10,7 @@ -- For bugs, questions, whatever, please email me: Alberto Gómez Corona agocorona@gmail.com module Main where-import Control.Workflow.Text+import Control.Workflow --import Debug.Trace --import Data.Typeable import Control.Concurrent
Demos/docAprobal.hs view
@@ -22,10 +22,10 @@ -}-import Control.Workflow.Text+import Control.Workflow -import Data.Persistent.Queue.Text-import Control.Workflow.Text.Patterns+import Data.Persistent.Queue+import Control.Workflow.Patterns import Data.Typeable import System.Exit
+ Demos/pr.hs view
@@ -0,0 +1,26 @@++module Main where+import Control.Workflow+import Control.Concurrent(threadDelay)+import System.IO (hFlush,stdout)+import Data.Vector as V+import System.IO.Unsafe+import Control.Concurrent.MVar++count= unsafePerformIO $ newMVar 0++-- delayed evaluation of logged step values++instance Indexable (Vector a) where+ key= const "Vector"++mcount cart= do+ i <- step $ modifyMVar count (\n-> threadDelay 1000000>> print cart >> return (n+1, n `mod` 3))++ let newCart= cart V.// [(i, cart V.! i + 1 )]++ mcount newCart+ return ()++main= start "count" mcount (V.fromList [0,0,0 :: Int])+
Demos/sequence.hs view
@@ -1,6 +1,6 @@ module Main where-import Control.Workflow.Text+import Control.Workflow import Control.Concurrent(threadDelay) import System.IO (hFlush,stdout)
Workflow.cabal view
@@ -1,5 +1,5 @@ name: Workflow-version: 0.5.8.2+version: 0.6.0.0 build-type: Simple license: BSD3@@ -35,6 +35,10 @@ . New in this release, .+ * 0.6.0.0 Changes in ghc 7.4 forces to drop the abstract serializer+ now Text and binary versions are the same.+ For binary serialization, use 'showBinary' and 'readpBinary'+ . * 0.5.8.2 minor changes for the MFlow package . * 0.5.8.1 solved a bug that caused a "casting failure"@@ -70,29 +74,24 @@ extra-tmp-files: exposed-modules:- Control.Workflow.Binary- Control.Workflow.Binary.Patterns- Control.Workflow.Text- Control.Workflow.Text.Patterns - Data.Persistent.Queue.Text- Data.Persistent.Queue.Binary+ Control.Workflow+ Control.Workflow.Patterns + Data.Persistent.Queue++ other-modules:- Control.Workflow.Text.TextDefs- Control.Workflow.Binary.BinDefs Control.Workflow.IDynamic Control.Workflow.Stat- Control.Workflow.GenSerializer + extra-source-files:- Control/Workflow/Workflow.inc.hs- Control/Workflow/Patterns.inc.hs- Data/Persistent/Queue/Queue.inc.hs Demos/docAprobal.hs Demos/sequence.hs Demos/Fact.hs Demos/Inspect.hs+ Demos/pr.hs exposed: True buildable: True