Merge pull request #99 from yandex/feature/85

GH-85: enlarge buffers for phout aggregator
yandex · Feb 22, 2018 · da8659a · da8659a
2 parents 7d539f8 + 405145a
commit da8659a
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 7 deletions.
diff --git a/core/aggregator/netsample/phout.go b/core/aggregator/netsample/phout.go
@@ -8,15 +8,30 @@ import (
 	"strconv"
 	"time"
 
+	"github.com/c2h5oh/datasize"
 	"github.com/pkg/errors"
 	"github.com/spf13/afero"
 
 	"github.com/yandex/pandora/core"
+	"github.com/yandex/pandora/core/coreutil"
 )
 
 type PhoutConfig struct {
-	Destination string // Destination file name
-	Id          bool   // Print ammo ids if true.
+	Destination     string                    // Destination file name
+	Id              bool                      // Print ammo ids if true.
+	FlushTime       time.Duration             `config:"flush-time"`
+	SampleQueueSize int                       `config:"sample-queue-size"`
+	Buffer          coreutil.BufferSizeConfig `config:",squash"`
+}
+
+func DefaultPhoutConfig() PhoutConfig {
+	return PhoutConfig{
+		FlushTime:       time.Second,
+		SampleQueueSize: 256 * 1024,
+		Buffer: coreutil.BufferSizeConfig{
+			BufferSize: 8 * datasize.MB,
+		},
+	}
 }
 
 func NewPhout(fs afero.Fs, conf PhoutConfig) (a Aggregator, err error) {
@@ -31,8 +46,8 @@ func NewPhout(fs afero.Fs, conf PhoutConfig) (a Aggregator, err error) {
 	}
 	a = &phoutAggregator{
 		config: conf,
-		sink:   make(chan *Sample, 32*1024),
-		writer: bufio.NewWriterSize(file, 512*1024),
+		sink:   make(chan *Sample, conf.SampleQueueSize),
+		writer: bufio.NewWriterSize(file, conf.Buffer.BufferSizeOrDefault()),
 		buf:    make([]byte, 0, 1024),
 		file:   file,
 	}

diff --git a/core/aggregator/netsample/phout_test.go b/core/aggregator/netsample/phout_test.go
@@ -30,7 +30,8 @@ var _ = Describe("Phout", func() {
 
 	BeforeEach(func() {
 		fs = afero.NewMemMapFs()
-		conf = PhoutConfig{Destination: fileName}
+		conf = DefaultPhoutConfig()
+		conf.Destination = fileName
 		ctx, cancel = context.WithCancel(context.Background())
 	})
 	JustBeforeEach(func() {

diff --git a/core/coreutil/buffer_size_config.go b/core/coreutil/buffer_size_config.go
@@ -9,7 +9,7 @@ import (
 	"github.com/c2h5oh/datasize"
 )
 
-const DefaultBufferSize = 256 * 1024
+const DefaultBufferSize = 512 * 1024
 const MinimalBufferSize = 4 * 1024
 
 // BufferSizeConfig SHOULD be used to configure buffer size.

diff --git a/core/import/import.go b/core/import/import.go
@@ -72,7 +72,7 @@ func Import(fs afero.Fs) {
 	register.Aggregator("phout", func(conf netsample.PhoutConfig) (core.Aggregator, error) {
 		a, err := netsample.NewPhout(fs, conf)
 		return netsample.WrapAggregator(a), err
-	})
+	}, netsample.DefaultPhoutConfig)
 	register.Aggregator("jsonlines", aggregator.NewJSONLinesAggregator, aggregator.DefaultJSONLinesAggregatorConfig)
 	register.Aggregator("json", aggregator.NewJSONLinesAggregator, aggregator.DefaultJSONLinesAggregatorConfig) // TODO(skipor): should be done via alias, but we don't have them yet
 	register.Aggregator("log", aggregator.NewLog)