Kiu/VictoriaMetrics
Kiu/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-15 00:13:30 +01:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

app/vmagent: fix data race when accessing writeRequest.lastFlushTime

Browse Source
This commit is contained in:
Aliaksandr Valialkin 2020-09-03 12:10:47 +03:00
parent 80a9dc79fe
commit de216bab41
1 changed files with 7 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
app/vmagent/remotewrite/pendingseries.go
View File

@ -3,6 +3,7 @@ package remotewrite
import ( import (
"flag" "flag"
"sync" "sync"
"sync/atomic"
"time" "time"
"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil" "github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
@ -69,7 +70,7 @@ func (ps *pendingSeries) periodicFlusher() {
case <-ps.stopCh: case <-ps.stopCh:
mustStop = true mustStop = true
case <-ticker.C: case <-ticker.C:
if fasttime.UnixTimestamp()-ps.wr.lastFlushTime < uint64(flushSeconds) { if fasttime.UnixTimestamp()-atomic.LoadUint64(&ps.wr.lastFlushTime) < uint64(flushSeconds) {
continue continue
} }
} }
@ -80,10 +81,12 @@ func (ps *pendingSeries) periodicFlusher() {
} }
type writeRequest struct { type writeRequest struct {
wr prompbmarshal.WriteRequest // Move lastFlushTime to the top of the struct in order to guarantee atomic access on 32-bit architectures.
pushBlock func(block []byte)
lastFlushTime uint64 lastFlushTime uint64
wr prompbmarshal.WriteRequest
pushBlock func(block []byte)
tss []prompbmarshal.TimeSeries tss []prompbmarshal.TimeSeries
labels []prompbmarshal.Label labels []prompbmarshal.Label
@ -114,7 +117,7 @@ func (wr *writeRequest) reset() {
func (wr *writeRequest) flush() { func (wr *writeRequest) flush() {
wr.wr.Timeseries = wr.tss wr.wr.Timeseries = wr.tss
wr.lastFlushTime = fasttime.UnixTimestamp() atomic.StoreUint64(&wr.lastFlushTime, fasttime.UnixTimestamp())
pushWriteRequest(&wr.wr, wr.pushBlock) pushWriteRequest(&wr.wr, wr.pushBlock)
wr.reset() wr.reset()
} }