diff --git a/Control/Workflow.hs b/Control/Workflow.hs new file mode 100644 --- /dev/null +++ b/Control/Workflow.hs @@ -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 ) + + + diff --git a/Control/Workflow/Binary.hs b/Control/Workflow/Binary.hs deleted file mode 100644 --- a/Control/Workflow/Binary.hs +++ /dev/null @@ -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" diff --git a/Control/Workflow/Binary/BinDefs.hs b/Control/Workflow/Binary/BinDefs.hs deleted file mode 100644 --- a/Control/Workflow/Binary/BinDefs.hs +++ /dev/null @@ -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/" - diff --git a/Control/Workflow/Binary/Patterns.hs b/Control/Workflow/Binary/Patterns.hs deleted file mode 100644 --- a/Control/Workflow/Binary/Patterns.hs +++ /dev/null @@ -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" diff --git a/Control/Workflow/GenSerializer.hs b/Control/Workflow/GenSerializer.hs deleted file mode 100644 --- a/Control/Workflow/GenSerializer.hs +++ /dev/null @@ -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 diff --git a/Control/Workflow/IDynamic.hs b/Control/Workflow/IDynamic.hs --- a/Control/Workflow/IDynamic.hs +++ b/Control/Workflow/IDynamic.hs @@ -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) diff --git a/Control/Workflow/Patterns.hs b/Control/Workflow/Patterns.hs new file mode 100644 --- /dev/null +++ b/Control/Workflow/Patterns.hs @@ -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) + + + + + diff --git a/Control/Workflow/Patterns.inc.hs b/Control/Workflow/Patterns.inc.hs deleted file mode 100644 --- a/Control/Workflow/Patterns.inc.hs +++ /dev/null @@ -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) - - - diff --git a/Control/Workflow/Stat.hs b/Control/Workflow/Stat.hs --- a/Control/Workflow/Stat.hs +++ b/Control/Workflow/Stat.hs @@ -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 "-----------------------------------" + diff --git a/Control/Workflow/Text.hs b/Control/Workflow/Text.hs deleted file mode 100644 --- a/Control/Workflow/Text.hs +++ /dev/null @@ -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" - - diff --git a/Control/Workflow/Text/Patterns.hs b/Control/Workflow/Text/Patterns.hs deleted file mode 100644 --- a/Control/Workflow/Text/Patterns.hs +++ /dev/null @@ -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" - diff --git a/Control/Workflow/Text/TextDefs.hs b/Control/Workflow/Text/TextDefs.hs deleted file mode 100644 --- a/Control/Workflow/Text/TextDefs.hs +++ /dev/null @@ -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 "-----------------------------------" - - diff --git a/Control/Workflow/Workflow.inc.hs b/Control/Workflow/Workflow.inc.hs deleted file mode 100644 --- a/Control/Workflow/Workflow.inc.hs +++ /dev/null @@ -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 ) - diff --git a/Data/Persistent/Queue.hs b/Data/Persistent/Queue.hs new file mode 100644 --- /dev/null +++ b/Data/Persistent/Queue.hs @@ -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 + diff --git a/Data/Persistent/Queue/Binary.hs b/Data/Persistent/Queue/Binary.hs deleted file mode 100644 --- a/Data/Persistent/Queue/Binary.hs +++ /dev/null @@ -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" diff --git a/Data/Persistent/Queue/Queue.inc.hs b/Data/Persistent/Queue/Queue.inc.hs deleted file mode 100644 --- a/Data/Persistent/Queue/Queue.inc.hs +++ /dev/null @@ -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 diff --git a/Data/Persistent/Queue/Text.hs b/Data/Persistent/Queue/Text.hs deleted file mode 100644 --- a/Data/Persistent/Queue/Text.hs +++ /dev/null @@ -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" - - diff --git a/Demos/Fact.hs b/Demos/Fact.hs --- a/Demos/Fact.hs +++ b/Demos/Fact.hs @@ -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 diff --git a/Demos/Inspect.hs b/Demos/Inspect.hs --- a/Demos/Inspect.hs +++ b/Demos/Inspect.hs @@ -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 diff --git a/Demos/docAprobal.hs b/Demos/docAprobal.hs --- a/Demos/docAprobal.hs +++ b/Demos/docAprobal.hs @@ -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 diff --git a/Demos/pr.hs b/Demos/pr.hs new file mode 100644 --- /dev/null +++ b/Demos/pr.hs @@ -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]) + diff --git a/Demos/sequence.hs b/Demos/sequence.hs --- a/Demos/sequence.hs +++ b/Demos/sequence.hs @@ -1,6 +1,6 @@ module Main where -import Control.Workflow.Text +import Control.Workflow import Control.Concurrent(threadDelay) import System.IO (hFlush,stdout) diff --git a/Workflow.cabal b/Workflow.cabal --- a/Workflow.cabal +++ b/Workflow.cabal @@ -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