Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-20 07:19:17 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

lib/logstorage: postpone initialization of per-shard stateSizeBudget until the first call to pipeProcessor.writeBlock()

Browse Source 
This simplifies pipeProcessor initialization logic a bit.
This also doesn't mangle the original maxStateSize value, which is used in error messages when the state size exceeds maxStateSize.
This commit is contained in:
Aliaksandr Valialkin 2024-09-29 10:16:14 +02:00
parent a18d4f79d4
commit b5d94f06f5
No known key found for this signature in database
GPG Key ID: 52C003EE2BCDB9EB
6 changed files with 5 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
lib/logstorage/pipe_sort.go
View File

@ -116,11 +116,9 @@ func newPipeSortProcessor(ps *pipeSort, workersCount int, stopCh <-chan struct{}
for i := range shards { for i := range shards {
shards[i] = pipeSortProcessorShard{ shards[i] = pipeSortProcessorShard{
pipeSortProcessorShardNopad: pipeSortProcessorShardNopad{ pipeSortProcessorShardNopad: pipeSortProcessorShardNopad{
ps: ps, ps: ps,
stateSizeBudget: stateSizeBudgetChunk,
}, },
} }
maxStateSize -= stateSizeBudgetChunk
} }
psp := &pipeSortProcessor{ psp := &pipeSortProcessor{

4
lib/logstorage/pipe_sort_topk.go
View File

@ -20,11 +20,9 @@ func newPipeTopkProcessor(ps *pipeSort, workersCount int, stopCh <-chan struct{}
for i := range shards { for i := range shards {
shards[i] = pipeTopkProcessorShard{ shards[i] = pipeTopkProcessorShard{
pipeTopkProcessorShardNopad: pipeTopkProcessorShardNopad{ pipeTopkProcessorShardNopad: pipeTopkProcessorShardNopad{
ps: ps, ps: ps,
stateSizeBudget: stateSizeBudgetChunk,
}, },
} }
maxStateSize -= stateSizeBudgetChunk
} }
ptp := &pipeTopkProcessor{ ptp := &pipeTopkProcessor{

3
lib/logstorage/pipe_stats.go
View File

@ -162,11 +162,8 @@ func (ps *pipeStats) newPipeProcessor(workersCount int, stopCh <-chan struct{},
shards[i] = pipeStatsProcessorShard{ shards[i] = pipeStatsProcessorShard{
pipeStatsProcessorShardNopad: pipeStatsProcessorShardNopad{ pipeStatsProcessorShardNopad: pipeStatsProcessorShardNopad{
ps: ps, ps: ps,
stateSizeBudget: stateSizeBudgetChunk,
}, },
} }
maxStateSize -= stateSizeBudgetChunk
} }
psp := &pipeStatsProcessor{ psp := &pipeStatsProcessor{

4
lib/logstorage/pipe_stream_context.go
View File

@ -74,11 +74,9 @@ func (pc *pipeStreamContext) newPipeProcessor(workersCount int, stopCh <-chan st
for i := range shards { for i := range shards {
shards[i] = pipeStreamContextProcessorShard{ shards[i] = pipeStreamContextProcessorShard{
pipeStreamContextProcessorShardNopad: pipeStreamContextProcessorShardNopad{ pipeStreamContextProcessorShardNopad: pipeStreamContextProcessorShardNopad{
pc: pc, pc: pc,
stateSizeBudget: stateSizeBudgetChunk,
}, },
} }
maxStateSize -= stateSizeBudgetChunk
} }
pcp := &pipeStreamContextProcessor{ pcp := &pipeStreamContextProcessor{

4
lib/logstorage/pipe_top.go
View File

@ -79,11 +79,9 @@ func (pt *pipeTop) newPipeProcessor(workersCount int, stopCh <-chan struct{}, ca
for i := range shards { for i := range shards {
shards[i] = pipeTopProcessorShard{ shards[i] = pipeTopProcessorShard{
pipeTopProcessorShardNopad: pipeTopProcessorShardNopad{ pipeTopProcessorShardNopad: pipeTopProcessorShardNopad{
pt: pt, pt: pt,
stateSizeBudget: stateSizeBudgetChunk,
}, },
} }
maxStateSize -= stateSizeBudgetChunk
} }
ptp := &pipeTopProcessor{ ptp := &pipeTopProcessor{

4
lib/logstorage/pipe_uniq.go
View File

@ -74,11 +74,9 @@ func (pu *pipeUniq) newPipeProcessor(workersCount int, stopCh <-chan struct{}, c
for i := range shards { for i := range shards {
shards[i] = pipeUniqProcessorShard{ shards[i] = pipeUniqProcessorShard{
pipeUniqProcessorShardNopad: pipeUniqProcessorShardNopad{ pipeUniqProcessorShardNopad: pipeUniqProcessorShardNopad{
pu: pu, pu: pu,
stateSizeBudget: stateSizeBudgetChunk,
}, },
} }
maxStateSize -= stateSizeBudgetChunk
} }
pup := &pipeUniqProcessor{ pup := &pipeUniqProcessor{