Optimized and increased read performance by ~2-3x by memory mapping the read-only datafiles

.drone.yml

@@ -5,7 +5,7 @@ steps:
   - name: build
     image: golang:latest
     commands:
-      - go test -v -short -cover -coverprofile=coverage.txt -coverpkg=$(go list) ./...
+      - go test -v -short -cover -coverprofile=coverage.txt -coverpkg=$(go list) .
 
   - name: coverage
     image: plugins/codecov

Makefile

@@ -34,13 +34,17 @@ release:
 	@./tools/release.sh
 
 profile: build
-	@go test -cpuprofile cpu.prof -memprofile mem.prof -v -bench ./...
+	@go test -cpuprofile cpu.prof -memprofile mem.prof -v -bench .
 
 bench: build
-	@go test -v -benchmem -bench=. ./...
+	@go test -v -benchmem -bench=. .
 
 test: build
-	@go test -v -cover -coverprofile=coverage.txt -covermode=atomic -coverpkg=$(shell go list) -race ./...
+	@go test -v \
+		-cover -coverprofile=coverage.txt -covermode=atomic \
+		-coverpkg=$(shell go list) \
+		-race \
+		.
 
 clean:
 	@git clean -f -d -X

README.md

@@ -94,32 +94,32 @@ Benchmarks run on a 11" Macbook with a 1.4Ghz Intel Core i7:
 ```
 $ make bench
 ...
-BenchmarkGet/128B-4         	  300000	      5178 ns/op	     400 B/op	       5 allocs/op
-BenchmarkGet/256B-4         	  300000	      5273 ns/op	     656 B/op	       5 allocs/op
-BenchmarkGet/512B-4         	  200000	      5368 ns/op	    1200 B/op	       5 allocs/op
-BenchmarkGet/1K-4           	  200000	      5800 ns/op	    2288 B/op	       5 allocs/op
-BenchmarkGet/2K-4           	  200000	      6766 ns/op	    4464 B/op	       5 allocs/op
-BenchmarkGet/4K-4           	  200000	      7857 ns/op	    9072 B/op	       5 allocs/op
-BenchmarkGet/8K-4           	  200000	      9538 ns/op	   17776 B/op	       5 allocs/op
-BenchmarkGet/16K-4          	  100000	     13188 ns/op	   34928 B/op	       5 allocs/op
-BenchmarkGet/32K-4          	  100000	     21620 ns/op	   73840 B/op	       5 allocs/op
+BenchmarkGet/128B-4         	  500000	      2537 ns/op	     672 B/op	       7 allocs/op
+BenchmarkGet/256B-4         	  500000	      2629 ns/op	    1056 B/op	       7 allocs/op
+BenchmarkGet/512B-4         	  500000	      2773 ns/op	    1888 B/op	       7 allocs/op
+BenchmarkGet/1K-4           	  500000	      3202 ns/op	    3552 B/op	       7 allocs/op
+BenchmarkGet/2K-4           	  300000	      3904 ns/op	    6880 B/op	       7 allocs/op
+BenchmarkGet/4K-4           	  300000	      5678 ns/op	   14048 B/op	       7 allocs/op
+BenchmarkGet/8K-4           	  200000	      8948 ns/op	   27360 B/op	       7 allocs/op
+BenchmarkGet/16K-4          	  100000	     14635 ns/op	   53472 B/op	       7 allocs/op
+BenchmarkGet/32K-4          	   50000	     28292 ns/op	  114912 B/op	       7 allocs/op
 
-BenchmarkPut/128B-4         	  200000	      7875 ns/op	     409 B/op	       6 allocs/op
-BenchmarkPut/256B-4         	  200000	      8712 ns/op	     538 B/op	       6 allocs/op
-BenchmarkPut/512B-4         	  200000	      9832 ns/op	     829 B/op	       6 allocs/op
-BenchmarkPut/1K-4           	  100000	     13105 ns/op	    1410 B/op	       6 allocs/op
-BenchmarkPut/2K-4           	  100000	     18601 ns/op	    2572 B/op	       6 allocs/op
-BenchmarkPut/4K-4           	   50000	     36631 ns/op	    5151 B/op	       6 allocs/op
-BenchmarkPut/8K-4           	   30000	     56128 ns/op	    9798 B/op	       6 allocs/op
-BenchmarkPut/16K-4          	   20000	     83209 ns/op	   18834 B/op	       6 allocs/op
-BenchmarkPut/32K-4          	   10000	    135899 ns/op	   41517 B/op	       6 allocs/op
+BenchmarkPut/128B-4         	  200000	      8173 ns/op	     409 B/op	       6 allocs/op
+BenchmarkPut/256B-4         	  200000	      8404 ns/op	     538 B/op	       6 allocs/op
+BenchmarkPut/512B-4         	  200000	      9741 ns/op	     829 B/op	       6 allocs/op
+BenchmarkPut/1K-4           	  100000	     13118 ns/op	    1411 B/op	       6 allocs/op
+BenchmarkPut/2K-4           	  100000	     17982 ns/op	    2573 B/op	       6 allocs/op
+BenchmarkPut/4K-4           	   50000	     35477 ns/op	    5154 B/op	       6 allocs/op
+BenchmarkPut/8K-4           	   30000	     54021 ns/op	    9804 B/op	       6 allocs/op
+BenchmarkPut/16K-4          	   20000	     96551 ns/op	   18849 B/op	       6 allocs/op
+BenchmarkPut/32K-4          	   10000	    129957 ns/op	   41561 B/op	       7 allocs/op
 
-BenchmarkScan-4             	 1000000	      1851 ns/op	     493 B/op	      25 allocs/op
+BenchmarkScan-4             	 1000000	      2011 ns/op	     493 B/op	      25 allocs/op
 ```
 
 For 128B values:
 
-* ~200,000 reads/sec
+* ~400,000 reads/sec
 * ~130,000 writes/sec
 
 The full benchmark above shows linear performance as you increase key/value sizes.

bitcask.go

@@ -89,7 +89,7 @@ func (b *Bitcask) Get(key string) ([]byte, error) {
 		df = b.datafiles[item.FileID]
 	}
 
-	e, err := df.ReadAt(item.Offset)
+	e, err := df.ReadAt(item.Offset, item.Size)
 	if err != nil {
 		return nil, err
 	}
@@ -117,12 +117,12 @@ func (b *Bitcask) Put(key string, value []byte) error {
 		return ErrValueTooLarge
 	}
 
-	offset, err := b.put(key, value)
+	offset, n, err := b.put(key, value)
 	if err != nil {
 		return err
 	}
 
-	item := b.keydir.Add(key, b.curr.FileID(), offset)
+	item := b.keydir.Add(key, b.curr.FileID(), offset, n)
 	b.trie.Add(key, item)
 
 	return nil
@@ -131,7 +131,7 @@ func (b *Bitcask) Put(key string, value []byte) error {
 // Delete deletes the named key. If the key doesn't exist or an I/O error
 // occurs the error is returned.
 func (b *Bitcask) Delete(key string) error {
-	_, err := b.put(key, []byte{})
+	_, _, err := b.put(key, []byte{})
 	if err != nil {
 		return err
 	}
@@ -177,7 +177,7 @@ func (b *Bitcask) Fold(f func(key string) error) error {
 	return nil
 }
 
-func (b *Bitcask) put(key string, value []byte) (int64, error) {
+func (b *Bitcask) put(key string, value []byte) (int64, int64, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
@@ -185,12 +185,12 @@ func (b *Bitcask) put(key string, value []byte) (int64, error) {
 	if size >= int64(b.config.maxDatafileSize) {
 		err := b.curr.Close()
 		if err != nil {
-			return -1, err
+			return -1, 0, err
 		}
 
 		df, err := internal.NewDatafile(b.path, b.curr.FileID(), true)
 		if err != nil {
-			return -1, err
+			return -1, 0, err
 		}
 
 		b.datafiles = append(b.datafiles, df)
@@ -198,7 +198,7 @@ func (b *Bitcask) put(key string, value []byte) (int64, error) {
 		id := b.curr.FileID() + 1
 		curr, err := internal.NewDatafile(b.path, id, false)
 		if err != nil {
-			return -1, err
+			return -1, 0, err
 		}
 		b.curr = curr
 	}
@@ -255,7 +255,7 @@ func Merge(path string, force bool) error {
 		defer df.Close()
 
 		for {
-			e, err := df.Read()
+			e, n, err := df.Read()
 			if err != nil {
 				if err == io.EOF {
 					break
@@ -269,7 +269,7 @@ func Merge(path string, force bool) error {
 				continue
 			}
 
-			keydir.Add(e.Key, ids[i], e.Offset)
+			keydir.Add(e.Key, ids[i], e.Offset, n)
 		}
 
 		tempdf, err := internal.NewDatafile(temp, id, false)
@@ -280,12 +280,12 @@ func Merge(path string, force bool) error {
 
 		for key := range keydir.Keys() {
 			item, _ := keydir.Get(key)
-			e, err := df.ReadAt(item.Offset)
+			e, err := df.ReadAt(item.Offset, item.Size)
 			if err != nil {
 				return err
 			}
 
-			_, err = tempdf.Write(e)
+			_, _, err = tempdf.Write(e)
 			if err != nil {
 				return err
 			}
@@ -365,12 +365,12 @@ func Open(path string, options ...Option) (*Bitcask, error) {
 
 			for key := range hint.Keys() {
 				item, _ := hint.Get(key)
-				_ = keydir.Add(key, item.FileID, item.Offset)
+				_ = keydir.Add(key, item.FileID, item.Offset, item.Size)
 				trie.Add(key, item)
 			}
 		} else {
 			for {
-				e, err := df.Read()
+				e, n, err := df.Read()
 				if err != nil {
 					if err == io.EOF {
 						break
@@ -384,7 +384,7 @@ func Open(path string, options ...Option) (*Bitcask, error) {
 					continue
 				}
 
-				item := keydir.Add(e.Key, ids[i], e.Offset)
+				item := keydir.Add(e.Key, ids[i], e.Offset, n)
 				trie.Add(e.Key, item)
 			}
 		}

go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/tidwall/redcon v1.0.0
 	github.com/ugorji/go/codec v0.0.0-20190320090025-2dc34c0b8780 // indirect
 	golang.org/x/crypto v0.0.0-20190320223903-b7391e95e576 // indirect
+	golang.org/x/exp v0.0.0-20190321205749-f0864edee7f3
 	golang.org/x/net v0.0.0-20190320064053-1272bf9dcd53 // indirect
 	golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6 // indirect
 	golang.org/x/sys v0.0.0-20190322080309-f49334f85ddc // indirect

go.sum

@@ -1,5 +1,6 @@
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.12+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
@@ -84,6 +85,10 @@ golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnf
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190320223903-b7391e95e576 h1:aUX/1G2gFSs4AsJJg2cL3HuoRhCSCz733FE5GUSuaT4=
 golang.org/x/crypto v0.0.0-20190320223903-b7391e95e576/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/exp v0.0.0-20190321205749-f0864edee7f3 h1:Ep4L2ibjtJcW6IP73KbcJAU0cpNKsLNSSP2jE1xlCys=
+golang.org/x/exp v0.0.0-20190321205749-f0864edee7f3/go.mod h1:ZjyILWgesfNpC6sMxTJOJm9Kp84zZh5NQWvqDGG3Qr8=
+golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
+golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190320064053-1272bf9dcd53/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTmV7VDcZyvRZ+QQXkXTZQ=
@@ -93,12 +98,14 @@ golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a h1:1n5lsVfiQW3yfsRGu98756EH1YthsFqr/5mxHduZW2A=
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190312061237-fead79001313/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190322080309-f49334f85ddc h1:4gbWbmmPFp4ySWICouJl6emP0MyS31yy9SrTlAGFT+g=
 golang.org/x/sys v0.0.0-20190322080309-f49334f85ddc/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20181030221726-6c7e314b6563/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190321232350-e250d351ecad/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

internal/datafile.go

@@ -1,12 +1,14 @@
 package internal
 
 import (
+	"bytes"
 	"fmt"
 	"os"
 	"path/filepath"
 	"sync"
 
 	"github.com/pkg/errors"
+	"golang.org/x/exp/mmap"
 
 	pb "github.com/prologic/bitcask/internal/proto"
 	"github.com/prologic/bitcask/internal/streampb"
@@ -17,7 +19,8 @@ const (
 )
 
 var (
-	ErrReadonly = errors.New("error: read only datafile")
+	ErrReadonly  = errors.New("error: read only datafile")
+	ErrReadError = errors.New("error: read error")
 )
 
 type Datafile struct {
@@ -25,6 +28,7 @@ type Datafile struct {
 
 	id     int
 	r      *os.File
+	ra     *mmap.ReaderAt
 	w      *os.File
 	offset int64
 	dec    *streampb.Decoder
@@ -34,6 +38,7 @@ type Datafile struct {
 func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
 	var (
 		r   *os.File
+		ra  *mmap.ReaderAt
 		w   *os.File
 		err error
 	)
@@ -56,6 +61,11 @@ func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
 		return nil, errors.Wrap(err, "error calling Stat()")
 	}
 
+	ra, err = mmap.Open(fn)
+	if err != nil {
+		return nil, err
+	}
+
 	offset := stat.Size()
 
 	dec := streampb.NewDecoder(r)
@@ -64,6 +74,7 @@ func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
 	return &Datafile{
 		id:     id,
 		r:      r,
+		ra:     ra,
 		w:      w,
 		offset: offset,
 		dec:    dec,
@@ -81,6 +92,10 @@ func (df *Datafile) Name() string {
 
 func (df *Datafile) Close() error {
 	if df.w == nil {
+		err := df.ra.Close()
+		if err != nil {
+			return err
+		}
 		return df.r.Close()
 	}
 
@@ -104,28 +119,45 @@ func (df *Datafile) Size() int64 {
 	return df.offset
 }
 
-func (df *Datafile) Read() (e pb.Entry, err error) {
+func (df *Datafile) Read() (e pb.Entry, n int64, err error) {
 	df.Lock()
 	defer df.Unlock()
 
-	return e, df.dec.Decode(&e)
-}
-
-func (df *Datafile) ReadAt(index int64) (e pb.Entry, err error) {
-	df.Lock()
-	defer df.Unlock()
-
-	_, err = df.r.Seek(index, os.SEEK_SET)
+	n, err = df.dec.Decode(&e)
 	if err != nil {
 		return
 	}
 
-	return e, df.dec.Decode(&e)
+	return
 }
 
-func (df *Datafile) Write(e pb.Entry) (int64, error) {
+func (df *Datafile) ReadAt(index, size int64) (e pb.Entry, err error) {
+	var n int
+
+	b := make([]byte, size)
+
 	if df.w == nil {
-		return -1, ErrReadonly
+		n, err = df.ra.ReadAt(b, index)
+	} else {
+		n, err = df.r.ReadAt(b, index)
+	}
+	if err != nil {
+		return
+	}
+	if int64(n) != size {
+		err = ErrReadError
+		return
+	}
+
+	buf := bytes.NewBuffer(b)
+	dec := streampb.NewDecoder(buf)
+	_, err = dec.Decode(&e)
+	return
+}
+
+func (df *Datafile) Write(e pb.Entry) (int64, int64, error) {
+	if df.w == nil {
+		return -1, 0, ErrReadonly
 	}
 
 	df.Lock()
@@ -135,9 +167,9 @@ func (df *Datafile) Write(e pb.Entry) (int64, error) {
 
 	n, err := df.enc.Encode(&e)
 	if err != nil {
-		return -1, err
+		return -1, 0, err
 	}
 	df.offset += n
 
-	return e.Offset, nil
+	return e.Offset, n, nil
 }

internal/keydir.go

@@ -11,6 +11,7 @@ import (
 type Item struct {
 	FileID int
 	Offset int64
+	Size   int64
 }
 
 type Keydir struct {
@@ -24,10 +25,11 @@ func NewKeydir() *Keydir {
 	}
 }
 
-func (k *Keydir) Add(key string, fileid int, offset int64) Item {
+func (k *Keydir) Add(key string, fileid int, offset, size int64) Item {
 	item := Item{
 		FileID: fileid,
 		Offset: offset,
+		Size:   size,
 	}
 
 	k.Lock()

internal/streampb/stream.go

@@ -66,12 +66,12 @@ type Decoder struct {
 
 // Decode takes a proto.Message and unmarshals the next payload in the
 // underlying io.Reader. It returns an EOF when it's done.
-func (d *Decoder) Decode(v proto.Message) error {
+func (d *Decoder) Decode(v proto.Message) (int64, error) {
 	prefixBuf := make([]byte, prefixSize)
 
 	_, err := io.ReadFull(d.r, prefixBuf)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	n := binary.BigEndian.Uint64(prefixBuf)
@@ -82,11 +82,11 @@ func (d *Decoder) Decode(v proto.Message) error {
 	for idx < n {
 		m, err := d.r.Read(buf[idx:n])
 		if err != nil {
-			return errors.Wrap(translateError(err), "failed reading marshaled data")
+			return 0, errors.Wrap(translateError(err), "failed reading marshaled data")
 		}
 		idx += uint64(m)
 	}
-	return proto.Unmarshal(buf[:n], v)
+	return int64(idx + prefixSize), proto.Unmarshal(buf[:n], v)
 }
 
 func translateError(err error) error {