mirror of https://github.com/voidlizard/hbs2
bytestring offloading
This commit is contained in:
parent
79788fd134
commit
500ad351a5
|
@ -1,4 +1,5 @@
|
||||||
{-# Language RecordWildCards #-}
|
{-# Language RecordWildCards #-}
|
||||||
|
{-# Language MultiWayIf #-}
|
||||||
module HBS2.Storage.NCQ3.Internal where
|
module HBS2.Storage.NCQ3.Internal where
|
||||||
|
|
||||||
import HBS2.Storage.NCQ3.Internal.Prelude
|
import HBS2.Storage.NCQ3.Internal.Prelude
|
||||||
|
@ -170,13 +171,19 @@ ncqPutBS0 wait ncq@NCQStorage{..} mtp mhref bs' = ncqOperation ncq (pure $ fromM
|
||||||
let work = do
|
let work = do
|
||||||
let bs = ncqMakeSectionBS mtp h bs'
|
let bs = ncqMakeSectionBS mtp h bs'
|
||||||
let shard = ncqGetShard ncq h
|
let shard = ncqGetShard ncq h
|
||||||
zero <- newTVarIO Nothing
|
|
||||||
|
|
||||||
atomically do
|
atomically do
|
||||||
upd <- stateTVar shard $ flip HM.alterF h \case
|
upd <- readTVar shard <&> HM.lookup h >>= \case
|
||||||
Nothing -> (True, Just (NCQEntry bs zero))
|
Nothing -> do
|
||||||
Just e | ncqEntryData e /= bs -> (True, Just (NCQEntry bs zero))
|
here <- newTVar (EntryHere bs)
|
||||||
| otherwise -> (False, Just e)
|
modifyTVar shard (HM.insert h (NCQEntry here))
|
||||||
|
pure True
|
||||||
|
|
||||||
|
Just (NCQEntry e) -> readTVar e >>= \case
|
||||||
|
EntryHere bs'' | bs == bs''-> pure False
|
||||||
|
| otherwise -> writeTVar e (EntryHere bs) >> pure True
|
||||||
|
|
||||||
|
EntryThere{} -> writeTVar e (EntryHere bs) >> pure True
|
||||||
|
|
||||||
when upd do
|
when upd do
|
||||||
modifyTVar ncqWriteQ (|> h)
|
modifyTVar ncqWriteQ (|> h)
|
||||||
|
@ -287,14 +294,28 @@ instance IsTomb Location where
|
||||||
(_, Right (T, _)) -> True
|
(_, Right (T, _)) -> True
|
||||||
_ -> False
|
_ -> False
|
||||||
|
|
||||||
|
instance IsTomb FileLocation where
|
||||||
|
ncqIsTomb (FileLocation _ _ s) = ncqIsTombEntrySize s
|
||||||
|
|
||||||
ncqGetEntryBS :: MonadUnliftIO m => NCQStorage -> Location -> m (Maybe ByteString)
|
ncqGetEntryBS :: MonadUnliftIO m => NCQStorage -> Location -> m (Maybe ByteString)
|
||||||
ncqGetEntryBS me = \case
|
ncqGetEntryBS me = \case
|
||||||
InMemory bs -> pure $ Just bs
|
InMemory bs -> pure $ Just bs
|
||||||
InFossil (FileLocation fk off size) -> ncqWithState me $ const do
|
|
||||||
try @_ @SomeException (ncqGetCachedData me fk) >>= \case
|
InFossil l@(FileLocation fk off size) -> flip fix (0 :: Int) \next i -> do
|
||||||
Left e -> err (viaShow e) >> pure Nothing
|
ncqWithState me $ const do
|
||||||
Right (CachedData mmap) -> do
|
try @_ @SomeException (ncqGetCachedData me fk) >>= \case
|
||||||
pure $ Just $ BS.take (fromIntegral size) $ BS.drop (fromIntegral off) mmap
|
Left e -> err (viaShow e) >> pure Nothing
|
||||||
|
Right (CachedData mmap) -> do
|
||||||
|
|
||||||
|
if | BS.length mmap >= fromIntegral off + fromIntegral size -> do
|
||||||
|
pure $ Just $ BS.take (fromIntegral size) $ BS.drop (fromIntegral off) mmap
|
||||||
|
|
||||||
|
| i < 1 -> do
|
||||||
|
atomically (ncqDelCachedDataSTM me fk) >> next (succ i)
|
||||||
|
|
||||||
|
| otherwise -> do
|
||||||
|
err $ red "can't remap fossil" <+> pretty l
|
||||||
|
pure Nothing
|
||||||
|
|
||||||
ncqEntrySize :: forall a . Integral a => Location -> a
|
ncqEntrySize :: forall a . Integral a => Location -> a
|
||||||
ncqEntrySize = \case
|
ncqEntrySize = \case
|
||||||
|
|
|
@ -130,9 +130,13 @@ ncqIndexFile n ts' fk = runMaybeT do
|
||||||
|
|
||||||
nwayHashScanAll nw bs $ \_ k _ -> do
|
nwayHashScanAll nw bs $ \_ k _ -> do
|
||||||
unless (k == emptyKey) $ atomically $ void $ runMaybeT do
|
unless (k == emptyKey) $ atomically $ void $ runMaybeT do
|
||||||
NCQEntry _ tfk <- MaybeT $ ncqLookupEntrySTM n (coerce k)
|
e <- MaybeT $ ncqLookupEntrySTM n (coerce k)
|
||||||
fk' <- MaybeT $ readTVar tfk
|
|
||||||
guard (coerce fk == flKey fk') -- remove only own stuff
|
fk' <- MaybeT $ case snd e of
|
||||||
|
EntryHere{} -> pure Nothing
|
||||||
|
EntryThere l -> pure $ Just (flKey l)
|
||||||
|
|
||||||
|
guard (coerce fk == fk') -- remove only own stuff
|
||||||
lift $ ncqAlterEntrySTM n (coerce k) (const Nothing)
|
lift $ ncqAlterEntrySTM n (coerce k) (const Nothing)
|
||||||
|
|
||||||
pure dest
|
pure dest
|
||||||
|
|
|
@ -18,9 +18,11 @@ ncqGetShard :: NCQStorage -> HashRef -> Shard
|
||||||
ncqGetShard ncq@NCQStorage{..} h = ncqMemTable ! ncqShardIdx ncq h
|
ncqGetShard ncq@NCQStorage{..} h = ncqMemTable ! ncqShardIdx ncq h
|
||||||
{-# INLINE ncqGetShard #-}
|
{-# INLINE ncqGetShard #-}
|
||||||
|
|
||||||
|
ncqLookupEntrySTM :: NCQStorage -> HashRef -> STM (Maybe (NCQEntry, NCQEntryL))
|
||||||
ncqLookupEntrySTM :: NCQStorage -> HashRef -> STM (Maybe NCQEntry)
|
ncqLookupEntrySTM ncq h = readTVar (ncqGetShard ncq h)
|
||||||
ncqLookupEntrySTM ncq h = readTVar (ncqGetShard ncq h) <&> HM.lookup h
|
<&> HM.lookup h >>= \case
|
||||||
|
Nothing -> pure Nothing
|
||||||
|
Just e@(NCQEntry v)-> Just . (e,) <$> readTVar v
|
||||||
|
|
||||||
ncqAlterEntrySTM :: NCQStorage
|
ncqAlterEntrySTM :: NCQStorage
|
||||||
-> HashRef
|
-> HashRef
|
||||||
|
|
|
@ -67,7 +67,8 @@ ncqStorageRun ncq@NCQStorage{..} = flip runContT pure do
|
||||||
|
|
||||||
atomically (ncqLookupEntrySTM ncq h) >>= \case
|
atomically (ncqLookupEntrySTM ncq h) >>= \case
|
||||||
Nothing -> none
|
Nothing -> none
|
||||||
Just e -> answer (Just (InMemory (ncqEntryData e))) >> exit ()
|
Just (_, EntryHere bs) -> answer (Just (InMemory bs)) >> exit ()
|
||||||
|
Just (_, EntryThere loc) -> answer (Just $ InFossil loc) >> exit ()
|
||||||
|
|
||||||
ContT $ ncqWithState ncq
|
ContT $ ncqWithState ncq
|
||||||
|
|
||||||
|
@ -195,10 +196,10 @@ ncqStorageRun ncq@NCQStorage{..} = flip runContT pure do
|
||||||
Just (Right chu) -> do
|
Just (Right chu) -> do
|
||||||
ws <- for chu $ \h -> do
|
ws <- for chu $ \h -> do
|
||||||
atomically (ncqLookupEntrySTM ncq h) >>= \case
|
atomically (ncqLookupEntrySTM ncq h) >>= \case
|
||||||
Just (NCQEntry bs w) -> do
|
Just (NCQEntry w, EntryHere bs) -> do
|
||||||
let off = fromIntegral total'
|
off <- fromIntegral <$> liftIO (fdSeek fh RelativeSeek 0)
|
||||||
n <- lift (appendSection fh bs)
|
n <- lift (appendSection fh bs)
|
||||||
atomically (writeTVar w (Just (FileLocation fk off (fromIntegral n))))
|
atomically (writeTVar w (EntryThere (FileLocation fk off (fromIntegral n))))
|
||||||
pure n
|
pure n
|
||||||
|
|
||||||
_ -> pure 0
|
_ -> pure 0
|
||||||
|
|
|
@ -43,20 +43,21 @@ data FileLocation =
|
||||||
}
|
}
|
||||||
deriving stock (Eq,Ord)
|
deriving stock (Eq,Ord)
|
||||||
|
|
||||||
|
|
||||||
data Location =
|
data Location =
|
||||||
InFossil {-# UNPACK #-} !FileLocation
|
InFossil {-# UNPACK #-} !FileLocation
|
||||||
| InMemory {-# UNPACK #-} !ByteString
|
| InMemory {-# UNPACK #-} !ByteString
|
||||||
|
|
||||||
data NCQEntry =
|
newtype NCQEntry = NCQEntry (TVar NCQEntryL)
|
||||||
NCQEntry
|
-- NCQEntry
|
||||||
{ ncqEntryData :: !ByteString
|
-- { ncqEntryData :: !ByteString
|
||||||
, ncqDumped :: !(TVar (Maybe FileLocation))
|
-- , ncqDumped :: !(TVar (Maybe FileLocation))
|
||||||
}
|
-- }
|
||||||
|
|
||||||
type NCQOffset = Word64
|
data NCQEntryL = EntryHere !ByteString | EntryThere !FileLocation
|
||||||
|
|
||||||
|
type NCQOffset = Word64
|
||||||
type NCQFileSize = NCQOffset
|
type NCQFileSize = NCQOffset
|
||||||
type NCQSize = Word32
|
type NCQSize = Word32
|
||||||
|
|
||||||
data Fact = P PData -- pending, not indexed
|
data Fact = P PData -- pending, not indexed
|
||||||
deriving stock (Eq,Ord,Data)
|
deriving stock (Eq,Ord,Data)
|
||||||
|
@ -151,6 +152,9 @@ instance Semigroup NCQState where
|
||||||
facts = ncqStateFacts a <> ncqStateFacts b
|
facts = ncqStateFacts a <> ncqStateFacts b
|
||||||
|
|
||||||
|
|
||||||
|
instance Pretty FileLocation where
|
||||||
|
pretty (FileLocation f o s) = parens ("file-location" <+> pretty f <+> pretty o <+> pretty s)
|
||||||
|
|
||||||
instance Pretty Location where
|
instance Pretty Location where
|
||||||
pretty = \case
|
pretty = \case
|
||||||
InFossil (FileLocation k o s) -> parens $ "in-fossil" <+> pretty k <+> pretty o <+> pretty s
|
InFossil (FileLocation k o s) -> parens $ "in-fossil" <+> pretty k <+> pretty o <+> pretty s
|
||||||
|
@ -216,9 +220,10 @@ ncqDeferredWriteOpSTM NCQStorage{..} work = do
|
||||||
nw <- readTVar ncqWrites <&> (`mod` V.length ncqWriteOps)
|
nw <- readTVar ncqWrites <&> (`mod` V.length ncqWriteOps)
|
||||||
writeTQueue (ncqWriteOps ! nw) work
|
writeTQueue (ncqWriteOps ! nw) work
|
||||||
|
|
||||||
|
{- HLINT Ignore "Eta reduction"-}
|
||||||
|
|
||||||
logErr :: forall x a m . (Pretty x, MonadUnliftIO m) => x -> m a -> m a
|
logErr :: forall x a m . (Pretty x, MonadUnliftIO m) => x -> m a -> m a
|
||||||
logErr loc m = handle (\(e::SomeException) -> err (pretty loc <> ":" <> viaShow e) >> throwIO e) m
|
logErr loc m = handle (\(e::SomeException) -> err (pretty loc <> ":" <> viaShow e) >> throwIO e) m
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -440,7 +440,6 @@ ncq3Tests = do
|
||||||
|
|
||||||
entry $ bindMatch "test:ncq3:concurrent1" $ nil_ $ \case
|
entry $ bindMatch "test:ncq3:concurrent1" $ nil_ $ \case
|
||||||
[ LitIntVal tn, LitIntVal n ] -> do
|
[ LitIntVal tn, LitIntVal n ] -> do
|
||||||
debug $ "ncq2:concurrent1" <+> pretty tn <+> pretty n
|
|
||||||
runTest $ testNCQ3Concurrent1 False ( fromIntegral tn) (fromIntegral n)
|
runTest $ testNCQ3Concurrent1 False ( fromIntegral tn) (fromIntegral n)
|
||||||
e -> throwIO $ BadFormException @C (mkList e)
|
e -> throwIO $ BadFormException @C (mkList e)
|
||||||
|
|
||||||
|
|
|
@ -1423,7 +1423,7 @@ internalEntries = do
|
||||||
entry $ bindMatch "coalesce" $ \case
|
entry $ bindMatch "coalesce" $ \case
|
||||||
[a] -> pure a
|
[a] -> pure a
|
||||||
[a,b] | isFalse b -> pure a
|
[a,b] | isFalse b -> pure a
|
||||||
[a,_] -> pure a
|
| otherwise -> pure b
|
||||||
_ -> pure nil
|
_ -> pure nil
|
||||||
|
|
||||||
entry $ bindAlias "nvl" "coalesce"
|
entry $ bindAlias "nvl" "coalesce"
|
||||||
|
|
Loading…
Reference in New Issue