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

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

@@ -119,6 +119,7 @@ module Database.LSMTree (
   BloomFilterAlloc (AllocFixed, AllocRequestFPR),
   FencePointerIndexType (OrdinaryIndex, CompactIndex),
   DiskCachePolicy (..),
+  MergeBatchSize (..),
 
   -- ** Table Configuration Overrides #table_configuration_overrides#
   OverrideDiskCachePolicy (..),
@@ -214,9 +215,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 (..))
@@ -99,12 +103,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.
 --
@@ -133,6 +139,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 +245,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 +394,41 @@ 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.
+
+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 n }
+                        (LevelNo _i) =
+    MR.CreditThreshold (MR.UnspentCredits (MR.MergeCredits (max 1 n)))

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

@@ -22,6 +22,7 @@ import           Codec.CBOR.Decoding
 import           Codec.CBOR.Encoding
 import           Codec.CBOR.Read
 import           Codec.CBOR.Write
+import           Control.Monad (unless)
 import           Control.Monad.Class.MonadThrow (Exception (displayException),
                      MonadThrow (..))
 import           Data.Bifunctor (Bifunctor (..))
@@ -274,27 +275,34 @@ 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@V0 = do
-      _ <- decodeListLenOf 7
+      n <- decodeListLen
+      unless (n >= 7 && n <= 8) $
+        fail "TableConfig: expected record of length 7 or 8"
       confMergePolicy <- decodeVersioned v
       confMergeSchedule <- decodeVersioned v
       confSizeRatio <- decodeVersioned v
       confWriteBufferAlloc <- decodeVersioned v
       confBloomFilterAlloc <- decodeVersioned v
       confFencePointerIndex <- decodeVersioned v
       confDiskCachePolicy <- decodeVersioned v
+      confMergeBatchSize <- if n == 8
+                              then decodeVersioned v
+                              else pure (confMergeBatchSize defaultTableConfig)
       pure TableConfig {..}
 
 -- MergePolicy
@@ -492,6 +500,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 V0 = 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 (..),

test/Test/Database/LSMTree/Internal/Snapshot/Codec.hs

@@ -230,11 +230,11 @@ instance Arbitrary SnapshotRun where
 
 instance Arbitrary TableConfig where
   arbitrary =
-          TableConfig <$> arbitrary <*> arbitrary <*> arbitrary
-      <*> arbitrary   <*> arbitrary <*> arbitrary <*> arbitrary
-  shrink (TableConfig a b c d e f g) =
-      [ TableConfig a' b' c' d' e' f' g'
-      | (a', b', c', d', e', f', g') <- shrink (a, b, c, d, e, f, g) ]
+          TableConfig <$> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary
+                      <*> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary
+  shrink (TableConfig a b c d e f g h) =
+      [ TableConfig a' b' c' d' e' f' g' h'
+      | (a', b', c', d', e', f', g', h') <- shrink (a, b, c, d, e, f, g, h) ]
 
 instance Arbitrary MergePolicy where
   arbitrary = pure LazyLevelling
@@ -273,6 +273,10 @@ instance Arbitrary MergeSchedule where
   arbitrary = elements [OneShot, Incremental]
   shrink _ = []
 
+instance Arbitrary MergeBatchSize where
+  arbitrary = MergeBatchSize <$> arbitrary
+  shrink (MergeBatchSize n) = map MergeBatchSize (shrink n)
+
 {-------------------------------------------------------------------------------
   Arbitrary: SnapLevels
 -------------------------------------------------------------------------------}

test/Test/Database/LSMTree/Internal/Snapshot/Codec/Golden.hs

@@ -15,8 +15,8 @@ import           Data.Typeable
 import qualified Data.Vector as V
 import           Database.LSMTree.Internal.Config (BloomFilterAlloc (..),
                      DiskCachePolicy (..), FencePointerIndexType (..),
-                     MergePolicy (..), MergeSchedule (..), SizeRatio (..),
-                     TableConfig (..), WriteBufferAlloc (..))
+                     MergeBatchSize (..), MergePolicy (..), MergeSchedule (..),
+                     SizeRatio (..), TableConfig (..), WriteBufferAlloc (..))
 import           Database.LSMTree.Internal.MergeSchedule
                      (MergePolicyForLevel (..), NominalCredits (..),
                      NominalDebt (..))
@@ -276,7 +276,8 @@ instance EnumGolden SnapshotLabel where
         SnapshotLabel{} -> ()
 
 instance EnumGolden TableConfig where
-  singGolden = TableConfig singGolden singGolden singGolden singGolden singGolden singGolden singGolden
+  singGolden = TableConfig singGolden singGolden singGolden singGolden
+                           singGolden singGolden singGolden singGolden
     where
       _coveredAllCases = \case
         TableConfig{} -> ()
@@ -329,6 +330,9 @@ instance EnumGolden MergeSchedule where
         OneShot{} -> ()
         Incremental{} -> ()
 
+instance EnumGolden MergeBatchSize where
+  enumGolden = map MergeBatchSize [ 1, 1000 ]
+
 instance EnumGolden (SnapLevels SnapshotRun) where
   singGolden = SnapLevels singGolden
     where

test/Test/Database/LSMTree/StateMachine.hs

@@ -94,8 +94,7 @@ import qualified Database.LSMTree.Class as Class
 import           Database.LSMTree.Extras (showPowersOf)
 import           Database.LSMTree.Extras.Generators (KeyForIndexCompact)
 import           Database.LSMTree.Extras.NoThunks (propNoThunks)
-import qualified Database.LSMTree.Internal.Config as R
-                     (TableConfig (TableConfig))
+import qualified Database.LSMTree.Internal.Config as R (TableConfig (..))
 import           Database.LSMTree.Internal.Serialise (SerialisedBlob,
                      SerialisedValue)
 import qualified Database.LSMTree.Internal.Types as R.Types
@@ -226,6 +225,8 @@ instance Arbitrary R.TableConfig where
         ]
     confWriteBufferAlloc <- QC.arbitrary
     confFencePointerIndex <- QC.arbitrary
+    confMergeBatchSize <- QC.sized $ \sz ->
+                            R.MergeBatchSize <$> QC.chooseInt (1, sz)
     pure $ R.TableConfig {
         R.confMergePolicy       = R.LazyLevelling
       , R.confSizeRatio         = R.Four
@@ -234,6 +235,7 @@ instance Arbitrary R.TableConfig where
       , confFencePointerIndex
       , R.confDiskCachePolicy   = R.DiskCacheNone
       , confMergeSchedule
+      , confMergeBatchSize
       }
 
   shrink R.TableConfig{..} =