Convert Run to use BlobFile

Where before it contained a file handle and a ref counter, it now just
contains a BlobFile (which itself is the pair of a handle and counter).

Author: dcoutts

src/Database/LSMTree/Internal/Merge.hs

@@ -190,7 +190,7 @@ finaliser var b rs = do
 -- this function with async exceptions masked. Otherwise, these resources can
 -- leak.
 complete ::
-     (MonadFix m, MonadSTM m, MonadST m, MonadThrow m)
+     (MonadFix m, MonadSTM m, MonadST m, MonadMask m)
   => Merge m h
   -> m (Run m h)
 complete Merge{..} = do
@@ -218,7 +218,7 @@ complete Merge{..} = do
 --
 -- Note: run with async exceptions masked. See 'complete'.
 stepsToCompletion ::
-      (MonadCatch m, MonadFix m, MonadSTM m, MonadST m)
+      (MonadMask m, MonadFix m, MonadSTM m, MonadST m)
    => Merge m h
    -> Int
    -> m (Run m h)
@@ -237,7 +237,7 @@ stepsToCompletion m stepBatchSize = go
 --
 -- Note: run with async exceptions masked. See 'complete'.
 stepsToCompletionCounted ::
-     (MonadCatch m, MonadFix m, MonadSTM m, MonadST m)
+     (MonadMask m, MonadFix m, MonadSTM m, MonadST m)
   => Merge m h
   -> Int
   -> m (Int, Run m h)

src/Database/LSMTree/Internal/MergeSchedule.hs

@@ -869,7 +869,7 @@ levelIsFull sr rs = V.length rs + 1 >= (sizeRatioInt sr)
 
 {-# SPECIALISE mergeRuns :: ResolveSerialisedValue -> HasFS IO h -> HasBlockIO IO h -> RunDataCaching -> RunBloomFilterAlloc -> RunFsPaths -> Merge.Level -> V.Vector (Run IO h) -> IO (Run IO h) #-}
 mergeRuns ::
-     (MonadCatch m, MonadFix m, MonadST m, MonadSTM m)
+     (MonadMask m, MonadFix m, MonadST m, MonadSTM m)
   => ResolveSerialisedValue
   -> HasFS m h
   -> HasBlockIO m h

src/Database/LSMTree/Internal/Run.hs

@@ -66,7 +66,9 @@ import           Data.BloomFilter (Bloom)
 import qualified Data.ByteString.Short as SBS
 import           Data.Foldable (for_)
 import           Data.Word (Word64)
-import           Database.LSMTree.Internal.BlobRef (BlobRef (..), BlobSpan (..))
+import           Database.LSMTree.Internal.BlobFile hiding (removeReference)
+import qualified Database.LSMTree.Internal.BlobFile as BlobFile
+import           Database.LSMTree.Internal.BlobRef hiding (removeReference)
 import           Database.LSMTree.Internal.BloomFilter (bloomFilterFromSBS)
 import qualified Database.LSMTree.Internal.CRC32C as CRC
 import           Database.LSMTree.Internal.Entry (NumEntries (..))
@@ -111,7 +113,7 @@ data Run m h = Run {
       -- | The file handle for the BLOBs file. This file is opened
       -- read-only and is accessed in a normal style using buffered
       -- I\/O, reading arbitrary file offset and length spans.
-    , runBlobFile       :: !(FS.Handle h)
+    , runBlobFile       :: !(BlobFile m h)
     , runRunDataCaching :: !RunDataCaching
     , runHasFS          :: !(HasFS m h)
     , runHasBlockIO     :: !(HasBlockIO m h)
@@ -143,10 +145,10 @@ removeReferenceN r = RC.removeReferenceN (runRefCounter r)
 
 -- | Helper function to make a 'BlobRef' that points into a 'Run'.
 mkBlobRefForRun :: Run m h -> BlobSpan -> BlobRef m (FS.Handle h)
-mkBlobRefForRun Run{runBlobFile, runRefCounter} blobRefSpan =
+mkBlobRefForRun Run{runBlobFile} blobRefSpan =
     BlobRef {
-      blobRefFile  = runBlobFile,
-      blobRefCount = runRefCounter,
+      blobRefFile  = blobFileHandle runBlobFile,
+      blobRefCount = blobFileRefCounter runBlobFile,
       blobRefSpan
     }
 
@@ -158,18 +160,18 @@ mkBlobRefForRun Run{runBlobFile, runRefCounter} blobRefSpan =
 --
 -- TODO: Once snapshots are implemented, files should get removed, but for now
 -- we want to be able to re-open closed runs from disk.
-close :: (MonadSTM m, MonadThrow m) => Run m h -> m ()
+close :: (MonadSTM m, MonadMask m, PrimMonad m) => Run m h -> m ()
 close Run {..} = do
     -- TODO: removing files should drop them from the page cache, but until we
     -- have proper snapshotting we are keeping the files around. Because of
     -- this, we instruct the OS to drop all run-related files from the page
     -- cache
     FS.hDropCacheAll runHasBlockIO runKOpsFile
-    FS.hDropCacheAll runHasBlockIO runBlobFile
+    FS.hDropCacheAll runHasBlockIO (blobFileHandle runBlobFile)
 
     FS.hClose runHasFS runKOpsFile
       `finally`
-        FS.hClose runHasFS runBlobFile
+         BlobFile.removeReference runBlobFile
 
 -- | Should this run cache key\/ops data in memory?
 data RunDataCaching = CacheRunData | NoCacheRunData
@@ -205,7 +207,7 @@ setRunDataCaching hbio runKOpsFile NoCacheRunData = do
   -> RunBuilder IO h
   -> IO (Run IO h) #-}
 fromMutable ::
-     (MonadFix m, MonadST m, MonadSTM m, MonadThrow m)
+     (MonadFix m, MonadST m, MonadSTM m, MonadMask m)
   => RunDataCaching
   -> RefCount
   -> RunBuilder m h
@@ -214,7 +216,8 @@ fromMutable runRunDataCaching refCount builder = do
     (runHasFS, runHasBlockIO, runRunFsPaths, runFilter, runIndex, runNumEntries) <-
       Builder.unsafeFinalise (runRunDataCaching == NoCacheRunData) builder
     runKOpsFile <- FS.hOpen runHasFS (runKOpsPath runRunFsPaths) FS.ReadMode
-    runBlobFile <- FS.hOpen runHasFS (runBlobPath runRunFsPaths) FS.ReadMode
+    runBlobFile <- newBlobFile runHasFS
+               =<< FS.hOpen runHasFS (runBlobPath runRunFsPaths) FS.ReadMode
     setRunDataCaching runHasBlockIO runKOpsFile runRunDataCaching
     rec runRefCounter <- RC.unsafeMkRefCounterN refCount (Just $ close r)
         let !r = Run { .. }
@@ -236,7 +239,7 @@ fromMutable runRunDataCaching refCount builder = do
 -- immediately when they are added to the write buffer, avoiding the need to do
 -- it here.
 fromWriteBuffer ::
-     (MonadFix m, MonadST m, MonadSTM m, MonadThrow m)
+     (MonadFix m, MonadST m, MonadSTM m, MonadMask m)
   => HasFS m h
   -> HasBlockIO m h
   -> RunDataCaching
@@ -274,7 +277,7 @@ data FileFormatError = FileFormatError FS.FsPath String
 -- checksum ('ChecksumError') or can't be parsed ('FileFormatError').
 openFromDisk ::
      forall m h.
-     (MonadFix m, MonadSTM m, MonadThrow m, PrimMonad m)
+     (MonadFix m, MonadSTM m, MonadMask m, PrimMonad m)
   => HasFS m h
   -> HasBlockIO m h
   -> RunDataCaching
@@ -299,7 +302,8 @@ openFromDisk fs hbio runRunDataCaching runRunFsPaths = do
         =<< readCRC (forRunIndex expectedChecksums) (forRunIndex paths)
 
     runKOpsFile <- FS.hOpen fs (runKOpsPath runRunFsPaths) FS.ReadMode
-    runBlobFile <- FS.hOpen fs (runBlobPath runRunFsPaths) FS.ReadMode
+    runBlobFile <- newBlobFile fs
+               =<< FS.hOpen fs (runBlobPath runRunFsPaths) FS.ReadMode
     setRunDataCaching hbio runKOpsFile runRunDataCaching
     rec runRefCounter <- RC.unsafeMkRefCounterN (RefCount 1) (Just $ close r)
         let !r = Run

test/Test/Database/LSMTree/Internal/Merge.hs

@@ -12,6 +12,7 @@ import qualified Data.Vector as V
 import           Database.LSMTree.Extras
 import           Database.LSMTree.Extras.Generators (KeyForIndexCompact)
 import           Database.LSMTree.Extras.RunData
+import qualified Database.LSMTree.Internal.BlobFile as BlobFile
 import qualified Database.LSMTree.Internal.Entry as Entry
 import qualified Database.LSMTree.Internal.Merge as Merge
 import           Database.LSMTree.Internal.PageAcc (entryWouldFitInPage)
@@ -73,9 +74,9 @@ prop_MergeDistributes fs hbio level stepSize (SmallList rds) =
       withRun fs hbio (simplePath 1) (RunData $ mergeWriteBuffers level $ fmap unRunData rds') $ \rhs -> do
 
         lhsKOpsFile <- FS.hGetAll fs (Run.runKOpsFile lhs)
-        lhsBlobFile <- FS.hGetAll fs (Run.runBlobFile lhs)
+        lhsBlobFile <- FS.hGetAll fs (BlobFile.blobFileHandle (Run.runBlobFile lhs))
         rhsKOpsFile <- FS.hGetAll fs (Run.runKOpsFile rhs)
-        rhsBlobFile <- FS.hGetAll fs (Run.runBlobFile rhs)
+        rhsBlobFile <- FS.hGetAll fs (BlobFile.blobFileHandle (Run.runBlobFile rhs))
 
         lhsKOps <- readKOps Nothing lhs
         rhsKOps <- readKOps Nothing rhs

test/Test/Database/LSMTree/Internal/Run.hs

@@ -25,6 +25,7 @@ import           Test.Tasty.QuickCheck
 import           Control.RefCount (RefCount (..), readRefCount)
 import           Database.LSMTree.Extras.Generators (KeyForIndexCompact (..))
 import           Database.LSMTree.Extras.RunData
+import           Database.LSMTree.Internal.BlobFile (BlobFile (..))
 import           Database.LSMTree.Internal.BlobRef (BlobSpan (..))
 import qualified Database.LSMTree.Internal.CRC32C as CRC
 import           Database.LSMTree.Internal.Entry
@@ -195,8 +196,8 @@ prop_WriteAndOpen fs hbio wb =
         (FS.handlePath (runKOpsFile written))
         (FS.handlePath (runKOpsFile loaded))
       assertEqual "blob file"
-        (FS.handlePath (runBlobFile written))
-        (FS.handlePath (runBlobFile loaded))
+        (FS.handlePath (blobFileHandle (runBlobFile written)))
+        (FS.handlePath (blobFileHandle (runBlobFile loaded)))
 
       -- make sure runs get closed again
       removeReference loaded