WIP: implement scheduled merges

Author: jorisdral

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

@@ -527,7 +527,7 @@ instance NoThunks a => NoThunks (StrictTVar IO a) where
 instance NoThunks a => NoThunks (StrictMVar IO a) where
   showTypeOf (_ :: Proxy (StrictMVar IO a)) = "StrictMVar IO"
   wNoThunks ctx var = do
-      x <- readMVar var
+      !x <- readMVar var -- TODO: undo
       noThunks ctx x
 
 {-------------------------------------------------------------------------------

src/Database/LSMTree/Internal.hs

@@ -1211,6 +1211,7 @@ snapshot resolve snap label th = do
                     (RW.unsafeAcquireWriteAccess (tableContent thEnv))
                     (atomically . RW.unsafeReleaseWriteAccess (tableContent thEnv))
                     $ \reg content -> do
+        supplyCredits (Entry.unNumEntries $ case confWriteBufferAlloc conf of AllocNumEntries x -> x) (tableLevels content)
         content' <- flushWriteBuffer
               (TraceMerge `contramap` tableTracer th)
               conf

src/Database/LSMTree/Internal/MergeSchedule.hs

@@ -706,11 +706,18 @@ addRunToLevels tr conf@TableConfig{..} resolve hfs hbio root uc r0 reg levels =
       traceWith tr $ AtLevel ln $ TraceExpectCompletedMergeSingleRun (runNumber $ Run.runRunFsPaths r)
       pure r
     expectCompletedMerge ln (MergingRun _ _ var) = do
-      withMVar var $ \case
-        CompletedMerge r -> do
+      modifyMVarMasked var $ \case
+        x@(CompletedMerge r) -> do
           traceWith tr $ AtLevel ln $ TraceExpectCompletedMerge (runNumber $ Run.runRunFsPaths r)
-          pure r
-        OngoingMerge _rs _ _m -> error "expectCompletedMerge: OngoingMerge not yet supported" -- TODO: implement.
+          pure (x, r)
+        OngoingMerge _rs _ _m -> do
+          -- RefCount n <- Merge.readRefCount m
+          -- let !n' = fromIntegralChecked n
+          -- V.forM_ rs $ \r -> Run.removeReferenceN r n'
+          -- r <- Merge.complete m
+          -- Merge.removeReferenceN m n'
+          -- pure (CompletedMerge r, r)
+          error "expectCompletedMerge: OngoingMerge not yet supported" -- TODO: implement.
 
     newMerge :: MergePolicyForLevel
              -> Merge.Level

src/Database/LSMTree/Internal/Snapshot.hs

@@ -122,7 +122,9 @@ runNumber r = Paths.runNumber (Run.runRunFsPaths r)
   -> HasFS IO h
   -> HasBlockIO IO h
   -> TableConfig
+  -> UniqCounter IO
   -> SessionRoot
+  -> ResolveSerialisedValue
   -> SnapLevels
   -> IO (Levels IO h)
   #-}