Skip to content

Commit c024537

Browse files
kdermekderme
committed
Regroup modules
1 parent e7e0f4b commit c024537

File tree

9 files changed

+73
-57
lines changed

9 files changed

+73
-57
lines changed

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

+4-3
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,6 @@ library
5454
Cardano.DbSync.Config.Node
5555
Cardano.DbSync.Config.Shelley
5656
Cardano.DbSync.Config.Types
57-
Cardano.DbSync.Database
5857
Cardano.DbSync.DbAction
5958
Cardano.DbSync.Error
6059

@@ -110,14 +109,14 @@ library
110109

111110
Cardano.DbSync.Metrics
112111

112+
Cardano.DbSync.Block
113113
Cardano.DbSync.Cache
114114
Cardano.DbSync.Cache.Epoch
115115
Cardano.DbSync.Cache.FIFO
116116
Cardano.DbSync.Cache.LRU
117117
Cardano.DbSync.Cache.Stake
118118
Cardano.DbSync.Cache.Types
119119
Cardano.DbSync.Cache.Util
120-
Cardano.DbSync.Default
121120
Cardano.DbSync.Epoch
122121

123122
Cardano.DbSync.Rollback
@@ -133,7 +132,9 @@ library
133132
Cardano.DbSync.LocalStateQuery
134133
Cardano.DbSync.StateQuery
135134
Cardano.DbSync.Sync
136-
Cardano.DbSync.Threads
135+
Cardano.DbSync.Threads.Database
136+
Cardano.DbSync.Threads.EpochStake
137+
Cardano.DbSync.Threads.Stake
137138
Cardano.DbSync.Tracing.ToObjectOrphans
138139
Cardano.DbSync.Types
139140

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

+3-2
Original file line numberDiff line numberDiff line change
@@ -33,15 +33,16 @@ import Cardano.DbSync.Api.Types (InsertOptions (..), RunMigration, SyncEnv (..),
3333
import Cardano.DbSync.Config (configureLogging)
3434
import Cardano.DbSync.Config.Cardano
3535
import Cardano.DbSync.Config.Types
36-
import Cardano.DbSync.Database
3736
import Cardano.DbSync.DbAction
3837
import Cardano.DbSync.Era
3938
import Cardano.DbSync.Error
4039
import Cardano.DbSync.Ledger.State
4140
import Cardano.DbSync.OffChain (runFetchOffChainPoolThread, runFetchOffChainVoteThread)
4241
import Cardano.DbSync.Rollback (unsafeRollback)
4342
import Cardano.DbSync.Sync (runSyncNodeClient)
44-
import Cardano.DbSync.Threads
43+
import Cardano.DbSync.Threads.Database
44+
import Cardano.DbSync.Threads.EpochStake
45+
import Cardano.DbSync.Threads.Stake
4546
import Cardano.DbSync.Tracing.ToObjectOrphans ()
4647
import Cardano.DbSync.Types
4748
import Cardano.DbSync.Util.Constraint (queryIsJsonbInSchema)

‎cardano-db-sync/src/Cardano/DbSync/Default.hs renamed to ‎cardano-db-sync/src/Cardano/DbSync/Block.hs

+2-2
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
{-# LANGUAGE NoImplicitPrelude #-}
88
{-# OPTIONS_GHC -Wno-unused-matches #-}
99

10-
module Cardano.DbSync.Default (
10+
module Cardano.DbSync.Block (
1111
insertListBlocks,
1212
) where
1313

@@ -82,8 +82,8 @@ applyAndInsertBlockMaybe syncEnv tracer cblk = do
8282
, ". Time to restore consistency."
8383
]
8484
rollbackFromBlockNo syncEnv (blockNo cblk)
85-
insertBlock syncEnv cblk applyRes True tookSnapshot
8685
liftIO $ setConsistentLevel syncEnv Consistent
86+
insertBlock syncEnv cblk applyRes True tookSnapshot
8787
Right blockId | Just (adaPots, slotNo, epochNo) <- getAdaPots applyRes -> do
8888
replaced <- lift $ DB.replaceAdaPots blockId $ mkAdaPots blockId slotNo epochNo adaPots
8989
if replaced

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

+15
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,21 @@ import Data.Either.Combinators
2929
import qualified Data.Map.Strict as Map
3030
import Database.Persist.Postgresql (SqlBackend)
3131

32+
-- | TO be called only by the stake thread
33+
resolveInsertRewardAccount ::
34+
forall m.
35+
(MonadBaseControl IO m, MonadIO m) =>
36+
SyncEnv ->
37+
CacheAction ->
38+
RewAccount ->
39+
ReaderT SqlBackend m DB.StakeAddressId
40+
resolveInsertRewardAccount syncEnv cacheUA ra = do
41+
eiStakeId <- queryStakeAddrWithCacheRetBs syncEnv cacheUA False ra -- read only
42+
case eiStakeId of
43+
Right stakeId -> pure stakeId
44+
Left (_, bs) -> insertStakeAddress ra (Just bs)
45+
46+
-- | TO be called only by the stake thread
3247
-- If the address already exists in the table, it will not be inserted again (due to
3348
-- the uniqueness constraint) but the function will return the 'StakeAddressId'.
3449
insertStakeAddress ::

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

+1-18
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,11 @@ module Cardano.DbSync.DbAction (
1010
writeDbActionQueue,
1111
waitRollback,
1212
waitRestartState,
13-
waitDoneInit,
14-
runAndSetDone,
1513
) where
1614

1715
import Cardano.DbSync.Types
1816
import Cardano.Prelude
19-
import Control.Concurrent.Class.MonadSTM.Strict (StrictTMVar, StrictTVar, newEmptyTMVarIO, newTVarIO, readTVar, readTVarIO, takeTMVar, writeTVar)
17+
import Control.Concurrent.Class.MonadSTM.Strict (StrictTMVar, StrictTVar, newEmptyTMVarIO, newTVarIO, takeTMVar)
2018
import qualified Control.Concurrent.STM as STM
2119
import Control.Concurrent.STM.TBQueue (TBQueue)
2220
import qualified Control.Concurrent.STM.TBQueue as TBQ
@@ -53,21 +51,6 @@ waitRestartState tc = do
5351
writeDbActionQueue tc $ DbRestartState resultVar
5452
atomically $ takeTMVar resultVar
5553

56-
waitDoneInit :: ThreadChannels -> IO ()
57-
waitDoneInit tc = atomically $ do
58-
isDone <- readTVar (tcDoneInit tc)
59-
if isDone then pure () else retry
60-
61-
runAndSetDone :: ThreadChannels -> IO Bool -> IO Bool
62-
runAndSetDone tc action = do
63-
isDone <- readTVarIO (tcDoneInit tc)
64-
if isDone
65-
then pure True
66-
else do
67-
fl <- action
68-
atomically $ writeTVar (tcDoneInit tc) fl
69-
pure fl
70-
7154
lengthDbActionQueue :: ThreadChannels -> STM Natural
7255
lengthDbActionQueue = STM.lengthTBQueue . tcQueue
7356

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

-1
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,6 @@ import Cardano.Client.Subscription (subscribe)
3030
import Cardano.DbSync.Api
3131
import Cardano.DbSync.Api.Types (ConsistentLevel (..), LedgerEnv (..), SyncEnv (..), envLedgerEnv, envNetworkMagic, envOptions)
3232
import Cardano.DbSync.Config
33-
import Cardano.DbSync.Database
3433
import Cardano.DbSync.DbAction
3534
import Cardano.DbSync.LocalStateQuery
3635
import Cardano.DbSync.Metrics

‎cardano-db-sync/src/Cardano/DbSync/Database.hs renamed to ‎cardano-db-sync/src/Cardano/DbSync/Threads/Database.hs

+2-2
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
{-# LANGUAGE TypeFamilies #-}
44
{-# LANGUAGE NoImplicitPrelude #-}
55

6-
module Cardano.DbSync.Database (
6+
module Cardano.DbSync.Threads.Database (
77
DbAction (..),
88
ThreadChannels,
99
lengthDbActionQueue,
@@ -15,7 +15,7 @@ import Cardano.BM.Trace (logDebug, logError, logInfo)
1515
import Cardano.DbSync.Api
1616
import Cardano.DbSync.Api.Types (ConsistentLevel (..), LedgerEnv (..), SyncEnv (..))
1717
import Cardano.DbSync.DbAction
18-
import Cardano.DbSync.Default
18+
import Cardano.DbSync.Block
1919
import Cardano.DbSync.Error
2020
import Cardano.DbSync.Ledger.State
2121
import Cardano.DbSync.Ledger.Types (CardanoLedgerState (..), SnapshotPoint (..))

‎cardano-db-sync/src/Cardano/DbSync/Threads.hs renamed to ‎cardano-db-sync/src/Cardano/DbSync/Threads/EpochStake.hs

+1-29
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
{-# LANGUAGE FlexibleContexts #-}
22
{-# LANGUAGE OverloadedStrings #-}
33

4-
module Cardano.DbSync.Threads where
4+
module Cardano.DbSync.Threads.EpochStake where
55

66
import Cardano.BM.Trace (logInfo)
77
import qualified Cardano.Db as DB
@@ -21,8 +21,6 @@ import Control.Monad.Trans.Class
2121
import Control.Monad.Trans.Except.Extra (runExceptT)
2222
import Database.Persist.Postgresql (IsolationLevel (..), runSqlConnWithIsolation, withPostgresqlConn)
2323

24-
import Cardano.DbSync.Cache.Types
25-
2624
runEpochStakeThread ::
2725
SyncEnv ->
2826
IO ()
@@ -56,29 +54,3 @@ runESLoop syncEnv lenv =
5654

5755
estakeChan = leEpochStakeChans lenv
5856
trce = getTrace syncEnv
59-
60-
runStakeThread :: SyncEnv -> IO ()
61-
runStakeThread syncEnv = do
62-
logInfo trce "Running Event thread"
63-
logException trce "runEpochStakeThread: " (runStakeLoop syncEnv)
64-
logInfo trce "Shutting Event thread"
65-
where
66-
trce = getTrace syncEnv
67-
68-
runStakeLoop :: SyncEnv -> IO ()
69-
runStakeLoop syncEnv =
70-
DB.runIohkLogging trce $
71-
withPostgresqlConn (envConnectionString syncEnv) actionDB
72-
where
73-
actionDB backend = runSqlConnWithIsolation (forever loopAction) backend Serializable
74-
75-
loopAction = do
76-
action <- liftIO $ atomically $ TBQ.readTBQueue (scPriorityQueue stakeChan)
77-
case action of
78-
QueryInsertStake _sk _ca _resVar -> undefined
79-
CacheStake _ _ _ -> pure ()
80-
BulkPrefetch _ -> pure ()
81-
CommitStake -> DB.transactionCommit
82-
83-
stakeChan = envStakeChans syncEnv
84-
trce = getTrace syncEnv

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

+45
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
{-# LANGUAGE FlexibleContexts #-}
2+
{-# LANGUAGE OverloadedStrings #-}
3+
4+
module Cardano.DbSync.Threads.Stake where
5+
6+
import Cardano.BM.Trace (logInfo)
7+
import qualified Cardano.Db as DB
8+
import Cardano.DbSync.Api
9+
import Cardano.DbSync.Api.Types
10+
import Cardano.DbSync.Util
11+
import Control.Concurrent.Class.MonadSTM.Strict
12+
import qualified Control.Concurrent.STM.TBQueue as TBQ
13+
import Control.Monad
14+
import Control.Monad.IO.Class (liftIO)
15+
import Database.Persist.Postgresql (IsolationLevel (..), runSqlConnWithIsolation, withPostgresqlConn)
16+
import Cardano.DbSync.Cache.Stake
17+
import Cardano.DbSync.Cache.Types
18+
19+
runStakeThread :: SyncEnv -> IO ()
20+
runStakeThread syncEnv = do
21+
logInfo trce "Running Event thread"
22+
logException trce "runStakeThread: " (runStakeLoop syncEnv)
23+
logInfo trce "Shutting Event thread"
24+
where
25+
trce = getTrace syncEnv
26+
27+
runStakeLoop :: SyncEnv -> IO ()
28+
runStakeLoop syncEnv =
29+
DB.runIohkLogging trce $
30+
withPostgresqlConn (envConnectionString syncEnv) actionDB
31+
where
32+
actionDB backend = runSqlConnWithIsolation (forever loopAction) backend Serializable
33+
34+
loopAction = do
35+
action <- liftIO $ atomically $ TBQ.readTBQueue (scPriorityQueue stakeChan)
36+
case action of
37+
QueryInsertStake rewardAcc ca resVar -> do
38+
stakeId <- resolveInsertRewardAccount syncEnv ca rewardAcc
39+
liftIO $ atomically $ writeTMVar resVar stakeId
40+
CacheStake _ _ _ -> pure ()
41+
BulkPrefetch _ -> pure ()
42+
CommitStake -> DB.transactionCommit
43+
44+
stakeChan = envStakeChans syncEnv
45+
trce = getTrace syncEnv

0 commit comments

Comments
 (0)