SciFlow 0.6.2 → 0.7.0
raw patch · 26 files changed
+1378/−1444 lines, 26 filesdep +binarydep +constraintsdep +cryptohash-sha256dep −aesondep −cerealdep −cereal-text
Dependencies added: binary, constraints, cryptohash-sha256, distributed-process, distributed-process-monad-control, hashable, memory, monad-control, network-transport, network-transport-tcp, stm, unordered-containers
Dependencies removed: aeson, cereal, cereal-text, containers, data-default-class, directory, drmaa, executable-path, fgl, graphviz, lens, network, optparse-applicative, split, temporary, th-lift, transformers, yaml
Files
- LICENSE +1/−1
- README.md +1/−45
- SciFlow.cabal +33/−44
- src/Control/Arrow/Async.hs +35/−0
- src/Control/Arrow/Free.hs +177/−0
- src/Control/Workflow.hs +115/−0
- src/Control/Workflow/Coordinator.hs +84/−0
- src/Control/Workflow/Coordinator/Local.hs +48/−0
- src/Control/Workflow/DataStore.hs +130/−0
- src/Control/Workflow/Interpreter/Exec.hs +114/−0
- src/Control/Workflow/Interpreter/FunctionTable.hs +71/−0
- src/Control/Workflow/Interpreter/Graph.hs +90/−0
- src/Control/Workflow/Language.hs +168/−0
- src/Control/Workflow/Language/TH.hs +111/−0
- src/Control/Workflow/Language/TH/Internal.hs +66/−0
- src/Control/Workflow/Types.hs +74/−0
- src/Control/Workflow/Utils.hs +60/−0
- src/Scientific/Workflow.hs +0/−77
- src/Scientific/Workflow/Internal/Builder.hs +0/−368
- src/Scientific/Workflow/Internal/Builder/Types.hs +0/−113
- src/Scientific/Workflow/Internal/DB.hs +0/−121
- src/Scientific/Workflow/Internal/Utils.hs +0/−109
- src/Scientific/Workflow/Main.hs +0/−259
- src/Scientific/Workflow/Main/Options.hs +0/−120
- src/Scientific/Workflow/Types.hs +0/−130
- src/Scientific/Workflow/Visualize.hs +0/−57
LICENSE view
@@ -1,4 +1,4 @@-Copyright (c) 2015-2016 Kai Zhang+Copyright (c) 2015-2019 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
@@ -1,45 +1,1 @@-Scientific workflow management system-=====================================--Introduction---------------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 and safe: Provide a simple and flexible way to design type safe-computational pipelines in Haskell.--2. Automatic Checkpointing: The states of intermediate steps are automatically-logged, allowing easy restart upon failures.--3. Parallelism and grid computing support.--Examples-----------See examples in the "examples" directory for more details.--Use `ghc main.hs -threaded` to compile the examples.-And type `./main --help` to see available commands.--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 drmaa` flag.-Use `./main run --remote` to submit jobs to remote machines.--Featured applications-----------------------[Here](https://github.com/Taiji-pipeline)-are some bioinformatics pipelines built with SciFlow.+See https://github.com/kaizhang/SciFlow for README.
SciFlow.cabal view
@@ -1,15 +1,13 @@ name: SciFlow-version: 0.6.2+version: 0.7.0 synopsis: Scientific workflow management system-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.+description: SciFlow is a DSL for building type-safe computational workflows.+ SciFlow supports distributed computing through Cloud Haskell. license: MIT license-file: LICENSE author: Kai Zhang maintainer: kai@kzhang.org-copyright: (c) 2015-2018 Kai Zhang+copyright: (c) 2015-2019 Kai Zhang category: Control build-type: Simple cabal-version: >=1.10@@ -17,59 +15,50 @@ extra-source-files: README.md -Flag drmaa- Description: Enable DRMAA integration- Default: False- library ghc-options: -Wall exposed-modules:- Scientific.Workflow- Scientific.Workflow.Main- Scientific.Workflow.Main.Options- Scientific.Workflow.Types- Scientific.Workflow.Visualize- Scientific.Workflow.Internal.Builder- Scientific.Workflow.Internal.Builder.Types- Scientific.Workflow.Internal.DB- Scientific.Workflow.Internal.Utils+ Control.Workflow+ Control.Workflow.Language+ Control.Workflow.Language.TH+ Control.Workflow.Interpreter.Exec+ Control.Workflow.Interpreter.FunctionTable+ Control.Workflow.Interpreter.Graph+ Control.Workflow.Coordinator+ Control.Workflow.Coordinator.Local+ Control.Workflow.DataStore+ Control.Workflow.Types+ Control.Workflow.Utils other-modules:- Paths_SciFlow-- if flag(drmaa)- CPP-Options: -DDRMAA_ENABLED- build-depends: drmaa >=0.2.0+ Control.Arrow.Free+ Control.Arrow.Async+ Control.Workflow.Language.TH.Internal build-depends:- base >=4.7 && <5.0+ base >= 4.7 && < 5.0 , bytestring- , aeson- , containers- , cereal- , cereal-text- , directory- , data-default-class+ , binary+ , constraints+ , cryptohash-sha256+ , distributed-process == 0.7.*+ , distributed-process-monad-control , exceptions- , executable-path- , fgl- , graphviz- , lens >=4.0+ , hashable , lifted-async , mtl- , network- , optparse-applicative >=0.14.0.0+ , memory+ , monad-control+ , network-transport == 0.5.*+ , network-transport-tcp , rainbow+ , stm , sqlite-simple- , split- , th-lift- , th-lift-instances- , time- , temporary , text , template-haskell- , transformers- , yaml+ , time+ , th-lift-instances+ , unordered-containers hs-source-dirs: src default-language: Haskell2010
+ src/Control/Arrow/Async.hs view
@@ -0,0 +1,35 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE UndecidableInstances #-}+-- | Asynchronous arrows over monads with MonadBaseControl IO, using+-- lifted-async.+module Control.Arrow.Async where++import Control.Arrow+import Control.Category+import Control.Concurrent.Async.Lifted+import Control.Monad.Trans.Control (MonadBaseControl)+import Prelude hiding (id, (.))++newtype AsyncA m a b = AsyncA { runAsyncA :: a -> m b }++instance Monad m => Category (AsyncA m) where+ id = AsyncA return+ (AsyncA f) . (AsyncA g) = AsyncA (\b -> g b >>= f)++-- | @since 2.01+instance MonadBaseControl IO m => Arrow (AsyncA m) where+ arr f = AsyncA (return . f)+ first (AsyncA f) = AsyncA (\ ~(b,d) -> f b >>= \c -> return (c,d))+ second (AsyncA f) = AsyncA (\ ~(d,b) -> f b >>= \c -> return (d,c))+ (AsyncA f) *** (AsyncA g) = AsyncA $ \ ~(a,b) ->+ withAsync (f a) $ \c ->+ withAsync (g b) $ \d ->+ waitBoth c d++instance MonadBaseControl IO m => ArrowChoice (AsyncA m) where+ left f = f +++ arr id+ right f = arr id +++ f+ f +++ g = (f >>> arr Left) ||| (g >>> arr Right)+ AsyncA f ||| AsyncA g = AsyncA (either f g)
+ src/Control/Arrow/Free.hs view
@@ -0,0 +1,177 @@+-- {-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE Arrows #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE NoImplicitPrelude #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE Rank2Types #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}++-- | Various varieties of free arrow constructions.+--+-- For all of these constructions, there are only two important functions:+-- - 'eval' evaluates the free arrow in the context of another arrow.+-- - 'effect' lifts the underlying effect into the arrow.+--+-- The class 'FreeArrowLike', which is not exported from this module, exists+-- to allow these to be defined generally.+--+-- This module also defines some arrow combinators which are not exposed by+-- the standard arrow library.+module Control.Arrow.Free+ ( Free+ , Choice+ , effect+ , eval+ -- * Arrow functions+ , mapA+ , mapSeqA+ , filterA+ , type (~>)+ ) where++import Control.Arrow+import Control.Category+import Data.Bool (Bool)+import Data.Constraint (Constraint, Dict (..), mapDict, weaken1,+ weaken2)+import Data.Either (Either (..))+import Data.Function (const, flip, ($))+import Data.List (uncons)+import Data.Maybe (maybe)+import Data.Monoid (Monoid)+import Data.Tuple (uncurry)++-- | A natural transformation on type constructors of two arguments.+type x ~> y = forall a b. x a b -> y a b++--------------------------------------------------------------------------------+-- FreeArrowLike+--------------------------------------------------------------------------------++-- | Small class letting us define `eval` and `effect` generally over+-- multiple free structures+class FreeArrowLike fal where+ type Ctx fal :: (k -> k -> *) -> Constraint+ effect :: eff a b -> fal eff a b+ eval :: forall eff arr a b. ((Ctx fal) arr)+ => (eff ~> arr)+ -> fal eff a b+ -> arr a b++-- | Annoying hackery to let us tuple constraints and still use 'effect'+-- and 'eval'+class Join ( a :: k -> Constraint) (b :: k -> Constraint) (x :: k) where+ ctx :: Dict (a x, b x)+instance (a x, b x) => Join a b x where+ ctx = Dict++--------------------------------------------------------------------------------+-- Arrow+--------------------------------------------------------------------------------++-- | Freely generated arrows over an effect.+data Free eff a b where+ Pure :: (a -> b) -> Free eff a b+ Effect :: eff a b -> Free eff a b+ Seq :: Free eff a b -> Free eff b c -> Free eff a c+ Par :: Free eff a1 b1 -> Free eff a2 b2 -> Free eff (a1, a2) (b1, b2)++instance Category (Free eff) where+ id = Pure id+ (.) = flip Seq++instance Arrow (Free eff) where+ arr = Pure+ first f = Par f id+ second f = Par id f+ (***) = Par++instance FreeArrowLike Free where+ type Ctx Free = Arrow+ -- | Lift an effect into an arrow.+ effect :: eff a b -> Free eff a b+ effect = Effect++ -- | Evaluate given an implicit arrow+ eval :: forall eff arr a0 b0. (Arrow arr)+ => (eff ~> arr)+ -> Free eff a0 b0+ -> arr a0 b0+ eval exec = go+ where+ go :: forall a b. Free eff a b -> arr a b+ go freeA = case freeA of+ Pure f -> arr f+ Seq f1 f2 -> go f2 . go f1+ Par f1 f2 -> go f1 *** go f2+ Effect eff -> exec eff++--------------------------------------------------------------------------------+-- ArrowChoice+--------------------------------------------------------------------------------++-- | Freely generated `ArrowChoice` over an effect.+newtype Choice eff a b = Choice {+ runChoice :: forall ac. ArrowChoice ac => (eff ~> ac) -> ac a b+}++instance Category (Choice eff) where+ id = Choice $ const id+ Choice f . Choice g = Choice $ \x -> f x . g x++instance Arrow (Choice eff) where+ arr a = Choice $ const $ arr a+ first (Choice a) = Choice $ \f -> first (a f)+ second (Choice a) = Choice $ \f -> second (a f)+ (Choice a) *** (Choice b) = Choice $ \f -> a f *** b f++instance ArrowChoice (Choice eff) where+ left (Choice a) = Choice $ \f -> left (a f)+ right (Choice a) = Choice $ \f -> right (a f)+ (Choice a) ||| (Choice b) = Choice $ \f -> a f ||| b f++instance FreeArrowLike Choice where+ type Ctx Choice = ArrowChoice+ effect :: eff a b -> Choice eff a b+ effect a = Choice $ \f -> f a++ eval :: forall eff arr a0 b0. (ArrowChoice arr)+ => (eff ~> arr)+ -> Choice eff a0 b0+ -> arr a0 b0+ eval f a = runChoice a f++--------------------------------------------------------------------------------+-- Functions+--------------------------------------------------------------------------------++-- | Map an arrow over a list.+mapA :: ArrowChoice a => a b c -> a [b] [c]+mapA f = arr (maybe (Left ()) Right . uncons)+ >>> (arr (const []) ||| ((f *** mapA f) >>> arr (uncurry (:))))++-- | Map an arrow over a list, forcing sequencing between each element.+mapSeqA :: ArrowChoice a => a b c -> a [b] [c]+mapSeqA f = arr (maybe (Left ()) Right . uncons)+ >>> (arr (const []) ||| ((first f >>> second (mapSeqA f)) >>> arr (uncurry (:))))++-- | Filter a list given an arrow filter+filterA :: ArrowChoice a => a b Bool -> a [b] [b]+filterA f = proc xs ->+ case xs of+ [] -> returnA -< []+ (y:ys) -> do+ b <- f -< y+ if b then+ (second (filterA f) >>> arr (uncurry (:))) -< (y,ys)+ else+ filterA f -< ys++
+ src/Control/Workflow.hs view
@@ -0,0 +1,115 @@+--------------------------------------------------------------------------------+-- |+-- Module : Control.Workflow+-- Copyright : (c) 2015-2019 Kai Zhang+-- License : MIT+-- Maintainer : kai@kzhang.org+-- Stability : experimental+-- Portability : portable+--+-- DSL for building computational workflows. Example:+--+-- > {-# LANGUAGE TemplateHaskell #-}+-- > {-# LANGUAGE OverloadedStrings #-}+-- >+-- > import Control.Monad.Reader+-- > import Control.Concurrent (threadDelay)+-- > import Control.Lens+-- > import System.Environment+-- > import qualified Data.HashMap.Strict as M+-- > import Network.Transport.TCP+-- > import Data.Proxy (Proxy(..))+-- >+-- > import Control.Workflow+-- > import Control.Workflow.Coordinator.Local+-- > import Control.Workflow.Coordinator.Drmaa+-- >+-- > s0 :: () -> ReaderT Int IO [Int]+-- > s0 = return . const [1..10]+-- >+-- > s1 :: Int -> ReaderT Int IO Int+-- >s1 i = (i*) <$> ask +-- >+-- > s2 = return . (!!1)+-- > s3 = return . (!!2)+-- > s4 = return . (!!3)+-- > s5 = return . (!!4)+-- > s6 (a,b,c,d) = liftIO $ threadDelay 10000000 >> print [a,b,c,d]+-- > +-- > build "wf" [t| SciFlow Int |] $ do+-- > node "S0" 's0 $ return ()+-- > nodePar "S1" 's1 $ return ()+-- > ["S0"] ~> "S1"+-- > +-- > node "S2" 's2 $ memory .= 30+-- > node "S3" 's3 $ memory .= 30+-- > node "S4" 's4 $ nCore .= 4+-- > node "S5" 's5 $ queue .= Just "gpu"+-- > ["S0"] ~> "S2"+-- > ["S0"] ~> "S3"+-- > ["S0"] ~> "S4"+-- > ["S0"] ~> "S5"+-- > +-- > node "S6" 's6 $ return ()+-- > ["S2", "S3", "S4", "S5"] ~> "S6"++-- > main :: IO ()+-- > main = do+-- > let serverAddr = "192.168.0.1"+-- > port = 8888+-- > storePath = "sciflow.db"+-- > resources = ResourceConfig $ M.fromList+-- > [("S6", Resource (Just 2) Nothing Nothing)]+-- > [mode] <- getArgs+-- > case mode of+-- > -- Run on local machine+-- > "local" -> withCoordinator (LocalConfig 5) $ \coord -> do+-- > Right transport <- createTransport (defaultTCPAddr serverAddr $ show port)+-- > defaultTCPParameters+-- > withStore storePath $ \store -> +-- > runSciFlow coord transport store (ResourceConfig M.empty) 2 wf+-- > -- Using the DRMAA backend+-- > "drmaa" -> do+-- > config <- getDefaultDrmaaConfig ["slave"]+-- > withCoordinator config $ \coord -> do+-- > Right transport <- createTransport (defaultTCPAddr serverAddr $ show port)+-- > defaultTCPParameters+-- > withStore storePath $ \store -> +-- > runSciFlow coord transport store resources 2 wf+-- > -- DRMAA workers+-- > "slave" -> startClient (Proxy :: Proxy Drmaa)+-- > (mkNodeId serverAddr port) $ _function_table wf+--------------------------------------------------------------------------------++module Control.Workflow+ ( -- * Workflow types+ SciFlow(..)+ , ResourceConfig(..)+ , Resource(..)++ -- * Construct workflow+ , node+ , nodePar+ , nCore+ , memory+ , queue+ , (~>)+ , path+ , namespace+ , build+ + -- * Run workflow+ , withCoordinator+ , withStore+ , startClient+ , runSciFlow+ , mkNodeId+ ) where++import Control.Workflow.Types+import Control.Workflow.Language+import Control.Workflow.Language.TH+import Control.Workflow.Interpreter.Exec+import Control.Workflow.DataStore+import Control.Workflow.Coordinator+import Control.Workflow.Utils
+ src/Control/Workflow/Coordinator.hs view
@@ -0,0 +1,84 @@+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeFamilyDependencies #-}+--------------------------------------------------------------------------------+-- |+-- Module : Control.Workflow.Coordinator+-- Copyright : (c) 2019 Kai Zhang+-- License : MIT+-- Maintainer : kai@kzhang.org+-- Stability : experimental+-- Portability : portable+--+-- Coordinator needs to be able to discover new workers and send commands+-- to workers. The implementation of Coordinator thus contains server and+-- client parts. Server-side codes are executed by `withCoordinator` and +-- client-side codes are executed by `initClient`.+--+--------------------------------------------------------------------------------++module Control.Workflow.Coordinator+ ( Signal(..)+ , Worker(..)+ , WorkerStatus(..)+ , Coordinator(..)+ ) where++import Data.Binary (Binary)+import Control.Monad.Catch (MonadMask)+import GHC.Generics (Generic)+import Control.Monad.IO.Class (MonadIO)+import Control.Distributed.Process+import GHC.Conc (STM)+import Data.Proxy (Proxy(..))++import Control.Workflow.Types++class Coordinator coordinator where+ -- | Configuration+ type Config coordinator = config | config -> coordinator++ -- | Initialize Coordinator on the server.+ withCoordinator :: (MonadMask m, MonadIO m)+ => Config coordinator -> (coordinator -> m a) -> m a++ -- | Server initiation process+ initiate :: coordinator -> Process ()++ -- | Server shutdown process+ shutdown :: coordinator -> Process ()++ startClient :: Proxy coordinator -> NodeId -> FunctionTable -> IO ()++ -- | Get all workers.+ getWorkers :: coordinator -> STM [Worker]++ -- | Reserve a free worker. This function should block+ -- until a worker is reserved.+ reserve :: coordinator -> Maybe Resource -> Process ProcessId++ -- | Set a worker free so that it can be assigned other jobs.+ freeWorker :: MonadIO m => coordinator -> ProcessId -> m ()++-- | A worker.+data Worker = Worker+ { _worker_id :: ProcessId+ , _worker_status :: WorkerStatus+ , _worker_config :: Maybe Resource+ } deriving (Generic, Show)++instance Binary Worker++-- | The status of a worker.+data WorkerStatus = Idle+ | Working+ | ErrorExit String+ deriving (Eq, Generic, Show)++instance Binary WorkerStatus++data Signal = Shutdown deriving (Generic)++instance Binary Signal
+ src/Control/Workflow/Coordinator/Local.hs view
@@ -0,0 +1,48 @@+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE LambdaCase #-}++module Control.Workflow.Coordinator.Local+ ( LocalConfig(..)+ , Local+ ) where++import Control.Monad.IO.Class (liftIO)+import Control.Distributed.Process+import Control.Concurrent.STM+import Control.Concurrent (threadDelay)++import Control.Workflow.Coordinator++data LocalConfig = LocalConfig+ { _queue_size :: Int }++type WorkerCounter = TMVar Int++data Local = Local WorkerCounter LocalConfig++instance Coordinator Local where+ type Config Local = LocalConfig++ withCoordinator config f =+ (Local <$> liftIO (newTMVarIO 0) <*> return config) >>= f++ initiate _ = return ()+ shutdown _ = return ()+ startClient _ _ _ = return ()+ getWorkers _ = return []++ reserve (Local counter config) _ = liftIO tryReserve >> getSelfPid+ where+ tryReserve = do+ n <- atomically $ takeTMVar counter+ if n < _queue_size config+ then atomically $ putTMVar counter $ n + 1+ else do+ atomically $ putTMVar counter n+ threadDelay 1000000+ tryReserve+ + freeWorker (Local counter _) _ = liftIO $ atomically $ do+ n <- takeTMVar counter+ putTMVar counter $ n - 1
+ src/Control/Workflow/DataStore.hs view
@@ -0,0 +1,130 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}+module Control.Workflow.DataStore+ ( DataStore(..)+ , Key+ , mkKey+ , JobStatus(..)+ , openStore+ , closeStore+ , withStore+ , markPending+ , markFailed+ , saveItem+ , fetchItem+ , delRecord+ , queryStatus+ ) where++import Control.Monad (unless)+import Control.Concurrent.MVar (withMVar, newMVar, MVar)+import Control.Monad.Catch (MonadMask, bracket)+import Control.Monad.Identity (Identity(..))+import Data.Binary (Binary, encode, decode)+import Data.Typeable (Typeable, typeOf)+import Control.Monad.IO.Class (MonadIO, liftIO)+import Database.SQLite.Simple+import GHC.Generics (Generic)+import qualified Data.Text as T+import qualified Data.ByteString.Char8 as B+import qualified Crypto.Hash.SHA256 as C+import Data.ByteArray.Encoding (convertToBase, Base(..))++newtype DataStore = DataStore { _db_conn :: MVar Connection }++data Key = Key { _hash :: B.ByteString, _name :: T.Text }++instance Show Key where+ show (Key hash jn) = T.unpack jn <> "(" <> h <> ")"+ where+ h = B.unpack (B.take 4 hash) <> ".."++mkKey :: (Typeable i, Binary i) => i -> T.Text -> Key+mkKey input nm = Key hash nm+ where+ hash = convertToBase Base16 $ C.hashlazy $ encode (nm, show $ typeOf input, input)+{-# INLINE mkKey #-}++data JobStatus = Pending+ | Complete+ | Failed String+ deriving (Eq, Generic, Show)++instance Binary JobStatus++openStore :: MonadIO m => FilePath -> m DataStore+openStore root = liftIO $ do+ db <- open root+ itemExist <- hasTable db "item_db"+ metaExist <- hasTable db "meta_db"+ unless (itemExist && metaExist) $ do+ execute_ db "CREATE TABLE item_db(hash TEXT PRIMARY KEY, data BLOB)"+ execute_ db+ "CREATE TABLE meta_db(hash TEXT PRIMARY KEY, jobname TEXT, status BLOB)"+ execute_ db "CREATE INDEX jobname_index ON meta_db(jobname)"+ DataStore <$> newMVar db+{-# INLINE openStore #-}++closeStore :: MonadIO m => DataStore -> m ()+closeStore (DataStore db) = liftIO $ withMVar db close+{-# INLINE closeStore #-}++withStore :: (MonadIO m, MonadMask m)+ => FilePath -> (DataStore -> m a) -> m a+withStore root = bracket (openStore root) closeStore+{-# INLINE withStore #-}++markPending :: MonadIO m => DataStore -> Key -> m ()+markPending (DataStore store) (Key k n) = liftIO $ withMVar store $ \db ->+ execute db "REPLACE INTO meta_db VALUES (?, ?, ?)" (k, n, encode Pending)+{-# INLINE markPending #-}++markFailed :: MonadIO m => DataStore -> Key -> String -> m ()+markFailed (DataStore store) (Key k n) msg = liftIO $ withMVar store $ \db ->+ execute db "REPLACE INTO meta_db VALUES (?, ?, ?)" (k, n, encode $ Failed msg)+{-# INLINE markFailed #-}++queryStatus :: MonadIO m => DataStore -> Key -> m (Maybe JobStatus)+queryStatus (DataStore store) (Key k _) = liftIO $ withMVar store $ \db ->+ query db "SELECT status FROM meta_db WHERE hash=?" [k] >>= \case+ [Only result] -> return $ Just $ decode result+ _ -> return Nothing+{-# INLINE queryStatus #-}++saveItem :: (MonadIO m, Binary a) => DataStore -> Key -> a -> m ()+saveItem (DataStore store) (Key k n) res = liftIO $ withMVar store $ \db -> do+ execute db "REPLACE INTO item_db VALUES (?, ?)" (k, encode res)+ execute db "REPLACE INTO meta_db VALUES (?, ?, ?)" (k, n, encode Complete)+{-# INLINE saveItem #-}++fetchItem :: (MonadIO m, Binary a) => DataStore -> Key -> m a+fetchItem (DataStore store) (Key k _) = liftIO $ withMVar store $ \db -> do+ query db "SELECT data FROM item_db WHERE hash=?" [k] >>= \case+ [Only result] -> return $ decode result+ _ -> error "Item not found"+{-# INLINE fetchItem #-}++delRecord :: MonadIO m => DataStore -> Key -> m ()+delRecord (DataStore store) (Key k _) = liftIO $ withMVar store $ \db -> do+ execute db "DELETE FROM meta_db WHERE hash= ?" [k]+ execute db "DELETE FROM item_db WHERE hash= ?" [k]+{-# INLINE delRecord #-}++-------------------------------------------------------------------------------+-- Low level functions+-------------------------------------------------------------------------------++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])+{-# INLINE hasTable #-}++{-+getKeys :: Connection -> IO [Key]+getKeys db = concat <$> query_ db+ (Query $ T.pack $ printf "SELECT pid FROM %s;" dbTableName)+-}
+ src/Control/Workflow/Interpreter/Exec.hs view
@@ -0,0 +1,114 @@+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE RankNTypes #-}++module Control.Workflow.Interpreter.Exec (runSciFlow) where++import Control.Arrow.Async+import Control.Arrow.Free (eval)+import Control.Monad.Reader+import Control.Monad.Except (ExceptT, throwError, runExceptT)+import Control.Monad.Catch (SomeException(..), handleAll)+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Trans (lift)+import Control.Distributed.Process (kill, processNodeId, call)+import Control.Distributed.Process.Node (forkProcess, runProcess, newLocalNode, LocalNode)+import Control.Distributed.Process.MonadBaseControl ()+import qualified Data.HashMap.Strict as M+import Network.Transport (Transport)+import Data.Binary (Binary(..), encode, decode)+import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar++import Control.Workflow.Types+import Control.Workflow.Utils+import Control.Workflow.Coordinator+import Control.Workflow.DataStore++runSciFlow :: (Coordinator coordinator, Binary env)+ => coordinator -- ^ Coordinator backend+ -> Transport -- ^ Cloud Haskell transport+ -> DataStore -- ^ Local cache+ -> ResourceConfig -- ^ Job resource configuration+ -> env -- ^ Optional environmental variables+ -> SciFlow env+ -> IO ()+runSciFlow coord transport store resource env sciflow = do+ nd <- newLocalNode transport $ _rtable $ _function_table sciflow+ pidInit <- forkProcess nd $ initiate coord+ runProcess nd $ do+ res <- liftIO $ flip runReaderT resource $ runExceptT $+ runAsyncA (execFlow nd coord store env sciflow) () + shutdown coord+ case res of+ Left ex -> errorS "Program exit with errors"+ Right _ -> infoS "Program finish successfully"+ kill pidInit "Exit"+{-# INLINE runSciFlow #-}++type FlowMonad = ExceptT String (ReaderT ResourceConfig IO)++-- | Flow interpreter.+execFlow :: forall coordinator env . (Coordinator coordinator, Binary env)+ => LocalNode+ -> coordinator+ -> DataStore+ -> env+ -> SciFlow env+ -> AsyncA FlowMonad () ()+execFlow localNode coord store env sciflow = eval (AsyncA . runFlow') $ _flow sciflow+ where+ runFlow' (Step w) = runJob localNode coord store (_function_table sciflow) env w+{-# INLINE execFlow #-}++runJob :: (Coordinator coordinator, Binary env)+ => LocalNode+ -> coordinator+ -> DataStore+ -> FunctionTable+ -> env+ -> Job env i o+ -> (i -> FlowMonad o)+runJob localNode coord store rf env Job{..} = runAsyncA $ eval ( \(Action _) ->+ AsyncA $ \i -> do+ let chash = mkKey i _job_name+ cleanUp (SomeException ex) = do+ throwError $ show ex+ input | _job_parallel = encode [i]+ | otherwise = encode i+ decode' x | _job_parallel = let [r] = decode x in r+ | otherwise = decode x+ go = queryStatus store chash >>= \case+ Just Pending -> liftIO (threadDelay 1000) >> go+ Just (Failed msg) -> throwError msg+ Just Complete -> fetchItem store chash+ Nothing -> handleAll cleanUp $ do+ -- A Hack, because `runProcess` cannot return value.+ result <- liftIO newEmptyMVar+ infoS $ show chash <> ": Running ..."+ jobRes <- lift $ reader (M.lookup _job_name . _resource_config) >>= \case+ Nothing -> return _job_resource+ r -> return r+ liftIO $ runProcess localNode $ do+ pid <- reserve coord jobRes+ call (_dict rf) (processNodeId pid)+ ((_table rf) (_job_name, encode env, input)) >>=+ liftIO . putMVar result+ freeWorker coord pid+ liftIO (takeMVar result) >>= \case+ Left msg -> do+ errorS $ show chash <> " Failed: " <> msg+ throwError msg+ Right r -> do+ let res = decode' r+ saveItem store chash res+ infoS $ show chash <> ": Complete!"+ return res+ go+ ) _job_action+{-# INLINE runJob #-}
+ src/Control/Workflow/Interpreter/FunctionTable.hs view
@@ -0,0 +1,71 @@+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE LambdaCase #-}+-- This interpreter treat a workflow as a function lookup table. +-- Given a (key, state, input) tuple where state and input are in binary format+-- , the table will return the output in binary format.++module Control.Workflow.Interpreter.FunctionTable+ ( mkDict+ , mkFunTable+ ) where++import Data.Binary+import Language.Haskell.TH+import qualified Data.ByteString.Lazy as B+import qualified Data.Text as T+import Control.Arrow.Free (Free, eval)+import Control.Arrow+import Control.Distributed.Process.Closure (functionTDict, mkClosure, remotableDecl)+import Control.Distributed.Process.Node (initRemoteTable)+import Control.Monad.Reader+import Control.Monad.Catch (SomeException(..), catch)+import qualified Control.Category as C+import Control.Distributed.Process (Process)+import Control.Concurrent.MVar++import Control.Workflow.Types++mkFunTable :: String -- ^ Name of the table+ -> String -- ^ SciFlow env i o+ -> Q [Dec]+mkFunTable nm wf = remotableDecl [sig, fun, dec]+ where+ funName = mkName $ nm ++ "__dict"+ sig = fmap return $ funName `sigD`+ [t| (T.Text, B.ByteString, B.ByteString) -> Process (Either String B.ByteString) |]+ fun = [d| $(varP funName) = liftIO . mkDict $(varE $ mkName wf) |]+ dec = [d| $(varP $ mkName nm) = FunctionTable $(mkClosure funName) $(functionTDict funName) $ __remoteTableDecl initRemoteTable |]++-- | Function table+type Dictionary = (T.Text, B.ByteString, B.ByteString) -- ^ Input+ -> IO (Either String B.ByteString) -- ^ Output++mkDict :: Binary env+ => Free (Flow env) i o+ -> Dictionary+mkDict flow (nm, env, input) = do+ res <- newMVar $ Left ""+ unA $ eval (go res) flow+ readMVar res+ where+ go res (Step job) = A $ modifyMVar_ res $ \case+ Left msg -> if nm == _job_name job+ then catch (runJob job) $ \(SomeException e) -> return $ Left $ show e+ else return $ Left msg+ x -> return x+ where+ runJob job = Right . encode <$>+ runReaderT (f (decode input)) (decode env)+ where+ f = runKleisli $ eval (Kleisli . _unAction) $ _job_action job++-- | Helper type+data A a b = A { unA :: IO () }++instance C.Category A where+ id = A $ return ()+ A f . A g = A (g >> f)++instance Arrow A where+ arr _ = A $ return ()+ A f *** A g = A (f >> g)
+ src/Control/Workflow/Interpreter/Graph.hs view
@@ -0,0 +1,90 @@+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Workflow.Interpreter.Graph+ ( mkGraph+ , Graph(..)+ , Node(..)+ , Edge(..)+ ) where++import Control.Arrow (Arrow(..))+import Control.Arrow.Free (Free, eval)+import Control.Category+import qualified Data.Text as T+import Prelude hiding (id, (.))+import qualified Data.HashSet as S+import Data.Hashable (Hashable(..))++import Control.Workflow.Types++mkGraph :: SciFlow env -> Graph+mkGraph flow = Graph ns $ map (\(a,b) -> Edge (_id a) $ _id b) es+ where+ ns = S.toList $ S.fromList $ concatMap (\(a,b) -> [a,b]) es+ es = S.toList $ S.fromList $ toEdges $ toDiagram $ _flow flow+{-# INLINE mkGraph #-}++data Graph = Graph+ { _nodes :: [Node]+ , _edges :: [Edge] }++data Node = Node+ { _id :: T.Text+ , _label :: T.Text+ , _descr :: T.Text }++instance Hashable Node where+ hashWithSalt s = hashWithSalt s . _id++instance Eq Node where+ a == b = _id a == _id b++data Edge = Edge+ { _from :: T.Text+ , _to :: T.Text }++toEdges :: Diagram a b -> [(Node, Node)]+toEdges (Seq f g) = map (\[a,b] -> (a,b)) (sequence [lastD f, headD g]) +++ toEdges f ++ toEdges g+toEdges (Par f g) = toEdges f ++ toEdges g+toEdges _ = []+{-# INLINE toEdges #-}++headD :: Diagram a b -> [Node]+headD (S nd) = [nd]+headD (Seq Ident g) = headD g+headD (Seq f _) = headD f+headD (Par f g) = headD f ++ headD g+headD _ = []+{-# INLINE headD #-}++lastD :: Diagram a b -> [Node]+lastD (S nd) = [nd]+lastD (Seq f Ident) = lastD f+lastD (Seq _ g) = lastD g+lastD (Par f g) = lastD f ++ lastD g+lastD _ = []+{-# INLINE lastD #-}++toDiagram :: Free (Flow env) a b -> Diagram a b+toDiagram = eval toDiagram'+ where+ toDiagram' (Step Job{..}) = S (Node _job_name _job_name _job_descr)+{-# INLINE toDiagram #-}+ +data Diagram a b where+ Ident :: Diagram a b+ S :: Node -> Diagram a b+ Seq :: Diagram a b -> Diagram b c -> Diagram a c+ Par :: Diagram a b -> Diagram c d -> Diagram (a,c) (b,d)+ +instance Category Diagram where+ id = Ident+ (.) = flip Seq+ +instance Arrow Diagram where+ arr = const Ident+ (***) = Par+
+ src/Control/Workflow/Language.hs view
@@ -0,0 +1,168 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Workflow.Language+ ( -- * Defining workflows+ node+ , nodePar+ , (~>)+ , path+ , namespace+ , Workflow(..)+ , Builder++ -- * Lens for Attributes + , doc+ , nCore+ , memory+ , queue+ , Node(..)+ , NodeAttributes++ , THExp(..)+ ) where++import Control.Arrow+import qualified Data.Text as T+import Control.Monad.State.Lazy (State)+import qualified Data.HashMap.Strict as M+import Control.Monad.State.Lazy (modify, execState)+import Data.Maybe (isNothing)+import Language.Haskell.TH (ExpQ, Name, varE)++import Control.Workflow.Types (Resource(..))++-- | A computation node.+data Node = Node+ { _node_function :: ExpQ -- ^ a function with type: a -> ReaderT env IO b+ , _node_job_resource :: Maybe Resource -- ^ Computational resource config+ , _node_parallel :: Bool -- ^ Should the job be run in parallel+ , _node_doc :: T.Text -- ^ Documentation+ }++data NodeAttributes = NodeAttributes+ { _doc :: T.Text -- ^ documentation+ , _nCore :: Int+ , _memory :: Int+ , _queue :: Maybe String }++-- | Node description.+doc :: Functor f => (T.Text -> f T.Text) -> NodeAttributes -> f NodeAttributes+doc x y = fmap (\newX -> y { _doc = newX }) (x (_doc y))+{-# INLINE doc #-}++-- | Number of cores.+nCore :: Functor f => (Int -> f Int) -> NodeAttributes -> f NodeAttributes+nCore x y = fmap (\newX -> y { _nCore = newX }) (x (_nCore y))+{-# INLINE nCore #-}++-- | Total memory.+memory :: Functor f => (Int -> f Int) -> NodeAttributes -> f NodeAttributes+memory x y = fmap (\newX -> y { _memory = newX }) (x (_memory y))+{-# INLINE memory #-}++-- | Job queue.+queue :: Functor f+ => (Maybe String -> f (Maybe String))+ -> NodeAttributes -> f NodeAttributes+queue x y = fmap (\newX -> y { _queue = newX }) (x (_queue y))+{-# INLINE queue #-}++mkNode :: THExp q+ => q -- ^ Template Haskell expression representing+ -- functions with type @a -> IO b@.+ -> State NodeAttributes ()+ -> Node+mkNode fun attrSetter = Node (mkExp fun) res False $ _doc attr+ where+ res | isNothing core && isNothing mem && isNothing (_queue attr) = Nothing+ | otherwise = Just $ Resource core mem $ _queue attr+ core = if _nCore attr > 1 then Just $ _nCore attr else Nothing+ mem = if _memory attr > 0 then Just $ _memory attr else Nothing+ attr = execState attrSetter $ NodeAttributes+ { _doc = ""+ , _nCore = 1+ , _memory = -1+ , _queue = Nothing }+{-# INLINE mkNode #-}++-- | Workflow declaration, containing a map of nodes and their parental processes.+data Workflow = Workflow+ { _nodes :: M.HashMap T.Text Node+ , _parents :: M.HashMap T.Text [T.Text] }++instance Semigroup Workflow where+ x <> y = Workflow (_nodes x <> _nodes y) (_parents x <> _parents y)++type Builder = State Workflow++-- | Define a step.+node :: THExp q+ => T.Text -- ^ Node id+ -> q -- ^ Template Haskell expression representing+ -- functions with type @a -> ReaderT env IO b@.+ -> State NodeAttributes () -- ^ Option setter+ -> Builder ()+node i f attrSetter = modify $ \wf ->+ wf{ _nodes = M.insertWith undefined i nd $ _nodes wf }+ where+ nd = mkNode f attrSetter+{-# INLINE node #-}++-- | Define a step that will be executed in parallel, i.e.,+-- @a -> m b@ becomes @[a] -> m [b]@.+nodePar :: THExp q+ => T.Text -- ^ Node id+ -> q -- ^ Template Haskell expression representing+ -- functions with type @a -> ReaderT env IO b@.+ -> State NodeAttributes ()+ -> Builder ()+nodePar i f attrSetter = modify $ \wf ->+ wf{ _nodes = M.insertWith undefined i nd{_node_parallel=True} $ _nodes wf }+ where+ nd = mkNode f attrSetter+{-# INLINE nodePar #-}++linkFromTo :: [T.Text] -> T.Text -> Builder ()+linkFromTo ps to = modify $ \wf ->+ wf{ _parents = M.insertWith undefined to ps $ _parents wf }+{-# INLINE linkFromTo #-}++-- | Connect nodes.+-- Example:+--+-- > node "step1" [| \() -> return 1 |] $ return ()+-- > node "step2" [| \() -> return 2 |] $ return ()+-- > node "step3" [| \(x, y) -> x * y |] $ return ()+-- > ["step1", "step2"] ~> "step3"+(~>) :: [T.Text] -> T.Text -> Builder ()+(~>) = linkFromTo+{-# INLINE (~>) #-}++-- | @'path' [a, b, c]@ is equivalent to @[a] ~> b >> [b] ~> c@+path :: [T.Text] -> Builder ()+path ns = sequence_ $ zipWith linkFromTo (map return $ init ns) $ tail ns+{-# 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 = modify (st <>)+ where+ st = execState (builder >> addPrefix) $ Workflow M.empty M.empty+ addPrefix = modify $ \Workflow{..} ->+ let nodes = M.fromList $ map (first add) $ M.toList _nodes + parents = M.fromList $ map (add *** map add) $ M.toList _parents+ in Workflow nodes parents+ add x = prefix <> "_" <> x+{-# INLINE namespace #-}++class THExp q where+ mkExp :: q -> ExpQ++instance THExp Name where+ mkExp = varE++instance THExp ExpQ where+ mkExp = id
+ src/Control/Workflow/Language/TH.hs view
@@ -0,0 +1,111 @@+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RecordWildCards #-}++module Control.Workflow.Language.TH (build) where++import Control.Arrow.Free (mapA, effect)+import Control.Arrow (arr)+import qualified Data.Text as T+import Language.Haskell.TH+import Instances.TH.Lift ()+import qualified Data.HashMap.Strict as M+import qualified Data.HashSet as S+import Control.Monad.State.Lazy (StateT, get, put, lift, execStateT, execState)++import Control.Workflow.Language+import Control.Workflow.Types+import Control.Workflow.Interpreter.FunctionTable (mkFunTable)+import Control.Workflow.Language.TH.Internal++-- | Generate template haskell codes to build the workflow.+build :: String -- ^ The name of the compiled workflow.+ -> TypeQ -- ^ The workflow signature.+ -> Builder () -- ^ Worflow builder.+ -> Q [Dec]+build name sig builder = compile name sig wf+ where+ wf = execState builder $ Workflow M.empty M.empty+{-# INLINE build #-}++-- 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.+compile :: String -- ^ The name of the compiled workflow+ -> TypeQ -- ^ The function signature+ -> Workflow+ -> Q [Dec]+compile name sig wf = do+ d1 <- defFlow wfName+ d2 <- mkFunTable (name ++ "__Table") (name ++ "__Flow")+ -- the function signature+ wf_signature <- (mkName name) `sigD` sig+ d3 <- [d| $(varP $ mkName name) = SciFlow $(varE wfName) $(varE tableName) |]+ return $ d1 ++ d2 ++ (wf_signature:d3)+ where+ tableName = mkName $ name ++ "__Table"+ wfName = mkName $ name ++ "__Flow"+ defFlow nm = do+ -- step function definitions+ res <- mapM (mkDefs wf) $ getSinks wf+ let funDecs = M.elems $ M.fromList $ concatMap snd res++ -- main definition+ main <- link (map fst res) [| arr $ const () |]++ return [ValD (VarP nm) (NormalB main) funDecs]+{-# INLINE compile #-}++type FunDef = (String, Dec)++-- Create function definitions for the target node and its ancestors.+-- Return the function name of the target node and all relevant function+-- definitions.+mkDefs :: Workflow+ -> T.Text+ -> Q (String, [FunDef])+mkDefs wf x = do+ funDefs <- execStateT (define x) M.empty+ return (fst $ M.lookupDefault undefined x funDefs, M.elems funDefs)+ where+ define :: T.Text+ -> StateT (M.HashMap T.Text FunDef) Q ()+ define nid = do+ mapM_ define ps+ funDefs <- get + let parentNames = map (fst . flip (M.lookupDefault undefined) funDefs) ps+ e <- lift $ link parentNames $ mkJob nid $+ M.lookupDefault (errMsg nid) nid $ _nodes wf+ let dec = (ndName, ValD (VarP $ mkName ndName) (NormalB e) [])+ put $ M.insert nid dec funDefs+ where+ ps = M.lookupDefault [] nid $ _parents wf+ ndName = T.unpack $ "f_" <> nid+ errMsg = error . ("Node not found: " ++) . T.unpack+{-# INLINE mkDefs #-}+ +-- | Get all the sinks, i.e., nodes with no children.+getSinks :: Workflow -> [T.Text]+getSinks wf = filter (\x -> not $ S.member x ps) $ M.keys $ _nodes wf+ where+ ps = S.fromList $ concat $ M.elems $ _parents wf+{-# INLINE getSinks #-}++mkJob :: T.Text -> Node -> ExpQ+mkJob nid Node{..}+ | _node_parallel = [| step $ Job+ { _job_name = nid+ , _job_descr = _node_doc+ , _job_resource = _node_job_resource+ , _job_parallel = True+ , _job_action = mapA $ effect $ Action $_node_function + } |]+ | otherwise = [| step $ Job+ { _job_name = nid+ , _job_descr = _node_doc+ , _job_resource = _node_job_resource+ , _job_parallel = False+ , _job_action = effect $ Action $_node_function + } |]+{-# INLINE mkJob #-}
+ src/Control/Workflow/Language/TH/Internal.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE TemplateHaskell #-}++module Control.Workflow.Language.TH.Internal (link) where++import Control.Arrow+import Data.List (foldr1)+import Language.Haskell.TH++link :: [String] -- a list of parents+ -> ExpQ -- child+ -> ExpQ+link [] x = x+link [s] x = [| $(varE $ mkName s) >>> $x |]+link [s1,s2] x = [| linkA2 $(varE $ mkName s1) $(varE $ mkName s2) $x |]+link [s1,s2,s3] x = [| linkA3 $(varE $ mkName s1) $(varE $ mkName s2) + $(varE $ mkName s3) $x |]+link [s1,s2,s3,s4] x = [| linkA4 $(varE $ mkName s1) $(varE $ mkName s2) + $(varE $ mkName s3) $(varE $ mkName s4) $x |]+link [s1,s2,s3,s4,s5] x = [| linkA5 $(varE $ mkName s1) $(varE $ mkName s2) + $(varE $ mkName s3) $(varE $ mkName s4) $(varE $ mkName s5) $x |]+link xs x = linkAN (map mkName xs) x+{-# INLINE link #-}++linkA2 :: Arrow arr => arr a b1 -> arr a b2 -> arr (b1, b2) c -> arr a c+linkA2 a1 a2 f = (a1 &&& a2) >>> f+{-# INLINE linkA2 #-}++linkA3 :: Arrow arr => arr a b1 -> arr a b2 -> arr a b3+ -> arr (b1, b2, b3) c+ -> arr a c+linkA3 a1 a2 a3 f = (a1 &&& a2 &&& a3) >>>+ arr (\(b1,(b2,b3)) -> (b1,b2,b3)) >>> f+{-# INLINE linkA3 #-}++linkA4 :: Arrow arr => arr a b1 -> arr a b2 -> arr a b3 -> arr a b4+ -> arr (b1, b2, b3, b4) c+ -> arr a c+linkA4 a1 a2 a3 a4 f = (a1 &&& a2 &&& a3 &&& a4) >>>+ arr (\(b1,(b2,(b3,b4))) -> (b1,b2,b3,b4)) >>> f+{-# INLINE linkA4 #-}++linkA5 :: Arrow arr => arr a b1 -> arr a b2 -> arr a b3 -> arr a b4 -> arr a b5+ -> arr (b1, b2, b3, b4, b5) c+ -> arr a c+linkA5 a1 a2 a3 a4 a5 f = (a1 &&& a2 &&& a3 &&& a4 &&& a5) >>>+ arr (\(b1,(b2,(b3,(b4,b5)))) -> (b1,b2,b3,b4,b5)) >>> f+{-# INLINE linkA5 #-}++linkAN :: [Name] -- ^ a list of Arrows+ -> ExpQ+ -> ExpQ+linkAN as f = [| $arr1 >>> arr $arr2 >>> $f |]+ where+ arr1 = return $ foldr1 g $ map VarE as+ where+ g x1 x2 = AppE (AppE (VarE '(&&&)) x1) x2+ arr2 = return $ LamE [tuple1] $ TupE $ map VarE vars+ where+ tuple1 = go $ map VarP vars+ where+ go [x] = x+ go (x:xs) = TupP [x, go xs]+ go _ = error "empty list"+ vars = map (\i -> mkName $ "x" ++ show i) ([1..n] :: [Int])+ n = length as+{-# INLINE linkAN #-}
+ src/Control/Workflow/Types.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE DeriveLift #-}+{-# LANGUAGE GADTs #-}+module Control.Workflow.Types+ ( SciFlow(..)+ , FunctionTable(..)+ , ResourceConfig(..)+ , Resource(..)+ , Job(..)+ , Action(..)+ , Flow(..)+ , step+ ) where++import Data.Binary (Binary)+import Control.Monad.Reader+import GHC.Generics (Generic)+import qualified Data.Text as T+import Data.Typeable (Typeable)+import Control.Arrow.Free (Free, Choice, effect)+import qualified Data.ByteString.Lazy as B+import qualified Data.HashMap.Strict as M+import Control.Distributed.Process.Serializable (SerializableDict)+import Control.Distributed.Process (Process, RemoteTable, Closure, Static)+import Language.Haskell.TH.Syntax (Lift)++-- | The core type, containing the workflow represented as a free arrow and +-- a function table for remote execution.+data SciFlow env = SciFlow+ { _flow :: Free (Flow env) () ()+ , _function_table :: FunctionTable }++-- | The function table that can be sent to remote.+data FunctionTable = FunctionTable+ { _table :: (T.Text, B.ByteString, B.ByteString)+ -> Closure (Process (Either String B.ByteString))+ , _dict :: Static (SerializableDict (Either String B.ByteString))+ , _rtable :: RemoteTable }++-- | Global job specific resource configuration. This will overwrite any+-- existing configuration.+newtype ResourceConfig = ResourceConfig+ { _resource_config :: M.HashMap T.Text Resource }++-- | The basic component/step of a workflow.+data Job env i o = Job+ { _job_name :: T.Text -- ^ The name of the job+ , _job_descr :: T.Text -- ^ The description of the job+ , _job_resource :: Maybe Resource -- ^ The computational resource needed+ , _job_parallel :: Bool -- ^ Whether to run this step in parallel+ , _job_action :: Choice (Action env) i o } -- ^ The action to run++data Action env i o where+ Action :: (Typeable i, Typeable o, Binary i, Binary o) =>+ { _unAction :: i -> ReaderT env IO o -- ^ The function to run+ } -> Action env i o++-- | Free arrow side effect.+data Flow env i o where+ Step :: (Binary i, Binary o) => Job env i o -> Flow env i o++step :: (Binary i, Binary o) => Job env i o -> Free (Flow env) i o+step job = effect $ Step job++-- | Computational resource+data Resource = Resource+ { _num_cpu :: Maybe Int -- ^ The number of CPU needed+ , _total_memory :: Maybe Int -- ^ Memory in GB+ , _submit_queue :: Maybe String -- ^ Job submitting queue+ } deriving (Eq, Generic, Show, Lift)++instance Binary Resource
+ src/Control/Workflow/Utils.hs view
@@ -0,0 +1,60 @@+{-# LANGUAGE OverloadedStrings #-}++module Control.Workflow.Utils+ ( infoS+ , warnS+ , errorS+ , mkNodeId+ ) where++import qualified Data.ByteString.Char8 as B+import Network.Transport (EndPointAddress(..))+import Control.Distributed.Process (NodeId(..))+import Data.Time (defaultTimeLocale, formatTime,+ getZonedTime)+import Rainbow+import System.IO+import Control.Monad.IO.Class (MonadIO, liftIO)+ +-- | Pretty print info messages.+infoS :: MonadIO m => String -> m ()+infoS txt = liftIO $ do+ t <- getTime+ let prefix = bold $ chunk ("[INFO]" ++ t ++ " ") & fore green+ msg = B.concat $ chunksToByteStrings toByteStringsColors8+ [prefix, chunk txt & fore green]+ B.hPutStrLn stderr msg+{-# INLINE infoS #-}+ +-- | Pretty print error messages.+errorS :: MonadIO m => String -> m ()+errorS txt = liftIO $ 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+{-# INLINE errorS #-}+ +-- | Pretty print warning messages.+warnS :: MonadIO m => String -> m ()+warnS txt = liftIO $ 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+{-# INLINE warnS #-}++-- | Get current time.+getTime :: IO String+getTime = formatTime defaultTimeLocale "[%m-%d %H:%M]" <$> getZonedTime+{-# INLINE getTime #-}++-- | Construct node id given server address and port.+mkNodeId :: String -- ^ Server address+ -> Int -- ^ Server port+ -> NodeId+mkNodeId ip port = NodeId $ EndPointAddress $ B.intercalate ":" $+ [B.pack ip, B.pack $ show $ port, "0"]+{-# INLINE mkNodeId #-}
− src/Scientific/Workflow.hs
@@ -1,77 +0,0 @@-{-|-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--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.--Features:--1. Easy to use and safe: Provide a simple and flexible way to design type safe-computational pipelines in Haskell.--2. Automatic Checkpointing: The states of intermediate steps are automatically-logged, allowing easy restart upon failures.--3. Parallelism and grid computing support.--Example:--> 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- , Workflow(..)- ) where--import Scientific.Workflow.Internal.Builder-import Scientific.Workflow.Internal.Builder.Types-import Scientific.Workflow.Main-import Scientific.Workflow.Types
− src/Scientific/Workflow/Internal/Builder.hs
@@ -1,368 +0,0 @@-{-# 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 = modify (st <>)- where- st = execState (builder >> addPrefix) ([], [])- 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- fmap deserialize $ 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 (serialize 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 <- fmap deserialize $ 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 (serialize r) $ wfState^.database- putMVar nodeSt $ Special Skip- return r-{-# INLINE handleSpecialMode #-}
− src/Scientific/Workflow/Internal/Builder/Types.hs
@@ -1,113 +0,0 @@-{-# 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
@@ -1,121 +0,0 @@-{-# LANGUAGE ConstraintKinds #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE UndecidableInstances #-}-module Scientific.Workflow.Internal.DB- ( openDB- , closeDB- , readData- , 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)------------------------------------------------------------------------------------- Data Serialization------------------------------------------------------------------------------------- | @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----- | An abstract type representing the database used to store states of workflow-newtype WorkflowDB = WorkflowDB Connection-type Key = T.Text-type Val = B.ByteString---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 :: Key -> WorkflowDB -> IO Val-readData pid (WorkflowDB db) = do- [Only result] <- query db (Query $ T.pack $- printf "SELECT data FROM %s WHERE pid=?" dbTableName) [pid]- return result-{-# INLINE readData #-}--updateData :: Key -> Val -> WorkflowDB -> IO ()-updateData pid result (WorkflowDB db) = execute db (Query $ T.pack $- printf "UPDATE %s SET data=? WHERE pid=?" dbTableName) (result, pid)-{-# INLINE updateData #-}--saveData :: Key -> Val -> WorkflowDB -> IO ()-saveData pid result (WorkflowDB db) = execute db (Query $ T.pack $- printf "INSERT INTO %s VALUES (?, ?)" dbTableName) (pid, result)-{-# INLINE saveData #-}--isFinished :: Key -> 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 [Key]-getKeys (WorkflowDB db) = concat <$> query_ db (Query $ T.pack $- printf "SELECT pid FROM %s;" dbTableName)-{-# INLINE getKeys #-}--delRecord :: Key -> 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
@@ -1,109 +0,0 @@-{-# 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
@@ -1,259 +0,0 @@-{-# LANGUAGE CPP #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TemplateHaskell #-}--module Scientific.Workflow.Main- ( defaultMain- , defaultMainOpts- , mainWith- , MainOpts(..)- , runWorkflow- ) where--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)--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, decodeEither)--#ifdef DRMAA_ENABLED-import DRMAA (withSession)-#endif--import Language.Haskell.TH-import qualified Language.Haskell.TH.Lift as T-import Options.Applicative hiding (Success)-import Text.Printf (printf)--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--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 -- ^ Short description about the program.- , workflowConfigType :: Maybe Name -- ^ The type of workflow config. Default is- -- @Nothing@ which let type inference do its job.- }--T.deriveLift ''MainOpts--defaultMainOpts :: MainOpts-defaultMainOpts = MainOpts- { preAction = 'id- , programHeader = printf "SciFlow-%s" (showVersion version)- , workflowConfigType = Nothing- }--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 $ wfType ++ wf_q ++ main_q- where- wfType = case workflowConfigType opts of- Nothing -> []- Just ty -> [SigD (mkName wfName) $ AppT (ConT $ mkName "Workflow") $- ConT ty]- wfName = "sciFlowDefaultMain"- dag = nmap (\x -> (_nodePid x, _nodeAttr x)) $ mkDAG builder-{-# INLINE mainWith #-}--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- then initialize $ runWorkflow wf runOpts-#endif- else runWorkflow wf runOpts{runOnRemote = False}-- View isRaw -> if isRaw- then B.putStr $ encode dag- else T.putStrLn $ drawWorkflow dag-- Cat opts pid -> runWorkflow wf defaultRunOpt- { dbFile = dbPath opts- , nThread = 4- , runMode = Review $ T.pack pid- , configuration = fromMaybe [] $ configFile opts }-- Write opts pid input -> runWorkflow wf defaultRunOpt- { dbFile = dbPath opts- , nThread = 4- , runMode = Replace (T.pack pid) input- , configuration = fromMaybe [] $ configFile opts }-- Delete opts pid -> bracket (openDB $ dbPath opts) closeDB- (delRecord $ T.pack pid)-- 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- 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 (DynFunction fn) -> do- printf "Recovering node: %s.\n" pid- c <- B.readFile $ T.unpack fl- dat <- return (readYaml c) `asTypeOf` fn undefined- saveData pid dat db- Nothing -> printf "Cannot identify node: %s. Skipped.\n" pid- -}-- {--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 (DynFunction fn) -> do- printf "Saving node: %s.\n" pid- dat <- readData pid db `asTypeOf` fn undefined- B.writeFile fl $ showYaml dat- Nothing -> return ()- -}--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-- 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 ()-- let initState = WorkflowState db pidStateMap availableThreads- (runOnRemote opts) logS-- config <- case configuration opts of- [] -> return def- fls -> do- r <- decodeEither . B.unlines <$> mapM B.readFile fls- case r of- Left err -> error err- Right x -> return x-- 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)--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)--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
@@ -1,120 +0,0 @@-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
@@ -1,130 +0,0 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE Rank2Types #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE UndecidableInstances #-}--module Scientific.Workflow.Types- ( Workflow(..)- , PID- , NodeState(..)- , SpecialMode(..)- , ProcState- , WorkflowState(..)- , database- , procStatus- , procParaControl- , remote- , logServer- , Processor- , RunMode(..)- , RunOpt(..)- , defaultRunOpt- , Parallel(..)- , WorkflowConfig- ) where--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)--import Scientific.Workflow.Internal.Builder.Types (Attribute)-import Scientific.Workflow.Internal.DB (WorkflowDB (..))---- | The id of a node-type PID = T.Text---- | The result of a computation node-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- { _database :: WorkflowDB- , _procStatus :: M.Map PID (MVar NodeState, Attribute)- , _procParaControl :: MVar () -- ^ Concurrency controller- , _remote :: Bool -- ^ Global remote switch- , _logServer :: Maybe Socket -- ^ Server for logging- }--makeLenses ''WorkflowState--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 config = Workflow- { _worflow_dag :: Gr PID Int- , _worflow_pidToAttr :: M.Map T.Text Attribute- , _workflow :: Processor config () ()- }---- | Options-data RunOpt = RunOpt- { dbFile :: FilePath- , nThread :: Int -- ^ number of concurrent processes- , runOnRemote :: Bool- , runMode :: RunMode- , configuration :: [FilePath]- , selected :: Maybe [PID] -- ^ Should run only selected nodes- , logServerAddr :: Maybe String- }--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 config r = Parallel { runParallel :: (ProcState config) r}--instance Functor (Parallel config) where- fmap f (Parallel a) = Parallel $ f <$> a--instance Applicative (Parallel config) where- pure = Parallel . pure- Parallel fs <*> Parallel as = Parallel $- (\(f, a) -> f a) <$> concurrently fs as--instance S.Serialize (Gr (PID, Attribute) Int)
− src/Scientific/Workflow/Visualize.hs
@@ -1,57 +0,0 @@-{-# LANGUAGE OverloadedStrings #-}-module Scientific.Workflow.Visualize- ( drawWorkflow- ) where--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.Internal.Builder.Types (Attribute, note)-import Scientific.Workflow.Types---- | 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 (_, (i, attr)) = [G.Label $ G.HtmlLabel label]- where- label = H.Table $ H.HTable (Just []) tableAttr $ header : H.HorizontalRule :- map toLine (wrap 45 $ if T.null (attr^.note) then "Empty" else 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]