lib/{storage,mergeset}: use time.Ticker instead of time.Timer where appropriate

It has been appeared that time.Timer was used in places where time.Ticker must be used instead.
This could result in blocked goroutines as in the https://github.com/VictoriaMetrics/VictoriaMetrics/issues/316 .
This commit is contained in:
Aliaksandr Valialkin 2020-02-13 12:55:58 +02:00
parent 8162d58dbd
commit eceaf13e5e
6 changed files with 36 additions and 36 deletions

View file

@ -216,13 +216,13 @@ func (idxbc *indexBlockCache) MustClose() {
// cleaner periodically cleans least recently used items. // cleaner periodically cleans least recently used items.
func (idxbc *indexBlockCache) cleaner() { func (idxbc *indexBlockCache) cleaner() {
t := time.NewTimer(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for { for {
select { select {
case <-t.C: case <-ticker.C:
idxbc.cleanByTimeout() idxbc.cleanByTimeout()
case <-idxbc.cleanerStopCh: case <-idxbc.cleanerStopCh:
t.Stop()
return return
} }
} }
@ -373,13 +373,13 @@ func (ibc *inmemoryBlockCache) MustClose() {
// cleaner periodically cleans least recently used items. // cleaner periodically cleans least recently used items.
func (ibc *inmemoryBlockCache) cleaner() { func (ibc *inmemoryBlockCache) cleaner() {
t := time.NewTimer(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for { for {
select { select {
case <-t.C: case <-ticker.C:
ibc.cleanByTimeout() ibc.cleanByTimeout()
case <-ibc.cleanerStopCh: case <-ibc.cleanerStopCh:
t.Stop()
return return
} }
} }
@ -468,8 +468,9 @@ func (ibc *inmemoryBlockCache) Misses() uint64 {
func init() { func init() {
go func() { go func() {
t := time.NewTimer(time.Second) ticker := time.NewTicker(time.Second)
for tm := range t.C { defer ticker.Stop()
for tm := range ticker.C {
t := uint64(tm.Unix()) t := uint64(tm.Unix())
atomic.StoreUint64(&currentTimestamp, t) atomic.StoreUint64(&currentTimestamp, t)
} }

View file

@ -403,18 +403,17 @@ func (tb *Table) startRawItemsFlusher() {
} }
func (tb *Table) rawItemsFlusher() { func (tb *Table) rawItemsFlusher() {
t := time.NewTimer(rawItemsFlushInterval) ticker := time.NewTicker(rawItemsFlushInterval)
defer ticker.Stop()
for { for {
select { select {
case <-tb.stopCh: case <-tb.stopCh:
return return
case <-t.C: case <-ticker.C:
t.Reset(rawItemsFlushInterval)
}
tb.flushRawItems(false) tb.flushRawItems(false)
} }
} }
}
const convertToV1280FileName = "converted-to-v1.28.0" const convertToV1280FileName = "converted-to-v1.28.0"

View file

@ -223,13 +223,13 @@ func (ibc *indexBlockCache) MustClose(isBig bool) {
// cleaner periodically cleans least recently used items. // cleaner periodically cleans least recently used items.
func (ibc *indexBlockCache) cleaner() { func (ibc *indexBlockCache) cleaner() {
t := time.NewTimer(5 * time.Second) ticker := time.NewTicker(5 * time.Second)
defer ticker.Stop()
for { for {
select { select {
case <-t.C: case <-ticker.C:
ibc.cleanByTimeout() ibc.cleanByTimeout()
case <-ibc.cleanerStopCh: case <-ibc.cleanerStopCh:
t.Stop()
return return
} }
} }
@ -317,8 +317,9 @@ func (ibc *indexBlockCache) Len() uint64 {
func init() { func init() {
go func() { go func() {
t := time.NewTimer(time.Second) ticker := time.NewTicker(time.Second)
for tm := range t.C { defer ticker.Stop()
for tm := range ticker.C {
t := uint64(tm.Unix()) t := uint64(tm.Unix())
atomic.StoreUint64(&currentTimestamp, t) atomic.StoreUint64(&currentTimestamp, t)
} }

View file

@ -688,17 +688,17 @@ func (pt *partition) startRawRowsFlusher() {
} }
func (pt *partition) rawRowsFlusher() { func (pt *partition) rawRowsFlusher() {
t := time.NewTimer(rawRowsFlushInterval) ticker := time.NewTicker(rawRowsFlushInterval)
defer ticker.Stop()
for { for {
select { select {
case <-pt.stopCh: case <-pt.stopCh:
return return
case <-t.C: case <-ticker.C:
t.Reset(rawRowsFlushInterval)
}
pt.flushRawRows(false) pt.flushRawRows(false)
} }
} }
}
func (pt *partition) flushRawRows(isFinal bool) { func (pt *partition) flushRawRows(isFinal bool) {
pt.rawRows.flush(pt, isFinal) pt.rawRows.flush(pt, isFinal)
@ -736,23 +736,22 @@ func (pt *partition) startInmemoryPartsFlusher() {
} }
func (pt *partition) inmemoryPartsFlusher() { func (pt *partition) inmemoryPartsFlusher() {
t := time.NewTimer(inmemoryPartsFlushInterval) ticker := time.NewTicker(inmemoryPartsFlushInterval)
defer ticker.Stop()
var pwsBuf []*partWrapper var pwsBuf []*partWrapper
var err error var err error
for { for {
select { select {
case <-pt.stopCh: case <-pt.stopCh:
return return
case <-t.C: case <-ticker.C:
t.Reset(inmemoryPartsFlushInterval)
}
pwsBuf, err = pt.flushInmemoryParts(pwsBuf[:0], false) pwsBuf, err = pt.flushInmemoryParts(pwsBuf[:0], false)
if err != nil { if err != nil {
logger.Panicf("FATAL: cannot flush inmemory parts: %s", err) logger.Panicf("FATAL: cannot flush inmemory parts: %s", err)
} }
} }
} }
}
func (pt *partition) flushInmemoryParts(dstPws []*partWrapper, force bool) ([]*partWrapper, error) { func (pt *partition) flushInmemoryParts(dstPws []*partWrapper, force bool) ([]*partWrapper, error) {
currentTime := time.Now() currentTime := time.Now()

View file

@ -443,15 +443,15 @@ func (s *Storage) startCurrHourMetricIDsUpdater() {
var currHourMetricIDsUpdateInterval = time.Second * 10 var currHourMetricIDsUpdateInterval = time.Second * 10
func (s *Storage) currHourMetricIDsUpdater() { func (s *Storage) currHourMetricIDsUpdater() {
t := time.NewTimer(currHourMetricIDsUpdateInterval) ticker := time.NewTicker(currHourMetricIDsUpdateInterval)
defer ticker.Stop()
for { for {
select { select {
case <-s.stop: case <-s.stop:
s.updateCurrHourMetricIDs() s.updateCurrHourMetricIDs()
return return
case <-t.C: case <-ticker.C:
s.updateCurrHourMetricIDs() s.updateCurrHourMetricIDs()
t.Reset(currHourMetricIDsUpdateInterval)
} }
} }
} }

View file

@ -375,13 +375,13 @@ func (tb *table) startRetentionWatcher() {
} }
func (tb *table) retentionWatcher() { func (tb *table) retentionWatcher() {
t := time.NewTimer(time.Minute) ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for { for {
select { select {
case <-tb.stop: case <-tb.stop:
return return
case <-t.C: case <-ticker.C:
t.Reset(time.Minute)
} }
minTimestamp := timestampFromTime(time.Now()) - tb.retentionMilliseconds minTimestamp := timestampFromTime(time.Now()) - tb.retentionMilliseconds