SciFlow 0.4.1 → 0.5.0

raw patch · 13 files changed

+697/−269 lines, 13 filesdep +cerealdep +directorydep +drmaadep −data-default-classbinary-addedPVP ok

version bump matches the API change (PVP)

Dependencies added: cereal, directory, drmaa, executable-path, lifted-async, rainbow, sqlite-simple

Dependencies removed: data-default-class

API changes (from Hackage documentation)

Files

• README.md +30/−19
• SciFlow.cabal +23/−9
• example.png binary
• examples/Functions.hs +0/−51
• examples/Main.hs +0/−18
• examples/example.png binary
• src/Scientific/Workflow.hs +23/−17
• src/Scientific/Workflow/Builder.hs +141/−76
• src/Scientific/Workflow/DB.hs +72/−18
• src/Scientific/Workflow/Main.hs +239/−0
• src/Scientific/Workflow/Types.hs +104/−52
• src/Scientific/Workflow/Utils.hs +56/−0
• src/Scientific/Workflow/Visualize.hs +9/−9

@@ -1,10 +1,27 @@ Scientific workflow management system ===================================== -A scientific workflow is a series of computational steps which usually can be presented as a Directed Acyclic Graph (DAG).+Introduction+------------ -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.)+SciFlow is a workflow management system for working with big data pipelines locally+or in a grid computing system. +Most scientific computing pipelines are composed of many computational steps, and each of them involves heavy computation and IO operations. A workflow management system can+help user design complex computing patterns and track the states of computation.+The ability to recover from failures is crucial in large pipelines as they usually+take days or weeks to finish.++Features:++1. Easy to use: A simple and flexible way to specify computational pipelines in Haskell.++2. Automatic Checkpointing: The result of each intermediate step is stored, allowing easy restart upon failures.++3. Parallelism and grid computing support: Independent computational steps will run concurrently. And users can decide whether to run steps locally or on remote compute nodes in a grid system.++Here is a simple example. (Since we use template haskell, we need to divide this small program into two files.)+ ```haskell --------------------------------------------------- -- File 1: MyModule.hs@@ -66,26 +83,20 @@ --------------------------------------------------- {-# 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+import Scientific.Workflow.Main -main :: IO ()-main = do-    (cmd:args) <- getArgs-    case cmd of-        "run" -> runWorkflow wf def-        "view" -> T.putStrLn $ renderBuilder F.builder+defaultMain F.builder ``` -The workflow can be visualized by running `runghc main.hs view | dot -Tpng > example.png`.+Use `ghc main.hs -threaded` to compile the program. And type `./main --help` to+see available commands. For example, the workflow can be visualized by running+`./main view | dot -Tpng > example.png`, as shown below. -![example](examples/example.png)+![example](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.+To run the workflow, simply type `./main run`. The program will create a sqlite database to store intermediate results. If being terminated prematurely, the program will use the saved data to continue from the last step.++To enable grid compute engine support, you need to have DRMAA C library installed+and compile the SciFlow with `-f sge` flag. Use `./main run --remote` to submit jobs+to remote machines.

@@ -1,5 +1,5 @@ name:                SciFlow-version:             0.4.1+version:             0.5.0 synopsis:            Scientific workflow management system description:   SciFlow is to help programmers design complex workflows@@ -16,44 +16,58 @@  extra-source-files:   README.md-  examples/Functions.hs-  examples/Main.hs-  examples/example.png+  example.png -Flag Debug+Flag debug   Description: Enable debug support   Default:     False +Flag sge+  Description: Enable SGE support+  Default:     False+ library   ghc-options: -Wall   exposed-modules:     Scientific.Workflow     Scientific.Workflow.Builder     Scientific.Workflow.DB+    Scientific.Workflow.Main     Scientific.Workflow.Types+    Scientific.Workflow.Utils     Scientific.Workflow.Visualize +  other-modules:+    Paths_SciFlow+   if flag(debug)     CPP-Options: -DDEBUG-  else-    CPP-Options: -DNDEBUG +  if flag(sge)+    CPP-Options: -DSGE+    build-depends: drmaa+   build-depends:       base >=4.0 && <5.0     , bytestring-    , data-default-class+    , containers+    , cereal+    , directory+    , executable-path     , fgl     , graphviz     , lens >=4.0+    , lifted-async     , mtl     , optparse-applicative+    , rainbow     , shelly+    , sqlite-simple     , split     , th-lift     , text     , template-haskell     , transformers-    , containers     , yaml    hs-source-dirs:      src

binary file changed (absent → 26711 bytes)

@@ -1,51 +0,0 @@-{-# 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"

@@ -1,18 +0,0 @@-{-# 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

binary file changed (26711 → absent bytes)

@@ -1,28 +1,34 @@ module Scientific.Workflow     ( runWorkflow-    , getWorkflowState     , module Scientific.Workflow.Builder     , module Scientific.Workflow.Types     ) where +import           Control.Concurrent          (forkIO)+import           Control.Concurrent.MVar+import           Control.Exception           (bracket, displayException) import           Control.Monad.State-import Control.Monad.Trans.Except-import Control.Exception (displayException)+import           Control.Monad.Trans.Except+import qualified Data.Map                    as M+import qualified Data.Set                    as S  import           Scientific.Workflow.Builder+import           Scientific.Workflow.DB import           Scientific.Workflow.Types-import System.IO+import           Scientific.Workflow.Utils+import           Text.Printf                 (printf) -runWorkflow :: [Workflow] -> State RunOpt () -> IO ()-runWorkflow wfs setOpt = do-    config <- getWorkflowState $ _dbPath opt-    foldM_ f config wfs-  where-    opt = execState setOpt defaultRunOpt-    f config (Workflow wf) = do-        result <- runExceptT $ runStateT (wf ()) config-        case result of-            Right (_, config') -> return config'-            Left ex -> do-                hPutStrLn stderr $ displayException ex-                return config+runWorkflow :: Workflow -> RunOpt -> IO ()+runWorkflow (Workflow pids _ wf) opts = bracket (openDB $ database opts) closeDB $ \db -> do+    ks <- S.fromList <$> getKeys db+    pidStateMap <- flip M.traverseWithKey pids $ \pid attr -> do+        v <- if pid `S.member` ks then newMVar Success else newMVar Scheduled+        return (v, attr)+    para <- newEmptyMVar+    _ <- forkIO $ replicateM_ (nThread opts) $ putMVar para ()+    let initState = WorkflowState db pidStateMap para $ runOnRemote opts+    result <- runExceptT $ evalStateT (wf ()) initState+    case result of+        Right _ -> return ()+        Left (pid, ex) -> error' $ printf "\"%s\" failed. The error was: %s"+            pid (displayException ex)

@@ -12,45 +12,46 @@     , Builder     , buildWorkflow     , buildWorkflowPart-    , getWorkflowState     , mkDAG     ) where -import Control.Lens ((^.), (%~), _1, _2, _3, at, (.=))-import Control.Exception (try, displayException)+import Control.Lens ((^.), (%~), _1, _2, _3, at)+import Control.Exception (try) import Control.Monad.Trans.Except (throwE) import           Control.Monad.State+import Control.Concurrent.MVar+import Control.Concurrent (forkIO)+import Control.Concurrent.Async.Lifted (concurrently, mapConcurrently) import qualified Data.Text           as T-import Data.Graph.Inductive.Graph-    ( mkGraph-    , lab-    , labNodes-    , outdeg-    , lpre-    , labnfilter-    , gmap-    , suc-    , subgraph )+import Data.Graph.Inductive.Graph ( mkGraph, lab, labNodes, labEdges, outdeg+                                  , lpre, labnfilter, nfilter, gmap, suc ) import Data.Graph.Inductive.PatriciaTree (Gr) import Data.List (sortBy)-import qualified Data.Map as M-import Data.Maybe (fromJust)+import Data.Maybe (fromJust, fromMaybe) import Data.Ord (comparing)-import qualified Data.Map                    as M+import qualified Data.Map as M+import Text.Printf (printf)  import           Language.Haskell.TH import qualified Language.Haskell.TH.Lift as T  import Scientific.Workflow.Types import Scientific.Workflow.DB+import Scientific.Workflow.Utils (debug, runRemote, defaultRemoteOpts) -import Debug.Trace (traceM) ++T.deriveLift ''M.Map+T.deriveLift ''Attribute+ instance T.Lift T.Text where   lift t = [| T.pack $(T.lift $ T.unpack t) |] +instance T.Lift (Gr (PID, Attribute) Int) where+  lift gr = [| uncurry mkGraph $(T.lift (labNodes gr, labEdges gr)) |] --- | The order of edges++-- | The order of incoming edges of a node type EdgeOrd = Int  -- | A computation node@@ -59,9 +60,13 @@ -- | Links between computational nodes type Edge = (PID, PID, EdgeOrd) +type Function = (PID, ExpQ)+ type Builder = State ([Node], [Edge]) --- | Declare a computational node++-- | Declare a computational node. The function must have the signature:+-- (DBData a, DBData b) => a -> IO b node :: ToExpQ q      => PID                  -- ^ node id      -> q                    -- ^ function@@ -73,6 +78,12 @@     newNode = (p, (toExpQ fn, attr)) {-# INLINE node #-} ++-- | Declare a function that can be called on remote+--function :: ToExpQ q => T.Text -> q -> Builder ()+--function funcName fn =++ -- | many-to-one generalized link function link :: [PID] -> PID -> Builder () link xs t = modify $ _2 %~ (zip3 xs (repeat t) [0..] ++)@@ -90,33 +101,29 @@     f a t = link [a] t >> return t {-# INLINE path #-} --- | Build the workflow.-buildWorkflow :: String+-- | Build the workflow. This function will first create functions defined in+-- the builder. These pieces will then be assembled to form a function that will+-- execute each individual function in a correct order, named $prefix$_sciflow.+-- Lastly, a function table will be created with the name $prefix$_function_table.+buildWorkflow :: String     -- ^ prefix               -> Builder ()               -> Q [Dec]-buildWorkflow wfName b = mkWorkflow wfName $ mkDAG b+buildWorkflow prefix b = mkWorkflow prefix $ 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 ()+buildWorkflowPart :: FilePath   -- ^ path to the db                   -> String                   -> Builder ()                   -> Q [Dec]-buildWorkflowPart setOpt wfName b = do-    st <- runIO $ getWorkflowState $ opt^.dbPath+buildWorkflowPart db wfName b = do+    st <- runIO $ getWorkflowState db     mkWorkflow wfName $ trimDAG st $ mkDAG b   where-    opt = execState setOpt defaultRunOpt--getWorkflowState :: FilePath -> IO WorkflowState-getWorkflowState dir = do-    db <- openDB dir-    ks <- getKeys db-    pSt <- mapM (flip isFinished db) ks-    let pSts = M.fromList $ zipWith (\k s ->-                 if s then (k, Success) else (k, Scheduled)) ks pSt-    return $ WorkflowState db pSts-{-# INLINE getWorkflowState #-}+    getWorkflowState dir = do+        db <- openDB dir+        ks <- getKeys db+        return $ M.fromList $ zip ks $ repeat Success  -- | Objects that can be converted to ExpQ class ToExpQ a where@@ -145,69 +152,127 @@ {-# INLINE mkDAG #-}  -- | Remove nodes that are executed before from a DAG.-trimDAG :: WorkflowState -> DAG -> DAG+trimDAG :: (M.Map T.Text NodeResult) -> DAG -> DAG trimDAG st dag = gmap revise gr   where-    revise context@(linkTo, nd, lab, linkFrom)+    revise context@(linkTo, _, lab, _)         | done (fst lab) && null linkTo = _3._2._1 %~ e $ context         | otherwise = context       where-        e x = [| const undefined >=> $(x) |]+        e x = [| (\() -> 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 = case M.lookup x (st^.procStatus) of+    done x = case M.lookup x st 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-    decWf <- [d| $(varP $ mkName wfName) = $(fmap ListE $ mapM linking leafNodes)-             |]-    return $ decNode ++ decWf++-- Generate codes from a DAG. This function will create functions defined in+-- the builder. These pieces will be assembled to form a function that will+-- execute each individual function in a correct order.+-- Lastly, a function table will be created with the name $prefix$_function_table.+mkWorkflow :: String   -- prefix+           -> DAG -> Q [Dec]+mkWorkflow workflowName dag = do+    -- write node funcitons+    functions <- fmap concat $ forM computeNodes $ \(p, (fn,_)) -> [d|+        $(varP $ mkName $ T.unpack p) = mkProc p $(fn) |]++    -- function table+    funcTable <-+        [d| $(varP $ mkName functionTableName) = M.fromList+                $( fmap ListE $ forM computeNodes $ \(p, (fn, _)) ->+                [| (T.unpack p, Closure $(fn)) |] ) |]++    -- define workflows+    workflows <-+        [d| $(varP $ mkName workflowName) = Workflow pids+                $(varE $ mkName functionTableName)+                $(connect sinks [| const $ return () |]) |]++    return $ functions ++ funcTable ++ workflows   where-    f (p, (fn, _)) = [d| $(varP $ mkName $ T.unpack p) = mkProc p $(fn) |]-    ns = labNodes dag-    leafNodes = filter ((==0) . outdeg dag . fst) ns-    linking nd = [| Workflow $(go nd) |]+    functionTableName = workflowName ++ "_function_table"+    computeNodes = snd $ unzip $ labNodes dag+    pids = M.fromList $ map (\(i, x) -> (i, snd x)) computeNodes+    sinks = labNodes $ nfilter ((==0) . outdeg dag) dag++    backTrack sink = connect sources $ mkNodeVar sink       where-        go n = connect inNodes n-          where-            inNodes = map (\(x,_) -> (x, fromJust $ lab dag x)) $-                      sortBy (comparing snd) $ lpre dag $ fst n-        define n = varE $ mkName (T.unpack $ (snd n) ^. _1)-        connect [] t = define t-        connect [s1] t = [| $(go s1) >=> $(define t) |]-        connect xs t = [| $(foldl g e0 $ tail xs) >=> $(define t) |]-          where-            e0 = [| (fmap.fmap) $(conE (tupleDataName $ length xs)) $(go $ head xs) |]-            g acc x = [| ((<*>) . fmap (<*>)) $(acc) $(go x) |]+        sources = map (\(x,_) -> (x, fromJust $ lab dag x)) $+            sortBy (comparing snd) $ lpre dag $ fst sink++    connect [] sink = sink+    connect [source] sink = [| $(backTrack source) >=> $(sink) |]+    connect sources sink = [| fmap runParallel $(foldl g e0 $ sources)+        >=> $(sink) |]+      where+        e0 = [| (pure. pure) $(conE (tupleDataName $ length sources)) |]+        g acc x = [| ((<*>) . fmap (<*>)) $(acc) $ fmap Parallel $(backTrack x) |]+    mkNodeVar = varE . mkName . T.unpack . fst . snd {-# INLINE mkWorkflow #-} -mkProc :: Serializable b => PID -> (a -> IO b) -> (Processor a b)+mkProc :: (BatchData' (IsList a b) a b, BatchData a b, DBData a, DBData b)+       => PID -> (a -> IO b) -> (Processor a b) mkProc pid f = \input -> do-    st <- get-    case M.findWithDefault Scheduled pid (st^.procStatus) of-        Fail ex -> lift $ throwE ex-        Success -> do-            r <- liftIO $ readData pid $ st^.db-            return r+    wfState <- get+    let (pSt, attr) = M.findWithDefault (error "Impossible") pid $ wfState^.procStatus++    pStValue <- liftIO $ takeMVar pSt+    case pStValue of+        (Fail ex) -> liftIO (putMVar pSt pStValue) >> lift (throwE (pid, ex))+        Success -> liftIO $ do+            putMVar pSt pStValue++#ifdef DEBUG+            debug $ printf "Recovering saved node: %s" pid+#endif++            readData pid $ wfState^.db         Scheduled -> do+            _ <- liftIO $ takeMVar $ wfState^.procParaControl+ #ifdef DEBUG-            traceM $ "Running node: " ++ T.unpack pid+            debug $ printf "Running node: %s" pid #endif -            result <- liftIO $ try $ f input+            let sendToRemote = fromMaybe (wfState^.remote) (attr^.submitToRemote)+            result <- liftIO $ try $ case () of+                _ | attr^.batch > 0 -> do+                    let (mkBatch, combineResult) = batchFunction f $ attr^.batch+                        input' = mkBatch input+                    combineResult <$> if sendToRemote+                        then mapConcurrently (runRemote defaultRemoteOpts pid) input'+                        else mapM f input'+                  | otherwise -> if sendToRemote+                      then runRemote defaultRemoteOpts pid input+                      else 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+                    _ <- liftIO $ do+                        putMVar pSt $ Fail ex+                        forkIO $ putMVar (wfState^.procParaControl) ()+                    lift (throwE (pid, ex))+                Right r -> liftIO $ do+                    saveData pid r $ wfState^.db+                    putMVar pSt Success+                    _ <- forkIO $ putMVar (wfState^.procParaControl) ()                     return r {-# INLINE mkProc #-}++++--------------------------------------------------------------------------------++newtype Parallel a = Parallel { runParallel :: ProcState a}++instance Functor Parallel where+    fmap f (Parallel a) = Parallel $ f <$> a++instance Applicative Parallel where+    pure = Parallel . pure+    Parallel fs <*> Parallel as = Parallel $+        (\(f, a) -> f a) <$> concurrently fs as

@@ -1,40 +1,94 @@ {-# LANGUAGE OverloadedStrings #-} module Scientific.Workflow.DB     ( openDB+    , closeDB     , readData+    , readDataByteString+    , saveDataByteString     , saveData+    , updateData+    , delRecord     , isFinished     , getKeys     ) where -import Scientific.Workflow.Types-import           Shelly              (fromText, lsT, shelly, test_f, mkdir_p)-import qualified Data.ByteString     as B-import qualified Data.Text           as T+import qualified Data.ByteString           as B+import qualified Data.Text                 as T+import           Database.SQLite.Simple+import           Scientific.Workflow.Types+import           Text.Printf               (printf) +dbTableName :: String+dbTableName = "SciFlowDB"++createTable :: Connection -> String -> IO ()+createTable db tablename =+    execute_ db $ Query $ T.pack $ printf+        "CREATE TABLE %s(pid TEXT PRIMARY KEY, data BLOB)" tablename++hasTable :: Connection -> String -> IO Bool+hasTable db tablename = do+    r <- query db+        "SELECT name FROM sqlite_master WHERE type='table' AND name=?" [tablename]+    return $ not $ null (r :: [Only T.Text])+ openDB :: FilePath -> IO WorkflowDB-openDB dir = do-    shelly $ mkdir_p $ fromText $ T.pack dir-    return $ WorkflowDB dir+openDB dbFile = do+    db <- open dbFile+    exist <- hasTable db dbTableName+    if exist+        then return $ WorkflowDB db+        else do+            createTable db dbTableName+            return $ WorkflowDB db {-# INLINE openDB #-} -readData :: Serializable r => PID -> WorkflowDB -> IO r-readData p (WorkflowDB dir) = deserialize <$> B.readFile fl-  where fl = dir ++ "/" ++ T.unpack p+closeDB :: WorkflowDB -> IO ()+closeDB (WorkflowDB db) = close db+{-# INLINE closeDB #-}++readData :: DBData r => PID -> WorkflowDB -> IO r+readData pid (WorkflowDB db) = do+    [Only result] <- query db (Query $ T.pack $+        printf "SELECT data FROM %s WHERE pid=?" dbTableName) [pid]+    return $ deserialize result {-# INLINE readData #-} -saveData :: Serializable r => PID -> r -> WorkflowDB -> IO ()-saveData p result (WorkflowDB dir) = B.writeFile fl $ serialize result-  where fl = dir ++ "/" ++ T.unpack p+readDataByteString :: PID -> WorkflowDB -> IO B.ByteString+readDataByteString pid (WorkflowDB db) = do+    [Only result] <- query db (Query $ T.pack $+        printf "SELECT data FROM %s WHERE pid=?" dbTableName) [pid]+    return result+{-# INLINE readDataByteString #-}++saveData :: DBData r => PID -> r -> WorkflowDB -> IO ()+saveData pid result (WorkflowDB db) = execute db (Query $ T.pack $+    printf "INSERT INTO %s VALUES (?, ?)" dbTableName) (pid, serialize result) {-# INLINE saveData #-} +updateData :: DBData r => PID -> r -> WorkflowDB -> IO ()+updateData pid result (WorkflowDB db) = execute db (Query $ T.pack $+    printf "UPDATE %s SET data=? WHERE pid=?" dbTableName) (serialize result, pid)+{-# INLINE updateData #-}++saveDataByteString :: PID -> B.ByteString -> WorkflowDB -> IO ()+saveDataByteString pid result (WorkflowDB db) = execute db (Query $ T.pack $+    printf "INSERT INTO %s VALUES (?, ?)" dbTableName) (pid, result)+{-# INLINE saveDataByteString #-}+ isFinished :: PID -> WorkflowDB -> IO Bool-isFinished p (WorkflowDB dir) = shelly $ test_f $ fromText $ T.pack fl-  where fl = dir ++ "/" ++ T.unpack p+isFinished pid (WorkflowDB db) = do+    result <- query db (Query $ T.pack $+        printf "SELECT pid FROM %s WHERE pid = ?" dbTableName) [pid]+    return $ not $ null (result :: [Only T.Text]) {-# INLINE isFinished #-}  getKeys :: WorkflowDB -> IO [PID]-getKeys (WorkflowDB dir) = f <$> shelly (lsT $ fromText $ T.pack dir)-  where-    f = map (snd . T.breakOnEnd "/")+getKeys (WorkflowDB db) = concat <$> query_ db (Query $ T.pack $+    printf "SELECT pid FROM %s;" dbTableName) {-# INLINE getKeys #-}++delRecord :: PID -> WorkflowDB -> IO ()+delRecord pid (WorkflowDB db) =+    execute db (Query $ T.pack $ printf+        "DELETE FROM %s WHERE pid = ?" dbTableName) [pid]

@@ -0,0 +1,239 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell   #-}++module Scientific.Workflow.Main+    ( defaultMain+    , defaultMainOpts+    , mainWith+    , MainOpts(..)+    ) where++import           Control.Monad                     (forM_)+import qualified Data.ByteString.Char8             as B+import           Data.Graph.Inductive.Graph        (nmap)+import           Data.Graph.Inductive.PatriciaTree (Gr)+import qualified Data.Map                          as M+import qualified Data.Text                         as T+import qualified Data.Text.Lazy.IO                 as T+import           Language.Haskell.TH+import qualified Language.Haskell.TH.Lift          as T+import           Options.Applicative               hiding (runParser)+import           Shelly                            (fromText, lsT, mkdir_p,+                                                    rm_f, shelly)+import           Text.Printf                       (printf)++import           Scientific.Workflow+import           Scientific.Workflow.DB+import           Scientific.Workflow.Visualize++import           Data.Version                      (showVersion)+import           Paths_SciFlow                     (version)+++data CMD = Run GlobalOpts Int Bool+         | View+         | Cat GlobalOpts String+         | Write GlobalOpts String FilePath+         | Delete GlobalOpts String+         | Recover GlobalOpts FilePath+         | DumpDB GlobalOpts FilePath+         | Call String String String++data GlobalOpts = GlobalOpts+    { dbPath :: FilePath }++globalParser :: Parser GlobalOpts+globalParser = GlobalOpts+           <$> strOption+               ( long "db-path"+              <> value "sciflow.db"+              <> metavar "DB_PATH" )++runParser :: Parser CMD+runParser = Run+    <$> globalParser+    <*> option auto+        ( short 'N'+       <> value 1+       <> metavar "CORES"+       <> help "The number of concurrent processes." )+    <*> switch+        ( long "remote"+       <> help "Submit jobs to remote machines.")+runExe initialize (Run opts n r) wf+    | r = initialize $ runWorkflow wf $ RunOpt (dbPath opts) n True+    | otherwise = runWorkflow wf $ RunOpt (dbPath opts) n False+runExe _ _ _ = undefined+{-# INLINE runExe #-}++viewParser :: Parser CMD+viewParser = pure View+viewExe = T.putStrLn . renderBuilder+{-# INLINE viewExe #-}++catParser :: Parser CMD+catParser = Cat+        <$> globalParser+        <*> strArgument+            (metavar "NODE_ID")+catExe (Cat opts pid) (Workflow _ ft _) = do+    db <- openDB $ dbPath opts+    case M.lookup pid ft of+        Just (Closure fn) -> do+            dat <- head [readData (T.pack pid) db, fn undefined]+            B.putStr $ showYaml dat+        Nothing -> return ()+catExe _ _ = undefined+{-# INLINE catExe #-}++writeParser :: Parser CMD+writeParser = Write+          <$> globalParser+          <*> strArgument+              (metavar "NODE_ID")+          <*> strArgument+              (metavar "INPUT_FILE")+writeExe (Write opts pid input) (Workflow _ ft _) = do+    db <- openDB $ dbPath opts+    c <- B.readFile input+    case M.lookup pid ft of+        Just (Closure fn) -> do+            dat <- head [return $ readYaml c, fn undefined]+            updateData (T.pack pid) dat db+        Nothing -> return ()+writeExe _ _ = undefined+{-# INLINE writeExe #-}++rmParser :: Parser CMD+rmParser = Delete+       <$> globalParser+       <*> strArgument+           (metavar "NODE_ID")+rmExe (Delete opts pid) = do+    db <- openDB $ dbPath opts+    delRecord (T.pack pid) db+rmExe _ = undefined+{-# INLINE rmExe #-}++recoverParser :: Parser CMD+recoverParser = Recover+            <$> globalParser+            <*> strArgument+                (metavar "BACKUP")+recoverExe (Recover opts dir) (Workflow _ ft _) = do+    fls <- shelly $ lsT $ fromText $ T.pack dir+    shelly $ rm_f $ fromText $ T.pack $ dbPath opts+    db <- openDB $ dbPath opts+    forM_ fls $ \fl -> do+        let pid = snd $ T.breakOnEnd "/" fl+        case M.lookup (T.unpack pid) ft of+            Just (Closure fn) -> do+                printf "Recovering node: %s.\n" pid+                c <- B.readFile $ T.unpack fl+                dat <- head [return $ readYaml c, fn undefined]+                saveData pid dat db+            Nothing -> printf "Cannot identify node: %s. Skipped.\n" pid+recoverExe _ _ = undefined+{-# INLINE recoverExe #-}++dumpDBParser :: Parser CMD+dumpDBParser = DumpDB+           <$> globalParser+           <*> strArgument+               (metavar "OUTPUT_DIR")+dumpDBExe (DumpDB opts dir) (Workflow _ ft _) = do+    shelly $ mkdir_p $ fromText $ T.pack dir+    db <- openDB $ dbPath opts+    nodes <- getKeys db+    forM_ nodes $ \pid -> do+        let fl = dir ++ "/" ++ T.unpack pid+        case M.lookup (T.unpack pid) ft of+            Just (Closure fn) -> do+                printf "Saving node: %s.\n" pid+                dat <- head [readData pid db, fn undefined]+                B.writeFile fl $ showYaml dat+            Nothing -> return ()+dumpDBExe _ _ = undefined+{-# INLINE dumpDBExe #-}++callParser :: Parser CMD+callParser = Call+         <$> strArgument mempty+         <*> strArgument mempty+         <*> strArgument mempty+callExe (Call pid inputFl outputFl) (Workflow _ ft _) = case M.lookup pid ft of+    Just (Closure fn) -> do+        input <- deserialize <$> B.readFile inputFl+        output <- serialize <$> fn input+        B.writeFile outputFl output+    Nothing -> undefined+callExe _ _ = undefined+{-# INLINE callExe #-}+++mainFunc :: (IO () -> IO ()) -- initialization function+         -> Gr (PID, Attribute) Int -> Workflow+         -> String  -- program header+         -> IO ()+mainFunc initialize dag wf h = execParser opts >>= execute+  where+    execute cmd@(Run _ _ _) = runExe initialize cmd wf+    execute View = viewExe dag+    execute cmd@(Cat _ _) = catExe cmd wf+    execute cmd@(Write _ _ _) = writeExe cmd wf+    execute cmd@(Delete _ _) = rmExe cmd+    execute cmd@(Recover _ _) = recoverExe cmd wf+    execute cmd@(DumpDB _ _) = dumpDBExe cmd wf+    execute cmd@(Call _ _ _) = callExe cmd wf++    opts = info (helper <*> parser) $ fullDesc <> header h+    parser = subparser $ (+        command "run" (info (helper <*> runParser) $+            fullDesc <> progDesc "run workflow")+     <> command "view" (info (helper <*> viewParser) $+            fullDesc <> progDesc "view workflow")+     <> command "cat" (info (helper <*> catParser) $+            fullDesc <> progDesc "display the result of a node")+     <> command "write" (info (helper <*> writeParser) $+            fullDesc <> progDesc "write the result to a node")+     <> command "rm" (info (helper <*> rmParser) $+            fullDesc <> progDesc "delete the result of a node.")+     <> command "recover" (info (helper <*> recoverParser) $+            fullDesc <> progDesc "Recover database from backup.")+     <> command "backup" (info (helper <*> dumpDBParser) $+            fullDesc <> progDesc "Backup database.")+     <> command "execFunc" (info (helper <*> callParser) $+            fullDesc <> progDesc "Do not call this directly.")+     )+++data MainOpts = MainOpts+    { preAction :: Name    -- ^ An action to be execute before the workflow. The+                           -- action should have type: IO () -> IO ().+                            -- ^ i.e., some initialization processes.+    , programHeader :: String+    }++T.deriveLift ''MainOpts++defaultMainOpts :: MainOpts+defaultMainOpts = MainOpts+    { preAction = 'id+    , programHeader = printf "SciFlow-%s" (showVersion version)+    }++defaultMain :: Builder () -> Q [Dec]+defaultMain = mainWith defaultMainOpts++mainWith :: MainOpts -> Builder () -> Q [Dec]+mainWith opts builder = do+    wf_q <- buildWorkflow wfName builder+    main_q <- [d| main = mainFunc $(varE $ preAction opts) dag+                    $(varE $ mkName wfName) (programHeader opts)+              |]+    return $ wf_q ++ main_q+  where+    wfName = "sciFlowDefaultMain"+    dag = nmap (\(a,(_,b)) -> (a,b)) $ mkDAG builder+{-# INLINE mainWith #-}

@@ -1,92 +1,144 @@-{-# LANGUAGE GADTs #-}-{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleInstances    #-}+{-# LANGUAGE OverloadedStrings    #-}+{-# LANGUAGE TemplateHaskell      #-} {-# LANGUAGE UndecidableInstances #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE MultiParamTypeClasses #-}  module Scientific.Workflow.Types     ( WorkflowDB(..)     , Workflow(..)+    , Closure(..)     , PID-    , ProcState(..)+    , NodeResult(..)+    , ProcState     , WorkflowState(..)     , db     , procStatus+    , procParaControl+    , remote     , Processor     , RunOpt(..)-    , defaultRunOpt-    , dbPath-    , Serializable(..)-    , Attribute+    , BatchData(..)+    , BatchData'(..)+    , IsList+    , DBData(..)+    , Attribute(..)+    , AttributeSetter     , defaultAttribute     , label     , note-    , def+    , batch+    , submitToRemote     ) where -import Control.Lens (makeLenses)+import qualified Data.Serialize as S+import           Control.Concurrent.MVar+import           Control.Exception          (SomeException)+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-import Data.Maybe (fromJust)-import Data.Yaml (FromJSON, ToJSON, encode, decode)--class Serializable a where-    serialize :: a -> B.ByteString-    deserialize :: B.ByteString -> a+import           Control.Monad.Trans.Except (ExceptT)+import qualified Data.ByteString            as B+import qualified Data.Map                   as M+import           Data.Maybe                 (fromJust)+import qualified Data.Text                  as T+import           Data.Yaml                  (FromJSON, ToJSON, decode, encode)+import           Database.SQLite.Simple     (Connection)+import Data.List.Split (chunksOf) -instance (FromJSON a, ToJSON a) => Serializable a where-    serialize = encode-    deserialize = fromJust . decode+data HTrue+data HFalse --- | An abstract type representing the database used to store states of workflow-data WorkflowDB  = WorkflowDB FilePath+type family IsList a b where+    IsList [a] [b] = HTrue+    IsList a b = HFalse --- | The id of a node-type PID = T.Text+class BatchData' flag a b where+    batchFunction' :: flag -> (a -> IO b) -> Int -> (a -> [a], [b] -> b) --- | The state of a computation node-data ProcState = Success-               | Scheduled-               | Fail SomeException+instance BatchData' HTrue [a] [b] where+    batchFunction' _ _ i = (chunksOf i, concat) -data WorkflowState = WorkflowState-    { _db         :: WorkflowDB-    , _procStatus :: M.Map PID ProcState-    }+instance BatchData' HFalse a b where+    batchFunction' _ _ _ = (return, head) -makeLenses ''WorkflowState+class BatchData a b where+    batchFunction :: (a -> IO b) -> Int -> (a -> [a], [b] -> b) -type Processor a b = a -> StateT WorkflowState (ExceptT SomeException IO) b+instance (IsList a b ~ flag, BatchData' flag a b) => BatchData a b where+    batchFunction = batchFunction' (undefined :: flag) -data Workflow where-    Workflow :: (Processor () o) -> Workflow+class DBData a where+    serialize :: a -> B.ByteString+    deserialize :: B.ByteString -> a+    showYaml :: a -> B.ByteString+    readYaml :: B.ByteString -> a -data RunOpt = RunOpt-    { _dbPath :: FilePath-    }+instance (FromJSON a, ToJSON a, S.Serialize a) => DBData a where+    serialize = S.encode+    deserialize = fromEither . S.decode+      where+        fromEither (Right x) = x+        fromEither _ = error "decode failed"+    showYaml = encode+    readYaml = fromJust . decode -makeLenses ''RunOpt+-- | An abstract type representing the database used to store states of workflow+newtype WorkflowDB  = WorkflowDB Connection -defaultRunOpt :: RunOpt-defaultRunOpt = RunOpt-    { _dbPath = "wfDB" }+-- | The id of a node+type PID = T.Text +-- | Node attribute data Attribute = Attribute     { _label :: T.Text  -- ^ short description-    , _note :: T.Text   -- ^ long description+    , _note  :: T.Text   -- ^ long description+    , _batch :: Int+    , _submitToRemote :: Maybe Bool  -- ^ overwrite the global option     } +makeLenses ''Attribute+ defaultAttribute :: Attribute defaultAttribute = Attribute     { _label = ""     , _note = ""+    , _batch = -1+    , _submitToRemote = Nothing     } -makeLenses ''Attribute+type AttributeSetter = State Attribute () -def :: State a ()-def = return ()+-- | The result of a computation node+data NodeResult = Success+                | Fail SomeException+                | Scheduled++data WorkflowState = WorkflowState+    { _db          :: WorkflowDB+    , _procStatus  :: M.Map PID (MVar NodeResult, Attribute)+    , _procParaControl :: MVar () -- ^ concurrency controller+    , _remote :: Bool+    }++makeLenses ''WorkflowState++type ProcState b = StateT WorkflowState (ExceptT (PID, SomeException) IO) b+type Processor a b = a -> ProcState b+++data Closure where+    Closure :: (DBData a, DBData b) => (a -> IO b) -> Closure++-- | A Workflow is a DAG+data Workflow = Workflow (M.Map T.Text Attribute)+                         (M.Map String Closure)+                         (Processor () ())++data RunOpt = RunOpt+    { database :: FilePath+    , nThread :: Int      -- ^ number of concurrent processes+    , runOnRemote :: Bool+    }

@@ -0,0 +1,56 @@+{-# LANGUAGE CPP #-}+module Scientific.Workflow.Utils where++import qualified Data.ByteString.Char8         as B+import qualified Data.Text                     as T+import           Debug.Trace                   (traceM)+import           Rainbow+import           System.IO++import           Scientific.Workflow.Types     (DBData (..))+import           System.Directory              (getCurrentDirectory)+import           System.Environment.Executable (getExecutablePath)++#ifdef SGE+import           DRMAA                         (DrmaaAttribute (..),+                                                defaultDrmaaConfig, drmaaRun,+                                                withTmpFile)+#endif++debug :: Monad m => String -> m ()+debug txt = traceM $ B.unpack $ B.concat $+    chunksToByteStrings toByteStringsColors8 [prefix, chunk txt & fore green]+  where+    prefix = bold $ chunk "[DEBUG] " & fore green++error' :: String -> IO ()+error' txt = B.hPutStrLn stderr $ B.concat $+    chunksToByteStrings toByteStringsColors8 [prefix, chunk txt & fore red]+  where+    prefix = bold $ chunk "[ERROR] " & fore red++data RemoteOpts = RemoteOpts+    { extraParams :: String+    }++defaultRemoteOpts :: RemoteOpts+defaultRemoteOpts = RemoteOpts+    { extraParams = ""+    }++runRemote :: (DBData a, DBData b) => RemoteOpts -> T.Text -> a -> IO b+#ifdef SGE+runRemote opts pid input = withTmpFile tmpDir $ \inputFl -> withTmpFile tmpDir $+    \outputFl -> do+        exePath <- getExecutablePath+        wd <- getCurrentDirectory+        let config = defaultDrmaaConfig{drmaa_wd=wd, drmaa_native=extraParams opts}++        B.writeFile inputFl $ serialize input+        drmaaRun exePath ["execFunc", T.unpack pid, inputFl, outputFl] config :: IO ()+        deserialize <$> B.readFile outputFl+  where+    tmpDir = "./"+#else+runRemote = error "SGE support was not turned on."+#endif

@@ -8,20 +8,20 @@ import qualified Data.Text as T import qualified Data.Text.Lazy      as TL -import Data.GraphViz-import Data.GraphViz.Printing-import Data.GraphViz.Attributes.Complete+import qualified Data.GraphViz as G+import qualified Data.GraphViz.Printing as G+import qualified Data.GraphViz.Attributes.Complete as G+import Data.Graph.Inductive.PatriciaTree (Gr) -import Scientific.Workflow.Types (note)+import Scientific.Workflow.Types import Scientific.Workflow.Builder  -- | Print the computation graph-renderBuilder :: Builder () -> TL.Text-renderBuilder b = renderDot . toDot $ graphToDot param dag+renderBuilder :: Gr (PID, Attribute) Int -> TL.Text+renderBuilder dag = G.renderDot . G.toDot $ G.graphToDot param dag   where-    fmtnode (_, (p, (_, attr))) = [Label $ StrLabel $ TL.fromStrict lab]+    fmtnode (_, (p, attr)) = [G.Label $ G.StrLabel $ TL.fromStrict lab]       where         lab | T.null (attr^.label) = p             | otherwise = attr^.label-    dag = mkDAG b-    param = nonClusteredParams{fmtNode = fmtnode}+    param = G.nonClusteredParams{G.fmtNode = fmtnode}