Extract also LMDB backend

Author: jasagredo

cabal.project

@@ -20,6 +20,7 @@ index-state:
 
 packages:
   ouroboros-consensus
+  ouroboros-consensus-lmdb
   ouroboros-consensus-lsm
   ouroboros-consensus-cardano
   ouroboros-consensus-protocol

ouroboros-consensus-cardano/src/ouroboros-consensus-cardano/Ouroboros/Consensus/Cardano/StreamingLedgerTables.hs renamed to ouroboros-consensus-cardano/app/Ouroboros/Consensus/Cardano/StreamingLedgerTables.hs

@@ -160,8 +160,6 @@ library
     contra-tracer,
     crypton,
     deepseq,
-    directory,
-    filepath,
     formatting >=6.3 && <7.3,
     fs-api,
     measures,
@@ -700,6 +698,9 @@ executable immdb-server
 executable snapshot-converter
   import: common-exe
   hs-source-dirs: app
+  other-modules:
+      Ouroboros.Consensus.Cardano.StreamingLedgerTables
+
   main-is: snapshot-converter.hs
   build-depends:
     ansi-terminal,
@@ -711,6 +712,8 @@ executable snapshot-converter
     mtl,
     optparse-applicative,
     ouroboros-consensus,
+    ouroboros-consensus-lsm,
+    ouroboros-consensus-lmdb,
     ouroboros-consensus-cardano,
     ouroboros-consensus-cardano:unstable-cardano-tools,
     resource-registry,

ouroboros-consensus-cardano/ouroboros-consensus-cardano.cabal

@@ -53,7 +53,6 @@ module Ouroboros.Consensus.Node
   , pattern DoDiskSnapshotChecksum
   , pattern NoDoDiskSnapshotChecksum
   , ChainSyncIdleTimeout (..)
-  , LedgerDbBackendArgs (..)
 
     -- * Internal helpers
   , mkNodeKernelArgs
@@ -127,8 +126,6 @@ import qualified Ouroboros.Consensus.Storage.ChainDB as ChainDB
 import qualified Ouroboros.Consensus.Storage.ChainDB.Impl.Args as ChainDB
 import Ouroboros.Consensus.Storage.LedgerDB.Args
 import Ouroboros.Consensus.Storage.LedgerDB.Snapshots
-import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.Args as V2
-import qualified Ouroboros.Consensus.Storage.LedgerDB.V2.LSM as LSM
 import Ouroboros.Consensus.Util.Args
 import Ouroboros.Consensus.Util.IOLike
 import Ouroboros.Consensus.Util.Orphans ()
@@ -320,7 +317,7 @@ data LowLevelRunNodeArgs m addrNTN addrNTC blk
   , llrnMaxClockSkew :: InFutureCheck.ClockSkew
   -- ^ Maximum clock skew
   , llrnPublicPeerSelectionStateVar :: StrictSTM.StrictTVar m (PublicPeerSelectionState addrNTN)
-  , llrnLdbFlavorArgs :: Complete LedgerDbFlavorArgs m
+  , llrnLdbFlavorArgs :: LedgerDbBackendArgs m blk
   -- ^ The flavor arguments
   }
 
@@ -378,7 +375,7 @@ data
   , -- Ad hoc values to replace default ChainDB configurations
     srnSnapshotPolicyArgs :: SnapshotPolicyArgs
   , srnQueryBatchSize :: QueryBatchSize
-  , srnLedgerDbBackendArgs :: LedgerDbBackendArgs m
+  , srnLedgerDbBackendArgs :: LedgerDbBackendArgs m blk
   }
 
 {-------------------------------------------------------------------------------
@@ -819,7 +816,7 @@ openChainDB ::
   (ChainDB.RelativeMountPoint -> SomeHasFS m) ->
   -- | Volatile FS, see 'NodeDatabasePaths'
   (ChainDB.RelativeMountPoint -> SomeHasFS m) ->
-  Complete LedgerDbFlavorArgs m ->
+  LedgerDbBackendArgs m blk ->
   -- | A set of default arguments (possibly modified from 'defaultArgs')
   Incomplete ChainDbArgs m blk ->
   -- | Customise the 'ChainDbArgs'
@@ -1007,7 +1004,7 @@ stdLowLevelRunNodeArgsIO
     }
   $(SafeWildCards.fields 'StdRunNodeArgs) = do
     llrnBfcSalt <- stdBfcSaltIO
-    (lsmSalt, llrnRng) <- genWord64 <$> newStdGen
+    llrnRng <- newStdGen
     pure
       LowLevelRunNodeArgs
         { llrnBfcSalt
@@ -1052,21 +1049,21 @@ stdLowLevelRunNodeArgsIO
             InFutureCheck.defaultClockSkew
         , llrnPublicPeerSelectionStateVar =
             Diffusion.dcPublicPeerSelectionVar srnDiffusionConfiguration
-        , llrnLdbFlavorArgs =
-            case srnLedgerDbBackendArgs of
-              V1LMDB args -> LedgerDbFlavorArgsV1 args
-              V2InMemory -> LedgerDbFlavorArgsV2 (V2.V2Args V2.InMemoryHandleArgs)
-              V2LSM path ->
-                LedgerDbFlavorArgsV2
-                  ( V2.V2Args
-                      ( V2.LSMHandleArgs
-                          ( V2.LSMArgs
-                              (mkFsPath $ splitDirectories path)
-                              lsmSalt
-                              (LSM.stdMkBlockIOFS (nonImmutableDbPath srnDatabasePath))
-                          )
-                      )
-                  )
+        , llrnLdbFlavorArgs = srnLedgerDbBackendArgs
+        -- case srnLedgerDbBackendArgs of
+        --   V1LMDB args -> LedgerDbFlavorArgsV1 args
+        --   V2InMemory -> LedgerDbFlavorArgsV2 (V2.V2Args V2.InMemoryHandleArgs)
+        --   V2LSM path ->
+        --     LedgerDbFlavorArgsV2
+        --       ( V2.V2Args
+        --           ( V2.LSMHandleArgs
+        --               ( V2.LSMArgs
+        --                   (mkFsPath $ splitDirectories path)
+        --                   lsmSalt
+        --                   (LSM.stdMkBlockIOFS (nonImmutableDbPath srnDatabasePath))
+        --               )
+        --           )
+        --       )
         }
    where
     networkMagic :: NetworkMagic

ouroboros-consensus-diffusion/src/ouroboros-consensus-diffusion/Ouroboros/Consensus/Node.hs

@@ -0,0 +1,93 @@
+cabal-version: 3.0
+name: ouroboros-consensus-lmdb
+version: 0.1.0.0
+synopsis: LMDB LedgerDB V1 backend for ouroboros-consensus
+description: LMDB LedgerDB V1 backend for ouroboros-consensus.
+license: Apache-2.0
+license-files:
+  LICENSE
+  NOTICE
+
+copyright:
+  2019-2023 Input Output Global Inc (IOG), INTERSECT 2023-2024.
+
+author: IOG Engineering Team
+maintainer: operations@iohk.io
+category: Network
+build-type: Simple
+extra-doc-files:
+  CHANGELOG.md
+
+source-repository head
+  type: git
+  location: https://github.com/IntersectMBO/ouroboros-consensus
+
+flag asserts
+  description: Enable assertions
+  manual: False
+  default: False
+
+common common-lib
+  default-language: Haskell2010
+  ghc-options:
+    -Wall
+    -Wcompat
+    -Wincomplete-uni-patterns
+    -Wincomplete-record-updates
+    -Wpartial-fields
+    -Widentities
+    -Wredundant-constraints
+    -Wmissing-export-lists
+    -Wunused-packages
+    -Wno-unticked-promoted-constructors
+
+  if flag(asserts)
+    ghc-options: -fno-ignore-asserts
+    cpp-options: -DENABLE_ASSERTIONS
+
+common common-test
+  import: common-lib
+  ghc-options:
+    -threaded
+    -rtsopts
+
+common common-bench
+  import: common-test
+  ghc-options:
+    -threaded
+    -rtsopts
+
+  -- We use this option to avoid skewed results due to changes in cache-line
+  -- alignment. See
+  -- https://github.com/Bodigrim/tasty-bench#comparison-against-baseline
+  ghc-options: -fproc-alignment=64
+
+library
+  import: common-lib
+  hs-source-dirs: src
+  exposed-modules:
+    Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB
+    Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Bridge
+    Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Status
+
+  build-depends:
+    base >=4.14 && <4.22,
+    bytestring >=0.10 && <0.13,
+    cardano-lmdb >=0.4,
+    cardano-lmdb-simple ^>=0.8,
+    cardano-slotting,
+    containers >=0.5 && <0.8,
+    contra-tracer,
+    fs-api ^>=0.4,
+    io-classes ^>=1.8.0.1,
+    mempack,
+    sop-core,
+    nothunks ^>=0.2,
+    ouroboros-consensus,
+    rawlock ^>=0.1.1,
+    serialise ^>=0.2,
+    mtl,
+    streaming,
+    text,
+
+  build-depends: text >=1.2.5.0 && <2.2

ouroboros-consensus-lmdb/ouroboros-consensus-lmdb.cabal

@@ -3,14 +3,19 @@
 {-# LANGUAGE DeriveGeneric #-}
 {-# LANGUAGE DerivingVia #-}
 {-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
+{-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE TypeOperators #-}
+{-# LANGUAGE UndecidableInstances #-}
 
 -- | A 'BackingStore' implementation based on [LMDB](http://www.lmdb.tech/doc/).
 module Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB
@@ -29,12 +34,13 @@ module Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB
   , withDbSeqNoRWMaybeNull
   ) where
 
-import Cardano.Slotting.Slot (SlotNo, WithOrigin (At))
+import Cardano.Slotting.Slot (WithOrigin (At))
 import qualified Codec.Serialise as S (Serialise (..))
 import qualified Control.Concurrent.Class.MonadSTM.TVar as IOLike
 import Control.Monad (forM_, unless, void, when)
 import qualified Control.Monad.Class.MonadSTM as IOLike
 import Control.Monad.IO.Class (MonadIO (liftIO))
+import Control.Monad.Trans (lift)
 import qualified Control.Tracer as Trace
 import Data.Bifunctor (first)
 import Data.Functor (($>), (<&>))
@@ -43,6 +49,7 @@ import Data.Map (Map)
 import qualified Data.Map.Strict as Map
 import Data.MemPack
 import Data.Proxy
+import qualified Data.SOP.Dict as Dict
 import qualified Data.Set as Set
 import qualified Data.Text as Strict
 import qualified Database.LMDB.Simple as LMDB
@@ -52,11 +59,14 @@ import qualified Database.LMDB.Simple.Internal as LMDB.Internal
 import qualified Database.LMDB.Simple.TransactionHandle as TrH
 import GHC.Generics (Generic)
 import GHC.Stack (HasCallStack)
-import Ouroboros.Consensus.Ledger.Tables
+import Ouroboros.Consensus.Block
+import Ouroboros.Consensus.Ledger.Basics
 import qualified Ouroboros.Consensus.Ledger.Tables.Diff as Diff
+import Ouroboros.Consensus.Storage.LedgerDB.API
 import Ouroboros.Consensus.Storage.LedgerDB.Snapshots
   ( SnapshotBackend (..)
   )
+import Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore
 import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.API as API
 import qualified Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Bridge as Bridge
 import Ouroboros.Consensus.Storage.LedgerDB.V1.BackingStore.Impl.LMDB.Status
@@ -70,10 +80,14 @@ import Ouroboros.Consensus.Util.IOLike
   , IOLike
   , MonadCatch (..)
   , MonadThrow (..)
+  , PrimState
   , bracket
   )
 import Ouroboros.Consensus.Util.IndexedMemPack
+import qualified Streaming as S
+import qualified Streaming.Prelude as S
 import qualified System.FS.API as FS
+import System.FS.IO
 
 {-------------------------------------------------------------------------------
   Database definition
@@ -793,3 +807,104 @@ prettyPrintLMDBErr = \case
   LMDBErrNotADir path ->
     "The path " <> show path <> " should be a directory but it is a file instead."
   LMDBErrClosed -> "The database has been closed."
+
+{-------------------------------------------------------------------------------
+  Backend
+-------------------------------------------------------------------------------}
+
+data LMDB
+
+instance
+  ( HasLedgerTables l
+  , MonadIO m
+  , IOLike m
+  , MemPackIdx l EmptyMK ~ l EmptyMK
+  ) =>
+  Backend m LMDB l
+  where
+  data Args m LMDB
+    = LMDBBackingStoreArgs FilePath LMDBLimits (Dict.Dict MonadIOPrim m)
+  data Trace m LMDB
+    = OnDiskBackingStoreInitialise LMDB.Limits
+    | OnDiskBackingStoreTrace BackingStoreTrace
+    deriving (Eq, Show)
+
+  isRightBackendForSnapshot _ _ UTxOHDLMDBSnapshot = True
+  isRightBackendForSnapshot _ _ _ = False
+
+  newBackingStoreInitialiser trcr (LMDBBackingStoreArgs fs limits Dict.Dict) =
+    newLMDBBackingStore
+      (SomeBackendTrace . OnDiskBackingStoreTrace >$< trcr)
+      limits
+      (LiveLMDBFS $ FS.SomeHasFS $ ioHasFS $ FS.MountPoint fs)
+
+{-------------------------------------------------------------------------------
+  Streaming
+-------------------------------------------------------------------------------}
+
+instance (Ord (TxIn l), GetTip l, Monad m) => StreamingBackend m LMDB l where
+  data SinkArgs m LMDB l
+    = SinkLMDB
+        -- \| Chunk size
+        Int
+        -- \| bsWrite
+        ( SlotNo ->
+          (l EmptyMK, l EmptyMK) ->
+          LedgerTables l DiffMK ->
+          m ()
+        )
+        (l EmptyMK -> m ())
+
+  data YieldArgs m LMDB l
+    = YieldLMDB
+        Int
+        (LedgerBackingStoreValueHandle m l)
+
+  yield _ (YieldLMDB chunkSize valueHandle) = yieldLmdbS chunkSize valueHandle
+  sink _ (SinkLMDB chunkSize write copy) = sinkLmdbS chunkSize write copy
+
+sinkLmdbS ::
+  forall m l.
+  (Ord (TxIn l), GetTip l, Monad m) =>
+  Int ->
+  (SlotNo -> (l EmptyMK, l EmptyMK) -> LedgerTables l DiffMK -> m ()) ->
+  (l EmptyMK -> m ()) ->
+  Sink m l
+sinkLmdbS writeChunkSize bs copyTo hint s = do
+  r <- go writeChunkSize mempty s
+  lift $ copyTo hint
+  pure (fmap (,Nothing) r)
+ where
+  sl = withOrigin (error "unreachable") id $ pointSlot $ getTip hint
+
+  go 0 m s' = do
+    lift $ bs sl (hint, hint) (LedgerTables $ DiffMK $ Diff.fromMapInserts m)
+    go writeChunkSize mempty s'
+  go n m s' = do
+    mbs <- S.uncons s'
+    case mbs of
+      Nothing -> do
+        lift $ bs sl (hint, hint) (LedgerTables $ DiffMK $ Diff.fromMapInserts m)
+        S.effects s'
+      Just ((k, v), s'') ->
+        go (n - 1) (Map.insert k v m) s''
+
+yieldLmdbS ::
+  Monad m =>
+  Int ->
+  LedgerBackingStoreValueHandle m l ->
+  Yield m l
+yieldLmdbS readChunkSize bsvh hint k = do
+  r <- k (go (RangeQuery Nothing readChunkSize))
+  lift $ S.effects r
+ where
+  go p = do
+    (LedgerTables (ValuesMK values), mx) <- lift $ S.lift $ bsvhRangeRead bsvh hint p
+    case mx of
+      Nothing -> pure $ pure Nothing
+      Just x -> do
+        S.each $ Map.toList values
+        go (RangeQuery (Just . LedgerTables . KeysMK $ Set.singleton x) readChunkSize)
+
+class (MonadIO m, PrimState m ~ PrimState IO) => MonadIOPrim m
+instance (MonadIO m, PrimState m ~ PrimState IO) => MonadIOPrim m