packages feed

concurrent-output (empty) → 1.0.0

raw patch · 9 files changed

+761/−0 lines, 9 filesdep +MissingHdep +asyncdep +basesetup-changed

Dependencies added: MissingH, async, base, bytestring, containers, directory, exceptions, process, stm, transformers, unix

Files

+ CHANGELOG view
@@ -0,0 +1,5 @@+concurrent-output (1.0.0) unstable; urgency=medium++  * First release.++ -- Joey Hess <id@joeyh.name>  Wed, 28 Oct 2015 21:01:23 -0400
+ Control/Concurrent/Output.hs view
@@ -0,0 +1,333 @@+{-# LANGUAGE BangPatterns #-}+{-# OPTIONS_GHC -fno-warn-tabs #-}++-- | +-- Copyright: 2013 Joey Hess <id@joeyh.name>+-- License: BSD-2-clause+-- +-- Concurrent output handling.+--+-- > import Control.Concurrent.Async+-- > import Control.Concurrent.Output+-- >+-- > main = withConcurrentOutput $+-- > 	outputConcurrent "washed the car\n"+-- > 		`concurrently`+-- >	outputConcurrent "walked the dog\n"+-- >		`concurrently`+-- > 	createProcessConcurrent (proc "ls" [])++module Control.Concurrent.Output (+	withConcurrentOutput,+	flushConcurrentOutput,+	outputConcurrent,+	createProcessConcurrent,+	waitForProcessConcurrent,+	lockOutput,+) where++import System.IO+import System.Posix.IO+import System.Directory+import System.Exit+import Control.Monad+import Control.Monad.IO.Class (liftIO, MonadIO)+import Control.Applicative+import System.IO.Unsafe (unsafePerformIO)+import Control.Concurrent+import Control.Concurrent.STM+import Control.Concurrent.Async+import Data.Maybe+import Data.List+import Data.Monoid+import qualified Data.ByteString as B+import qualified System.Process as P+import qualified Data.Set as S++import Utility.Monad+import Utility.Exception+import Utility.FileSystemEncoding++data OutputHandle = OutputHandle+	{ outputLock :: TMVar Lock+	, outputBuffer :: TMVar Buffer+	, outputThreads :: TMVar (S.Set (Async ()))+	}++data Lock = Locked++-- | A shared global variable for the OutputHandle.+{-# NOINLINE globalOutputHandle #-}+globalOutputHandle :: MVar OutputHandle+globalOutputHandle = unsafePerformIO $ +	newMVar =<< OutputHandle+		<$> newEmptyTMVarIO+		<*> newTMVarIO []+		<*> newTMVarIO S.empty++-- | Gets the global OutputHandle.+getOutputHandle :: IO OutputHandle+getOutputHandle = readMVar globalOutputHandle++-- | Holds a lock while performing an action that will display output.+-- While this is running, other threads that try to lockOutput will block,+-- and calls to `outputConcurrent` and `createProcessConcurrent`+-- will result in that concurrent output being buffered and not+-- displayed until the action is done.+lockOutput :: (MonadIO m, MonadMask m) => m a -> m a+lockOutput = bracket_ (liftIO takeOutputLock) (liftIO dropOutputLock)++-- | Blocks until we have the output lock.+takeOutputLock :: IO ()+takeOutputLock = void $ takeOutputLock' True++-- | Tries to take the output lock, without blocking.+tryTakeOutputLock :: IO Bool+tryTakeOutputLock = takeOutputLock' False++withLock :: (TMVar Lock -> STM a) -> IO a+withLock a = do+	lck <- outputLock <$> getOutputHandle+	atomically (a lck)++takeOutputLock' :: Bool -> IO Bool+takeOutputLock' block = do+	locked <- withLock $ \l -> do+		v <- tryTakeTMVar l+		case v of+			Just Locked+				| block -> retry+				| otherwise -> do+					-- Restore value we took.+					putTMVar l Locked+					return False+			Nothing -> do+				putTMVar l Locked+				return True+	when locked $ do+		bv <- outputBuffer <$> getOutputHandle+		buf <- atomically $ swapTMVar bv []+		emitBuffer stdout buf+	return locked++-- | Only safe to call after taking the output lock.+dropOutputLock :: IO ()+dropOutputLock = withLock $ void . takeTMVar++-- | Use this around any IO actions that use `outputConcurrent`+-- or `createProcessConcurrent`+--+-- This is necessary to ensure that buffered concurrent output actually+-- gets displayed before the program exits.+withConcurrentOutput :: IO a -> IO a+withConcurrentOutput a = a `finally` flushConcurrentOutput++-- | Blocks until any processes started by `createProcessConcurrent` have+-- finished, and any buffered output is displayed.+flushConcurrentOutput :: IO ()+flushConcurrentOutput = do+	-- Wait for all outputThreads to finish.+	v <- outputThreads <$> getOutputHandle+	atomically $ do+		r <- takeTMVar v+		if r == S.empty+			then putTMVar v r+			else retry+	-- Take output lock to ensure that nothing else is currently+	-- generating output, and flush any buffered output.+	lockOutput $ return ()++-- | Displays a string to stdout, and flush output so it's displayed.+--+-- Uses locking to ensure that the whole string is output atomically+-- even when other threads are concurrently generating output.+--+-- When something else is writing to the console at the same time, this does+-- not block. It buffers the string, so it will be displayed once the other+-- writer is done.+outputConcurrent :: String -> IO ()+outputConcurrent s = bracket setup cleanup go+  where+	setup = tryTakeOutputLock+	cleanup False = return ()+	cleanup True = dropOutputLock+	go True = do+		putStr s+		hFlush stdout+	go False = do+		bv <- outputBuffer <$> getOutputHandle+		oldbuf <- atomically $ takeTMVar bv+		newbuf <- addBuffer (Output (B.pack (decodeW8NUL s))) oldbuf+		atomically $ putTMVar bv newbuf++-- | This must be used to wait for processes started with +-- `createProcessConcurrent`.+--+-- This is necessary because `System.Process.waitForProcess` has a+-- race condition when two threads check the same process. If the race+-- is triggered, one thread will successfully wait, but the other+-- throws a DoesNotExist exception.+waitForProcessConcurrent :: P.ProcessHandle -> IO ExitCode+waitForProcessConcurrent h = do+	v <- tryWhenExists (P.waitForProcess h)+	case v of+		Just r -> return r+		Nothing -> maybe (waitForProcessConcurrent h) return =<< P.getProcessExitCode h++-- | Wrapper around `System.Process.createProcess` that prevents +-- multiple processes that are running concurrently from writing+-- to stdout/stderr at the same time.+--+-- If the process does not output to stdout or stderr, it's run+-- by createProcess entirely as usual. Only processes that can generate+-- output are handled specially:+--+-- A process is allowed to write to stdout and stderr in the usual+-- way, assuming it can successfully take the output lock.+--+-- When the output lock is held (by another concurrent process,+-- or because `outputConcurrent` is being called at the same time),+-- the process is instead run with its stdout and stderr+-- redirected to a buffer. The buffered output will be displayed as soon+-- as the output lock becomes free.+createProcessConcurrent :: P.CreateProcess -> IO (Maybe Handle, Maybe Handle, Maybe Handle, P.ProcessHandle) +createProcessConcurrent p+	| willOutput (P.std_out p) || willOutput (P.std_err p) =+		ifM tryTakeOutputLock+			( firstprocess+			, concurrentprocess+			)+	| otherwise = P.createProcess p+  where+	rediroutput ss h+		| willOutput ss = P.UseHandle h+		| otherwise = ss++	firstprocess = do+		r@(_, _, _, h) <- P.createProcess p+			`onException` dropOutputLock+		-- Wait for the process to exit and drop the lock.+		void $ async $ do+			void $ tryIO $ waitForProcessConcurrent h+			dropOutputLock+		return r+	+	concurrentprocess = do+		(toouth, fromouth) <- pipe+		(toerrh, fromerrh) <- pipe+		let p' = p+			{ P.std_out = rediroutput (P.std_out p) toouth+			, P.std_err = rediroutput (P.std_err p) toerrh+			}+		r <- P.createProcess p'+		outbuf <- setupBuffer stdout toouth (P.std_out p) fromouth+		errbuf <- setupBuffer stderr toerrh (P.std_err p) fromerrh+		void $ async $ bufferWriter [outbuf, errbuf]+		return r++	pipe = do+		(from, to) <- createPipe+		(,) <$> fdToHandle to <*> fdToHandle from++willOutput :: P.StdStream -> Bool+willOutput P.Inherit = True+willOutput _ = False++-- Built up with newest seen output first.+type Buffer = [BufferedActivity]++data BufferedActivity+	= ReachedEnd+	| Output B.ByteString+	| InTempFile FilePath+	deriving (Eq)++setupBuffer :: Handle -> Handle -> P.StdStream -> Handle -> IO (Handle, MVar Buffer, TMVar ())+setupBuffer h toh ss fromh = do+	hClose toh+	buf <- newMVar []+	bufsig <- atomically newEmptyTMVar+	void $ async $ outputDrainer ss fromh buf bufsig+	return (h, buf, bufsig)++-- Drain output from the handle, and buffer it.+outputDrainer :: P.StdStream -> Handle -> MVar Buffer -> TMVar () -> IO ()+outputDrainer ss fromh buf bufsig+	| willOutput ss = go+	| otherwise = atend+  where+	go = do+		v <- tryIO $ B.hGetSome fromh 1048576+		case v of+			Right b | not (B.null b) -> do+				modifyMVar_ buf $ addBuffer (Output b)+				changed+				go+			_ -> atend+	atend = do+		modifyMVar_ buf $ pure . (ReachedEnd :)+		changed+		hClose fromh+	changed = atomically $ do+		void $ tryTakeTMVar bufsig+		putTMVar bufsig ()++-- Wait to lock output, and once we can, display everything +-- that's put into the buffers, until the end.+bufferWriter :: [(Handle, MVar Buffer, TMVar ())] -> IO ()+bufferWriter ts = do+	worker <- async $ void $ lockOutput $ mapConcurrently go ts+	v <- outputThreads <$> getOutputHandle+	atomically $ do+		s <- takeTMVar v+		putTMVar v (S.insert worker s)+	void $ async $ do+		void $ waitCatch worker+		atomically $ do+			s <- takeTMVar v+			putTMVar v (S.delete worker s)+  where+	go v@(outh, buf, bufsig) = do+		void $ atomically $ takeTMVar bufsig+		l <- takeMVar buf+		putMVar buf []+		emitBuffer outh l+		if any (== ReachedEnd) l+			then return ()+			else go v++emitBuffer :: Handle -> Buffer -> IO ()+emitBuffer outh l = forM_ (reverse l) $ \ba -> case ba of+	Output b -> do+		B.hPut outh b+		hFlush outh+	InTempFile tmp -> do+		B.hPut outh =<< B.readFile tmp+		void $ tryWhenExists $ removeFile tmp+	ReachedEnd -> return ()++-- Adds a value to the Buffer. When adding Output to a Handle, it's cheaper+-- to combine it with any already buffered Output to that same Handle.+--+-- When the total buffered Output exceeds 1 mb in size, it's moved out of+-- memory, to a temp file. This should only happen rarely, but is done to+-- avoid some verbose process unexpectedly causing excessive memory use.+addBuffer :: BufferedActivity -> Buffer -> IO Buffer+addBuffer (Output b) buf+	| B.length b' <= 1048576 = return (Output b' : other)+	| otherwise = do+		tmpdir <- getTemporaryDirectory+		(tmp, h) <- openTempFile tmpdir "output.tmp"+		B.hPut h b'+		hClose h+		return (InTempFile tmp : other)+  where+	!b' = B.concat (mapMaybe getOutput this) <> b+	!(this, other) = partition isOutput buf+	isOutput v = case v of+		Output _ -> True+		_ -> False+	getOutput v = case v of+		Output b'' -> Just b''+		_ -> Nothing+addBuffer v buf = return (v:buf)
+ LICENSE view
@@ -0,0 +1,22 @@+Copyright 2015 Joey Hess <id@joeyh.name>.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:+1. Redistributions of source code must retain the above copyright+   notice, this list of conditions and the following disclaimer.+2. Redistributions in binary form must reproduce the above copyright+   notice, this list of conditions and the following disclaimer in the+   documentation and/or other materials provided with the distribution.++THIS SOFTWARE IS PROVIDED BY AUTHORS AND CONTRIBUTORS ``AS IS'' AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE+ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS+OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)+HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT+LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY+OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF+SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,5 @@+{- cabal setup file -}++import Distribution.Simple++main = defaultMain
+ Utility/Data.hs view
@@ -0,0 +1,19 @@+{- utilities for simple data types+ -+ - Copyright 2013 Joey Hess <id@joeyh.name>+ -+ - License: BSD-2-clause+ -}++{-# OPTIONS_GHC -fno-warn-tabs #-}++module Utility.Data where++{- First item in the list that is not Nothing. -}+firstJust :: Eq a => [Maybe a] -> Maybe a+firstJust ms = case dropWhile (== Nothing) ms of+	[] -> Nothing+	(md:_) -> md++eitherToMaybe :: Either a b -> Maybe b+eitherToMaybe = either (const Nothing) Just
+ Utility/Exception.hs view
@@ -0,0 +1,98 @@+{- Simple IO exception handling (and some more)+ -+ - Copyright 2011-2015 Joey Hess <id@joeyh.name>+ -+ - License: BSD-2-clause+ -}++{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -fno-warn-tabs #-}++module Utility.Exception (+	module X,+	catchBoolIO,+	catchMaybeIO,+	catchDefaultIO,+	catchMsgIO,+	catchIO,+	tryIO,+	bracketIO,+	catchNonAsync,+	tryNonAsync,+	tryWhenExists,+	catchHardwareFault,+) where++import Control.Monad.Catch as X hiding (Handler)+import qualified Control.Monad.Catch as M+import Control.Exception (IOException, AsyncException)+import Control.Monad+import Control.Monad.IO.Class (liftIO, MonadIO)+import System.IO.Error (isDoesNotExistError, ioeGetErrorType)+import GHC.IO.Exception (IOErrorType(..))++import Utility.Data++{- Catches IO errors and returns a Bool -}+catchBoolIO :: MonadCatch m => m Bool -> m Bool+catchBoolIO = catchDefaultIO False++{- Catches IO errors and returns a Maybe -}+catchMaybeIO :: MonadCatch m => m a -> m (Maybe a)+catchMaybeIO a = catchDefaultIO Nothing $ a >>= (return . Just)++{- Catches IO errors and returns a default value. -}+catchDefaultIO :: MonadCatch m => a -> m a -> m a+catchDefaultIO def a = catchIO a (const $ return def)++{- Catches IO errors and returns the error message. -}+catchMsgIO :: MonadCatch m => m a -> m (Either String a)+catchMsgIO a = do+	v <- tryIO a+	return $ either (Left . show) Right v++{- catch specialized for IO errors only -}+catchIO :: MonadCatch m => m a -> (IOException -> m a) -> m a+catchIO = M.catch++{- try specialized for IO errors only -}+tryIO :: MonadCatch m => m a -> m (Either IOException a)+tryIO = M.try++{- bracket with setup and cleanup actions lifted to IO.+ -+ - Note that unlike catchIO and tryIO, this catches all exceptions. -}+bracketIO :: (MonadMask m, MonadIO m) => IO v -> (v -> IO b) -> (v -> m a) -> m a+bracketIO setup cleanup = bracket (liftIO setup) (liftIO . cleanup)++{- Catches all exceptions except for async exceptions.+ - This is often better to use than catching them all, so that+ - ThreadKilled and UserInterrupt get through.+ -}+catchNonAsync :: MonadCatch m => m a -> (SomeException -> m a) -> m a+catchNonAsync a onerr = a `catches`+	[ M.Handler (\ (e :: AsyncException) -> throwM e)+	, M.Handler (\ (e :: SomeException) -> onerr e)+	]++tryNonAsync :: MonadCatch m => m a -> m (Either SomeException a)+tryNonAsync a = go `catchNonAsync` (return . Left)+  where+	go = do+		v <- a+		return (Right v)++{- Catches only DoesNotExist exceptions, and lets all others through. -}+tryWhenExists :: MonadCatch m => m a -> m (Maybe a)+tryWhenExists a = do+	v <- tryJust (guard . isDoesNotExistError) a+	return (eitherToMaybe v)++{- Catches only exceptions caused by hardware faults.+ - Ie, disk IO error. -}+catchHardwareFault :: MonadCatch m => m a -> (IOException -> m a) -> m a+catchHardwareFault a onhardwareerr = catchIO a onlyhw+  where+	onlyhw e+		| ioeGetErrorType e == HardwareFault = onhardwareerr e+		| otherwise = throwM e
+ Utility/FileSystemEncoding.hs view
@@ -0,0 +1,165 @@+{- GHC File system encoding handling.+ -+ - Copyright 2012-2014 Joey Hess <id@joeyh.name>+ -+ - License: BSD-2-clause+ -}++{-# LANGUAGE CPP #-}+{-# OPTIONS_GHC -fno-warn-tabs #-}++module Utility.FileSystemEncoding (+	fileEncoding,+	withFilePath,+	md5FilePath,+	decodeBS,+	encodeBS,+	decodeW8,+	encodeW8,+	encodeW8NUL,+	decodeW8NUL,+	truncateFilePath,+) where++import qualified GHC.Foreign as GHC+import qualified GHC.IO.Encoding as Encoding+import Foreign.C+import System.IO+import System.IO.Unsafe+import qualified Data.Hash.MD5 as MD5+import Data.Word+import Data.Bits.Utils+import Data.List.Utils+import qualified Data.ByteString.Lazy as L+#ifdef mingw32_HOST_OS+import qualified Data.ByteString.Lazy.UTF8 as L8+#endif++import Utility.Exception++{- Sets a Handle to use the filesystem encoding. This causes data+ - written or read from it to be encoded/decoded the same+ - as ghc 7.4 does to filenames etc. This special encoding+ - allows "arbitrary undecodable bytes to be round-tripped through it".+ -}+fileEncoding :: Handle -> IO ()+#ifndef mingw32_HOST_OS+fileEncoding h = hSetEncoding h =<< Encoding.getFileSystemEncoding+#else+{- The file system encoding does not work well on Windows,+ - and Windows only has utf FilePaths anyway. -}+fileEncoding h = hSetEncoding h Encoding.utf8+#endif++{- Marshal a Haskell FilePath into a NUL terminated C string using temporary+ - storage. The FilePath is encoded using the filesystem encoding,+ - reversing the decoding that should have been done when the FilePath+ - was obtained. -}+withFilePath :: FilePath -> (CString -> IO a) -> IO a+withFilePath fp f = Encoding.getFileSystemEncoding+	>>= \enc -> GHC.withCString enc fp f++{- Encodes a FilePath into a String, applying the filesystem encoding.+ -+ - There are very few things it makes sense to do with such an encoded+ - string. It's not a legal filename; it should not be displayed.+ - So this function is not exported, but instead used by the few functions+ - that can usefully consume it.+ -+ - This use of unsafePerformIO is belived to be safe; GHC's interface+ - only allows doing this conversion with CStrings, and the CString buffer+ - is allocated, used, and deallocated within the call, with no side+ - effects.+ -+ - If the FilePath contains a value that is not legal in the filesystem+ - encoding, rather than thowing an exception, it will be returned as-is.+ -}+{-# NOINLINE _encodeFilePath #-}+_encodeFilePath :: FilePath -> String+_encodeFilePath fp = unsafePerformIO $ do+	enc <- Encoding.getFileSystemEncoding+	GHC.withCString enc fp (GHC.peekCString Encoding.char8)+		`catchNonAsync` (\_ -> return fp)++{- Encodes a FilePath into a Md5.Str, applying the filesystem encoding. -}+md5FilePath :: FilePath -> MD5.Str+md5FilePath = MD5.Str . _encodeFilePath++{- Decodes a ByteString into a FilePath, applying the filesystem encoding. -}+decodeBS :: L.ByteString -> FilePath+#ifndef mingw32_HOST_OS+decodeBS = encodeW8NUL . L.unpack+#else+{- On Windows, we assume that the ByteString is utf-8, since Windows+ - only uses unicode for filenames. -}+decodeBS = L8.toString+#endif++{- Encodes a FilePath into a ByteString, applying the filesystem encoding. -}+encodeBS :: FilePath -> L.ByteString+#ifndef mingw32_HOST_OS+encodeBS = L.pack . decodeW8NUL+#else+encodeBS = L8.fromString+#endif++{- Converts a [Word8] to a FilePath, encoding using the filesystem encoding.+ -+ - w82c produces a String, which may contain Chars that are invalid+ - unicode. From there, this is really a simple matter of applying the+ - file system encoding, only complicated by GHC's interface to doing so.+ -+ - Note that the encoding stops at any NUL in the input. FilePaths+ - do not normally contain embedded NUL, but Haskell Strings may.+ -}+{-# NOINLINE encodeW8 #-}+encodeW8 :: [Word8] -> FilePath+encodeW8 w8 = unsafePerformIO $ do+	enc <- Encoding.getFileSystemEncoding+	GHC.withCString Encoding.char8 (w82s w8) $ GHC.peekCString enc++{- Useful when you want the actual number of bytes that will be used to+ - represent the FilePath on disk. -}+decodeW8 :: FilePath -> [Word8]+decodeW8 = s2w8 . _encodeFilePath++{- Like encodeW8 and decodeW8, but NULs are passed through unchanged. -}+encodeW8NUL :: [Word8] -> FilePath+encodeW8NUL = join nul . map encodeW8 . split (s2w8 nul)+  where+	nul = ['\NUL']++decodeW8NUL :: FilePath -> [Word8]+decodeW8NUL = join (s2w8 nul) . map decodeW8 . split nul+  where+	nul = ['\NUL']++{- Truncates a FilePath to the given number of bytes (or less),+ - as represented on disk.+ -+ - Avoids returning an invalid part of a unicode byte sequence, at the+ - cost of efficiency when running on a large FilePath.+ -}+truncateFilePath :: Int -> FilePath -> FilePath+#ifndef mingw32_HOST_OS+truncateFilePath n = go . reverse+  where+	go f =+		let bytes = decodeW8 f+		in if length bytes <= n+			then reverse f+			else go (drop 1 f)+#else+{- On Windows, count the number of bytes used by each utf8 character. -}+truncateFilePath n = reverse . go [] n . L8.fromString+  where+	go coll cnt bs+		| cnt <= 0 = coll+		| otherwise = case L8.decode bs of+			Just (c, x) | c /= L8.replacement_char ->+				let x' = fromIntegral x+				in if cnt - x' < 0+					then coll+					else go (c:coll) (cnt - x') (L8.drop 1 bs)+			_ -> coll+#endif
+ Utility/Monad.hs view
@@ -0,0 +1,71 @@+{- monadic stuff+ -+ - Copyright 2010-2012 Joey Hess <id@joeyh.name>+ -+ - License: BSD-2-clause+ -}++{-# OPTIONS_GHC -fno-warn-tabs #-}++module Utility.Monad where++import Data.Maybe+import Control.Monad++{- Return the first value from a list, if any, satisfying the given+ - predicate -}+firstM :: Monad m => (a -> m Bool) -> [a] -> m (Maybe a)+firstM _ [] = return Nothing+firstM p (x:xs) = ifM (p x) (return $ Just x , firstM p xs)++{- Runs the action on values from the list until it succeeds, returning+ - its result. -}+getM :: Monad m => (a -> m (Maybe b)) -> [a] -> m (Maybe b)+getM _ [] = return Nothing+getM p (x:xs) = maybe (getM p xs) (return . Just) =<< p x++{- Returns true if any value in the list satisfies the predicate,+ - stopping once one is found. -}+anyM :: Monad m => (a -> m Bool) -> [a] -> m Bool+anyM p = liftM isJust . firstM p++allM :: Monad m => (a -> m Bool) -> [a] -> m Bool+allM _ [] = return True+allM p (x:xs) = p x <&&> allM p xs++{- Runs an action on values from a list until it succeeds. -}+untilTrue :: Monad m => [a] -> (a -> m Bool) -> m Bool+untilTrue = flip anyM++{- if with a monadic conditional. -}+ifM :: Monad m => m Bool -> (m a, m a) -> m a+ifM cond (thenclause, elseclause) = do+	c <- cond+	if c then thenclause else elseclause++{- short-circuiting monadic || -}+(<||>) :: Monad m => m Bool -> m Bool -> m Bool+ma <||> mb = ifM ma ( return True , mb )++{- short-circuiting monadic && -}+(<&&>) :: Monad m => m Bool -> m Bool -> m Bool+ma <&&> mb = ifM ma ( mb , return False )++{- Same fixity as && and || -}+infixr 3 <&&>+infixr 2 <||>++{- Runs an action, passing its value to an observer before returning it. -}+observe :: Monad m => (a -> m b) -> m a -> m a+observe observer a = do+	r <- a+	_ <- observer r+	return r++{- b `after` a runs first a, then b, and returns the value of a -}+after :: Monad m => m b -> m a -> m a+after = observe . const++{- do nothing -}+noop :: Monad m => m ()+noop = return ()
+ concurrent-output.cabal view
@@ -0,0 +1,43 @@+Name: concurrent-output+Version: 1.0.0+Cabal-Version: >= 1.8+License: BSD3+Maintainer: Joey Hess <id@joeyh.name>+Author: Joey Hess+Stability: Stable+Copyright: 2015 Joey Hess+License-File: LICENSE+Build-Type: Simple+Category: Concurrency+Synopsis: handling concurrent output+Description:+ Provides a simple interface for writing concurrent programs that+ need to output a lot of status messages to the console, and/or run+ concurrent external commands that output to the console.+Extra-Source-Files:+  CHANGELOG++Library+  GHC-Options: -Wall -fno-warn-tabs -O2+  Build-Depends: base (>= 4.5), base < 5+    , bytestring (>= 0.10.0), bytestring (< 0.11)+    , async (>= 2.0), async (< 2.1)+    , stm (>= 2.0), stm (< 2.5)+    , process (>= 1.2.0), process (< 1.4.0)+    , unix (>= 2.7.0), unix (< 2.8.0)+    , directory (>= 1.2.0), directory (< 1.3.0)+    , containers (>= 0.5.0), containers (< 0.6.0)+    , transformers (>= 0.3.0), transformers (< 0.5.0)+    , exceptions (>= 0.8.0), exceptions (< 0.9.0)+    , MissingH (>= 1.3.0), MissingH (< 1.4.0)+  Exposed-Modules:+    Control.Concurrent.Output+  Other-Modules:+    Utility.Monad+    Utility.Data+    Utility.Exception+    Utility.FileSystemEncoding++source-repository head+  type: git+  location: git://git.joeyh.name/concurrent-output.git