Skip to content

Commit 92c9850

Browse files
kdermekderme
committed
Add separate atomic connections for datum and scripts
Not done yet for Address variant
1 parent 04576b6 commit 92c9850

File tree

17 files changed

+253
-104
lines changed

17 files changed

+253
-104
lines changed

‎cardano-db-sync/cardano-db-sync.cabal

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,7 @@ library
135135
Cardano.DbSync.Threads.Database
136136
Cardano.DbSync.Threads.EpochStake
137137
Cardano.DbSync.Threads.Ledger
138+
Cardano.DbSync.Threads.MultiAsset
138139
Cardano.DbSync.Threads.Rewards
139140
Cardano.DbSync.Threads.Stake
140141
Cardano.DbSync.Threads.TxInResolve
@@ -217,6 +218,7 @@ library
217218
, prometheus
218219
, psqueues
219220
, random-shuffle
221+
, resourcet
220222
, resource-pool
221223
, scientific
222224
, serialise

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

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ import qualified Cardano.Crypto as Crypto
2929
import qualified Cardano.Db as DB
3030
import qualified Cardano.Db as Db
3131
import Cardano.DbSync.Api
32-
import Cardano.DbSync.Api.Types (InsertOptions (..), RunMigration, SyncEnv (..), SyncOptions (..), envLedgerEnv)
32+
import Cardano.DbSync.Api.Types
3333
import Cardano.DbSync.Config (configureLogging)
3434
import Cardano.DbSync.Config.Cardano
3535
import Cardano.DbSync.Config.Types
@@ -54,7 +54,7 @@ import Control.Monad.Extra (whenJust)
5454
import qualified Data.Strict.Maybe as Strict
5555
import qualified Data.Text as Text
5656
import Data.Version (showVersion)
57-
import Database.Persist.Postgresql (ConnectionString, withPostgresqlConn)
57+
import Database.Persist.Postgresql (ConnectionString)
5858
import qualified Ouroboros.Consensus.HardFork.Simple as HardFork
5959
import Ouroboros.Network.NodeToClient (IOManager, withIOManager)
6060
import Paths_cardano_db_sync (version)
@@ -167,17 +167,17 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
167167
logInfo trce $ "Using alonzo genesis file from: " <> (show . unGenesisFile $ dncAlonzoGenesisFile syncNodeConfigFromFile)
168168

169169
Db.runIohkLogging trce $
170-
withPostgresqlConn dbConnString $
171-
\backend -> liftIO $ do
170+
withDBSyncConnections dbConnString $
171+
\backends -> liftIO $ do
172172
runOrThrowIO $ runExceptT $ do
173173
genCfg <- readCardanoGenesisConfig syncNodeConfigFromFile
174-
isJsonbInSchema <- queryIsJsonbInSchema backend
174+
isJsonbInSchema <- queryIsJsonbInSchema (mainBackend backends)
175175
logProtocolMagicId trce $ genesisProtocolMagicId genCfg
176176
syncEnv <-
177177
ExceptT $
178178
mkSyncEnvFromConfig
179179
trce
180-
backend
180+
backends
181181
dbConnString
182182
syncOptions
183183
genCfg
@@ -197,7 +197,7 @@ runSyncNode metricsSetters trce iomgr dbConnString runMigrationFnc syncNodeConfi
197197
liftIO $ runExtraMigrationsMaybe syncEnv
198198
unless useLedger $ liftIO $ do
199199
logInfo trce "Migrating to a no ledger schema"
200-
Db.noLedgerMigrations backend trce
200+
Db.noLedgerMigrations (mainBackend backends) trce
201201
insertValidateGenesisDist syncEnv (dncNetworkName syncNodeConfigFromFile) genCfg (useShelleyInit syncNodeConfigFromFile)
202202

203203
-- communication channel between datalayer thread and chainsync-client thread

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

Lines changed: 54 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,9 @@ module Cardano.DbSync.Api (
2626
getHasConsumedOrPruneTxOut,
2727
getSkipTxIn,
2828
getPrunes,
29+
withDBSyncConnections,
30+
withScriptConnection,
31+
withDatumConnection,
2932
mkSyncEnvFromConfig,
3033
getInsertOptions,
3134
getTrace,
@@ -41,7 +44,7 @@ import qualified Cardano.Chain.Genesis as Byron
4144
import Cardano.Crypto.ProtocolMagic (ProtocolMagicId (..))
4245
import qualified Cardano.Db as DB
4346
import Cardano.DbSync.Api.Types
44-
import Cardano.DbSync.Cache.Types (CacheCapacity (..), newEmptyCache, newStakeChannels, useNoCache)
47+
import Cardano.DbSync.Cache.Types (CacheCapacity (..), newEmptyCache, newMAChannels, newStakeChannels, useNoCache)
4548
import Cardano.DbSync.Config.Cardano
4649
import Cardano.DbSync.Config.Shelley
4750
import Cardano.DbSync.Config.Types
@@ -65,9 +68,12 @@ import Control.Concurrent.Class.MonadSTM.Strict (
6568
writeTVar,
6669
)
6770
import qualified Control.Concurrent.Class.MonadSTM.Strict.TBQueue as TBQ
71+
import Control.Concurrent.MVar
72+
import Control.Monad.Logger (LoggingT, MonadLoggerIO)
73+
import Control.Monad.Trans.Resource (MonadUnliftIO)
6874
import qualified Data.Strict.Maybe as Strict
6975
import Data.Time.Clock (getCurrentTime)
70-
import Database.Persist.Postgresql (ConnectionString, createPostgresqlPool)
76+
import Database.Persist.Postgresql (ConnectionString, createPostgresqlPool, withPostgresqlConn)
7177
import Database.Persist.Sql (SqlBackend)
7278
import Ouroboros.Consensus.Block.Abstract (BlockProtocol)
7379
import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
@@ -248,7 +254,7 @@ writePrefetch syncEnv cblock = do
248254

249255
mkSyncEnv ::
250256
Trace IO Text ->
251-
SqlBackend ->
257+
DbConnections ->
252258
ConnectionString ->
253259
SyncOptions ->
254260
ProtocolInfo CardanoBlock ->
@@ -259,7 +265,7 @@ mkSyncEnv ::
259265
SyncNodeParams ->
260266
RunMigration ->
261267
IO SyncEnv
262-
mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runMigrationFnc = do
268+
mkSyncEnv trce backends connectionString syncOptions protoInfo nw nwMagic systemStart syncNodeConfigFromFile syncNP runMigrationFnc = do
263269
dbCNamesVar <- newTVarIO =<< dbConstraintNamesExists backend
264270
cache <-
265271
if soptCache syncOptions
@@ -278,8 +284,9 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
278284
indexesVar <- newTVarIO $ enpForceIndexes syncNP
279285
bts <- getBootstrapInProgress trce (isTxOutConsumedBootstrap' syncNodeConfigFromFile) backend
280286
bootstrapVar <- newTVarIO bts
281-
-- Offline Pool + Anchor queues
282287
cChans <- newStakeChannels
288+
maChans <- newMAChannels
289+
-- Offline Pool + Anchor queues
283290
opwq <- newTBQueueIO 1000
284291
oprq <- newTBQueueIO 1000
285292
oawq <- newTBQueueIO 1000
@@ -309,7 +316,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
309316

310317
pure $
311318
SyncEnv
312-
{ envBackend = backend
319+
{ envBackends = backends
313320
, envPool = pool
314321
, envBootstrap = bootstrapVar
315322
, envCache = cache
@@ -323,6 +330,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
323330
, envLedgerEnv = ledgerEnvType
324331
, envNetworkMagic = nwMagic
325332
, envStakeChans = cChans
333+
, envMAChans = maChans
326334
, envOffChainPoolResultQueue = oprq
327335
, envOffChainPoolWorkQueue = opwq
328336
, envOffChainVoteResultQueue = oarq
@@ -335,10 +343,47 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
335343
where
336344
hasLedger' = hasLedger . sioLedger . dncInsertOptions
337345
isTxOutConsumedBootstrap' = isTxOutConsumedBootstrap . sioTxOut . dncInsertOptions
346+
backend = mainBackend backends
347+
348+
withDBSyncConnections ::
349+
(MonadUnliftIO m, MonadLoggerIO m) =>
350+
ConnectionString ->
351+
(DbConnections -> m a) ->
352+
m a
353+
withDBSyncConnections connStr action =
354+
withPostgresqlConn connStr $ \mainConn ->
355+
withPostgresqlConn connStr $ \scriptConn ->
356+
withPostgresqlConn connStr $ \datumConn -> do
357+
scr <- liftIO $ newMVar scriptConn
358+
dt <- liftIO $ newMVar datumConn
359+
action $ DbConnections mainConn scr dt
360+
361+
withScriptConnection ::
362+
SyncEnv ->
363+
ReaderT SqlBackend (LoggingT IO) a ->
364+
IO a
365+
withScriptConnection = withGivenConnection scriptBackend
366+
367+
withDatumConnection ::
368+
SyncEnv ->
369+
ReaderT SqlBackend (LoggingT IO) a ->
370+
IO a
371+
withDatumConnection = withGivenConnection datumBackend
372+
373+
withGivenConnection ::
374+
(DbConnections -> MVar SqlBackend) ->
375+
SyncEnv ->
376+
ReaderT SqlBackend (LoggingT IO) a ->
377+
IO a
378+
withGivenConnection toConn syncEnv action = do
379+
withMVar connVar $ \conn ->
380+
DB.runDbLogging conn (getTrace syncEnv) action
381+
where
382+
connVar = toConn $ envBackends syncEnv
338383

339384
mkSyncEnvFromConfig ::
340385
Trace IO Text ->
341-
SqlBackend ->
386+
DbConnections ->
342387
ConnectionString ->
343388
SyncOptions ->
344389
GenesisConfig ->
@@ -347,7 +392,7 @@ mkSyncEnvFromConfig ::
347392
-- | run migration function
348393
RunMigration ->
349394
IO (Either SyncNodeError SyncEnv)
350-
mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeConfigFromFile syncNodeParams runMigrationFnc =
395+
mkSyncEnvFromConfig trce backends connectionString syncOptions genCfg syncNodeConfigFromFile syncNodeParams runMigrationFnc =
351396
case genCfg of
352397
GenesisCardano _ bCfg sCfg _ _
353398
| unProtocolMagicId (Byron.configProtocolMagicId bCfg) /= Shelley.sgNetworkMagic (scConfig sCfg) ->
@@ -374,7 +419,7 @@ mkSyncEnvFromConfig trce backend connectionString syncOptions genCfg syncNodeCon
374419
Right
375420
<$> mkSyncEnv
376421
trce
377-
backend
422+
backends
378423
connectionString
379424
syncOptions
380425
(fst $ mkProtocolInfoCardano genCfg [])

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,15 @@ module Cardano.DbSync.Api.Types (
99
RunMigration,
1010
ConsistentLevel (..),
1111
CurrentEpochNo (..),
12+
DbConnections (..),
1213
Prefetch (..),
1314
PrefetchTxId (..),
15+
envBackend,
1416
newPrefetch,
1517
) where
1618

1719
import qualified Cardano.Db as DB
18-
import Cardano.DbSync.Cache.Types (CacheStatus, StakeChannels)
20+
import Cardano.DbSync.Cache.Types (CacheStatus, MAChannels, StakeChannels)
1921
import Cardano.DbSync.Config.Types (SyncNodeConfig)
2022
import qualified Cardano.DbSync.Era.Shelley.Generic.Tx.Types as Generic
2123
import Cardano.DbSync.Ledger.Types (HasLedgerEnv)
@@ -30,6 +32,7 @@ import Cardano.DbSync.Types (
3032
import Cardano.Slotting.Slot (EpochNo (..))
3133
import Control.Concurrent.Class.MonadSTM.Strict (StrictTVar, newTBQueueIO, newTVarIO)
3234
import Control.Concurrent.Class.MonadSTM.Strict.TBQueue (StrictTBQueue)
35+
import Control.Concurrent.MVar
3336
import Data.Map (Map)
3437
import Data.Pool (Pool)
3538
import qualified Data.Strict.Maybe as Strict
@@ -41,7 +44,7 @@ import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
4144
import Ouroboros.Network.Magic (NetworkMagic (..))
4245

4346
data SyncEnv = SyncEnv
44-
{ envBackend :: !SqlBackend
47+
{ envBackends :: !DbConnections
4548
, envPool :: !(Pool SqlBackend)
4649
, envPrefetch :: !Prefetch
4750
, envCache :: !CacheStatus
@@ -55,6 +58,7 @@ data SyncEnv = SyncEnv
5558
, envLedgerEnv :: !LedgerEnv
5659
, envNetworkMagic :: !NetworkMagic
5760
, envStakeChans :: !StakeChannels
61+
, envMAChans :: !MAChannels
5862
, envOffChainPoolResultQueue :: !(StrictTBQueue IO OffChainPoolResult)
5963
, envOffChainPoolWorkQueue :: !(StrictTBQueue IO OffChainPoolWorkQueue)
6064
, envOffChainVoteResultQueue :: !(StrictTBQueue IO OffChainVoteResult)
@@ -65,6 +69,9 @@ data SyncEnv = SyncEnv
6569
, envSystemStart :: !SystemStart
6670
}
6771

72+
envBackend :: SyncEnv -> SqlBackend
73+
envBackend = mainBackend . envBackends
74+
6875
data SyncOptions = SyncOptions
6976
{ soptEpochAndCacheEnabled :: !Bool
7077
, soptAbortOnInvalid :: !Bool
@@ -108,6 +115,12 @@ newtype CurrentEpochNo = CurrentEpochNo
108115
{ cenEpochNo :: Strict.Maybe EpochNo
109116
}
110117

118+
data DbConnections = DbConnections
119+
{ mainBackend :: !SqlBackend
120+
, scriptBackend :: !(MVar SqlBackend)
121+
, datumBackend :: !(MVar SqlBackend)
122+
}
123+
111124
data PrefetchTxId = PrefetchTxIdBlock CardanoBlock | PrefetchTxIdBlocks [CardanoBlock]
112125

113126
data Prefetch = Prefetch

0 commit comments

Comments
 (0)