Skip to content

Commit 06bb521

Browse files
iuwqyiriuwqyir
committed
clickhouse and poller settings
1 parent b0a4ef9 commit 06bb521

9 files changed

+249
-162
lines changed

‎cmd/root.go

+16-2
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,7 @@ func init() {
5555
rootCmd.PersistentFlags().Int("poller-from-block", 0, "From which block to start polling")
5656
rootCmd.PersistentFlags().Bool("poller-force-from-block", false, "Force the poller to start from the block specified in `poller-from-block`")
5757
rootCmd.PersistentFlags().Int("poller-until-block", 0, "Until which block to poll")
58+
rootCmd.PersistentFlags().Int("poller-parallel-pollers", 5, "Maximum number of parallel pollers")
5859
rootCmd.PersistentFlags().Bool("committer-enabled", true, "Toggle committer")
5960
rootCmd.PersistentFlags().Int("committer-blocks-per-commit", 10, "How many blocks to commit each interval")
6061
rootCmd.PersistentFlags().Int("committer-interval", 1000, "How often to commit blocks in milliseconds")
@@ -76,6 +77,8 @@ func init() {
7677
rootCmd.PersistentFlags().String("storage-orchestrator-clickhouse-host", "", "Clickhouse host for orchestrator storage")
7778
rootCmd.PersistentFlags().String("storage-orchestrator-clickhouse-username", "", "Clickhouse username for orchestrator storage")
7879
rootCmd.PersistentFlags().String("storage-orchestrator-clickhouse-password", "", "Clickhouse password for orchestrator storage")
80+
rootCmd.PersistentFlags().Bool("storage-orchestrator-clickhouse-asyncInsert", false, "Clickhouse async insert for orchestrator storage")
81+
rootCmd.PersistentFlags().Int("storage-orchestrator-clickhouse-maxRowsPerInsert", 100000, "Clickhouse max rows per insert for orchestrator storage")
7982
rootCmd.PersistentFlags().Int("storage-orchestrator-memory-maxItems", 0, "Max items for orchestrator memory storage")
8083
rootCmd.PersistentFlags().Int("storage-orchestrator-redis-poolSize", 0, "Redis pool size for orchestrator storage")
8184
rootCmd.PersistentFlags().String("storage-orchestrator-redis-addr", "", "Redis address for orchestrator storage")
@@ -85,8 +88,12 @@ func init() {
8588
rootCmd.PersistentFlags().String("storage-main-clickhouse-host", "", "Clickhouse host for main storage")
8689
rootCmd.PersistentFlags().String("storage-main-clickhouse-username", "", "Clickhouse username for main storage")
8790
rootCmd.PersistentFlags().String("storage-main-clickhouse-password", "", "Clickhouse password for main storage")
91+
rootCmd.PersistentFlags().Bool("storage-main-clickhouse-asyncInsert", false, "Clickhouse async insert for main storage")
92+
rootCmd.PersistentFlags().Int("storage-main-clickhouse-maxRowsPerInsert", 100000, "Clickhouse max rows per insert for main storage")
8893
rootCmd.PersistentFlags().String("storage-staging-clickhouse-username", "", "Clickhouse username for staging storage")
8994
rootCmd.PersistentFlags().String("storage-staging-clickhouse-password", "", "Clickhouse password for staging storage")
95+
rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-asyncInsert", false, "Clickhouse async insert for staging storage")
96+
rootCmd.PersistentFlags().Int("storage-staging-clickhouse-maxRowsPerInsert", 100000, "Clickhouse max rows per insert for staging storage")
9097
rootCmd.PersistentFlags().String("api-host", "localhost:3000", "API host")
9198
viper.BindPFlag("rpc.url", rootCmd.PersistentFlags().Lookup("rpc-url"))
9299
viper.BindPFlag("rpc.blocks.blocksPerRequest", rootCmd.PersistentFlags().Lookup("rpc-blocks-blocksPerRequest"))
@@ -107,6 +114,7 @@ func init() {
107114
viper.BindPFlag("poller.fromBlock", rootCmd.PersistentFlags().Lookup("poller-from-block"))
108115
viper.BindPFlag("poller.forceFromBlock", rootCmd.PersistentFlags().Lookup("poller-force-from-block"))
109116
viper.BindPFlag("poller.untilBlock", rootCmd.PersistentFlags().Lookup("poller-until-block"))
117+
viper.BindPFlag("poller.parallelPollers", rootCmd.PersistentFlags().Lookup("poller-parallel-pollers"))
110118
viper.BindPFlag("committer.enabled", rootCmd.PersistentFlags().Lookup("committer-enabled"))
111119
viper.BindPFlag("committer.blocksPerCommit", rootCmd.PersistentFlags().Lookup("committer-blocks-per-commit"))
112120
viper.BindPFlag("committer.interval", rootCmd.PersistentFlags().Lookup("committer-interval"))
@@ -122,18 +130,24 @@ func init() {
122130
viper.BindPFlag("storage.staging.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-database"))
123131
viper.BindPFlag("storage.staging.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-host"))
124132
viper.BindPFlag("storage.staging.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-port"))
133+
viper.BindPFlag("storage.staging.clickhouse.username", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-username"))
134+
viper.BindPFlag("storage.staging.clickhouse.password", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-password"))
135+
viper.BindPFlag("storage.staging.clickhouse.asyncInsert", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-asyncInsert"))
136+
viper.BindPFlag("storage.staging.clickhouse.maxRowsPerInsert", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-maxRowsPerInsert"))
125137
viper.BindPFlag("storage.main.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-database"))
126138
viper.BindPFlag("storage.main.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-host"))
127139
viper.BindPFlag("storage.main.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-port"))
128140
viper.BindPFlag("storage.main.clickhouse.username", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-username"))
129141
viper.BindPFlag("storage.main.clickhouse.password", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-password"))
130-
viper.BindPFlag("storage.staging.clickhouse.username", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-username"))
131-
viper.BindPFlag("storage.staging.clickhouse.password", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-password"))
142+
viper.BindPFlag("storage.main.clickhouse.asyncInsert", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-asyncInsert"))
143+
viper.BindPFlag("storage.main.clickhouse.maxRowsPerInsert", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-maxRowsPerInsert"))
132144
viper.BindPFlag("storage.orchestrator.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-database"))
133145
viper.BindPFlag("storage.orchestrator.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-host"))
134146
viper.BindPFlag("storage.orchestrator.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-port"))
135147
viper.BindPFlag("storage.orchestrator.clickhouse.username", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-username"))
136148
viper.BindPFlag("storage.orchestrator.clickhouse.password", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-password"))
149+
viper.BindPFlag("storage.orchestrator.clickhouse.asyncInsert", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-asyncInsert"))
150+
viper.BindPFlag("storage.orchestrator.clickhouse.maxRowsPerInsert", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-maxRowsPerInsert"))
137151
viper.BindPFlag("storage.orchestrator.memory.maxItems", rootCmd.PersistentFlags().Lookup("storage-orchestrator-memory-maxItems"))
138152
viper.BindPFlag("storage.orchestrator.redis.poolSize", rootCmd.PersistentFlags().Lookup("storage-orchestrator-redis-poolSize"))
139153
viper.BindPFlag("storage.orchestrator.redis.addr", rootCmd.PersistentFlags().Lookup("storage-orchestrator-redis-addr"))

‎configs/config.go

+15-12
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,13 @@ type LogConfig struct {
1414
}
1515

1616
type PollerConfig struct {
17-
Enabled bool `mapstructure:"enabled"`
18-
Interval int `mapstructure:"interval"`
19-
BlocksPerPoll int `mapstructure:"blocksPerPoll"`
20-
FromBlock int `mapstructure:"fromBlock"`
21-
ForceFromBlock bool `mapstructure:"forceFromBlock"`
22-
UntilBlock int `mapstructure:"untilBlock"`
17+
Enabled bool `mapstructure:"enabled"`
18+
Interval int `mapstructure:"interval"`
19+
BlocksPerPoll int `mapstructure:"blocksPerPoll"`
20+
FromBlock int `mapstructure:"fromBlock"`
21+
ForceFromBlock bool `mapstructure:"forceFromBlock"`
22+
UntilBlock int `mapstructure:"untilBlock"`
23+
ParallelPollers int `mapstructure:"parallelPollers"`
2324
}
2425

2526
type CommitterConfig struct {
@@ -63,12 +64,14 @@ type StorageConnectionConfig struct {
6364
}
6465

6566
type ClickhouseConfig struct {
66-
Host string `mapstructure:"host"`
67-
Port int `mapstructure:"port"`
68-
Username string `mapstructure:"username"`
69-
Password string `mapstructure:"password"`
70-
Database string `mapstructure:"database"`
71-
DisableTLS bool `mapstructure:"disableTLS"`
67+
Host string `mapstructure:"host"`
68+
Port int `mapstructure:"port"`
69+
Username string `mapstructure:"username"`
70+
Password string `mapstructure:"password"`
71+
Database string `mapstructure:"database"`
72+
DisableTLS bool `mapstructure:"disableTLS"`
73+
AsyncInsert bool `mapstructure:"asyncInsert"`
74+
MaxRowsPerInsert int `mapstructure:"maxRowsPerInsert"`
7275
}
7376

7477
type MemoryConfig struct {

‎internal/orchestrator/poller.go

+5-5
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ type Poller struct {
2525
storage storage.IStorage
2626
lastPolledBlock *big.Int
2727
pollUntilBlock *big.Int
28+
parallelPollers int
2829
}
2930

3031
type BlockNumberWithError struct {
@@ -62,19 +63,18 @@ func NewPoller(rpc rpc.IRPCClient, storage storage.IStorage) *Poller {
6263
storage: storage,
6364
lastPolledBlock: lastPolledBlock,
6465
pollUntilBlock: untilBlock,
66+
parallelPollers: config.Cfg.Poller.ParallelPollers,
6567
}
6668
}
6769

6870
func (p *Poller) Start() {
6971
interval := time.Duration(p.triggerIntervalMs) * time.Millisecond
7072
ticker := time.NewTicker(interval)
7173

72-
// TODO: make this configurable?
73-
const numWorkers = 5
74-
tasks := make(chan struct{}, numWorkers)
74+
tasks := make(chan struct{}, p.parallelPollers)
7575
var blockRangeMutex sync.Mutex
7676

77-
for i := 0; i < numWorkers; i++ {
77+
for i := 0; i < p.parallelPollers; i++ {
7878
go func() {
7979
for range tasks {
8080
blockRangeMutex.Lock()
@@ -155,7 +155,7 @@ func (p *Poller) getEndBlockForRange(startBlock *big.Int, latestBlock *big.Int)
155155
endBlock = latestBlock
156156
}
157157
if p.reachedPollLimit(endBlock) {
158-
log.Debug().Msgf("End block %s is greater than poll until block %s, setting to poll until block", endBlock, p.pollUntilBlock)
158+
log.Debug().Msgf("End block %s is greater than or equal to poll until block %s, setting range end to poll until block", endBlock, p.pollUntilBlock)
159159
endBlock = p.pollUntilBlock
160160
}
161161
return endBlock

0 commit comments

Comments
 (0)