Introduce chunking in the logs.

In the background logger we introduce chunking, a process when
we read a number of messages from the queue and process that chunk
and flush afterwards. It approach can lead to better performance,
for example if you have hLogStdout, and hFlush as a flush you may
may still use BufferredHandle instead of line one, without introducing
delays in writing logs.

Currently there are 2 ways to write consume log messages:
  1. old one (Just 1) in limit parameter of the Capacity - each message
     if written and processed one by one
  2. `Nothing` unbounded (actually bounded by the TQueue size)
  3. `Just i` bounded

Default choice is `Just 32`, it's an arbitrary decision that works in
practice for our company, but we have not spend much time investigating.

Author: qnikst

CHANGELOG.markdown

@@ -6,6 +6,8 @@
 
   - 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
 

co-log-concurrent.cabal

@@ -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/

src/Colog/Concurrent.hs

@@ -53,14 +53,15 @@ module Colog.Concurrent
          -- $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.Monad (forever, join)
 import Control.Monad.IO.Class (MonadIO (..))
 import Data.Foldable (for_)
+import Numeric.Natural (Natural)
 
 import Colog.Concurrent.Internal (BackgroundWorker (..), Capacity (..), mkCapacity)
 import Colog.Core.Action (LogAction (..))
@@ -140,12 +141,16 @@ See 'forkBackgroundLogger' for more details.
 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..."
@@ -156,16 +161,17 @@ withBackgroundLogger
     :: 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)
 defCapacity :: Capacity
-defCapacity = Capacity 4096
+defCapacity = Capacity 4096 (Just 32)
 
 
 {- $extended-api
@@ -218,19 +224,30 @@ __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.
 -}
-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 $ unLogAction logAction
+      flush)
     (\_ ->
        (do msgs <- atomically $ many $ readTBQueue queue
-           for_ msgs $ unLogAction logAction)
+           for_ msgs $ 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'
@@ -280,9 +297,13 @@ happening.
 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

src/Colog/Concurrent/Internal.hs

@@ -24,16 +24,24 @@ import Numeric.Natural (Natural)
 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
 
 -- | Create new capacity.
 --
 -- @since 0.5.0.0
-mkCapacity :: Natural -> Capacity
-mkCapacity = Capacity . fromIntegral
+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
 process.