From b26b00f72bb6f7f8dee57b7ed1b14378c5cef7b8 Mon Sep 17 00:00:00 2001
From: Alex Sharov <AskAlexSharov@gmail.com>
Date: Thu, 25 Jun 2020 14:34:40 +0700
Subject: [PATCH] increase IO buffers size (#686)

* increase IO buffers size

* found SSD threshold
---
 common/etl/buffers.go      | 1 +
 common/etl/dataprovider.go | 9 ++++-----
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/common/etl/buffers.go b/common/etl/buffers.go
index 94592280c..863376a7c 100644
--- a/common/etl/buffers.go
+++ b/common/etl/buffers.go
@@ -16,6 +16,7 @@ const (
 	SortableOldestAppearedBuffer
 
 	BufferOptimalSize = 256 * 1024 * 1024 /* 256 mb | var because we want to sometimes change it from tests */
+	BufIOSize         = 64 * 4096         // 64 pages | default is 1 page | increasing further doesn't show speedup on SSD
 )
 
 type Buffer interface {
diff --git a/common/etl/dataprovider.go b/common/etl/dataprovider.go
index 8fb5f45d7..bfdd079b7 100644
--- a/common/etl/dataprovider.go
+++ b/common/etl/dataprovider.go
@@ -35,12 +35,11 @@ func FlushToDisk(encoder Encoder, currentKey []byte, b Buffer, datadir string) (
 	if err != nil {
 		return nil, err
 	}
+	defer bufferFile.Sync() //nolint:errcheck
 
-	w := bufio.NewWriter(bufferFile)
+	w := bufio.NewWriterSize(bufferFile, BufIOSize)
 	defer w.Flush() //nolint:errcheck
-	defer func() {
-		bufferFile.Sync() //nolint:errcheck
-	}()
+
 	encoder.Reset(w)
 	for _, entry := range b.GetEntries() {
 		err = writeToDisk(encoder, entry.key, entry.value)
@@ -66,7 +65,7 @@ func (p *fileDataProvider) Next(decoder Decoder) ([]byte, []byte, error) {
 		if err != nil {
 			return nil, nil, err
 		}
-		p.reader = bufio.NewReader(p.file)
+		p.reader = bufio.NewReaderSize(p.file, BufIOSize)
 	}
 	decoder.Reset(p.reader)
 	return readElementFromDisk(decoder)