endurance test + multiple fixes

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal.hs

@@ -67,6 +67,7 @@ ncqStorageOpen fp upd = do
   ncqStateUse       <- newTVarIO mempty
   ncqServiceSem     <- atomically $ newTSem 1
   ncqFileLock       <- newTVarIO Nothing
+  ncqCurrentFossils <- newTVarIO mempty
 
   let ncq = NCQStorage{..} & upd
 
@@ -231,7 +232,7 @@ ncqTryLoadState me@NCQStorage{..} = do
 
       if not corrupted then do
         debug $ yellow "indexing" <+> pretty dataFile
-        ncqIndexFile me dataFile
+        ncqIndexFile me Nothing dataFile
       else do
 
         o <- ncqFileTryRecover path

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal/Files.hs

@@ -7,6 +7,8 @@ import HBS2.Storage.NCQ3.Internal.Types
 import System.Posix.Files qualified as PFS
 import Data.List qualified as List
 
+{- HLINT ignore "Eta reduce" -}
+
 
 removeFile :: MonadIO m => FilePath -> m ()
 removeFile fp = do
@@ -49,29 +51,36 @@ ncqListFilesBy  me@NCQStorage{..} filt = do
 
   pure $ List.sortOn ( Down . fst ) r
 
+ncqFindMinPairOfBy :: forall fa m . (ToFileName fa, MonadUnliftIO m)
+                   => NCQStorage
+                   -> (fa -> Bool)                         -- ^ eligible predicate
+                   -> [fa]
+                   -> m (Maybe (NCQFileSize, fa, fa))
+ncqFindMinPairOfBy sto eligible lst =
+  go lst Nothing
+  where
+    go :: [fa] -> Maybe (NCQFileSize, fa, fa) -> m (Maybe (NCQFileSize, fa, fa))
+    go (a:b:rest) best = do
+      best' <- if eligible a && eligible b
+                 then do
+                   let pa = ncqGetFileName sto a
+                   let pb = ncqGetFileName sto b
+                   s1 <- fsize pa
+                   s2 <- fsize pb
+                   let sz = fromIntegral (s1 + s2)
+                   pure $ case best of
+                     Nothing                -> Just (sz, a, b)
+                     Just (sz0,_,_) | sz<sz0 -> Just (sz, a, b)
+                     _                       -> best
+                 else pure best
+      go (b:rest) best'
+    go _ best = pure best
+
+    fsize s = liftIO (PFS.getFileStatus s) <&> PFS.fileSize
+
 ncqFindMinPairOf ::  forall fa m . (ToFileName fa, MonadUnliftIO m)
                  => NCQStorage
                  -> [fa]
                  -> m (Maybe (NCQFileSize, fa, fa))
-ncqFindMinPairOf sto lst = do
+ncqFindMinPairOf sto lst = ncqFindMinPairOfBy sto (const True) lst
-
-  let files = fmap (\x -> (x, ncqGetFileName sto x)) lst
-
-  flip fix (files, Nothing) $ \next (fs, r) -> do
-    case fs of
-      [] ->  pure r
-      [ _ ] -> pure r
-      ( s1 : s2 : ss ) -> do
-        size1 <- fsize (snd s1)
-        size2 <- fsize (snd s2)
-        let size = fromIntegral $ size1 + size2
-
-        case r of
-          Nothing -> next (s2 : ss, Just (size, fst s1, fst s2) )
-          e@(Just (size0, _, _)) | size0 > size -> next (s2 : ss, Just (size, fst s1, fst s2) )
-                                 | otherwise -> next (s2:ss, e)
-
-  where fsize s = liftIO (PFS.getFileStatus s) <&> PFS.fileSize
-
-
 

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal/Fossil.hs

@@ -1,3 +1,4 @@
+{-# Language RecordWildCards #-}
 module HBS2.Storage.NCQ3.Internal.Fossil where
 
 import HBS2.Storage.NCQ3.Internal.Prelude
@@ -7,6 +8,7 @@ import HBS2.Storage.NCQ3.Internal.Index
 import HBS2.Storage.NCQ3.Internal.State
 
 import Data.HashSet qualified as HS
+import Data.HashMap.Strict qualified as HM
 import Data.List qualified as List
 import Data.ByteString.Lazy qualified as LBS
 import Data.ByteString qualified as BS
@@ -61,6 +63,7 @@ ncqFossilMergeStep :: forall m . MonadUnliftIO m
                -> m Bool
 
 ncqFossilMergeStep me@NCQStorage{..}  = withSem ncqServiceSem $ flip runContT pure $ callCC \exit -> do
+  tmax <- liftIO getPOSIXTime >>= newTVarIO
 
   debug "ncqFossilMergeStep"
 
@@ -69,7 +72,12 @@ ncqFossilMergeStep me@NCQStorage{..}  = withSem ncqServiceSem $ flip runContT pu
            <&> fmap DataFile .  HS.toList . ncqStateFiles
            <&> List.sortOn Down
 
-  r' <- lift $ ncqFindMinPairOf me files
+  NCQState{..}  <- readTVarIO ncqState
+
+  let tss = ncqStateIndex & fmap (\(Down x, y) -> (y, realToFrac x :: POSIXTime)) & HM.fromList
+
+  cur <- readTVarIO ncqCurrentFossils
+  r' <- lift $ ncqFindMinPairOfBy me (\x -> not (HS.member (coerce x) cur)) files
 
   r@(sumSize, f1, f2) <- ContT $ maybe1 r' (pure False)
 
@@ -84,6 +92,7 @@ ncqFossilMergeStep me@NCQStorage{..}  = withSem ncqServiceSem $ flip runContT pu
   ContT $ bracket none $ const do
           removeFile outFile
 
+
   liftIO $ withBinaryFileAtomic outFile WriteMode $ \fwh -> do
     fd <- handleToFd fwh
 
@@ -96,8 +105,7 @@ ncqFossilMergeStep me@NCQStorage{..}  = withSem ncqServiceSem $ flip runContT pu
           Nothing  -> pure False
           Just (InMemory{}) -> pure False
           Just (InFossil fk oi si) -> do
-            let skip = fk > fik || (fk == fik && o < fromIntegral oi)
+            let beWritten = fk == fik && o == fromIntegral oi
-            let beWritten = not skip
 
             -- let c = if skip then green else id
             -- when (si == ncqTombEntrySize) do
@@ -107,9 +115,11 @@ ncqFossilMergeStep me@NCQStorage{..}  = withSem ncqServiceSem $ flip runContT pu
             --               <+> "write" <+> c (pretty beWritten)
 
             atomically do
+              tj <- readTVar tmax
+              modifyTVar tmax (max (fromMaybe tj (HM.lookup fk tss)))
               here <- readTVar already <&> HS.member k
               let proceed = not here && beWritten
-              modifyTVar already (HS.insert k)
+              when proceed $ modifyTVar already (HS.insert k)
               pure proceed
 
     appendTailSection fd
@@ -126,7 +136,8 @@ ncqFossilMergeStep me@NCQStorage{..}  = withSem ncqServiceSem $ flip runContT pu
   ncqStateUpdate me do
     ncqStateAddFact (P (PData f3 ss))
 
-  lift $ ncqIndexFile me f3
+  ts <- readTVarIO tmax
+  lift $ ncqIndexFile me (Just ts) f3
 
   ncqStateUpdate me do
     ncqStateDelDataFile (coerce f1)

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal/Index.hs

@@ -84,8 +84,12 @@ ncqLocate :: MonadUnliftIO m => NCQStorage -> HashRef -> m (Maybe Location)
 ncqLocate me href = ncqOperation me (pure Nothing) do
   ncqLocate_ True me href
 
-ncqIndexFile :: MonadUnliftIO m => NCQStorage -> DataFile FileKey -> m (Maybe FilePath)
+ncqIndexFile :: MonadUnliftIO m
-ncqIndexFile n fk = runMaybeT do
+             => NCQStorage
+             -> Maybe POSIXTime
+             -> DataFile FileKey
+             -> m (Maybe FilePath)
+ncqIndexFile n ts' fk = runMaybeT do
 
   let fp   = toFileName fk & ncqGetFileName n
   fki <- ncqGetNewFileKey n IndexFile
@@ -110,7 +114,7 @@ ncqIndexFile n fk = runMaybeT do
   moveFile result dest
 
   stat <- liftIO $ PFS.getFileStatus dest
-  let ts = PFS.modificationTimeHiRes stat
+  let ts = fromMaybe (PFS.modificationTimeHiRes stat) ts'
 
   midx <- liftIO (nwayHashMMapReadOnly dest)
 

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal/Run.hs

@@ -49,7 +49,9 @@ ncqStorageRun ncq@NCQStorage{..} = flip runContT pure do
           if not stop then STM.retry else pure Nothing
 
     maybe1 what none $ \(fk :: FileKey) -> do
-      ncqIndexFile ncq (DataFile fk) >> loop
+      ncqIndexFile ncq Nothing (DataFile fk)
+      atomically $ modifyTVar ncqCurrentFossils (HS.delete fk)
+      loop
 
   let shLast = V.length ncqWriteOps - 1
   spawnActivity $ pooledForConcurrentlyN_ (V.length ncqWriteOps) [0..shLast] $ \i -> do
@@ -212,6 +214,8 @@ ncqStorageRun ncq@NCQStorage{..} = flip runContT pure do
     openNewDataFile = do
       fk <- ncqGetNewFileKey ncq DataFile
 
+      atomically $ modifyTVar ncqCurrentFossils (HS.insert fk)
+
       ncqStateUpdate ncq (ncqStateAddDataFile fk)
 
       let fname = ncqGetFileName ncq (DataFile fk)

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal/Sweep.hs

@@ -15,10 +15,11 @@ import Data.HashMap.Strict qualified as HM
 ncqLiveKeysSTM :: NCQStorage -> STM (HashSet FileKey)
 ncqLiveKeysSTM NCQStorage{..} = do
 
-  s0     <- readTVar ncqState
+  s0      <- readTVar ncqState
-  merged <- readTVar ncqStateUse <&> (s0<>) . foldMap fst . HM.elems
+  merged  <- readTVar ncqStateUse <&> (s0<>) . foldMap fst . HM.elems
+  current <- readTVar ncqCurrentFossils
 
-  pure $ HS.fromList $ universeBi @_ @FileKey merged
+  pure $ current <> HS.fromList (universeBi @_ @FileKey merged)
 
 ncqLiveKeys :: forall m . MonadIO m => NCQStorage -> m (HashSet FileKey)
 ncqLiveKeys  ncq = atomically $ ncqLiveKeysSTM  ncq

hbs2-storage-ncq/lib/HBS2/Storage/NCQ3/Internal/Types.hs

@@ -93,6 +93,7 @@ data NCQStorage =
   , ncqState          :: TVar NCQState
   , ncqStateKey       :: TVar FileKey
   , ncqStateUse       :: TVar (HashMap FileKey (NCQState, TVar Int))
+  , ncqCurrentFossils :: TVar (HashSet FileKey)
   , ncqWrites         :: TVar Int
   , ncqWriteEMA       :: TVar Double  -- for writes-per-seconds
   , ncqWriteQ         :: TVar (Seq HashRef)

hbs2-tests/scripts/ncq3/t1.ss

@@ -0,0 +1,19 @@
+test:root temp
+test:dir:keep
+set! w:getblk 100
+set! w:storm  2
+set! w:putblk 90
+set! w:blk 65536
+
+println "w:blk" w:blk
+println "go"
+
+; test:ncq3:endurance:inproc 200000
+test:ncq3:endurance:inproc 300000
+
+
+
+
+
+
+