SciFlow 0.5.1 → 0.5.3.1
raw patch · 7 files changed
+281/−194 lines, 7 filesdep +exceptions
Dependencies added: exceptions
Files
- SciFlow.cabal +3/−2
- src/Scientific/Workflow.hs +27/−5
- src/Scientific/Workflow/Builder.hs +53/−87
- src/Scientific/Workflow/Main.hs +34/−36
- src/Scientific/Workflow/Types.hs +154/−60
- src/Scientific/Workflow/Utils.hs +10/−3
- src/Scientific/Workflow/Visualize.hs +0/−1
SciFlow.cabal view
@@ -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
src/Scientific/Workflow.hs view
@@ -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 ()
src/Scientific/Workflow/Builder.hs view
@@ -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
src/Scientific/Workflow/Main.hs view
@@ -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
src/Scientific/Workflow/Types.hs view
@@ -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)
src/Scientific/Workflow/Utils.hs view
@@ -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 = "./"
src/Scientific/Workflow/Visualize.hs view
@@ -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