diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -1,10 +1,27 @@
 Scientific workflow management system
 =====================================
 
-A scientific workflow is a series of computational steps which usually can be presented as a Directed Acyclic Graph (DAG).
+Introduction
+------------
 
-SciFlow is to help programmers design complex workflows with ease. Here is a trivial example. (Since we use template haskell, we need to divide this small program into two parts.)
+SciFlow is a workflow management system for working with big data pipelines locally
+or in a grid computing system.
 
+Most scientific computing pipelines are composed of many computational steps, and each of them involves heavy computation and IO operations. A workflow management system can
+help user design complex computing patterns and track the states of computation.
+The ability to recover from failures is crucial in large pipelines as they usually
+take days or weeks to finish.
+
+Features:
+
+1. Easy to use: A simple and flexible way to specify computational pipelines in Haskell.
+
+2. Automatic Checkpointing: The result of each intermediate step is stored, allowing easy restart upon failures.
+
+3. Parallelism and grid computing support: Independent computational steps will run concurrently. And users can decide whether to run steps locally or on remote compute nodes in a grid system.
+
+Here is a simple example. (Since we use template haskell, we need to divide this small program into two files.)
+
 ```haskell
 ---------------------------------------------------
 -- File 1: MyModule.hs
@@ -66,26 +83,20 @@
 ---------------------------------------------------
 {-# LANGUAGE TemplateHaskell #-}
 
-import System.Environment
-
 import qualified Functions as F
-import qualified Data.Text.Lazy.IO as T
-
-import Scientific.Workflow
-import Scientific.Workflow.Visualize
-
-buildWorkflow "wf" F.builder
+import Scientific.Workflow.Main
 
-main :: IO ()
-main = do
-    (cmd:args) <- getArgs
-    case cmd of
-        "run" -> runWorkflow wf def
-        "view" -> T.putStrLn $ renderBuilder F.builder
+defaultMain F.builder
 ```
 
-The workflow can be visualized by running `runghc main.hs view | dot -Tpng > example.png`.
+Use `ghc main.hs -threaded` to compile the program. And type `./main --help` to
+see available commands. For example, the workflow can be visualized by running
+`./main view | dot -Tpng > example.png`, as shown below.
 
-![example](examples/example.png)
+![example](example.png)
 
-To run the workflow, simply type `runghc main.hs run`. The program will create a directory to store results of each step. If being terminated prematurely, the program will use the saved data to continue from the last step.
+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 sge` flag. Use `./main run --remote` to submit jobs
+to remote machines.
diff --git a/SciFlow.cabal b/SciFlow.cabal
--- a/SciFlow.cabal
+++ b/SciFlow.cabal
@@ -1,5 +1,5 @@
 name:                SciFlow
-version:             0.4.1
+version:             0.5.0
 synopsis:            Scientific workflow management system
 description:
   SciFlow is to help programmers design complex workflows
@@ -16,44 +16,58 @@
 
 extra-source-files:
   README.md
-  examples/Functions.hs
-  examples/Main.hs
-  examples/example.png
+  example.png
 
-Flag Debug
+Flag debug
   Description: Enable debug support
   Default:     False
 
+Flag sge
+  Description: Enable SGE support
+  Default:     False
+
 library
   ghc-options: -Wall
   exposed-modules:
     Scientific.Workflow
     Scientific.Workflow.Builder
     Scientific.Workflow.DB
+    Scientific.Workflow.Main
     Scientific.Workflow.Types
+    Scientific.Workflow.Utils
     Scientific.Workflow.Visualize
 
+  other-modules:
+    Paths_SciFlow
+
   if flag(debug)
     CPP-Options: -DDEBUG
-  else
-    CPP-Options: -DNDEBUG
 
+  if flag(sge)
+    CPP-Options: -DSGE
+    build-depends: drmaa
+
   build-depends:
       base >=4.0 && <5.0
     , bytestring
-    , data-default-class
+    , containers
+    , cereal
+    , directory
+    , executable-path
     , fgl
     , graphviz
     , lens >=4.0
+    , lifted-async
     , mtl
     , optparse-applicative
+    , rainbow
     , shelly
+    , sqlite-simple
     , split
     , th-lift
     , text
     , template-haskell
     , transformers
-    , containers
     , yaml
 
   hs-source-dirs:      src
diff --git a/example.png b/example.png
new file mode 100644
Binary files /dev/null and b/example.png differ
diff --git a/examples/Functions.hs b/examples/Functions.hs
deleted file mode 100644
--- a/examples/Functions.hs
+++ /dev/null
@@ -1,51 +0,0 @@
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TemplateHaskell #-}
-module Functions
-    (builder) where
-
-import Control.Lens ((^.), (.=))
-import qualified Data.Text as T
-import Shelly hiding (FilePath)
-import Text.Printf (printf)
-
-import Scientific.Workflow
-
-create :: () -> IO FilePath
-create _ = do
-    writeFile "hello.txt" "hello world"
-    return "hello.txt"
-
-countWords :: FilePath -> IO Int
-countWords fl = do
-    content <- readFile fl
-    return $ length $ words content
-
-countChars :: FilePath -> IO Int
-countChars fl = do
-    content <- readFile fl
-    return $ sum $ map length $ words content
-
-output :: (Int, Int) -> IO Bool
-output (ws, cs) = do
-    putStrLn $ printf "Number of words: %d" ws
-    putStrLn $ printf "Number of characters: %d" cs
-    return True
-
-cleanUp :: (Bool, FilePath) -> IO ()
-cleanUp (toBeRemoved, fl) = if toBeRemoved
-    then shelly $ rm $ fromText $ T.pack fl
-    else return ()
-
--- builder monad
-builder :: Builder ()
-builder = do
-    node "step0" 'create $ label .= "write something to a file"
-    node "step1" 'countWords $ label .= "word count"
-    node "step2" 'countChars $ label .= "character count"
-    node "step3" 'output $ label .= "print"
-    node "step4" 'cleanUp $ label .= "remove the file"
-
-    ["step0"] ~> "step1"
-    ["step0"] ~> "step2"
-    ["step1", "step2"] ~> "step3"
-    ["step3", "step0"] ~> "step4"
diff --git a/examples/Main.hs b/examples/Main.hs
deleted file mode 100644
--- a/examples/Main.hs
+++ /dev/null
@@ -1,18 +0,0 @@
-{-# LANGUAGE TemplateHaskell #-}
-
-import System.Environment
-
-import qualified Functions as F
-import qualified Data.Text.Lazy.IO as T
-
-import Scientific.Workflow
-import Scientific.Workflow.Visualize
-
-buildWorkflow "wf" F.builder
-
-main :: IO ()
-main = do
-    (cmd:args) <- getArgs
-    case cmd of
-        "run" -> runWorkflow wf def
-        "view" -> T.putStrLn $ renderBuilder F.builder
diff --git a/examples/example.png b/examples/example.png
deleted file mode 100644
Binary files a/examples/example.png and /dev/null differ
diff --git a/src/Scientific/Workflow.hs b/src/Scientific/Workflow.hs
--- a/src/Scientific/Workflow.hs
+++ b/src/Scientific/Workflow.hs
@@ -1,28 +1,34 @@
 module Scientific.Workflow
     ( runWorkflow
-    , getWorkflowState
     , module Scientific.Workflow.Builder
     , module Scientific.Workflow.Types
     ) where
 
+import           Control.Concurrent          (forkIO)
+import           Control.Concurrent.MVar
+import           Control.Exception           (bracket, displayException)
 import           Control.Monad.State
-import Control.Monad.Trans.Except
-import Control.Exception (displayException)
+import           Control.Monad.Trans.Except
+import qualified Data.Map                    as M
+import qualified Data.Set                    as S
 
 import           Scientific.Workflow.Builder
+import           Scientific.Workflow.DB
 import           Scientific.Workflow.Types
-import System.IO
+import           Scientific.Workflow.Utils
+import           Text.Printf                 (printf)
 
-runWorkflow :: [Workflow] -> State RunOpt () -> IO ()
-runWorkflow wfs setOpt = do
-    config <- getWorkflowState $ _dbPath opt
-    foldM_ f config wfs
-  where
-    opt = execState setOpt defaultRunOpt
-    f config (Workflow wf) = do
-        result <- runExceptT $ runStateT (wf ()) config
-        case result of
-            Right (_, config') -> return config'
-            Left ex -> do
-                hPutStrLn stderr $ displayException ex
-                return config
+runWorkflow :: Workflow -> RunOpt -> IO ()
+runWorkflow (Workflow pids _ wf) opts = bracket (openDB $ database opts) closeDB $ \db -> do
+    ks <- S.fromList <$> getKeys db
+    pidStateMap <- flip M.traverseWithKey pids $ \pid attr -> do
+        v <- if pid `S.member` ks then newMVar Success else newMVar Scheduled
+        return (v, attr)
+    para <- newEmptyMVar
+    _ <- forkIO $ replicateM_ (nThread opts) $ putMVar para ()
+    let initState = WorkflowState db pidStateMap para $ runOnRemote opts
+    result <- runExceptT $ evalStateT (wf ()) initState
+    case result of
+        Right _ -> return ()
+        Left (pid, ex) -> error' $ printf "\"%s\" failed. The error was: %s"
+            pid (displayException ex)
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
@@ -12,45 +12,46 @@
     , Builder
     , buildWorkflow
     , buildWorkflowPart
-    , getWorkflowState
     , mkDAG
     ) where
 
-import Control.Lens ((^.), (%~), _1, _2, _3, at, (.=))
-import Control.Exception (try, displayException)
+import Control.Lens ((^.), (%~), _1, _2, _3, at)
+import Control.Exception (try)
 import Control.Monad.Trans.Except (throwE)
 import           Control.Monad.State
+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
-    , outdeg
-    , lpre
-    , labnfilter
-    , gmap
-    , suc
-    , subgraph )
+import Data.Graph.Inductive.Graph ( mkGraph, lab, labNodes, labEdges, outdeg
+                                  , lpre, labnfilter, nfilter, gmap, suc )
 import Data.Graph.Inductive.PatriciaTree (Gr)
 import Data.List (sortBy)
-import qualified Data.Map as M
-import Data.Maybe (fromJust)
+import Data.Maybe (fromJust, fromMaybe)
 import Data.Ord (comparing)
-import qualified Data.Map                    as M
+import qualified Data.Map as M
+import Text.Printf (printf)
 
 import           Language.Haskell.TH
 import qualified Language.Haskell.TH.Lift as T
 
 import Scientific.Workflow.Types
 import Scientific.Workflow.DB
+import Scientific.Workflow.Utils (debug, runRemote, defaultRemoteOpts)
 
-import Debug.Trace (traceM)
 
+
+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 edges
+
+-- | The order of incoming edges of a node
 type EdgeOrd = Int
 
 -- | A computation node
@@ -59,9 +60,13 @@
 -- | Links between computational nodes
 type Edge = (PID, PID, EdgeOrd)
 
+type Function = (PID, ExpQ)
+
 type Builder = State ([Node], [Edge])
 
--- | Declare a computational node
+
+-- | Declare a computational node. The function must have the signature:
+-- (DBData a, DBData b) => a -> IO b
 node :: ToExpQ q
      => PID                  -- ^ node id
      -> q                    -- ^ function
@@ -73,6 +78,12 @@
     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..] ++)
@@ -90,33 +101,29 @@
     f a t = link [a] t >> return t
 {-# INLINE path #-}
 
--- | Build the workflow.
-buildWorkflow :: String
+-- | 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 $prefix$_sciflow.
+-- Lastly, a function table will be created with the name $prefix$_function_table.
+buildWorkflow :: String     -- ^ prefix
               -> Builder ()
               -> Q [Dec]
-buildWorkflow wfName b = mkWorkflow wfName $ mkDAG b
+buildWorkflow prefix b = mkWorkflow prefix $ mkDAG b
 
 -- | Build only a part of the workflow that has not been executed. This is used
 -- during development for fast compliation.
-buildWorkflowPart :: State RunOpt ()
+buildWorkflowPart :: FilePath   -- ^ path to the db
                   -> String
                   -> Builder ()
                   -> Q [Dec]
-buildWorkflowPart setOpt wfName b = do
-    st <- runIO $ getWorkflowState $ opt^.dbPath
+buildWorkflowPart db wfName b = do
+    st <- runIO $ getWorkflowState db
     mkWorkflow wfName $ trimDAG st $ mkDAG b
   where
-    opt = execState setOpt defaultRunOpt
-
-getWorkflowState :: FilePath -> IO WorkflowState
-getWorkflowState dir = do
-    db <- openDB dir
-    ks <- getKeys db
-    pSt <- mapM (flip isFinished db) ks
-    let pSts = M.fromList $ zipWith (\k s ->
-                 if s then (k, Success) else (k, Scheduled)) ks pSt
-    return $ WorkflowState db pSts
-{-# INLINE getWorkflowState #-}
+    getWorkflowState dir = do
+        db <- openDB dir
+        ks <- getKeys db
+        return $ M.fromList $ zip ks $ repeat Success
 
 -- | Objects that can be converted to ExpQ
 class ToExpQ a where
@@ -145,69 +152,127 @@
 {-# INLINE mkDAG #-}
 
 -- | Remove nodes that are executed before from a DAG.
-trimDAG :: WorkflowState -> DAG -> DAG
+trimDAG :: (M.Map T.Text NodeResult) -> DAG -> DAG
 trimDAG st dag = gmap revise gr
   where
-    revise context@(linkTo, nd, lab, linkFrom)
+    revise context@(linkTo, _, lab, _)
         | done (fst lab) && null linkTo = _3._2._1 %~ e $ context
         | otherwise = context
       where
-        e x = [| const undefined >=> $(x) |]
+        e x = [| (\() -> undefined) >=> $(x) |]
     gr = labnfilter f dag
       where
         f (i, (x,_)) = (not . done) x || any (not . done) children
           where children = map (fst . fromJust . lab dag) $ suc dag i
-    done x = case M.lookup x (st^.procStatus) of
+    done x = case M.lookup x st of
         Just Success -> True
         _ -> False
 {-# INLINE trimDAG #-}
 
--- | Generate codes from a DAG
-mkWorkflow :: String -> DAG -> Q [Dec]
-mkWorkflow wfName dag = do
-    decNode <- concat <$> mapM (f . snd) ns
-    decWf <- [d| $(varP $ mkName wfName) = $(fmap ListE $ mapM linking leafNodes)
-             |]
-    return $ decNode ++ decWf
+
+-- 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 $prefix$_function_table.
+mkWorkflow :: String   -- prefix
+           -> 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
+    workflows <-
+        [d| $(varP $ mkName workflowName) = Workflow pids
+                $(varE $ mkName functionTableName)
+                $(connect sinks [| const $ return () |]) |]
+
+    return $ functions ++ funcTable ++ workflows
   where
-    f (p, (fn, _)) = [d| $(varP $ mkName $ T.unpack p) = mkProc p $(fn) |]
-    ns = labNodes dag
-    leafNodes = filter ((==0) . outdeg dag . fst) ns
-    linking nd = [| Workflow $(go nd) |]
+    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
       where
-        go n = connect inNodes n
-          where
-            inNodes = map (\(x,_) -> (x, fromJust $ lab dag x)) $
-                      sortBy (comparing snd) $ lpre dag $ fst n
-        define n = varE $ mkName (T.unpack $ (snd n) ^. _1)
-        connect [] t = define t
-        connect [s1] t = [| $(go s1) >=> $(define t) |]
-        connect xs t = [| $(foldl g e0 $ tail xs) >=> $(define t) |]
-          where
-            e0 = [| (fmap.fmap) $(conE (tupleDataName $ length xs)) $(go $ head xs) |]
-            g acc x = [| ((<*>) . fmap (<*>)) $(acc) $(go x) |]
+        sources = map (\(x,_) -> (x, fromJust $ lab dag x)) $
+            sortBy (comparing snd) $ lpre dag $ fst sink
+
+    connect [] sink = sink
+    connect [source] sink = [| $(backTrack source) >=> $(sink) |]
+    connect sources sink = [| fmap runParallel $(foldl g e0 $ sources)
+        >=> $(sink) |]
+      where
+        e0 = [| (pure. pure) $(conE (tupleDataName $ length sources)) |]
+        g acc x = [| ((<*>) . fmap (<*>)) $(acc) $ fmap Parallel $(backTrack x) |]
+    mkNodeVar = varE . mkName . T.unpack . fst . snd
 {-# INLINE mkWorkflow #-}
 
-mkProc :: Serializable b => PID -> (a -> IO b) -> (Processor a b)
+mkProc :: (BatchData' (IsList a b) a b, BatchData a b, DBData a, DBData b)
+       => PID -> (a -> IO b) -> (Processor a b)
 mkProc pid f = \input -> do
-    st <- get
-    case M.findWithDefault Scheduled pid (st^.procStatus) of
-        Fail ex -> lift $ throwE ex
-        Success -> do
-            r <- liftIO $ readData pid $ st^.db
-            return r
+    wfState <- get
+    let (pSt, attr) = M.findWithDefault (error "Impossible") pid $ wfState^.procStatus
+
+    pStValue <- liftIO $ takeMVar pSt
+    case pStValue of
+        (Fail ex) -> liftIO (putMVar pSt pStValue) >> lift (throwE (pid, ex))
+        Success -> liftIO $ do
+            putMVar pSt pStValue
+
+#ifdef DEBUG
+            debug $ printf "Recovering saved node: %s" pid
+#endif
+
+            readData pid $ wfState^.db
         Scheduled -> do
+            _ <- liftIO $ takeMVar $ wfState^.procParaControl
+
 #ifdef DEBUG
-            traceM $ "Running node: " ++ T.unpack pid
+            debug $ printf "Running node: %s" pid
 #endif
 
-            result <- liftIO $ try $ f input
+            let sendToRemote = fromMaybe (wfState^.remote) (attr^.submitToRemote)
+            result <- liftIO $ 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'
+                  | otherwise -> if sendToRemote
+                      then runRemote defaultRemoteOpts pid input
+                      else f input
             case result of
                 Left ex -> do
-                    (procStatus . at pid) .= Just (Fail ex)
-                    lift $ throwE ex
-                Right r -> do
-                    liftIO $ saveData pid r $ st^.db
-                    (procStatus . at pid) .= Just Success
+                    _ <- liftIO $ do
+                        putMVar pSt $ Fail ex
+                        forkIO $ putMVar (wfState^.procParaControl) ()
+                    lift (throwE (pid, ex))
+                Right r -> liftIO $ do
+                    saveData pid r $ wfState^.db
+                    putMVar pSt Success
+                    _ <- forkIO $ putMVar (wfState^.procParaControl) ()
                     return r
 {-# 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/DB.hs b/src/Scientific/Workflow/DB.hs
--- a/src/Scientific/Workflow/DB.hs
+++ b/src/Scientific/Workflow/DB.hs
@@ -1,40 +1,94 @@
 {-# LANGUAGE OverloadedStrings #-}
 module Scientific.Workflow.DB
     ( openDB
+    , closeDB
     , readData
+    , readDataByteString
+    , saveDataByteString
     , saveData
+    , updateData
+    , delRecord
     , isFinished
     , getKeys
     ) where
 
-import Scientific.Workflow.Types
-import           Shelly              (fromText, lsT, shelly, test_f, mkdir_p)
-import qualified Data.ByteString     as B
-import qualified Data.Text           as T
+import qualified Data.ByteString           as B
+import qualified Data.Text                 as T
+import           Database.SQLite.Simple
+import           Scientific.Workflow.Types
+import           Text.Printf               (printf)
 
+dbTableName :: String
+dbTableName = "SciFlowDB"
+
+createTable :: Connection -> String -> IO ()
+createTable db tablename =
+    execute_ db $ Query $ T.pack $ printf
+        "CREATE TABLE %s(pid TEXT PRIMARY KEY, data BLOB)" tablename
+
+hasTable :: Connection -> String -> IO Bool
+hasTable db tablename = do
+    r <- query db
+        "SELECT name FROM sqlite_master WHERE type='table' AND name=?" [tablename]
+    return $ not $ null (r :: [Only T.Text])
+
 openDB :: FilePath -> IO WorkflowDB
-openDB dir = do
-    shelly $ mkdir_p $ fromText $ T.pack dir
-    return $ WorkflowDB dir
+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 #-}
 
-readData :: Serializable r => PID -> WorkflowDB -> IO r
-readData p (WorkflowDB dir) = deserialize <$> B.readFile fl
-  where fl = dir ++ "/" ++ T.unpack p
+closeDB :: WorkflowDB -> IO ()
+closeDB (WorkflowDB db) = close db
+{-# INLINE closeDB #-}
+
+readData :: DBData r => PID -> WorkflowDB -> IO r
+readData pid (WorkflowDB db) = do
+    [Only result] <- query db (Query $ T.pack $
+        printf "SELECT data FROM %s WHERE pid=?" dbTableName) [pid]
+    return $ deserialize result
 {-# INLINE readData #-}
 
-saveData :: Serializable r => PID -> r -> WorkflowDB -> IO ()
-saveData p result (WorkflowDB dir) = B.writeFile fl $ serialize result
-  where fl = dir ++ "/" ++ T.unpack p
+readDataByteString :: PID -> WorkflowDB -> IO B.ByteString
+readDataByteString pid (WorkflowDB db) = do
+    [Only result] <- query db (Query $ T.pack $
+        printf "SELECT data FROM %s WHERE pid=?" dbTableName) [pid]
+    return result
+{-# INLINE readDataByteString #-}
+
+saveData :: DBData r => PID -> r -> WorkflowDB -> IO ()
+saveData pid result (WorkflowDB db) = execute db (Query $ T.pack $
+    printf "INSERT INTO %s VALUES (?, ?)" dbTableName) (pid, serialize result)
 {-# INLINE saveData #-}
 
+updateData :: DBData r => PID -> r -> WorkflowDB -> IO ()
+updateData pid result (WorkflowDB db) = execute db (Query $ T.pack $
+    printf "UPDATE %s SET data=? WHERE pid=?" dbTableName) (serialize result, pid)
+{-# INLINE updateData #-}
+
+saveDataByteString :: PID -> B.ByteString -> WorkflowDB -> IO ()
+saveDataByteString pid result (WorkflowDB db) = execute db (Query $ T.pack $
+    printf "INSERT INTO %s VALUES (?, ?)" dbTableName) (pid, result)
+{-# INLINE saveDataByteString #-}
+
 isFinished :: PID -> WorkflowDB -> IO Bool
-isFinished p (WorkflowDB dir) = shelly $ test_f $ fromText $ T.pack fl
-  where fl = dir ++ "/" ++ T.unpack p
+isFinished pid (WorkflowDB db) = do
+    result <- query db (Query $ T.pack $
+        printf "SELECT pid FROM %s WHERE pid = ?" dbTableName) [pid]
+    return $ not $ null (result :: [Only T.Text])
 {-# INLINE isFinished #-}
 
 getKeys :: WorkflowDB -> IO [PID]
-getKeys (WorkflowDB dir) = f <$> shelly (lsT $ fromText $ T.pack dir)
-  where
-    f = map (snd . T.breakOnEnd "/")
+getKeys (WorkflowDB db) = concat <$> query_ db (Query $ T.pack $
+    printf "SELECT pid FROM %s;" dbTableName)
 {-# INLINE getKeys #-}
+
+delRecord :: PID -> WorkflowDB -> IO ()
+delRecord pid (WorkflowDB db) =
+    execute db (Query $ T.pack $ printf
+        "DELETE FROM %s WHERE pid = ?" dbTableName) [pid]
diff --git a/src/Scientific/Workflow/Main.hs b/src/Scientific/Workflow/Main.hs
new file mode 100644
--- /dev/null
+++ b/src/Scientific/Workflow/Main.hs
@@ -0,0 +1,239 @@
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TemplateHaskell   #-}
+
+module Scientific.Workflow.Main
+    ( defaultMain
+    , defaultMainOpts
+    , mainWith
+    , 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)
+import qualified Data.Map                          as M
+import qualified Data.Text                         as T
+import qualified Data.Text.Lazy.IO                 as T
+import           Language.Haskell.TH
+import qualified Language.Haskell.TH.Lift          as T
+import           Options.Applicative               hiding (runParser)
+import           Shelly                            (fromText, lsT, mkdir_p,
+                                                    rm_f, shelly)
+import           Text.Printf                       (printf)
+
+import           Scientific.Workflow
+import           Scientific.Workflow.DB
+import           Scientific.Workflow.Visualize
+
+import           Data.Version                      (showVersion)
+import           Paths_SciFlow                     (version)
+
+
+data CMD = Run GlobalOpts Int Bool
+         | View
+         | Cat GlobalOpts String
+         | Write GlobalOpts String FilePath
+         | Delete GlobalOpts String
+         | Recover GlobalOpts FilePath
+         | DumpDB GlobalOpts FilePath
+         | Call String String String
+
+data GlobalOpts = GlobalOpts
+    { dbPath :: FilePath }
+
+globalParser :: Parser GlobalOpts
+globalParser = GlobalOpts
+           <$> strOption
+               ( long "db-path"
+              <> value "sciflow.db"
+              <> metavar "DB_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.")
+runExe initialize (Run opts n r) wf
+    | r = initialize $ runWorkflow wf $ RunOpt (dbPath opts) n True
+    | otherwise = runWorkflow wf $ RunOpt (dbPath opts) n False
+runExe _ _ _ = undefined
+{-# INLINE runExe #-}
+
+viewParser :: Parser CMD
+viewParser = pure View
+viewExe = T.putStrLn . renderBuilder
+{-# INLINE viewExe #-}
+
+catParser :: Parser CMD
+catParser = Cat
+        <$> 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 _ _ = undefined
+{-# INLINE catExe #-}
+
+writeParser :: Parser CMD
+writeParser = Write
+          <$> globalParser
+          <*> strArgument
+              (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 _ _ = undefined
+{-# INLINE writeExe #-}
+
+rmParser :: Parser CMD
+rmParser = Delete
+       <$> globalParser
+       <*> strArgument
+           (metavar "NODE_ID")
+rmExe (Delete opts pid) = do
+    db <- openDB $ dbPath opts
+    delRecord (T.pack pid) db
+rmExe _ = undefined
+{-# INLINE rmExe #-}
+
+recoverParser :: Parser CMD
+recoverParser = Recover
+            <$> 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
+    db <- openDB $ dbPath opts
+    forM_ fls $ \fl -> do
+        let pid = snd $ T.breakOnEnd "/" fl
+        case M.lookup (T.unpack pid) ft of
+            Just (Closure fn) -> do
+                printf "Recovering node: %s.\n" pid
+                c <- B.readFile $ T.unpack fl
+                dat <- head [return $ readYaml c, fn undefined]
+                saveData pid dat db
+            Nothing -> printf "Cannot identify node: %s. Skipped.\n" pid
+recoverExe _ _ = undefined
+{-# INLINE recoverExe #-}
+
+dumpDBParser :: Parser CMD
+dumpDBParser = DumpDB
+           <$> globalParser
+           <*> strArgument
+               (metavar "OUTPUT_DIR")
+dumpDBExe (DumpDB opts dir) (Workflow _ ft _) = do
+    shelly $ mkdir_p $ fromText $ T.pack dir
+    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 (Closure fn) -> do
+                printf "Saving node: %s.\n" pid
+                dat <- head [readData pid db, fn undefined]
+                B.writeFile fl $ showYaml dat
+            Nothing -> return ()
+dumpDBExe _ _ = undefined
+{-# INLINE dumpDBExe #-}
+
+callParser :: Parser CMD
+callParser = Call
+         <$> strArgument mempty
+         <*> 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
+callExe _ _ = undefined
+{-# INLINE callExe #-}
+
+
+mainFunc :: (IO () -> IO ()) -- initialization function
+         -> Gr (PID, Attribute) Int -> Workflow
+         -> String  -- program header
+         -> IO ()
+mainFunc initialize dag wf h = execParser opts >>= execute
+  where
+    execute cmd@(Run _ _ _) = runExe initialize cmd wf
+    execute View = viewExe dag
+    execute cmd@(Cat _ _) = catExe cmd wf
+    execute cmd@(Write _ _ _) = writeExe cmd wf
+    execute cmd@(Delete _ _) = rmExe cmd
+    execute cmd@(Recover _ _) = recoverExe cmd wf
+    execute cmd@(DumpDB _ _) = dumpDBExe cmd wf
+    execute cmd@(Call _ _ _) = callExe cmd wf
+
+    opts = info (helper <*> parser) $ fullDesc <> header h
+    parser = subparser $ (
+        command "run" (info (helper <*> runParser) $
+            fullDesc <> progDesc "run workflow")
+     <> command "view" (info (helper <*> viewParser) $
+            fullDesc <> progDesc "view workflow")
+     <> command "cat" (info (helper <*> catParser) $
+            fullDesc <> progDesc "display the result of a node")
+     <> command "write" (info (helper <*> writeParser) $
+            fullDesc <> progDesc "write the result to a node")
+     <> command "rm" (info (helper <*> rmParser) $
+            fullDesc <> progDesc "delete the result of a node.")
+     <> command "recover" (info (helper <*> recoverParser) $
+            fullDesc <> progDesc "Recover database from backup.")
+     <> command "backup" (info (helper <*> dumpDBParser) $
+            fullDesc <> progDesc "Backup database.")
+     <> command "execFunc" (info (helper <*> callParser) $
+            fullDesc <> progDesc "Do not call this directly.")
+     )
+
+
+data MainOpts = MainOpts
+    { preAction :: Name    -- ^ An action to be execute before the workflow. The
+                           -- action should have type: IO () -> IO ().
+                            -- ^ i.e., some initialization processes.
+    , programHeader :: String
+    }
+
+T.deriveLift ''MainOpts
+
+defaultMainOpts :: MainOpts
+defaultMainOpts = MainOpts
+    { preAction = 'id
+    , programHeader = printf "SciFlow-%s" (showVersion version)
+    }
+
+defaultMain :: Builder () -> Q [Dec]
+defaultMain = mainWith defaultMainOpts
+
+mainWith :: MainOpts -> Builder () -> Q [Dec]
+mainWith opts builder = do
+    wf_q <- buildWorkflow wfName builder
+    main_q <- [d| main = mainFunc $(varE $ preAction opts) dag
+                    $(varE $ mkName wfName) (programHeader opts)
+              |]
+    return $ wf_q ++ main_q
+  where
+    wfName = "sciFlowDefaultMain"
+    dag = nmap (\(a,(_,b)) -> (a,b)) $ mkDAG builder
+{-# INLINE mainWith #-}
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,92 +1,144 @@
-{-# LANGUAGE GADTs #-}
-{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FlexibleInstances    #-}
+{-# LANGUAGE OverloadedStrings    #-}
+{-# LANGUAGE TemplateHaskell      #-}
 {-# LANGUAGE UndecidableInstances #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
 
 module Scientific.Workflow.Types
     ( WorkflowDB(..)
     , Workflow(..)
+    , Closure(..)
     , PID
-    , ProcState(..)
+    , NodeResult(..)
+    , ProcState
     , WorkflowState(..)
     , db
     , procStatus
+    , procParaControl
+    , remote
     , Processor
     , RunOpt(..)
-    , defaultRunOpt
-    , dbPath
-    , Serializable(..)
-    , Attribute
+    , BatchData(..)
+    , BatchData'(..)
+    , IsList
+    , DBData(..)
+    , Attribute(..)
+    , AttributeSetter
     , defaultAttribute
     , label
     , note
-    , def
+    , batch
+    , submitToRemote
     ) where
 
-import Control.Lens (makeLenses)
+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 Control.Exception (SomeException)
-import qualified Data.ByteString     as B
-import qualified Data.Map            as M
-import qualified Data.Text           as T
-import Data.Maybe (fromJust)
-import Data.Yaml (FromJSON, ToJSON, encode, decode)
-
-class Serializable a where
-    serialize :: a -> B.ByteString
-    deserialize :: B.ByteString -> a
+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)
 
-instance (FromJSON a, ToJSON a) => Serializable a where
-    serialize = encode
-    deserialize = fromJust . decode
+data HTrue
+data HFalse
 
--- | An abstract type representing the database used to store states of workflow
-data WorkflowDB  = WorkflowDB FilePath
+type family IsList a b where
+    IsList [a] [b] = HTrue
+    IsList a b = HFalse
 
--- | The id of a node
-type PID = T.Text
+class BatchData' flag a b where
+    batchFunction' :: flag -> (a -> IO b) -> Int -> (a -> [a], [b] -> b)
 
--- | The state of a computation node
-data ProcState = Success
-               | Scheduled
-               | Fail SomeException
+instance BatchData' HTrue [a] [b] where
+    batchFunction' _ _ i = (chunksOf i, concat)
 
-data WorkflowState = WorkflowState
-    { _db         :: WorkflowDB
-    , _procStatus :: M.Map PID ProcState
-    }
+instance BatchData' HFalse a b where
+    batchFunction' _ _ _ = (return, head)
 
-makeLenses ''WorkflowState
+class BatchData a b where
+    batchFunction :: (a -> IO b) -> Int -> (a -> [a], [b] -> b)
 
-type Processor a b = a -> StateT WorkflowState (ExceptT SomeException IO) b
+instance (IsList a b ~ flag, BatchData' flag a b) => BatchData a b where
+    batchFunction = batchFunction' (undefined :: flag)
 
-data Workflow where
-    Workflow :: (Processor () o) -> Workflow
+class DBData a where
+    serialize :: a -> B.ByteString
+    deserialize :: B.ByteString -> a
+    showYaml :: a -> B.ByteString
+    readYaml :: B.ByteString -> a
 
-data RunOpt = RunOpt
-    { _dbPath :: FilePath
-    }
+instance (FromJSON a, ToJSON a, S.Serialize a) => DBData a where
+    serialize = S.encode
+    deserialize = fromEither . S.decode
+      where
+        fromEither (Right x) = x
+        fromEither _ = error "decode failed"
+    showYaml = encode
+    readYaml = fromJust . decode
 
-makeLenses ''RunOpt
+-- | An abstract type representing the database used to store states of workflow
+newtype WorkflowDB  = WorkflowDB Connection
 
-defaultRunOpt :: RunOpt
-defaultRunOpt = RunOpt
-    { _dbPath = "wfDB" }
+-- | The id of a node
+type PID = T.Text
 
+-- | Node attribute
 data Attribute = Attribute
     { _label :: T.Text  -- ^ short description
-    , _note :: T.Text   -- ^ long description
+    , _note  :: T.Text   -- ^ long description
+    , _batch :: Int
+    , _submitToRemote :: Maybe Bool  -- ^ overwrite the global option
     }
 
+makeLenses ''Attribute
+
 defaultAttribute :: Attribute
 defaultAttribute = Attribute
     { _label = ""
     , _note = ""
+    , _batch = -1
+    , _submitToRemote = Nothing
     }
 
-makeLenses ''Attribute
+type AttributeSetter = State Attribute ()
 
-def :: State a ()
-def = return ()
+-- | The result of a computation node
+data NodeResult = Success
+                | Fail SomeException
+                | Scheduled
+
+data WorkflowState = WorkflowState
+    { _db          :: WorkflowDB
+    , _procStatus  :: M.Map PID (MVar NodeResult, Attribute)
+    , _procParaControl :: MVar () -- ^ concurrency controller
+    , _remote :: Bool
+    }
+
+makeLenses ''WorkflowState
+
+type ProcState b = StateT WorkflowState (ExceptT (PID, SomeException) IO) b
+type Processor a b = a -> ProcState b
+
+
+data Closure where
+    Closure :: (DBData a, DBData b) => (a -> IO b) -> Closure
+
+-- | A Workflow is a DAG
+data Workflow = Workflow (M.Map T.Text Attribute)
+                         (M.Map String Closure)
+                         (Processor () ())
+
+data RunOpt = RunOpt
+    { database :: FilePath
+    , nThread :: Int      -- ^ number of concurrent processes
+    , runOnRemote :: Bool
+    }
diff --git a/src/Scientific/Workflow/Utils.hs b/src/Scientific/Workflow/Utils.hs
new file mode 100644
--- /dev/null
+++ b/src/Scientific/Workflow/Utils.hs
@@ -0,0 +1,56 @@
+{-# LANGUAGE CPP #-}
+module Scientific.Workflow.Utils where
+
+import qualified Data.ByteString.Char8         as B
+import qualified Data.Text                     as T
+import           Debug.Trace                   (traceM)
+import           Rainbow
+import           System.IO
+
+import           Scientific.Workflow.Types     (DBData (..))
+import           System.Directory              (getCurrentDirectory)
+import           System.Environment.Executable (getExecutablePath)
+
+#ifdef SGE
+import           DRMAA                         (DrmaaAttribute (..),
+                                                defaultDrmaaConfig, drmaaRun,
+                                                withTmpFile)
+#endif
+
+debug :: Monad m => String -> m ()
+debug txt = traceM $ B.unpack $ B.concat $
+    chunksToByteStrings toByteStringsColors8 [prefix, chunk txt & fore green]
+  where
+    prefix = bold $ chunk "[DEBUG] " & fore green
+
+error' :: String -> IO ()
+error' txt = B.hPutStrLn stderr $ B.concat $
+    chunksToByteStrings toByteStringsColors8 [prefix, chunk txt & fore red]
+  where
+    prefix = bold $ chunk "[ERROR] " & fore red
+
+data RemoteOpts = RemoteOpts
+    { extraParams :: String
+    }
+
+defaultRemoteOpts :: RemoteOpts
+defaultRemoteOpts = RemoteOpts
+    { extraParams = ""
+    }
+
+runRemote :: (DBData a, DBData b) => RemoteOpts -> T.Text -> a -> IO b
+#ifdef SGE
+runRemote opts pid input = withTmpFile tmpDir $ \inputFl -> withTmpFile tmpDir $
+    \outputFl -> do
+        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 ()
+        deserialize <$> B.readFile outputFl
+  where
+    tmpDir = "./"
+#else
+runRemote = error "SGE support was not turned on."
+#endif
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
@@ -8,20 +8,20 @@
 import qualified Data.Text as T
 import qualified Data.Text.Lazy      as TL
 
-import Data.GraphViz
-import Data.GraphViz.Printing
-import Data.GraphViz.Attributes.Complete
+import qualified Data.GraphViz as G
+import qualified Data.GraphViz.Printing as G
+import qualified Data.GraphViz.Attributes.Complete as G
+import Data.Graph.Inductive.PatriciaTree (Gr)
 
-import Scientific.Workflow.Types (note)
+import Scientific.Workflow.Types
 import Scientific.Workflow.Builder
 
 -- | Print the computation graph
-renderBuilder :: Builder () -> TL.Text
-renderBuilder b = renderDot . toDot $ graphToDot param dag
+renderBuilder :: Gr (PID, Attribute) Int -> TL.Text
+renderBuilder dag = G.renderDot . G.toDot $ G.graphToDot param dag
   where
-    fmtnode (_, (p, (_, attr))) = [Label $ StrLabel $ TL.fromStrict lab]
+    fmtnode (_, (p, attr)) = [G.Label $ G.StrLabel $ TL.fromStrict lab]
       where
         lab | T.null (attr^.label) = p
             | otherwise = attr^.label
-    dag = mkDAG b
-    param = nonClusteredParams{fmtNode = fmtnode}
+    param = G.nonClusteredParams{G.fmtNode = fmtnode}