add LRU cache to stake address select & inserts

Author: Cmdv

cardano-db-sync/src/Cardano/DbSync/Api.hs

@@ -97,6 +97,7 @@ import Ouroboros.Consensus.Protocol.Abstract (ConsensusProtocol)
 import Ouroboros.Network.Block (BlockNo (..), Point (..))
 import Ouroboros.Network.Magic (NetworkMagic (..))
 import qualified Ouroboros.Network.Point as Point
+import Cardano.DbSync.Cache.LRU (LRUCacheCapacity(..))
 
 setConsistentLevel :: SyncEnv -> ConsistentLevel -> IO ()
 setConsistentLevel env cst = do
@@ -382,7 +383,15 @@ mkSyncEnv ::
   IO SyncEnv
 mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP ranMigrations runMigrationFnc = do
   dbCNamesVar <- newTVarIO =<< dbConstraintNamesExists backend
-  cache <- if soptCache syncOptions then newEmptyCache 250000 50000 else pure useNoCache
+  cache <-
+    if soptCache syncOptions
+      then newEmptyCache
+            LRUCacheCapacity
+            { lirCapacityStakeHashRaw = 300000,
+              lruCapacityDatum = 250000,
+              lruCapacityMultiAsset = 50000
+            }
+      else pure cacheStatusNoCache
   consistentLevelVar <- newTVarIO Unchecked
   fixDataVar <- newTVarIO $ if ranMigrations then DataFixRan else NoneFixRan
   indexesVar <- newTVarIO $ enpForceIndexes syncNP

cardano-db-sync/src/Cardano/DbSync/Cache.hs

@@ -83,106 +83,139 @@ getCacheStatistics cs =
 
 queryOrInsertRewardAccount ::
   (MonadBaseControl IO m, MonadIO m) =>
+  Trace IO Text ->
   CacheStatus ->
   CacheUpdateAction ->
   Ledger.RewardAccount StandardCrypto ->
   ReaderT SqlBackend m DB.StakeAddressId
-queryOrInsertRewardAccount cacheStatus cacheUA rewardAddr = do
-  eiAddrId <- queryRewardAccountWithCacheRetBs cacheStatus cacheUA rewardAddr
+queryOrInsertRewardAccount trce cacheStatus cacheUA rewardAddr = do
+  eiAddrId <- queryRewardAccountWithCacheRetBs trce cacheStatus rewardAddr
   case eiAddrId of
-    Left (_err, bs) -> insertStakeAddress rewardAddr (Just bs)
+    Left (_err, bs) -> insertStakeAddress cacheStatus rewardAddr (Just bs)
     Right addrId -> pure addrId
 
 queryOrInsertStakeAddress ::
   (MonadBaseControl IO m, MonadIO m) =>
+  Trace IO Text ->
   CacheStatus ->
   CacheUpdateAction ->
   Network ->
   StakeCred ->
   ReaderT SqlBackend m DB.StakeAddressId
-queryOrInsertStakeAddress cacheStatus cacheUA nw cred =
-  queryOrInsertRewardAccount cacheStatus cacheUA $ Ledger.RewardAccount nw cred
+queryOrInsertStakeAddress trce cacheStatus cacheUA nw cred =
+  queryOrInsertRewardAccount trce cacheStatus cacheUA $ Ledger.RewardAccount nw cred
 
 -- If the address already exists in the table, it will not be inserted again (due to
 -- the uniqueness constraint) but the function will return the 'StakeAddressId'.
 insertStakeAddress ::
   (MonadBaseControl IO m, MonadIO m) =>
+  CacheStatus ->
   Ledger.RewardAccount StandardCrypto ->
   Maybe ByteString ->
   ReaderT SqlBackend m DB.StakeAddressId
-insertStakeAddress rewardAddr stakeCredBs =
-  DB.insertStakeAddress $
-    DB.StakeAddress
-      { DB.stakeAddressHashRaw = addrBs
-      , DB.stakeAddressView = Generic.renderRewardAccount rewardAddr
-      , DB.stakeAddressScriptHash = Generic.getCredentialScriptHash $ Ledger.raCredential rewardAddr
-      }
+insertStakeAddress cacheStatus rewardAddr stakeCredBs = do
+  addrId <- DB.insertStakeAddress $
+      DB.StakeAddress
+        { DB.stakeAddressHashRaw = addrBs
+        , DB.stakeAddressView = Generic.renderRewardAccount rewardAddr
+        , DB.stakeAddressScriptHash = Generic.getCredentialScriptHash $ Ledger.raCredential rewardAddr
+        }
+  case cacheStatus of
+    NoCache -> pure addrId
+    CacheActive ci -> do
+      liftIO $ atomically $ modifyTVar (cStakeRawHashes ci) $
+        LRU.insert addrBs addrId
+      pure addrId
   where
     addrBs = fromMaybe (Ledger.serialiseRewardAccount rewardAddr) stakeCredBs
 
 queryRewardAccountWithCacheRetBs ::
   forall m.
   MonadIO m =>
+  Trace IO Text ->
   CacheStatus ->
   CacheUpdateAction ->
   Ledger.RewardAccount StandardCrypto ->
   ReaderT SqlBackend m (Either (DB.LookupFail, ByteString) DB.StakeAddressId)
-queryRewardAccountWithCacheRetBs cacheStatus cacheUA rwdAcc =
-  queryStakeAddrWithCacheRetBs cacheStatus cacheUA (Ledger.raNetwork rwdAcc) (Ledger.raCredential rwdAcc)
+queryRewardAccountWithCacheRetBs trce cacheStatus cacheUA rwdAcc =
+  queryStakeAddrWithCacheRetBs trce cacheStatus cacheUA (Ledger.raNetwork rwdAcc) (Ledger.raCredential rwdAcc)
 
 queryStakeAddrWithCache ::
   forall m.
   MonadIO m =>
+  Trace IO Text ->
   CacheStatus ->
   CacheUpdateAction ->
   Network ->
   StakeCred ->
   ReaderT SqlBackend m (Either DB.LookupFail DB.StakeAddressId)
-queryStakeAddrWithCache cacheStatus cacheUA nw cred =
-  mapLeft fst <$> queryStakeAddrWithCacheRetBs cacheStatus cacheUA nw cred
+queryStakeAddrWithCache trce cacheStatus cacheUA nw cred =
+  mapLeft fst <$> queryStakeAddrWithCacheRetBs trce cacheStatus cacheUA nw cred
 
 queryStakeAddrWithCacheRetBs ::
   forall m.
   MonadIO m =>
+  Trace IO Text ->
   CacheStatus ->
   CacheUpdateAction ->
   Network ->
   StakeCred ->
   ReaderT SqlBackend m (Either (DB.LookupFail, ByteString) DB.StakeAddressId)
-queryStakeAddrWithCacheRetBs cacheStatus cacheUA nw cred = do
+queryStakeAddrWithCacheRetBs trce cacheStatus cacheUA nw cred = do
+  let !bs = Ledger.serialiseRewardAccount (Ledger.RewardAccount nw cred)
   case cacheStatus of
     NoCache -> do
-      let !bs = Ledger.serialiseRewardAccount (Ledger.RewardAccount nw cred)
       mapLeft (,bs) <$> queryStakeAddress bs
-    ActiveCache ci -> do
-      mp <- liftIO $ readTVarIO (cStakeCreds ci)
-      (mAddrId, mp') <- queryStakeAddrAux cacheUA mp (cStats ci) nw cred
-      liftIO $ atomically $ writeTVar (cStakeCreds ci) mp'
-      pure mAddrId
+    CacheActive ci -> do
+      currentCache <- liftIO $ readTVarIO (cStakeRawHashes ci)
+      let cacheSize = LRU.getSize currentCache
+      newCache <-
+            if cacheSize < 1
+              then do
+                liftIO $ logInfo trce "----------------- Cache is empty. Querying all addresses. ---------"
+                queryRes <- DB.queryLatestAddresses cacheSize
+                pure $ LRU.fromList queryRes currentCache
+                -- convert the results into the cache
+              else pure currentCache
+      case LRU.lookup bs newCache of
+        Just (addrId, mp') -> do
+          liftIO $ hitCreds (cStats ci)
+          liftIO $ atomically $ writeTVar (cStakeRawHashes ci) mp'
+          pure $ Right addrId
+        Nothing -> do
+          liftIO $ missCreds (cStats ci)
+          liftIO $ atomically $ writeTVar (cStakeRawHashes ci) newCache
+          queryRes <- mapLeft (,bs) <$> queryStakeAddress bs
+          case queryRes of
+            Left _ -> pure queryRes
+            Right stakeAddrsId -> do
+              liftIO $ atomically $ modifyTVar (cStakeRawHashes ci) $
+               LRU.insert bs stakeAddrsId
+              pure $ Right stakeAddrsId
 
-queryStakeAddrAux ::
-  MonadIO m =>
-  CacheUpdateAction ->
-  StakeAddrCache ->
-  StrictTVar IO CacheStatistics ->
-  Network ->
-  StakeCred ->
-  ReaderT SqlBackend m (Either (DB.LookupFail, ByteString) DB.StakeAddressId, StakeAddrCache)
-queryStakeAddrAux cacheUA mp sts nw cred =
-  case Map.lookup cred mp of
-    Just addrId -> do
-      liftIO $ hitCreds sts
-      case cacheUA of
-        EvictAndUpdateCache -> pure (Right addrId, Map.delete cred mp)
-        _ -> pure (Right addrId, mp)
-    Nothing -> do
-      liftIO $ missCreds sts
-      let !bs = Ledger.serialiseRewardAccount (Ledger.RewardAccount nw cred)
-      mAddrId <- mapLeft (,bs) <$> queryStakeAddress bs
-      case (mAddrId, cacheUA) of
-        (Right addrId, UpdateCache) -> pure (Right addrId, Map.insert cred addrId mp)
-        (Right addrId, _) -> pure (Right addrId, mp)
-        (err, _) -> pure (err, mp)
+-- queryStakeAddrAux ::
+--   MonadIO m =>
+--   CacheNew ->
+--   StakeAddrCache ->
+--   StrictTVar IO CacheStatistics ->
+--   Network ->
+--   StakeCred ->
+--   ReaderT SqlBackend m (Either (DB.LookupFail, ByteString) DB.StakeAddressId, StakeAddrCache)
+-- queryStakeAddrAux cacheNew mp sts nw cred =
+--   case Map.lookup cred mp of
+--     Just addrId -> do
+--       liftIO $ hitCreds sts
+--       case cacheNew of
+--         EvictAndReturn -> pure (Right addrId, Map.delete cred mp)
+--         _ -> pure (Right addrId, mp)
+--     Nothing -> do
+--       liftIO $ missCreds sts
+--       let !bs = Ledger.serialiseRewardAccount (Ledger.RewardAccount nw cred)
+--       mAddrId <- mapLeft (,bs) <$> queryStakeAddress bs
+--       case (mAddrId, cacheNew) of
+--         (Right addrId, CacheNew) -> pure (Right addrId, Map.insert cred addrId mp)
+--         (Right addrId, _) -> pure (Right addrId, mp)
+--         (err, _) -> pure (err, mp)
 
 queryPoolKeyWithCache ::
   MonadIO m =>
@@ -372,7 +405,7 @@ queryDatum ::
 queryDatum cacheStatus hsh = do
   case cacheStatus of
     NoCache -> DB.queryDatum $ Generic.dataHashToBytes hsh
-    ActiveCache ci -> do
+    CacheActive ci -> do
       mp <- liftIO $ readTVarIO (cDatum ci)
       case LRU.lookup hsh mp of
         Just (datumId, mp') -> do

cardano-db-sync/src/Cardano/DbSync/Cache/Epoch.hs

@@ -26,25 +26,27 @@ import Database.Persist.Postgresql (SqlBackend)
 -- Epoch Cache
 -------------------------------------------------------------------------------------
 readCacheEpoch :: MonadIO m => CacheStatus -> m (Maybe CacheEpoch)
-readCacheEpoch cache =
-  case cache of
+readCacheEpoch cacheStatus =
+  case cacheStatus of
     NoCache -> pure Nothing
     ActiveCache ci -> do
       cacheEpoch <- liftIO $ readTVarIO (cEpoch ci)
       pure $ Just cacheEpoch
 
 readEpochBlockDiffFromCache :: MonadIO m => CacheStatus -> m (Maybe EpochBlockDiff)
-readEpochBlockDiffFromCache cache =
-  case cache of
+
+readEpochBlockDiffFromCache cacheStatus =
+  case cacheStatus of
     NoCache -> pure Nothing
     ActiveCache ci -> do
       cE <- liftIO $ readTVarIO (cEpoch ci)
       case (ceMapEpoch cE, ceEpochBlockDiff cE) of
         (_, epochInternal) -> pure epochInternal
 
 readLastMapEpochFromCache :: CacheStatus -> IO (Maybe DB.Epoch)
-readLastMapEpochFromCache cache =
-  case cache of
+
+readLastMapEpochFromCache cacheStatus =
+  case cacheStatus of
     NoCache -> pure Nothing
     ActiveCache ci -> do
       cE <- readTVarIO (cEpoch ci)
@@ -57,22 +59,23 @@ readLastMapEpochFromCache cache =
           Just (_, ep) -> pure $ Just ep
 
 rollbackMapEpochInCache :: MonadIO m => CacheInternal -> DB.BlockId -> m (Either SyncNodeError ())
-rollbackMapEpochInCache cache blockId = do
-  cE <- liftIO $ readTVarIO (cEpoch cache)
+rollbackMapEpochInCache cacheInternal blockId = do
+  cE <- liftIO $ readTVarIO (cEpoch cacheInternal)
   -- split the map and delete anything after blockId including it self as new blockId might be
   -- given when inserting the block again when doing rollbacks.
   let (newMapEpoch, _) = split blockId (ceMapEpoch cE)
-  writeToCache cache (CacheEpoch newMapEpoch (ceEpochBlockDiff cE))
+  writeToCache cacheInternal (CacheEpoch newMapEpoch (ceEpochBlockDiff cE))
 
 writeEpochBlockDiffToCache ::
   MonadIO m =>
   CacheStatus ->
   EpochBlockDiff ->
   ReaderT SqlBackend m (Either SyncNodeError ())
-writeEpochBlockDiffToCache cache epCurrent =
-  case cache of
-    NoCache -> pure $ Left $ SNErrDefault "writeEpochBlockDiffToCache: CacheStatus is NoCache"
-    ActiveCache ci -> do
+
+writeEpochBlockDiffToCache cacheStatus epCurrent =
+  case cacheStatus of
+    NoCache -> pure $ Left $ SNErrDefault "writeEpochBlockDiffToCache: Cache is NoCache"
+    CacheActive ci -> do
       cE <- liftIO $ readTVarIO (cEpoch ci)
       case (ceMapEpoch cE, ceEpochBlockDiff cE) of
         (epochLatest, _) -> writeToCache ci (CacheEpoch epochLatest (Just epCurrent))
@@ -86,17 +89,17 @@ writeToMapEpochCache ::
   CacheStatus ->
   DB.Epoch ->
   ReaderT SqlBackend m (Either SyncNodeError ())
-writeToMapEpochCache syncEnv cache latestEpoch = do
+writeToMapEpochCache syncEnv cacheStatus latestEpoch = do
   -- this can also be tought of as max rollback number
   let securityParam =
         case envLedgerEnv syncEnv of
           HasLedger hle -> getSecurityParameter $ leProtocolInfo hle
           NoLedger nle -> getSecurityParameter $ nleProtocolInfo nle
-  case cache of
-    NoCache -> pure $ Left $ SNErrDefault "writeToMapEpochCache: CacheStatus is NoCache"
+  case cacheStatus of
+    NoCache -> pure $ Left $ SNErrDefault "writeToMapEpochCache: Cache is NoCache"
     ActiveCache ci -> do
       -- get EpochBlockDiff so we can use the BlockId we stored when inserting blocks
-      epochInternalCE <- readEpochBlockDiffFromCache cache
+      epochInternalCE <- readEpochBlockDiffFromCache cacheStatus
       case epochInternalCE of
         Nothing -> pure $ Left $ SNErrDefault "writeToMapEpochCache: No epochInternalEpochCache"
         Just ei -> do