Make the MergeBatchSize an adjustable parameter in TableConfig

Previously it was hard coded to be the same as the write buffer size.

Document what it means as a new tunable parameter.

Setting this low (1) is important for getting good parallel work balance
on the pipelined WP8 benchmark. It is a crucial change that makes the
pipelined version actually improve performance. Previously it would only
get about a 5 to 10% improvement.

Author: dcoutts

README.md

@@ -356,6 +356,12 @@ The *disk cache policy* determines if lookup operations use the OS page
 cache. Caching may improve the performance of lookups and updates if
 database access follows certain patterns.
 
+`confMergeBatchSize`  
+The merge batch size balances the maximum latency of individual update
+operations, versus the latency of a sequence of update operations.
+Bigger batches improves overall performance but some updates will take a
+lot longer than others. The default is to use a large batch size.
+
 ##### Fine-tuning: Merge Policy, Size Ratio, and Write Buffer Size <span id="fine_tuning_data_layout" class="anchor"></span>
 
 The configuration parameters `confMergePolicy`, `confSizeRatio`, and
@@ -647,6 +653,33 @@ locality if it is likely to access entries that have nearby keys.
   does not have good spatial or temporal locality. For instance, if the
   access pattern is uniformly random.
 
+##### Fine-tuning: Merge Batch Size <span id="fine_tuning_merge_batch_size" class="anchor"></span>
+
+The *merge batch size* is a micro-tuning parameter, and in most cases
+you do need to think about it and can leave it at its default.
+
+When using the `Incremental` merge schedule, merging is done in batches.
+This is a trade-off: larger batches tends to mean better overall
+performance but the downside is that while most updates (inserts,
+deletes, upserts) are fast, some are slower (when a batch of merging
+work has to be done).
+
+If you care most about the maximum latency of updates, then use a small
+batch size. If you don't care about latency of individual operations,
+just the latency of the overall sequence of operations then use a large
+batch size. The default is to use a large batch size, the same size as
+the write buffer itself. The minimum batch size is 1. The maximum batch
+size is the size of the write buffer `confWriteBufferAlloc`.
+
+Note that the actual batch size is the minimum of this configuration
+parameter and the size of the batch of operations performed (e.g.
+`inserts`). So if you consistently use large batches, you can use a
+batch size of 1 and the merge batch size will always be determined by
+the operation batch size.
+
+A further reason why it may be preferable to use minimal batch sizes is
+to get good parallel work balance, when using parallelism.
+
 ### References
 
 The implementation of LSM-trees in this package draws inspiration from:

lsm-tree.cabal

@@ -183,6 +183,12 @@ description:
       The /disk cache policy/ determines if lookup operations use the OS page cache.
       Caching may improve the performance of lookups and updates if database access follows certain patterns.
 
+  [@confMergeBatchSize@]
+      The merge batch size balances the maximum latency of individual update
+      operations, versus the latency of a sequence of update operations. Bigger
+      batches improves overall performance but some updates will take a lot
+      longer than others. The default is to use a large batch size.
+
   ==== Fine-tuning: Merge Policy, Size Ratio, and Write Buffer Size #fine_tuning_data_layout#
 
   The configuration parameters @confMergePolicy@, @confSizeRatio@, and @confWriteBufferAlloc@ affect how the table organises its data.
@@ -429,6 +435,31 @@ description:
   *   Use the @DiskCacheNone@ policy if the database's access pattern has does not have good spatial or temporal locality.
       For instance, if the access pattern is uniformly random.
 
+  ==== Fine-tuning: Merge Batch Size #fine_tuning_merge_batch_size#
+
+  The /merge batch size/ is a micro-tuning parameter, and in most cases you do
+  need to think about it and can leave it at its default.
+
+  When using the 'Incremental' merge schedule, merging is done in batches. This
+  is a trade-off: larger batches tends to mean better overall performance but the
+  downside is that while most updates (inserts, deletes, upserts) are fast, some
+  are slower (when a batch of merging work has to be done).
+
+  If you care most about the maximum latency of updates, then use a small batch
+  size. If you don't care about latency of individual operations, just the
+  latency of the overall sequence of operations then use a large batch size. The
+  default is to use a large batch size, the same size as the write buffer itself.
+  The minimum batch size is 1. The maximum batch size is the size of the write
+  buffer 'confWriteBufferAlloc'.
+
+  Note that the actual batch size is the minimum of this configuration
+  parameter and the size of the batch of operations performed (e.g. 'inserts').
+  So if you consistently use large batches, you can use a batch size of 1 and
+  the merge batch size will always be determined by the operation batch size.
+
+  A further reason why it may be preferable to use minimal batch sizes is to get
+  good parallel work balance, when using parallelism.
+
   == References
 
   The implementation of LSM-trees in this package draws inspiration from:

src-extras/Database/LSMTree/Extras/NoThunks.hs

@@ -659,6 +659,9 @@ deriving anyclass instance NoThunks DiskCachePolicy
 deriving stock instance Generic MergeSchedule
 deriving anyclass instance NoThunks MergeSchedule
 
+deriving stock instance Generic MergeBatchSize
+deriving anyclass instance NoThunks MergeBatchSize
+
 {-------------------------------------------------------------------------------
   RWVar
 -------------------------------------------------------------------------------}

src/Database/LSMTree.hs

@@ -109,7 +109,8 @@ module Database.LSMTree (
     confBloomFilterAlloc,
     confFencePointerIndex,
     confDiskCachePolicy,
-    confMergeSchedule
+    confMergeSchedule,
+    confMergeBatchSize
   ),
   defaultTableConfig,
   MergePolicy (LazyLevelling),
@@ -119,6 +120,7 @@ module Database.LSMTree (
   BloomFilterAlloc (AllocFixed, AllocRequestFPR),
   FencePointerIndexType (OrdinaryIndex, CompactIndex),
   DiskCachePolicy (..),
+  MergeBatchSize (..),
 
   -- ** Table Configuration Overrides #table_configuration_overrides#
   OverrideDiskCachePolicy (..),
@@ -214,9 +216,10 @@ import qualified Database.LSMTree.Internal.BlobRef as Internal
 import           Database.LSMTree.Internal.Config
                      (BloomFilterAlloc (AllocFixed, AllocRequestFPR),
                      DiskCachePolicy (..), FencePointerIndexType (..),
-                     LevelNo (..), MergePolicy (..), MergeSchedule (..),
-                     SizeRatio (..), TableConfig (..), WriteBufferAlloc (..),
-                     defaultTableConfig, serialiseKeyMinimalSize)
+                     LevelNo (..), MergeBatchSize (..), MergePolicy (..),
+                     MergeSchedule (..), SizeRatio (..), TableConfig (..),
+                     WriteBufferAlloc (..), defaultTableConfig,
+                     serialiseKeyMinimalSize)
 import           Database.LSMTree.Internal.Config.Override
                      (OverrideDiskCachePolicy (..))
 import           Database.LSMTree.Internal.Entry (NumEntries (..))

src/Database/LSMTree/Internal/Config.hs

@@ -26,12 +26,16 @@ module Database.LSMTree.Internal.Config (
   , diskCachePolicyForLevel
     -- * Merge schedule
   , MergeSchedule (..)
+    -- * Merge batch size
+  , MergeBatchSize (..)
+  , creditThresholdForLevel
   ) where
 
 import           Control.DeepSeq (NFData (..))
 import           Database.LSMTree.Internal.Index (IndexType)
 import qualified Database.LSMTree.Internal.Index as Index
                      (IndexType (Compact, Ordinary))
+import qualified Database.LSMTree.Internal.MergingRun as MR
 import qualified Database.LSMTree.Internal.RawBytes as RB
 import           Database.LSMTree.Internal.Run (RunDataCaching (..))
 import           Database.LSMTree.Internal.RunAcc (RunBloomFilterAlloc (..))
@@ -90,6 +94,12 @@ For a detailed discussion of fine-tuning the table configuration, see [Fine-tuni
 [@confDiskCachePolicy :: t'DiskCachePolicy'@]
     The /disk cache policy/ supports caching lookup operations using the OS page cache.
     Caching may improve the performance of lookups and updates if database access follows certain patterns.
+
+[@confMergeBatchSize :: t'MergeBatchSize'@]
+    The merge batch size balances the maximum latency of individual update
+    operations, versus the latency of a sequence of update operations. Bigger
+    batches improves overall performance but some updates will take a lot
+    longer than others. The default is to use a large batch size.
 -}
 data TableConfig = TableConfig {
     confMergePolicy       :: !MergePolicy
@@ -99,12 +109,14 @@ data TableConfig = TableConfig {
   , confBloomFilterAlloc  :: !BloomFilterAlloc
   , confFencePointerIndex :: !FencePointerIndexType
   , confDiskCachePolicy   :: !DiskCachePolicy
+  , confMergeBatchSize    :: !MergeBatchSize
   }
   deriving stock (Show, Eq)
 
 instance NFData TableConfig where
-  rnf (TableConfig a b c d e f g) =
-      rnf a `seq` rnf b `seq` rnf c `seq` rnf d `seq` rnf e `seq` rnf f `seq` rnf g
+  rnf (TableConfig a b c d e f g h) =
+      rnf a `seq` rnf b `seq` rnf c `seq` rnf d `seq`
+      rnf e `seq` rnf f `seq` rnf g `seq` rnf h
 
 -- | The 'defaultTableConfig' defines reasonable defaults for all 'TableConfig' parameters.
 --
@@ -122,6 +134,8 @@ instance NFData TableConfig where
 -- OrdinaryIndex
 -- >>> confDiskCachePolicy defaultTableConfig
 -- DiskCacheAll
+-- >>> confMergeBatchSize defaultTableConfig
+-- MergeBatchSize 20000
 --
 defaultTableConfig :: TableConfig
 defaultTableConfig =
@@ -133,6 +147,7 @@ defaultTableConfig =
       , confBloomFilterAlloc  = AllocRequestFPR 1.0e-3
       , confFencePointerIndex = OrdinaryIndex
       , confDiskCachePolicy   = DiskCacheAll
+      , confMergeBatchSize    = MergeBatchSize 20_000 -- same as write buffer
       }
 
 data RunLevelNo = RegularLevel LevelNo | UnionLevel
@@ -238,6 +253,8 @@ data MergeSchedule =
     The 'Incremental' merge schedule spreads out the merging work over time.
     This is less efficient than the 'OneShot' merge schedule, but has a consistent workload.
     Using the 'Incremental' merge schedule, the worst-case disk I\/O complexity of the update operations is /logarithmic/ in the size of the table.
+    This 'Incremental' merge schedule still uses batching to improve performance.
+    The batch size can be controlled using the 'MergeBatchSize'.
     -}
   | Incremental
   deriving stock (Eq, Show)
@@ -385,3 +402,50 @@ diskCachePolicyForLevel policy levelNo =
         RegularLevel l | l <= LevelNo n -> CacheRunData
                        | otherwise      -> NoCacheRunData
         UnionLevel                      -> NoCacheRunData
+
+{-------------------------------------------------------------------------------
+  Merge batch size
+-------------------------------------------------------------------------------}
+
+{- |
+The /merge batch size/ is a micro-tuning parameter, and in most cases you do
+need to think about it and can leave it at its default.
+
+When using the 'Incremental' merge schedule, merging is done in batches. This
+is a trade-off: larger batches tends to mean better overall performance but the
+downside is that while most updates (inserts, deletes, upserts) are fast, some
+are slower (when a batch of merging work has to be done).
+
+If you care most about the maximum latency of updates, then use a small batch
+size. If you don't care about latency of individual operations, just the
+latency of the overall sequence of operations then use a large batch size. The
+default is to use a large batch size, the same size as the write buffer itself.
+The minimum batch size is 1. The maximum batch size is the size of the write
+buffer 'confWriteBufferAlloc'.
+
+Note that the actual batch size is the minimum of this configuration
+parameter and the size of the batch of operations performed (e.g. 'inserts').
+So if you consistently use large batches, you can use a batch size of 1 and
+the merge batch size will always be determined by the operation batch size.
+
+A further reason why it may be preferable to use minimal batch sizes is to get
+good parallel work balance, when using parallelism.
+-}
+newtype MergeBatchSize = MergeBatchSize Int
+  deriving stock (Show, Eq, Ord)
+  deriving newtype (NFData)
+
+-- TODO: the thresholds for doing merge work should be different for each level,
+-- and ideally all-pairs co-prime.
+creditThresholdForLevel :: TableConfig -> LevelNo -> MR.CreditThreshold
+creditThresholdForLevel TableConfig {
+                           confMergeBatchSize   = MergeBatchSize mergeBatchSz,
+                           confWriteBufferAlloc = AllocNumEntries writeBufferSz
+                         }
+                        (LevelNo _i) =
+    MR.CreditThreshold
+  . MR.UnspentCredits
+  . MR.MergeCredits
+  . max 1
+  . min writeBufferSz
+  $ mergeBatchSz

src/Database/LSMTree/Internal/Config/Override.hs

@@ -91,16 +91,8 @@ instance Override DiskCachePolicy SnapshotMetaData where
         in  fmap (override rdc) smt
 
 instance Override DiskCachePolicy TableConfig where
-  override confDiskCachePolicy' TableConfig {..}
-    = TableConfig
-        { confMergePolicy,
-          confMergeSchedule,
-          confSizeRatio,
-          confWriteBufferAlloc,
-          confBloomFilterAlloc,
-          confFencePointerIndex,
-          confDiskCachePolicy = confDiskCachePolicy'
-        }
+  override confDiskCachePolicy' tc =
+    tc { confDiskCachePolicy = confDiskCachePolicy' }
 
 instance Override DiskCachePolicy (SnapLevels SnapshotRun) where
   override dcp (SnapLevels (vec :: V.Vector (SnapLevel SnapshotRun))) =

src/Database/LSMTree/Internal/IncomingRun.hs

@@ -218,13 +218,6 @@ supplyCreditsIncomingRun conf ln (Merging _ nominalDebt nominalCreditsVar mr)
     -- use atomic operations for its counters). We could potentially simplify
     -- MergingRun by dispensing with batching for the MergeCredits counters.
 
--- TODO: the thresholds for doing merge work should be different for each level,
--- maybe co-prime?
-creditThresholdForLevel :: TableConfig -> LevelNo -> MR.CreditThreshold
-creditThresholdForLevel conf (LevelNo _i) =
-    let AllocNumEntries x = confWriteBufferAlloc conf
-    in  MR.CreditThreshold (MR.UnspentCredits (MergeCredits x))
-
 -- | Deposit nominal credits in the local credits var, ensuring the total
 -- credits does not exceed the total debt.
 --

src/Database/LSMTree/Internal/Snapshot/Codec.hs

@@ -294,27 +294,44 @@ instance Encode TableConfig where
         , confBloomFilterAlloc = bloomFilterAlloc
         , confFencePointerIndex = fencePointerIndex
         , confDiskCachePolicy = diskCachePolicy
+        , confMergeBatchSize = mergeBatchSize
         }
       ) =
-      encodeListLen 7
+      encodeListLen 8
         <> encode mergePolicy
         <> encode mergeSchedule
         <> encode sizeRatio
         <> encode writeBufferAlloc
         <> encode bloomFilterAlloc
         <> encode fencePointerIndex
         <> encode diskCachePolicy
+        <> encode mergeBatchSize
 
 instance DecodeVersioned TableConfig where
-  decodeVersioned v = do
-      _ <- decodeListLenOf 7
+  decodeVersioned v@V0 = do
+      decodeListLenOf 7
       confMergePolicy <- decodeVersioned v
       confMergeSchedule <- decodeVersioned v
       confSizeRatio <- decodeVersioned v
       confWriteBufferAlloc <- decodeVersioned v
       confBloomFilterAlloc <- decodeVersioned v
       confFencePointerIndex <- decodeVersioned v
       confDiskCachePolicy <- decodeVersioned v
+      let confMergeBatchSize = case confWriteBufferAlloc of
+                                 AllocNumEntries n -> MergeBatchSize n
+      pure TableConfig {..}
+
+  -- We introduced the confMergeBatchSize in V1
+  decodeVersioned v@V1 = do
+      decodeListLenOf 8
+      confMergePolicy <- decodeVersioned v
+      confMergeSchedule <- decodeVersioned v
+      confSizeRatio <- decodeVersioned v
+      confWriteBufferAlloc <- decodeVersioned v
+      confBloomFilterAlloc <- decodeVersioned v
+      confFencePointerIndex <- decodeVersioned v
+      confDiskCachePolicy <- decodeVersioned v
+      confMergeBatchSize <- decodeVersioned v
       pure TableConfig {..}
 
 -- MergePolicy
@@ -494,6 +511,14 @@ instance DecodeVersioned MergeSchedule where
         1 -> pure Incremental
         _ -> fail ("[MergeSchedule] Unexpected tag: " <> show tag)
 
+-- MergeBatchSize
+
+instance Encode MergeBatchSize where
+  encode (MergeBatchSize n) = encodeInt n
+
+instance DecodeVersioned MergeBatchSize where
+  decodeVersioned _v = MergeBatchSize <$> decodeInt
+
 {-------------------------------------------------------------------------------
   Encoding and decoding: SnapLevels
 -------------------------------------------------------------------------------}

src/Database/LSMTree/Simple.hs

@@ -111,6 +111,7 @@ module Database.LSMTree.Simple (
     FencePointerIndexType (OrdinaryIndex, CompactIndex),
     DiskCachePolicy (..),
     MergeSchedule (..),
+    MergeBatchSize (..),
 
     -- ** Table Configuration Overrides #table_configuration_overrides#
     OverrideDiskCachePolicy (..),
@@ -165,9 +166,9 @@ import           Data.Vector (Vector)
 import           Data.Void (Void)
 import           Database.LSMTree (BloomFilterAlloc, CursorClosedError (..),
                      DiskCachePolicy, FencePointerIndexType,
-                     InvalidSnapshotNameError (..), MergePolicy, MergeSchedule,
-                     OverrideDiskCachePolicy (..), Range (..), RawBytes,
-                     ResolveAsFirst (..), SerialiseKey (..),
+                     InvalidSnapshotNameError (..), MergeBatchSize, MergePolicy,
+                     MergeSchedule, OverrideDiskCachePolicy (..), Range (..),
+                     RawBytes, ResolveAsFirst (..), SerialiseKey (..),
                      SerialiseKeyOrderPreserving, SerialiseValue (..),
                      SessionClosedError (..), SizeRatio,
                      SnapshotCorruptedError (..),