diff options
Diffstat (limited to 'CmdLine')
| -rw-r--r-- | CmdLine/Action.hs | 86 | ||||
| -rw-r--r-- | CmdLine/GitAnnex/Options.hs | 8 | ||||
| -rw-r--r-- | CmdLine/Seek.hs | 2 |
3 files changed, 89 insertions, 7 deletions
diff --git a/CmdLine/Action.hs b/CmdLine/Action.hs index b566621bb..2838e4ff8 100644 --- a/CmdLine/Action.hs +++ b/CmdLine/Action.hs @@ -1,6 +1,6 @@ {- git-annex command-line actions - - - Copyright 2010-2014 Joey Hess <id@joeyh.name> + - Copyright 2010-2015 Joey Hess <id@joeyh.name> - - Licensed under the GNU GPL version 3 or higher. -} @@ -9,10 +9,15 @@ module CmdLine.Action where import Common.Annex import qualified Annex +import Annex.Concurrent import Types.Command import qualified Annex.Queue +import Messages.Internal +import Types.Messages -type CommandActionRunner = CommandStart -> CommandCleanup +import Control.Concurrent.Async +import Control.Exception (throwIO) +import Data.Either {- Runs a command, starting with the check stage, and then - the seek stage. Finishes by running the continutation, and @@ -22,6 +27,7 @@ performCommandAction Command { cmdseek = seek, cmdcheck = c, cmdname = name } pa mapM_ runCheck c Annex.changeState $ \s -> s { Annex.errcounter = 0 } seek params + finishCommandActions cont showerrcount =<< Annex.getState Annex.errcounter where @@ -33,9 +39,77 @@ performCommandAction Command { cmdseek = seek, cmdcheck = c, cmdname = name } pa - including by throwing IO errors (but other errors terminate the whole - command). - - - This should only be run in the seek stage. -} -commandAction :: CommandActionRunner -commandAction a = account =<< tryIO go + - When concurrency is enabled, a thread is forked off to run the action + - in the background, as soon as a free slot is available. + + - This should only be run in the seek stage. + -} +commandAction :: CommandStart -> Annex () +commandAction a = withOutputType go + where + go (ParallelOutput n) = do + ws <- Annex.getState Annex.workers + (st, ws') <- if null ws + then do + st <- dupState + return (st, replicate (n-1) (Left st)) + else do + l <- liftIO $ drainTo (n-1) ws + findFreeSlot l + w <- liftIO $ async $ snd <$> Annex.run st run + Annex.changeState $ \s -> s { Annex.workers = Right w:ws' } + go _ = run + run = void $ includeCommandAction a + +{- Waits for any forked off command actions to finish. + - + - Merge together the cleanup actions of all the AnnexStates used by + - threads, into the current Annex's state, so they'll run at shutdown. + - + - Also merge together the errcounters of the AnnexStates. + -} +finishCommandActions :: Annex () +finishCommandActions = do + l <- liftIO . drainTo 0 =<< Annex.getState Annex.workers + forM_ (lefts l) mergeState + +{- Wait for Asyncs from the list to finish, replacing them with their + - final AnnexStates, until the list of remaining Asyncs is not larger + - than the specified size, then returns the new list. + - + - If the action throws an exception, it is propigated, but first + - all other actions are waited for, to allow for a clean shutdown. + -} +drainTo + :: Int + -> [Either Annex.AnnexState (Async Annex.AnnexState)] + -> IO [Either Annex.AnnexState (Async Annex.AnnexState)] +drainTo sz l + | null as || sz >= length as = return l + | otherwise = do + (done, ret) <- waitAnyCatch as + let as' = filter (/= done) as + case ret of + Left e -> do + void $ drainTo 0 (map Left sts ++ map Right as') + throwIO e + Right st -> do + drainTo sz $ map Left (st:sts) ++ map Right as' + where + (sts, as) = partitionEithers l + +findFreeSlot :: [Either Annex.AnnexState (Async Annex.AnnexState)] -> Annex (Annex.AnnexState, [Either Annex.AnnexState (Async Annex.AnnexState)]) +findFreeSlot = go [] + where + go c [] = do + st <- dupState + return (st, c) + go c (Left st:rest) = return (st, c ++ rest) + go c (v:rest) = go (v:c) rest + +{- Like commandAction, but without the concurrency. -} +includeCommandAction :: CommandStart -> CommandCleanup +includeCommandAction a = account =<< tryIO go where go = do Annex.Queue.flushWhenFull @@ -53,7 +127,7 @@ commandAction a = account =<< tryIO go {- Runs a single command action through the start, perform and cleanup - stages, without catching errors. Useful if one command wants to run - part of another command. -} -callCommandAction :: CommandActionRunner +callCommandAction :: CommandStart -> CommandCleanup callCommandAction = start where start = stage $ maybe skip perform diff --git a/CmdLine/GitAnnex/Options.hs b/CmdLine/GitAnnex/Options.hs index be1c74ede..38fa93090 100644 --- a/CmdLine/GitAnnex/Options.hs +++ b/CmdLine/GitAnnex/Options.hs @@ -138,6 +138,14 @@ jsonOption :: Option jsonOption = Option ['j'] ["json"] (NoArg (Annex.setOutput JSONOutput)) "enable JSON output" +jobsOption :: Option +jobsOption = Option ['J'] ["jobs"] (ReqArg set paramNumber) + "enable concurrent jobs" + where + set s = case readish s of + Nothing -> error "Bad --jobs number" + Just n -> Annex.setOutput (ParallelOutput n) + timeLimitOption :: Option timeLimitOption = Option ['T'] ["time-limit"] (ReqArg Limit.addTimeLimit paramTime) diff --git a/CmdLine/Seek.hs b/CmdLine/Seek.hs index 96076261f..3166ab83d 100644 --- a/CmdLine/Seek.hs +++ b/CmdLine/Seek.hs @@ -84,7 +84,7 @@ withFilesInRefs a = mapM_ go case v of Nothing -> noop Just k -> whenM (matcher $ MatchingKey k) $ - void $ commandAction $ a f k + commandAction $ a f k withPathContents :: ((FilePath, FilePath) -> CommandStart) -> CommandSeek withPathContents a params = do |