diff --git a/Control/Workflow.hs b/Control/Workflow.hs --- a/Control/Workflow.hs +++ b/Control/Workflow.hs @@ -110,6 +110,8 @@ , newWFRef , stepWFRef , readWFRef +, getNRefs +, getWFRef -- * State manipulation , writeWFRef , moveState @@ -180,7 +182,7 @@ import Data.TCache import Data.TCache.Defs import Data.RefSerialize -import Control.Workflow.IDynamic +import Data.Persistent.IDynamic import Unsafe.Coerce import System.Mem.StableName import Control.Workflow.Stat @@ -658,7 +660,7 @@ else do -- has been running but not running now mst <- readDBRef sref stat' <- case mst of - Nothing -> error $ "getState: Workflow not found: "++ key + Nothing -> return stat1 -- error $ "getState: Workflow not found: "++ key Just s' -> do -- the thread may have been killed by an exception when running s <- case recover s' of @@ -771,16 +773,16 @@ -- :: (Serialize a, Typeable a, Monad m) -- => Int -- ^ the step number. If negative, count from the current state backwards -- -> Workflow m a -- ^ return the n-tn intermediate step result ---getStep i= WF(\s -> do --- let ind= index s --- stat= state s +--getStep i= WF(\s -> do -- +-- let stat= state s +-- -- return (s, if i > 0 && i < stat then fromIDyn $ versions s !! (stat -i-1) -- else if i <= 0 && i > -stat then fromIDyn $ versions s !! (stat - ind +i-1) -- else error "getStep: wrong index") -- ) --- | Return the list of object keys that are running for a workflow +-- | Return the keys of the workflows that are running with a given prefix getWFKeys :: String -> IO [String] getWFKeys wfname= do mwfs <- atomically $ readDBRef tvRunningWfs @@ -911,26 +913,43 @@ => a -> Workflow m (WFRef a) newWFRef x= stepWFRef (return x) >>= return . fst --- | Execute an step but return a reference to the result besides the result itself +-- | Execute an step and return a reference to the result besides the result itself -- stepWFRef :: ( Serialize a , Typeable a , MonadIO m) => m a -> Workflow m (WFRef a,a) stepWFRef exp= do - r <- step exp -- !> "stepWFRef" - + r <- step exp -- !> "stepWFRef" WF(\s@Stat{..} -> do - - - let (n,flag)= if recover then (state - (L.length versions) -1 ,False) + let (n,flag)= if recover + then (state - (L.length versions) -1 ,False) else (state -1 ,True) let ref = WFRef n self let s'= s{references= (n,(toIDyn r,flag)):references } liftIO $ atomically $ writeDBRef self s' r `seq` return (s',(ref,r)) ) +getNRefs wfname= do + st <- getResource stat0{wfName= wfname} `onNothing` error ("Workflow not found: "++ wfname) + return $ L.length $ references st + +getWFRef :: Int -> String -> WFRef a +getWFRef n s= WFRef n $ getDBRef s +--getReference ::(Monad m +-- , Serialize a +-- , Typeable a) +-- => Int -> Workflow m (Maybe a) +--getReference n = WF $ \st -> +-- case L.lookup n $! references st of +-- Just (r,_) -> return(st, Just $ fromIDyn r) +-- Nothing -> do +-- let n1= state st - n +-- return (st, Just . fromIDyn $ versions st !! n1) + + + -- | Read the content of a Workflow reference. Note that its result is not in the Workflow monad readWFRef :: ( Serialize a , Typeable a) @@ -1159,6 +1178,18 @@ Just x -> do 'logWF' $ "received" ++ show x ... @ -} + +--longWait :: Integer -> Workflow m a -> Workflow m a +--longWait time wf= +-- WF $ \s -> do +-- flag <- getTimeoutFlag time +-- forkIO $ atomically $ do +-- b <- readTVar tv +-- if b == False then retry else return () +-- start (wfName s) wf +-- myThreadId >>= killThread + + waitUntilSTM :: TVar Bool -> STM() waitUntilSTM tv = do b <- readTVar tv diff --git a/Control/Workflow/IDynamic.hs b/Control/Workflow/IDynamic.hs deleted file mode 100644 --- a/Control/Workflow/IDynamic.hs +++ /dev/null @@ -1,147 +0,0 @@ - {-# OPTIONS -XExistentialQuantification - -XOverlappingInstances - -XUndecidableInstances - -XScopedTypeVariables - -XDeriveDataTypeable - -XTypeSynonymInstances - -XIncoherentInstances - -XOverloadedStrings - -XMultiParamTypeClasses - -XFunctionalDependencies - -XFlexibleInstances #-} -{- | -IDynamic is a indexable and serializable version of Dynamic. (See @Data.Dynamic@). It is used as containers of objects -in the cache so any new datatype can be incrementally stored without recompilation. -IDimamic provices methods for safe casting, besides serializaton, deserialirezation and retrieval by key. --} -module Control.Workflow.IDynamic where -import Data.Typeable -import Unsafe.Coerce -import System.IO.Unsafe -import Data.TCache -import Data.TCache.Defs -import Data.RefSerialize -import Data.Char (showLitChar) - -import Data.ByteString.Lazy.Char8 as B - -import Data.Word -import Numeric (showHex, readHex) -import Control.Exception(handle, SomeException, ErrorCall) -import Control.Monad(replicateM) -import Data.Word -import Control.Concurrent.MVar -import Data.IORef -import Data.Map as M(empty) -import Data.RefSerialize -import Data.HashTable as HT - - - - -data IDynamic = IDyn (IORef IDynType) - -data IDynType= forall a w r.(Typeable a, Serialize a) - => DRight !a - | DLeft !(ByteString ,(Context, ByteString)) - - - deriving Typeable - -newtype Save= Save ByteString deriving Typeable - -tosave d@(IDyn r)= unsafePerformIO $ do - mr<- readIORef r - case mr of - DRight _ -> return d - DLeft (s,_) -> writeIORef r (DRight $ Save s) >> return d - - -instance Serialize Save where - showp (Save s)= insertString s - readp = error "readp not impremented for Save" - - -errorfied str str2= error $ str ++ ": IDynamic object not reified: "++ str2 - - - -dynPrefix= "Dyn" -dynPrefixSp= append (pack dynPrefix) " " -notreified = pack $ dynPrefix ++" 0" - - - -instance Serialize IDynamic where - - showp (IDyn t)= - case unsafePerformIO $ readIORef t of - DRight x -> do --- insertString $ pack dynPrefix - c <- getWContext - showpx <- rshowps x --- showpText . fromIntegral $ B.length showpx - showp $ unpack showpx - - DLeft (showpx,_) -> -- error $ "IDynamic not reified :: "++ unpack showpx --- insertString notreified - insertString $ encode showpx - where - encode = pack . show . unpack - - readp = lexeme (do --- symbol dynPrefix --- n <- readpText --- s <- takep n - - s <- rreadp :: STR String - - c <- getRContext - return . IDyn . unsafePerformIO . newIORef $ DLeft ( pack s, c)) - "IDynamic" - - - -instance Show IDynamic where - show (IDyn r) = - let t= unsafePerformIO $ readIORef r - in case t of - DRight x -> "IDyn (" ++ ( unpack . runW $ showp x) ++ ")" - DLeft (s, _) -> "IDyns " ++ unpack s - - - - - -toIDyn x= IDyn . unsafePerformIO . newIORef $ DRight x - - -fromIDyn :: (Typeable a , Serialize a)=> IDynamic -> a -fromIDyn x=r where - r = case safeFromIDyn x of - Nothing -> error $ "fromIDyn: casting failure for data " - ++ show x ++ " to type: " - ++ (show $ typeOf r) - Just v -> v - - -safeFromIDyn :: (Typeable a, Serialize a) => IDynamic -> Maybe a -safeFromIDyn (IDyn r)=unsafePerformIO $ do - t<- readIORef r - case t of - DRight x -> return $ cast x - - DLeft (str, c) -> - handle (\(e :: SomeException) -> return Nothing) $ -- !> ("safeFromIDyn : "++ show e)) $ - do - let v= runRC c rreadp str - writeIORef r $! DRight v -- !> ("***reified "++ unpack str) - return $! Just v -- !> ("*** end reified " ++ unpack str) - - ---main= print (safeFromIDyn $ IDyn $ unsafePerformIO $ newIORef $ DLeft $ (pack "1", (unsafePerformIO $ HT.new (==) HT.hashInt, pack "")) :: Maybe Int) - -reifyM :: (Typeable a,Serialize a) => IDynamic -> a -> IO a -reifyM dyn v = do - let v'= fromIDyn dyn - return $ v' `seq` v' diff --git a/Control/Workflow/Patterns.hs b/Control/Workflow/Patterns.hs --- a/Control/Workflow/Patterns.hs +++ b/Control/Workflow/Patterns.hs @@ -90,12 +90,12 @@ import Control.Exception.Extensible (Exception,SomeException) import Data.RefSerialize import Control.Workflow.Stat - +import qualified Data.Vector as V import Data.TCache import Debug.Trace - +import Data.Maybe -newtype ActionWF a= ActionWF (WFRef(Maybe a)) -- (WFRef (String, Bool)) +data ActionWF a= ActionWF (WFRef(Maybe a)) ThreadId -- (WFRef (String, Bool)) -- | spawn a list of independent workflow 'actions' with a seed value 'a' -- The results are reduced by `merge` or `select` @@ -107,9 +107,9 @@ split actions a = mapM (\ac -> do mv <- newWFRef Nothing - fork (ac a >>= \v -> (step . liftIO . atomicallySync . writeWFRef mv . Just) v ) + th<- fork (ac a >>= \v -> (step . liftIO . atomicallySync . writeWFRef mv . Just) v ) - return $ ActionWF mv ) + return $ ActionWF mv th ) actions @@ -121,13 +121,12 @@ , Typeable b , Serialize a, Serialize b) => ([a] -> io b) -> [ActionWF a] -> Workflow io b -merge cond results= step $ mapM (\(ActionWF mv ) -> readWFRef1 mv ) results >>= cond -- !> "cond" +merge cond results= step $ mapM (\(ActionWF mv _ ) -> liftIO (atomically $ readWFRef1 mv) ) results >>= cond -- !> "cond" -readWFRef1 :: ( MonadIO io - , Serialize a +readWFRef1 :: ( Serialize a , Typeable a) - => WFRef (Maybe a) -> io a -readWFRef1 r = liftIO . atomically $ do + => WFRef (Maybe a) -> STM a +readWFRef1 r = do mv <- readWFRef r @@ -138,10 +137,11 @@ data Select - = Select - | Discard - | FinishDiscard - | FinishSelect + = Select -- ^ select the source output + | Discard -- ^ Discard the source output + | Continue -- ^ Continue the source process + | FinishDiscard -- ^ Discard this output, kill all and return the selected outputs + | FinishSelect -- ^ Select this output, kill all and return the selected outputs deriving(Typeable, Read, Show) instance Exception Select @@ -150,38 +150,42 @@ -- The check filter, can select , discard or finish the entire computation before -- the timeout is reached. When the computation finalizes, it kill all -- the pending workflows and return the list of selected outputs --- the timeout is in seconds and is no limited to a single execution, so it can proceed for years. +-- the timeout is in seconds and it is is in the workflow monad, +-- so it is possible to restart the process if interrupted, +-- so it can proceed for years. -- -- This is necessary for the modelization of real-life institutional cycles such are political elections -- A timeout of 0 means no timeout. select :: ( Serialize a - , Serialize [a] +-- , Serialize [a] , Typeable a , HasFork io , CMC.MonadCatchIO io) => Integer - -> (a -> io Select) + -> (a -> STM Select) -> [ActionWF a] -> Workflow io [a] select timeout check actions= do - res <- liftIO $ newMVar [] + res <- liftIO $ newTVarIO $ V.generate(length actions) (const Nothing) flag <- getTimeoutFlag timeout parent <- liftIO myThreadId checThreads <- liftIO $ newEmptyMVar count <- liftIO $ 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 - liftIO $ throwTo parent FinishDiscard - FinishSelect -> do - addRes r - liftIO $ throwTo parent FinishDiscard + let check' (ActionWF ac _) i = do + liftIO . atomically $ do + r <- readWFRef1 ac + b <- check r + case b of + Discard -> return () + Select -> addRes i r + Continue -> addRes i r >> retry + FinishDiscard -> do + unsafeIOToSTM $ throwTo parent FinishDiscard + FinishSelect -> do + addRes i r + unsafeIOToSTM $ throwTo parent FinishDiscard n <- liftIO $ CMC.block $ do n <- takeMVar count @@ -195,7 +199,7 @@ `CMC.catch` (\(e :: Select) -> liftIO $ throwTo parent e) - ws <- mapM ( fork . check') actions + ws <- mapM (\(ac,i) -> fork $ check' ac i) $ zip actions [0..] liftIO $ putMVar checThreads ws liftIO $ atomically $ do @@ -206,17 +210,18 @@ throw FinishDiscard where - addRes r= liftIO $ CMC.block $ do - l <- takeMVar res - putMVar res $ r : l + addRes i r= do + l <- readTVar res + writeTVar res $ l V.// [(i, Just r)] let killall = liftIO $ do ws <- readMVar checThreads - liftIO $ mapM_ killThread ws -- !> "KILLALL" + liftIO $ mapM_ killThread ws + liftIO $ mapM_ (\(ActionWF _ th) -> killThread th)actions -- !> "KILLALL" step $ CMC.catch process -- (WF $ \s -> process >>= \ r -> return (s, r)) (\(e :: Select)-> do - liftIO $ readMVar res + liftIO $ return . catMaybes . V.toList =<< atomically ( readTVar res) ) `CMC.finally` killall @@ -233,7 +238,7 @@ -- @ vote :: ( Serialize b - , Serialize [b] +-- , Serialize [b] , Typeable b , HasFork io , CMC.MonadCatchIO io) @@ -243,7 +248,7 @@ -> a -> Workflow io c vote timeout actions comp x= - split actions x >>= select timeout (const $ return Select) >>= comp + split actions x >>= select timeout (const $ return Continue) >>= comp -- | sum the outputs of a list of workflows according with its monoid definition @@ -251,7 +256,7 @@ -- @ sumUp timeout actions = vote timeout actions (return . mconcat) @ sumUp :: ( Serialize b - , Serialize [b] +-- , Serialize [b] , Typeable b , Monoid b , HasFork io @@ -291,7 +296,7 @@ return refs - mapM (\r -> readWFRef1 r >>= print) refs + mapM (\r -> liftIO (atomically $ readWFRef1 r) >>= print) refs diff --git a/Control/Workflow/Stat.hs b/Control/Workflow/Stat.hs --- a/Control/Workflow/Stat.hs +++ b/Control/Workflow/Stat.hs @@ -7,6 +7,7 @@ -XOverloadedStrings -XRecordWildCards -XScopedTypeVariables + -XPatternGuards #-} module Control.Workflow.Stat where @@ -21,13 +22,13 @@ import Control.Concurrent.STM(TVar, newTVarIO) import Data.IORef import Data.RefSerialize -import Control.Workflow.IDynamic +import Data.Persistent.IDynamic import Control.Monad(replicateM) import qualified Data.ByteString.Lazy.Char8 as B hiding (index) import Data.ByteString.Char8(findSubstring) -import Control.Workflow.IDynamic +import Data.Persistent.IDynamic import Control.Concurrent import Control.Exception(catch,bracket,SomeException) import System.IO.Error @@ -37,7 +38,6 @@ import Control.Monad --import Debug.Trace --- --(!>) = flip trace data WF s m l = WF { st :: s -> m (s,l) } @@ -99,7 +99,7 @@ showp stat@Stat{..} = do s <- showps $ Prelude.reverse versions - let l= show (B.length s + lenLen) ++" "++ show state + let l= show (B.length s + 1 + lenLen) ++" "++ show state insertString . B.pack $ l ++ take (fromIntegral lenLen - length l - 2) (repeat ' ')++ "\r\n" insertString s header stat @@ -205,9 +205,9 @@ | otherwise= do hSeek h AbsoluteSeek 0 -- !> ("WRITING partial " ++ wfName ) let s = runWC context $ insertString "\r\n" >> showpe written ( reverse $ take (state - written) versions) - let l= show (seek -3 + B.length s) ++" "++ show state - B.hPut h . B.pack $ l ++ take (fromIntegral lenLen - length l - 2) (repeat ' ') ++ "\r\n" - hSeek h AbsoluteSeek (fromIntegral seek - 3) + let l= show (seek - 5 + B.length s) ++" "++ show state + B.hPut h . B.pack $ l ++ take (fromIntegral lenLen - length l - 2) (repeat ' ') ++ "\r\n" + hSeek h AbsoluteSeek (fromIntegral seek - 5) B.hPut h s writeHeader h writeContext h @@ -320,20 +320,14 @@ -instance Indexable String where - key= id - -instance Indexable Int where - key= show - -instance Indexable Integer where - key= show +wFRefStr = "WFRef" +-- | default instances -instance Indexable () where - key _= "void" +instance (Show a, Read a )=> Serialize a where + showp= showpText + readp= readpText -wFRefStr = "WFRef" instance Serialize (WFRef a) where showp (WFRef n ref)= do diff --git a/Demos/pr.hs b/Demos/pr.hs deleted file mode 100644 --- a/Demos/pr.hs +++ /dev/null @@ -1,17 +0,0 @@ - -module Main where -import Control.Workflow - - - -main = do - syncWrite SyncManual - getName >>= putStrLn - main - -getName= exec1nc "test" $ do - name <- step $ do - putStrLn "your name?" - getLine - - return $ "hello " ++ name diff --git a/Workflow.cabal b/Workflow.cabal --- a/Workflow.cabal +++ b/Workflow.cabal @@ -1,5 +1,5 @@ name: Workflow -version: 0.7.0.7 +version: 0.8.0.0 cabal-version: >= 1.6 build-type: Simple license: BSD3 @@ -28,7 +28,7 @@ data-dir: "" extra-source-files: Demos/Fact.hs Demos/Inspect.hs - Demos/docAprobal.hs Demos/pr.hs Demos/sequence.hs + Demos/docAprobal.hs Demos/sequence.hs @@ -36,7 +36,7 @@ build-depends: MonadCatchIO-mtl -any, RefSerialize ==0.3.0.* , TCache -any && <1.0, base >=4 && <5, binary -any, bytestring -any, containers -any, directory -any, extensible-exceptions -any, - mtl -any, old-time -any, stm >2 + mtl -any, old-time -any, stm >2, vector -any exposed-modules: Control.Workflow Control.Workflow.Configuration @@ -47,8 +47,8 @@ MultiParamTypeClasses ExistentialQuantification TypeSynonymInstances RecordWildCards DeriveDataTypeable hs-source-dirs: . - other-modules: Control.Workflow.Stat Control.Workflow.Stat - Control.Workflow.IDynamic + other-modules: Control.Workflow.Stat + source-repository head type : git