diff --git a/BuildBox.hs b/BuildBox.hs --- a/BuildBox.hs +++ b/BuildBox.hs @@ -10,6 +10,7 @@ , module BuildBox.Command.Environment , module BuildBox.Command.File , module BuildBox.Command.Darcs + , module BuildBox.Control.Gang , module BuildBox.Cron , module BuildBox.FileFormat.BuildResults , module BuildBox.IO.Directory @@ -28,6 +29,7 @@ import BuildBox.Command.Environment import BuildBox.Command.File import BuildBox.Command.Darcs +import BuildBox.Control.Gang import BuildBox.Cron import BuildBox.FileFormat.BuildResults import BuildBox.IO.Directory diff --git a/BuildBox/Build.hs b/BuildBox/Build.hs --- a/BuildBox/Build.hs +++ b/BuildBox/Build.hs @@ -8,6 +8,7 @@ -- * Building , runBuild + , runBuildWithState , runBuildPrint , runBuildPrintWithState , successfully diff --git a/BuildBox/Build/Base.hs b/BuildBox/Build/Base.hs --- a/BuildBox/Build/Base.hs +++ b/BuildBox/Build/Base.hs @@ -26,7 +26,7 @@ evalStateT (runErrorT build) s --- | Like 'runBuild`, but Run a build command, reporting whether it succeeded to the console. +-- | Like 'runBuild`, but report whether it succeeded to the console. -- If it succeeded then return Just the result, else Nothing. runBuildPrint :: FilePath -> Build a -> IO (Maybe a) runBuildPrint scratchDir build @@ -35,7 +35,20 @@ runBuildPrintWithState s build --- | Like `runBuildPrintWithConfig` but also takes a `BuildConfig`. +-- | Like `runBuild` but also takes a `BuildState`. +runBuildWithState :: BuildState -> Build a -> IO (Maybe a) +runBuildWithState s build + = do result <- evalStateT (runErrorT build) s + case result of + Left err + -> do putStrLn $ render $ ppr err + return $ Nothing + + Right x + -> do return $ Just x + + +-- | Like `runBuildPrint` but also takes a `BuildState`. runBuildPrintWithState :: BuildState -> Build a -> IO (Maybe a) runBuildPrintWithState s build = do result <- evalStateT (runErrorT build) s diff --git a/BuildBox/Command/System.hs b/BuildBox/Command/System.hs --- a/BuildBox/Command/System.hs +++ b/BuildBox/Command/System.hs @@ -208,6 +208,8 @@ trace $ "systemTeeIO stderr: " ++ Log.toString logErr trace $ "systemTeeIO: All done" + hClose hOutRead + hClose hErrRead code `seq` logOut `seq` logErr `seq` return (code, logOut, logErr) diff --git a/BuildBox/Command/System/Internals.hs b/BuildBox/Command/System/Internals.hs --- a/BuildBox/Command/System/Internals.hs +++ b/BuildBox/Command/System/Internals.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE PatternGuards, BangPatterns #-} +{-# LANGUAGE PatternGuards, BangPatterns, NamedFieldPuns #-} module BuildBox.Command.System.Internals ( streamIn , streamOuts) @@ -7,26 +7,78 @@ import Control.Concurrent import Control.Concurrent.STM.TChan import Control.Monad.STM +import Control.Monad +import Foreign.Ptr +import Data.IORef +import Data.Char +import Data.Word import Data.ByteString.Char8 (ByteString) -import qualified Data.ByteString.Char8 as BS +import qualified Data.ByteString.Internal as BS +import qualified Data.ByteString.Char8 as BS +import GHC.IO.Handle.Internals +import GHC.IO.Handle.Types +import GHC.IO.Buffer +import GHC.IO.BufferedIO as Buffered +import Foreign.Marshal.Utils (copyBytes) + + -- | Continually read lines from a handle and write them to this channel. -- When the handle hits EOF then write `Nothing` to the channel. streamIn :: Handle -> TChan (Maybe ByteString) -> IO () streamIn !hRead !chan - = do eof <- hIsEOF hRead - if eof - then do - atomically $ writeTChan chan Nothing - return () - - else do - str <- BS.hGetLine hRead - atomically $ writeTChan chan (Just str) - streamIn hRead chan + = streamIn' False hRead chan +streamIn' + :: Bool -- ^ whether the last line ended in a newline character + -> Handle -- ^ handle to read lines from + -> TChan (Maybe ByteString) -- ^ channel to write lines to + -> IO () +streamIn' !gotNewLine !hRead !chan + = uponM (hIsEOF hRead) + -- On EOF, if the last line ended in a newline then write a final + -- empty ByteString to the channel to signify this. + (do when gotNewLine + $ atomically $ writeTChan chan (Just BS.empty) + atomically $ writeTChan chan Nothing + return ()) + + -- Block until there is a line available from the channel. + -- The line may or may not end in a newline character, depending + -- on whether there was a literal newline in the source file, + -- or the file was flushed. + (do str <- hGetLineNL hRead + + if BS.null str + then -- hmm. The file was supposedly non-empty, but we didn't get + -- any data. Let's just try again. + streamIn' gotNewLine hRead chan + + else do + -- Check whether we got an actual newline character on the + -- end of the string. + let hasNewLine + | BS.last str == '\n' = True + | otherwise = False + + -- For string ending in newline characters, we don't want to + -- push the newline to the consumer, but we need to remember + -- if we've seen one to handle the end-of-file condition properly. + let str' + | hasNewLine = BS.init str + | otherwise = str + + atomically $ writeTChan chan (Just str') + streamIn' hasNewLine hRead chan) + +uponM :: Monad m => m Bool -> m a -> m a -> m a +uponM c x y + = do b <- c + if b then x else y + + -- | Continually read lines from some channels and write them to handles. -- When all the channels return `Nothing` then we're done. -- When we're done, signal this fact on the semaphore. @@ -50,7 +102,7 @@ streamOuts' False prev [] = do threadDelay 100000 streamOuts' False [] prev - + -- try to read from the current chan. streamOuts' !active !prev (!x@(!chan, !mHandle, !qsem) : rest) = do @@ -80,4 +132,81 @@ | otherwise -> streamOuts' True (prev ++ [x]) rest + + +-- Code hacked out of Data.ByteString library. +-- | Like `hGetLine`, but return the newline charater on the end of the string, +-- if there was one. This allows us to distinguish between lines that were flushed +-- from the IO buffer due to a newline character, and partial lines that were +-- flushed manually, and don't have a newline. +hGetLineNL :: Handle -> IO ByteString +hGetLineNL h = + wantReadableHandle_ "BuildBox.Command.System.Internals" h $ + \ h_@Handle__{haByteBuffer} -> do + flushCharReadBuffer h_ + buf <- readIORef haByteBuffer + if isEmptyBuffer buf + then fill h_ buf 0 [] + else haveBuf h_ buf 0 [] + where + + fill h_@Handle__{haByteBuffer,haDevice} buf len xss = + len `seq` do + (r,buf') <- Buffered.fillReadBuffer haDevice buf + if r == 0 + then do writeIORef haByteBuffer buf{ bufR=0, bufL=0 } + if len > 0 + then mkBigPS len xss + else ioe_EOF + else haveBuf h_ buf' len xss + + haveBuf h_@Handle__{haByteBuffer} + buf@Buffer{ bufRaw=raw, bufR=w, bufL=r } + len xss = + do + off <- findEOL r w raw + let new_len = len + off - r + xs <- mkPS raw r off + + -- if eol == True, then off is the offset of the '\n' + -- otherwise off == w and the buffer is now empty. + if off /= w + then do + if (w == off + 1) + then writeIORef haByteBuffer buf{ bufL=0, bufR=0 } + else writeIORef haByteBuffer buf{ bufL = off + 1 } + + -- Produce the final string. + -- If r == w then we've found the end-of file, but there + -- was no newline character on the input. + if r == w + then mkBigPS new_len (xs:xss) + else mkBigPS new_len (BS.pack "\n" : xs : xss) + + else do + fill h_ buf{ bufL=0, bufR=0 } new_len (xs:xss) + + -- Try to find an end-of-line character in the buffer, if there is one. + -- If not we'll have r == w. + findEOL r w raw + | r == w = return w + | otherwise + = do + c <- readWord8Buf raw r + if c == fromIntegral (ord '\n') + then return r + else findEOL (r+1) w raw + + +mkPS :: RawBuffer Word8 -> Int -> Int -> IO ByteString +mkPS buf start end = + BS.create len $ \p -> + withRawBuffer buf $ \pbuf -> do + copyBytes p (pbuf `plusPtr` start) len + where + len = end - start + +mkBigPS :: Int -> [ByteString] -> IO ByteString +mkBigPS _ [ps] = return ps +mkBigPS _ pss = return $! BS.concat (reverse pss) diff --git a/BuildBox/Control/Gang.hs b/BuildBox/Control/Gang.hs new file mode 100644 --- /dev/null +++ b/BuildBox/Control/Gang.hs @@ -0,0 +1,211 @@ + +-- | A gang consisting of a fixed number of threads that can run actions in parallel. +-- Good for constructing parallel test frameworks. +module BuildBox.Control.Gang + ( Gang + , GangState(..) + + , forkGangActions + , joinGang + , pauseGang + , resumeGang + , flushGang + , killGang + + , getGangState + , waitForGangState) +where +import Control.Concurrent +import Data.IORef +import qualified Data.Set as Set +import Data.Set (Set) + +-- Gang ----------------------------------------------------------------------- +-- | Abstract gang of threads. +data Gang + = Gang + { gangThreads :: Int + , gangThreadsAvailable :: QSemN + , gangState :: IORef GangState + , gangActionsRunning :: IORef Int + , gangThreadsRunning :: IORef (Set ThreadId) } + + +data GangState + = -- | Gang is running and starting new actions. + GangRunning + + -- | Gang may be running already started actions, + -- but no new ones are being started. + | GangPaused + + -- | Gang is waiting for currently running actions to finish, + -- but not starting new ones. + | GangFlushing + + -- | Gang is finished, all the actions have completed. + | GangFinished + + -- | Gang was killed, all the threads are dead (or dying). + | GangKilled + deriving (Show, Eq) + + +-- | Get the state of a gang. +getGangState :: Gang -> IO GangState +getGangState gang + = readIORef (gangState gang) + + +-- | Block until all actions have finished executing, +-- or the gang is killed. +joinGang :: Gang -> IO () +joinGang gang + = do state <- readIORef (gangState gang) + if state == GangFinished || state == GangKilled + then return () + else do + threadDelay 10000 + joinGang gang + + +-- | Block until already started actions have completed, but don't start any more. +-- Gang state changes to `GangFlushing`. +flushGang :: Gang -> IO () +flushGang gang + = do writeIORef (gangState gang) GangFlushing + waitForGangState gang GangFinished + + +-- | Pause a gang. Actions that have already been started continue to run, +-- but no more will be started until a `resumeGang` command is issued. +-- Gang state changes to `GangPaused`. +pauseGang :: Gang -> IO () +pauseGang gang + = writeIORef (gangState gang) GangPaused + + +-- | Resume a paused gang, which allows it to continue starting new actions. +-- Gang state changes to `GangRunning`. +resumeGang :: Gang -> IO () +resumeGang gang + = writeIORef (gangState gang) GangRunning + + +-- | Kill all the threads in a gang. +-- Gang stage changes to `GangKilled`. +killGang :: Gang -> IO () +killGang gang + = do writeIORef (gangState gang) GangKilled + tids <- readIORef (gangThreadsRunning gang) + mapM_ killThread $ Set.toList tids + + +-- | Block until the gang is in the given state. +waitForGangState :: Gang -> GangState -> IO () +waitForGangState gang waitState + = do state <- readIORef (gangState gang) + if state == waitState + then return () + else do + threadDelay 10000 + waitForGangState gang waitState + + +-- | Fork a new gang to run the given actions. +-- This function returns immediately, with the gang executing in the background. +-- Gang state starts as `GangRunning` then transitions to `GangFinished`. +-- To block until all the actions are finished use `joinGang`. +forkGangActions + :: Int -- ^ Number of worker threads in the gang \/ maximum number + -- of actions to execute concurrenty. + -> [IO ()] -- ^ Actions to run. They are started in-order, but may finish + -- out-of-order depending on the run time of the individual action. + -> IO Gang + +forkGangActions threads actions + = do semThreads <- newQSemN threads + refState <- newIORef GangRunning + refActionsRunning <- newIORef 0 + refThreadsRunning <- newIORef (Set.empty) + let gang + = Gang + { gangThreads = threads + , gangThreadsAvailable = semThreads + , gangState = refState + , gangActionsRunning = refActionsRunning + , gangThreadsRunning = refThreadsRunning } + + _ <- forkIO $ gangLoop gang actions + return gang + + +-- | Run actions on a gang. +gangLoop :: Gang -> [IO ()] -> IO () +gangLoop gang [] + = do -- Wait for all the threads to finish. + waitQSemN + (gangThreadsAvailable gang) + (gangThreads gang) + + -- Signal that the gang is finished running actions. + writeIORef (gangState gang) GangFinished + + +gangLoop gang actions@(action:actionsRest) + = do state <- readIORef (gangState gang) + case state of + GangRunning + -> do -- Wait for a worker thread to become available. + waitQSemN (gangThreadsAvailable gang) 1 + gangLoop_withWorker gang action actionsRest + + GangPaused + -> do threadDelay 10000 + gangLoop gang actions + + GangFlushing + -> do actionsRunning <- readIORef (gangActionsRunning gang) + if actionsRunning == 0 + then writeIORef (gangState gang) GangFinished + else do + threadDelay 10000 + gangLoop gang [] + + GangFinished -> return () + GangKilled -> return () + +-- we have an available worker +gangLoop_withWorker :: Gang -> IO () -> [IO ()] -> IO () +gangLoop_withWorker gang action actionsRest + = do -- See if we're supposed to be starting actions or not. + state <- readIORef (gangState gang) + case state of + GangRunning + -> do -- fork off the first action + tid <- forkOS $ do + -- run the action (and wait for it to complete) + action + + -- signal that a new worker is available + signalQSemN (gangThreadsAvailable gang) 1 + + -- remove our ThreadId from the set of running ThreadIds. + tid <- myThreadId + atomicModifyIORef (gangThreadsRunning gang) + (\tids -> (Set.delete tid tids, ())) + + -- Add the ThreadId of the freshly forked thread to the set + -- of running ThreadIds. We'll need this set if we want to kill + -- the gang. + atomicModifyIORef (gangThreadsRunning gang) + (\tids -> (Set.insert tid tids, ())) + + -- handle the rest of the actions. + gangLoop gang actionsRest + + -- someone issued flush or pause command while we + -- were waiting for a worker, so don't start next action. + _ -> do + signalQSemN (gangThreadsAvailable gang) 1 + gangLoop gang (action:actionsRest) diff --git a/BuildBox/Data/Log.hs b/BuildBox/Data/Log.hs --- a/BuildBox/Data/Log.hs +++ b/BuildBox/Data/Log.hs @@ -39,9 +39,7 @@ toString ll = BS.unpack $ BS.intercalate (BS.pack "\n") - $ F.toList (ll Seq.>< Seq.singleton BS.empty) - -- hack to get the last "\n" on the end. - + $ F.toList ll -- | O(n) Convert a `String` to a `Log`. fromString :: String -> Log diff --git a/buildbox.cabal b/buildbox.cabal --- a/buildbox.cabal +++ b/buildbox.cabal @@ -1,5 +1,5 @@ Name: buildbox -Version: 1.5.1.1 +Version: 1.5.2.1 License: BSD3 License-file: LICENSE Author: Ben Lippmeier @@ -61,6 +61,7 @@ BuildBox.Command.Network BuildBox.Command.System BuildBox.Command.Sleep + BuildBox.Control.Gang BuildBox.Cron.Schedule BuildBox.Cron BuildBox.Data.Log