Workflow 0.7.0.7 → 0.8.0.0
raw patch · 6 files changed
+104/−238 lines, 6 filesdep +vector
Dependencies added: vector
Files
- Control/Workflow.hs +43/−12
- Control/Workflow/IDynamic.hs +0/−147
- Control/Workflow/Patterns.hs +44/−39
- Control/Workflow/Stat.hs +12/−18
- Demos/pr.hs +0/−17
- Workflow.cabal +5/−5
Control/Workflow.hs view
@@ -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
− Control/Workflow/IDynamic.hs
@@ -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'
Control/Workflow/Patterns.hs view
@@ -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
Control/Workflow/Stat.hs view
@@ -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
− Demos/pr.hs
@@ -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
Workflow.cabal view
@@ -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