Increased read performance by ~3-4x by removing another unnecessary I/O operation (Seek)

internal/datafile.go

@@ -1,12 +1,14 @@
 package internal
 
 import (
+	"bytes"
 	"fmt"
 	"os"
 	"path/filepath"
 	"sync"
 
 	"github.com/pkg/errors"
+	log "github.com/sirupsen/logrus"
 
 	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 {
@@ -104,28 +107,40 @@ 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) {
+	log.WithField("index", index).WithField("size", size).Debug("ReadAt")
+
+	b := make([]byte, size)
+	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, ErrReadonly
+		return -1, 0, ErrReadonly
 	}
 
 	df.Lock()
@@ -135,9 +150,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 {