diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -1,5 +1,13 @@ # Revision history for Z-IO +## 0.7.0.0 -- 2020-03-09 + +* Change resource `Pool` to keyed by default, add `SimplePool`. +* Add `Semigroup` instance to `Logger`. +* Add `clearInputBuffer/clearOutputBuffer` to `Z.IO.Buffered`. +* Add `catchSync/ingoreSync` to `Z.IO.Exception`. +* Add `putStdLn/printStdLn` back. + ## 0.6.4.0 -- 2020-02-20 * Add `initProcess'` to kill process while finish using the process resource by default. diff --git a/Z-IO.cabal b/Z-IO.cabal --- a/Z-IO.cabal +++ b/Z-IO.cabal @@ -1,6 +1,6 @@ cabal-version: 2.4 name: Z-IO -version: 0.6.4.0 +version: 0.7.0.0 synopsis: Simple and high performance IO toolkit for Haskell description: Simple and high performance IO toolkit for Haskell, including @@ -125,7 +125,7 @@ , time >=1.9 && <=2.0 , unix-time >=0.4.7 && <=0.5 , unordered-containers ^>=0.2 - , Z-Data ^>=0.6 + , Z-Data ^>=0.7 default-language: Haskell2010 default-extensions: diff --git a/Z/IO/Buffered.hs b/Z/IO/Buffered.hs --- a/Z/IO/Buffered.hs +++ b/Z/IO/Buffered.hs @@ -21,6 +21,7 @@ , newBufferedInput' , readBuffer, readBufferText , unReadBuffer + , clearInputBuffer , readParser , readParseChunks , readExactly @@ -34,6 +35,7 @@ , writeBuffer , writeBuilder , flushBuffer + , clearOutputBuffer -- * common buffer size , V.defaultChunkSize , V.smallChunkSize @@ -229,6 +231,10 @@ return (T.validate (V.fromArr arr s i)) else return (T.validate bs) +-- | Clear already buffered input. +clearInputBuffer :: BufferedInput -> IO () +clearInputBuffer BufferedInput{..} = writeIORef bufPushBack V.empty + -- | Read exactly N bytes. -- -- If EOF reached before N bytes read, an 'OtherError' with name 'EINCOMPLETE' will be thrown. @@ -449,3 +455,6 @@ withMutablePrimArrayContents outputBuffer $ \ ptr -> bufOutput ptr i writePrimIORef bufIndex 0 +-- | Clear already buffered output. +clearOutputBuffer :: BufferedOutput -> IO () +clearOutputBuffer BufferedOutput{..} = writePrimIORef bufIndex 0 diff --git a/Z/IO/Exception.hs b/Z/IO/Exception.hs --- a/Z/IO/Exception.hs +++ b/Z/IO/Exception.hs @@ -70,6 +70,9 @@ , throwOtherError , unwrap , unwrap' + -- * Sync exception tools + , catchSync + , ignoreSync -- * Re-exports , module Control.Exception , HasCallStack @@ -204,6 +207,25 @@ {-# INLINABLE unwrap' #-} unwrap' _ _ (Just x) = return x unwrap' n d Nothing = throwOtherError n d + +-- | Check if the given exception is synchronous +-- +isSyncException :: Exception e => e -> Bool +isSyncException e = + case fromException (toException e) of + Just (SomeAsyncException _) -> False + Nothing -> True + +-- | Same as upstream 'C.catch', but will not catch asynchronous exceptions +-- +catchSync :: Exception e => IO a -> (e -> IO a) -> IO a +catchSync f g = f `catch` \ e -> + if isSyncException e then g e else throwIO e + +-- | Ingore all synchronous exceptions. +-- +ignoreSync :: IO a -> IO () +ignoreSync f = catchSync (void f) (\ (_ :: SomeException) -> return ()) -------------------------------------------------------------------------------- diff --git a/Z/IO/Logger.hs b/Z/IO/Logger.hs --- a/Z/IO/Logger.hs +++ b/Z/IO/Logger.hs @@ -106,6 +106,7 @@ ------------------------------------------------------------------------------- +-- | Formatter used by `Logger`. type LogFormatter = B.Builder () -- ^ data\/time string(second precision) -> Level -- ^ log level -> B.Builder () -- ^ log content @@ -113,29 +114,24 @@ -> ThreadId -- ^ logging thread id -> B.Builder () --- | Extensible logger type. +-- | The `Logger` type. data Logger = Logger - { loggerPushBuilder :: B.Builder () -> IO () - -- ^ Push log into buffer - , flushLogger :: IO () - -- ^ Flush logger's buffer to output device - , flushLoggerThrottled :: IO () - -- ^ Throttled flush, e.g. use 'throttleTrailing_' from "Z.IO.LowResTimer" - , loggerTSCache :: IO (B.Builder ()) - -- ^ An IO action return a formatted date\/time string - , loggerFmt :: LogFormatter - -- ^ Log formatter - , loggerLevel :: {-# UNPACK #-} !Level - -- ^ Output logs if level is equal or higher than this value. - } + (Level -> Bool -> CallStack -> B.Builder () -> IO ()) -- ^ logging function + (IO ()) -- ^ manually flush +-- | For composing different loggers +instance Semigroup Logger where + Logger log1 flush1 <> Logger log2 flush2 = Logger + (\ l b cs bu -> log1 l b cs bu >> log2 l b cs bu) + (flush1 >> flush2) + -- | Logger config type used in this module. data LoggerConfig = LoggerConfig { loggerMinFlushInterval :: {-# UNPACK #-} !Int -- ^ Minimal flush interval, see Notes on 'debug' , loggerLineBufSize :: {-# UNPACK #-} !Int -- ^ Buffer size to build each log line - , loggerConfigLevel :: {-# UNPACK #-} !Level + , loggerLevel :: {-# UNPACK #-} !Level -- ^ Config log's filter level , loggerFormatter :: LogFormatter -- ^ Log formatter @@ -179,10 +175,15 @@ logsRef <- newIORef [] let flush = flushLogIORef oLock logsRef throttledFlush <- throttleTrailing_ loggerMinFlushInterval flush - return $ Logger (pushLogIORef logsRef loggerLineBufSize) - flush throttledFlush defaultTSCache loggerFormatter - loggerConfigLevel + return $ Logger (\ level flushNow cstack bu -> + when (level >= loggerLevel) $ do + ts <- defaultTSCache + tid <- myThreadId + (pushLogIORef logsRef loggerLineBufSize) $ loggerFormatter ts level bu cstack tid + if flushNow then flush else throttledFlush + ) flush + -- | Make a new logger write to 'stderrBuf'. newStdLogger :: LoggerConfig -> IO Logger newStdLogger config = newLogger config stderrBuf @@ -304,7 +305,9 @@ -- | Manually flush global logger. flushDefaultLogger :: IO () -flushDefaultLogger = getDefaultLogger >>= flushLogger +flushDefaultLogger = do + (Logger _ flush) <- getDefaultLogger + flush -- | Flush global logger when program exits. withDefaultLogger :: IO () -> IO () @@ -358,6 +361,7 @@ -- Level other than built-in ones, are formatted in decimal numeric format, i.e. -- @defaultLevelFmt 60 == "LEVEL60"@ defaultLevelFmt :: Level -> B.Builder () +{-# INLINE defaultLevelFmt #-} defaultLevelFmt level = case level of CRITICAL -> "CRITICAL" FATAL -> "FATAL" @@ -368,18 +372,23 @@ level' -> "LEVEL" >> B.int level' debug :: HasCallStack => B.Builder () -> IO () +{-# INLINE debug #-} debug = otherLevel_ DEBUG False callStack info :: HasCallStack => B.Builder () -> IO () +{-# INLINE info #-} info = otherLevel_ INFO False callStack warning :: HasCallStack => B.Builder () -> IO () +{-# INLINE warning #-} warning = otherLevel_ WARNING False callStack fatal :: HasCallStack => B.Builder () -> IO () +{-# INLINE fatal #-} fatal = otherLevel_ FATAL True callStack critical :: HasCallStack => B.Builder () -> IO () +{-# INLINE critical #-} critical = otherLevel_ CRITICAL True callStack otherLevel :: HasCallStack @@ -387,25 +396,31 @@ -> Bool -- ^ flush immediately? -> B.Builder () -- ^ log content -> IO () +{-# INLINE otherLevel #-} otherLevel level flushNow bu = otherLevel_ level flushNow callStack bu otherLevel_ :: Level -> Bool -> CallStack -> B.Builder () -> IO () +{-# INLINE otherLevel_ #-} otherLevel_ level flushNow cstack bu = do - logger <- getDefaultLogger - otherLevelTo_ level flushNow cstack logger bu + (Logger f _) <- getDefaultLogger + f level flushNow cstack bu -------------------------------------------------------------------------------- debugTo :: HasCallStack => Logger -> B.Builder () -> IO () +{-# INLINE debugTo #-} debugTo = otherLevelTo_ DEBUG False callStack infoTo :: HasCallStack => Logger -> B.Builder () -> IO () +{-# INLINE infoTo #-} infoTo = otherLevelTo_ INFO False callStack warningTo :: HasCallStack => Logger -> B.Builder () -> IO () +{-# INLINE warningTo #-} warningTo = otherLevelTo_ WARNING False callStack fatalTo :: HasCallStack => Logger -> B.Builder () -> IO () +{-# INLINE fatalTo #-} fatalTo = otherLevelTo_ FATAL True callStack otherLevelTo :: HasCallStack @@ -414,16 +429,11 @@ -> Bool -- ^ flush immediately? -> B.Builder () -- ^ log content -> IO () -otherLevelTo logger level flushNow = - otherLevelTo_ level flushNow callStack logger +{-# INLINE otherLevelTo #-} +otherLevelTo logger level flushNow = otherLevelTo_ level flushNow callStack logger otherLevelTo_ :: Level -> Bool -> CallStack -> Logger -> B.Builder () -> IO () -otherLevelTo_ level flushNow cstack logger bu = when (level >= loggerLevel logger) $ do - ts <- loggerTSCache logger - tid <- myThreadId - (loggerPushBuilder logger) $ (loggerFmt logger) ts level bu cstack tid - if flushNow - then flushLogger logger - else flushLoggerThrottled logger +{-# INLINE otherLevelTo_ #-} +otherLevelTo_ level flushNow cs (Logger f _) = f level flushNow cs foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> CInt diff --git a/Z/IO/LowResTimer.hs b/Z/IO/LowResTimer.hs --- a/Z/IO/LowResTimer.hs +++ b/Z/IO/LowResTimer.hs @@ -229,7 +229,7 @@ where timeoutAThread tid = void . forkIO $ throwTo tid (TimeOutException tid undefined) --- | Similar to 'timeoutLowRes', but raise a 'TimeOutException' to current thread +-- | Similar to 'timeoutLowRes', but throw an async 'TimeOutException' to current thread -- instead of return 'Nothing' if timeout. timeoutLowResEx :: HasCallStack => Int -- ^ timeout in unit of 0.1s @@ -245,9 +245,14 @@ where timeoutAThread tid = void . forkIO $ throwTo tid (TimeOutException tid callStack) --- | see 'timeoutLowResEx' on 'TimeOutException', this exception is not a sub-exception type of 'SomeIOException'. +-- | see 'timeoutLowResEx' on 'TimeOutException'. +-- +-- This exception is not a sub-exception type of 'SomeIOException', +-- but a sub-exception type of 'TimeOutException'. data TimeOutException = TimeOutException ThreadId CallStack deriving Show -instance Exception TimeOutException +instance Exception TimeOutException where + toException = asyncExceptionToException + fromException = asyncExceptionFromException -- | Similiar to 'threadDelay', suspends the current thread for a given number of deciseconds. @@ -323,7 +328,7 @@ go nextList tListRef counter EQ -> do -- if round number is equal to 0, fire it atomicSubCounter_ counter 1 - catch action ( \ (_ :: SomeException) -> return () ) -- well, we really don't want timers break our loop + catchSync action ( \ (_ :: SomeException) -> return () ) -- well, we really don't want timers break our loop go nextList tListRef counter GT -> do -- if round number is larger than 0, put it back for another round atomicModifyIORef' tListRef $ \ tlist -> (TimerItem roundCounter action tlist, ()) diff --git a/Z/IO/Process.hsc b/Z/IO/Process.hsc --- a/Z/IO/Process.hsc +++ b/Z/IO/Process.hsc @@ -166,7 +166,7 @@ initProcess' opt = initResource (spawn opt) $ \ (s0, s1, s2, pstate) -> do m_pid <- getProcessPID pstate - maybe (return ()) (flip killPID SIGTERM) m_pid + forM_ m_pid (`killPID` SIGTERM) _ <- waitProcessExit pstate forM_ s0 closeUVStream forM_ s1 closeUVStream diff --git a/Z/IO/Resource.hs b/Z/IO/Resource.hs --- a/Z/IO/Resource.hs +++ b/Z/IO/Resource.hs @@ -11,7 +11,7 @@ . The 'Applicative' and 'Monad' instance is especially useful when you want safely combine multiple resources. -A high performance resource pool based on STM is also provided. +A high performance resource pool is also provided. -} @@ -24,19 +24,25 @@ , withResource' -- * Resource pool , Pool - , PoolState(..) , initPool - , withResourceInPool - , poolStat, poolInUse + , withPool + , SimplePool + , initSimplePool + , withSimplePool + , statPool -- * Re-export , liftIO ) where -import Control.Concurrent.STM +import Control.Concurrent import Control.Monad -import qualified Control.Monad.Catch as MonadCatch +import qualified Control.Monad.Catch as MonadCatch import Control.Monad.IO.Class +import qualified Data.Map.Strict as M import Z.Data.PrimRef.PrimIORef +import Z.Data.Array +import qualified Z.Data.Vector as V +import Data.IORef import Z.IO.LowResTimer import Z.IO.Exception @@ -159,143 +165,145 @@ -------------------------------------------------------------------------------- --- | A single resource pool entry. -data Entry a = Entry - (a, IO ()) -- the resource and clean up action - {-# UNPACK #-} !Int -- the life remaining - -data PoolState = PoolClosed | PoolScanning | PoolEmpty deriving (Eq, Show) +-- | A entry linked-list annotated with size. +data Entry res + = EntryNil + | EntryCons + (res, IO ()) -- the resource and clean up action + {-# UNPACK #-} !Int -- size from this point on + {-# UNPACK #-} !Int -- the life remaining + (Entry res) -- next entry --- | A high performance resource pool based on STM. +-- | A high performance resource pool. -- --- We choose to not divide pool into strips due to the difficults in resource balancing. If there --- is a high contention on resource (see 'statPool'), just increase the maximum number of resources --- can be opened. +-- The Pool is first divided by GHC runtime capabilities, each capability maintains a map from key to living +-- resource list. Resource are fetched from living list first, create on demand if there's no living resource. -- -data Pool a = Pool - { _poolResource :: Resource a - , _poolLimit :: Int - , _poolIdleTime :: Int - , _poolEntries :: TVar [Entry a] - , _poolInUse :: TVar Int - , _poolState :: TVar PoolState +data Pool key res = Pool + { _poolResource :: key -> Resource res -- ^ how to get a resource + , _poolLimitPerKey :: {-# UNPACK #-} !Int -- ^ max number for resource we keep alive after used + , _poolIdleTime :: {-# UNPACK #-} !Int -- ^ max idle time for resource we keep alive + , _poolArray :: {-# UNPACK #-} !(UnliftedArray (IORef (Maybe (M.Map key (Entry res))))) } +-- | Dump the status of pool. +statPool :: Pool key res -> IO (SmallArray (M.Map key Int)) +statPool (Pool _ _ _ arr) = (`V.traverseVec` arr) $ \ resMapRef -> do + mResMap <- readIORef resMapRef + case mResMap of + Just resMap -> return $ (`fmap` resMap) ( \ es -> + case es of EntryCons _ siz _ _ -> siz + _ -> 0) + _ -> throwECLOSED + -- | Initialize a resource pool with given 'Resource' -- -- Like other initXXX functions, this function won't open a resource pool until you use 'withResource'. --- And this resource pool follow the same resource management pattern like other resources. --- -initPool :: Resource a - -> Int -- ^ maximum number of resources can be opened +initPool :: (key -> Resource res) + -> Int -- ^ maximum number of resources per local pool per key to be maintained. -> Int -- ^ amount of time after which an unused resource can be released (in seconds). - -> Resource (Pool a) -initPool res limit itime = initResource createPool closePool + -> Resource (Pool key res) +initPool resf limit itime = initResource createPool closePool where createPool = do - entries <- newTVarIO [] - inuse <- newTVarIO 0 - state <- newTVarIO PoolEmpty - return (Pool res limit itime entries inuse state) - - closePool (Pool _ _ _ entries _ state) = join . atomically $ do - c <- readTVar state - if c == PoolClosed - then return (return ()) - else do - writeTVar state PoolClosed - return (do - es <- readTVarIO entries - forM_ es $ \ (Entry (_, close) _) -> - MonadCatch.handleAll (\ _ -> return ()) close) + numCaps <- getNumCapabilities + marr <- newArr numCaps + forM_ [0..numCaps-1] $ \ i -> do + writeArr marr i =<< newIORef (Just M.empty) + arr <- unsafeFreezeArr marr + return (Pool resf limit itime arr) --- | Get a resource pool's 'PoolState' --- --- This function is useful when debug, under load lots of 'PoolEmpty' may indicate --- contention on resources, i.e. the limit on maximum number of resources can be opened --- should be adjusted to a higher number. On the otherhand, lots of 'PoolScanning' --- may indicate there're too much free resources. --- -poolStat :: Pool a -> IO PoolState -poolStat pool = readTVarIO (_poolState pool) + closePool (Pool _ _ _ localPoolArr) = do + -- close all existed resource + (`V.traverseVec_` localPoolArr) $ \ resMapRef -> + atomicModifyIORef resMapRef $ \ mResMap -> + case mResMap of + Just resMap -> (Nothing, mapM_ closeEntry resMap) + _ -> (Nothing, return ()) --- | Get how many resource is being used within a resource pool. --- --- This function is useful when debug, under load in use number alway reaches limit may indicate --- contention on resources, i.e. the limit on maximum number of resources can be opened --- should be adjusted to a higher number. --- -poolInUse :: Pool a -> IO Int -poolInUse pool = readTVarIO (_poolInUse pool) + closeEntry (EntryCons (_, close) _ _ _) = ignoreSync close + closeEntry EntryNil = return () -- | Open resource inside a given resource pool and do some computation. -- -- This function is thread safe, concurrently usage will be guaranteed --- to get different resource. If exception happens, +-- to get different resource. If exception happens during computation, -- resource will be closed(not return to pool). -withResourceInPool :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack) - => Pool a -> (a -> m b) -> m b -withResourceInPool (Pool res limit itime entries inuse state) k = +withPool :: (MonadCatch.MonadMask m, MonadIO m, Ord key, HasCallStack) + => Pool key res -> key -> (res -> m a) -> m a +withPool (Pool resf limitPerKey itime arr) key f = do + !resMapRef <- indexArr arr . fst <$> liftIO (threadCapability =<< myThreadId) fst <$> MonadCatch.generalBracket - (liftIO takeFromPool) + (liftIO $ takeFromPool resMapRef) (\ r@(_, close) exit -> case exit of - MonadCatch.ExitCaseSuccess _ -> liftIO (returnToPool r) - _ -> liftIO $ do - atomically $ modifyTVar' inuse (subtract 1) - close) - (\ (a, _) -> k a) + MonadCatch.ExitCaseSuccess _ -> liftIO (returnToPool resMapRef r) + _ -> liftIO close) + (\ (a, _) -> f a) where - takeFromPool = join . atomically $ do - c <- readTVar state - if c == PoolClosed - then throwECLOSEDSTM - else do - es <- readTVar entries - case es of - ((Entry a _):es') -> do - writeTVar entries es' - return (return a) - _ -> do - i <- readTVar inuse - when (i == limit) retry - modifyTVar' inuse (+1) - return (acquire res `onException` - atomically (modifyTVar' inuse (subtract 1))) + takeFromPool resMapRef = + join . atomicModifyIORef' resMapRef $ \ mResMap -> + case mResMap of + Just resMap -> + case M.lookup key resMap of + Just (EntryCons a _ _ es') -> + (Just $! M.adjust (const es') key resMap, return a) + _ -> (Just resMap, acquire (resf key)) + _ -> (Nothing, throwECLOSED) - returnToPool a = join . atomically $ do - c <- readTVar state - case c of - PoolClosed -> return (snd a) - PoolEmpty -> do - modifyTVar' entries (Entry a itime:) - writeTVar state PoolScanning - return (void $ registerLowResTimer 10 scanPool) - _ -> do - modifyTVar' entries (Entry a itime:) - return (return ()) + returnToPool resMapRef r = do + join . atomicModifyIORef' resMapRef $ \ mResMap -> + case mResMap of + Just resMap -> + case M.lookup key resMap of + Just (EntryCons _ siz _ _) -> + if siz < limitPerKey + -- if entries under given key do not exceed limit, we prepend res back to entries + then (Just $! M.adjust (EntryCons r (siz+1) itime) key resMap, return ()) + -- otherwise we close it + else (Just resMap, snd r) + _ -> (Just $! M.insert key (EntryCons r 1 itime EntryNil) resMap, + scanLocalPool resMapRef) + _ -> (Nothing, snd r) - scanPool = do - join . atomically $ do - c <- readTVar state - if c == PoolClosed - then return (return ()) - else do - es <- readTVar entries - if (null es) - then do - writeTVar state PoolEmpty - return (return ()) - else do - let (deadNum, dead, living) = age es 0 [] [] - writeTVar entries living - modifyTVar' inuse (subtract deadNum) - return (do - forM_ dead $ \ (_, close) -> - MonadCatch.handleAll (\ _ -> return ()) close - void $ registerLowResTimer 10 scanPool) + scanLocalPool resMapRef = do + registerLowResTimer_ 10 . join . atomicModifyIORef' resMapRef $ \ mResMap -> + case mResMap of + Just resMap -> + case M.lookup key resMap of + -- this is where we clean up empty keys + Just EntryNil -> (Just $! M.delete key resMap, return ()) + Just es -> do + let (dead, living) = age es 0 [] EntryNil + case living of + -- no living resources any more, stop scanning + EntryNil -> (Just $! M.delete key resMap, + forM_ dead (ignoreSync . snd)) + _ -> (Just $! M.adjust (const living) key resMap, + (do forM_ dead (ignoreSync . snd) + scanLocalPool resMapRef)) + -- no living resources under given key, stop scanning + _ -> (Just resMap, return ()) + _ -> (Nothing, return ()) - age ((Entry a life):es) !deadNum dead living - | life > 1 = age es deadNum dead (Entry a (life-1):living) - | otherwise = age es (deadNum+1) (a:dead) living - age _ !deadNum dead living = (deadNum, dead, living) + age (EntryCons a _ life es) !livingNum dead living + | life > 1 = let !livingNum' = (livingNum+1) + in age es livingNum' dead (EntryCons a livingNum' (life-1) living) + | otherwise = age es livingNum (a:dead) living + age _ _ dead living = (dead, living) + +-- | Simple resource pool where lookup via key is not needed. +type SimplePool res = Pool () res + +-- | Initialize a 'SimplePool'. +initSimplePool :: Resource res + -> Int -- ^ maximum number of resources per local pool to be maintained. + -> Int -- ^ amount of time after which an unused resource can be released (in seconds). + -> Resource (SimplePool res) +initSimplePool f = initPool (const f) + +-- | Open resource with 'SimplePool', see 'withPool' +-- +withSimplePool :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack) + => SimplePool res -> (res -> m a) -> m a +withSimplePool pool = withPool pool () diff --git a/Z/IO/StdStream.hs b/Z/IO/StdStream.hs --- a/Z/IO/StdStream.hs +++ b/Z/IO/StdStream.hs @@ -25,7 +25,7 @@ putStd "hello world!" -- Raw mode - setStdinTTYMode UV_TTY_MODE_RAW + setStdinTTYMode TTY_MODE_RAW forever $ do withMVar stdinBuf $ \ i -> withMVar stdoutBuf $ \ o -> do bs <- readBuffer i @@ -41,11 +41,12 @@ , getStdStreamFD , isStdStreamTTY , setStdinTTYMode + , withRawStdin , getStdoutWinSize , stdin, stdout, stderr , stdinBuf, stdoutBuf, stderrBuf -- * utils - , readStd, printStd, putStd + , readStd, printStd, putStd, printStdLn, putStdLn -- * re-export , withMVar -- * Constant @@ -238,12 +239,17 @@ -- | Change terminal's mode if stdin is connected to a terminal, -- do nothing if stdout is not connected to TTY. +-- setStdinTTYMode :: TTYMode -> IO () setStdinTTYMode mode = case stdin of StdStream True hdl _ uvm -> withUVManager' uvm . throwUVIfMinus_ $ uv_tty_set_mode hdl mode _ -> return () +-- | Set stdin to raw mode before run IO, set back to normal after. +withRawStdin :: IO a -> IO a +withRawStdin = bracket_ (setStdinTTYMode TTY_MODE_RAW) (setStdinTTYMode TTY_MODE_NORMAL) + -- | Get terminal's output window size in (width, height) format, -- return (-1, -1) if stdout is not connected to TTY. getStdoutWinSize :: HasCallStack => IO (CInt, CInt) @@ -257,17 +263,27 @@ -------------------------------------------------------------------------------- --- | Print a 'Print' and flush to stdout, with a linefeed. +-- | Print a 'Print' and flush to stdout, without linefeed. printStd :: (HasCallStack, T.Print a) => a -> IO () printStd s = putStd (T.toUTF8Builder s) --- | Print a 'Builder' and flush to stdout, with a linefeed. +-- | Print a 'Builder' and flush to stdout, without linefeed. putStd :: HasCallStack => B.Builder a -> IO () putStd b = withMVar stdoutBuf $ \ o -> do + writeBuilder o b + flushBuffer o + +-- | Print a 'Print' and flush to stdout, with a linefeed. +printStdLn :: (HasCallStack, T.Print a) => a -> IO () +printStdLn s = putStdLn (T.toUTF8Builder s) + +-- | Print a 'Builder' and flush to stdout, with a linefeed. +putStdLn :: HasCallStack => B.Builder a -> IO () +putStdLn b = withMVar stdoutBuf $ \ o -> do writeBuilder o (b >> B.char8 '\n') flushBuffer o --- | Read a line from stdin +-- | Read a line from stdin(in normal mode). -- -- This function will throw 'ECLOSED' when meet EOF, which may cause trouble if stdin is connected -- to a file, use 'readLine' instead. diff --git a/Z/IO/StdStream/Ansi.hs b/Z/IO/StdStream/Ansi.hs --- a/Z/IO/StdStream/Ansi.hs +++ b/Z/IO/StdStream/Ansi.hs @@ -23,7 +23,7 @@ -- * Control codes cursorUp, cursorDown, cursorForward, cursorBackward, cursorDownLine, cursorUpLine , - setCursorColumn, setCursorPosition, saveCursor, restoreCursor, + setCursorColumn, setCursorPosition, saveCursor, restoreCursor, getCursorPosition, clearFromCursorToScreenEnd, clearFromCursorToScreenBeginning, clearScreen, clearFromCursorToLineEnd, clearFromCursorToLineBeginning, clearLine, scrollPageUp, scrollPageDown, @@ -47,9 +47,13 @@ ) where import qualified Z.Data.Builder as B +import qualified Z.Data.Parser as P import qualified Z.Data.Text as T +import Z.Data.ASCII import Data.Word import GHC.Generics +import Z.IO.StdStream +import Z.IO.Buffered csi :: [Int] -- ^ List of parameters for the control sequence -> B.Builder () -- ^ Character(s) that identify the control function @@ -72,6 +76,20 @@ cursorDownLine n = csi [n] (B.char8 'E') cursorUpLine n = csi [n] (B.char8 'F') +getCursorPosition :: IO (Int, Int) +getCursorPosition = do + withRawStdin . withMVar stdinBuf $ \ i -> do + clearInputBuffer i + putStd (csi [] "6n") + readParser (do + P.word8 ESC + P.word8 BRACKET_LEFT + !n <- P.int + P.word8 SEMICOLON + !m <- P.int + P.word8 LETTER_R + return (m, n)) i + -- | Code to move the cursor to the specified column. The column numbering is -- 1-based (that is, the left-most column is numbered 1). setCursorColumn :: Int -- ^ 1-based column to move to @@ -88,7 +106,6 @@ saveCursor, restoreCursor :: B.Builder () saveCursor = B.char8 '\ESC' >> B.char8 '7' restoreCursor = B.char8 '\ESC' >> B.char8 '8' - clearFromCursorToScreenEnd, clearFromCursorToScreenBeginning, clearScreen :: B.Builder () clearFromCursorToLineEnd, clearFromCursorToLineBeginning, clearLine :: B.Builder () diff --git a/Z/IO/UV/Manager.hs b/Z/IO/UV/Manager.hs --- a/Z/IO/UV/Manager.hs +++ b/Z/IO/UV/Manager.hs @@ -85,7 +85,7 @@ toUTF8BuilderP p uvm = T.parenWhen (p > 10) $ "UVManager on capability " >> T.int (uvmCap uvm) -uvManagerArray :: IORef (Array UVManager) +uvManagerArray :: IORef (SmallArray UVManager) {-# NOINLINE uvManagerArray #-} uvManagerArray = unsafePerformIO $ do numCaps <- getNumCapabilities diff --git a/test/Z/IO/ResourceSpec.hs b/test/Z/IO/ResourceSpec.hs --- a/test/Z/IO/ResourceSpec.hs +++ b/test/Z/IO/ResourceSpec.hs @@ -21,27 +21,18 @@ workerCounter <- newCounter 0 let res = initResource (atomicAddCounter_ resCounter 1) (\ _ -> atomicSubCounter_ resCounter 1) - resPool = initPool res 100 1 + resPool = initSimplePool res 100 1 + R.withResource resPool $ \ pool -> do - forM_ [1..200] $ \ k -> forkIO. R.withResourceInPool pool $ \ i -> do + forM_ [1..200] $ \ k -> forkIO. R.withSimplePool pool $ \ i -> do atomicAddCounter_ workerCounter 1 - r <- readPrimIORef resCounter - assertEqual "pool should limit max usage" True (r <= 100) threadDelay 100000 threadDelay 1000000 - -- first 100 worker quickly get resources - -- then hold for 1s, rest 100 worker have to wait, and so on - -- so here we wait for 5s to make sure every worker got a resource - -- we used to use replicateConcurrently_ from async, but it's - -- not really neccessary r <- readPrimIORef resCounter assertEqual "pool should keep returned resources alive" 100 r - s <- poolStat pool - assertEqual "pool should be scanning returned resources" PoolScanning s - threadDelay 5000000 -- after 5s, 200 thread should release all resources w <- readPrimIORef workerCounter @@ -50,28 +41,19 @@ r <- readPrimIORef resCounter assertEqual "pool should reap unused resources" 0 r - s <- poolStat pool - assertEqual "pool should stop scanning returned resources" PoolEmpty s - -- Let's test again writePrimIORef workerCounter 0 - forM_ [1..200] $ \ k -> forkIO. R.withResourceInPool pool $ \ i -> do + forM_ [1..200] $ \ k -> forkIO. R.withSimplePool pool $ \ i -> do atomicAddCounter_ workerCounter 1 - r <- readPrimIORef resCounter - assertEqual "pool should limit max usage" True (r <= 100) threadDelay 100000 - threadDelay 1000000 r <- readPrimIORef resCounter assertEqual "pool should keep returned resources alive" 100 r - s <- poolStat pool - assertEqual "pool should be scanning returned resources" PoolScanning s - threadDelay 5000000 w <- readPrimIORef workerCounter @@ -80,31 +62,20 @@ r <- readPrimIORef resCounter assertEqual "pool should reap unused resources" 0 r - s <- poolStat pool - assertEqual "pool should stop scanning returned resources" PoolEmpty s - it "resource pool under exceptions" $ do resCounter <- newCounter 0 let res = initResource (atomicAddCounter' resCounter 1) (\ _ -> atomicSubCounter_ resCounter 1) - resPool = initPool res 100 1 + resPool = initSimplePool res 100 1 R.withResource resPool $ \ pool -> do - forM_ [1..200] $ \ k -> forkIO. R.withResourceInPool pool $ \ i -> do - r <- readPrimIORef resCounter + forM_ [1..200] $ \ k -> forkIO. R.withSimplePool pool $ \ i -> do threadDelay 100000 when (even i) (throwIO WorkerException) - assertEqual "pool should limit max usage" True (r <= 100) threadDelay 1000000 - s <- poolStat pool - assertEqual "pool should be scanning returned resources" PoolScanning s - threadDelay 5000000 r <- readPrimIORef resCounter assertEqual "pool should reap unused resources" 0 r - - s <- poolStat pool - assertEqual "pool should stop scanning returned resources" PoolEmpty s