SciFlow 0.4.0 → 0.4.1
raw patch · 9 files changed
+233/−30 lines, 9 filesdep +transformersbinary-added
Dependencies added: transformers
Files
- README.md +91/−0
- SciFlow.cabal +9/−6
- examples/Functions.hs +51/−0
- examples/Main.hs +18/−0
- examples/example.png binary
- src/Scientific/Workflow.hs +9/−2
- src/Scientific/Workflow/Builder.hs +45/−16
- src/Scientific/Workflow/Types.hs +8/−3
- src/Scientific/Workflow/Visualize.hs +2/−3
+ README.md view
@@ -0,0 +1,91 @@+Scientific workflow management system+=====================================++A scientific workflow is a series of computational steps which usually can be presented as a Directed Acyclic Graph (DAG).++SciFlow is to help programmers design complex workflows with ease. Here is a trivial example. (Since we use template haskell, we need to divide this small program into two parts.)++```haskell+---------------------------------------------------+-- File 1: MyModule.hs+---------------------------------------------------+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+module Functions+ (builder) where++import Control.Lens ((^.), (.=))+import qualified Data.Text as T+import Shelly hiding (FilePath)+import Text.Printf (printf)++import Scientific.Workflow++create :: () -> IO FilePath+create _ = do+ writeFile "hello.txt" "hello world"+ return "hello.txt"++countWords :: FilePath -> IO Int+countWords fl = do+ content <- readFile fl+ return $ length $ words content++countChars :: FilePath -> IO Int+countChars fl = do+ content <- readFile fl+ return $ sum $ map length $ words content++output :: (Int, Int) -> IO Bool+output (ws, cs) = do+ putStrLn $ printf "Number of words: %d" ws+ putStrLn $ printf "Number of characters: %d" cs+ return True++cleanUp :: (Bool, FilePath) -> IO ()+cleanUp (toBeRemoved, fl) = if toBeRemoved+ then shelly $ rm $ fromText $ T.pack fl+ else return ()++-- builder monad+builder :: Builder ()+builder = do+ node "step0" 'create $ label .= "write something to a file"+ node "step1" 'countWords $ label .= "word count"+ node "step2" 'countChars $ label .= "character count"+ node "step3" 'output $ label .= "print"+ node "step4" 'cleanUp $ label .= "remove the file"++ ["step0"] ~> "step1"+ ["step0"] ~> "step2"+ ["step1", "step2"] ~> "step3"+ ["step3", "step0"] ~> "step4"++---------------------------------------------------+-- File 2: main.hs+---------------------------------------------------+{-# LANGUAGE TemplateHaskell #-}++import System.Environment++import qualified Functions as F+import qualified Data.Text.Lazy.IO as T++import Scientific.Workflow+import Scientific.Workflow.Visualize++buildWorkflow "wf" F.builder++main :: IO ()+main = do+ (cmd:args) <- getArgs+ case cmd of+ "run" -> runWorkflow wf def+ "view" -> T.putStrLn $ renderBuilder F.builder+```++The workflow can be visualized by running `runghc main.hs view | dot -Tpng > example.png`.++++To run the workflow, simply type `runghc main.hs run`. The program will create a directory to store results of each step. If being terminated prematurely, the program will use the saved data to continue from the last step.
SciFlow.cabal view
@@ -1,8 +1,5 @@--- Initial SciFlow.cabal generated by cabal init. For further--- documentation, see http://haskell.org/cabal/users-guide/- name: SciFlow-version: 0.4.0+version: 0.4.1 synopsis: Scientific workflow management system description: SciFlow is to help programmers design complex workflows@@ -12,12 +9,17 @@ license-file: LICENSE author: Kai Zhang maintainer: kai@kzhang.org-copyright: (c) 2015 Kai Zhang+copyright: (c) 2016 Kai Zhang category: Control build-type: Simple--- extra-source-files: cabal-version: >=1.10 +extra-source-files:+ README.md+ examples/Functions.hs+ examples/Main.hs+ examples/example.png+ Flag Debug Description: Enable debug support Default: False@@ -50,6 +52,7 @@ , th-lift , text , template-haskell+ , transformers , containers , yaml
+ examples/Functions.hs view
@@ -0,0 +1,51 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+module Functions+ (builder) where++import Control.Lens ((^.), (.=))+import qualified Data.Text as T+import Shelly hiding (FilePath)+import Text.Printf (printf)++import Scientific.Workflow++create :: () -> IO FilePath+create _ = do+ writeFile "hello.txt" "hello world"+ return "hello.txt"++countWords :: FilePath -> IO Int+countWords fl = do+ content <- readFile fl+ return $ length $ words content++countChars :: FilePath -> IO Int+countChars fl = do+ content <- readFile fl+ return $ sum $ map length $ words content++output :: (Int, Int) -> IO Bool+output (ws, cs) = do+ putStrLn $ printf "Number of words: %d" ws+ putStrLn $ printf "Number of characters: %d" cs+ return True++cleanUp :: (Bool, FilePath) -> IO ()+cleanUp (toBeRemoved, fl) = if toBeRemoved+ then shelly $ rm $ fromText $ T.pack fl+ else return ()++-- builder monad+builder :: Builder ()+builder = do+ node "step0" 'create $ label .= "write something to a file"+ node "step1" 'countWords $ label .= "word count"+ node "step2" 'countChars $ label .= "character count"+ node "step3" 'output $ label .= "print"+ node "step4" 'cleanUp $ label .= "remove the file"++ ["step0"] ~> "step1"+ ["step0"] ~> "step2"+ ["step1", "step2"] ~> "step3"+ ["step3", "step0"] ~> "step4"
+ examples/Main.hs view
@@ -0,0 +1,18 @@+{-# LANGUAGE TemplateHaskell #-}++import System.Environment++import qualified Functions as F+import qualified Data.Text.Lazy.IO as T++import Scientific.Workflow+import Scientific.Workflow.Visualize++buildWorkflow "wf" F.builder++main :: IO ()+main = do+ (cmd:args) <- getArgs+ case cmd of+ "run" -> runWorkflow wf def+ "view" -> T.putStrLn $ renderBuilder F.builder
+ examples/example.png view
binary file changed (absent → 26711 bytes)
src/Scientific/Workflow.hs view
@@ -6,9 +6,12 @@ ) where import Control.Monad.State+import Control.Monad.Trans.Except+import Control.Exception (displayException) import Scientific.Workflow.Builder import Scientific.Workflow.Types+import System.IO runWorkflow :: [Workflow] -> State RunOpt () -> IO () runWorkflow wfs setOpt = do@@ -17,5 +20,9 @@ where opt = execState setOpt defaultRunOpt f config (Workflow wf) = do- (_, config') <- runStateT (wf ()) config- return config'+ result <- runExceptT $ runStateT (wf ()) config+ case result of+ Right (_, config') -> return config'+ Left ex -> do+ hPutStrLn stderr $ displayException ex+ return config
src/Scientific/Workflow/Builder.hs view
@@ -17,6 +17,8 @@ ) where import Control.Lens ((^.), (%~), _1, _2, _3, at, (.=))+import Control.Exception (try, displayException)+import Control.Monad.Trans.Except (throwE) import Control.Monad.State import qualified Data.Text as T import Data.Graph.Inductive.Graph@@ -48,14 +50,23 @@ lift t = [| T.pack $(T.lift $ T.unpack t) |] +-- | The order of edges type EdgeOrd = Int++-- | A computation node type Node = (PID, (ExpQ, Attribute))++-- | Links between computational nodes type Edge = (PID, PID, EdgeOrd) type Builder = State ([Node], [Edge]) -- | Declare a computational node-node :: ToExpQ q => PID -> q -> State Attribute () -> Builder ()+node :: ToExpQ q+ => PID -- ^ node id+ -> q -- ^ function+ -> State Attribute () -- ^ Attribues+ -> Builder () node p fn setAttr = modify $ _1 %~ (newNode:) where attr = execState setAttr defaultAttribute@@ -79,11 +90,14 @@ f a t = link [a] t >> return t {-# INLINE path #-} +-- | Build the workflow. buildWorkflow :: String -> Builder () -> Q [Dec] buildWorkflow wfName b = mkWorkflow wfName $ mkDAG b +-- | Build only a part of the workflow that has not been executed. This is used+-- during development for fast compliation. buildWorkflowPart :: State RunOpt () -> String -> Builder ()@@ -100,7 +114,7 @@ ks <- getKeys db pSt <- mapM (flip isFinished db) ks let pSts = M.fromList $ zipWith (\k s ->- if s then (k, Finished) else (k, Scheduled)) ks pSt+ if s then (k, Success) else (k, Scheduled)) ks pSt return $ WorkflowState db pSts {-# INLINE getWorkflowState #-} @@ -116,32 +130,39 @@ type DAG = Gr Node EdgeOrd +-- | Contruct a DAG representing the workflow mkDAG :: Builder () -> DAG mkDAG b = mkGraph ns' es' where ns' = map (\x -> (pid2nid $ fst x, x)) ns es' = map (\(fr, t, o) -> (pid2nid fr, pid2nid t, o)) es (ns, es) = execState b ([], [])- pid2nid p = M.findWithDefault (error "mkDAG") p m+ pid2nid p = M.findWithDefault errMsg p m where m = M.fromListWithKey err $ zip (map fst ns) [0..] err k _ _ = error $ "multiple instances for: " ++ T.unpack k+ errMsg = error $ "mkDAG: cannot identify node: " ++ T.unpack p {-# INLINE mkDAG #-} +-- | Remove nodes that are executed before from a DAG. trimDAG :: WorkflowState -> DAG -> DAG trimDAG st dag = gmap revise gr where- revise c | done (c^._3._1) && null (c^._1) = _3._2._1 %~ e $ c- | otherwise = c- where e x = [| const undefined >=> $(x) |]+ revise context@(linkTo, nd, lab, linkFrom)+ | done (fst lab) && null linkTo = _3._2._1 %~ e $ context+ | otherwise = context+ where+ e x = [| const undefined >=> $(x) |] gr = labnfilter f dag where f (i, (x,_)) = (not . done) x || any (not . done) children where children = map (fst . fromJust . lab dag) $ suc dag i- done x = getStatus x == Finished- getStatus x = M.findWithDefault Scheduled x $ st^.procStatus+ done x = case M.lookup x (st^.procStatus) of+ Just Success -> True+ _ -> False {-# INLINE trimDAG #-} +-- | Generate codes from a DAG mkWorkflow :: String -> DAG -> Q [Dec] mkWorkflow wfName dag = do decNode <- concat <$> mapM (f . snd) ns@@ -168,17 +189,25 @@ {-# INLINE mkWorkflow #-} mkProc :: Serializable b => PID -> (a -> IO b) -> (Processor a b)-mkProc p f = \input -> do+mkProc pid f = \input -> do st <- get- case M.findWithDefault Scheduled p (st^.procStatus) of- Finished -> lift $ readData p $ st^.db+ case M.findWithDefault Scheduled pid (st^.procStatus) of+ Fail ex -> lift $ throwE ex+ Success -> do+ r <- liftIO $ readData pid $ st^.db+ return r Scheduled -> do #ifdef DEBUG- traceM $ "Running node: " ++ T.unpack p+ traceM $ "Running node: " ++ T.unpack pid #endif- result <- lift $ f input- lift $ saveData p result $ st^.db - (procStatus . at p) .= Just Finished- return result+ result <- liftIO $ try $ f input+ case result of+ Left ex -> do+ (procStatus . at pid) .= Just (Fail ex)+ lift $ throwE ex+ Right r -> do+ liftIO $ saveData pid r $ st^.db+ (procStatus . at pid) .= Just Success+ return r {-# INLINE mkProc #-}
src/Scientific/Workflow/Types.hs view
@@ -26,6 +26,8 @@ import Control.Lens (makeLenses) import Control.Monad.State+import Control.Monad.Trans.Except (ExceptT)+import Control.Exception (SomeException) import qualified Data.ByteString as B import qualified Data.Map as M import qualified Data.Text as T@@ -40,13 +42,16 @@ serialize = encode deserialize = fromJust . decode +-- | An abstract type representing the database used to store states of workflow data WorkflowDB = WorkflowDB FilePath +-- | The id of a node type PID = T.Text -data ProcState = Finished+-- | The state of a computation node+data ProcState = Success | Scheduled- deriving (Eq)+ | Fail SomeException data WorkflowState = WorkflowState { _db :: WorkflowDB@@ -55,7 +60,7 @@ makeLenses ''WorkflowState -type Processor a b = a -> StateT WorkflowState IO b+type Processor a b = a -> StateT WorkflowState (ExceptT SomeException IO) b data Workflow where Workflow :: (Processor () o) -> Workflow
src/Scientific/Workflow/Visualize.hs view
@@ -4,9 +4,7 @@ ) where import Control.Lens-import Scientific.Workflow.Types-import Shelly (fromText, lsT, shelly, test_f, mkdir_p)-import qualified Data.ByteString as B+import Scientific.Workflow.Types (label) import qualified Data.Text as T import qualified Data.Text.Lazy as TL @@ -17,6 +15,7 @@ import Scientific.Workflow.Types (note) import Scientific.Workflow.Builder +-- | Print the computation graph renderBuilder :: Builder () -> TL.Text renderBuilder b = renderDot . toDot $ graphToDot param dag where