concurrent-output, first pass

Output without -Jn should be unchanged from before. With -Jn,
concurrent-output is used for messages, but regions are not used yet, so
it's a mess.

BuildFlags.hs

@@ -65,6 +65,9 @@ buildFlags = filter (not . null)
 #else
 #warning Building without XMPP.
 #endif
+#ifdef WITH_CONCURRENTOUTPUT
+	, "ConcurrentOutput"
+#endif
 #ifdef WITH_DNS
 	, "DNS"
 #endif

CmdLine.hs

@@ -24,6 +24,7 @@ import Annex.Action
 import Annex.Environment
 import Command
 import Types.Messages
+import Messages.Concurrent
 
 {- Runs the passed command line. -}
 dispatch :: Bool -> CmdParams -> [Command] -> [GlobalOption] -> [(String, String)] -> IO Git.Repo -> String -> String -> IO ()
@@ -45,8 +46,9 @@ dispatch fuzzyok allargs allcmds globaloptions fields getgitrepo progname progde
 			whenM (annexDebug <$> Annex.getGitConfig) $
 				liftIO enableDebugOutput
 			startup
-			performCommandAction cmd seek $
-				shutdown $ cmdnocommit cmd
+			withConcurrentOutput $
+				performCommandAction cmd seek $
+					shutdown $ cmdnocommit cmd
 	go (Left norepo) = do
 		let ingitrepo = \a -> a =<< Git.Config.global
 		-- Parse command line with full cmdparser first,

CmdLine/Action.hs

@@ -47,7 +47,7 @@ performCommandAction Command { cmdcheck = c, cmdname = name } seek cont = do
 commandAction :: CommandStart -> Annex ()
 commandAction a = withOutputType go 
   where
-	go (ParallelOutput n) = do
+	go (ConcurrentOutput n) = do
 		ws <- Annex.getState Annex.workers
 		(st, ws') <- if null ws
 			then do

CmdLine/GitAnnex/Options.hs

@@ -283,7 +283,7 @@ jsonOption = globalFlag (Annex.setOutput JSONOutput)
 	)
 
 jobsOption :: GlobalOption
-jobsOption = globalSetter (Annex.setOutput . ParallelOutput) $ 
+jobsOption = globalSetter (Annex.setOutput . ConcurrentOutput) $ 
 	option auto
 		( long "jobs" <> short 'J' <> metavar paramNumber
 		<> help "enable concurrent jobs"

Messages.hs

@@ -52,16 +52,15 @@ import Types.Key
 import qualified Annex
 
 showStart :: String -> FilePath -> Annex ()
-showStart command file = handleMessage (JSON.start command $ Just file) $
-	flushed $ putStr $ command ++ " " ++ file ++ " "
+showStart command file = outputMessage (JSON.start command $ Just file) $
+	command ++ " " ++ file ++ " "
 
 showStart' :: String -> Key -> Maybe FilePath -> Annex ()
 showStart' command key afile = showStart command $
 	fromMaybe (key2file key) afile
 
 showNote :: String -> Annex ()
-showNote s = handleMessage (JSON.note s) $
-	flushed $ putStr $ "(" ++ s ++ ") "
+showNote s = outputMessage (JSON.note s) $ "(" ++ s ++ ") "
 
 showAction :: String -> Annex ()
 showAction s = showNote $ s ++ "..."
@@ -76,7 +75,7 @@ showSideAction m = Annex.getState Annex.output >>= go
 			Annex.changeState $ \s -> s { Annex.output = st' }
 		| sideActionBlock st == InBlock = return ()
 		| otherwise = p
-	p = handleMessage q $ putStrLn $ "(" ++ m ++ "...)"
+	p = outputMessage q $ "(" ++ m ++ "...)\n"
 			
 showStoringStateAction :: Annex ()
 showStoringStateAction = showSideAction "recording state in git"
@@ -101,11 +100,10 @@ doSideAction' b a = do
 {- Make way for subsequent output of a command. -}
 showOutput :: Annex ()
 showOutput = unlessM commandProgressDisabled $
-	handleMessage q $ putStr "\n"
+	outputMessage q "\n"
 
 showLongNote :: String -> Annex ()
-showLongNote s = handleMessage (JSON.note s) $
-	putStrLn $ '\n' : indent s
+showLongNote s = outputMessage (JSON.note s) ('\n' : indent s ++ "\n")
 
 showEndOk :: Annex ()
 showEndOk = showEndResult True
@@ -114,7 +112,7 @@ showEndFail :: Annex ()
 showEndFail = showEndResult False
 
 showEndResult :: Bool -> Annex ()
-showEndResult ok = handleMessage (JSON.end ok) $ putStrLn $ endResult ok
+showEndResult ok = outputMessage (JSON.end ok) $ endResult ok ++ "\n"
 
 endResult :: Bool -> String
 endResult True = "ok"
@@ -129,11 +127,10 @@ warning = warning' True . indent
 warning' :: Bool -> String -> Annex ()
 warning' makeway w = do
 	when makeway $
-		handleMessage q $ putStr "\n"
-	liftIO $ do
-		hFlush stdout
-		hPutStrLn stderr w
+		outputMessage q "\n"
+	outputError (w ++ "\n")
 
+{- Not concurrent output safe. -}
 warningIO :: String -> IO ()
 warningIO w = do
 	putStr "\n"
@@ -145,7 +142,10 @@ indent = intercalate "\n" . map (\l -> "  " ++ l) . lines
 
 {- Shows a JSON fragment only when in json mode. -}
 maybeShowJSON :: JSON a => [(String, a)] -> Annex ()
-maybeShowJSON v = handleMessage (JSON.add v) q
+maybeShowJSON v = withOutputType $ liftIO . go
+  where
+	go JSONOutput = JSON.add v
+	go _ = return ()
 
 {- Shows a complete JSON value, only when in json mode. -}
 showFullJSON :: JSON a => [(String, a)] -> Annex Bool
@@ -157,19 +157,19 @@ showFullJSON v = withOutputType $ liftIO . go
 {- Performs an action that outputs nonstandard/customized output, and
  - in JSON mode wraps its output in JSON.start and JSON.end, so it's
  - a complete JSON document.
- - This is only needed when showStart and showEndOk is not used. -}
+ - This is only needed when showStart and showEndOk is not used.
+ -}
 showCustom :: String -> Annex Bool -> Annex ()
 showCustom command a = do
-	handleMessage (JSON.start command Nothing) q
+	outputMessage (JSON.start command Nothing) ""
 	r <- a
-	handleMessage (JSON.end r) q
+	outputMessage (JSON.end r) ""
 
 showHeader :: String -> Annex ()
-showHeader h = handleMessage q $
-	flushed $ putStr $ h ++ ": "
+showHeader h = outputMessage q $ (h ++ ": ")
 
 showRaw :: String -> Annex ()
-showRaw s = handleMessage q $ putStrLn s
+showRaw = outputMessage q
 
 setupConsole :: IO ()
 setupConsole = do
@@ -207,6 +207,6 @@ debugEnabled = do
 commandProgressDisabled :: Annex Bool
 commandProgressDisabled = withOutputType $ \t -> return $ case t of
 	QuietOutput -> True
-	ParallelOutput _ -> True
 	JSONOutput -> True
 	NormalOutput -> False
+	ConcurrentOutput _ -> True

Messages/Concurrent.hs

@@ -0,0 +1,33 @@
+{- git-annex concurrent output
+ -
+ - Copyright 2015 Joey Hess <id@joeyh.name>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+{-# LANGUAGE CPP #-}
+
+module Messages.Concurrent where
+
+import Common.Annex
+import Messages.Internal
+import Types.Messages
+
+#ifdef WITH_CONCURRENTOUTPUT
+import qualified System.Console.Concurrent as Console
+#endif
+
+{- Enable concurrent output when that has been requested.
+ -
+ - This should only be run once per git-annex lifetime, with
+ - everything that might generate messages run inside it.
+ -}
+withConcurrentOutput :: Annex a -> Annex a
+#ifdef WITH_CONCURRENTOUTPUT
+withConcurrentOutput a = withOutputType go
+  where
+	go (ConcurrentOutput _) = Console.withConcurrentOutput a
+	go _ = a
+#else
+withConcurrentOutput = id
+#endif

Messages/Internal.hs

@@ -5,6 +5,8 @@
  - Licensed under the GNU GPL version 3 or higher.
  -}
 
+{-# LANGUAGE CPP #-}
+
 module Messages.Internal where
 
 import Common
@@ -12,14 +14,38 @@ import Types
 import Types.Messages
 import qualified Annex
 
-handleMessage :: IO () -> IO () -> Annex ()
-handleMessage json normal = withOutputType go
+#ifdef WITH_CONCURRENTOUTPUT
+import System.Console.Concurrent
+#endif
+
+outputMessage :: IO () -> String -> Annex ()
+outputMessage json s = withOutputType go
   where
-	go NormalOutput = liftIO normal
+	go NormalOutput = liftIO $
+		flushed $ putStr s
 	go QuietOutput = q
-	go (ParallelOutput _) = q
+	go (ConcurrentOutput _) = liftIO $
+#ifdef WITH_CONCURRENTOUTPUT
+		outputConcurrent s
+#else
+		q
+#endif
 	go JSONOutput = liftIO $ flushed json
 
+outputError :: String -> Annex ()
+outputError s = withOutputType go
+  where
+	go (ConcurrentOutput _) = liftIO $
+#ifdef WITH_CONCURRENTOUTPUT
+		errorConcurrent s
+#else
+		q
+#endif
+	go _ = liftIO $ do
+		hFlush stdout
+		hPutStr stderr s
+		hFlush stderr
+
 q :: Monad m => m ()
 q = noop
 

Messages/Progress.hs

@@ -17,15 +17,14 @@ import Types
 import Types.Messages
 import Types.Key
 
-#ifdef WITH_ASCIIPROGRESS
-import System.Console.AsciiProgress
-import qualified System.Console.Terminal.Size as Terminal
+#ifdef WITH_CONCURRENTOUTPUT
+import System.Console.Concurrent
+import System.Console.Regions
 import Control.Concurrent
-#else
+#endif
 import Data.Progress.Meter
 import Data.Progress.Tracker
 import Data.Quantity
-#endif
 
 {- Shows a progress meter while performing a transfer of a key.
  - The action is passed a callback to use to update the meter. -}
@@ -36,7 +35,7 @@ metered combinemeterupdate key af a = case keySize key of
   where
 	go _ QuietOutput = nometer
 	go _ JSONOutput = nometer
-#ifdef WITH_ASCIIPROGRESS
+#if 0
 	go size _ = do
 		showOutput
 		liftIO $ putStrLn ""
@@ -65,8 +64,8 @@ metered combinemeterupdate key af a = case keySize key of
 
 		return r
 #else
-	-- Old progress bar code, not suitable for parallel output.
-	go _ (ParallelOutput _) = do
+	-- Old progress bar code, not suitable for concurrent output.
+	go _ (ConcurrentOutput _) = do
 		r <- nometer
 		liftIO $ putStrLn $ fromMaybe (key2file key) af
 		return r
@@ -79,7 +78,7 @@ metered combinemeterupdate key af a = case keySize key of
 		return r
 #endif
 
-#ifdef WITH_ASCIIPROGRESS
+#if 0
 	pupdate pg n = do
 		let i = fromBytesProcessed n
 		sofar <- stCompleted <$> getProgressStats pg
@@ -95,24 +94,17 @@ metered combinemeterupdate key af a = case keySize key of
 
 	nometer = a (const noop)
 
-#ifdef WITH_ASCIIPROGRESS
-	truncatepretty n s
-		| length s > n = take (n-2) s ++ ".."
-		| otherwise = s
-#endif
-
-{- Use when the progress meter is only desired for parallel
- - mode; as when a command's own progress output is preferred. -}
-parallelMetered :: Maybe MeterUpdate -> Key -> AssociatedFile -> (MeterUpdate -> Annex a) -> Annex a
-parallelMetered combinemeterupdate key af a = withOutputType go
+{- Use when the progress meter is only desired for concurrent
+ - output; as when a command's own progress output is preferred. -}
+concurrentMetered :: Maybe MeterUpdate -> Key -> AssociatedFile -> (MeterUpdate -> Annex a) -> Annex a
+concurrentMetered combinemeterupdate key af a = withOutputType go
   where
-	go (ParallelOutput _) = metered combinemeterupdate key af a
+	go (ConcurrentOutput _) = metered combinemeterupdate key af a
 	go _ = a (fromMaybe (const noop) combinemeterupdate)
 
 {- Progress dots. -}
 showProgressDots :: Annex ()
-showProgressDots = handleMessage q $
-	flushed $ putStr "."
+showProgressDots = outputMessage q "."
 
 {- Runs a command, that may output progress to either stdout or
  - stderr, as well as other messages.
@@ -149,5 +141,7 @@ mkStderrRelayer = do
 mkStderrEmitter :: Annex (String -> IO ())
 mkStderrEmitter = withOutputType go
   where
-	go (ParallelOutput _) = return $ \s -> hPutStrLn stderr ("E: " ++ s)
+#ifdef WITH_CONCURRENTOUTPUT
+	go (ConcurrentOutput _) = return errorConcurrent
+#endif
 	go _ = return (hPutStrLn stderr)

Remote/Git.hs

@@ -421,7 +421,7 @@ lockKey r key callback
 
 {- Tries to copy a key's content from a remote's annex to a file. -}
 copyFromRemote :: Remote -> Key -> AssociatedFile -> FilePath -> MeterUpdate -> Annex (Bool, Verification)
-copyFromRemote r key file dest p = parallelMetered (Just p) key file $
+copyFromRemote r key file dest p = concurrentMetered (Just p) key file $
 	copyFromRemote' r key file dest
 
 copyFromRemote' :: Remote -> Key -> AssociatedFile -> FilePath -> MeterUpdate -> Annex (Bool, Verification)
@@ -522,7 +522,7 @@ copyFromRemoteCheap r key af file
 			)
 	| Git.repoIsSsh (repo r) =
 		ifM (Annex.Content.preseedTmp key file)
-			( fst <$> parallelMetered Nothing key af
+			( fst <$> concurrentMetered Nothing key af
 				(copyFromRemote' r key af file)
 			, return False
 			)
@@ -533,7 +533,7 @@ copyFromRemoteCheap _ _ _ _ = return False
 
 {- Tries to copy a key's content to a remote's annex. -}
 copyToRemote :: Remote -> Key -> AssociatedFile -> MeterUpdate -> Annex Bool
-copyToRemote r key file p = parallelMetered (Just p) key file $ copyToRemote' r key file
+copyToRemote r key file p = concurrentMetered (Just p) key file $ copyToRemote' r key file
 
 copyToRemote' :: Remote -> Key -> AssociatedFile -> MeterUpdate -> Annex Bool
 copyToRemote' r key file p

Types/Messages.hs

@@ -9,7 +9,7 @@ module Types.Messages where
 
 import Data.Default
 
-data OutputType = NormalOutput | QuietOutput | ParallelOutput Int | JSONOutput
+data OutputType = NormalOutput | QuietOutput | ConcurrentOutput Int | JSONOutput
 
 data SideActionBlock = NoBlock | StartBlock | InBlock
 	deriving (Eq)

debian/changelog

@@ -1,3 +1,9 @@
+git-annex (5.20151102.2) UNRELEASED; urgency=medium
+
+  * Use concurrent-output library for -Jn mode.
+
+ -- Joey Hess <id@joeyh.name>  Wed, 04 Nov 2015 12:50:20 -0400
+
 git-annex (5.20151102.1) unstable; urgency=medium
 
   * Avoid installing desktop file and program file if cabal install

git-annex.cabal

@@ -93,8 +93,8 @@ Flag DesktopNotify
 Flag TorrentParser
   Description: Use haskell torrent library to parse torrent files
 
-Flag AsciiProgress
-  Description: Use ascii-progress library (experimental)
+Flag ConcurrentOutput
+  Description: Use concurrent-output library
   Default: False
 
 Flag EKG
@@ -278,9 +278,9 @@ Executable git-annex
     Build-Depends: esqueleto, persistent-sqlite, persistent, persistent-template
     CPP-Options: -DWITH_DATABASE
 
-  if flag(AsciiProgress)
-    Build-Depends: ascii-progress (<= 0.2.1.2), terminal-size
-    CPP-Options: -DWITH_ASCIIPROGRESS
+  if flag(ConcurrentOutput)
+    Build-Depends: concurrent-output (>= 1.4.1)
+    CPP-Options: -DWITH_CONCURRENTOUTPUT
 
   if flag(EKG)
     Build-Depends: ekg