diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,30 @@ +Copyright (c) 2013, Ryan Newton + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + + * 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. + + * Neither the name of Ryan Newton nor the names of other + contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT +OWNER 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. diff --git a/Scripting/Parallel/ThreadPool.hs b/Scripting/Parallel/ThreadPool.hs new file mode 100644 --- /dev/null +++ b/Scripting/Parallel/ThreadPool.hs @@ -0,0 +1,67 @@ + +module Scripting.Parallel.ThreadPool + ( + -- * Helpers for parallel scripting, i.e. prepping input to `hydraPrint` + parForM + ) + where + +import Data.IORef +import Control.Monad (forM) +import Control.Concurrent +import Control.Concurrent.Chan +import qualified Control.Concurrent.Async as A +import Prelude as P + +import qualified System.IO.Streams as S +-- import System.IO.Streams (InputStream, OutputStream) +import System.IO.Streams.Concurrent (chanToInput, chanToOutput) + +-------------------------------------------------------------------------------- + + +-- | A helper for parallel scripting. Run work items in parallel on N worker threads +-- (a thread pool), creating only ONE output stream per worker, not one per parallel +-- task. Specifically, each invocation of the user's function is given an +-- OutputStream that it holds a "lock" on -- it is the only thread accessing that +-- output stream. +-- +-- This function returns immediately with +-- (1) a list of input streams that will carry results on the fly, as they are produced, and +-- (2) a barrier action that waits for all parallel work to be finished and yields the final results. +-- The first list is `numWorkers` long, and the second is `numTasks`. +-- +-- Additional contracts: +parForM :: Int -> [a] -> (S.OutputStream c -> a -> IO b) -> IO ([S.InputStream c], IO [b]) +parForM numWorkers inputs action = + do let numTasks = P.length inputs + answers <- sequence$ P.replicate numTasks newEmptyMVar + workLeft <- newIORef (P.zip inputs answers) + chans <- sequence $ P.replicate numWorkers newChan + resultStrms <- mapM chanToInput chans + outStrms <- mapM chanToOutput chans + ------------ + -- That didn't seem to work.... let's try a different way of creating an input/output pair. + ------------ + asyncs <- forM (zip outStrms [(0::Int)..]) $ \ (strm,idx) -> + A.async $ do +-- B.hPutStrLn System.IO.stderr (B.pack$ "INSIDE ASYNC, "++show idx++" of "++show (P.length outStrms)) + let pfloop = do -- Pop work off the queue: + x <- atomicModifyIORef workLeft + (\ls -> if P.null ls + then ([], Nothing) + else (P.tail ls, Just ((\ (h:_)->h) ls))) +-- B.hPutStrLn System.IO.stderr (B.pack$ "POPPED work "++show (idx, fmap (const ()) x)) + case x of + Nothing -> S.write Nothing strm -- End the stream. + Just (input,mv) -> + do result <- action strm input + putMVar mv result + pfloop + pfloop + let barrier = do + mapM_ A.wait asyncs -- For exception handling. + mapM readMVar answers + + return (resultStrms, barrier) + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/SystemTest.hs b/SystemTest.hs new file mode 100644 --- /dev/null +++ b/SystemTest.hs @@ -0,0 +1,66 @@ +{-# LANGUAGE CPP #-} +-- This isn't a unit test. + +module Main where + +import UI.HydraPrint +import qualified System.IO.Streams as S +import System.IO.Streams.Concurrent as CS +import Control.Concurrent +import Control.Concurrent.Chan +import Control.Monad (forM_) +import Data.ByteString.Char8 as B +import Data.IORef +import Prelude as P +import System.Environment (getArgs) + +--baseStrmRate = 1000 -- Seconds +-- baseStrmRate = 500 +-- baseStrmRate = 100 + +-- windowLife = 1000 +windowLife = 1500 + +main :: IO () +main = do + args <- getArgs + let baseStrmRate = + case args of + [b] -> read b + _ -> 100 + + c1 <- newIORef 1 + s1 <- S.makeInputStream $ do + c <- readIORef (c1 :: IORef Int) + threadDelay$ 2000 * baseStrmRate + writeIORef c1 (c+1) + return$ Just$ B.pack$ show (100 * c) ++"\n" + + strmChan <- newChan + writeChan strmChan (Just ("Strm-0",s1)) + + -- Add a new stream every 3 seconds, they each expire after 5. + ---------------------------------------- + forkIO $ do + forM_ [1.. 3] $ \ix -> do + dbgLogLn "[strmsrc] Waiting before spawning additional strm..." + threadDelay$ 3000 * windowLife + c2 <- newIORef 1 + sN <- S.makeInputStream $ do + c <- readIORef (c2 :: IORef Int) + threadDelay$ 1000 * baseStrmRate + writeIORef c2 (c+1) + if c >= (8 * windowLife) `quot` baseStrmRate + then return Nothing + else return$ Just$ B.pack$ show ix ++":"++ show (10 * c) ++"\n" + dbgLogLn$"[strmsrc] Spawning additional stream! "++show ix + writeChan strmChan (Just ("strm-"++show ix, sN)) + dbgLogLn "[strmsrc] Waiting before cutting off stream source..." + threadDelay$ 8 * 1000 * windowLife + dbgLogLn "[strmsrc] Shutting off stream source!" + writeChan strmChan Nothing + dbgLogLn "[strmsrc] All done, exiting thread." + ---------------------------------------- + + strmSrc <- CS.chanToInput strmChan + hydraPrint defaultHydraConf strmSrc diff --git a/TestFramework.hs b/TestFramework.hs new file mode 100644 --- /dev/null +++ b/TestFramework.hs @@ -0,0 +1,8 @@ + + +module Main where + +import Test.Framework (defaultMain) +import UI.HydraPrint (testSuite) + +main = defaultMain [testSuite] diff --git a/UI/HydraPrint.hs b/UI/HydraPrint.hs new file mode 100644 --- /dev/null +++ b/UI/HydraPrint.hs @@ -0,0 +1,974 @@ +{-# LANGUAGE OverloadedStrings, NamedFieldPuns #-} +{-# LANGUAGE CPP #-} + +#define UPDATE_ALL_ALWAYS + +#ifndef NOTESTING +{-# LANGUAGE TemplateHaskell #-} +#endif + +-- | A simple utility to multiplex of *dynamic* collection of text streams. As the +-- number of streams varies, the multiplexing of the terminal output does too. +module UI.HydraPrint + ( + -- * hydraPrint and friends + hydraPrint, hydraPrintStatic, + HydraConf(..), defaultHydraConf, DeleteWinWhen(..) + + -- * Types + + -- * Tiling behavior + -- computeTiling, applyTiling + +#ifndef NOTESTING + -- * Testing + , testSuite +#endif + + -- TEMPORARY: + , dbgLogLn + ) + where + +import Data.IORef +import Data.Time.Clock +import Data.Word +import Data.Char (ord) +import Data.Map as M +import Data.List as L +import qualified Data.ByteString.Char8 as B +import Data.ByteString.Char8 (ByteString) +import Data.ByteString.Unsafe (unsafeUseAsCStringLen) +import Prelude as P hiding (unzip4) +import Control.Monad +import Control.Concurrent (threadDelay) +import Control.Monad.IO.Class (liftIO, MonadIO) +import Control.Exception +import Foreign.C.String (withCAStringLen) + +import qualified System.IO.Streams as S +import System.IO.Streams (InputStream, OutputStream) +import System.IO.Streams.Concurrent (concurrentMerge) + + +#if 0 +import UI.HSCurses.CursesHelper as CH +import UI.HSCurses.Curses as C hiding (s1,s3,tl,ls) +#else +import UI.NCurses hiding (Event) +import qualified UI.NCurses as C +#endif + +-- import Control.Monad.State +-- import Control.Monad.Reader + +import System.IO (hFlush, hPutStrLn, stderr, openFile, IOMode(WriteMode), Handle) +import System.IO.Unsafe (unsafePerformIO) +import System.IO.Error (isDoesNotExistError) +import System.Directory (removeFile) +import System.Environment (getEnvironment) + +import Control.Applicative +import qualified Data.Foldable as F +import qualified Data.List.NonEmpty as NE +import Data.List.NonEmpty (NonEmpty((:|))) + +#ifndef NOTESTING +import Test.QuickCheck hiding (NonEmpty) +import Test.HUnit (Assertion, assertEqual, assertBool) +import Test.Framework (Test, defaultMain) +import Test.Framework.Providers.HUnit (testCase) +import Test.Framework.Providers.QuickCheck2 (testProperty) +import Test.Framework.TH (testGroupGenerator) +#endif + +dbg :: Bool +dbg = case P.lookup "HYDRA_DEBUG" theEnv of + Nothing -> False + Just "" -> False + Just "0" -> False + Just "False" -> False + Just _ -> True + +theEnv :: [(String, String)] +theEnv = unsafePerformIO$ getEnvironment + +io :: MonadIO m => IO a -> m a +io x = liftIO x + +-------------------------------------------------------------------------------- +-- User visible configuration +-------------------------------------------------------------------------------- + +data HydraConf = + HydraConf + { +-- majorMode :: -- | Interleaved, windows, or serialized. + deleteWhen :: DeleteWinWhen, + useColor :: Bool -- Use colors if they are supported. + + -- TODO: Introduce environment variables: + -- HYDRA_COLOR + -- HYDRA_DELETEWHEN + -- ... + } + +-- | How long should we wait after a stream goes dry to close the window associated +-- with it? If 'Never' is selected, then the window will stay until a new stream +-- causes the screen to reconfigure, or hydraPrint exits. +data DeleteWinWhen = Never + | After Seconds + | Immediately + +type Seconds = Double + +-- | Like the name says, a default set of options for passing to `hydraPrint`. +defaultHydraConf :: HydraConf +defaultHydraConf = + HydraConf + { + deleteWhen = After 3.0, + useColor = True + } + +-------------------------------------------------------------------------------- +-- Internal Types +-------------------------------------------------------------------------------- + + +-- | The state of an active `hydraPrint` computation. +data MPState = + MPState + { + activeStrms :: M.Map StreamID WindowWidget, + + -- | Windows whose streams have ended but are not yet removed (deleteWhen). + -- Each entry stores its timeout. + dyingStrms :: [(StreamID, Seconds, WindowWidget)], + + -- | Streams that are gone and have no widget. + deadStrms :: [StreamHistory], + + -- | All active windows. Need to be explicitly deleted. + windows :: [CWindow], + + -- | All color IDs, cycle through these for new windows. + colorIDs :: [ColorID] + -- Log: TODO: could log stream create/delete events and their times. + } + +-- | All the state for a widget, that persists beyond the creation and destruction of +-- individual NCurses windows (and even the reinitialization of the whole system). +-- +-- These text widgets currently do NOT support line wrapping. They crop to the right +-- and at the bottom. +data WindowWidget = + WindowWidget + { + -- "Methods" + ---------------------------------------- + -- | The current and previous text in the widget. + hist :: StreamHistory, + -- | Get the current size of the writable area. + textSizeYX :: IO (Word,Word), + + -- | Replace a line within the window, clearing the rest of the line if the + -- string is too short, and cropping it if it is too long. The `Word` argument + -- is a zero-based index into the writable area of the window. Drawing off the + -- end of the window will be ignored. + putLine :: ByteString -> Curses (), + -- putLineN :: Word -> String -> IO () + + -- | Just redraw it. + repaint :: Curses (), + + setWin :: CWindow -> Curses (), + + -- "Private" state: + ---------------------------------------- + winRef :: IORef CWindow + } + +-- | The history of a stream. The view changes, but the underlying stream histories +-- persist. +data StreamHistory = + StreamHistory { + -- | The name of a stream might identify a client hostname, or a subprogram run, + -- or a file being compiled. These are displayed so as to help distinguish + -- different windows from one another, especially as the layout changes. + streamName :: String, + -- | A (reverse) list of lines with the most recently produced at the head. + revHist :: IORef [ByteString] + } + +-- | Most of the computation for this module happens in the context of a global, +-- mutable state. +-- type MP a = StateT (IORef MPState) IO a + +-- | Position of a window: (Height,Width, PosY, PosX) +-- The same order as accepted by `newWin`. +type WinPos = (Word,Word,Word,Word) + +-- | Along with the raw pointer, remember the size at which a window was created: +data CWindow = CWindow C.Window WinPos (String,ColorID) + deriving Show + +-------------------------------------------------------------------------------- + +-- These are all the basic colors currently exported by UI.NCurses, except black: +-- We make sure the basic, white-on-black option is first. +allColors :: [Color] +--allColors = [ColorWhite, ColorGreen, ColorCyan, ColorMagenta, ColorYellow, ColorRed, ColorBlue] +allColors = [ColorGreen, ColorCyan, ColorMagenta, ColorYellow, ColorRed, ColorBlue] + +-- TODO: Pair attributes with them: +-- AttributeStandout +-- AttributeUnderline +-- AttributeReverse +-- AttributeBlink +-- AttributeDim +-- AttributeBold +-- AttributeAltCharset +-- AttributeInvisible +-- AttributeProtect + +-- Return a finite list of color IDs, we rotate through these. +initColors :: Curses [ColorID] +initColors = do + supports <- supportsColor + if supports then do + cdc <- canDefineColor + mx <- maxColorID + -- Here we make the backgrounds actually black instead of grey: + when cdc $ defineColor ColorBlack 0 0 0 + ls <- mapM (\(idx, x) -> newColorID x ColorBlack idx) + (zip [1..mx] allColors) + return$ defaultColorID : ls + else + return [defaultColorID] + +-- | Create a new batch of NCurses windows (deleting the old ones) and display the +-- current state of a set of stream histories. +createWindows :: [(String,ColorID)] -> Word -> Curses ([CWindow],Word,Word) +createWindows names num = do + (curY,curX) <- screenSize + let (nX,nY) = computeTiling num + panelDims = applyTiling (i2w curY, i2w curX) (nY,nX) + ws <- forM (P.zip names (NE.toList panelDims)) $ + \ ((name,colorID), tup@(hght,wid, posY, posX)) -> do + w1 <- newWindow (w2i hght) (w2i wid) (w2i posY) (w2i posX) + + let msg = ("CreatedWindow: at "++show tup++", name "++name) + -- when dbg $ do dbgLogLn msg + -- moveCursor 1 2 + -- drawString msg + -- drawBox Nothing Nothing + let cwin = CWindow w1 tup (name,colorID) +#ifndef UPDATE_ALL_ALWAYS + updateWindow w1$ drawNamedBorder cwin +#endif + return cwin + return (ws,nX,nY) + +-- For blanking out inactive windows. +blankChar :: Char +blankChar = ' ' + +{- +-- | Use the simple method of writing blanks to clear. Convention: overwrite the +-- lower & right borders, but not the top/left. +clearWindow :: CWindow -> IO () +clearWindow (CWindow wp (hght,wid,_,_)) = do + let + width' = wid - borderLeft -- - borderRight + blank = P.replicate (w2i width') blankChar + forM_ [borderTop .. hght - borderBottom - 1 ] $ \ yind -> do +-- forM_ [borderTop .. hght - 1] $ \ yind -> do +-- forM_ [hght - 1 .. hght - 1] $ \ yind -> do + wMove wp (w2i yind) (w2i borderLeft) + wAddStr wp blank +-- blit wp blank + writeToCorner wp (w2i$ hght-1) (w2i borderLeft) blank + wnoutRefresh wp +-} + +-- Nah, this won't do it... odd that there's no clear or fill function? + +#if 0 +clearWindow :: CWindow -> Curses () +clearWindow (CWindow wp _) = do + updateWindow wp $ + setBackground (Glyph ' ' []) +#else +clearWindow :: CWindow -> Curses () +clearWindow (CWindow wp (hght,wid,_,_) _) = updateWindow wp $ do + let + width' = wid - borderLeft -- - borderRight + blank = P.replicate (w2i width') blankChar + io$ evaluate hght + io$ evaluate wid + io$ evaluate wp + forM_ [borderTop .. hght - borderBottom - 1 ] $ \ yind -> do + moveCursor (w2i yind) (w2i borderLeft) + drawString blank + return () + writeToCorner (w2i$ hght-1) (w2i borderLeft) blank +-- wnoutRefresh wp +#endif + +-- | Write out a string that goes all the way to the bottom/right corner. +writeToCorner :: Int -> Int -> String -> Update () +writeToCorner y x str = do + let len = P.length str + moveCursor (fromIntegral y) (fromIntegral x) + drawString (P.init str) + -- Uh oh! 'ncurses' doesn't expose winsch either. SKIP IT for now: +-- moveCursor y (len-1) +-- throwIfErr_ "winsch" $ winsch wp (fromIntegral$ ord$ P.last str) + return () + + +{- +-- | Write out a string that goes all the way to the bottom/right corner. +writeToCorner :: Window -> Int -> Int -> String -> IO () +writeToCorner wp y x str = do + -- I'm getting a Curses error if I try to write the lower-right corner character!?9 + let len = P.length str + wMove wp y x + wAddStr wp (P.init str) + wMove wp y (len-1) + ------------ + -- Hack: Even waddch directly is throwing an error in the lower right + -- corner... Hmm. A hack is to ignore the error code. + --throwIfErr_ "waddch" $ + -- waddch wp (fromIntegral$ ord blankChar) -- Don't advance the cursor. + ------------ + -- This seems to be a known issue: + -- http://lists.gnu.org/archive/html/bug-ncurses/2007-09/msg00002.html + throwIfErr_ "winsch" $ winsch wp (fromIntegral$ ord blankChar) + return () + +wAddCh :: Window -> Char -> IO () +wAddCh wp ch = throwIfErr_ "waddch" $ waddch wp (fromIntegral$ ord ch) + +wInsCh :: Window -> Char -> IO () +wInsCh wp ch = throwIfErr_ "winsch" $ winsch wp (fromIntegral$ ord ch) + +blit :: Window -> String -> IO () +blit wp s = + -- Ignore all non-ascii at the moment: + withCAStringLen s $ \ (s',len) -> + throwIfErr_ "waddchnstr" $ waddchnstr wp s' (fromIntegral len) + +blitB :: Window -> ByteString -> IO () +blitB wp s = + -- Ignore all non-ascii at the moment: +-- B.useAsCStringLen s $ \ (s',len) -> + unsafeUseAsCStringLen s $ \ (s',len) -> + throwIfErr_ "waddchnstr" $ waddchnstr wp s' (fromIntegral len) + +-- This SHOULDNT be necessary, but I'm having problems with blanking and blinking +-- otherwise. +redrawAll :: [CWindow] -> IO () +redrawAll wins = do +-- forM_ wins $ \ (CWindow wp _) -> do +-- wBorder wp defaultBorder +-- wRefresh wp -- TODO: use wnoutrefresh instead +-- wnoutRefresh wp + C.update +-} +redrawAll :: [CWindow] -> Curses () +redrawAll wins = do + forM_ wins $ \ cwin@(CWindow wp _ _) -> do +#ifdef UPDATE_ALL_ALWAYS + updateWindow wp $ do + -- HACK: This shouldn't be necessary, but I have problems with windows + -- appearing and then disappearing: + -------------------- +-- moveCursor 1 1 +-- drawString " " + drawNamedBorder cwin + -- drawBox Nothing Nothing + -------------------- + return () +#endif + return () + C.render + + +-- How many characters to avoid at the edges of window, for the border: +borderTop :: Word +borderTop = if dbg then 2 else 1 +-- borderTop = 2 +borderBottom :: Word +borderBottom = 1 +borderLeft :: Word +borderLeft = 1 +borderRight :: Word +borderRight = 1 + +-- | Create a new, persistent scrolling text widget. +createWindowWidget :: String -> IO WindowWidget +createWindowWidget streamName = do -- ioStrm + revHist <- newIORef [] + winRef <- newIORef (error "winRef field uninialized. Call setWin.") + let hist = StreamHistory{streamName, revHist} + putLine bstr = do + oldhist <- io$ readIORef revHist + let msg = bstr -- `B.append` (B.pack (" ")) + let newhist = msg : oldhist + io$ writeIORef revHist newhist + repaint + + -- Redraw all text AND the border: + repaint = do + cwin@(CWindow wp (y,x,_,_) _) <- io$ readIORef winRef + updateWindow wp $ do + newhist <- io$ readIORef revHist + let y' = y - borderTop - borderBottom + shown = P.take (w2i y') newhist + padY = y' - i2w(P.length shown) + forM_ (P.zip [1..] shown) $ \ (ind,oneline) -> do + moveCursor (w2i (y - borderBottom - ind - padY)) (w2i borderLeft) + let padded = oneline `B.append` + B.replicate (w2i x - B.length oneline) ' ' + cropped = B.take (w2i (x - borderLeft - borderRight)) padded + drawString (B.unpack cropped) + ------ Line is put! ---- + drawNamedBorder cwin + + textSizeYX = do + CWindow _ (y,x,_,_) _ <- readIORef winRef + return (y,x) + + setWin cwin@(CWindow wp _ _) = do + io$ writeIORef winRef cwin + updateWindow wp $ + drawNamedBorder cwin + return () + + obj = WindowWidget { hist, textSizeYX, putLine, + setWin, winRef, repaint } + return obj + +drawNamedBorder :: CWindow -> Update () +drawNamedBorder (CWindow wp (hght,wid,y,_) (name,winColor)) = do +-- wBorder wp defaultBorder + setColor defaultColorID + drawBox Nothing Nothing + setColor winColor + let isTop = (y == 0) + -- name' = llCorner : name ++ [lrCorner] + name' = "[" ++ name ++ "]" + mid = wid `quot` 2 + strt = max 0 (w2i mid - (fromIntegral (P.length name' `quot` 2))) + if isTop then + moveCursor 0 strt + else + moveCursor (w2i$ hght-1) strt + drawString (take (w2i wid) name') + +dbgLn :: String -> IO () +dbgLn s = when dbg$ + do dbgLogLn s + P.putStrLn s + +dbgLogLn :: String -> IO () +dbgLogLn s = when dbg$ + do B.hPutStrLn dbgLog (B.pack s) + hFlush dbgLog + +dbgLog :: Handle +dbgLog = unsafePerformIO $ do + let file = "/tmp/hydraprint_debug.log" + removeIfExists file + openFile file WriteMode + +-------------------------------------------------------------------------------- + +-- | This simply takes the names of the desired initial windows, and sets things up. +initAndRunCurses :: HydraConf -> [String] -> (MPState -> Curses a) -> IO a +initAndRunCurses HydraConf{useColor} names action = runCurses $ do + setCursorMode CursorInvisible + cids <- if useColor then initColors + else return [defaultColorID] + -- _ <- leaveOk True + wids <- forM names $ \ sname -> + io$ createWindowWidget sname + (wins,_,_) <- createWindows (zip names cids) (i2w$ P.length names) + sequence$ zipWith setWin wids wins + action$ MPState { activeStrms= M.fromList (zip [0..] wids), + dyingStrms= [], + deadStrms= [], + windows = wins, + colorIDs= cids -- tail cids ++ [head cids] + } + +-------------------------------------------------------------------------------- + + + + +-------------------------------------------------------------------------------- + + +-- | Take a fixed list of input streams. This variant preemptively splits the screen +-- into exactly one panel per stream. +hydraPrintStatic :: HydraConf -> [(String, InputStream ByteString)] -> IO () +hydraPrintStatic _ [] = return () +hydraPrintStatic conf srcs = do + -- Because of how steadyStat is structured, we need to peel off the LAST stream: + let (nameL,strmL) = P.last srcs + (names,strms) = unzip (P.init srcs) + strms' <- sequence$ zipWith preProcess [0..] strms + merged <- concurrentMerge strms' + -- We set up all but the LAST stream, and then go into steady state: + initAndRunCurses conf names $ \ initMPS -> do + steadyState conf{deleteWhen=Never} initMPS (i2w$ P.length names) (nameL,strmL) merged +-- steadyState conf initSt 1 (s2name,s2) merge2 + +-------------------------------------------------------------------------------- + +-- | Takes a /source/ of input streams, which may be added dynamically. A stream +-- that joins dynamically, exits once it issues an end-of-stream. +-- +-- `hydraPrint` is a blocking call that doesn't return until ALL streams that +-- appear produce an end-of-stream, AND the stream-source itself reaches +-- end-of-stream. +hydraPrint :: HydraConf -> InputStream (String, InputStream ByteString) -> IO () +hydraPrint conf strmSrc = phase0 conf =<< S.map NewStream strmSrc +----------------------------------------PHASE0---------------------------------------- +-- Nothing to do before there is at least ONE stream... +phase0 :: HydraConf -> InputStream Event -> IO () +phase0 conf strmSrc' = do + dbgLn $ "phase0: blocking for event." + ms1 <- S.read strmSrc' + case ms1 of + Nothing -> do + dbgLn $ "phase0: stream ended" + return () + Just (NewStream (s1name,s1)) -> do + dbgLn $ "phase0: new (first) stream! ("++s1name++") Moving to phase1." + s1' <- preProcess 0 s1 + -- Next, we need a "select/epoll". We use concurrentMerge. + merge1 <- concurrentMerge [strmSrc', s1'] + phase1 conf s1name merge1 + _ -> error "hydraPrint: Internal error. Unexpected event." +----------------------------------------PHASE1---------------------------------------- +-- Initially, we start in "cooked" (non-ncurses) mode, and stay there as long as +-- there is only one output stream. +phase1 :: HydraConf -> String -> InputStream Event -> IO () +phase1 conf s1name merge1 = do + dbgLn $ "phase1: blocking for event." + nxt <- S.read merge1 + case nxt of + Nothing -> do + dbgLn $ "Streams ended in phase1!" + return () + Just (NewStrLine _ (StrmElt ln)) -> do + B.putStrLn ln + phase1 conf s1name merge1 + Just (NewStrLine sid EOS) -> do + dbgLn $ "Got stream EOS! ID "++show sid + phase0 conf merge1 + Just (NewStream (s2name,s2)) -> do + dbgLn $ "Got newStream! "++s2name++". Transition to steady state..." -- (press enter) + ---------------------- + -- PROBLEM: We don't want to do a nested runCurses here... + -- cursesEvts <- io$ S.makeInputStream $ fmap (Just . CursesKeyEvent) + -- (C.getEvent defaultWindow Nothing) + -- Warning, because the curses events go into a concurrentMerge, they will keep + -- being read into Haskell, irrespective of what this "main" thread does. +-- merge2 <- io$ concurrentMerge [merge1, cursesEvts] + heartbeat <- timer 200 -- 200ms heartbeat + merge2 <- concurrentMerge [ merge1, heartbeat ] + ---------------------- + initAndRunCurses conf [s1name] $ \ initMPS -> + -- The first stream as ID 0, so this next one has ID 1: + steadyState conf initMPS 1 (s2name,s2) merge2 + Just (CursesKeyEvent _) -> error "Internal error. Shouldn't see Curses event here." + +----------------------------------------PHASE3---------------------------------------- +-- Re-enter this loop every time there is a new stream. +steadyState :: HydraConf -> MPState -> StreamID -> (String,InputStream ByteString) -> InputStream Event -> Curses () +steadyState conf state0@MPState{activeStrms,windows} sidCnt (newName,newStrm) merged = do + -- First, deal with the new stream. + newStrm' <- io$ preProcess sidCnt newStrm + merged' <- io$ concurrentMerge [merged, newStrm'] + widg <- io$ createWindowWidget newName + let active2 = M.insert sidCnt widg activeStrms + windows2 <- reCreate active2 windows + let state1 = state0{activeStrms=active2, windows=windows2} + -- redraw is next, as soon as we call loop: + + -- Second, enter an event loop: + let loop mps@MPState{activeStrms, dyingStrms, deadStrms, windows} = do +-- TEMP: +-- mapM_ repaint (M.elems activeStrms) + redrawAll windows + + nxt <- io$ S.read merged' + case nxt of + Nothing -> return () + Just HeartBeat-> do + -----------------Poll dying windows------------------- + let deadLp mps' [] = loop mps' -- mps{dyingStrms=remain} + deadLp mps' ((sid,timeOut,widg):tl) = do + now <- secsToday + -- io $ dbgPrnt $ "Checking dying time "++show timeOut++" against now: "++show now + if now >= timeOut + -- TODO: Remove duplicate code here: + then do let MPState{activeStrms,dyingStrms,deadStrms} = mps' + let active' = M.delete sid activeStrms + case P.reverse windows of + [] -> error "hydraPrint: Internal error. Expecting list of windows to be non-empty." + lst:_ -> clearWindow lst + windows' <- reCreate active' windows + deadLp mps'{activeStrms=active', + dyingStrms= P.filter (\(a,_,_) -> a /= sid) dyingStrms, + deadStrms = hist widg : deadStrms, + windows = windows' + } + tl + else deadLp mps' tl + -- Call this below to actually do the poll. + pollAndContinue mps = deadLp mps dyingStrms + + -------------------Poll key event------------------- + win <- defaultWindow + let keyLoop hit mps = do + mevt <- getEvent win (Just 0) + case mevt of + Nothing -> do when hit $ do + mapM_ repaint (M.elems activeStrms) + return () + pollAndContinue mps + Just evt -> do + io$ dbgPrnt$ " [dbg] Got curses event: "++show evt + case evt of + EventResized -> do +-- windows' <- reCreate activeStrms windows +-- keyLoop hit mps{windows=windows'} + keyLoop hit mps + EventCharacter 'q' -> return () + _ -> do +-- mapM_ repaint (M.elems activeStrms) +-- C.render + keyLoop True mps + keyLoop False mps +-- hWaitForInput stdin (1000) + + Just (NewStrLine sid (StrmElt ln)) -> do + io$ dbgLogLn (B.unpack ln) + putLine (activeStrms!sid) ln + loop mps + Just (NewStrLine sid EOS) -> do + io$ dbgPrnt $ " [dbg] Stream ID "++ show sid++" got end-of-stream " + case deleteWhen conf of + Never -> loop mps -- Don't change *anything*.. + After secs -> do + now <- secsToday + let dyingStrms' = (sid, secs + now, activeStrms!sid) : dyingStrms + -- FIXME: don't use winRef: + (CWindow wp _ _) <- io$ readIORef (winRef (activeStrms!sid)) + updateWindow wp $ return () + + loop mps{ dyingStrms=dyingStrms' } + Immediately -> do + let active' = M.delete sid activeStrms + + -- Deleting always shifts down the LAST window (should improve this) + case P.reverse windows of + [] -> error "hydraPrint: Internal error. Expecting list of windows to be non-empty." + lst:_ -> clearWindow lst + windows' <- reCreate active' windows + loop mps{ activeStrms = active', + deadStrms= hist (activeStrms!sid) : deadStrms, + windows = windows' } + Just (NewStream (s2name,s2)) -> do + io$ dbgPrnt $ " [dbg] NEW stream: "++ show s2name + steadyState conf mps (sidCnt+1) (s2name,s2) merged' + Just (CursesKeyEvent key) -> do + case key of +{- + KeyChar 'q' -> do + CH.end + dbgLn " [dbg] NCurses finished." + KeyChar 'p' -> do + -- Pause until another key is hit. + _ <- C.getCh + loop mps + KeyResize -> do + C.endWin + C.update + windows' <- reCreate activeStrms windows +-- redrawAll windows' + loop mps{windows=windows'} +-} + _ -> do io$ dbgPrnt $ " [dbg] CURSES Key event: "++show key + loop mps + loop state1 + where + dbgPrnt s = when dbg $ do + dbgLogLn s + -- AGAIN, a problem with the Curses monad here... we want to call this before curses is initialized: + -- putLine (P.head$ M.elems activeStrms) (B.pack s) + -- redrawAll windows + + reCreate active' oldWins = do + let names = P.map (streamName . hist) $ M.elems active' + numactive = fromIntegral (M.size active') + withcolors = P.zip names (P.cycle (colorIDs state0)) + (ws,numHoriz,numVert) <- createWindows withcolors numactive + -- Guaranteed to be in ascending key order, which in our case is + -- first-stream-to-join first. + forM_ (P.zip ws (M.assocs active')) $ \ (win,(sid,wid)) -> do + setWin wid win + -- Actually delete the old windows: +-- forM_ oldWins (\ (CWindow w _) -> delWin w) + forM_ oldWins (\ (CWindow w _ _) -> closeWindow w) + -- io$ dbgPrnt$ " [dbg] Deleted windows: "++show (P.map (\ (CWindow w _) -> w) oldWins) + -- ++ " created "++ show(P.map (\ (CWindow w _) -> w) ws) + ---------------------------------------- + -- Erase the bit of border which may be unused: + (nLines,nCols) <- screenSize + dummies <- case reverse ws of + [] -> return [] + (CWindow wp (hght,wid,y,x) _ : _) -> + let lastCol = w2i$ x + wid - 1 in + if (lastCol < nCols - 1) then do + ----------- First wipe the horizontal lower border: + let startX = lastCol+1 + remainingX = fromIntegral$ nCols - startX + dummy <- newWindow 1 remainingX (w2i$ y+hght-1) startX + let dummyCW = CWindow dummy (1, i2w remainingX, (w2i$ y+hght-1), i2w startX) ("",defaultColorID) + -- wclear dummy; wnoutRefresh dummy + io$ dbgPrnt$ "Dummy horiz (screen "++show (nLines,nCols)++"): "++show dummyCW +-- clearWindow dummyCW + ----------- Then the vertical right border: + let startY = (w2i$ y+1) + remainingY = fromIntegral$ nLines - startY - 1 + dummy2 <- newWindow remainingY 1 startY (nCols-1) + let dummy2CW = CWindow dummy2 (i2w remainingY, 1, i2w startY, i2w (nCols-1)) ("",defaultColorID) + io$ dbgPrnt$ "Dummy vert: "++show dummy2CW + -- wclear dummy2; wnoutRefresh dummy2 +-- clearWindow dummy2CW + return [dummy,dummy2] + else return [] + return ws + +-- Helper: import a bytestring into our system. +preProcess :: StreamID -> InputStream ByteString -> IO (InputStream Event) +preProcess id s = do + s' <- S.lines s + s'' <- liftStream s' + S.map (NewStrLine id) s'' + + +type StreamID = Word + + +-- | There are three relevant kinds of events inside our inner loop. +data Event = NewStream (String, InputStream ByteString) + | NewStrLine {-# UNPACK #-} !StreamID (Lifted ByteString) + | CursesKeyEvent Key + | HeartBeat +-- deriving (Show,Eq,Read,Ord) + +instance Show Event where + show (NewStream (s,_)) = "NewStream "++s + show (NewStrLine sid str) = "NewStrLine "++show sid++" "++show str + show (CursesKeyEvent k) = "CursesKeyEvent "++ show k + +-------------------------------------------------------------------------------- +-- Tiling behavior +-------------------------------------------------------------------------------- + +-- | If at least `n` windows are required, this computes the x-by-y tiling such that +-- `x * y >= n`. It returns `(x,y)` where `x` represents the number of horizontal +-- tiles and `y` the number of vertical. +computeTiling :: Word -> (Word,Word) +computeTiling reqWins = + if (n' - 1) * n' >= reqWins + then (n' - 1, n') + else (n', n') + where + n :: Double + n = sqrt (fromIntegral reqWins) + n' = ceiling n + +-- | Split a space into a given X-by-Y tile arrangement, leaving room for borders. +applyTiling :: (Word, Word) -> (Word, Word) -> NonEmpty WinPos +applyTiling _ a2@(splitsY,splitsX) + | splitsX < 1 || splitsY < 1 = + error$"applyTiling: cannot split ZERO ways in either dimension: "++show(a2) +applyTiling (screenY,screenX) (splitsY,splitsX) = NE.fromList$ + [ (height,width, yStrt, xStrt) + | (yStrt,height) <- doDim screenY splitsY + , (xStrt,width) <- doDim screenX splitsX ] + where + -- This is used both for horizontal and vertical, but I use horizontal + -- terminology below: + doDim :: Word -> Word -> [(Word,Word)] + doDim screen splits = P.zip starts widths' + -- Every window must "pay" for its left border, the rightmost border is paid for + -- globally, hence the minus-one here: + where + -- Every window must "pay" for its left border, the rightmost border is paid for + -- globally, hence the minus-one here: + usable = screen - 1 + (each,left) = usable `quotRem` splits + -- Here we distribute the remainder as evenly as possible: + widths = let (hd,tl) = L.splitAt (fromIntegral left) + (L.replicate (w2i splits) each) in + (L.map (+1) hd) ++ tl + -- Starting positions are based on the raw widths not counting overlap + starts = L.init$ L.scanl (+) 0 widths + -- Final widths get bumped to include their rightmost border: + widths' = L.map (+1) widths + +-------------------------------------------------------------------------------- + +test :: IO () +test = do + -- Weird, what will happen: +-- inlines <- S.lines S.stdin + s1 <- S.fromList ["hi","there","you","blah"] + s2 <- S.fromList ["aaa","bbb","ccc"] + + Just x <- S.read s1 + Just y <- S.read s1 + P.putStrLn$"READ FIRST INPUTs: "++show (x,y) + S.unRead y s1 + S.unRead x s1 + _ <- P.getLine + + hydraPrint defaultHydraConf =<< S.fromList [("s1",s1),("s2",s2)] + +-------------------------------------------------------------------------------- +-- Missing bits that should be elsewhere: +-------------------------------------------------------------------------------- + +-- | 'unzip4' for `NonEmpty` lists +unzip4 :: Functor f => f (a,b,c,d) -> (f a, f b, f c, f d) +unzip4 xs = ((\(x,_,_,_) -> x) <$> xs, + (\(_,x,_,_) -> x) <$> xs, + (\(_,_,x,_) -> x) <$> xs, + (\(_,_,_,x) -> x) <$> xs) + + +-- | Annoyingly, many libraries use Int where negative values are not allowed. +i2w :: (Show n, Integral n) => n -> Word +i2w i | i < 0 = error$"i2w: Cannot convert negative Int to Word: "++show i +i2w i = fromIntegral i + +w2i :: (Show n, Integral n) => Word -> n +w2i w = if i < 0 + then error$"w2i: Cannot convert Word to Int: "++show w + else i + where + i = fromIntegral w + + +timer :: Int -> IO (S.InputStream Event) +timer milles = S.makeInputStream g + where + g = do -- putStrLn "TIMER" + threadDelay$ milles * 1000 + return (Just HeartBeat) + + +secsToday :: Curses Double +secsToday = do + now <- io getCurrentTime + return (fromRational$ toRational$ utctDayTime now) + +-- | io-streams do not by default support tee or fan-out, because all reads "pop". +-- This broadcasts an InputStream to two new InputStreams, each of which will +-- receive every element. +-- +-- One useful application of this is creating additional copies of a stream before +-- it is connected to a downstream operator like +-- `System.IO.Streams.Concurrent.concurrentMerge`. Once that connection happens, +-- even if one stops demanding output from the `concurrentMerge`, one cannot know +-- how many elements it popped from the original input stream already. +dupStream :: InputStream a -> IO (InputStream a, InputStream a) +dupStream = error "dupStream unimplemented" + + +-- | This makes the EOS into an /explicit/, penultimate message. This way it survives +-- `concurrentMerge`. +liftStream :: InputStream a -> IO (InputStream (Lifted a)) +liftStream ins = + do flag <- newIORef True + S.makeInputStream $ do + x <- S.read ins + flg <- readIORef flag + case x of + Just y -> return (Just (StrmElt y)) + Nothing | flg -> do writeIORef flag False + return (Just EOS) + | otherwise -> return Nothing + +-- | Datatype for reifying end-of-stream. +data Lifted a = EOS | StrmElt a + deriving (Show,Eq,Read,Ord) + + +-------------------------------------------------------------------------------- +-- Tests +-------------------------------------------------------------------------------- + +-- Returns (inclusive,exclusive) bounds. +boundingBox :: NE.NonEmpty WinPos -> WinPos +boundingBox wps = (maxY,maxX, minY,minX) + where + minY = F.foldl1 min ys + minX = F.foldl1 min xs + maxY = F.foldl1 max (NE.zipWith (+) hs ys) + maxX = F.foldl1 max (NE.zipWith (+) ws xs) + (hs,ws,ys,xs) = UI.HydraPrint.unzip4 wps + +t0 :: NonEmpty WinPos +t0 = applyTiling (48,173) (3,2) + +t1 :: WinPos +t1 = boundingBox t0 + +#ifndef NOTESTING + +case_t0 :: Assertion +case_t0 = assertBool "Basic tiling example" + (noprop_goodtiling (48,173) (3,2)) + +-- DISABLING: This hangs under quickcheck. +noprop_goodtiling :: (Word,Word) -> (Word,Word) -> Bool +noprop_goodtiling (y,x) (splitY,splitX) = + if (y>0 && x>0 && splitY>0 && splitX > 0) + then (h == y && w == x) + else True -- Trivially. + where + tiles = applyTiling (y,x) (splitY,splitX) + (h,w,0,0) = boundingBox tiles + +testSuite :: Test +testSuite = $(testGroupGenerator) + +instance (Arbitrary a) => Arbitrary (NE.NonEmpty a) where + arbitrary = (:|) <$> arbitrary <*> arbitrary + shrink x = NE.fromList <$> shrink (NE.toList x) + +case_lift :: Assertion +case_lift = do + x <- liftStream =<< S.fromList [1..4] + y <- S.toList x + assertEqual "eq" [StrmElt 1,StrmElt 2,StrmElt 3,StrmElt 4,EOS] y + + +#endif + +removeIfExists :: FilePath -> IO () +removeIfExists fileName = removeFile fileName `catch` handleExists + where handleExists e + | isDoesNotExistError e = return () + | otherwise = throwIO e diff --git a/examples/make/Makefile b/examples/make/Makefile new file mode 100644 --- /dev/null +++ b/examples/make/Makefile @@ -0,0 +1,23 @@ + + + + +one: two three + $(eval PIPE := $(shell hydra-head -p)) + echo "Task one start." >> $(PIPE) + sleep 1 + echo "Task one finished." >> $(PIPE) + +two: + $(eval PIPE := $(shell hydra-head -p)) + echo "Task two start!" >> $(PIPE) + sleep 2 + echo "Task two finished!" >> $(PIPE) + + +three: + $(eval PIPE := $(shell hydra-head -p)) + echo "Task three start!" >> $(PIPE) + sleep 3 + echo "Task three finished!" >> $(PIPE) + diff --git a/examples/make/README b/examples/make/README new file mode 100644 --- /dev/null +++ b/examples/make/README @@ -0,0 +1,5 @@ + +Just run ./run.sh. +Don't run the Makefile directly. + +It requires that "hydra-view" and "hydra-head" are in the PATH. diff --git a/examples/make/run.sh b/examples/make/run.sh new file mode 100644 --- /dev/null +++ b/examples/make/run.sh @@ -0,0 +1,5 @@ +#!/bin/bash + +# Here's the simple approach, bring up the view and then as its +# "continuation", call the parallel Make. +hydra-view -- make -j diff --git a/hydra-head.hs b/hydra-head.hs new file mode 100644 --- /dev/null +++ b/hydra-head.hs @@ -0,0 +1,147 @@ + +-- | The client, goes with hydra-view. + +module Main where + +import Data.IORef +import Data.Word +import qualified Data.ByteString.Char8 as B +import Control.Monad (when, forM_) +import Control.Concurrent (threadDelay) +import System.Console.GetOpt (getOpt, ArgOrder(Permute), OptDescr(Option), ArgDescr(..), usageInfo) +import System.Environment (getArgs, getEnvironment) +import System.IO (IOMode(..), openFile, hClose, hPutStrLn) +import System.IO.Unsafe (unsafePerformIO) +import System.Directory (doesDirectoryExist, doesFileExist, removeFile) +import System.FilePath ((),takeFileName) +import System.Posix.Files (createNamedPipe) +import System.Posix.Types (CMode(..)) +import System.Process (runInteractiveCommand) +import System.Exit (exitFailure, exitSuccess) +import System.Random + +import Prelude as P + +import UI.HydraPrint (hydraPrint, defaultHydraConf) +import qualified System.IO.Streams as S +import System.IO.Streams.Concurrent (concurrentMerge) + +-------------------------------------------------------------------------------- + +-- | Extra usage docs beyond the flag info. +usageStr :: String +usageStr = unlines $ + [ "\n Hydra-head works with 'hydra-view' to add a new stream." + , " " + , " There are two main modes. Hydra-head either returns immediately," + , " producing the name of a pipe you can use. Or it pipes its " + , " standard input to the hydra-view session." + , " " + , " If you use the named pipe method, hydra-view will continue monitoring" + , " the pipe until you explicitly DELETE it. Thus the pipe can be used to " + , " aggregate the output of multiple commands." + ] + +-- | Datatype for command line options. +data Flag = + ReturnPipe -- | Return the name of a pipe immediately. + | SessionID FilePath -- | Use a session ID to match-up with the view server. + | ShowHelp + deriving (Eq,Read,Ord,Show) + +isSessionID :: Flag -> Bool +isSessionID (SessionID _ ) = True +isSessionID _ = False + +-- | Command line options. +cli_options :: [OptDescr Flag] +cli_options = + [ + Option ['s'] ["session"] (ReqArg SessionID "STRING") + "Use a sessionID to avoid collision with other hydra-view servers." + , Option ['p'] ["pipe"] (NoArg ReturnPipe) + "Return immediately [the path of] a named pipe for subsequent use." + , Option ['h'] ["help"] (NoArg ShowHelp) + "Show help and exit." + ] + +-------------------------------------------------------------------------------- +-- FIXME, factor these out: + +theEnv :: [(String, String)] +theEnv = unsafePerformIO getEnvironment + +-- Here we make some attempt to work on Windows: +defaultTempDir :: String +defaultTempDir = unsafePerformIO $ do + b <- doesDirectoryExist "/tmp/" + if b then return "/tmp/" else + case lookup "TEMP" theEnv of + Nothing -> error "hydra-view: Could not find a temporary directory to put the pipe source." + Just d -> return d + +-- | There's a simple policy on where to put the pipes, so that other clients can +-- find it. +defaultPipeSrc :: String +defaultPipeSrc = defaultTempDir ++ "hydra-view.pipe" + +sessionPipe :: String -> String +sessionPipe id = + defaultTempDir "hydra-view_session_"++id++".pipe" + +-- +-------------------------------------------------------------------------------- + + +main :: IO () +main = do + cli_args <- getArgs + let (options,restargs,errs) = getOpt Permute cli_options cli_args + showUsage = do putStrLn "USAGE: hydra-head [OPTIONS] -- commands to run" + putStrLn$ usageStr + putStr$ usageInfo " OPTIONS:" cli_options + when (not (null errs)) $ do + putStrLn$ "Errors parsing command line options:" + mapM_ (putStr . (" "++)) errs + showUsage + exitFailure + when (ShowHelp `elem` options) $ do showUsage; exitSuccess + + let pipePerms = CMode 0o777 + openPipe p = do b <- doesFileExist p + when b $ removeFile p + createNamedPipe p pipePerms + + getName g = do let suffix :: Word64 + (suffix,g') = random g + let path = defaultTempDir "hydra-head_tmp_pipe_"++show suffix + b <- doesFileExist path + if b then getName g' + else return path + + serverPipe = case filter isSessionID options of + [] -> defaultPipeSrc + [SessionID id] -> sessionPipe id + x -> error "hydra-head: Multiple sessions specified: "++show x + + stdGen <- getStdGen + newPipe <- getName stdGen + openPipe newPipe + + -- Tell the server about the new pipe: + hnd <- openFile serverPipe AppendMode + hPutStrLn hnd newPipe + hClose hnd + + if ReturnPipe `elem` options then + putStrLn newPipe + else do + -- SO terribly hackish. I get this error if I don't wait a bit: +-- hydra-head.exe: /tmp/hydra-head_tmp_pipe_14447421285220617689: openFile: does not exist (Device not configured) + threadDelay$ 10 * 1000 + outH <- openFile newPipe AppendMode + out <- S.handleToOutputStream outH + S.connect S.stdin out + hClose outH + removeFile newPipe + return () diff --git a/hydra-print.cabal b/hydra-print.cabal new file mode 100644 --- /dev/null +++ b/hydra-print.cabal @@ -0,0 +1,79 @@ +-- Initial hydra-print.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/ + +name: hydra-print +version: 0.1.0.0 +synopsis: NCurses interface to view multiple ByteString streams in parallel. +description: + + Output from parallel subprocesses is an ugly, many-headed beast. + Piping though hydra-print provides a NCurses interface to a dynamic + set output streams, displayed without interleaving. This is a + common problem, for example, with parallel compiles, or + data-processing scripts. + . + Specifically, this small library will split the screen when + parallelism occurs, and unsplit it when the parallelism is + finished (i.e. dynamically growing and losing \"heads\"). + The @io-streams@ library is used for all stream inputs. + +license: BSD3 +license-file: LICENSE +author: Ryan Newton +maintainer: Ryan Newton +copyright: 2013, Ryan Newton +category: Concurrency +build-type: Simple +cabal-version: >=1.8 + +extra-source-files: + SystemTest.hs + examples/make/Makefile + examples/make/run.sh + examples/make/README + +Library + Exposed-modules: UI.HydraPrint + Scripting.Parallel.ThreadPool + CPP-options: -DNOTESTING + Source-repository head + type: git + location: https://github.com/rrnewton/hydra-print + build-depends: base <= 4.7, transformers, containers, text, bytestring, io-streams, + semigroups, mtl, ncurses, directory, async +-- TEMP: I seem to need to include these even though they are used only by the EXECUTABLES: + , unix, process, time, random + -- build-depends base ==4.6.*, transformers ==0.3.*, containers ==0.5.*, text ==0.11.*, + -- mtl ==2.1.*, semigroups ==0.9.*, + -- bytestring ==0.10.*, io-streams ==1.0.*, hscurses ==1.4.* + +Test-Suite test-hydra-print + type: exitcode-stdio-1.0 + main-is: TestFramework.hs + -- DUPLICATED from above + build-depends: base <= 4.7, transformers, containers, text, bytestring, io-streams, + semigroups, mtl, ncurses, directory, async, + unix, process, time, random, + -- Additional dependencies: + --------------------------- + QuickCheck ==2.5.*, + HUnit ==1.2.*, + test-framework ==0.8.*, + test-framework-hunit ==0.3.*, + test-framework-quickcheck2 ==0.3.*, + test-framework-th ==0.2.* + + +Executable hydra-view + main-is: hydra-view.hs + CPP-options: -DNOTESTING + build-depends: base <= 4.7, transformers, containers, text, bytestring, io-streams, + semigroups, mtl, ncurses, directory, async, unix, + filepath, process, time, random + +Executable hydra-head + main-is: hydra-head.hs + CPP-options: -DNOTESTING + build-depends: base <= 4.7, transformers, containers, text, bytestring, io-streams, + semigroups, mtl, ncurses, directory, async, unix, + filepath, process, time, random diff --git a/hydra-view.hs b/hydra-view.hs new file mode 100644 --- /dev/null +++ b/hydra-view.hs @@ -0,0 +1,193 @@ +{-# LANGUAGE CPP #-} +{-# LANGUAGE NamedFieldPuns #-} + +-- | The server, sets up a named pipe to receive the names of named pipes. + +module Main where + +import Data.IORef +import qualified Data.Set as Set +import qualified Data.ByteString.Char8 as B +import Control.Monad (when, forM_) +import Control.Concurrent (threadDelay, forkIO) +-- import Control.Concurrent.Chan +import System.Console.GetOpt (getOpt, ArgOrder(Permute), OptDescr(Option), ArgDescr(..), usageInfo) +import System.Environment (getArgs, getEnvironment) +import System.IO (IOMode(..), openFile, hClose, hPutStrLn, stderr) +import System.IO.Unsafe (unsafePerformIO) +import System.Directory (doesDirectoryExist, doesFileExist, removeFile) +import System.FilePath ((),takeFileName) +import System.Posix.Files (createNamedPipe) +import System.Posix.Types (CMode(..)) +import System.Process (runInteractiveCommand, terminateProcess) +import System.Exit (exitFailure, exitSuccess) + +import UI.HydraPrint (hydraPrint, defaultHydraConf, HydraConf(..)) +import qualified System.IO.Streams as S +import System.IO.Streams.Concurrent (concurrentMerge) + +-------------------------------------------------------------------------------- + +-- | Extra usage docs beyond the flag info. +usageStr :: String +usageStr = unlines $ + [ "\n Hydra-view provides multi-\"headed\" output. That is, it provides a" + , " way to view a dynamic collection of output streams." + , " " + , " Hydra-view works by creating a named pipe to which file names are sent." + , " Each one one of those filenames is a 'stream source', usually a named-" + , " pipe itself. Each such stream gets its own panel while it is producing " + , " output. " + , " " + , " See the command 'hydra-head' for a convenient way to connect new" + , " streams to an existing hydra-view session." + ] + + +theEnv :: [(String, String)] +theEnv = unsafePerformIO getEnvironment + +-- | There's a simple policy on where to put the pipes, so that other clients can +-- find it. +defaultPipeSrc :: String +defaultPipeSrc = defaultTempDir ++ "hydra-view.pipe" + +-- Here we make some attempt to work on Windows: +defaultTempDir :: String +defaultTempDir = unsafePerformIO $ do + b <- doesDirectoryExist "/tmp/" + if b then return "/tmp/" else + case lookup "TEMP" theEnv of + Nothing -> error "hydra-view: Could not find a temporary directory to put the pipe source." + Just d -> return d + +sessionPipe :: String -> String +sessionPipe id = + defaultTempDir "hydra-view_session_"++id++".pipe" + +-- | Datatype for command line options. +data Flag = + PipeSrc FilePath -- | Use a user-provided file as the source of pipes. + | SessionID FilePath -- | Use a session ID to match-up with clients. + | ShowHelp + | NoColor + deriving (Eq,Read,Ord,Show) + +isSrcFlag :: Flag -> Bool +isSrcFlag (PipeSrc _) = True +isSrcFlag (SessionID _) = True +isSrcFlag _ = False + +-- | Command line options. +cli_options :: [OptDescr Flag] +cli_options = + [ Option ['p'] ["pipesrc"] (ReqArg PipeSrc "FILENAME") + "Use a specific file as the source of streams." + , Option ['s'] ["session"] (ReqArg SessionID "STRING") + "Use a sessionID to avoid collision with other hydra-view servers." + , Option [] ["no-color"] (NoArg NoColor) + "Don't use color to distinguish panels." + , Option ['h'] ["help"] (NoArg ShowHelp) + "Show help and exit." + ] + +main :: IO () +main = do + cli_args <- getArgs + let (options,restargs,errs) = getOpt Permute cli_options cli_args + showUsage = do putStrLn "USAGE: hydra-view [OPTIONS] -- commands to run" + putStrLn$ usageStr + putStr$ usageInfo " OPTIONS:" cli_options + when (not (null errs)) $ do + putStrLn$ "Errors parsing command line options:" + mapM_ (putStr . (" "++)) errs + showUsage + exitFailure + + let pipePerms = CMode 0o777 + openPipe p = do b <- doesFileExist p + when b $ removeFile p + createNamedPipe p pipePerms + when (ShowHelp `elem` options) $ do showUsage; exitSuccess + pipe <- case filter isSrcFlag options of + [PipeSrc file] -> return file -- Pre-existing! + [SessionID id] -> do let p = sessionPipe id + openPipe p + return p + [] -> do openPipe defaultPipeSrc + return defaultPipeSrc + _ -> error$"hydra-view: too many options! Can only set the pipe to use once: "++show options + +-- (_, outH, _, _pid) <- runInteractiveCommand$ tailCmd++pipe + (_, outH, _, _pid) <- runInteractiveCommand$ "tail -f "++pipe + strmSrc <- S.lines =<< S.handleToInputStream outH + + openHnds <- newIORef [] + -- Lazy resource cleanup: keeps a set of open files, and collects old ones as new + -- ones are added: + let addNPoll newpth pr = do + open <- atomicModifyIORef openHnds + (\ls -> let new = (newpth,pr) : ls in (new,new)) + forM_ open $ \ (_pth,(hnd,strm)) -> do + b <- S.atEOF strm + when b $ hClose hnd + return () + + -- Store closed pipes based on their filename: + closedPipes <- newIORef Set.empty + + -- Read new pipes to get bytestring streams: + let rd pth = do +-- putStrLn$ " NEW string on pipe! "++show pth + let str = B.unpack pth + strmName = takeFileName str + b <- doesFileExist str + if not b then do + nl <- S.nullInput + return (strmName, nl) + else do + (_, strmH, _, pid) <- runInteractiveCommand$ tailCmd++str + strm <- S.lines =<< S.handleToInputStream strmH + strm' <- S.map (`B.append` (B.pack "\n")) strm + + -- Run a timer to poll if the files still exist, and close the relevant streams if + -- not. + forkIO + (let loop = do + threadDelay$ 200 * 1000 + b <- doesFileExist str + when (not b) $ do +-- dbgPrnt$ " [dbg] Terminating process: "++show pid + terminateProcess pid + loop + in loop) + return (strmName,strm') + srcs <- S.mapM rd strmSrc + + srcs' <- + case restargs of + [] -> do putStrLn$ " [hydra-view] No command to run; send named-pipe names to: "++show pipe + return srcs + _ -> do + let cmd = unwords restargs + putStrLn$" [hydra-view] First output stream from command: "++show cmd + (_stdinH, stdoutH, stderrH, _pid) <- runInteractiveCommand cmd + inS <- S.lines =<< S.handleToInputStream stdoutH + errS <- S.lines =<< S.handleToInputStream stderrH + merged <- concurrentMerge [inS,errS] + merged' <- S.map (`B.append` (B.pack "\n")) merged -- Unlines it. + singleton <- S.fromList [("main",merged')] + S.appendInputStream singleton srcs + + let useColor = not$ NoColor `elem` options + conf = defaultHydraConf { useColor } + + hydraPrint conf srcs' + return () + +pollFileDelay :: Int +pollFileDelay = 200 * 1000 + +-- tailCmd = "tail -f " +-- tailCmd = "tail -c +0 -F " +tailCmd = "tail -c +0 -f "