github
/
prometheus
mirror of https://github.com/prometheus/prometheus
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Avoid deadlock when processing duplicate series record (#9170) 

* Avoid deadlock when processing duplicate series record

`processWALSamples()` needs to be able to send on its output channel
before it can read the input channel, so reads to allow this in case the
output channel is full.

Signed-off-by: Bryan Boreham <bjboreham@gmail.com>

* processWALSamples: update comment

Previous text seems to relate to an earlier implementation.

Signed-off-by: Bryan Boreham <bjboreham@gmail.com>
pull/9184/head
Bryan Boreham 2021-08-10 09:02:42 +01:00 committed by GitHub
parent fac1b57334
commit 7407457243
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 11 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
tsdb/head_wal.go
View File

@ -228,9 +228,17 @@ Outer:
// It is possible that some old sample is being processed in processWALSamples that // It is possible that some old sample is being processed in processWALSamples that
// could cause race below. So we wait for the goroutine to empty input the buffer and finish // could cause race below. So we wait for the goroutine to empty input the buffer and finish
// processing all old samples after emptying the buffer. // processing all old samples after emptying the buffer.
select {
case <-outputs[idx]: // allow output side to drain to avoid deadlock
default:
}
inputs[idx] <- []record.RefSample{} inputs[idx] <- []record.RefSample{}
for len(inputs[idx]) != 0 { for len(inputs[idx]) != 0 {
time.Sleep(1 * time.Millisecond) time.Sleep(1 * time.Millisecond)
select {
case <-outputs[idx]: // allow output side to drain to avoid deadlock
default:
}
} }
// Checking if the new m-mapped chunks overlap with the already existing ones. // Checking if the new m-mapped chunks overlap with the already existing ones.
@ -351,9 +359,9 @@ Outer:
return nil return nil
} }
// processWALSamples adds a partition of samples it receives to the head and passes // processWALSamples adds the samples it receives to the head and passes
// them on to other workers. // the buffer received to an output channel for reuse.
// Samples before the mint timestamp are discarded. // Samples before the minValidTime timestamp are discarded.
func (h *Head) processWALSamples( func (h *Head) processWALSamples(
minValidTime int64, minValidTime int64,
input <-chan []record.RefSample, output chan<- []record.RefSample, input <-chan []record.RefSample, output chan<- []record.RefSample,