Maximize disk utilization during Open (#682)

During badger.Open, we read the SSTables from disk and load their index. Previously, this was being done serially, where we open one table at a time, and then we were using 64 goroutines to read the first keys of each block within the SSTable.

With this PR, this has been switched around. Using a serial reader to read the block keys provides the same performance as using the 64 goroutines to do random reads (given MemoryMap or LoadToRAM). So, we use a simpler serial reader for `readIndex`.

This PR uses 3 goroutines to load multiple tables at a time. Using 3 goroutines has shown to provide maximum disk read throughput utilization, a sweet spot between SSDs and HDDs. Using 8 goroutines to load tables is worse on HDDs than just using 1, due to too much random IOPS.

On internal SSD, average read throughput for master is ~400 MBps. With this PR, the throughput is >500MBps, which is the same throughput as possible via cat (i.e. best possible). During these tests, care was taken to ensure that disk cache was flushed out, using:

```
#!/bin/sh
echo 3 | sudo tee /proc/sys/vm/drop_caches
sudo blockdev --flushbufs /dev/sdb1
```

Changes:
* Attempt at making open tables concurrent. But, my testing shows it to be slower than opening one table at a time. Probably largely due to the fact that I'm testing on external rotational HDD. (Initially 8 goroutines were being used. Using 1 resulted in subpar disk utilization. Using 3 proved ideal in the test set).
* Saw a huge perf improvement when reading the tables serially to generate block index (turned out to be the same).
* Use concurrency of 2 to open tables. Serial algo for read index (later switched to 3 goroutines).
* Improve logging
* Self-review

Author: manishrjain

badger/cmd/fill.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/badger"
+	"github.com/dgraph-io/badger/options"
 	"github.com/dgraph-io/badger/y"
 	"github.com/spf13/cobra"
 )
@@ -56,6 +57,8 @@ func fill(cmd *cobra.Command, args []string) error {
 	opts.Truncate = truncate
 	opts.SyncWrites = false
 	opts.CompactL0OnClose = force
+	opts.TableLoadingMode = options.FileIO
+	opts.ValueLogLoadingMode = options.FileIO
 
 	db, err := badger.Open(opts)
 	if err != nil {

badger/cmd/info.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/dgraph-io/badger"
+	"github.com/dgraph-io/badger/options"
 	"github.com/dgraph-io/badger/table"
 	"github.com/dgraph-io/badger/y"
 	humanize "github.com/dustin/go-humanize"
@@ -77,6 +78,7 @@ func dur(src, dst time.Time) string {
 func tableInfo(dir, valueDir string) error {
 	// Open DB
 	opts := badger.DefaultOptions
+	opts.TableLoadingMode = options.MemoryMap
 	opts.Dir = sstDir
 	opts.ValueDir = vlogDir
 	opts.ReadOnly = true

badger/main.go

@@ -37,5 +37,6 @@ func main() {
 		}
 	}()
 	runtime.SetBlockProfileRate(100)
+	runtime.GOMAXPROCS(128)
 	cmd.Execute()
 }

levels.go

@@ -22,6 +22,8 @@ import (
 	"math/rand"
 	"os"
 	"sort"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"golang.org/x/net/trace"
@@ -101,33 +103,89 @@ func newLevelsController(kv *DB, mf *Manifest) (*levelsController, error) {
 	}
 
 	// Some files may be deleted. Let's reload.
+	var flags uint32 = y.Sync
+	if kv.opt.ReadOnly {
+		flags |= y.ReadOnly
+	}
+
+	var mu sync.Mutex
 	tables := make([][]*table.Table, kv.opt.MaxLevels)
 	var maxFileID uint64
-	for fileID, tableManifest := range mf.Tables {
-		fname := table.NewFilename(fileID, kv.opt.Dir)
-		var flags uint32 = y.Sync
-		if kv.opt.ReadOnly {
-			flags |= y.ReadOnly
-		}
-		fd, err := y.OpenExistingFile(fname, flags)
-		if err != nil {
-			closeAllTables(tables)
-			return nil, errors.Wrapf(err, "Opening file: %q", fname)
-		}
 
-		t, err := table.OpenTable(fd, kv.opt.TableLoadingMode)
-		if err != nil {
-			closeAllTables(tables)
-			return nil, errors.Wrapf(err, "Opening table: %q", fname)
+	// Make errCh non-blocking for iteration over mf.Tables.
+	errCh := make(chan error, len(mf.Tables))
+
+	// We found that using 3 goroutines allows disk throughput to be utilized to its max.
+	// Disk utilization is the main thing we should focus on, while trying to read the data. That's
+	// the one factor that remains constant between HDD and SSD.
+	throttleCh := make(chan struct{}, 3)
+	flushThrottle := func() error {
+		close(throttleCh)
+		for range throttleCh {
+		}
+		close(errCh)
+		for err := range errCh {
+			if err != nil {
+				return err
+			}
 		}
+		return nil
+	}
 
-		level := tableManifest.Level
-		tables[level] = append(tables[level], t)
+	start := time.Now()
+	var numOpened int32
+	tick := time.NewTicker(3 * time.Second)
+	defer tick.Stop()
 
+	for fileID, tableManifest := range mf.Tables {
+		fname := table.NewFilename(fileID, kv.opt.Dir)
+	THROTTLE:
+		for {
+			select {
+			case throttleCh <- struct{}{}:
+				break THROTTLE
+			case <-tick.C:
+				Infof("%d tables out of %d opened in %s\n", atomic.LoadInt32(&numOpened),
+					len(mf.Tables), time.Since(start).Round(time.Millisecond))
+			case err := <-errCh:
+				if err != nil {
+					flushThrottle()
+					closeAllTables(tables)
+					return nil, err
+				}
+			}
+		}
 		if fileID > maxFileID {
 			maxFileID = fileID
 		}
+		go func(fname string, level int) {
+			defer func() {
+				<-throttleCh
+				atomic.AddInt32(&numOpened, 1)
+			}()
+			fd, err := y.OpenExistingFile(fname, flags)
+			if err != nil {
+				errCh <- errors.Wrapf(err, "Opening file: %q", fname)
+				return
+			}
+
+			t, err := table.OpenTable(fd, kv.opt.TableLoadingMode)
+			if err != nil {
+				errCh <- errors.Wrapf(err, "Opening table: %q", fname)
+				return
+			}
+
+			mu.Lock()
+			tables[level] = append(tables[level], t)
+			mu.Unlock()
+		}(fname, int(tableManifest.Level))
+	}
+	if err := flushThrottle(); err != nil {
+		closeAllTables(tables)
+		return nil, err
 	}
+	Infof("All %d tables opened in %s\n", atomic.LoadInt32(&numOpened),
+		time.Since(start).Round(time.Millisecond))
 	s.nextFileID = maxFileID + 1
 	for i, tbls := range tables {
 		s.levels[i].initTables(tbls)

table/table.go

@@ -17,8 +17,11 @@
 package table
 
 import (
+	"bufio"
+	"bytes"
 	"encoding/binary"
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"path/filepath"
@@ -150,7 +153,7 @@ func OpenTable(fd *os.File, loadingMode options.FileLoadingMode) (*Table, error)
 		}
 	}
 
-	if err := t.readIndex(); err != nil {
+	if err := t.readIndex(loadingMode); err != nil {
 		return nil, y.Wrap(err)
 	}
 
@@ -200,7 +203,7 @@ func (t *Table) readNoFail(off int, sz int) []byte {
 	return res
 }
 
-func (t *Table) readIndex() error {
+func (t *Table) readIndex(loadingMode options.FileLoadingMode) error {
 	readPos := t.tableSize
 
 	// Read bloom filter.
@@ -240,54 +243,39 @@ func (t *Table) readIndex() error {
 		t.blockIndex = append(t.blockIndex, ko)
 	}
 
-	che := make(chan error, len(t.blockIndex))
-	blocks := make(chan int, len(t.blockIndex))
-
-	for i := 0; i < len(t.blockIndex); i++ {
-		blocks <- i
-	}
-
-	for i := 0; i < 64; i++ { // Run 64 goroutines.
-		go func() {
-			var h header
-
-			for index := range blocks {
-				ko := &t.blockIndex[index]
-
-				offset := ko.offset
-				buf, err := t.read(offset, h.Size())
-				if err != nil {
-					che <- errors.Wrap(err, "While reading first header in block")
-					continue
-				}
-
-				h.Decode(buf)
-				y.AssertTruef(h.plen == 0, "Key offset: %+v, h.plen = %d", *ko, h.plen)
-
-				offset += h.Size()
-				buf = make([]byte, h.klen)
-				var out []byte
-				if out, err = t.read(offset, int(h.klen)); err != nil {
-					che <- errors.Wrap(err, "While reading first key in block")
-					continue
-				}
-				y.AssertTrue(len(buf) == copy(buf, out))
-
-				ko.key = buf
-				che <- nil
-			}
-		}()
-	}
-	close(blocks) // to stop reading goroutines
-
-	var readError error
-	for i := 0; i < len(t.blockIndex); i++ {
-		if err := <-che; err != nil && readError == nil {
-			readError = err
+	// Execute this index read serially, because all disks are orders of magnitude faster when read
+	// serially compared to executing random reads.
+	var h header
+	var offset int
+	var r *bufio.Reader
+	if loadingMode == options.LoadToRAM {
+		// We already read the table to put it into t.mmap. So, no point reading it again from disk.
+		// Instead use the read buffer.
+		r = bufio.NewReader(bytes.NewReader(t.mmap))
+	} else {
+		if _, err := t.fd.Seek(0, io.SeekStart); err != nil {
+			return err
 		}
+		r = bufio.NewReader(t.fd)
 	}
-	if readError != nil {
-		return readError
+	hbuf := make([]byte, h.Size())
+	for idx := range t.blockIndex {
+		ko := &t.blockIndex[idx]
+		if _, err := r.Discard(ko.offset - offset); err != nil {
+			return err
+		}
+		offset = ko.offset
+		if _, err := io.ReadFull(r, hbuf); err != nil {
+			return err
+		}
+		offset += len(hbuf)
+		h.Decode(hbuf)
+		y.AssertTrue(h.plen == 0)
+		ko.key = make([]byte, h.klen)
+		if _, err := io.ReadFull(r, ko.key); err != nil {
+			return err
+		}
+		offset += len(ko.key)
 	}
 
 	return nil

util.go

@@ -17,6 +17,7 @@
 package badger
 
 import (
+	"encoding/hex"
 	"io/ioutil"
 	"math/rand"
 	"sync/atomic"
@@ -77,8 +78,9 @@ func (s *levelHandler) validate() error {
 
 		if y.CompareKeys(s.tables[j-1].Biggest(), s.tables[j].Smallest()) >= 0 {
 			return errors.Errorf(
-				"Inter: %q vs %q: level=%d j=%d numTables=%d",
-				string(s.tables[j-1].Biggest()), string(s.tables[j].Smallest()), s.level, j, numTables)
+				"Inter: Biggest(j-1) \n%s\n vs Smallest(j): \n%s\n: level=%d j=%d numTables=%d",
+				hex.Dump(s.tables[j-1].Biggest()), hex.Dump(s.tables[j].Smallest()),
+				s.level, j, numTables)
 		}
 
 		if y.CompareKeys(s.tables[j].Smallest(), s.tables[j].Biggest()) > 0 {