diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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. diff --git a/SciFlow.cabal b/SciFlow.cabal --- a/SciFlow.cabal +++ b/SciFlow.cabal @@ -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 diff --git a/src/Control/Arrow/Async.hs b/src/Control/Arrow/Async.hs new file mode 100644 --- /dev/null +++ b/src/Control/Arrow/Async.hs @@ -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) diff --git a/src/Control/Arrow/Free.hs b/src/Control/Arrow/Free.hs new file mode 100644 --- /dev/null +++ b/src/Control/Arrow/Free.hs @@ -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 + + diff --git a/src/Control/Workflow.hs b/src/Control/Workflow.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow.hs @@ -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 diff --git a/src/Control/Workflow/Coordinator.hs b/src/Control/Workflow/Coordinator.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Coordinator.hs @@ -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 diff --git a/src/Control/Workflow/Coordinator/Local.hs b/src/Control/Workflow/Coordinator/Local.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Coordinator/Local.hs @@ -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 diff --git a/src/Control/Workflow/DataStore.hs b/src/Control/Workflow/DataStore.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/DataStore.hs @@ -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) +-} diff --git a/src/Control/Workflow/Interpreter/Exec.hs b/src/Control/Workflow/Interpreter/Exec.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Interpreter/Exec.hs @@ -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 #-} diff --git a/src/Control/Workflow/Interpreter/FunctionTable.hs b/src/Control/Workflow/Interpreter/FunctionTable.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Interpreter/FunctionTable.hs @@ -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) diff --git a/src/Control/Workflow/Interpreter/Graph.hs b/src/Control/Workflow/Interpreter/Graph.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Interpreter/Graph.hs @@ -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 + diff --git a/src/Control/Workflow/Language.hs b/src/Control/Workflow/Language.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Language.hs @@ -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 diff --git a/src/Control/Workflow/Language/TH.hs b/src/Control/Workflow/Language/TH.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Language/TH.hs @@ -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 #-} diff --git a/src/Control/Workflow/Language/TH/Internal.hs b/src/Control/Workflow/Language/TH/Internal.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Language/TH/Internal.hs @@ -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 #-} diff --git a/src/Control/Workflow/Types.hs b/src/Control/Workflow/Types.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Types.hs @@ -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 diff --git a/src/Control/Workflow/Utils.hs b/src/Control/Workflow/Utils.hs new file mode 100644 --- /dev/null +++ b/src/Control/Workflow/Utils.hs @@ -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 #-} diff --git a/src/Scientific/Workflow.hs b/src/Scientific/Workflow.hs deleted file mode 100644 --- a/src/Scientific/Workflow.hs +++ /dev/null @@ -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 diff --git a/src/Scientific/Workflow/Internal/Builder.hs b/src/Scientific/Workflow/Internal/Builder.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Internal/Builder.hs +++ /dev/null @@ -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 #-} diff --git a/src/Scientific/Workflow/Internal/Builder/Types.hs b/src/Scientific/Workflow/Internal/Builder/Types.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Internal/Builder/Types.hs +++ /dev/null @@ -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) diff --git a/src/Scientific/Workflow/Internal/DB.hs b/src/Scientific/Workflow/Internal/DB.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Internal/DB.hs +++ /dev/null @@ -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] diff --git a/src/Scientific/Workflow/Internal/Utils.hs b/src/Scientific/Workflow/Internal/Utils.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Internal/Utils.hs +++ /dev/null @@ -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 diff --git a/src/Scientific/Workflow/Main.hs b/src/Scientific/Workflow/Main.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Main.hs +++ /dev/null @@ -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 diff --git a/src/Scientific/Workflow/Main/Options.hs b/src/Scientific/Workflow/Main/Options.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Main/Options.hs +++ /dev/null @@ -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 diff --git a/src/Scientific/Workflow/Types.hs b/src/Scientific/Workflow/Types.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Types.hs +++ /dev/null @@ -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) diff --git a/src/Scientific/Workflow/Visualize.hs b/src/Scientific/Workflow/Visualize.hs deleted file mode 100644 --- a/src/Scientific/Workflow/Visualize.hs +++ /dev/null @@ -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]