diff --git a/SciFlow.cabal b/SciFlow.cabal --- a/SciFlow.cabal +++ b/SciFlow.cabal @@ -1,5 +1,5 @@ name: SciFlow -version: 0.5.1 +version: 0.5.3.1 synopsis: Scientific workflow management system description: SciFlow is to help programmers design complex workflows @@ -20,7 +20,7 @@ Flag debug Description: Enable debug support - Default: False + Default: True Flag sge Description: Enable SGE support @@ -53,6 +53,7 @@ , containers , cereal , directory + , exceptions , executable-path , fgl , graphviz diff --git a/src/Scientific/Workflow.hs b/src/Scientific/Workflow.hs --- a/src/Scientific/Workflow.hs +++ b/src/Scientific/Workflow.hs @@ -11,6 +11,7 @@ import Control.Monad.Trans.Except import qualified Data.Map as M import qualified Data.Set as S +import Data.Yaml (decodeFile) import Scientific.Workflow.Builder import Scientific.Workflow.DB @@ -19,14 +20,35 @@ import Text.Printf (printf) runWorkflow :: Workflow -> RunOpt -> IO () -runWorkflow (Workflow pids _ wf) opts = bracket (openDB $ database opts) closeDB $ \db -> do +runWorkflow (Workflow pids wf) opts = bracket (openDB $ database opts) closeDB $ \db -> do ks <- S.fromList <$> getKeys db - pidStateMap <- flip M.traverseWithKey pids $ \pid attr -> do - v <- if pid `S.member` ks then newMVar Success else newMVar Scheduled - return (v, attr) + pidStateMap <- flip M.traverseWithKey pids $ \pid attr -> case runMode opts of + Normal -> do + v <- if pid `S.member` ks then newMVar Success else newMVar Scheduled + return (v, attr) + ExecSingle i input output -> do + v <- if pid == i then newMVar (EXE input output) else newMVar Skip + return (v, attr) + ReadSingle i -> do + v <- if pid == i then newMVar Read else newMVar Skip + return (v, attr) + WriteSingle i input -> do + v <- if pid == i then newMVar (Replace input) else newMVar Skip + return (v, attr) + para <- newEmptyMVar _ <- forkIO $ replicateM_ (nThread opts) $ putMVar para () - let initState = WorkflowState db pidStateMap para $ runOnRemote opts + + env <- case configuration opts of + Nothing -> return M.empty + Just fl -> do + r <- decodeFile fl + case r of + Nothing -> error "fail to parse configuration file" + Just x -> return x + + let initState = WorkflowState db pidStateMap para (runOnRemote opts) env + result <- runExceptT $ evalStateT (wf ()) initState case result of Right _ -> return () diff --git a/src/Scientific/Workflow/Builder.hs b/src/Scientific/Workflow/Builder.hs --- a/src/Scientific/Workflow/Builder.hs +++ b/src/Scientific/Workflow/Builder.hs @@ -9,61 +9,33 @@ , link , (~>) , path - , Builder , buildWorkflow , buildWorkflowPart , mkDAG ) where -import Control.Lens ((^.), (%~), _1, _2, _3, at) -import Control.Exception (try) +import Control.Lens ((^.), (%~), _1, _2, _3) import Control.Monad.Trans.Except (throwE) -import Control.Monad.State +import Control.Monad.State (lift, liftIO, (>=>), foldM_, execState, modify, State, get) import Control.Concurrent.MVar import Control.Concurrent (forkIO) -import Control.Concurrent.Async.Lifted (concurrently, mapConcurrently) import qualified Data.Text as T -import Data.Graph.Inductive.Graph ( mkGraph, lab, labNodes, labEdges, outdeg +import Data.Graph.Inductive.Graph ( mkGraph, lab, labNodes, outdeg , lpre, labnfilter, nfilter, gmap, suc ) import Data.Graph.Inductive.PatriciaTree (Gr) -import Data.List (sortBy) +import Data.List (sortBy, foldl') import Data.Maybe (fromJust, fromMaybe) +import qualified Data.ByteString as B import Data.Ord (comparing) import qualified Data.Map as M import Text.Printf (printf) - +import Control.Concurrent.Async.Lifted (mapConcurrently) import Language.Haskell.TH -import qualified Language.Haskell.TH.Lift as T +import Control.Monad.Catch (try) import Scientific.Workflow.Types import Scientific.Workflow.DB -import Scientific.Workflow.Utils (debug, runRemote, defaultRemoteOpts) - - - -T.deriveLift ''M.Map -T.deriveLift ''Attribute - -instance T.Lift T.Text where - lift t = [| T.pack $(T.lift $ T.unpack t) |] - -instance T.Lift (Gr (PID, Attribute) Int) where - lift gr = [| uncurry mkGraph $(T.lift (labNodes gr, labEdges gr)) |] - - --- | The order of incoming edges of a node -type EdgeOrd = Int - --- | A computation node -type Node = (PID, (ExpQ, Attribute)) - --- | Links between computational nodes -type Edge = (PID, PID, EdgeOrd) - -type Function = (PID, ExpQ) - -type Builder = State ([Node], [Edge]) - +import Scientific.Workflow.Utils (debug, runRemote, defaultRemoteOpts, RemoteOpts(..)) -- | Declare a computational node. The function must have the signature: -- (DBData a, DBData b) => a -> IO b @@ -78,14 +50,6 @@ newNode = (p, (toExpQ fn, attr)) {-# INLINE node #-} - -{- - Declare a function that can be called on remote -function :: ToExpQ q => T.Text -> q -> Builder () -function funcName fn = --} - - -- | many-to-one generalized link function link :: [PID] -> PID -> Builder () link xs t = modify $ _2 %~ (zip3 xs (repeat t) [0..] ++) @@ -139,6 +103,7 @@ type DAG = Gr Node EdgeOrd +-- TODO: check the graph is a valid DAG -- | Contruct a DAG representing the workflow mkDAG :: Builder () -> DAG mkDAG b = mkGraph ns' es' @@ -179,46 +144,39 @@ mkWorkflow :: String -- name -> DAG -> Q [Dec] mkWorkflow workflowName dag = do - -- write node funcitons - functions <- fmap concat $ forM computeNodes $ \(p, (fn,_)) -> [d| - $(varP $ mkName $ T.unpack p) = mkProc p $(fn) |] - - -- function table - funcTable <- - [d| $(varP $ mkName functionTableName) = M.fromList - $( fmap ListE $ forM computeNodes $ \(p, (fn, _)) -> - [| (T.unpack p, Closure $(fn)) |] ) |] - - -- define workflows + let expq = connect sinks [| const $ return () |] + -- define the workflow workflows <- - [d| $(varP $ mkName workflowName) = Workflow pids - $(varE $ mkName functionTableName) - $(connect sinks [| const $ return () |]) |] + [d| $(varP $ mkName workflowName) = Workflow pids $expq |] - return $ functions ++ funcTable ++ workflows + return workflows where - functionTableName = workflowName ++ "_function_table" computeNodes = snd $ unzip $ labNodes dag pids = M.fromList $ map (\(i, x) -> (i, snd x)) computeNodes sinks = labNodes $ nfilter ((==0) . outdeg dag) dag - backTrack sink = connect sources $ mkNodeVar sink + backTrack (i, (p, (fn, attr))) + | attr^.stateful = connect (fst $ unzip parents) [| mkProc p $fn |] + | otherwise = connect (fst $ unzip parents) [| mkProc p (liftIO . $fn) |] where - sources = map (\(x,_) -> (x, fromJust $ lab dag x)) $ - sortBy (comparing snd) $ lpre dag $ fst sink + parents = map ( \(x, o) -> ((x, fromJust $ lab dag x), o) ) $ + sortBy (comparing snd) $ lpre dag i connect [] sink = sink - connect [source] sink = [| $(backTrack source) >=> $(sink) |] - connect sources sink = [| fmap runParallel $(foldl g e0 $ sources) - >=> $(sink) |] + connect [source] sink = [| $expq >=> $sink |] where + expq = backTrack source + connect sources sink = [| fmap runParallel $expq >=> $sink |] + where + expq = foldl' g e0 $ sources e0 = [| (pure. pure) $(conE (tupleDataName $ length sources)) |] - g acc x = [| ((<*>) . fmap (<*>)) $(acc) $ fmap Parallel $(backTrack x) |] - mkNodeVar = varE . mkName . T.unpack . fst . snd + g acc x = + let expq = backTrack x + in [| ((<*>) . fmap (<*>)) $acc $ fmap Parallel $expq |] {-# INLINE mkWorkflow #-} mkProc :: (BatchData' (IsList a b) a b, BatchData a b, DBData a, DBData b) - => PID -> (a -> IO b) -> (Processor a b) + => PID -> (a -> ProcState b) -> (Processor a b) mkProc pid f = \input -> do wfState <- get let (pSt, attr) = M.findWithDefault (error "Impossible") pid $ wfState^.procStatus @@ -242,15 +200,19 @@ #endif let sendToRemote = fromMaybe (wfState^.remote) (attr^.submitToRemote) - result <- liftIO $ try $ case () of + remoteOpts = defaultRemoteOpts + { extraParams = attr^.remoteParam + , environment = wfState^.config + } + result <- try $ case () of _ | attr^.batch > 0 -> do let (mkBatch, combineResult) = batchFunction f $ attr^.batch input' = mkBatch input combineResult <$> if sendToRemote - then mapConcurrently (runRemote defaultRemoteOpts pid) input' - else mapM f input' + then liftIO $ mapConcurrently (runRemote remoteOpts pid) input' + else mapM f input' -- do not run in parallel in local machine | otherwise -> if sendToRemote - then runRemote defaultRemoteOpts pid input + then liftIO $ runRemote remoteOpts pid input else f input case result of Left ex -> do @@ -263,18 +225,22 @@ putMVar pSt Success _ <- forkIO $ putMVar (wfState^.procParaControl) () return r + Skip -> liftIO $ putMVar pSt pStValue >> return undefined + EXE input output -> do + c <- liftIO $ B.readFile input + r <- f $ deserialize c + liftIO $ B.writeFile output $ serialize r + liftIO $ putMVar pSt Skip + return undefined + Read -> liftIO $ do + r <- readData pid $ wfState^.db + B.putStr $ showYaml r + putMVar pSt Skip + return r + Replace input -> do + c <- liftIO $ B.readFile input + r <- return (readYaml c) `asTypeOf` f undefined + liftIO $ updateData pid r $ wfState^.db + liftIO $ putMVar pSt Skip + return r {-# INLINE mkProc #-} - - - --------------------------------------------------------------------------------- - -newtype Parallel a = Parallel { runParallel :: ProcState a} - -instance Functor Parallel where - fmap f (Parallel a) = Parallel $ f <$> a - -instance Applicative Parallel where - pure = Parallel . pure - Parallel fs <*> Parallel as = Parallel $ - (\(f, a) -> f a) <$> concurrently fs as diff --git a/src/Scientific/Workflow/Main.hs b/src/Scientific/Workflow/Main.hs --- a/src/Scientific/Workflow/Main.hs +++ b/src/Scientific/Workflow/Main.hs @@ -1,7 +1,7 @@ +{-# LANGUAGE CPP #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE CPP #-} module Scientific.Workflow.Main ( defaultMain @@ -10,7 +10,6 @@ , MainOpts(..) ) where -import Control.Monad (forM_) import qualified Data.ByteString.Char8 as B import Data.Graph.Inductive.Graph (nmap) import Data.Graph.Inductive.PatriciaTree (Gr) @@ -19,7 +18,7 @@ import qualified Data.Text.Lazy.IO as T #ifdef SGE -import DRMAA (withSGESession) +import DRMAA (withSGESession) #endif import Language.Haskell.TH @@ -44,10 +43,12 @@ | Delete GlobalOpts String | Recover GlobalOpts FilePath | DumpDB GlobalOpts FilePath - | Call String String String + | Call GlobalOpts String String String data GlobalOpts = GlobalOpts - { dbPath :: FilePath } + { dbPath :: FilePath + , configFile :: Maybe FilePath + } globalParser :: Parser GlobalOpts globalParser = GlobalOpts @@ -55,7 +56,11 @@ ( long "db-path" <> value "sciflow.db" <> metavar "DB_PATH" ) + <*> (optional . strOption) + ( long "config" + <> metavar "CONFIG_PATH" ) + runParser :: Parser CMD runParser = Run <$> globalParser @@ -69,11 +74,14 @@ <> help "Submit jobs to remote machines.") runExe initialize (Run opts n r) wf #ifdef SGE - | r = initialize $ withSGESession $ runWorkflow wf $ RunOpt (dbPath opts) n True + | r = initialize $ withSGESession $ runWorkflow wf $ + RunOpt (dbPath opts) n True Normal $ configFile opts #else - | r = initialize $ runWorkflow wf $ RunOpt (dbPath opts) n True + | r = initialize $ runWorkflow wf $ + RunOpt (dbPath opts) n True Normal $ configFile opts #endif - | otherwise = runWorkflow wf $ RunOpt (dbPath opts) n False + | otherwise = runWorkflow wf $ + RunOpt (dbPath opts) n False Normal $ configFile opts runExe _ _ _ = undefined {-# INLINE runExe #-} @@ -87,13 +95,8 @@ <$> globalParser <*> strArgument (metavar "NODE_ID") -catExe (Cat opts pid) (Workflow _ ft _) = do - db <- openDB $ dbPath opts - case M.lookup pid ft of - Just (Closure fn) -> do - dat <- head [readData (T.pack pid) db, fn undefined] - B.putStr $ showYaml dat - Nothing -> return () +catExe (Cat opts pid) wf = runWorkflow wf $ + RunOpt (dbPath opts) 10 False (ReadSingle $ T.pack pid) $ configFile opts catExe _ _ = undefined {-# INLINE catExe #-} @@ -104,14 +107,8 @@ (metavar "NODE_ID") <*> strArgument (metavar "INPUT_FILE") -writeExe (Write opts pid input) (Workflow _ ft _) = do - db <- openDB $ dbPath opts - c <- B.readFile input - case M.lookup pid ft of - Just (Closure fn) -> do - dat <- head [return $ readYaml c, fn undefined] - updateData (T.pack pid) dat db - Nothing -> return () +writeExe (Write opts pid input) wf = runWorkflow wf $ + RunOpt (dbPath opts) 10 False (WriteSingle (T.pack pid) input) $ configFile opts writeExe _ _ = undefined {-# INLINE writeExe #-} @@ -131,6 +128,7 @@ <$> globalParser <*> strArgument (metavar "BACKUP") + {- recoverExe (Recover opts dir) (Workflow _ ft _) = do fls <- shelly $ lsT $ fromText $ T.pack dir shelly $ rm_f $ fromText $ T.pack $ dbPath opts @@ -138,12 +136,13 @@ forM_ fls $ \fl -> do let pid = snd $ T.breakOnEnd "/" fl case M.lookup (T.unpack pid) ft of - Just (Closure fn) -> do + Just (DynFunction fn) -> do printf "Recovering node: %s.\n" pid c <- B.readFile $ T.unpack fl - dat <- head [return $ readYaml c, fn undefined] + dat <- return (readYaml c) `asTypeOf` fn undefined saveData pid dat db Nothing -> printf "Cannot identify node: %s. Skipped.\n" pid + -} recoverExe _ _ = undefined {-# INLINE recoverExe #-} @@ -152,6 +151,7 @@ <$> globalParser <*> strArgument (metavar "OUTPUT_DIR") + {- dumpDBExe (DumpDB opts dir) (Workflow _ ft _) = do shelly $ mkdir_p $ fromText $ T.pack dir db <- openDB $ dbPath opts @@ -159,25 +159,23 @@ forM_ nodes $ \pid -> do let fl = dir ++ "/" ++ T.unpack pid case M.lookup (T.unpack pid) ft of - Just (Closure fn) -> do + Just (DynFunction fn) -> do printf "Saving node: %s.\n" pid - dat <- head [readData pid db, fn undefined] + dat <- readData pid db `asTypeOf` fn undefined B.writeFile fl $ showYaml dat Nothing -> return () + -} dumpDBExe _ _ = undefined {-# INLINE dumpDBExe #-} callParser :: Parser CMD callParser = Call - <$> strArgument mempty + <$> globalParser <*> strArgument mempty <*> strArgument mempty -callExe (Call pid inputFl outputFl) (Workflow _ ft _) = case M.lookup pid ft of - Just (Closure fn) -> do - input <- deserialize <$> B.readFile inputFl - output <- serialize <$> fn input - B.writeFile outputFl output - Nothing -> undefined + <*> strArgument mempty +callExe (Call opts pid inputFl outputFl) wf = runWorkflow wf $ + RunOpt (dbPath opts) 10 False (ExecSingle (T.pack pid) inputFl outputFl) $ configFile opts callExe _ _ = undefined {-# INLINE callExe #-} @@ -195,7 +193,7 @@ execute cmd@(Delete _ _) = rmExe cmd execute cmd@(Recover _ _) = recoverExe cmd wf execute cmd@(DumpDB _ _) = dumpDBExe cmd wf - execute cmd@(Call _ _ _) = callExe cmd wf + execute cmd@(Call _ _ _ _) = callExe cmd wf opts = info (helper <*> parser) $ fullDesc <> header h parser = subparser $ ( @@ -219,7 +217,7 @@ data MainOpts = MainOpts - { preAction :: Name -- ^ An action to be execute before the workflow. The + { preAction :: Name -- ^ An action to be execute before the workflow. The -- action should have type: IO () -> IO (). -- ^ i.e., some initialization processes. , programHeader :: String diff --git a/src/Scientific/Workflow/Types.hs b/src/Scientific/Workflow/Types.hs --- a/src/Scientific/Workflow/Types.hs +++ b/src/Scientific/Workflow/Types.hs @@ -1,16 +1,15 @@ -{-# LANGUAGE FlexibleInstances #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE TemplateHaskell #-} -{-# LANGUAGE UndecidableInstances #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE GADTs #-} {-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TemplateHaskell #-} +{-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE UndecidableInstances #-} module Scientific.Workflow.Types ( WorkflowDB(..) , Workflow(..) - , Closure(..) , PID , NodeResult(..) , ProcState @@ -19,7 +18,13 @@ , procStatus , procParaControl , remote + , config + , getConfig + , getConfigMaybe + , getConfig' + , getConfigMaybe' , Processor + , RunMode(..) , RunOpt(..) , BatchData(..) , BatchData'(..) @@ -32,44 +37,39 @@ , note , batch , submitToRemote - ) where - -import qualified Data.Serialize as S -import Control.Concurrent.MVar -import Control.Exception (SomeException) -import Control.Lens (makeLenses) -import Control.Monad.State -import Control.Monad.Trans.Except (ExceptT) -import qualified Data.ByteString as B -import qualified Data.Map as M -import Data.Maybe (fromJust) -import qualified Data.Text as T -import Data.Yaml (FromJSON, ToJSON, decode, encode) -import Database.SQLite.Simple (Connection) -import Data.List.Split (chunksOf) - -data HTrue -data HFalse - -type family IsList a b where - IsList [a] [b] = HTrue - IsList a b = HFalse - -class BatchData' flag a b where - batchFunction' :: flag -> (a -> IO b) -> Int -> (a -> [a], [b] -> b) - -instance BatchData' HTrue [a] [b] where - batchFunction' _ _ i = (chunksOf i, concat) + , stateful + , remoteParam -instance BatchData' HFalse a b where - batchFunction' _ _ _ = (return, head) + , Parallel(..) -class BatchData a b where - batchFunction :: (a -> IO b) -> Int -> (a -> [a], [b] -> b) + -- * Builder types + , Node + , Edge + , EdgeOrd + , Builder + ) where -instance (IsList a b ~ flag, BatchData' flag a b) => BatchData a b where - batchFunction = batchFunction' (undefined :: flag) +import Control.Concurrent.Async.Lifted (concurrently) +import Control.Concurrent.MVar (MVar) +import Control.Exception (SomeException) +import Control.Lens (at, makeLenses, (^.)) +import Control.Monad.State (State, StateT, get) +import Control.Monad.Trans.Except (ExceptT) +import qualified Data.ByteString as B +import Data.Graph.Inductive.Graph (labEdges, labNodes, mkGraph) +import Data.Graph.Inductive.PatriciaTree (Gr) +import Data.List.Split (chunksOf) +import qualified Data.Map as M +import Data.Maybe (fromJust, fromMaybe) +import qualified Data.Serialize as S +import qualified Data.Text as T +import Data.Yaml (FromJSON, ToJSON, decode, + encode) +import Database.SQLite.Simple (Connection) +import Language.Haskell.TH +import qualified Language.Haskell.TH.Lift as T +-- | 'DBData' type class is used for data serialization. class DBData a where serialize :: a -> B.ByteString deserialize :: B.ByteString -> a @@ -93,10 +93,14 @@ -- | Node attribute data Attribute = Attribute - { _label :: T.Text -- ^ short description - , _note :: T.Text -- ^ long description - , _batch :: Int - , _submitToRemote :: Maybe Bool -- ^ overwrite the global option + { _label :: T.Text -- ^ Short description + , _note :: T.Text -- ^ Long description + , _batch :: Int -- ^ Batch size. If > 0, inputs will be divided + -- into batches. + , _submitToRemote :: Maybe Bool -- ^ Overwrite the global option + , _stateful :: Bool -- ^ Whether the node function has access + -- to internal states + , _remoteParam :: String } makeLenses ''Attribute @@ -107,20 +111,32 @@ , _note = "" , _batch = -1 , _submitToRemote = Nothing + , _stateful = False + , _remoteParam = "" } type AttributeSetter = State Attribute () -- | The result of a computation node -data NodeResult = Success - | Fail SomeException - | Scheduled +data NodeResult = Success -- ^ The node has been executed + | Fail SomeException -- ^ The node failed to finish + | Scheduled -- ^ The node will be executed + | Skip -- ^ The node will not be executed + | Read -- ^ Simply read the saved data from database + | Replace FilePath -- ^ Read the result from the input file + -- and save it to database. + | EXE FilePath FilePath -- ^ Read input from the input file and + -- save results to the output file. This is + -- used in remote mode. data WorkflowState = WorkflowState - { _db :: WorkflowDB - , _procStatus :: M.Map PID (MVar NodeResult, Attribute) - , _procParaControl :: MVar () -- ^ concurrency controller - , _remote :: Bool + { _db :: WorkflowDB + , _procStatus :: M.Map PID (MVar NodeResult, Attribute) + , _procParaControl :: MVar () -- ^ Concurrency controller + , _remote :: Bool -- ^ Global remote switch + , _config :: M.Map T.Text T.Text -- ^ Workflow configuration. This + -- is used to store environmental + -- variables. } makeLenses ''WorkflowState @@ -128,17 +144,95 @@ type ProcState b = StateT WorkflowState (ExceptT (PID, SomeException) IO) b type Processor a b = a -> ProcState b +getConfigMaybe :: T.Text -> ProcState (Maybe T.Text) +getConfigMaybe key = do + st <- get + return $ (st^.config) ^.at key -data Closure where - Closure :: (DBData a, DBData b) => (a -> IO b) -> Closure +getConfig :: T.Text -> ProcState T.Text +getConfig x = fmap (fromMaybe errMsg) $ getConfigMaybe x + where + errMsg = error $ "The Key " ++ show x ++ " doesn't exist!" --- | A Workflow is a DAG +getConfig' :: T.Text -> ProcState String +getConfig' = fmap T.unpack . getConfig + +getConfigMaybe' :: T.Text -> ProcState (Maybe String) +getConfigMaybe' = (fmap.fmap) T.unpack . getConfigMaybe + +-- | A Workflow is a stateful function data Workflow = Workflow (M.Map T.Text Attribute) - (M.Map String Closure) (Processor () ()) +-- | Options data RunOpt = RunOpt - { database :: FilePath - , nThread :: Int -- ^ number of concurrent processes - , runOnRemote :: Bool + { database :: FilePath + , nThread :: Int -- ^ number of concurrent processes + , runOnRemote :: Bool + , runMode :: RunMode + , configuration :: Maybe FilePath } + +data RunMode = Normal + | ExecSingle PID FilePath FilePath + | ReadSingle PID + | WriteSingle PID FilePath + +-- | Auxiliary type for concurrency support. +newtype Parallel a = Parallel { runParallel :: ProcState a} + +instance Functor Parallel where + fmap f (Parallel a) = Parallel $ f <$> a + +instance Applicative Parallel where + pure = Parallel . pure + Parallel fs <*> Parallel as = Parallel $ + (\(f, a) -> f a) <$> concurrently fs as + + +T.deriveLift ''M.Map +T.deriveLift ''Attribute + +instance T.Lift T.Text where + lift t = [| T.pack $(T.lift $ T.unpack t) |] + +instance T.Lift (Gr (PID, Attribute) Int) where + lift gr = [| uncurry mkGraph $(T.lift (labNodes gr, labEdges gr)) |] + + +-- | The order of incoming edges of a node +type EdgeOrd = Int + +-- | A computation node +type Node = (PID, (ExpQ, Attribute)) + +-- | Links between computational nodes +type Edge = (PID, PID, EdgeOrd) + +type Builder = State ([Node], [Edge]) + + + +data HTrue +data HFalse + +type family IsList a b where + IsList [a] [b] = HTrue + IsList a b = HFalse + +class BatchData' flag a b where + batchFunction' :: flag -> (a -> ProcState b) -> Int -> (a -> [a], [b] -> b) + +instance BatchData' HTrue [a] [b] where + batchFunction' _ _ i = (chunksOf i, concat) + +instance BatchData' HFalse a b where + batchFunction' _ _ _ = (return, head) + +-- | 'BatchData' represents inputs that can be divided into batches and processed +-- in parallel, i.e. list. +class BatchData a b where + batchFunction :: (a -> ProcState b) -> Int -> (a -> [a], [b] -> b) + +instance (IsList a b ~ flag, BatchData' flag a b) => BatchData a b where + batchFunction = batchFunction' (undefined :: flag) diff --git a/src/Scientific/Workflow/Utils.hs b/src/Scientific/Workflow/Utils.hs --- a/src/Scientific/Workflow/Utils.hs +++ b/src/Scientific/Workflow/Utils.hs @@ -6,6 +6,8 @@ import Debug.Trace (traceM) import Rainbow import System.IO +import qualified Data.Map as M +import Data.Yaml (encode) import Scientific.Workflow.Types (DBData (..)) import System.Directory (getCurrentDirectory) @@ -31,23 +33,28 @@ data RemoteOpts = RemoteOpts { extraParams :: String + , environment :: M.Map T.Text T.Text } defaultRemoteOpts :: RemoteOpts defaultRemoteOpts = RemoteOpts { extraParams = "" + , environment = M.empty } runRemote :: (DBData a, DBData b) => RemoteOpts -> T.Text -> a -> IO b #ifdef SGE -runRemote opts pid input = withTmpFile tmpDir $ \inputFl -> withTmpFile tmpDir $ - \outputFl -> do +runRemote opts pid input = withTmpFile tmpDir $ \inputFl -> + withTmpFile tmpDir $ \outputFl -> withTmpFile tmpDir $ \configFl -> do + B.writeFile configFl $ encode $ environment opts + exePath <- getExecutablePath wd <- getCurrentDirectory let config = defaultDrmaaConfig{drmaa_wd=wd, drmaa_native=extraParams opts} B.writeFile inputFl $ serialize input - drmaaRun exePath ["execFunc", T.unpack pid, inputFl, outputFl] config :: IO () + drmaaRun exePath [ "execFunc", "--config", configFl, T.unpack pid + , inputFl, outputFl ] config :: IO () deserialize <$> B.readFile outputFl where tmpDir = "./" diff --git a/src/Scientific/Workflow/Visualize.hs b/src/Scientific/Workflow/Visualize.hs --- a/src/Scientific/Workflow/Visualize.hs +++ b/src/Scientific/Workflow/Visualize.hs @@ -14,7 +14,6 @@ import Data.Graph.Inductive.PatriciaTree (Gr) import Scientific.Workflow.Types -import Scientific.Workflow.Builder -- | Print the computation graph renderBuilder :: Gr (PID, Attribute) Int -> TL.Text