mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-02-02 08:31:11 -08:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Remove WaitGroup and extra goroutine.

Browse source 
Signed-off-by: Tom Wilkie <tom.wilkie@gmail.com>
This commit is contained in:
Tom Wilkie 2018-02-01 13:20:38 +00:00
parent f3c61f8bb2
commit aa17263edd
1 changed files with 16 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
storage/remote/queue_manager.go
View file

@ -16,6 +16,7 @@ package remote
import ( import (
"math" "math"
"sync" "sync"
"sync/atomic"
"time" "time"
"golang.org/x/time/rate" "golang.org/x/time/rate"
@ -375,7 +376,7 @@ type shards struct {
qm *QueueManager qm *QueueManager
queues []chan *model.Sample queues []chan *model.Sample
done chan struct{} done chan struct{}
wg sync.WaitGroup running int32
} }
func (t *QueueManager) newShards(numShards int) *shards { func (t *QueueManager) newShards(numShards int) *shards {
@ -387,8 +388,8 @@ func (t *QueueManager) newShards(numShards int) *shards {
qm: t, qm: t,
queues: queues, queues: queues,
done: make(chan struct{}), done: make(chan struct{}),
running: int32(numShards),
} }
s.wg.Add(numShards)
return s return s
} }
@ -407,14 +408,8 @@ func (s *shards) stop() {
close(shard) close(shard)
} }
done := make(chan struct{})
go func() {
s.wg.Wait()
close(done)
}()
select { select {
case <-done: case <-s.done:
case <-time.After(stopFlushDeadline): case <-time.After(stopFlushDeadline):
level.Error(s.qm.logger).Log("msg", "Failed to flush all samples on shutdown") level.Error(s.qm.logger).Log("msg", "Failed to flush all samples on shutdown")
} }
@ -435,7 +430,12 @@ func (s *shards) enqueue(sample *model.Sample) bool {
} }
func (s *shards) runShard(i int) { func (s *shards) runShard(i int) {
defer s.wg.Done() defer func() {
if atomic.AddInt32(&s.running, -1) == 0 {
close(s.done)
}
}()
queue := s.queues[i] queue := s.queues[i]
// Send batches of at most MaxSamplesPerSend samples to the remote storage. // Send batches of at most MaxSamplesPerSend samples to the remote storage.