github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-31 15:06:26 +00:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Aliaksandr Valialkin 2024-05-25 22:20:27 +02:00
parent 1395e222d7
commit 8248afc9bd
No known key found for this signature in database
GPG key ID: 52C003EE2BCDB9EB
1 changed files with 6 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
lib/logstorage/pipe_unroll.go
View file

@ -74,9 +74,10 @@ func (pu *pipeUnroll) updateNeededFields(neededFields, unneededFields fieldsSet)
}
}
func (pu *pipeUnroll) newPipeProcessor(workersCount int, _ <-chan struct{}, _ func(), ppNext pipeProcessor) pipeProcessor {
func (pu *pipeUnroll) newPipeProcessor(workersCount int, stopCh <-chan struct{}, _ func(), ppNext pipeProcessor) pipeProcessor {
return &pipeUnrollProcessor{
pu: pu,
stopCh: stopCh,
ppNext: ppNext,
shards: make([]pipeUnrollProcessorShard, workersCount),
@ -85,6 +86,7 @@ func (pu *pipeUnroll) newPipeProcessor(workersCount int, _ <-chan struct{}, _ fu
type pipeUnrollProcessor struct {
pu *pipeUnroll
stopCh <-chan struct{}
ppNext pipeProcessor
shards []pipeUnrollProcessorShard
@ -139,6 +141,9 @@ func (pup *pipeUnrollProcessor) writeBlock(workerID uint, br *blockResult) {
fields := shard.fields
for rowIdx := range br.timestamps {
if bm.isSetBit(rowIdx) {
if needStop(pup.stopCh) {
return
}
shard.writeUnrolledFields(br, pu.fields, columnValues, rowIdx)
} else {
fields = fields[:0]