RepoMirrors/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus synced 2024-12-27 00:53:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Spawn grouping of fingerprints with free semaphore.

Browse Source 
The previous implementation spawned N goroutines to group samples
together and would not start work until the semaphore unblocked.
While this didn't leak, it polluted the scheduling space.  Thusly,
the routine only starts after a semaphore has been acquired.
This commit is contained in:
Matt T. Proud 2013-05-21 16:11:35 +02:00
parent 053f4296d0
commit 74a66fd938
1 changed files with 11 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
storage/metric/leveldb.go
View File

@ -30,9 +30,7 @@ import (
"time"
)
var (
leveldbChunkSize = flag.Int("leveldbChunkSize", 200, "Maximum number of samples stored under one key.")
const (
sortConcurrency = 2
)
@ -47,6 +45,8 @@ type LevelDBMetricPersistence struct {
}
var (
leveldbChunkSize = flag.Int("leveldbChunkSize", 200, "Maximum number of samples stored under one key.")
// These flag values are back of the envelope, though they seem sensible.
// Please re-evaluate based on your own needs.
curationRemarksCacheSize = flag.Int("curationRemarksCacheSize", 50*1024*1024, "The size for the curation remarks cache (bytes).")
@ -188,9 +188,7 @@ func (l *LevelDBMetricPersistence) AppendSample(sample model.Sample) (err error)
// together by their respective metric fingerprint, and finally sorts
// them chronologically.
func groupByFingerprint(samples model.Samples) map[model.Fingerprint]model.Samples {
var (
fingerprintToSamples = map[model.Fingerprint]model.Samples{}
)
fingerprintToSamples := map[model.Fingerprint]model.Samples{}
for _, sample := range samples {
fingerprint := *model.NewFingerprintFromMetric(sample.Metric)
@ -199,10 +197,8 @@ func groupByFingerprint(samples model.Samples) map[model.Fingerprint]model.Sampl
fingerprintToSamples[fingerprint] = samples
}
var (
sortingSemaphore = make(chan bool, sortConcurrency)
doneSorting sync.WaitGroup
)
sortingSemaphore := make(chan bool, sortConcurrency)
doneSorting := sync.WaitGroup{}
for i := 0; i < sortConcurrency; i++ {
sortingSemaphore <- true
@ -211,8 +207,8 @@ func groupByFingerprint(samples model.Samples) map[model.Fingerprint]model.Sampl
for _, samples := range fingerprintToSamples {
doneSorting.Add(1)
<-sortingSemaphore
go func(samples model.Samples) {
<-sortingSemaphore
sort.Sort(samples)
sortingSemaphore <- true
doneSorting.Done()
@ -538,16 +534,12 @@ func (l *LevelDBMetricPersistence) AppendSamples(samples model.Samples) (err err
recordOutcome(duration, err, map[string]string{operation: appendSamples, result: success}, map[string]string{operation: appendSamples, result: failure})
}(time.Now())
var (
fingerprintToSamples = groupByFingerprint(samples)
indexErrChan = make(chan error, 1)
watermarkErrChan = make(chan error, 1)
)
fingerprintToSamples := groupByFingerprint(samples)
indexErrChan := make(chan error, 1)
watermarkErrChan := make(chan error, 1)
go func(groups map[model.Fingerprint]model.Samples) {
var (
metrics = map[model.Fingerprint]model.Metric{}
)
metrics := map[model.Fingerprint]model.Metric{}
for fingerprint, samples := range groups {
metrics[fingerprint] = samples[0].Metric