diff --git a/ChangeLog.md b/ChangeLog.md --- a/ChangeLog.md +++ b/ChangeLog.md @@ -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`. diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright (c) 2017-2020, Haskell Z Team +Copyright (c) Project Z Contributors, 2017-2020 All rights reserved. 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.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 diff --git a/Z/IO/BIO.hs b/Z/IO/BIO.hs --- a/Z/IO/BIO.hs +++ b/Z/IO/BIO.hs @@ -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 diff --git a/Z/IO/Buffered.hs b/Z/IO/Buffered.hs --- a/Z/IO/Buffered.hs +++ b/Z/IO/Buffered.hs @@ -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 diff --git a/Z/IO/Exception.hs b/Z/IO/Exception.hs --- a/Z/IO/Exception.hs +++ b/Z/IO/Exception.hs @@ -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 -------------------------------------------------------------------------------- diff --git a/Z/IO/FileSystem.hs b/Z/IO/FileSystem.hs --- a/Z/IO/FileSystem.hs +++ b/Z/IO/FileSystem.hs @@ -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 () diff --git a/Z/IO/FileSystem/Threaded.hs b/Z/IO/FileSystem/Threaded.hs --- a/Z/IO/FileSystem/Threaded.hs +++ b/Z/IO/FileSystem/Threaded.hs @@ -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 () diff --git a/Z/IO/Logger.hs b/Z/IO/Logger.hs --- a/Z/IO/Logger.hs +++ b/Z/IO/Logger.hs @@ -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][:7:1]This a debug message\\n@ +-- @[FATAL][2021-02-01T15:03:30+0800][: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":":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 diff --git a/Z/IO/LowResTimer.hs b/Z/IO/LowResTimer.hs --- a/Z/IO/LowResTimer.hs +++ b/Z/IO/LowResTimer.hs @@ -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 diff --git a/Z/IO/Network/IPC.hs b/Z/IO/Network/IPC.hs --- a/Z/IO/Network/IPC.hs +++ b/Z/IO/Network/IPC.hs @@ -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 -------------------------------------------------------------------------------- diff --git a/Z/IO/Network/TCP.hs b/Z/IO/Network/TCP.hs --- a/Z/IO/Network/TCP.hs +++ b/Z/IO/Network/TCP.hs @@ -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 -------------------------------------------------------------------------------- diff --git a/Z/IO/Resource.hs b/Z/IO/Resource.hs --- a/Z/IO/Resource.hs +++ b/Z/IO/Resource.hs @@ -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 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 @@ -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 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 @@ -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