Parallelise preparing the txs

Also move some util functions

Author: kderme

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

@@ -215,6 +215,7 @@ library
                       , prometheus
                       , psqueues
                       , random-shuffle
+                      , resource-pool
                       , scientific
                       , serialise
                       , small-steps

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

@@ -67,7 +67,7 @@ import Control.Concurrent.Class.MonadSTM.Strict (
 import qualified Control.Concurrent.Class.MonadSTM.Strict.TBQueue as TBQ
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (getCurrentTime)
-import Database.Persist.Postgresql (ConnectionString)
+import Database.Persist.Postgresql (ConnectionString, createPostgresqlPool)
 import Database.Persist.Sql (SqlBackend)
 import Ouroboros.Consensus.Block.Abstract (BlockProtocol)
 import Ouroboros.Consensus.BlockchainTime.WallClock.Types (SystemStart (..))
@@ -279,6 +279,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
   oarq <- newTBQueueIO 1000
   epochVar <- newTVarIO initCurrentEpochNo
   epochSyncTime <- newTVarIO =<< getCurrentTime
+  pool <- DB.runIohkLogging trce $ createPostgresqlPool connectionString 5 -- TODO make configurable
   ledgerEnvType <-
     case (enpMaybeLedgerStateDir syncNP, hasLedger' syncNodeConfigFromFile) of
       (Just dir, True) ->
@@ -302,6 +303,7 @@ mkSyncEnv trce backend connectionString syncOptions protoInfo nw nwMagic systemS
   pure $
     SyncEnv
       { envBackend = backend
+      , envPool = pool
       , envBootstrap = bootstrapVar
       , envCache = cache
       , envPrefetch = prefetch

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

@@ -15,8 +15,7 @@ import Cardano.DbSync.Era.Shelley.Generic.Tx.Babbage (fromTxOut)
 import Cardano.DbSync.Era.Shelley.Generic.Tx.Types (DBPlutusScript)
 import qualified Cardano.DbSync.Era.Shelley.Generic.Util as Generic
 import Cardano.DbSync.Era.Universal.Insert.Grouped
-import Cardano.DbSync.Era.Universal.Insert.Tx (insertTxOut)
-import Cardano.DbSync.Era.Util (liftLookupFail)
+import qualified Cardano.DbSync.Era.Universal.Insert.Tx as Prepare
 import Cardano.DbSync.Error
 import Cardano.DbSync.Ledger.State
 import Cardano.DbSync.Types
@@ -171,7 +170,7 @@ prepareTxOut syncEnv (TxIn txIntxId (TxIx index), txOut) = do
   let txHashByteString = Generic.safeHashToByteString $ unTxId txIntxId
   let genTxOut = fromTxOut index txOut
   txId <- liftLookupFail "prepareTxOut" $ queryTxIdWithCache cache txIntxId
-  insertTxOut syncEnv iopts (txId, txHashByteString) genTxOut
+  Prepare.prepareTxOut syncEnv iopts (txId, txHashByteString) genTxOut
   where
     cache = envCache syncEnv
     iopts = soptInsertOptions $ envOptions syncEnv

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

@@ -31,6 +31,7 @@ import Cardano.Slotting.Slot (EpochNo (..))
 import Control.Concurrent.Class.MonadSTM.Strict (StrictTVar, newTBQueueIO, newTVarIO)
 import Control.Concurrent.Class.MonadSTM.Strict.TBQueue (StrictTBQueue)
 import Data.Map (Map)
+import Data.Pool (Pool)
 import qualified Data.Strict.Maybe as Strict
 import Data.Time.Clock (UTCTime)
 import Data.Word (Word64)
@@ -41,6 +42,7 @@ import Ouroboros.Network.Magic (NetworkMagic (..))
 
 data SyncEnv = SyncEnv
   { envBackend :: !SqlBackend
+  , envPool :: !(Pool SqlBackend)
   , envPrefetch :: !Prefetch
   , envCache :: !CacheStatus
   , envConnectionString :: !ConnectionString