-
Notifications
You must be signed in to change notification settings - Fork 5.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Performance refactor of running_output buffers
- Loading branch information
Showing
7 changed files
with
529 additions
and
198 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| package buffer | ||
|
|
||
| import ( | ||
| "github.com/influxdata/telegraf" | ||
| ) | ||
|
|
||
| // Buffer is an object for storing metrics in a circular buffer. | ||
| type Buffer struct { | ||
| buf chan telegraf.Metric | ||
| // total dropped metrics | ||
| drops int | ||
| // total metrics added | ||
| total int | ||
| } | ||
|
|
||
| // NewBuffer returns a Buffer | ||
| // size is the maximum number of metrics that Buffer will cache. If Add is | ||
| // called when the buffer is full, then the oldest metric(s) will be dropped. | ||
| func NewBuffer(size int) *Buffer { | ||
| return &Buffer{ | ||
| buf: make(chan telegraf.Metric, size), | ||
| } | ||
| } | ||
|
|
||
| // IsEmpty returns true if Buffer is empty. | ||
| func (b *Buffer) IsEmpty() bool { | ||
| return len(b.buf) == 0 | ||
| } | ||
|
|
||
| // Len returns the current length of the buffer. | ||
| func (b *Buffer) Len() int { | ||
| return len(b.buf) | ||
| } | ||
|
|
||
| // Drops returns the total number of dropped metrics that have occured in this | ||
| // buffer since instantiation. | ||
| func (b *Buffer) Drops() int { | ||
| return b.drops | ||
| } | ||
|
|
||
| // Total returns the total number of metrics that have been added to this buffer. | ||
| func (b *Buffer) Total() int { | ||
| return b.total | ||
| } | ||
|
|
||
| // Add adds metrics to the buffer. | ||
| func (b *Buffer) Add(metrics ...telegraf.Metric) { | ||
| for i, _ := range metrics { | ||
| b.total++ | ||
| select { | ||
| case b.buf <- metrics[i]: | ||
| default: | ||
| b.drops++ | ||
| <-b.buf | ||
| b.buf <- metrics[i] | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Batch returns a batch of metrics of size batchSize. | ||
| // the batch will be of maximum length batchSize. It can be less than batchSize, | ||
| // if the length of Buffer is less than batchSize. | ||
| func (b *Buffer) Batch(batchSize int) []telegraf.Metric { | ||
| n := min(len(b.buf), batchSize) | ||
| out := make([]telegraf.Metric, n) | ||
| for i := 0; i < n; i++ { | ||
| out[i] = <-b.buf | ||
| } | ||
| return out | ||
| } | ||
|
|
||
| func min(a, b int) int { | ||
| if b < a { | ||
| return b | ||
| } | ||
| return a | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| package buffer | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/influxdata/telegraf" | ||
| "github.com/influxdata/telegraf/testutil" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| var metricList = []telegraf.Metric{ | ||
| testutil.TestMetric(2, "mymetric1"), | ||
| testutil.TestMetric(1, "mymetric2"), | ||
| testutil.TestMetric(11, "mymetric3"), | ||
| testutil.TestMetric(15, "mymetric4"), | ||
| testutil.TestMetric(8, "mymetric5"), | ||
| } | ||
|
|
||
| func BenchmarkAddMetrics(b *testing.B) { | ||
| buf := NewBuffer(10000) | ||
| m := testutil.TestMetric(1, "mymetric") | ||
| for n := 0; n < b.N; n++ { | ||
| buf.Add(m) | ||
| } | ||
| } | ||
|
|
||
| func TestNewBufferBasicFuncs(t *testing.T) { | ||
| b := NewBuffer(10) | ||
|
|
||
| assert.True(t, b.IsEmpty()) | ||
| assert.Zero(t, b.Len()) | ||
| assert.Zero(t, b.Drops()) | ||
| assert.Zero(t, b.Total()) | ||
|
|
||
| m := testutil.TestMetric(1, "mymetric") | ||
| b.Add(m) | ||
| assert.False(t, b.IsEmpty()) | ||
| assert.Equal(t, b.Len(), 1) | ||
| assert.Equal(t, b.Drops(), 0) | ||
| assert.Equal(t, b.Total(), 1) | ||
|
|
||
| b.Add(metricList...) | ||
| assert.False(t, b.IsEmpty()) | ||
| assert.Equal(t, b.Len(), 6) | ||
| assert.Equal(t, b.Drops(), 0) | ||
| assert.Equal(t, b.Total(), 6) | ||
| } | ||
|
|
||
| func TestDroppingMetrics(t *testing.T) { | ||
| b := NewBuffer(10) | ||
|
|
||
| // Add up to the size of the buffer | ||
| b.Add(metricList...) | ||
| b.Add(metricList...) | ||
| assert.False(t, b.IsEmpty()) | ||
| assert.Equal(t, b.Len(), 10) | ||
| assert.Equal(t, b.Drops(), 0) | ||
| assert.Equal(t, b.Total(), 10) | ||
|
|
||
| // Add 5 more and verify they were dropped | ||
| b.Add(metricList...) | ||
| assert.False(t, b.IsEmpty()) | ||
| assert.Equal(t, b.Len(), 10) | ||
| assert.Equal(t, b.Drops(), 5) | ||
| assert.Equal(t, b.Total(), 15) | ||
| } | ||
|
|
||
| func TestGettingBatches(t *testing.T) { | ||
| b := NewBuffer(20) | ||
|
|
||
| // Verify that the buffer returned is smaller than requested when there are | ||
| // not as many items as requested. | ||
| b.Add(metricList...) | ||
| batch := b.Batch(10) | ||
| assert.Len(t, batch, 5) | ||
|
|
||
| // Verify that the buffer is now empty | ||
| assert.True(t, b.IsEmpty()) | ||
| assert.Zero(t, b.Len()) | ||
| assert.Zero(t, b.Drops()) | ||
| assert.Equal(t, b.Total(), 5) | ||
|
|
||
| // Verify that the buffer returned is not more than the size requested | ||
| b.Add(metricList...) | ||
| batch = b.Batch(3) | ||
| assert.Len(t, batch, 3) | ||
|
|
||
| // Verify that buffer is not empty | ||
| assert.False(t, b.IsEmpty()) | ||
| assert.Equal(t, b.Len(), 2) | ||
| assert.Equal(t, b.Drops(), 0) | ||
| assert.Equal(t, b.Total(), 10) | ||
| } |
Oops, something went wrong.