co-log-concurrent 0.4.0.0 → 0.5.0.0

raw patch · 5 files changed

+114/−25 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.markdown +9/−0
• README.markdown +1/−0
• co-log-concurrent.cabal +1/−1
• src/Colog/Concurrent.hs +86/−22
• src/Colog/Concurrent/Internal.hs +17/−2

@@ -1,5 +1,14 @@ # Revision history for co-log-concurent +## 0.5.0.0 --++* API changes:++  - Introduce mkCapacity function, as previously it was not possible+    to create custom capacity.+  - Fork background logger now takes an explicit flush action. And+    reads logs in chunks.+ ## 0.4.0.0 -- 2020-04-18  * Library extracted from co-log.

@@ -58,6 +58,7 @@ main =   withBackgroundLogger     defCapacity+    (pure ())     logByteStringStdout $ \log ->       usingLoggerT log $ do         logMsg @ByteString "Starting application..."

@@ -1,6 +1,6 @@ cabal-version:       2.4 name:                co-log-concurrent-version:             0.4.0.0+version:             0.5.0.0 synopsis:            Asynchronous backend for co-log library description:         Buiding block for writing asynchronous logger pipelines. homepage:            https://github.com/qnikst/co-log-concurrent/

@@ -13,14 +13,14 @@ potential tradeoffs and describe if individual building blocks are affected or not. -  1. _Unbounded memory usage_ - if there is no backpressure mechanism the user threads,+  1. __Unbound memory usage__ – if there is no backpressure mechanism the user threads,      threads may generate more logs that can we can store at the same amount of time.      In such cases messages are accumulated in memory. It extends GC times and memory usage.-  2. _Persistence requirements_ - sometimes application may want to ensure that+  2. __Persistence requirements__ – sometimes application may want to ensure that      we persisted the logs before it moved to the next statement. It is not a case with      concurrent log systems in general; some we lose logs even the thread moves forward.      It may happen when the application exits before dumping all logs.-  3. _Non-precise logging_ - sometimes there may be anomalies when storing logs,+  3. __Non-precise logging__ – sometimes there may be anomalies when storing logs,      such as logs reordering or imprecise timestamps.  In case if your application is a subject of those problems you may@@ -29,18 +29,24 @@ -}  module Colog.Concurrent-       ( -- $general+       ( +         -- $general+          -- * Simple API.          -- $simple-api          withBackgroundLogger        , defCapacity-         -- * Extended API+         -- * Exceptions in messages.+         -- $exceptions++         -- * Extended API.          -- $extended-api          -- ** Background worker          -- $background-worker        , BackgroundWorker        , backgroundWorkerWrite        , killBackgroundLogger+       , mkCapacity          -- ** Background logger        , forkBackgroundLogger        , convertToLogAction@@ -52,31 +58,32 @@          -- $worker-thread-usage        ) where -import Control.Applicative (many)+import Control.Applicative (many, (<|>), some) import Control.Concurrent (forkFinally, killThread)-import Control.Concurrent.STM (atomically, check, newTVarIO, readTVar, writeTVar)+import Control.Concurrent.STM (STM, atomically, check, newTVarIO, readTVar, writeTVar) import Control.Concurrent.STM.TBQueue (newTBQueueIO, readTBQueue, writeTBQueue)-import Control.Exception (bracket, finally)+import Control.Exception (bracket, finally, mask_) import Control.Monad (forever, join) import Control.Monad.IO.Class (MonadIO (..)) import Data.Foldable (for_)+import Numeric.Natural (Natural) -import Colog.Concurrent.Internal (BackgroundWorker (..), Capacity (..))+import Colog.Concurrent.Internal (BackgroundWorker (..), Capacity (..), mkCapacity) import Colog.Core.Action (LogAction (..))   {- $general Concurrent logger consists of the following building blocks (see schema below). -  1. *Logger in the application thread*. The application runs it in the main thread,+  1. __Logger in the application thread__. The application runs it in the main thread,       and it has access to all the thread state. This logger can work in any @m@.-  2. *Communication channel with backpressure support*. In addition to the channel,+  2. __Communication channel with backpressure support__. In addition to the channel,       we have a converter that puts the user message to the communication channel.       This converter works in the user thread. Such a logger usually works in 'IO',       but it's possible to make it work in 'Control.Concurrent.STM.STM' as well.       At this point, the library provides only 'IO' version, but it can be lifted       to any 'MonadIO' by the user.-  3. *Logger thread*. It's a background thread that performs an actual synchronous+  3. __Logger thread__. It's a background thread that performs an actual synchronous      write to the log sinks. Loggers there do not have access to the users' thread      state. @@ -116,6 +123,7 @@  In this approach, the application concurrently writes logs to the logger, then the logger concurrently writing to all sinks.+ -}  {- $simple-api@@ -125,6 +133,7 @@ versions of the library. But the guarantee that we give is that no matter what implementation is, it keeps with reasonable defaults and can be applied to a generic application.+ -}  {- | @@ -139,12 +148,16 @@ application state or thread info, so you should only pass methods that serialize and dump data there. +@IO ()@ - flush provides a function to flush all the logs, it allows flush logs+by chunks, so @LogAction@ may not care about flushing.+ @ main :: IO () main =   'withBackgroundLogger'      'defCapacity'      'Colog.Actions.logByteStringStdout'+     ('pure' ())      (\log -> 'Colog.Monad.usingLoggerT' log $ __do__         'Colog.Monad.logMsg' \@ByteString "Starting application..."         'Colog.Monad.logMsg' \@ByteString "Finishing application..."@@ -155,18 +168,47 @@     :: MonadIO m     => Capacity  -- ^ Capacity of messages to handle; bounded channel size     -> LogAction IO msg  -- ^ Action that will be used in a forked thread+    -> IO () -- ^ Action to flush logs     -> (LogAction m msg -> IO a)  -- ^ Continuation action     -> IO a-withBackgroundLogger cap logger action =-   bracket (forkBackgroundLogger cap logger)+withBackgroundLogger cap logger flush action =+   bracket (forkBackgroundLogger cap logger flush)            killBackgroundLogger            (action . convertToLogAction) --- | Default capacity size, (4096)+-- | Default capacity size.+--+-- * 4096 messages that in flight+-- * Up to 32 messages can be read in a single chunk defCapacity :: Capacity-defCapacity = Capacity 4096+defCapacity = Capacity 4096 (Just 32) +{- $exceptions+It worth discussing how the library handles exceptions in messages.  User generates a+message in the application thread, but it's guaranteed to be evaluated only in+the logger thread. It means that on the contrary to the synchronious logger it's+possible to store exception in the message and it another thread, so the application+thread will never see it. +@+let message = showt (1/0)+unLogger logger message+@++In synchronous case exception will be rised in the thread, but in asynchronous thread+exception will happen somewhere else.++Currently the library does not take any action to prevent that. The problem is that+there is no safe strategy that will work in general case. There are some approaches though:++  1. Use deepseq to fully evaluate message before passing it to the logger.+     Doesn't work for all types, leads to additional computations.+  2. Require message to be WHNF-strict and call `seq` before sending message to the logger.+  3. Serialize message in the user thread.+-}+++ {- $extended-api  Extended API explains how asynchronous logging is working and provides basic@@ -216,20 +258,38 @@ __N.B.__ On exit, even in case of exception thread will dump all values that are in the queue. But it will stop doing that in case if another exception will happen.++*Exception handing*. 'forkBackoundLogger' forks a private that and does+not expect that someone sends it an asynchronous exception. It means that+any asynchronous exception is treated as a command to stop the worker.+Logger is not aware of synchronous exceptions as well, so handing of the+application specific exceptions should be a concern of the logging action.+ -}-forkBackgroundLogger :: Capacity -> LogAction IO msg -> IO (BackgroundWorker msg)-forkBackgroundLogger (Capacity cap) logAction = do+forkBackgroundLogger :: Capacity -> LogAction IO msg -> IO () -> IO (BackgroundWorker msg)+forkBackgroundLogger (Capacity cap lim) logAction flush = do   queue <- newTBQueueIO cap   isAlive <- newTVarIO True   tid <- forkFinally     (forever $ do-      msg <- atomically $ readTBQueue queue-      unLogAction logAction msg)+      msgs <- atomically $ fetch $ readTBQueue queue+      for_ msgs $ mask_ . unLogAction logAction+      flush)     (\_ ->        (do msgs <- atomically $ many $ readTBQueue queue-           for_ msgs $ unLogAction logAction)+           for_ msgs $ mask_ . unLogAction logAction+           flush)          `finally` atomically (writeTVar isAlive False))   pure $ BackgroundWorker tid (writeTBQueue queue) isAlive+  where+    fetch +      | Just n <- lim = someN n+      | otherwise = some+    someN :: Natural -> STM a -> STM [a]+    someN 0 _ = pure []+    someN n f = (:) <$> f <*> go n where+      go 0 = pure []+      go k = ((:) <$> f <*> go (k-1)) <|> pure []   {- | Convert a given 'BackgroundWorker msg' into a 'LogAction msg'@@ -279,9 +339,13 @@ When closed it will dump all pending messages, unless another asynchronous exception will arrive, or synchronous exception will happen during the logging.++Note. Limit parameter of capacity is ignored here as the function+performs IO actions and seems that doesn't benefit from the chunking.+However it may change in the future versions if proved to be wrong. -} mkBackgroundThread :: Capacity -> IO (BackgroundWorker (IO ()))-mkBackgroundThread (Capacity cap) = do+mkBackgroundThread (Capacity cap _lim) = do   queue <- newTBQueueIO cap   isAlive <- newTVarIO True   tid <- forkFinally

@@ -12,6 +12,7 @@ module Colog.Concurrent.Internal        ( BackgroundWorker (..)        , Capacity (..)+       , mkCapacity        ) where  import Control.Concurrent (ThreadId)@@ -23,9 +24,23 @@ differrent for the different GHC versions. -} #if MIN_VERSION_stm(2,5,0)-newtype Capacity = Capacity Natural+data Capacity = Capacity Natural (Maybe Natural) #else-newtype Capacity = Capacity Int+data Capacity = Capacity Int (Maybe Natural)+#endif++-- | Creates new capacity.+--+-- @since 0.5.0.0+mkCapacity+  :: Natural -- ^ Size of the queue. Number of the messages in flight+  -> Maybe Natural -- ^ Maximum number of messages that logger can read in a chunk.+  -> Capacity+mkCapacity n = Capacity (mk n)  where+#if MIN_VERSION_stm(2,5,0)+  mk = id+#else+  mk = fromIntegral #endif  {- | Wrapper for the background thread that may receive messages to