Z-IO 0.4.0.0 → 0.5.0.0

raw patch · 15 files changed

+252/−160 lines, 15 filesdep ~Z-DataPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: Z-Data

API changes (from Hackage documentation)

Files

• ChangeLog.md +7/−0
• LICENSE +1/−1
• Z-IO.cabal +3/−4
• Z/IO/BIO.hs +20/−17
• Z/IO/Buffered.hs +33/−6
• Z/IO/Exception.hs +10/−20
• Z/IO/FileSystem.hs +2/−2
• Z/IO/FileSystem/Threaded.hs +2/−2
• Z/IO/Logger.hs +56/−15
• Z/IO/LowResTimer.hs +25/−0
• Z/IO/Network/IPC.hs +27/−25
• Z/IO/Network/TCP.hs +32/−30
• Z/IO/Resource.hs +16/−16
• Z/IO/UV/Manager.hs +10/−9
• test/Z/IO/ResourceSpec.hs +8/−13

@@ -1,5 +1,12 @@ # Revision history for Z-IO +## 0.5.0.0  -- 2020-01-28++* Add `unwrap/unwrap'` to `Z.IO.Exception`.+* Add `readParseChunks` to `Z.IO.Buffered`, Change `readParser`'s type to match `readParseChunks`.+* Add `sourceParseChunksBufferedInput`, `sourceParseChunksInput` to `Z.IO.BIO`.+* Add `newJSONLogger/defaultJSONFmt` to `Z.IO.Logger`, provide simple JSON structured logging. + ## 0.3.0.0  -- 2020-12-29  * Add `getSystemTime'` to `Z.IO.Time`.

@@ -1,4 +1,4 @@-Copyright (c) 2017-2020, Haskell Z Team+Copyright (c) Project Z Contributors, 2017-2020  All rights reserved. 

@@ -1,6 +1,6 @@ cabal-version:              2.4 name:                       Z-IO-version:                    0.4.0.0+version:                    0.5.0.0 synopsis:                   Simple and high performance IO toolkit for Haskell description:                Simple and high performance IO toolkit for Haskell, including                             file system, network, ipc and more!@@ -8,8 +8,7 @@ license-file:               LICENSE author:                     Dong Han, Tao He maintainer:                 winterland1989@gmail.com-copyright:                  (c) Dong Han, 2017-2020-                            (c) Tao He, 2017-2019+copyright:                  (c) Project Z Contributors category:                   Data build-type:                 Simple homepage:                   https://github.com/haskell-Z/Z-IO@@ -126,7 +125,7 @@                           , stm                     == 2.5.*                           , unordered-containers    == 0.2.*                           , containers              == 0.6.*-                          , Z-Data                  == 0.4.* +                          , Z-Data                  == 0.5.*                            , time                    >= 1.9 && <= 2.0                           , unix-time               >= 0.4.7 && <= 0.5 

@@ -64,7 +64,8 @@   , sourceFromBuffered, sourceFromInput   , sourceTextFromBuffered, sourceTextFromInput   , sourceJSONFromBuffered, sourceJSONFromInput-  , sourceParsedBufferInput, sourceParsedInput+  , sourceParserBufferInput, sourceParserInput+  , sourceParseChunksBufferedInput, sourceParseChunksInput   -- ** Sink   , sinkToList   , sinkToBuffered@@ -454,25 +455,22 @@ -- Throw 'OtherError' with name "EJSON" if JSON value is not parsed or converted. sourceJSONFromBuffered :: forall a. (JSON.JSON a, HasCallStack) => BufferedInput -> Source a {-# INLINABLE sourceJSONFromBuffered #-}-sourceJSONFromBuffered bi = BIO{ pull = do-    bs <- readBuffer bi-    if V.null bs-       then return Nothing-       else do-           (rest, r) <- JSON.decodeChunks (readBuffer bi) bs-           unReadBuffer rest bi-           case r of Right v -> return (Just v)-                     Left e  -> throwOtherError "EJSON" (T.toText e) }+sourceJSONFromBuffered = sourceParseChunksBufferedInput JSON.decodeChunks  -- | Turn buffered input device into a packet source, throw 'OtherError' with name @EPARSE@ if parsing fail.-sourceParsedBufferInput :: HasCallStack => P.Parser a -> BufferedInput -> Source a-{-# INLINABLE sourceParsedBufferInput #-}-sourceParsedBufferInput p bi = BIO{ pull = do+sourceParserBufferInput :: HasCallStack => P.Parser a -> BufferedInput -> Source a+{-# INLINABLE sourceParserBufferInput #-}+sourceParserBufferInput p = sourceParseChunksBufferedInput (P.parseChunks p)++-- | Turn buffered input device into a packet source, throw 'OtherError' with name @EPARSE@ if parsing fail.+sourceParseChunksBufferedInput :: (HasCallStack, T.Print e) => P.ParseChunks IO V.Bytes e a -> BufferedInput -> Source a+{-# INLINABLE sourceParseChunksBufferedInput #-}+sourceParseChunksBufferedInput cp bi = BIO{ pull = do     bs <- readBuffer bi     if V.null bs        then return Nothing        else do-           (rest, r) <- P.parseChunks p (readBuffer bi) bs+           (rest, r) <- cp (readBuffer bi) bs            unReadBuffer rest bi            case r of Right v -> return (Just v)                      Left e  -> throwOtherError "EPARSE" (T.toText e) }@@ -502,9 +500,14 @@     liftIO (sourceFromInput f)  -- | Turn input device into a packet source.-sourceParsedInput :: (Input i, HasCallStack) => P.Parser a -> i -> IO (Source a)-{-# INLINABLE sourceParsedInput #-}-sourceParsedInput p i = sourceParsedBufferInput p <$> newBufferedInput i+sourceParserInput :: (Input i, HasCallStack) => P.Parser a -> i -> IO (Source a)+{-# INLINABLE sourceParserInput #-}+sourceParserInput p i = sourceParserBufferInput p <$> newBufferedInput i++-- | Turn input device into a packet source.+sourceParseChunksInput :: (T.Print e, Input i, HasCallStack) => P.ParseChunks IO V.Bytes e a -> i -> IO (Source a)+{-# INLINABLE sourceParseChunksInput #-}+sourceParseChunksInput p i = sourceParseChunksBufferedInput p <$> newBufferedInput i  -------------------------------------------------------------------------------- -- Sink

@@ -22,6 +22,7 @@   , readBuffer, readBufferText   , unReadBuffer   , readParser+  , readParseChunks   , readExactly   , readToMagic   , readLine@@ -101,10 +102,12 @@  -- | Open a new buffered input with 'V.defaultChunkSize' as buffer size. newBufferedInput :: Input i => i -> IO BufferedInput+{-# INLINABLE newBufferedInput #-} newBufferedInput = newBufferedInput' V.defaultChunkSize  -- | Open a new buffered output with 'V.defaultChunkSize' as buffer size. newBufferedOutput :: Output o => o -> IO BufferedOutput+{-# INLINABLE newBufferedOutput #-} newBufferedOutput = newBufferedOutput' V.defaultChunkSize  -- | Open a new buffered output with given buffer size, e.g. 'V.defaultChunkSize'.@@ -112,6 +115,7 @@                    => Int    -- ^ Output buffer size                    -> o                    -> IO BufferedOutput+{-# INLINABLE newBufferedInput' #-} newBufferedOutput' bufSiz o = do     index <- newPrimIORef 0     buf <- newPinnedPrimArray (max bufSiz 0)@@ -122,6 +126,7 @@                   => Int     -- ^ Input buffer size                   -> i                   -> IO BufferedInput+{-# INLINABLE newBufferedOutput' #-} newBufferedInput' bufSiz i = do     pb <- newIORef V.empty     buf <- newPinnedPrimArray (max bufSiz 0)@@ -138,6 +143,7 @@ -- otherwise we copy buffer into result and reuse buffer afterward. -- readBuffer :: HasCallStack => BufferedInput -> IO V.Bytes+{-# INLINABLE readBuffer #-} readBuffer BufferedInput{..} = do     pb <- readIORef bufPushBack     if V.null pb@@ -168,6 +174,7 @@ -- trailing bytes before EOF, an 'OtherError' with name 'EINCOMPLETE' will be thrown, if there're -- invalid UTF8 bytes, an 'OtherError' with name 'EINVALIDUTF8' will be thrown.` readBufferText :: HasCallStack => BufferedInput -> IO T.Text+{-# INLINABLE readBufferText #-} readBufferText BufferedInput{..} = do     pb <- readIORef bufPushBack     rbuf <- readIORef inputBuffer@@ -226,6 +233,7 @@ -- -- If EOF reached before N bytes read, an 'OtherError' with name 'EINCOMPLETE' will be thrown. readExactly :: HasCallStack => Int -> BufferedInput -> IO V.Bytes+{-# INLINABLE readExactly #-} readExactly n0 h0 = V.concat `fmap` (go h0 n0)   where     go h n = do@@ -249,6 +257,7 @@ -- This function will loop read until meet EOF('Input' device return 'V.empty'), -- Useful for reading small file into memory. readAll :: HasCallStack => BufferedInput -> IO [V.Bytes]+{-# INLINABLE readAll #-} readAll h = loop []   where     loop acc = do@@ -262,27 +271,40 @@ -- This function will loop read until meet EOF('Input' device return 'V.empty'), -- Useful for reading small file into memory. readAll' :: HasCallStack => BufferedInput -> IO V.Bytes+{-# INLINABLE readAll' #-} readAll' i = V.concat <$> readAll i  -- | Push bytes back into buffer(if not empty). -- unReadBuffer :: HasCallStack => V.Bytes -> BufferedInput -> IO ()+{-# INLINABLE unReadBuffer #-} unReadBuffer pb' BufferedInput{..} = unless (V.null pb') $ do     modifyIORef' bufPushBack (\ pb -> pb' `V.append` pb) --- | Read buffer and parse with 'Parser'.+-- | Read buffer and parse with 'P.ParseChunks'. -- -- This function will continuously draw data from input before parsing finish. Unconsumed -- bytes will be returned to buffer. ----- Either during parsing or before parsing, reach EOF will result in 'P.ParseError'.-readParser :: HasCallStack => P.Parser a -> BufferedInput -> IO (Either P.ParseError a)-readParser p i = do+-- Throw 'OtherError' with name @EPARSE@ if parsing failed.+readParseChunks :: (T.Print e, HasCallStack) => P.ParseChunks IO V.Bytes e a -> BufferedInput -> IO a+{-# INLINABLE readParseChunks #-}+readParseChunks cp i = do     bs <- readBuffer i-    (rest, r) <- P.parseChunks p (readBuffer i) bs+    (rest, r) <- cp (readBuffer i) bs     unReadBuffer rest i-    return r+    unwrap "EPARSE" r +-- | Read buffer and parse with 'P.Parser'.+--+-- This function will continuously draw data from input before parsing finish. Unconsumed+-- bytes will be returned to buffer.+--+-- Throw 'OtherError' with name @EPARSE@ if parsing failed.+readParser :: HasCallStack => P.Parser a -> BufferedInput -> IO a+{-# INLINABLE readParser #-}+readParser = readParseChunks . P.parseChunks+ {-| Read until reach a magic bytes, return bytes(including the magic bytes).  Empty bytes indicate EOF. if EOF is reached before meet a magic byte, partial bytes are returned.@@ -295,6 +317,7 @@ @ -} readToMagic :: HasCallStack => Word8 -> BufferedInput -> IO V.Bytes+{-# INLINABLE readToMagic #-} readToMagic magic0 h0 = V.concat <$> go h0 magic0   where     go h magic = do@@ -323,6 +346,7 @@ @ -} readLine :: HasCallStack => BufferedInput -> IO (Maybe V.Bytes)+{-# INLINABLE readLine #-} readLine i = do     bs@(V.PrimVector arr s l) <- readToMagic 10 i     if l == 0@@ -344,6 +368,7 @@ --   write buffer first, then try again. -- writeBuffer :: HasCallStack => BufferedOutput -> V.Bytes -> IO ()+{-# INLINABLE writeBuffer #-} writeBuffer o@BufferedOutput{..} v@(V.PrimVector ba s l) = do     i <- readPrimIORef bufIndex     bufSiz <- getSizeofMutablePrimArray outputBuffer@@ -372,6 +397,7 @@ -- Run 'B.Builder' with buffer if it can hold, write to device when buffer is full. -- writeBuilder :: HasCallStack => BufferedOutput -> B.Builder a -> IO ()+{-# INLINABLE writeBuilder #-} writeBuilder BufferedOutput{..} (B.Builder b) = do     i <- readPrimIORef bufIndex     originBufSiz <- getSizeofMutablePrimArray outputBuffer@@ -416,6 +442,7 @@ -- | Flush the buffer into output device(if buffer is not empty). -- flushBuffer :: HasCallStack => BufferedOutput -> IO ()+{-# INLINABLE flushBuffer #-} flushBuffer BufferedOutput{..} = do     i <- readPrimIORef bufIndex     when (i /= 0) $ do

@@ -65,10 +65,8 @@   , throwECLOSEDSTM   , throwUVError   , throwOtherError-    -- * Unwrapping Either/ / Maybe result-  , UnwrapException(..)   , unwrap-  , unwrapJust+  , unwrap'     -- * Re-exports   , module Control.Exception   , HasCallStack@@ -194,25 +192,17 @@  -------------------------------------------------------------------------------- --- | Exception for 'unwrap' \/ 'unwrapJust'.-data UnwrapException e-    = UnwrapEitherException CallStack e-    | UnwrapMaybeException CallStack-    deriving Show--instance (Typeable e, Show e) => Exception (UnwrapException e)---- | Try to unwrap a value from 'Right', throw @UnwrapException e@ if 'Left e'.-unwrap :: (HasCallStack, Show e, Typeable e) => Either e a -> IO a+-- | Try to unwrap a value from 'Right', throw @OtherError name desc@ with @desc == toText e@ if 'Left e'.+unwrap :: (HasCallStack, T.Print e) => T.Text -> Either e a -> IO a {-# INLINABLE unwrap #-}-unwrap (Right x) = return x-unwrap (Left e) = throwIO (UnwrapEitherException callStack e)+unwrap _ (Right x) = return x+unwrap n (Left e)  = throwOtherError n (T.toText e) --- | Try to unwrap a value from 'Just', throw @UnwrapException ()@ if 'Nothing'.-unwrapJust :: HasCallStack => Maybe a -> IO a-{-# INLINABLE unwrapJust #-}-unwrapJust (Just x) = return x-unwrapJust Nothing = throwIO (UnwrapMaybeException callStack :: UnwrapException ())+-- | Try to unwrap a value from 'Just', throw @OtherError name desc@ if 'Nothing'.+unwrap' :: HasCallStack => T.Text -> T.Text -> Maybe a -> IO a+{-# INLINABLE unwrap' #-}+unwrap' _ _ (Just x) = return x+unwrap' n d Nothing = throwOtherError n d  -------------------------------------------------------------------------------- 

@@ -295,9 +295,9 @@ writeTextFile filename content = writeFile filename (T.getUTF8Bytes content)  -- | Quickly open a file and read its content as a JSON value.--- Throw 'OtherError' with name @EJSON@ if JSON value is not parsed or converted.+-- Throw 'OtherError' with name @EPARSE@ if JSON value is not parsed. readJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> IO a-readJSONFile filename = unwrap . JSON.decode' =<< readFile filename+readJSONFile filename = unwrap "EPARSE" . JSON.decode' =<< readFile filename  -- | Quickly open a file and write a JSON Value. writeJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> a -> IO ()

@@ -311,9 +311,9 @@ writeTextFile filename content = writeFile filename (T.getUTF8Bytes content)  -- | Quickly open a file and read its content as a JSON value.--- Throw 'OtherError' with name @EJSON@ if JSON value is not parsed or converted.+-- Throw 'OtherError' with name @EPARSE@ if JSON value is not parsed. readJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> IO a-readJSONFile filename = unwrap . JSON.decode' =<< readFile filename+readJSONFile filename = unwrap "EPARSE" . JSON.decode' =<< readFile filename  -- | Quickly open a file and write a JSON Value. writeJSONFile :: (HasCallStack, JSON.JSON a) => CBytes -> a -> IO ()

@@ -46,6 +46,7 @@   , withDefaultLogger   , newLogger   , newColoredLogger+  , newJSONLogger     -- * logging functions   , debug   , info@@ -63,7 +64,7 @@   , defaultTSCache   , defaultFmtCallStack   , defaultLevelFmt-  , LogFormatter, defaultFmt, coloredFmt+  , LogFormatter, defaultFmt, defaultColoredFmt, defaultJSONFmt   , pushLogIORef, flushLogIORef     -- * Constants     -- ** Level@@ -79,9 +80,13 @@ import           Control.Concurrent.MVar import           Control.Monad import           Data.IORef+import           Foreign.C.Types         (CInt(..))+import           GHC.Conc.Sync           (ThreadId(..), myThreadId)+import           GHC.Exts                (ThreadId#) import           GHC.Stack import           System.IO.Unsafe        (unsafePerformIO) import qualified Z.Data.Builder          as B+import qualified Z.Data.JSON.Builder     as JB import qualified Z.Data.CBytes           as CB import           Z.Data.Vector.Base      as V import           Z.IO.Buffered@@ -95,6 +100,7 @@                   -> Level                  -- ^ log level                   -> B.Builder ()           -- ^ log content                   -> CallStack              -- ^ call stack trace+                  -> ThreadId               -- ^ logging thread id                   -> B.Builder ()  -- | Extensible logger type.@@ -141,8 +147,22 @@         t <- getSystemTime'         CB.toBuilder <$> formatSystemTime iso8061DateFormat t --- | Make a new simple logger.+-- | Make a new colored logger(connected to stderr). --+-- This logger will output colorized log if stderr is connected to TTY.+newColoredLogger :: LoggerConfig -> IO Logger+newColoredLogger LoggerConfig{..} = do+    logsRef <- newIORef []+    let flush = flushLogIORef stderrBuf logsRef+    throttledFlush <- throttleTrailing_ loggerMinFlushInterval flush+    return $ Logger (pushLogIORef logsRef loggerLineBufSize)+                    flush throttledFlush defaultTSCache+                    (if isStdStreamTTY stderr then defaultColoredFmt+                                              else defaultFmt)+                    loggerConfigLevel++-- | Make a new simple logger, see 'defaultFmt'.+-- newLogger :: LoggerConfig           -> MVar BufferedOutput           -> IO Logger@@ -154,18 +174,17 @@                     flush throttledFlush defaultTSCache defaultFmt                     loggerConfigLevel --- | Make a new colored logger(connected to stderr).+-- | Make a new structured JSON logger, see 'defaultJSONFmt' ----- This logger will output colorized log if stderr is connected to TTY.-newColoredLogger :: LoggerConfig -> IO Logger-newColoredLogger LoggerConfig{..} = do+newJSONLogger :: LoggerConfig+              -> MVar BufferedOutput+              -> IO Logger+newJSONLogger LoggerConfig{..} oLock = do     logsRef <- newIORef []-    let flush = flushLogIORef stderrBuf logsRef+    let flush = flushLogIORef oLock logsRef     throttledFlush <- throttleTrailing_ loggerMinFlushInterval flush     return $ Logger (pushLogIORef logsRef loggerLineBufSize)-                    flush throttledFlush defaultTSCache-                    (if isStdStreamTTY stderr then coloredFmt-                                              else defaultFmt)+                    flush throttledFlush defaultTSCache defaultJSONFmt                     loggerConfigLevel  -- | Use 'pushLogIORef' and 'pushLogIORef' to implement a simple 'IORef' based concurrent logger.@@ -196,20 +215,21 @@  -- | A default log formatter ----- @ [DEBUG][2020-10-09T07:44:14UTC][<interactive>:7:1]This a debug message\\n@+-- @[FATAL][2021-02-01T15:03:30+0800][<interactive>:31:1][thread#669]...@ defaultFmt :: LogFormatter-defaultFmt ts level content cstack = do+defaultFmt ts level content cstack (ThreadId tid#) = do     B.square (defaultLevelFmt level)     B.square ts     B.square $ defaultFmtCallStack cstack+    B.square $ "thread#" >> B.int (getThreadId tid#)     content     B.char8 '\n'  -- | A default colored log formatter -- -- DEBUG level is 'Cyan', WARNING level is 'Yellow', FATAL and CRITICAL level are 'Red'.-coloredFmt :: LogFormatter-coloredFmt ts level content cstack = do+defaultColoredFmt :: LogFormatter+defaultColoredFmt ts level content cstack (ThreadId tid#) = do     let blevel = defaultLevelFmt level     B.square (case level of         DEBUG    -> color Cyan blevel@@ -219,9 +239,27 @@         _        -> blevel)     B.square ts     B.square $ defaultFmtCallStack cstack+    B.square $ "thread#" >> B.int (getThreadId tid#)     content     B.char8 '\n' +-- | A default JSON log formatter.+--+-- @{"level":"FATAL","time":"2021-02-01T15:02:19+0800","loc":"<interactive>:27:1","theadId":606,"content":"..."}\\n@+defaultJSONFmt :: LogFormatter+defaultJSONFmt ts level content cstack (ThreadId tid#) = do+    B.curly $ do+        "level" `JB.kv` B.quotes (defaultLevelFmt level)+        B.comma+        "time" `JB.kv` B.quotes ts+        B.comma+        "loc" `JB.kv` B.quotes (defaultFmtCallStack cstack)+        B.comma+        "thead" `JB.kv`  B.int (getThreadId tid#)+        B.comma+        "content" `JB.kv` JB.string (B.unsafeBuildText content)+    B.char8 '\n'+ -- | Default stack formatter which fetch the logging source and location. defaultFmtCallStack :: CallStack -> B.Builder () defaultFmtCallStack cs =@@ -365,7 +403,10 @@ otherLevelTo_ :: Level -> Bool -> CallStack -> Logger -> B.Builder () -> IO () otherLevelTo_ level flushNow cstack logger bu = when (level >= loggerLevel logger) $ do     ts <- loggerTSCache logger-    (loggerPushBuilder logger) $ (loggerFmt logger) ts level bu cstack+    tid <- myThreadId+    (loggerPushBuilder logger) $ (loggerFmt logger) ts level bu cstack tid     if flushNow     then flushLogger logger     else flushLoggerThrottled logger++foreign import ccall unsafe "rts_getThreadId" getThreadId :: ThreadId# -> CInt

@@ -28,6 +28,7 @@   , cancelLowResTimer_   , timeoutLowRes   , timeoutLowResEx+  , threadDelayLowRes   , throttle   , throttle_   , throttleTrailing_@@ -121,6 +122,7 @@ -- | Get a 'LowResTimerManager' for current thread. -- getLowResTimerManager :: IO LowResTimerManager+{-# INLINABLE getLowResTimerManager #-} getLowResTimerManager = do     (cap, _) <- threadCapability =<< myThreadId     lrtmArray <- readIORef lowResTimerManager@@ -131,6 +133,7 @@ -- This is mostly for testing purpose. -- isLowResTimerManagerRunning :: LowResTimerManager -> IO Bool+{-# INLINABLE isLowResTimerManagerRunning #-} isLowResTimerManagerRunning (LowResTimerManager _ _ _ runningLock) = readMVar runningLock  -- | Register a new timer on current capability's timer manager, start the timing wheel if it's not turning.@@ -144,6 +147,7 @@ registerLowResTimer :: Int          -- ^ timeout in unit of 0.1s                     -> IO ()        -- ^ the action you want to perform, it should not block                     -> IO LowResTimer+{-# INLINABLE registerLowResTimer #-} registerLowResTimer t action = do     lrtm <- getLowResTimerManager     registerLowResTimerOn lrtm t action@@ -152,6 +156,7 @@ registerLowResTimer_ :: Int          -- ^ timeout in unit of 0.1s                      -> IO ()        -- ^ the action you want to perform, it should not block                      -> IO ()+{-# INLINABLE registerLowResTimer_ #-} registerLowResTimer_ t action = void (registerLowResTimer t action)  -- | Same as 'registerLowResTimer', but allow you choose timer manager.@@ -160,6 +165,7 @@                       -> Int          -- ^ timeout in unit of 0.1s                       -> IO ()        -- ^ the action you want to perform, it should not block                       -> IO LowResTimer+{-# INLINABLE registerLowResTimerOn #-} registerLowResTimerOn lrtm@(LowResTimerManager queue indexLock regCounter _) t action = do      let (round_, tick) = (max 0 t) `quotRem` queueSize@@ -186,6 +192,7 @@ -- A return value <= 0 indictate the timer is firing or fired. -- queryLowResTimer :: LowResTimer -> IO Int+{-# INLINABLE queryLowResTimer #-} queryLowResTimer (LowResTimer c) = readPrimIORef c  -- | Cancel a timer, return the remaining ticks.@@ -193,11 +200,13 @@ -- This function have no effect after the timer is fired. -- cancelLowResTimer :: LowResTimer -> IO Int+{-# INLINABLE cancelLowResTimer #-} cancelLowResTimer (LowResTimer c) = atomicOrCounter c (-1)  -- | @void . cancelLowResTimer@ -- cancelLowResTimer_ :: LowResTimer -> IO ()+{-# INLINABLE cancelLowResTimer_ #-} cancelLowResTimer_ = void . cancelLowResTimer  -- | similar to 'System.Timeout.timeout', this function put a limit on time which an IO can consume.@@ -208,6 +217,7 @@ timeoutLowRes :: Int    -- ^ timeout in unit of 0.1s               -> IO a               -> IO (Maybe a)+{-# INLINABLE timeoutLowRes #-} timeoutLowRes timeo io = do     mid <- myThreadId     catch@@ -225,6 +235,7 @@                 => Int    -- ^ timeout in unit of 0.1s                 -> IO a                 -> IO a+{-# INLINABLE timeoutLowResEx #-} timeoutLowResEx timeo io = do     mid <- myThreadId     timer <- registerLowResTimer timeo (timeoutAThread mid)@@ -238,10 +249,21 @@ data TimeOutException = TimeOutException ThreadId CallStack deriving Show instance Exception TimeOutException ++-- | Similiar to 'threadDelay', suspends the current thread for a given number of deciseconds.+--+threadDelayLowRes :: Int -> IO ()+{-# INLINABLE threadDelayLowRes #-}+threadDelayLowRes dsecs = mask_ $ do+    m <- newEmptyMVar+    t <- registerLowResTimer dsecs (putMVar m ())+    takeMVar m `onException` cancelLowResTimer_ t+ -------------------------------------------------------------------------------- -- | Check if low resolution timer manager loop is running, start loop if not. -- ensureLowResTimerManager :: LowResTimerManager -> IO ()+{-# INLINABLE ensureLowResTimerManager #-} ensureLowResTimerManager lrtm@(LowResTimerManager _ _ _ runningLock) = do     modifyMVar_ runningLock $ \ running -> do         unless running $ do@@ -321,6 +343,7 @@ throttle :: Int         -- ^ cache time in unit of 0.1s          -> IO a        -- ^ the original IO action          -> IO (IO a)   -- ^ throttled IO action+{-# INLINABLE throttle #-} throttle t action = do     resultCounter <- newCounter 0     resultRef <- newIORef =<< action@@ -343,6 +366,7 @@ throttle_ :: Int            -- ^ cache time in unit of 0.1s           -> IO ()          -- ^ the original IO action           -> IO (IO ())     -- ^ throttled IO action+{-# INLINABLE throttle_ #-} throttle_ t action = do     resultCounter <- newCounter 0     return $ do@@ -360,6 +384,7 @@ throttleTrailing_ :: Int                   -> IO ()        -- ^ the original IO action                   -> IO (IO ())   -- ^ throttled IO action+{-# INLINABLE throttleTrailing_ #-} throttleTrailing_ t action = do     resultCounter <- newCounter 0     return $ do

@@ -157,34 +157,36 @@ -- Step 3. -- After uv loop finishes, if we got some FDs, copy the FD buffer, fetch accepted FDs and fork worker threads. -                    -- we lock uv manager here in case of next uv_run overwrite current accept buffer-                    acceptBufCopy <- withUVManager' serverUVManager $ do-                        _ <- tryTakeMVar m-                        acceptCountDown <- peekBufferSizeTable serverUVManager serverSlot-                        pokeBufferSizeTable serverUVManager serverSlot (backLog-1)+                    -- we shouldn't receive asycn exceptions here otherwise accepted FDs are not closed+                    mask_$ do+                        -- we lock uv manager here in case of next uv_run overwrite current accept buffer+                        acceptBufCopy <- withUVManager' serverUVManager $ do+                            _ <- tryTakeMVar m+                            acceptCountDown <- peekBufferSizeTable serverUVManager serverSlot+                            pokeBufferSizeTable serverUVManager serverSlot (backLog-1) -                        -- if acceptCountDown count to -1, we should resume on haskell side-                        when (acceptCountDown == -1) (hs_uv_listen_resume serverHandle)+                            -- if acceptCountDown count to -1, we should resume on haskell side+                            when (acceptCountDown == -1) (hs_uv_listen_resume serverHandle) -                        -- copy accepted FDs-                        let acceptCount = backLog - 1 - acceptCountDown-                        acceptBuf' <- newPrimArray acceptCount-                        copyMutablePrimArray acceptBuf' 0 acceptBuf (acceptCountDown+1) acceptCount-                        unsafeFreezePrimArray acceptBuf'+                            -- copy accepted FDs+                            let acceptCount = backLog - 1 - acceptCountDown+                            acceptBuf' <- newPrimArray acceptCount+                            copyMutablePrimArray acceptBuf' 0 acceptBuf (acceptCountDown+1) acceptCount+                            unsafeFreezePrimArray acceptBuf' -                    -- fork worker thread-                    forM_ [0..sizeofPrimArray acceptBufCopy-1] $ \ i -> do-                        let fd = indexPrimArray acceptBufCopy i-                        if fd < 0-                        -- minus fd indicate a server error and we should close server-                        then throwUVIfMinus_ (return fd)-                        -- It's important to use the worker thread's mananger instead of server's one!-                        else void . forkBa $ do-                            uvm <- getUVManager-                            withResource (initUVStream (\ loop hdl -> do-                                throwUVIfMinus_ (uv_pipe_init loop hdl 0)-                                throwUVIfMinus_ (uv_pipe_open hdl fd)) uvm) $ \ uvs -> do-                                ipcServerWorker uvs+                        -- fork worker thread+                        forM_ [0..sizeofPrimArray acceptBufCopy-1] $ \ i -> do+                            let fd = indexPrimArray acceptBufCopy i+                            if fd < 0+                            -- minus fd indicate a server error and we should close server+                            then throwUVIfMinus_ (return fd)+                            -- It's important to use the worker thread's mananger instead of server's one!+                            else void . forkBa $ do+                                uvm <- getUVManager+                                withResource (initUVStream (\ loop hdl -> do+                                    throwUVIfMinus_ (uv_pipe_init loop hdl 0)+                                    throwUVIfMinus_ (uv_pipe_open hdl fd)) uvm) $ \ uvs -> do+                                    ipcServerWorker uvs  -------------------------------------------------------------------------------- 

@@ -170,39 +170,41 @@ -- Step 3. -- After uv loop finishes, if we got some FDs, copy the FD buffer, fetch accepted FDs and fork worker threads. -                    -- we lock uv manager here in case of next uv_run overwrite current accept buffer-                    acceptBufCopy <- withUVManager' serverUVManager $ do-                        _ <- tryTakeMVar m-                        acceptCountDown <- peekBufferSizeTable serverUVManager serverSlot-                        pokeBufferSizeTable serverUVManager serverSlot (backLog-1)+                    -- we shouldn't receive asycn exceptions here otherwise accepted FDs are not closed+                    mask_$ do+                        -- we lock uv manager here in case of next uv_run overwrite current accept buffer+                        acceptBufCopy <- withUVManager' serverUVManager $ do+                            _ <- tryTakeMVar m+                            acceptCountDown <- peekBufferSizeTable serverUVManager serverSlot+                            pokeBufferSizeTable serverUVManager serverSlot (backLog-1) -                        -- if acceptCountDown count to -1, we should resume on haskell side-                        when (acceptCountDown == -1) (hs_uv_listen_resume serverHandle)+                            -- if acceptCountDown count to -1, we should resume on haskell side+                            when (acceptCountDown == -1) (hs_uv_listen_resume serverHandle) -                        -- copy accepted FDs-                        let acceptCount = backLog - 1 - acceptCountDown-                        acceptBuf' <- newPrimArray acceptCount-                        copyMutablePrimArray acceptBuf' 0 acceptBuf (acceptCountDown+1) acceptCount-                        unsafeFreezePrimArray acceptBuf'+                            -- copy accepted FDs+                            let acceptCount = backLog - 1 - acceptCountDown+                            acceptBuf' <- newPrimArray acceptCount+                            copyMutablePrimArray acceptBuf' 0 acceptBuf (acceptCountDown+1) acceptCount+                            unsafeFreezePrimArray acceptBuf' -                    -- fork worker thread-                    forM_ [0..sizeofPrimArray acceptBufCopy-1] $ \ i -> do-                        let fd = indexPrimArray acceptBufCopy i-                        if fd < 0-                        -- minus fd indicate a server error and we should close server-                        then throwUVIfMinus_ (return fd)-                        -- It's important to use the worker thread's mananger instead of server's one!-                        else void . forkBa $ do-                            uvm <- getUVManager-                            withResource (initUVStream (\ loop hdl -> do-                                throwUVIfMinus_ (uv_tcp_init loop hdl)-                                throwUVIfMinus_ (uv_tcp_open hdl fd)) uvm) $ \ uvs -> do-                                -- safe without withUVManager-                                when tcpServerWorkerNoDelay . throwUVIfMinus_ $-                                    uv_tcp_nodelay (uvsHandle uvs) 1-                                when (tcpServerWorkerKeepAlive > 0) . throwUVIfMinus_ $-                                    uv_tcp_keepalive (uvsHandle uvs) 1 tcpServerWorkerKeepAlive-                                tcpServerWorker uvs+                        -- fork worker thread+                        forM_ [0..sizeofPrimArray acceptBufCopy-1] $ \ i -> do+                            let fd = indexPrimArray acceptBufCopy i+                            if fd < 0+                            -- minus fd indicate a server error and we should close server+                            then throwUVIfMinus_ (return fd)+                            -- It's important to use the worker thread's mananger instead of server's one!+                            else void . forkBa $ do+                                uvm <- getUVManager+                                withResource (initUVStream (\ loop hdl -> do+                                    throwUVIfMinus_ (uv_tcp_init loop hdl)+                                    throwUVIfMinus_ (uv_tcp_open hdl fd)) uvm) $ \ uvs -> do+                                    -- safe without withUVManager+                                    when tcpServerWorkerNoDelay . throwUVIfMinus_ $+                                        uv_tcp_nodelay (uvsHandle uvs) 1+                                    when (tcpServerWorkerKeepAlive > 0) . throwUVIfMinus_ $+                                        uv_tcp_keepalive (uvsHandle uvs) 1 tcpServerWorkerKeepAlive+                                    tcpServerWorker uvs  -------------------------------------------------------------------------------- 

@@ -26,7 +26,6 @@   , Pool   , PoolState(..)   , initPool-  , initInPool   , withResourceInPool   , poolStat, poolInUse   -- * Re-export@@ -229,14 +228,23 @@ poolInUse :: Pool a -> IO Int poolInUse pool = readTVarIO (_poolInUse pool) --- | Obtain the pooled resource inside a given resource pool.------ You shouldn't use 'withResource' with this resource after you closed the pool,--- a 'ResourceVanished' will be thrown.+-- | Open resource inside a given resource pool and do some computation. ---initInPool :: Pool a -> Resource a-initInPool (Pool res limit itime entries inuse state) =-    fst <$> initResource takeFromPool returnToPool+-- This function is thread safe, concurrently usage will be guaranteed+-- to get different resource. If exception happens,+-- 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 =+    fst <$> MonadCatch.generalBracket+        (liftIO takeFromPool)+        (\ r@(_, close) exit ->+            case exit of+                MonadCatch.ExitCaseSuccess _ -> liftIO (returnToPool r)+                _ -> liftIO $ do+                    atomically $ modifyTVar' inuse (subtract 1)+                    close)+        (\ (a, _) -> k a)   where     takeFromPool = join . atomically $ do         c <- readTVar state@@ -291,11 +299,3 @@         | 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)---- | Open resource inside a given resource pool and do some computation.----withResourceInPool :: (MonadCatch.MonadMask m, MonadIO m, HasCallStack)-                   => Pool a -> (a -> m b) -> m b-withResourceInPool pool = withResource res-  where-    res = initInPool pool

@@ -191,7 +191,7 @@                 throwUVIfMinus_ (hs_uv_wake_up_async loopData)                 return Nothing             else do-                r <- f loop+                !r <- f loop                 return (Just r)         case r of             Just r' -> return r'@@ -219,14 +219,15 @@         then yield >> poll                              -- we yield here, to let other threads do actual work         else do                                         -- otherwise we still yield once             yield                                       -- in case other threads can still progress-            e' <- withMVar runningLock $ \ _ -> step uvm False   -- now we do another non-blocking poll to make sure-            if e' > 0 then yield >> poll             -- if we got events somehow, we yield and go back-            else do                                 -- if there's still no events, we directly jump to safe blocking poll-                _ <- swapMVar runningLock True          -- after swap this lock, other thread can wake up us-                _ <- step uvm True                  -- by send async handler, and it's thread safe+            e' <- withMVar runningLock $ \ _ -> step uvm False   -- now we have done another non-blocking poll+            if e' > 0 then yield >> poll            -- if we got events somehow, we yield and go back+            else do                                 -- if there're still no events,+                                                    -- we directly jump to safe blocking poll+                _ <- swapMVar runningLock True      -- after swap this lock, other threads have to wake up us+                _ <- step uvm True                  -- by send async handler, thus libuv's states are safe                 _ <- swapMVar runningLock False--                yield                               -- we yield here, to let other threads do actual work+                                                    -- blocking poll only exits if there're events,+                yield                               -- so we yield here, to let other threads do actual work                 poll      -- call uv_run, return the event number@@ -255,7 +256,7 @@                 -- After step finished, other threads are free to take the same slot,                 -- thus can overwrite the buffer size table, i.e. the previous result.                 ---                r <- peekBufferSizeTable uvm slot+                !r <- peekBufferSizeTable uvm slot                 tryPutMVar lock r             return c 

@@ -23,12 +23,11 @@                                (\ _ -> atomicSubCounter_ resCounter 1)             resPool = initPool res 100 1         R.withResource resPool $ \ pool -> do-            let res = initInPool pool-            forM_ [1..1000] $ \ k -> forkIO. R.withResource res $ \ i -> do+            forM_ [1..200] $ \ k -> forkIO. R.withResourceInPool pool $ \ i -> do                 atomicAddCounter_ workerCounter 1                 r <- readPrimIORef resCounter-                threadDelay 100000                 assertEqual "pool should limit max usage" True (r <= 100)+                threadDelay 100000              threadDelay 1000000             -- first 100 worker quickly get resources@@ -43,10 +42,10 @@             s <- poolStat pool             assertEqual "pool should be scanning returned resources" PoolScanning s -            threadDelay 5000000  -- after 5s, 1000 thread should release all resources+            threadDelay 5000000  -- after 5s, 200 thread should release all resources              w <- readPrimIORef workerCounter-            assertEqual "worker should be able to get resource" 1000 w+            assertEqual "worker should be able to get resource" 200 w              r <- readPrimIORef resCounter             assertEqual "pool should reap unused resources" 0 r@@ -58,11 +57,11 @@              writePrimIORef workerCounter 0 -            forM_ [1..1000] $ \ k -> forkIO. R.withResource res $ \ i -> do+            forM_ [1..200] $ \ k -> forkIO. R.withResourceInPool pool $ \ i -> do                 atomicAddCounter_ workerCounter 1                 r <- readPrimIORef resCounter-                threadDelay 100000                 assertEqual "pool should limit max usage" True (r <= 100)+                threadDelay 100000               threadDelay 1000000@@ -76,7 +75,7 @@             threadDelay 5000000              w <- readPrimIORef workerCounter-            assertEqual "worker should be able to get resource" 1000 w+            assertEqual "worker should be able to get resource" 200 w              r <- readPrimIORef resCounter             assertEqual "pool should reap unused resources" 0 r@@ -90,18 +89,14 @@                                (\ _ -> atomicSubCounter_ resCounter 1)             resPool = initPool res 100 1         R.withResource resPool $ \ pool -> do-            let res = initInPool pool -            forM_ [1..1000] $ \ k -> forkIO. R.withResource res $ \ i -> do+            forM_ [1..200] $ \ k -> forkIO. R.withResourceInPool pool $ \ i -> do                 r <- readPrimIORef resCounter                 threadDelay 100000                 when (even i) (throwIO WorkerException)                 assertEqual "pool should limit max usage" True (r <= 100)              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