packages feed

SciFlow 0.5.3.1 → 0.6.0

raw patch · 16 files changed

+1264/−938 lines, 16 filesdep +aesondep +cereal-textdep +data-default-classdep −shellydep ~basedep ~drmaadep ~optparse-applicative

Dependencies added: aeson, cereal-text, data-default-class, network, temporary, th-lift-instances, time

Dependencies removed: shelly

Dependency ranges changed: base, drmaa, optparse-applicative

Files

LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2015 Kai Zhang+Copyright (c) 2015-2016 Kai Zhang  Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the
README.md view
@@ -4,99 +4,42 @@ Introduction ------------ -SciFlow is a workflow management system for working with big data pipelines locally-or in a grid computing system.+SciFlow is a DSL for building scientific workflows. Workflows built with SciFlow+can be run either on normal desktops or in grid computing environments that+support DRMAA.  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-----------------------------------------------------{-# 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+Features+-------- -cleanUp :: (Bool, FilePath) -> IO ()-cleanUp (toBeRemoved, fl) = if toBeRemoved-    then shelly $ rm $ fromText $ T.pack fl-    else return ()+1. Easy to use and safe: Provide a simple and flexible way to design type safe+computational pipelines in Haskell. --- 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"+2. Automatic Checkpointing: The states of intermediate steps are automatically+logged, allowing easy restart upon failures. -    ["step0"] ~> "step1"-    ["step0"] ~> "step2"-    ["step1", "step2"] ~> "step3"-    ["step3", "step0"] ~> "step4"+3. Parallelism and grid computing support. ------------------------------------------------------- File 2: main.hs-----------------------------------------------------{-# LANGUAGE TemplateHaskell #-}+Examples+-------- -import qualified Functions as F-import Scientific.Workflow.Main+See examples in the "examples" directory for more details. -defaultMain F.builder-```+Use `ghc main.hs -threaded` to compile the examples.+And type `./main --help` to see available commands. -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.+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. -![example](example.png)+To enable grid compute engine support, you need to have DRMAA C library+installed and compile the SciFlow with `-f drmaa` flag.+Use `./main run --remote` to submit jobs to remote machines. -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.+Featured applications+-------------------- -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.+[Here](https://github.com/Taiji-pipeline)+are some bioinformatics pipelines built with SciFlow.
SciFlow.cabal view
@@ -1,58 +1,55 @@ name:                SciFlow-version:             0.5.3.1+version:             0.6.0 synopsis:            Scientific workflow management system-description:-  SciFlow is to help programmers design complex workflows-  with ease.-+description:         SciFlow is a DSL for building scientific workflows.+                     Workflows built with SciFlow can be run either on desktop+                     computers or in grid computing environments that+                     support DRMAA. license:             MIT license-file:        LICENSE author:              Kai Zhang maintainer:          kai@kzhang.org-copyright:           (c) 2016 Kai Zhang+copyright:           (c) 2015-2017 Kai Zhang category:            Control build-type:          Simple cabal-version:       >=1.10  extra-source-files:   README.md-  example.png -Flag debug-  Description: Enable debug support-  Default:     True--Flag sge-  Description: Enable SGE support+Flag drmaa+  Description: Enable DRMAA integration   Default:     False  library   ghc-options: -Wall   exposed-modules:     Scientific.Workflow-    Scientific.Workflow.Builder-    Scientific.Workflow.DB     Scientific.Workflow.Main+    Scientific.Workflow.Main.Options     Scientific.Workflow.Types-    Scientific.Workflow.Utils     Scientific.Workflow.Visualize+    Scientific.Workflow.Internal.Builder+    Scientific.Workflow.Internal.Builder.Types+    Scientific.Workflow.Internal.DB+    Scientific.Workflow.Internal.Utils    other-modules:     Paths_SciFlow -  if flag(debug)-    CPP-Options: -DDEBUG--  if flag(sge)-    CPP-Options: -DSGE-    build-depends: drmaa+  if flag(drmaa)+    CPP-Options: -DDRMAA_ENABLED+    build-depends: drmaa >=0.2.0    build-depends:-      base >=4.0 && <5.0+      base >=4.7 && <5.0     , bytestring+    , aeson     , containers     , cereal+    , cereal-text     , directory+    , data-default-class     , exceptions     , executable-path     , fgl@@ -60,12 +57,15 @@     , lens >=4.0     , lifted-async     , mtl-    , optparse-applicative+    , network+    , optparse-applicative >=0.13.0.0     , rainbow-    , shelly     , sqlite-simple     , split     , th-lift+    , th-lift-instances+    , time+    , temporary     , text     , template-haskell     , transformers
− example.png

binary file changed (26711 → absent bytes)

src/Scientific/Workflow.hs view
@@ -1,56 +1,76 @@-module Scientific.Workflow-    ( runWorkflow-    , module Scientific.Workflow.Builder-    , module Scientific.Workflow.Types-    ) where+{-|+Module      : Scientific.Workflow+Description : Building type safe scientific workflows+Copyright   : (c) 2015-2017 Kai Zhang+License     : MIT+Maintainer  : kai@kzhang.org+Stability   : experimental+Portability : POSIX -import           Control.Concurrent          (forkIO)-import           Control.Concurrent.MVar-import           Control.Exception           (bracket, displayException)-import           Control.Monad.State-import           Control.Monad.Trans.Except-import qualified Data.Map                    as M-import qualified Data.Set                    as S-import Data.Yaml (decodeFile)+SciFlow is a DSL for building scientific workflows. Workflows built with SciFlow+can be run either on desktop computers or in grid computing environments that+support DRMAA. -import           Scientific.Workflow.Builder-import           Scientific.Workflow.DB-import           Scientific.Workflow.Types-import           Scientific.Workflow.Utils-import           Text.Printf                 (printf)+Features: -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 -> case runMode opts of-        Normal -> do-            v <- if pid `S.member` ks then newMVar Success else newMVar Scheduled-            return (v, attr)-        ExecSingle i input output -> do-            v <- if pid == i then newMVar (EXE input output) else newMVar Skip-            return (v, attr)-        ReadSingle i -> do-            v <- if pid == i then newMVar Read else newMVar Skip-            return (v, attr)-        WriteSingle i input -> do-            v <- if pid == i then newMVar (Replace input) else newMVar Skip-            return (v, attr)+1. Easy to use and safe: Provide a simple and flexible way to design type safe+computational pipelines in Haskell. -    para <- newEmptyMVar-    _ <- forkIO $ replicateM_ (nThread opts) $ putMVar para ()+2. Automatic Checkpointing: The states of intermediate steps are automatically+logged, allowing easy restart upon failures. -    env <- case configuration opts of-        Nothing -> return M.empty-        Just fl -> do-            r <- decodeFile fl-            case r of-                Nothing -> error "fail to parse configuration file"-                Just x -> return x+3. Parallelism and grid computing support. -    let initState = WorkflowState db pidStateMap para (runOnRemote opts) env+Example: -    result <- runExceptT $ evalStateT (wf ()) initState-    case result of-        Right _ -> return ()-        Left (pid, ex) -> error' $ printf "\"%s\" failed. The error was: %s"-            pid (displayException ex)+> import           Control.Lens             ((.=))+> import           Scientific.Workflow+>+> f :: Int -> Int+> f = (+1)+>+> defaultMain $ do+>     nodeS "step0" [| return . const [1..10] :: () -> WorkflowConfig () [Int] |] $ return ()+>     nodeP' 2 "step1" 'f $ note .= "run in parallel with batch size 2"+>     nodeP' 4 "step2" 'f $ note .= "run in parallel with batch size 4"+>     node' "step3" [| \(x, y) -> x ++ y |] $ return ()+>+>     ["step0"] ~> "step1"+>     ["step0"] ~> "step2"+>     ["step1", "step2"] ~> "step3"++-}+module Scientific.Workflow+    ( defaultMain+    , mainWith+    , defaultMainOpts+    , MainOpts(..)++    , Builder+    , namespace+    , node+    , node'+    , nodeS+    , nodeP+    , nodeP'+    , nodePS+    , nodeSharedP+    , nodeSharedP'+    , nodeSharedPS+    , link+    , (~>)+    , path++    , label+    , note+    , submitToRemote+    , remoteParam++    , ContextData(..)+    , WorkflowConfig+    ) where++import           Scientific.Workflow.Internal.Builder+import           Scientific.Workflow.Internal.Builder.Types+import           Scientific.Workflow.Main+import           Scientific.Workflow.Types
− src/Scientific/Workflow/Builder.hs 
@@ -1,246 +0,0 @@-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE CPP #-}--module Scientific.Workflow.Builder-    ( node-    , link-    , (~>)-    , path-    , buildWorkflow-    , buildWorkflowPart-    , mkDAG-    ) where--import Control.Lens ((^.), (%~), _1, _2, _3)-import Control.Monad.Trans.Except (throwE)-import           Control.Monad.State (lift, liftIO, (>=>), foldM_, execState, modify, State, get)-import Control.Concurrent.MVar-import Control.Concurrent (forkIO)-import qualified Data.Text           as T-import Data.Graph.Inductive.Graph ( mkGraph, lab, labNodes, outdeg-                                  , lpre, labnfilter, nfilter, gmap, suc )-import Data.Graph.Inductive.PatriciaTree (Gr)-import Data.List (sortBy, foldl')-import Data.Maybe (fromJust, fromMaybe)-import qualified Data.ByteString as B-import Data.Ord (comparing)-import qualified Data.Map as M-import Text.Printf (printf)-import Control.Concurrent.Async.Lifted (mapConcurrently)-import           Language.Haskell.TH-import Control.Monad.Catch (try)--import Scientific.Workflow.Types-import Scientific.Workflow.DB-import Scientific.Workflow.Utils (debug, runRemote, defaultRemoteOpts, RemoteOpts(..))---- | 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-     -> State Attribute ()   -- ^ Attribues-     -> Builder ()-node p fn setAttr = modify $ _1 %~ (newNode:)-  where-    attr = execState setAttr defaultAttribute-    newNode = (p, (toExpQ fn, attr))-{-# INLINE node #-}---- | many-to-one generalized link function-link :: [PID] -> PID -> Builder ()-link xs t = modify $ _2 %~ (zip3 xs (repeat t) [0..] ++)-{-# INLINE link #-}---- | (~>) = link.-(~>) :: [PID] -> PID -> Builder ()-(~>) = link-{-# INLINE (~>) #-}---- | singleton-path :: [PID] -> Builder ()-path ns = foldM_ f (head ns) $ tail ns-  where-    f a t = link [a] t >> return t-{-# INLINE path #-}---- | 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 $name$.--- Lastly, a function table will be created with the name $name$_function_table.-buildWorkflow :: String     -- ^ name-              -> Builder ()-              -> Q [Dec]-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 :: FilePath   -- ^ path to the db-                  -> String-                  -> Builder ()-                  -> Q [Dec]-buildWorkflowPart db wfName b = do-    st <- runIO $ getWorkflowState db-    mkWorkflow wfName $ trimDAG st $ mkDAG b-  where-    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-    toExpQ :: a -> ExpQ--instance ToExpQ Name where-    toExpQ = varE--instance ToExpQ ExpQ where-    toExpQ = id--type DAG = Gr Node EdgeOrd---- TODO: check the graph is a valid DAG--- | 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 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 :: (M.Map T.Text NodeResult) -> DAG -> DAG-trimDAG st dag = gmap revise gr-  where-    revise context@(linkTo, _, lab, _)-        | done (fst lab) && null linkTo = _3._2._1 %~ e $ context-        | otherwise = context-      where-        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 of-        Just Success -> True-        _ -> False-{-# INLINE trimDAG #-}----- 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 $name$_function_table.-mkWorkflow :: String   -- name-           -> DAG -> Q [Dec]-mkWorkflow workflowName dag = do-    let expq = connect sinks [| const $ return () |]-    -- define the workflow-    workflows <--        [d| $(varP $ mkName workflowName) = Workflow pids $expq |]--    return workflows-  where-    computeNodes = snd $ unzip $ labNodes dag-    pids = M.fromList $ map (\(i, x) -> (i, snd x)) computeNodes-    sinks = labNodes $ nfilter ((==0) . outdeg dag) dag--    backTrack (i, (p, (fn, attr)))-        | attr^.stateful = connect (fst $ unzip parents) [| mkProc p $fn |]-        | otherwise = connect (fst $ unzip parents) [| mkProc p (liftIO . $fn) |]-      where-        parents = map ( \(x, o) -> ((x, fromJust $ lab dag x), o) ) $-            sortBy (comparing snd) $ lpre dag i--    connect [] sink = sink-    connect [source] sink = [| $expq >=> $sink |]-      where-        expq = backTrack source-    connect sources sink = [| fmap runParallel $expq >=> $sink |]-      where-        expq = foldl' g e0 $ sources-        e0 = [| (pure. pure) $(conE (tupleDataName $ length sources)) |]-        g acc x =-            let expq = backTrack x-            in [| ((<*>) . fmap (<*>)) $acc $ fmap Parallel $expq |]-{-# INLINE mkWorkflow #-}--mkProc :: (BatchData' (IsList a b) a b, BatchData a b, DBData a, DBData b)-       => PID -> (a -> ProcState b) -> (Processor a b)-mkProc pid f = \input -> do-    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-            debug $ printf "Running node: %s" pid-#endif--            let sendToRemote = fromMaybe (wfState^.remote) (attr^.submitToRemote)-                remoteOpts = defaultRemoteOpts-                    { extraParams = attr^.remoteParam-                    , environment = wfState^.config-                    }-            result <- try $ case () of-                _ | attr^.batch > 0 -> do-                    let (mkBatch, combineResult) = batchFunction f $ attr^.batch-                        input' = mkBatch input-                    combineResult <$> if sendToRemote-                        then liftIO $ mapConcurrently (runRemote remoteOpts pid) input'-                        else mapM f input'  -- do not run in parallel in local machine-                  | otherwise -> if sendToRemote-                      then liftIO $ runRemote remoteOpts pid input-                      else f input-            case result of-                Left ex -> do-                    _ <- 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-        Skip -> liftIO $ putMVar pSt pStValue >> return undefined-        EXE input output -> do-            c <- liftIO $ B.readFile input-            r <- f $ deserialize c-            liftIO $ B.writeFile output $ serialize r-            liftIO $ putMVar pSt Skip-            return undefined-        Read -> liftIO $ do-            r <- readData pid $ wfState^.db-            B.putStr $ showYaml r-            putMVar pSt Skip-            return r-        Replace input -> do-            c <- liftIO $ B.readFile input-            r <- return (readYaml c) `asTypeOf` f undefined-            liftIO $ updateData pid r $ wfState^.db-            liftIO $ putMVar pSt Skip-            return r-{-# INLINE mkProc #-}
− src/Scientific/Workflow/DB.hs 
@@ -1,94 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-module Scientific.Workflow.DB-    ( openDB-    , closeDB-    , readData-    , readDataByteString-    , saveDataByteString-    , saveData-    , updateData-    , delRecord-    , isFinished-    , getKeys-    ) where--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 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 #-}--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 #-}--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 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 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]
+ src/Scientific/Workflow/Internal/Builder.hs view
@@ -0,0 +1,367 @@+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RecordWildCards #-}++module Scientific.Workflow.Internal.Builder+    ( node+    , node'+    , nodeS+    , nodeP+    , nodeP'+    , nodePS+    , nodeSharedP+    , nodeSharedP'+    , nodeSharedPS+    , link+    , (~>)+    , path+    , namespace+    , buildWorkflow+    , buildWorkflowPart+    , mkDAG+    , mkProc+    ) where++import Control.Monad.Identity (runIdentity)+import Data.Monoid ((<>))+import Control.Lens ((^.), (%~), _1, _2, _3, (&))+import Control.Monad.Trans.Except (throwE)+import Control.Monad.State (lift, liftIO, (>=>), foldM_, execState, modify, State)+import Control.Monad.Reader (ask)+import Control.Concurrent.MVar+import Control.Concurrent (forkIO)+import qualified Data.Text           as T+import Data.List.Split (chunksOf)+import Data.Yaml (ToJSON)+import Data.Graph.Inductive.Graph ( mkGraph, lab, labNodes, outdeg, nmap+                                  , lpre, labnfilter, nfilter, gmap, suc )+import Data.List (sortBy, foldl')+import Data.Maybe (fromJust, fromMaybe)+import qualified Data.ByteString as B+import Data.Ord (comparing)+import qualified Data.Map as M+import Control.Concurrent.Async.Lifted (mapConcurrently)+import           Language.Haskell.TH+import Control.Monad.Catch (try)++import Scientific.Workflow.Types+import Scientific.Workflow.Internal.Builder.Types+import Scientific.Workflow.Internal.DB+import Scientific.Workflow.Internal.Utils (sendLog, Log(..), runRemote, RemoteOpts(..))++nodeWith :: ToExpQ function+         => FunctionConfig+         -> PID                  -- ^ node id+         -> function             -- ^ function+         -> State Attribute ()   -- ^ Attribues+         -> Builder ()+nodeWith conf pid fn setAttr = modify $ _1 %~ (newNode:)+  where+    attr = execState setAttr defaultAttribute{_functionConfig = conf}+    newNode = Node pid (toExpQ fn) attr+{-# INLINE nodeWith #-}++-- | Declare an IO computational step.+node :: ToExpQ fun+     => PID                  -- ^ Node id+     -> fun                  -- ^ Template Haskell expression representing+                             -- functions with type @a -> IO b@.+     -> State Attribute ()   -- ^ Attribues+     -> Builder ()+node = nodeWith $ FunctionConfig None IOAction+{-# INLINE node #-}++-- | Declare a pure computational step.+node' :: ToExpQ fun+      => PID+      -> fun -- ^ Template Haskell expression representing+             -- functions with type @a -> b@.+      -> State Attribute ()+      -> Builder ()+node' = nodeWith $ FunctionConfig None Pure+{-# INLINE node' #-}++-- | Declare a stateful computational step.+nodeS :: ToExpQ fun+      => PID+      -> fun  -- ^ Template Haskell expression representing+              -- functions with type "@a -> 'WorkflowConfig' st b@".+      -> State Attribute () -> Builder ()+nodeS = nodeWith $ FunctionConfig None Stateful+{-# INLINE nodeS #-}++-- | Declare an IO and parallel computational step. This will turn functions+-- with type "@a -> IO b@" into functions with type "@[a] -> IO [b]@". And+-- @[a]@ will be processed in parallel with provided batch size.+-- Note: Currently, parallelism is available only when @"--remote"@ flag+-- is on.+nodeP :: ToExpQ fun+      => Int       -- ^ Batch size for parallel execution.+      -> PID+      -> fun+      -> State Attribute () -> Builder ()+nodeP n = nodeWith $ FunctionConfig (Standard n) IOAction+{-# INLINE nodeP #-}++-- | Same as @'nodeP'@ but work with pure functions.+nodeP' :: ToExpQ fun => Int -> PID -> fun -> State Attribute () -> Builder ()+nodeP' n = nodeWith $ FunctionConfig (Standard n) Pure+{-# INLINE nodeP' #-}++-- | Same as @'nodeP'@ but work with stateful functions.+nodePS :: ToExpQ fun => Int -> PID -> fun -> State Attribute () -> Builder ()+nodePS n = nodeWith $ FunctionConfig (Standard n) Stateful+{-# INLINE nodePS #-}++-- | Similar to @'nodeP'@ but work with inputs that are associated with a+-- shared context. Turn @'ContextData' context a -> 'IO' b@ into+-- @'ContextData' context [a] -> 'IO' [b]@.+nodeSharedP :: ToExpQ fun+            => Int+            -> PID+            -> fun   -- ^ Template Haskell expression representing+                     -- functions with type @'ContextData' context a -> 'IO' b@.+            -> State Attribute () -> Builder ()+nodeSharedP n = nodeWith $ FunctionConfig (ShareData n) IOAction+{-# INLINE nodeSharedP #-}++nodeSharedP' :: ToExpQ fun => Int -> PID -> fun -> State Attribute () -> Builder ()+nodeSharedP' n = nodeWith $ FunctionConfig (ShareData n) Pure+{-# INLINE nodeSharedP' #-}++nodeSharedPS :: ToExpQ fun => Int -> PID -> fun -> State Attribute () -> Builder ()+nodeSharedPS n = nodeWith $ FunctionConfig (ShareData n) Stateful+{-# INLINE nodeSharedPS #-}++-- | Declare the dependency between nodes.+-- Example:+--+-- > node' "step1" [| \() -> 1 :: Int |] $ return ()+-- > node' "step2" [| \() -> 2 :: Int |] $ return ()+-- > node' "step3" [| \(x, y) -> x * y |] $ return ()+-- > link ["step1", "step2"] "step3"+link :: [PID] -> PID -> Builder ()+link xs t = modify $ _2 %~ (zipWith3 Edge xs (repeat t) [0..] ++)+{-# INLINE link #-}++-- | @(~>) = 'link'@.+(~>) :: [PID] -> PID -> Builder ()+(~>) = link+{-# INLINE (~>) #-}++-- | "@'path' [a, b, c]@" is equivalent to "@'link' a b >> 'link' b c@"+path :: [PID] -> Builder ()+path ns = foldM_ f (head ns) $ tail ns+  where+    f a t = link [a] t >> return t+{-# INLINE path #-}++-- | Add a prefix to IDs of nodes for a given builder, i.e.,+-- @id@ becomes @prefix_id@.+namespace :: T.Text -> Builder () -> Builder ()+namespace prefix builder = builder >> addPrefix+  where+    addPrefix = modify $ \(nodes, edges) ->+        ( map (\x -> x{_nodePid = prefix <> "_" <> _nodePid x}) nodes+        , map (\x -> x{ _edgeFrom = prefix <> "_" <> _edgeFrom x+                      , _edgeTo = prefix <> "_" <> _edgeTo x }) edges )++-- | 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 $name$.+buildWorkflow :: String     -- ^ Name of the workflow+              -> Builder () -- ^ Builder+              -> Q [Dec]+buildWorkflow workflowName = mkWorkflow workflowName . mkDAG++-- | Build only a part of the workflow that has not been executed. This is used+-- during development for fast compliation.+buildWorkflowPart :: FilePath   -- ^ Path to the db+                  -> String     -- ^ Name of the workflow+                  -> Builder () -- ^ Builder+                  -> Q [Dec]+buildWorkflowPart dbPath wfName b = do+    st <- runIO $ getWorkflowState dbPath+    mkWorkflow wfName $ trimDAG st $ mkDAG b+  where+    getWorkflowState fl = do+        db <- openDB fl+        ks <- getKeys db+        return $ M.fromList $ zip ks $ repeat Success++-- TODO: check the graph is a valid DAG+-- | Contruct a DAG representing the workflow+mkDAG :: Builder () -> DAG+mkDAG builder = mkGraph ns' es'+  where+    ns' = map (\x -> (pid2nid $ _nodePid x, x)) ns+    es' = map (\Edge{..} -> (pid2nid _edgeFrom, pid2nid _edgeTo, _edgeOrd)) es+    (ns, es) = execState builder ([], [])+    pid2nid pid = M.findWithDefault+        (error $ "mkDAG: cannot identify node: " ++ T.unpack pid) pid $+        M.fromListWithKey+            (\k _ _ -> error $ "Multiple declaration for: " ++ T.unpack k) $+            zip (map _nodePid ns) [0..]+{-# INLINE mkDAG #-}++-- | Remove nodes that are executed before from a DAG.+trimDAG :: (M.Map T.Text NodeState) -> DAG -> DAG+trimDAG st dag = gmap revise gr+  where+    revise context@(linkTo, _, nodeLabel, _)+        | shallBuild (_nodePid nodeLabel) && null linkTo = context+        | otherwise = context & _3 %~+            ( \l -> l{_nodeFunction = feedEmptyInput (_nodeFunction l)} )+      where+        feedEmptyInput x = [| (\() -> undefined) >=> $(x) |]+    gr = labnfilter f dag+      where+        f (i, x) = shallBuild (_nodePid x) || any shallBuild children+          where children = map (_nodePid . fromJust . lab dag) $ suc dag i+    shallBuild x = case M.lookup x st of+        Just Success -> False+        _ -> True+{-# INLINE trimDAG #-}+++-- 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 $name$_function_table.+mkWorkflow :: String   -- name+           -> DAG -> Q [Dec]+mkWorkflow workflowName dag =+    [d| $(varP $ mkName workflowName) = Workflow dag' pids $workflowMain |]+  where+    workflowMain = connect sinks [| const $ return () |]+    dag' = nmap _nodePid dag+    computeNodes = snd $ unzip $ labNodes dag+    pids = M.fromList $ map (\Node{..} -> (_nodePid, _nodeAttr)) computeNodes+    sinks = labNodes $ nfilter ((==0) . outdeg dag) dag++    backTrack (i, Node{..}) = connect (fst $ unzip parents) [| $mkP $fun |]+      where+        parents = map ( \(x, o) -> ((x, fromJust $ lab dag x), o) ) $+            sortBy (comparing snd) $ lpre dag i+        fun = case _nodeAttr^.functionConfig of+            FunctionConfig _ Pure -> [| return . $_nodeFunction |]+            FunctionConfig _ IOAction -> [| liftIO . $_nodeFunction |]+            FunctionConfig _ Stateful -> [| (lift . lift) . $_nodeFunction |]+        mkP = case _nodeAttr^.functionConfig of+            FunctionConfig None _ -> [| mkProc _nodePid |]+            FunctionConfig (Standard n) _ -> [| mkProcListN n _nodePid |]+            FunctionConfig (ShareData n) _ -> [| mkProcListNWithContext n _nodePid |]++    connect [] sink = sink+    connect [source] sink = [| $(backTrack source) >=> $sink |]+    connect sources sink = [| fmap runParallel $expq >=> $sink |]+      where+        expq = foldl' g e0 $ sources+        e0 = [| (pure. pure) $(conE (tupleDataName $ length sources)) |]+        g acc x = [| ((<*>) . fmap (<*>)) $acc $ fmap Parallel $(backTrack x) |]+{-# INLINE mkWorkflow #-}++mkProc :: (DBData a, DBData b, ToJSON config)+       => PID -> (a -> (ProcState config) b) -> (Processor config a b)+mkProc = mkProcWith (return, runIdentity)+{-# INLINE mkProc #-}++mkProcListN :: (DBData a, DBData b, ToJSON config)+            => Int+            -> PID+            -> (a -> (ProcState config) b)+            -> (Processor config [a] [b])+mkProcListN n pid f = mkProcWith (chunksOf n, concat) pid $ mapM f+{-# INLINE mkProcListN #-}++mkProcListNWithContext :: (DBData a, DBData b, DBData c, ToJSON config)+                       => Int -> PID+                       -> (ContextData c a -> (ProcState config) b)+                       -> (Processor config (ContextData c [a]) [b])+mkProcListNWithContext n pid f = mkProcWith (toChunks, concat) pid f'+  where+    f' (ContextData c xs) = mapM f $ zipWith ContextData (repeat c) xs+    toChunks (ContextData c xs) = zipWith ContextData (repeat c) $ chunksOf n xs+{-# INLINE mkProcListNWithContext #-}++mkProcWith :: (Traversable t, DBData a, DBData b, ToJSON config)+           => (a -> t a, t b -> b) -> PID+           -> (a -> (ProcState config) b)+           -> (Processor config a b)+mkProcWith (box, unbox) pid f = \input -> do+    wfState <- ask+    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+            readData pid $ wfState^.database+        Scheduled -> do+            _ <- liftIO $ takeMVar $ wfState^.procParaControl++            liftIO $ sendLog (wfState^.logServer) $ Running pid++            config <- lift $ lift ask+            let sendToRemote = fromMaybe (wfState^.remote) (attr^.submitToRemote)+                remoteOpts = RemoteOpts+                    { extraParams = attr^.remoteParam+                    , environment = config+                    }+                input' = box input+            result <- try $ unbox <$> if sendToRemote+                then liftIO $ mapConcurrently (runRemote remoteOpts pid) input'+                else mapM f input'  -- disable parallel in local machine due to memory issue+            case result of+                Left ex -> do+                    _ <- liftIO $ do+                        putMVar pSt $ Fail ex+                        _ <- forkIO $ putMVar (wfState^.procParaControl) ()+                        sendLog (wfState^.logServer) $ Warn pid "Failed!"+                    lift (throwE (pid, ex))+                Right r -> liftIO $ do+                    saveData pid r $ wfState^.database+                    putMVar pSt Success+                    _ <- forkIO $ putMVar (wfState^.procParaControl) ()+                    sendLog (wfState^.logServer) $ Complete pid+                    return r++        Special mode -> handleSpecialMode mode wfState pSt pid f+{-# INLINE mkProcWith #-}++handleSpecialMode :: (DBData a, DBData b)+                  => SpecialMode+                  -> WorkflowState+                  -> MVar NodeState -> PID+                  -> (a -> (ProcState config) b)+                  -> (ProcState config) b+handleSpecialMode mode wfState nodeSt pid fn = case mode of+    Skip -> liftIO $ putMVar nodeSt (Special Skip) >> return undefined++    EXE inputData output -> do+        c <- liftIO $ B.readFile inputData+        r <- fn $ deserialize c+        liftIO $ B.writeFile output $ serialize r+        liftIO $ putMVar nodeSt $ Special Skip+        return r++    -- Read data stored in this node+    FetchData -> liftIO $ do+        r <- readData pid $ wfState^.database+        B.putStr $ showYaml r+        putMVar nodeSt $ Special Skip+        return r++    -- Replace data stored in this node+    WriteData inputData -> do+        c <- liftIO $ B.readFile inputData+        r <- return (readYaml c) `asTypeOf` fn undefined+        liftIO $ do+            updateData pid r $ wfState^.database+            putMVar nodeSt $ Special Skip+            return r+{-# INLINE handleSpecialMode #-}
+ src/Scientific/Workflow/Internal/Builder/Types.hs view
@@ -0,0 +1,113 @@+{-# LANGUAGE DeriveGeneric        #-}+{-# LANGUAGE FlexibleInstances    #-}+{-# LANGUAGE OverloadedStrings    #-}+{-# LANGUAGE TemplateHaskell      #-}+{-# LANGUAGE TypeSynonymInstances #-}+module Scientific.Workflow.Internal.Builder.Types where++import           Control.Lens                      (makeLenses)+import           Control.Monad.State               (State)+import           Data.Aeson.Types                  (defaultOptions,+                                                    genericParseJSON,+                                                    genericToEncoding)+import           Data.Graph.Inductive.PatriciaTree (Gr)+import           Data.Serialize                    (Serialize)+import           Data.Serialize.Text               ()+import           Data.Text                         (Text)+import           Data.Yaml                         (FromJSON (..), ToJSON (..))+import           GHC.Generics                      (Generic)+import           Instances.TH.Lift                 ()+import           Language.Haskell.TH               (ExpQ, Name, varE)+import           Language.Haskell.TH.Lift          (deriveLift)++-- | A computation node.+data Node = Node+    { _nodePid      :: Text+    , _nodeFunction :: ExpQ+    , _nodeAttr     :: Attribute+    }++-- | Links between computational nodes+data Edge = Edge+    { _edgeFrom :: Text+    , _edgeTo   :: Text+    , _edgeOrd  :: EdgeOrd  -- ^ Order of the edge+    }++type EdgeOrd = Int++type Builder = State ([Node], [Edge])++-- | Node attributes.+data Attribute = Attribute+    { _label          :: Text      -- ^ Short description+    , _note           :: Text      -- ^ Long description+    , _submitToRemote :: Maybe Bool  -- ^ Overwrite the global option+    , _remoteParam    :: String     -- ^ Parameters for to remote execution+    , _functionConfig :: FunctionConfig  -- ^ Usually not being used directly+    } deriving (Generic)++-- | The type of node function+data FunctionConfig = FunctionConfig ParallelMode FunctionType deriving (Generic)++data ParallelMode = None            -- ^ No parallelism.+                  | Standard Int    -- ^ Turn input @a@ into @[a]@ and process+                                    -- them in parallel.+                  | ShareData Int   -- ^ Assume the input is @ContextData d a@,+                                    -- where @d@ is shared and @a@ becomes @[a]@.+                  deriving (Generic)++data FunctionType = Pure       -- ^ The function is pure, i.e., @a -> b@.+                  | IOAction   -- ^ A IO function, i.e., @a -> IO b@.+                  | Stateful   -- ^ A function that has access to configuration,+                               -- i.e., @a -> WorkflowConfig config b@.+                  deriving (Generic)++instance Serialize Attribute+instance Serialize FunctionConfig+instance Serialize ParallelMode+instance Serialize FunctionType++deriveLift ''FunctionConfig+deriveLift ''ParallelMode+deriveLift ''FunctionType+deriveLift ''Attribute++makeLenses 'Attribute++defaultAttribute :: Attribute+defaultAttribute = Attribute+    { _label = ""+    , _note = ""+    , _submitToRemote = Nothing+    , _remoteParam = ""+    , _functionConfig = FunctionConfig None IOAction+    }++type AttributeSetter = State Attribute ()++type DAG = Gr Node EdgeOrd++-- | Objects that can be converted to ExpQ+class ToExpQ a where+    toExpQ :: a -> ExpQ++instance ToExpQ Name where+    toExpQ = varE++instance ToExpQ ExpQ where+    toExpQ = id++-- | Data and its environment.+data ContextData context dat = ContextData+    { _context :: context+    , _data    :: dat+    } deriving (Generic)++instance (FromJSON c, FromJSON d) => FromJSON (ContextData c d) where+    parseJSON = genericParseJSON defaultOptions++instance (ToJSON c, ToJSON d) => ToJSON (ContextData c d) where+    toEncoding = genericToEncoding defaultOptions++instance (Serialize c, Serialize d) => Serialize (ContextData c d)
+ src/Scientific/Workflow/Internal/DB.hs view
@@ -0,0 +1,129 @@+{-# LANGUAGE FlexibleInstances    #-}+{-# LANGUAGE OverloadedStrings    #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE ConstraintKinds #-}+module Scientific.Workflow.Internal.DB+    ( openDB+    , closeDB+    , readData+    , readDataByteString+    , saveDataByteString+    , saveData+    , updateData+    , delRecord+    , isFinished+    , getKeys+    , WorkflowDB(..)+    , DBData+    , serialize+    , deserialize+    , readYaml+    , showYaml+    ) where++import qualified Data.ByteString        as B+import           Data.Maybe             (fromJust)+import qualified Data.Serialize         as S+import qualified Data.Text              as T+import           Data.Yaml              (FromJSON (..), ToJSON (..), decode,+                                         encode)+import           Database.SQLite.Simple+import           Text.Printf            (printf)++-- | An abstract type representing the database used to store states of workflow+newtype WorkflowDB  = WorkflowDB Connection++-- | @DBData@ constraint is used for data serialization.+type DBData a = (FromJSON a, ToJSON a, S.Serialize a)++serialize :: DBData a => a -> B.ByteString+serialize = S.encode++deserialize :: DBData a => B.ByteString -> a+deserialize = fromEither . S.decode+  where+    fromEither (Right x) = x+    fromEither _         = error "decode failed"++showYaml :: DBData a => a -> B.ByteString+showYaml = encode++readYaml :: DBData a => B.ByteString -> a+readYaml = fromJust . decode++type PID = T.Text++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 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 #-}++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 #-}++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 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 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]
+ src/Scientific/Workflow/Internal/Utils.hs view
@@ -0,0 +1,109 @@+{-# LANGUAGE CPP           #-}+{-# LANGUAGE DeriveGeneric #-}+module Scientific.Workflow.Internal.Utils+    ( RemoteOpts(..)+    , Log(..)+    , runRemote+    , sendLog+    )where++import qualified Data.ByteString.Char8           as B+import qualified Data.Serialize                  as S+import           Data.Serialize.Text             ()+import qualified Data.Text                       as T+import           Data.Time                       (defaultTimeLocale, formatTime,+                                                  getZonedTime)+import           Data.Yaml                       (ToJSON, encode)+import           GHC.Generics                    (Generic)+import           Network.Socket                  (Socket)+import           Network.Socket.ByteString       (sendAll)+import           Rainbow+import           System.IO++import           Scientific.Workflow.Internal.DB++#ifdef DRMAA_ENABLED+import           DRMAA                           (DrmaaAttribute (..),+                                                  defaultDrmaaConfig, drmaaRun)+import           System.Environment.Executable   (getExecutablePath)+import           System.IO.Temp                  (withTempDirectory)+#endif++data Log = Running T.Text+         | Complete T.Text+         | Warn T.Text String+         | Error String+         | Exit+         deriving (Generic, Show)++instance S.Serialize Log++getTime :: IO String+getTime = do+    t <- getZonedTime+    return $ formatTime defaultTimeLocale "[%m-%d %H:%M]" t+{-# INLINE getTime #-}++sendLog :: Maybe Socket -> Log -> IO ()+sendLog sock msg = do+    case sock of+        Just s -> sendAll s $ S.encode msg+        _      -> return ()+    case msg of+        Running pid  -> logMsg $ T.unpack pid ++ ": Running..."+        Complete pid -> logMsg $ T.unpack pid ++ ": Finished!"+        Warn pid s   -> warnMsg $ T.unpack pid ++ ": " ++ s+        Error s  -> errorMsg s+        Exit -> return ()++logMsg :: String -> IO ()+logMsg txt = do+    t <- getTime+    let prefix = bold $ chunk ("[LOG]" ++ t ++ " ") & fore green+        msg = B.concat $ chunksToByteStrings toByteStringsColors8+            [prefix, chunk txt & fore green]+    B.hPutStrLn stderr msg++errorMsg :: String -> IO ()+errorMsg txt = do+    t <- getTime+    let prefix = bold $ chunk ("[ERROR]" ++ t ++ " ") & fore red+        msg = B.concat $ chunksToByteStrings toByteStringsColors8+            [prefix, chunk txt & fore red]+    B.hPutStrLn stderr msg++warnMsg :: String -> IO ()+warnMsg txt = do+    t <- getTime+    let prefix = bold $ chunk ("[WARN]" ++ t ++ " ") & fore yellow+        msg = B.concat $ chunksToByteStrings toByteStringsColors8+            [prefix, chunk txt & fore red]+    B.hPutStrLn stderr msg++data RemoteOpts config = RemoteOpts+    { extraParams :: String+    , environment :: config+    }++runRemote :: (DBData a, DBData b, ToJSON config)+          => RemoteOpts config -> T.Text -> a -> IO b+#ifdef DRMAA_ENABLED+runRemote opts pid input = withTempDirectory tmpDir "drmaa.tmp" $ \dir -> do+    let inputFl = dir ++ "/drmaa_input.tmp"+        outputFl = dir ++ "/drmaa_output.tmp"+        configFl = dir ++ "/drmaa_config.tmp"++    B.writeFile configFl $ encode $ environment opts++    exePath <- getExecutablePath+    let config = defaultDrmaaConfig{drmaa_native=extraParams opts}++    B.writeFile inputFl $ serialize input+    drmaaRun exePath [ "execFunc", "--config", configFl, T.unpack pid+        , inputFl, outputFl ] config :: IO ()+    deserialize <$> B.readFile outputFl+  where+    tmpDir = "./"+#else+runRemote = error "DRMAA support was not turned on."+#endif
src/Scientific/Workflow/Main.hs view
@@ -8,126 +8,144 @@     , defaultMainOpts     , mainWith     , MainOpts(..)+    , runWorkflow     ) where -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           Control.Concurrent                         (forkIO)+import           Control.Concurrent.MVar+import           Control.Exception                          (bracket,+                                                             displayException)+import           Control.Monad                              (replicateM_)+import           Control.Monad.Reader                       (runReaderT)+import           Control.Monad.Trans.Except                 (runExceptT)+import           Data.Default.Class                         (Default (..))+import           Data.Graph.Inductive.Graph                 (lab, labNodes,+                                                             nmap)+import           Data.Graph.Inductive.Query.DFS             (rdfs)+import qualified Data.Map                                   as M+import           Data.Maybe                                 (fromJust)+import qualified Data.Set                                   as S+import           Data.Tuple                                 (swap)+import           Network.Socket                             (Family (..),+                                                             SockAddr (..),+                                                             Socket,+                                                             SocketType (Stream),+                                                             close, connect,+                                                             defaultProtocol,+                                                             isConnected,+                                                             socket) -#ifdef SGE-import           DRMAA                             (withSGESession)+import qualified Data.ByteString.Char8                      as B+import           Data.Graph.Inductive.PatriciaTree          (Gr)+import           Data.Maybe                                 (fromMaybe)+import           Data.Serialize                             (encode)+import qualified Data.Text                                  as T+import qualified Data.Text.Lazy.IO                          as T+import           Data.Yaml                                  (FromJSON, decode)++#ifdef DRMAA_ENABLED+import           DRMAA                                      (withSession) #endif  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 qualified Language.Haskell.TH.Lift                   as T+import           Options.Applicative                        hiding (Success)+import           Text.Printf                                (printf) -import           Scientific.Workflow-import           Scientific.Workflow.DB+import           Data.Version                               (showVersion)+import           Paths_SciFlow                              (version)+import           Scientific.Workflow.Internal.Builder+import           Scientific.Workflow.Internal.Builder.Types+import           Scientific.Workflow.Internal.DB+import           Scientific.Workflow.Internal.Utils+import           Scientific.Workflow.Main.Options           (CMD (..),+                                                             GlobalOpts (..),+                                                             argsParser)+import           Scientific.Workflow.Types import           Scientific.Workflow.Visualize -import           Data.Version                      (showVersion)-import           Paths_SciFlow                     (version)-+data MainOpts = MainOpts+    { preAction     :: Name   -- ^ An action to be execute before the workflow.+                              -- The action should have type: @'IO' () -> 'IO' ()@.+                              -- e.g., some initialization processes.+    , programHeader :: String+    } -data CMD = Run GlobalOpts Int Bool-         | View-         | Cat GlobalOpts String-         | Write GlobalOpts String FilePath-         | Delete GlobalOpts String-         | Recover GlobalOpts FilePath-         | DumpDB GlobalOpts FilePath-         | Call GlobalOpts String String String+T.deriveLift ''MainOpts -data GlobalOpts = GlobalOpts-    { dbPath :: FilePath-    , configFile :: Maybe FilePath+defaultMainOpts :: MainOpts+defaultMainOpts = MainOpts+    { preAction = 'id+    , programHeader = printf "SciFlow-%s" (showVersion version)     } -globalParser :: Parser GlobalOpts-globalParser = GlobalOpts-           <$> strOption-               ( long "db-path"-              <> value "sciflow.db"-              <> metavar "DB_PATH" )-           <*> (optional . strOption)-               ( long "config"-              <> metavar "CONFIG_PATH" )+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 (\x -> (_nodePid x, _nodeAttr x)) $ mkDAG builder+{-# INLINE mainWith #-} -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-#ifdef SGE-    | r = initialize $ withSGESession $ runWorkflow wf $-        RunOpt (dbPath opts) n True Normal $ configFile opts+mainFunc :: (Default config, FromJSON config)+         => (IO () -> IO ()) -- initialization function+         -> Gr (PID, Attribute) Int -> Workflow config+         -> String  -- program header+         -> IO ()+mainFunc initialize dag wf h = execParser (argsParser h) >>= execute+  where+    execute cmd = case cmd of+        Run opts n r s logS ->+            let runOpts = defaultRunOpt+                    { dbFile = dbPath opts+                    , runOnRemote = True+                    , nThread = n+                    , configuration = fromMaybe [] $ configFile opts+                    , selected = fmap (map T.pack) s+                    , logServerAddr = logS }+            in if r+#ifdef DRMAA_ENABLED+                then initialize $ withSession $ runWorkflow wf runOpts #else-    | r = initialize $ runWorkflow wf $-        RunOpt (dbPath opts) n True Normal $ configFile opts+                then initialize $ runWorkflow wf runOpts #endif-    | otherwise = runWorkflow wf $-        RunOpt (dbPath opts) n False Normal $ configFile opts-runExe _ _ _ = undefined-{-# INLINE runExe #-}+                else runWorkflow wf runOpts{runOnRemote = False} -viewParser :: Parser CMD-viewParser = pure View-viewExe = T.putStrLn . renderBuilder-{-# INLINE viewExe #-}+        View isRaw -> if isRaw+            then B.putStr $ encode dag+            else T.putStrLn $ drawWorkflow dag -catParser :: Parser CMD-catParser = Cat-        <$> globalParser-        <*> strArgument-            (metavar "NODE_ID")-catExe (Cat opts pid) wf = runWorkflow wf $-        RunOpt (dbPath opts) 10 False (ReadSingle $ T.pack pid) $ configFile opts-catExe _ _ = undefined-{-# INLINE catExe #-}+        Cat opts pid -> runWorkflow wf defaultRunOpt+            { dbFile = dbPath opts+            , nThread = 4+            , runMode = Review $ T.pack pid+            , configuration = fromMaybe [] $ configFile opts } -writeParser :: Parser CMD-writeParser = Write-          <$> globalParser-          <*> strArgument-              (metavar "NODE_ID")-          <*> strArgument-              (metavar "INPUT_FILE")-writeExe (Write opts pid input) wf = runWorkflow wf $-    RunOpt (dbPath opts) 10 False (WriteSingle (T.pack pid) input) $ configFile opts-writeExe _ _ = undefined-{-# INLINE writeExe #-}+        Write opts pid input -> runWorkflow wf defaultRunOpt+            { dbFile = dbPath opts+            , nThread = 4+            , runMode = Replace (T.pack pid) input+            , configuration = fromMaybe [] $ configFile opts } -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 #-}+        Delete opts pid -> bracket (openDB $ dbPath opts) closeDB+            (delRecord $ T.pack pid) -recoverParser :: Parser CMD-recoverParser = Recover-            <$> globalParser-            <*> strArgument-                (metavar "BACKUP")+        Call opts pid inputFl outputFl -> runWorkflow wf defaultRunOpt+            { dbFile = dbPath opts+            , nThread = 4+            , runMode = Slave (T.pack pid) inputFl outputFl+            , configuration = fromMaybe [] $ configFile opts }++        Recover _ _  -> undefined+        DumpDB _ _   -> undefined+                 {- recoverExe (Recover opts dir) (Workflow _ ft _) = do     fls <- shelly $ lsT $ fromText $ T.pack dir@@ -143,14 +161,7 @@                 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@@ -165,83 +176,77 @@                 B.writeFile fl $ showYaml dat             Nothing -> return ()             -}-dumpDBExe _ _ = undefined-{-# INLINE dumpDBExe #-} -callParser :: Parser CMD-callParser = Call-         <$> globalParser-         <*> strArgument mempty-         <*> strArgument mempty-         <*> strArgument mempty-callExe (Call opts pid inputFl outputFl) wf = runWorkflow wf $-    RunOpt (dbPath opts) 10 False (ExecSingle (T.pack pid) inputFl outputFl) $ configFile opts-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+runWorkflow :: (Default config, FromJSON config)+            => Workflow config -> RunOpt -> IO ()+runWorkflow (Workflow gr pids wf) opts =+    bracket (mkConnection opts) cleanUp $ \(db, logS) -> do+        ks <- S.fromList <$> getKeys db+        let selection = case selected opts of+                Nothing -> Nothing+                Just xs -> let nodeMap = M.fromList $ map swap $ labNodes gr+                               nds = map (flip (M.findWithDefault undefined) nodeMap) xs+                           in Just $ S.fromList $ map (fromJust . lab gr) $ rdfs nds gr -    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.")-     )+        pidStateMap <- flip M.traverseWithKey pids $ \pid attr ->+            case runMode opts of+                Master -> do+                    v <- case fmap (S.member pid) selection of+                        Just False -> newMVar $ Special Skip+                        _ -> if pid `S.member` ks+                            then newMVar Success+                            else newMVar Scheduled+                    return (v, attr)+                Slave i input output -> do+                    v <- if pid == i+                        then newMVar $ Special $ EXE input output+                        else newMVar $ Special Skip+                    return (v, attr)+                Review i -> do+                    v <- if pid == i then newMVar (Special FetchData) else newMVar $ Special Skip+                    return (v, attr)+                Replace i input -> do+                    v <- if pid == i then newMVar (Special $ WriteData input) else newMVar $ Special Skip+                    return (v, attr) +        availableThreads <- newEmptyMVar+        _ <- forkIO $ replicateM_ (nThread opts) $ putMVar availableThreads () -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-    }+        let initState = WorkflowState db pidStateMap availableThreads+                (runOnRemote opts) logS -T.deriveLift ''MainOpts+        config <- case configuration opts of+            [] -> return def+            fls -> do+                r <- decode . B.unlines <$> mapM B.readFile fls+                case r of+                    Nothing -> error "fail to parse configuration file"+                    Just x  -> return x -defaultMainOpts :: MainOpts-defaultMainOpts = MainOpts-    { preAction = 'id-    , programHeader = printf "SciFlow-%s" (showVersion version)-    }+        result <- runReaderT (runExceptT $ runReaderT (wf ()) initState) config+        case result of+            Right _ -> return ()+            Left (pid, ex) -> sendLog logS $ Error $ printf "\"%s\" failed. The error was: %s."+                pid (displayException ex) -defaultMain :: Builder () -> Q [Dec]-defaultMain = mainWith defaultMainOpts+mkConnection :: RunOpt -> IO (WorkflowDB, Maybe Socket)+mkConnection opts = do+    db <- openDB $ dbFile opts+    logS <- case logServerAddr opts of+        Just addr -> do+            sock <- socket AF_UNIX Stream defaultProtocol+            connect sock $ SockAddrUnix addr+            connected <- isConnected sock+            if connected+                then return $ Just sock+                else error "Could not connect to socket!"+        Nothing -> return Nothing+    return (db, logS) -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 #-}+cleanUp :: (WorkflowDB, Maybe Socket) -> IO ()+cleanUp (db, sock) = do+    sendLog sock Exit+    case sock of+        Just s -> close s+        _      -> return ()+    closeDB db
+ src/Scientific/Workflow/Main/Options.hs view
@@ -0,0 +1,120 @@+module Scientific.Workflow.Main.Options+    ( CMD (..)+    , GlobalOpts(..)+    , argsParser+    ) where++import           Data.List.Split     (splitOn)+import           Data.Semigroup      ((<>))+import           Options.Applicative++argsParser :: String -> ParserInfo CMD+argsParser h = info (helper <*> parser) $ fullDesc <> header h+  where+    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 CMD = Run GlobalOpts Int Bool (Maybe [String]) (Maybe String)+         | View Bool+         | Cat GlobalOpts String+         | Write GlobalOpts String FilePath+         | Delete GlobalOpts String+         | Recover GlobalOpts FilePath+         | DumpDB GlobalOpts FilePath+         | Call GlobalOpts String String String++data GlobalOpts = GlobalOpts+    { dbPath     :: FilePath+    , configFile :: Maybe [FilePath]+    }++globalParser :: Parser GlobalOpts+globalParser = GlobalOpts+           <$> strOption+               ( long "db-path"+              <> value "sciflow.db"+              <> metavar "DB_PATH" )+           <*> (optional . option (splitOn "," <$> str))+               ( long "config"+              <> metavar "CONFIG_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.")+    <*> (optional . option (splitOn "," <$> str))+        ( long "select"+       <> metavar "SELECTED"+       <> help "Run only selected nodes.")+    <*> (optional . fmap f . strOption)+        ( long "log-server"+       <> metavar "Log_SERVER" )+  where+    f x = case x of+        ('\\' : '0' : rest) -> '\0' : rest+        x' -> x'++viewParser :: Parser CMD+viewParser = View <$> switch (long "raw")++catParser :: Parser CMD+catParser = Cat+        <$> globalParser+        <*> strArgument+            (metavar "NODE_ID")++writeParser :: Parser CMD+writeParser = Write+          <$> globalParser+          <*> strArgument+              (metavar "NODE_ID")+          <*> strArgument+              (metavar "INPUT_FILE")++rmParser :: Parser CMD+rmParser = Delete+       <$> globalParser+       <*> strArgument+           (metavar "NODE_ID")++recoverParser :: Parser CMD+recoverParser = Recover+            <$> globalParser+            <*> strArgument+                (metavar "BACKUP")++dumpDBParser :: Parser CMD+dumpDBParser = DumpDB+           <$> globalParser+           <*> strArgument+               (metavar "OUTPUT_DIR")++callParser :: Parser CMD+callParser = Call+         <$> globalParser+         <*> strArgument mempty+         <*> strArgument mempty+         <*> strArgument mempty
src/Scientific/Workflow/Types.hs view
@@ -1,238 +1,130 @@+{-# LANGUAGE DeriveGeneric         #-} {-# LANGUAGE FlexibleInstances     #-}-{-# LANGUAGE GADTs                 #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings     #-}-{-# LANGUAGE ScopedTypeVariables   #-}+{-# LANGUAGE Rank2Types            #-} {-# LANGUAGE TemplateHaskell       #-} {-# LANGUAGE TypeFamilies          #-} {-# LANGUAGE UndecidableInstances  #-}  module Scientific.Workflow.Types-    ( WorkflowDB(..)-    , Workflow(..)+    ( Workflow(..)     , PID-    , NodeResult(..)+    , NodeState(..)+    , SpecialMode(..)     , ProcState     , WorkflowState(..)-    , db+    , database     , procStatus     , procParaControl     , remote-    , config-    , getConfig-    , getConfigMaybe-    , getConfig'-    , getConfigMaybe'+    , logServer     , Processor     , RunMode(..)     , RunOpt(..)-    , BatchData(..)-    , BatchData'(..)-    , IsList-    , DBData(..)-    , Attribute(..)-    , AttributeSetter-    , defaultAttribute-    , label-    , note-    , batch-    , submitToRemote-    , stateful-    , remoteParam-+    , defaultRunOpt     , Parallel(..)--    -- * Builder types-    , Node-    , Edge-    , EdgeOrd-    , Builder+    , WorkflowConfig     ) where -import           Control.Concurrent.Async.Lifted   (concurrently)-import           Control.Concurrent.MVar           (MVar)-import           Control.Exception                 (SomeException)-import           Control.Lens                      (at, makeLenses, (^.))-import           Control.Monad.State               (State, StateT, get)-import           Control.Monad.Trans.Except        (ExceptT)-import qualified Data.ByteString                   as B-import           Data.Graph.Inductive.Graph        (labEdges, labNodes, mkGraph)-import           Data.Graph.Inductive.PatriciaTree (Gr)-import           Data.List.Split                   (chunksOf)-import qualified Data.Map                          as M-import           Data.Maybe                        (fromJust, fromMaybe)-import qualified Data.Serialize                    as S-import qualified Data.Text                         as T-import           Data.Yaml                         (FromJSON, ToJSON, decode,-                                                    encode)-import           Database.SQLite.Simple            (Connection)-import           Language.Haskell.TH-import qualified Language.Haskell.TH.Lift          as T---- | 'DBData' type class is used for data serialization.-class DBData a where-    serialize :: a -> B.ByteString-    deserialize :: B.ByteString -> a-    showYaml :: a -> B.ByteString-    readYaml :: B.ByteString -> a--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+import           Control.Concurrent.Async.Lifted            (concurrently)+import           Control.Concurrent.MVar                    (MVar)+import           Control.Exception                          (SomeException)+import           Control.Lens                               (makeLenses)+import           Control.Monad.Reader                       (ReaderT)+import           Control.Monad.Trans.Except                 (ExceptT)+import           Data.Graph.Inductive.Graph                 (labEdges, labNodes,+                                                             mkGraph)+import           Data.Graph.Inductive.PatriciaTree          (Gr)+import qualified Data.Map                                   as M+import qualified Data.Serialize                             as S+import           Data.Serialize.Text                        ()+import qualified Data.Text                                  as T+import qualified Language.Haskell.TH.Lift                   as T+import           Network.Socket                             (Socket) --- | An abstract type representing the database used to store states of workflow-newtype WorkflowDB  = WorkflowDB Connection+import           Scientific.Workflow.Internal.Builder.Types (Attribute)+import           Scientific.Workflow.Internal.DB            (WorkflowDB (..))  -- | The id of a node type PID = T.Text --- | Node attribute-data Attribute = Attribute-    { _label          :: T.Text      -- ^ Short description-    , _note           :: T.Text      -- ^ Long description-    , _batch          :: Int         -- ^ Batch size. If > 0, inputs will be divided-                                     -- into batches.-    , _submitToRemote :: Maybe Bool  -- ^ Overwrite the global option-    , _stateful       :: Bool        -- ^ Whether the node function has access-                                     -- to internal states-    , _remoteParam    :: String-    }--makeLenses ''Attribute--defaultAttribute :: Attribute-defaultAttribute = Attribute-    { _label = ""-    , _note = ""-    , _batch = -1-    , _submitToRemote = Nothing-    , _stateful = False-    , _remoteParam = ""-    }--type AttributeSetter = State Attribute ()- -- | The result of a computation node-data NodeResult = Success                -- ^ The node has been executed-                | Fail SomeException     -- ^ The node failed to finish-                | Scheduled              -- ^ The node will be executed-                | Skip                   -- ^ The node will not be executed-                | Read                   -- ^ Simply read the saved data from database-                | Replace FilePath         -- ^ Read the result from the input file-                                         -- and save it to database.-                | EXE FilePath FilePath  -- ^ Read input from the input file and-                                         -- save results to the output file. This is-                                         -- used in remote mode.+data NodeState = Success               -- ^ The node has been executed+               | Fail SomeException    -- ^ The node failed to finish+               | Scheduled             -- ^ The node will be executed+               | Special SpecialMode   -- ^ Indicate the workflow is currently+                                       -- running under special mode +data SpecialMode = Skip                -- ^ The node will not be executed+                 | FetchData           -- ^ Simply read the saved data from database+                 | WriteData FilePath  -- ^ Read the result from the input file+                                       -- and save it to database.+                 | EXE FilePath FilePath  -- ^ Read input from the input file and+                                          -- save results to the output file. This is+                                          -- used in remote mode.+ data WorkflowState = WorkflowState-    { _db              :: WorkflowDB-    , _procStatus      :: M.Map PID (MVar NodeResult, Attribute)+    { _database        :: WorkflowDB+    , _procStatus      :: M.Map PID (MVar NodeState, Attribute)     , _procParaControl :: MVar () -- ^ Concurrency controller     , _remote          :: Bool    -- ^ Global remote switch-    , _config          :: M.Map T.Text T.Text    -- ^ Workflow configuration. This-                                                 -- is used to store environmental-                                                 -- variables.+    , _logServer       :: Maybe Socket  -- ^ Server for logging     }  makeLenses ''WorkflowState -type ProcState b = StateT WorkflowState (ExceptT (PID, SomeException) IO) b-type Processor a b = a -> ProcState b--getConfigMaybe :: T.Text -> ProcState (Maybe T.Text)-getConfigMaybe key = do-    st <- get-    return $ (st^.config) ^.at key--getConfig :: T.Text -> ProcState T.Text-getConfig x = fmap (fromMaybe errMsg) $ getConfigMaybe x-  where-    errMsg = error $ "The Key " ++ show x ++ " doesn't exist!"--getConfig' :: T.Text -> ProcState String-getConfig' = fmap T.unpack . getConfig--getConfigMaybe' :: T.Text -> ProcState (Maybe String)-getConfigMaybe' = (fmap.fmap) T.unpack . getConfigMaybe+type ProcState config = ReaderT WorkflowState (+    ExceptT (PID, SomeException) (WorkflowConfig config) )+type WorkflowConfig config = ReaderT config IO+type Processor config a b = a -> (ProcState config) b  -- | A Workflow is a stateful function-data Workflow = Workflow (M.Map T.Text Attribute)-                         (Processor () ())+data Workflow config = Workflow+    { _worflow_dag       :: Gr PID Int+    , _worflow_pidToAttr :: M.Map T.Text Attribute+    , _workflow          :: Processor config () ()+    }  -- | Options data RunOpt = RunOpt-    { database      :: FilePath+    { dbFile        :: FilePath     , nThread       :: Int      -- ^ number of concurrent processes     , runOnRemote   :: Bool     , runMode       :: RunMode-    , configuration :: Maybe FilePath+    , configuration :: [FilePath]+    , selected      :: Maybe [PID]  -- ^ Should run only selected nodes+    , logServerAddr :: Maybe String     } -data RunMode = Normal-             | ExecSingle PID FilePath FilePath-             | ReadSingle PID-             | WriteSingle PID FilePath+defaultRunOpt :: RunOpt+defaultRunOpt = RunOpt+    { dbFile = "sciflow.db"+    , nThread  = 1+    , runOnRemote = False+    , runMode = Master+    , configuration = []+    , selected = Nothing+    , logServerAddr = Nothing+    } +data RunMode = Master                       -- ^ Run as the master process+             | Slave PID FilePath FilePath  -- ^ Run as a slave process+             | Review PID                   -- ^ Review the info stored in a node+             | Replace PID FilePath         -- ^ Replace the info stored in a node++instance (T.Lift a, T.Lift b) => T.Lift (Gr a b) where+  lift gr = [| uncurry mkGraph $(T.lift (labNodes gr, labEdges gr)) |]+ -- | Auxiliary type for concurrency support.-newtype Parallel a = Parallel { runParallel :: ProcState a}+newtype Parallel config r = Parallel { runParallel :: (ProcState config) r} -instance Functor Parallel where+instance Functor (Parallel config) where     fmap f (Parallel a) = Parallel $ f <$> a -instance Applicative Parallel where+instance Applicative (Parallel config) where     pure = Parallel . pure     Parallel fs <*> Parallel as = Parallel $         (\(f, a) -> f a) <$> concurrently fs as --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 incoming edges of a node-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])----data HTrue-data HFalse--type family IsList a b where-    IsList [a] [b] = HTrue-    IsList a b = HFalse--class BatchData' flag a b where-    batchFunction' :: flag -> (a -> ProcState b) -> Int -> (a -> [a], [b] -> b)--instance BatchData' HTrue [a] [b] where-    batchFunction' _ _ i = (chunksOf i, concat)--instance BatchData' HFalse a b where-    batchFunction' _ _ _ = (return, head)---- | 'BatchData' represents inputs that can be divided into batches and processed--- in parallel, i.e. list.-class BatchData a b where-    batchFunction :: (a -> ProcState b) -> Int -> (a -> [a], [b] -> b)--instance (IsList a b ~ flag, BatchData' flag a b) => BatchData a b where-    batchFunction = batchFunction' (undefined :: flag)+instance S.Serialize (Gr (PID, Attribute) Int)
− src/Scientific/Workflow/Utils.hs 
@@ -1,63 +0,0 @@-{-# 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 qualified Data.Map as M-import Data.Yaml (encode)--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-    , environment :: M.Map T.Text T.Text-    }--defaultRemoteOpts :: RemoteOpts-defaultRemoteOpts = RemoteOpts-    { extraParams = ""-    , environment = M.empty-    }--runRemote :: (DBData a, DBData b) => RemoteOpts -> T.Text -> a -> IO b-#ifdef SGE-runRemote opts pid input = withTmpFile tmpDir $ \inputFl ->-    withTmpFile tmpDir $ \outputFl -> withTmpFile tmpDir $ \configFl -> do-        B.writeFile configFl $ encode $ environment opts--        exePath <- getExecutablePath-        wd <- getCurrentDirectory-        let config = defaultDrmaaConfig{drmaa_wd=wd, drmaa_native=extraParams opts}--        B.writeFile inputFl $ serialize input-        drmaaRun exePath [ "execFunc", "--config", configFl, T.unpack pid-            , inputFl, outputFl ] config :: IO ()-        deserialize <$> B.readFile outputFl-  where-    tmpDir = "./"-#else-runRemote = error "SGE support was not turned on."-#endif
src/Scientific/Workflow/Visualize.hs view
@@ -1,26 +1,57 @@ {-# LANGUAGE OverloadedStrings #-} module Scientific.Workflow.Visualize-    ( renderBuilder+    ( drawWorkflow     ) where -import Control.Lens-import Scientific.Workflow.Types (label)-import qualified Data.Text as T-import qualified Data.Text.Lazy      as TL--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           Control.Lens                               ((^.))+import           Data.Graph.Inductive.PatriciaTree          (Gr)+import qualified Data.GraphViz                              as G+import qualified Data.GraphViz.Attributes.Complete          as G+import qualified Data.GraphViz.Attributes.HTML              as H+import qualified Data.GraphViz.Printing                     as G+import qualified Data.Text                                  as T+import qualified Data.Text.Lazy                             as TL -import Scientific.Workflow.Types+import           Scientific.Workflow.Internal.Builder.Types (Attribute, note)+import           Scientific.Workflow.Types --- | Print the computation graph-renderBuilder :: Gr (PID, Attribute) Int -> TL.Text-renderBuilder dag = G.renderDot . G.toDot $ G.graphToDot param dag+-- | Output the computation graph in dot code which can be visualize by Graphviz.+drawWorkflow :: Gr (PID, Attribute) Int -> TL.Text+drawWorkflow dag = G.renderDot . G.toDot $ G.graphToDot param dag   where-    fmtnode (_, (p, attr)) = [G.Label $ G.StrLabel $ TL.fromStrict lab]+    fmtnode (_, (i, attr)) = [G.Label $ G.HtmlLabel label]       where-        lab | T.null (attr^.label) = p-            | otherwise = attr^.label-    param = G.nonClusteredParams{G.fmtNode = fmtnode}+        label = H.Table $ H.HTable (Just []) tableAttr $ header : H.HorizontalRule :+            map toLine (wrap 45 $ attr^.note)+        header = H.Cells [H.LabelCell [] $ H.Text+            [ H.Format H.Bold $ [H.Font [H.PointSize 18] [H.Str $ TL.fromStrict i]]+            ]]+        tableAttr = [ H.Border 0+                    , H.CellPadding 0]+    param = G.nonClusteredParams+        { G.globalAttributes =+            [ G.GraphAttrs+                [ -- G.Ratio G.CompressRatio+                -- , G.Size $ G.GSize 7.20472 (Just 9.72441) True+                ]+            , G.NodeAttrs+                [ G.FillColor [G.WC (G.RGBA 190 174 212 100) Nothing]+                , G.Color [G.WC (G.RGBA 190 174 212 0) Nothing]+                , G.Style [G.SItem G.Filled [], G.SItem G.Rounded []]+                , G.Shape G.BoxShape+                , G.FontName "Anonymous Pro, Courier"+                , G.FontSize 16+                ]+            ]+        , G.fmtNode = fmtnode+        }+    toLine x = H.Cells [H.LabelCell [H.Align H.HLeft] $+        H.Text [H.Str $ TL.fromStrict x]]++wrap :: Int -> T.Text -> [T.Text]+wrap limit = concatMap (combine . foldl f (0, [], []) . T.words) . T.lines+  where+    f (count, acc, line) w = if count + T.length w >= limit+        then (0, [], line ++ [T.unwords $ acc ++ [w]])+        else (count + T.length w + 1, acc ++ [w], line)+    combine (_, acc, line) = line ++ [T.unwords acc]