Add docs for bitcask

.drone.yml

@@ -5,18 +5,10 @@ steps:
   - name: build
     image: golang:latest
     commands:
-      - go test -v -short -cover -coverprofile=coverage.txt ./...
+      - go test -v -short -cover -coverprofile=coverage.txt -coverpkg=$(go list) ./...
 
   - name: coverage
     image: plugins/codecov
     settings:
       token:
         from_secret: codecov-token
-
-  - name: notify
-    image: plugins/webhook
-    urls: https://msgbus.mills.io/ci.mills.io
-    when:
-      status:
-        - success
-        - failure

.goreleaser.yml

@@ -3,14 +3,14 @@ builds:
     binary: bitcask
     main: ./cmd/bitcask
     flags: -tags "static_build"
-    ldflags: -w -X .Version={{.Version}} -X .Commit={{.Commit}}
+    ldflags: -w -X github.com/prologic/bitcask/internal.Version={{.Version}} -X github.com/prologic/bitcask/internal.Commit={{.Commit}}
     env:
       - CGO_ENABLED=0
   - 
     binary: bitcaskd
     main: ./cmd/bitcaskd
     flags: -tags "static_build"
-    ldflags: -w -X .Version={{.Version}} -X .Commit={{.Commit}}
+    ldflags: -w -X github.com/prologic/bitcask/internal.Version={{.Version}} -X github.com/prologic/bitcask/internal.Commit={{.Commit}}
     env:
       - CGO_ENABLED=0
 sign:

Makefile

@@ -13,11 +13,11 @@ dev: build
 build: clean generate
 	@go build \
 		-tags "netgo static_build" -installsuffix netgo \
-		-ldflags "-w -X $(shell go list).Version=$(VERSION) -X $(shell go list).Commit=$(COMMIT)" \
+		-ldflags "-w -X $(shell go list)/internal.Version=$(VERSION) -X $(shell go list)/internal.Commit=$(COMMIT)" \
 		./cmd/bitcask/...
 	@go build \
 		-tags "netgo static_build" -installsuffix netgo \
-		-ldflags "-w -X $(shell go list).Version=$(VERSION) -X $(shell go list).Commit=$(COMMIT)" \
+		-ldflags "-w -X $(shell go list)/internal.Version=$(VERSION) -X $(shell go list)/internal.Commit=$(COMMIT)" \
 		./cmd/bitcaskd/...
 
 generate:
@@ -25,6 +25,7 @@ generate:
 
 install: build
 	@go install ./cmd/bitcask/...
+	@go install ./cmd/bitcaskd/...
 
 image:
 	@docker build -t prologic/bitcask .
@@ -39,7 +40,7 @@ bench: build
 	@go test -v -benchmem -bench=. ./...
 
 test: build
-	@go test -v -cover -coverprofile=coverage.txt -covermode=atomic -coverpkg=./... -race ./...
+	@go test -v -cover -coverprofile=coverage.txt -covermode=atomic -coverpkg=$(shell go list) -race ./...
 
 clean:
 	@git clean -f -d -X

README.md

@@ -6,13 +6,13 @@
 [![GoDoc](https://godoc.org/github.com/prologic/bitcask?status.svg)](https://godoc.org/github.com/prologic/bitcask) 
 [![Sourcegraph](https://sourcegraph.com/github.com/prologic/bitcask/-/badge.svg)](https://sourcegraph.com/github.com/prologic/bitcask?badge)
 
-A Bitcask (LSM+WAL) Key/Value Store written in Go.
+A high performance Key/Value store written in [Go](https://golang.org) with a predictable read/write performance and high throughput. Uses a [Bitcask](https://en.wikipedia.org/wiki/Bitcask) on-disk layout (LSM+WAL) similar to [Riak](https://riak.com/).
 
 ## Features
 
-* Embeddable
-* Builtin CLI
-* Builtin Redis-compatible server
+* Embeddable (`import "github.com/prologic/bitcask"`)
+* Builtin CLI (`bitcask`)
+* Builtin Redis-compatible server (`bitcaskd`)
 * Predictable read/write performance
 * Low latecny
 * High throughput (See: [Performance](README.md#Performance)
@@ -34,16 +34,13 @@ $ go get github.com/prologic/bitcask
 ```#!go
 package main
 
-import (
-    "log"
-
-    "github.com/prologic/bitcask"
-)
+import "github.com/prologic/bitcask"
 
 func main() {
     db, _ := bitcask.Open("/tmp/db")
+    defer db.Close()
     db.Set("Hello", []byte("World"))
-    db.Close()
+    val, _ := db.Get("hello")
 }
 ```
 
@@ -97,30 +94,33 @@ Benchmarks run on a 11" Macbook with a 1.4Ghz Intel Core i7:
 ```
 $ make bench
 ...
-BenchmarkGet/128B-4         	  200000	      5780 ns/op	     400 B/op	       5 allocs/op
-BenchmarkGet/256B-4         	  200000	      6138 ns/op	     656 B/op	       5 allocs/op
-BenchmarkGet/512B-4         	  200000	      5967 ns/op	    1200 B/op	       5 allocs/op
-BenchmarkGet/1K-4           	  200000	      6290 ns/op	    2288 B/op	       5 allocs/op
-BenchmarkGet/2K-4           	  200000	      6293 ns/op	    4464 B/op	       5 allocs/op
-BenchmarkGet/4K-4           	  200000	      7673 ns/op	    9072 B/op	       5 allocs/op
-BenchmarkGet/8K-4           	  200000	     10373 ns/op	   17776 B/op	       5 allocs/op
-BenchmarkGet/16K-4          	  100000	     14227 ns/op	   34928 B/op	       5 allocs/op
-BenchmarkGet/32K-4          	  100000	     25953 ns/op	   73840 B/op	       5 allocs/op
-BenchmarkPut/128B-4         	  100000	     17353 ns/op	     680 B/op	       5 allocs/op
-BenchmarkPut/256B-4         	  100000	     18620 ns/op	     808 B/op	       5 allocs/op
-BenchmarkPut/512B-4         	  100000	     19068 ns/op	    1096 B/op	       5 allocs/op
-BenchmarkPut/1K-4           	  100000	     23738 ns/op	    1673 B/op	       5 allocs/op
-BenchmarkPut/2K-4           	   50000	     25118 ns/op	    2826 B/op	       5 allocs/op
-BenchmarkPut/4K-4           	   50000	     44605 ns/op	    5389 B/op	       5 allocs/op
-BenchmarkPut/8K-4           	   30000	     55237 ns/op	   10001 B/op	       5 allocs/op
-BenchmarkPut/16K-4          	   20000	     78966 ns/op	   18972 B/op	       5 allocs/op
-BenchmarkPut/32K-4          	   10000	    116253 ns/op	   41520 B/op	       5 allocs/op
+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
+
+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
+
+BenchmarkScan-4             	 1000000	      1851 ns/op	     493 B/op	      25 allocs/op
 ```
 
 For 128B values:
 
-* ~180,000 reads/sec
-*  ~60,000 writes/sec
+* ~200,000 reads/sec
+* ~130,000 writes/sec
 
 The full benchmark above shows linear performance as you increase key/value sizes.
 

bitcask.go

@@ -2,42 +2,61 @@ package bitcask
 
 import (
 	"errors"
-	"fmt"
+	"hash/crc32"
 	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
-	"sort"
-	"strconv"
 	"strings"
-	"time"
 
 	"github.com/gofrs/flock"
-)
+	"github.com/prologic/trie"
 
-const (
-	DefaultMaxDatafileSize = 1 << 20 // 1MB
+	"github.com/prologic/bitcask/internal"
 )
 
 var (
-	ErrKeyNotFound       = errors.New("error: key not found")
-	ErrCannotAcquireLock = errors.New("error: cannot acquire lock")
+	// ErrKeyNotFound is the error returned when a key is not found
+	ErrKeyNotFound = errors.New("error: key not found")
+
+	// ErrKeyTooLarge is the error returned for a key that exceeds the
+	// maximum allowed key size (configured with WithMaxKeySize).
+	ErrKeyTooLarge = errors.New("error: key too large")
+
+	// ErrValueTooLarge is the error returned for a value that exceeds the
+	// maximum allowed value size (configured with WithMaxValueSize).
+	ErrValueTooLarge = errors.New("error: value too large")
+
+	// ErrChecksumFailed is the error returned if a key/valie retrieved does
+	// not match its CRC checksum
+	ErrChecksumFailed = errors.New("error: checksum failed")
+
+	// ErrDatabaseLocked is the error returned if the database is locked
+	// (typically opened by another process)
+	ErrDatabaseLocked = errors.New("error: database locked")
 )
 
+// Bitcask is a struct that represents a on-disk LSM and WAL data structure
+// and in-memory hash of key/value pairs as per the Bitcask paper and seen
+// in the Riak database.
 type Bitcask struct {
 	*flock.Flock
 
+	config    *config
 	path      string
-	curr      *Datafile
-	keydir    *Keydir
-	datafiles []*Datafile
-
-	maxDatafileSize int64
+	curr      *internal.Datafile
+	keydir    *internal.Keydir
+	datafiles []*internal.Datafile
+	trie      *trie.Trie
 }
 
+// Close closes the database and removes the lock. It is important to call
+// Close() as this is the only wat to cleanup the lock held by the open
+// database.
 func (b *Bitcask) Close() error {
 	defer func() {
 		b.Flock.Unlock()
+		os.Remove(b.Flock.Path())
 	}()
 
 	for _, df := range b.datafiles {
@@ -46,43 +65,68 @@ func (b *Bitcask) Close() error {
 	return b.curr.Close()
 }
 
+// Sync flushes all buffers to disk ensuring all data is written
 func (b *Bitcask) Sync() error {
 	return b.curr.Sync()
 }
 
+// Get retrieves the value of the given key. If the key is not found or an/I/O
+// error occurs a null byte slice is returend along with the error.
 func (b *Bitcask) Get(key string) ([]byte, error) {
-	var df *Datafile
+	var df *internal.Datafile
 
 	item, ok := b.keydir.Get(key)
 	if !ok {
 		return nil, ErrKeyNotFound
 	}
 
-	if item.FileID == b.curr.id {
+	if item.FileID == b.curr.FileID() {
 		df = b.curr
 	} else {
 		df = b.datafiles[item.FileID]
 	}
 
-	e, err := df.ReadAt(item.Index)
+	e, err := df.ReadAt(item.Offset)
 	if err != nil {
 		return nil, err
 	}
 
+	checksum := crc32.ChecksumIEEE(e.Value)
+	if checksum != e.Checksum {
+		return nil, ErrChecksumFailed
+	}
+
 	return e.Value, nil
 }
 
+// Has returns true if the key exists in the database, false otherwise.
+func (b *Bitcask) Has(key string) bool {
+	_, ok := b.keydir.Get(key)
+	return ok
+}
+
+// Put stores the key and value in the database.
 func (b *Bitcask) Put(key string, value []byte) error {
-	index, err := b.put(key, value)
+	if len(key) > b.config.maxKeySize {
+		return ErrKeyTooLarge
+	}
+	if len(value) > b.config.maxValueSize {
+		return ErrValueTooLarge
+	}
+
+	offset, err := b.put(key, value)
 	if err != nil {
 		return err
 	}
 
-	b.keydir.Add(key, b.curr.id, index, time.Now().Unix())
+	item := b.keydir.Add(key, b.curr.FileID(), offset)
+	b.trie.Add(key, item)
 
 	return nil
 }
 
+// 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{})
 	if err != nil {
@@ -90,10 +134,37 @@ func (b *Bitcask) Delete(key string) error {
 	}
 
 	b.keydir.Delete(key)
+	b.trie.Remove(key)
 
 	return nil
 }
 
+// Scan performa a prefix scan of keys matching the given prefix and calling
+// the function `f` with the keys found. If the function returns an error
+// no further keys are processed and the first error returned.
+func (b *Bitcask) Scan(prefix string, f func(key string) error) error {
+	keys := b.trie.PrefixSearch(prefix)
+	for _, key := range keys {
+		if err := f(key); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// Len returns the total number of keys in the database
+func (b *Bitcask) Len() int {
+	return b.keydir.Len()
+}
+
+// Keys returns all keys in the database as a channel of string(s)
+func (b *Bitcask) Keys() chan string {
+	return b.keydir.Keys()
+}
+
+// Fold iterates over all keys in the database calling the function `f` for
+// each key. If the function returns an error, no further keys are processed
+// and the error returned.
 func (b *Bitcask) Fold(f func(key string) error) error {
 	for key := range b.keydir.Keys() {
 		if err := f(key); err != nil {
@@ -109,72 +180,41 @@ func (b *Bitcask) put(key string, value []byte) (int64, error) {
 		return -1, err
 	}
 
-	if size >= b.maxDatafileSize {
+	if size >= int64(b.config.maxDatafileSize) {
 		err := b.curr.Close()
 		if err != nil {
 			return -1, err
 		}
 
-		df, err := NewDatafile(b.path, b.curr.id, true)
+		df, err := internal.NewDatafile(b.path, b.curr.FileID(), true)
+		if err != nil {
+			return -1, err
+		}
+
 		b.datafiles = append(b.datafiles, df)
 
-		id := b.curr.id + 1
-		curr, err := NewDatafile(b.path, id, false)
+		id := b.curr.FileID() + 1
+		curr, err := internal.NewDatafile(b.path, id, false)
 		if err != nil {
 			return -1, err
 		}
 		b.curr = curr
 	}
 
-	e := NewEntry(key, value)
+	e := internal.NewEntry(key, value)
 	return b.curr.Write(e)
 }
 
-func (b *Bitcask) setMaxDatafileSize(size int64) error {
-	b.maxDatafileSize = size
-	return nil
-}
-
-func WithMaxDatafileSize(size int64) func(*Bitcask) error {
-	return func(b *Bitcask) error {
-		return b.setMaxDatafileSize(size)
-	}
-}
-
-func getDatafiles(path string) ([]string, error) {
-	fns, err := filepath.Glob(fmt.Sprintf("%s/*.data", path))
-	if err != nil {
-		return nil, err
-	}
-	sort.Strings(fns)
-	return fns, nil
-}
-
-func parseIds(fns []string) ([]int, error) {
-	var ids []int
-	for _, fn := range fns {
-		fn = filepath.Base(fn)
-		ext := filepath.Ext(fn)
-		if ext != ".data" {
-			continue
-		}
-		id, err := strconv.ParseInt(strings.TrimSuffix(fn, ext), 10, 32)
-		if err != nil {
-			return nil, err
-		}
-		ids = append(ids, int(id))
-	}
-	sort.Ints(ids)
-	return ids, nil
-}
-
+// Merge merges all datafiles in the database creating hint files for faster
+// startup. Old keys are squashed and deleted keys removes. Call this function
+// periodically to reclaim disk space.
 func Merge(path string, force bool) error {
-	fns, err := getDatafiles(path)
+	fns, err := internal.GetDatafiles(path)
 	if err != nil {
 		return err
 	}
 
-	ids, err := parseIds(fns)
+	ids, err := internal.ParseIds(fns)
 	if err != nil {
 		return err
 	}
@@ -204,9 +244,9 @@ func Merge(path string, force bool) error {
 
 		id := ids[i]
 
-		keydir := NewKeydir()
+		keydir := internal.NewKeydir()
 
-		df, err := NewDatafile(path, id, true)
+		df, err := internal.NewDatafile(path, id, true)
 		if err != nil {
 			return err
 		}
@@ -227,10 +267,10 @@ func Merge(path string, force bool) error {
 				continue
 			}
 
-			keydir.Add(e.Key, ids[i], e.Index, e.Timestamp)
+			keydir.Add(e.Key, ids[i], e.Offset)
 		}
 
-		tempdf, err := NewDatafile(temp, id, false)
+		tempdf, err := internal.NewDatafile(temp, id, false)
 		if err != nil {
 			return err
 		}
@@ -238,7 +278,7 @@ func Merge(path string, force bool) error {
 
 		for key := range keydir.Keys() {
 			item, _ := keydir.Get(key)
-			e, err := df.ReadAt(item.Index)
+			e, err := df.ReadAt(item.Offset)
 			if err != nil {
 				return err
 			}
@@ -274,7 +314,10 @@ func Merge(path string, force bool) error {
 	return nil
 }
 
-func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
+// Open opens the database at the given path with optional options.
+// Options can be provided with the `WithXXX` functions that provide
+// configuration options as functions.
+func Open(path string, options ...Option) (*Bitcask, error) {
 	if err := os.MkdirAll(path, 0755); err != nil {
 		return nil, err
 	}
@@ -284,21 +327,23 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 		return nil, err
 	}
 
-	fns, err := getDatafiles(path)
+	fns, err := internal.GetDatafiles(path)
 	if err != nil {
 		return nil, err
 	}
 
-	ids, err := parseIds(fns)
+	ids, err := internal.ParseIds(fns)
 	if err != nil {
 		return nil, err
 	}
 
-	keydir := NewKeydir()
-	var datafiles []*Datafile
+	var datafiles []*internal.Datafile
+
+	keydir := internal.NewKeydir()
+	trie := trie.New()
 
 	for i, fn := range fns {
-		df, err := NewDatafile(path, ids[i], true)
+		df, err := internal.NewDatafile(path, ids[i], true)
 		if err != nil {
 			return nil, err
 		}
@@ -311,14 +356,15 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 			}
 			defer f.Close()
 
-			hint, err := NewKeydirFromBytes(f)
+			hint, err := internal.NewKeydirFromBytes(f)
 			if err != nil {
 				return nil, err
 			}
 
 			for key := range hint.Keys() {
 				item, _ := hint.Get(key)
-				keydir.Add(key, item.FileID, item.Index, item.Timestamp)
+				_ = keydir.Add(key, item.FileID, item.Offset)
+				trie.Add(key, item)
 			}
 		} else {
 			for {
@@ -336,7 +382,8 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 					continue
 				}
 
-				keydir.Add(e.Key, ids[i], e.Index, e.Timestamp)
+				item := keydir.Add(e.Key, ids[i], e.Offset)
+				trie.Add(e.Key, item)
 			}
 		}
 	}
@@ -346,23 +393,23 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 		id = ids[(len(ids) - 1)]
 	}
 
-	curr, err := NewDatafile(path, id, false)
+	curr, err := internal.NewDatafile(path, id, false)
 	if err != nil {
 		return nil, err
 	}
 
 	bitcask := &Bitcask{
 		Flock:     flock.New(filepath.Join(path, "lock")),
+		config:    newDefaultConfig(),
 		path:      path,
 		curr:      curr,
 		keydir:    keydir,
 		datafiles: datafiles,
-
-		maxDatafileSize: DefaultMaxDatafileSize,
+		trie:      trie,
 	}
 
-	for _, option := range options {
-		err = option(bitcask)
+	for _, opt := range options {
+		err = opt(bitcask.config)
 		if err != nil {
 			return nil, err
 		}
@@ -374,7 +421,7 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 	}
 
 	if !locked {
-		return nil, ErrCannotAcquireLock
+		return nil, ErrDatabaseLocked
 	}
 
 	return bitcask, nil

bitcask_test.go

@@ -3,6 +3,8 @@ package bitcask
 import (
 	"fmt"
 	"io/ioutil"
+	"reflect"
+	"sort"
 	"strings"
 	"sync"
 	"testing"
@@ -38,12 +40,48 @@ func TestAll(t *testing.T) {
 		assert.Equal([]byte("bar"), val)
 	})
 
+	t.Run("Len", func(t *testing.T) {
+		assert.Equal(1, db.Len())
+	})
+
+	t.Run("Has", func(t *testing.T) {
+		assert.True(db.Has("foo"))
+	})
+
+	t.Run("Keys", func(t *testing.T) {
+		keys := make([]string, 0)
+		for key := range db.Keys() {
+			keys = append(keys, key)
+		}
+		assert.Equal([]string{"foo"}, keys)
+	})
+
+	t.Run("Fold", func(t *testing.T) {
+		var (
+			keys   []string
+			values [][]byte
+		)
+
+		err := db.Fold(func(key string) error {
+			value, err := db.Get(key)
+			if err != nil {
+				return err
+			}
+			keys = append(keys, key)
+			values = append(values, value)
+			return nil
+		})
+		assert.NoError(err)
+		assert.Equal([]string{"foo"}, keys)
+		assert.Equal([][]byte{[]byte("bar")}, values)
+	})
+
 	t.Run("Delete", func(t *testing.T) {
 		err := db.Delete("foo")
 		assert.NoError(err)
 		_, err = db.Get("foo")
 		assert.Error(err)
-		assert.Equal(err.Error(), "error: key not found")
+		assert.Equal(ErrKeyNotFound, err)
 	})
 
 	t.Run("Sync", func(t *testing.T) {
@@ -90,7 +128,7 @@ func TestDeletedKeys(t *testing.T) {
 			assert.NoError(err)
 			_, err = db.Get("foo")
 			assert.Error(err)
-			assert.Equal("error: key not found", err.Error())
+			assert.Equal(ErrKeyNotFound, err)
 		})
 
 		t.Run("Sync", func(t *testing.T) {
@@ -118,7 +156,7 @@ func TestDeletedKeys(t *testing.T) {
 		t.Run("Get", func(t *testing.T) {
 			_, err = db.Get("foo")
 			assert.Error(err)
-			assert.Equal("error: key not found", err.Error())
+			assert.Equal(ErrKeyNotFound, err)
 		})
 
 		t.Run("Close", func(t *testing.T) {
@@ -128,7 +166,51 @@ func TestDeletedKeys(t *testing.T) {
 	})
 }
 
-func TestMerge(t *testing.T) {
+func TestMaxKeySize(t *testing.T) {
+	assert := assert.New(t)
+
+	testdir, err := ioutil.TempDir("", "bitcask")
+	assert.NoError(err)
+
+	var db *Bitcask
+
+	t.Run("Open", func(t *testing.T) {
+		db, err = Open(testdir, WithMaxKeySize(16))
+		assert.NoError(err)
+	})
+
+	t.Run("Put", func(t *testing.T) {
+		key := strings.Repeat(" ", 17)
+		value := []byte("foobar")
+		err = db.Put(key, value)
+		assert.Error(err)
+		assert.Equal(ErrKeyTooLarge, err)
+	})
+}
+
+func TestMaxValueSize(t *testing.T) {
+	assert := assert.New(t)
+
+	testdir, err := ioutil.TempDir("", "bitcask")
+	assert.NoError(err)
+
+	var db *Bitcask
+
+	t.Run("Open", func(t *testing.T) {
+		db, err = Open(testdir, WithMaxValueSize(16))
+		assert.NoError(err)
+	})
+
+	t.Run("Put", func(t *testing.T) {
+		key := "foo"
+		value := []byte(strings.Repeat(" ", 17))
+		err = db.Put(key, value)
+		assert.Error(err)
+		assert.Equal(ErrValueTooLarge, err)
+	})
+}
+
+func TestOpenMerge(t *testing.T) {
 	assert := assert.New(t)
 
 	testdir, err := ioutil.TempDir("", "bitcask")
@@ -141,7 +223,7 @@ func TestMerge(t *testing.T) {
 		)
 
 		t.Run("Open", func(t *testing.T) {
-			db, err = Open(testdir, WithMaxDatafileSize(1024))
+			db, err = Open(testdir, WithMaxDatafileSize(32))
 			assert.NoError(err)
 		})
 
@@ -199,6 +281,77 @@ func TestMerge(t *testing.T) {
 	})
 }
 
+func TestMergeOpen(t *testing.T) {
+	var (
+		db  *Bitcask
+		err error
+	)
+
+	assert := assert.New(t)
+
+	testdir, err := ioutil.TempDir("", "bitcask")
+	assert.NoError(err)
+
+	t.Run("Setup", func(t *testing.T) {
+		t.Run("Open", func(t *testing.T) {
+			db, err = Open(testdir, WithMaxDatafileSize(32))
+			assert.NoError(err)
+		})
+
+		t.Run("Put", func(t *testing.T) {
+			for i := 0; i < 1024; i++ {
+				err = db.Put(string(i), []byte(strings.Repeat(" ", 1024)))
+				assert.NoError(err)
+			}
+		})
+
+		t.Run("Get", func(t *testing.T) {
+			for i := 0; i < 32; i++ {
+				err = db.Put(string(i), []byte(strings.Repeat(" ", 1024)))
+				assert.NoError(err)
+				val, err := db.Get(string(i))
+				assert.NoError(err)
+				assert.Equal([]byte(strings.Repeat(" ", 1024)), val)
+			}
+		})
+
+		t.Run("Sync", func(t *testing.T) {
+			err = db.Sync()
+			assert.NoError(err)
+		})
+
+		t.Run("Close", func(t *testing.T) {
+			err = db.Close()
+			assert.NoError(err)
+		})
+	})
+
+	t.Run("Merge", func(t *testing.T) {
+		t.Run("Merge", func(t *testing.T) {
+			err = Merge(testdir, true)
+			assert.NoError(err)
+		})
+
+		t.Run("Open", func(t *testing.T) {
+			db, err = Open(testdir)
+			assert.NoError(err)
+		})
+
+		t.Run("Get", func(t *testing.T) {
+			for i := 0; i < 32; i++ {
+				val, err := db.Get(string(i))
+				assert.NoError(err)
+				assert.Equal([]byte(strings.Repeat(" ", 1024)), val)
+			}
+		})
+
+		t.Run("Close", func(t *testing.T) {
+			err = db.Close()
+			assert.NoError(err)
+		})
+	})
+}
+
 func TestConcurrent(t *testing.T) {
 	var (
 		db  *Bitcask
@@ -241,10 +394,9 @@ func TestConcurrent(t *testing.T) {
 			wg := &sync.WaitGroup{}
 
 			go f(wg, 2)
-			wg.Add(1)
-
 			go f(wg, 3)
-			wg.Add(1)
+			go f(wg, 5)
+			wg.Add(3)
 
 			wg.Wait()
 		})
@@ -264,10 +416,9 @@ func TestConcurrent(t *testing.T) {
 			wg := &sync.WaitGroup{}
 
 			go f(wg, 100)
-			wg.Add(1)
-
 			go f(wg, 100)
-			wg.Add(1)
+			go f(wg, 100)
+			wg.Add(3)
 
 			wg.Wait()
 		})
@@ -279,6 +430,58 @@ func TestConcurrent(t *testing.T) {
 	})
 }
 
+func TestScan(t *testing.T) {
+	assert := assert.New(t)
+
+	testdir, err := ioutil.TempDir("", "bitcask")
+	assert.NoError(err)
+
+	var db *Bitcask
+
+	t.Run("Setup", func(t *testing.T) {
+		t.Run("Open", func(t *testing.T) {
+			db, err = Open(testdir)
+			assert.NoError(err)
+		})
+
+		t.Run("Put", func(t *testing.T) {
+			var items = map[string][]byte{
+				"1":     []byte("1"),
+				"2":     []byte("2"),
+				"3":     []byte("3"),
+				"food":  []byte("pizza"),
+				"foo":   []byte("foo"),
+				"fooz":  []byte("fooz ball"),
+				"hello": []byte("world"),
+			}
+			for k, v := range items {
+				err = db.Put(k, v)
+				assert.NoError(err)
+			}
+		})
+	})
+
+	t.Run("Scan", func(t *testing.T) {
+		var (
+			vals     []string
+			expected = []string{
+				"foo",
+				"fooz ball",
+				"pizza",
+			}
+		)
+
+		err = db.Scan("fo", func(key string) error {
+			val, err := db.Get(key)
+			assert.NoError(err)
+			vals = append(vals, string(val))
+			return nil
+		})
+		sort.Strings(vals)
+		assert.Equal(expected, vals)
+	})
+}
+
 func TestLocking(t *testing.T) {
 	assert := assert.New(t)
 
@@ -291,7 +494,7 @@ func TestLocking(t *testing.T) {
 
 	_, err = Open(testdir)
 	assert.Error(err)
-	assert.Equal("error: cannot acquire lock", err.Error())
+	assert.Equal(ErrDatabaseLocked, err)
 }
 
 type benchmarkTestCase struct {
@@ -385,3 +588,47 @@ func BenchmarkPut(b *testing.B) {
 		})
 	}
 }
+
+func BenchmarkScan(b *testing.B) {
+	testdir, err := ioutil.TempDir("", "bitcask")
+	if err != nil {
+		b.Fatal(err)
+	}
+
+	db, err := Open(testdir)
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer db.Close()
+
+	var items = map[string][]byte{
+		"1":     []byte("1"),
+		"2":     []byte("2"),
+		"3":     []byte("3"),
+		"food":  []byte("pizza"),
+		"foo":   []byte("foo"),
+		"fooz":  []byte("fooz ball"),
+		"hello": []byte("world"),
+	}
+	for k, v := range items {
+		err := db.Put(k, v)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+
+	var expected = []string{"foo", "food", "fooz"}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		var keys []string
+		err = db.Scan("fo", func(key string) error {
+			keys = append(keys, key)
+			return nil
+		})
+		sort.Strings(keys)
+		if !reflect.DeepEqual(expected, keys) {
+			b.Fatal(fmt.Errorf("expected keys=#%v got=%#v", expected, keys))
+		}
+	}
+}

cmd/bitcask/del.go

@@ -35,6 +35,7 @@ func del(path, key string) int {
 		log.WithError(err).Error("error opening database")
 		return 1
 	}
+	defer db.Close()
 
 	err = db.Delete(key)
 	if err != nil {

cmd/bitcask/get.go

@@ -36,6 +36,7 @@ func get(path, key string) int {
 		log.WithError(err).Error("error opening database")
 		return 1
 	}
+	defer db.Close()
 
 	value, err := db.Get(key)
 	if err != nil {

cmd/bitcask/keys.go

@@ -34,6 +34,7 @@ func keys(path string) int {
 		log.WithError(err).Error("error opening database")
 		return 1
 	}
+	defer db.Close()
 
 	err = db.Fold(func(key string) error {
 		fmt.Printf("%s\n", key)

cmd/bitcask/root.go

@@ -8,13 +8,13 @@ import (
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
-	"github.com/prologic/bitcask"
+	"github.com/prologic/bitcask/internal"
 )
 
 // RootCmd represents the base command when called without any subcommands
 var RootCmd = &cobra.Command{
 	Use:     "bitcask",
-	Version: bitcask.FullVersion(),
+	Version: internal.FullVersion(),
 	Short:   "Command-line tools for bitcask",
 	Long: `This is the command-line tool to interact with a bitcask database.
 

cmd/bitcask/scan.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/prologic/bitcask"
+)
+
+var scanCmd = &cobra.Command{
+	Use:     "scan <prefix>",
+	Aliases: []string{"search", "find"},
+	Short:   "Perform a prefis scan for keys",
+	Long: `This performa a prefix scan for keys  starting with the given
+prefix. This uses a Trie to search for matching keys and returns all matched
+keys.`,
+	Args: cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		path := viper.GetString("path")
+
+		prefix := args[0]
+
+		os.Exit(scan(path, prefix))
+	},
+}
+
+func init() {
+	RootCmd.AddCommand(scanCmd)
+}
+
+func scan(path, prefix string) int {
+	db, err := bitcask.Open(path)
+	if err != nil {
+		log.WithError(err).Error("error opening database")
+		return 1
+	}
+	defer db.Close()
+
+	err = db.Scan(prefix, func(key string) error {
+		value, err := db.Get(key)
+		if err != nil {
+			log.WithError(err).Error("error reading key")
+			return err
+		}
+
+		fmt.Printf("%s\n", string(value))
+		log.WithField("key", key).WithField("value", value).Debug("key/value")
+		return nil
+	})
+	if err != nil {
+		log.WithError(err).Error("error scanning keys")
+		return 1
+	}
+
+	return 0
+}

cmd/bitcask/set.go

@@ -47,6 +47,7 @@ func set(path, key string, value io.Reader) int {
 		log.WithError(err).Error("error opening database")
 		return 1
 	}
+	defer db.Close()
 
 	data, err := ioutil.ReadAll(value)
 	if err != nil {

cmd/bitcaskd/main.go

@@ -10,13 +10,14 @@ import (
 	"github.com/tidwall/redcon"
 
 	"github.com/prologic/bitcask"
+	"github.com/prologic/bitcask/internal"
 )
 
 var (
 	bind            string
 	debug           bool
 	version         bool
-	maxDatafileSize int64
+	maxDatafileSize int
 )
 
 func init() {
@@ -30,7 +31,7 @@ func init() {
 
 	flag.StringVarP(&bind, "bind", "b", ":6379", "interface and port to bind to")
 
-	flag.Int64Var(&maxDatafileSize, "max-datafile-size", 1<<20, "maximum datafile size in bytes")
+	flag.IntVar(&maxDatafileSize, "max-datafile-size", 1<<20, "maximum datafile size in bytes")
 }
 
 func main() {
@@ -43,7 +44,7 @@ func main() {
 	}
 
 	if version {
-		fmt.Printf("bitcaskd version %s", bitcask.FullVersion())
+		fmt.Printf("bitcaskd version %s", internal.FullVersion())
 		os.Exit(0)
 	}
 
@@ -60,7 +61,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	log.WithField("bind", bind).WithField("path", path).Infof("starting bitcaskd v%s", bitcask.FullVersion())
+	log.WithField("bind", bind).WithField("path", path).Infof("starting bitcaskd v%s", internal.FullVersion())
 
 	err = redcon.ListenAndServe(bind,
 		func(conn redcon.Conn, cmd redcon.Command) {
@@ -95,6 +96,22 @@ func main() {
 				} else {
 					conn.WriteBulk(value)
 				}
+			case "keys":
+				conn.WriteArray(db.Len())
+				for key := range db.Keys() {
+					conn.WriteBulk([]byte(key))
+				}
+			case "exists":
+				if len(cmd.Args) != 2 {
+					conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
+					return
+				}
+				key := string(cmd.Args[1])
+				if db.Has(key) {
+					conn.WriteInt(1)
+				} else {
+					conn.WriteInt(0)
+				}
 			case "del":
 				if len(cmd.Args) != 2 {
 					conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")

entry.go

@@ -1,17 +0,0 @@
-package bitcask
-
-import (
-	"hash/crc32"
-
-	pb "github.com/prologic/bitcask/proto"
-)
-
-func NewEntry(key string, value []byte) pb.Entry {
-	crc := crc32.ChecksumIEEE(value)
-
-	return pb.Entry{
-		CRC:   crc,
-		Key:   key,
-		Value: value,
-	}
-}

go.mod

@@ -1,20 +1,18 @@
 module github.com/prologic/bitcask
 
 require (
+	github.com/BurntSushi/toml v0.3.1 // indirect
 	github.com/gofrs/flock v0.7.1
 	github.com/gogo/protobuf v1.2.1
 	github.com/golang/protobuf v1.2.0
-	github.com/gorilla/websocket v1.4.0 // indirect
-	github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7 // indirect
-	github.com/mitchellh/go-homedir v1.1.0
+	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/pkg/errors v0.8.1
-	github.com/prologic/msgbus v0.1.1
-	github.com/prometheus/client_golang v0.9.2 // indirect
+	github.com/prologic/trie v0.0.0-20190316011403-395e39dac705
 	github.com/sirupsen/logrus v1.3.0
 	github.com/spf13/cobra v0.0.3
 	github.com/spf13/pflag v1.0.3
 	github.com/spf13/viper v1.3.1
 	github.com/stretchr/testify v1.3.0
 	github.com/tidwall/redcon v0.9.0
-	gopkg.in/vmihailenco/msgpack.v2 v2.9.1
+	golang.org/x/sync v0.0.0-20181108010431-42b317875d0f // indirect
 )

go.sum

@@ -1,6 +1,6 @@
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
-github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973 h1:xJ4a3vCFaGF/jqvzLMYoU8P317H5OQ+Via4RmuPwCS0=
-github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -16,21 +16,16 @@ github.com/gogo/protobuf v1.2.1 h1:/s5zKNz0uPFCZ5hddgPdo2TK2TVrUNMn0OOX8/aZMTE=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
 github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
-github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
-github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7 h1:K//n/AqR5HjG3qxbrBCL4vJPW0MVFSs9CPK1OOJdRME=
-github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7/go.mod h1:2iMrUgbbvHEiQClaW2NsSzMyGHqN+rDFqY705q49KG0=
+github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
+github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/magiconair/properties v1.8.0 h1:LLgXmsheXeRoUOBOjtwPQCWIYqM/LU1ayDtDePerRcY=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
-github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
-github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
-github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
-github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2 h1:fmNYVwqnSfB9mZU6OS2O6GsXM+wcskZDuKQzvN1EDeE=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181zc=
@@ -39,15 +34,8 @@ github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prologic/msgbus v0.1.1/go.mod h1:B3Qu4/U2FP08x93jUzp9E8bl155+cIgDH2DUGRK6OZk=
-github.com/prometheus/client_golang v0.9.2 h1:awm861/B8OKDd2I/6o1dy3ra4BamzKhYOiGItCeZ740=
-github.com/prometheus/client_golang v0.9.2/go.mod h1:OsXs2jCmiKlQ1lTBmv21f2mNfw4xf/QclQDMrYNZzcM=
-github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910 h1:idejC8f05m9MGOsuEi1ATq9shN03HrxNkD/luQvxCv8=
-github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
-github.com/prometheus/common v0.0.0-20181126121408-4724e9255275 h1:PnBWHBf+6L0jOqq0gIVUe6Yk0/QMZ640k6NvkxcBf+8=
-github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
-github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a h1:9a8MnZMP0X2nLJdBg+pBmGgkJlSaKC2KaQmTCk1XDtE=
-github.com/prometheus/procfs v0.0.0-20181204211112-1dc9a6cbc91a/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
+github.com/prologic/trie v0.0.0-20190316011403-395e39dac705 h1:2J+cSlAeECj0lfMKSmM7n5OlIio+yLovaKLZJzwLc6U=
+github.com/prologic/trie v0.0.0-20190316011403-395e39dac705/go.mod h1:LFuDmpHJGmciXd8Rl5YMhVlLMps9gz2GtYLzwxrFhzs=
 github.com/sirupsen/logrus v1.3.0 h1:hI/7Q+DtNZ2kINb6qt/lS+IyXnHQe9e90POfeewL/ME=
 github.com/sirupsen/logrus v1.3.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
@@ -74,7 +62,7 @@ github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9 h1:mKdxBk7AujPs8kU4m80U72y/zjbZ3UcXC7dClwKbUI0=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
-golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/sync v0.0.0-20181108010431-42b317875d0f h1:Bl/8QSvNqXvPGPGXa2z5xUTmV7VDcZyvRZ+QQXkXTZQ=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a h1:1n5lsVfiQW3yfsRGu98756EH1YthsFqr/5mxHduZW2A=
@@ -82,8 +70,7 @@ golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5h
 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=
+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=
-gopkg.in/vmihailenco/msgpack.v2 v2.9.1 h1:kb0VV7NuIojvRfzwslQeP3yArBqJHW9tOl4t38VS1jM=
-gopkg.in/vmihailenco/msgpack.v2 v2.9.1/go.mod h1:/3Dn1Npt9+MYyLpYYXjInO/5jvMLamn+AEGwNEOatn8=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

internal/datafile.go

@@ -1,15 +1,15 @@
-package bitcask
+package internal
 
 import (
-	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
 	"sync"
-	"time"
 
-	pb "github.com/prologic/bitcask/proto"
-	"github.com/prologic/bitcask/streampb"
+	"github.com/pkg/errors"
+
+	pb "github.com/prologic/bitcask/internal/proto"
+	"github.com/prologic/bitcask/internal/streampb"
 )
 
 const (
@@ -23,11 +23,12 @@ var (
 type Datafile struct {
 	sync.RWMutex
 
-	id  int
-	r   *os.File
-	w   *os.File
-	dec *streampb.Decoder
-	enc *streampb.Encoder
+	id     int
+	r      *os.File
+	w      *os.File
+	offset int64
+	dec    *streampb.Decoder
+	enc    *streampb.Encoder
 }
 
 func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
@@ -50,19 +51,30 @@ func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
 	if err != nil {
 		return nil, err
 	}
+	stat, err := r.Stat()
+	if err != nil {
+		return nil, errors.Wrap(err, "error calling Stat()")
+	}
+
+	offset := stat.Size()
 
 	dec := streampb.NewDecoder(r)
 	enc := streampb.NewEncoder(w)
 
 	return &Datafile{
-		id:  id,
-		r:   r,
-		w:   w,
-		dec: dec,
-		enc: enc,
+		id:     id,
+		r:      r,
+		w:      w,
+		offset: offset,
+		dec:    dec,
+		enc:    enc,
 	}, nil
 }
 
+func (df *Datafile) FileID() int {
+	return df.id
+}
+
 func (df *Datafile) Name() string {
 	return df.r.Name()
 }
@@ -87,22 +99,9 @@ func (df *Datafile) Sync() error {
 }
 
 func (df *Datafile) Size() (int64, error) {
-	var (
-		stat os.FileInfo
-		err  error
-	)
-
-	if df.w == nil {
-		stat, err = df.r.Stat()
-	} else {
-		stat, err = df.w.Stat()
-	}
-
-	if err != nil {
-		return -1, err
-	}
-
-	return stat.Size(), nil
+	df.RLock()
+	defer df.RUnlock()
+	return df.offset, nil
 }
 
 func (df *Datafile) Read() (e pb.Entry, err error) {
@@ -129,23 +128,16 @@ func (df *Datafile) Write(e pb.Entry) (int64, error) {
 		return -1, ErrReadonly
 	}
 
-	stat, err := df.w.Stat()
-	if err != nil {
-		return -1, err
-	}
-
-	index := stat.Size()
-
-	e.Index = index
-	e.Timestamp = time.Now().Unix()
-
 	df.Lock()
-	err = df.enc.Encode(&e)
-	df.Unlock()
+	defer df.Unlock()
 
+	e.Offset = df.offset
+
+	n, err := df.enc.Encode(&e)
 	if err != nil {
 		return -1, err
 	}
+	df.offset += n
 
-	return index, nil
+	return e.Offset, nil
 }

internal/entry.go

@@ -0,0 +1,17 @@
+package internal
+
+import (
+	"hash/crc32"
+
+	pb "github.com/prologic/bitcask/internal/proto"
+)
+
+func NewEntry(key string, value []byte) pb.Entry {
+	checksum := crc32.ChecksumIEEE(value)
+
+	return pb.Entry{
+		Checksum: checksum,
+		Key:      key,
+		Value:    value,
+	}
+}

internal/keydir.go

@@ -1,4 +1,4 @@
-package bitcask
+package internal
 
 import (
 	"bytes"
@@ -9,9 +9,8 @@ import (
 )
 
 type Item struct {
-	FileID    int
-	Index     int64
-	Timestamp int64
+	FileID int
+	Offset int64
 }
 
 type Keydir struct {
@@ -25,15 +24,17 @@ func NewKeydir() *Keydir {
 	}
 }
 
-func (k *Keydir) Add(key string, fileid int, index, timestamp int64) {
-	k.Lock()
-	defer k.Unlock()
-
-	k.kv[key] = Item{
-		FileID:    fileid,
-		Index:     index,
-		Timestamp: timestamp,
+func (k *Keydir) Add(key string, fileid int, offset int64) Item {
+	item := Item{
+		FileID: fileid,
+		Offset: offset,
 	}
+
+	k.Lock()
+	k.kv[key] = item
+	k.Unlock()
+
+	return item
 }
 
 func (k *Keydir) Get(key string) (Item, bool) {
@@ -51,11 +52,17 @@ func (k *Keydir) Delete(key string) {
 	delete(k.kv, key)
 }
 
+func (k *Keydir) Len() int {
+	return len(k.kv)
+}
+
 func (k *Keydir) Keys() chan string {
 	ch := make(chan string)
 	go func() {
-		for k := range k.kv {
-			ch <- k
+		k.RLock()
+		defer k.RUnlock()
+		for key := range k.kv {
+			ch <- key
 		}
 		close(ch)
 	}()

internal/proto/entry.pb.go

@@ -19,11 +19,10 @@ var _ = math.Inf
 const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
 
 type Entry struct {
-	CRC                  uint32   `protobuf:"varint,1,opt,name=CRC,proto3" json:"CRC,omitempty"`
+	Checksum             uint32   `protobuf:"varint,1,opt,name=Checksum,proto3" json:"Checksum,omitempty"`
 	Key                  string   `protobuf:"bytes,2,opt,name=Key,proto3" json:"Key,omitempty"`
-	Index                int64    `protobuf:"varint,3,opt,name=Index,proto3" json:"Index,omitempty"`
+	Offset               int64    `protobuf:"varint,3,opt,name=Offset,proto3" json:"Offset,omitempty"`
 	Value                []byte   `protobuf:"bytes,4,opt,name=Value,proto3" json:"Value,omitempty"`
-	Timestamp            int64    `protobuf:"varint,5,opt,name=Timestamp,proto3" json:"Timestamp,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -33,7 +32,7 @@ func (m *Entry) Reset()         { *m = Entry{} }
 func (m *Entry) String() string { return proto.CompactTextString(m) }
 func (*Entry) ProtoMessage()    {}
 func (*Entry) Descriptor() ([]byte, []int) {
-	return fileDescriptor_entry_4f5906245d08394f, []int{0}
+	return fileDescriptor_entry_3e91842c99935ae2, []int{0}
 }
 func (m *Entry) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Entry.Unmarshal(m, b)
@@ -53,9 +52,9 @@ func (m *Entry) XXX_DiscardUnknown() {
 
 var xxx_messageInfo_Entry proto.InternalMessageInfo
 
-func (m *Entry) GetCRC() uint32 {
+func (m *Entry) GetChecksum() uint32 {
 	if m != nil {
-		return m.CRC
+		return m.Checksum
 	}
 	return 0
 }
@@ -67,9 +66,9 @@ func (m *Entry) GetKey() string {
 	return ""
 }
 
-func (m *Entry) GetIndex() int64 {
+func (m *Entry) GetOffset() int64 {
 	if m != nil {
-		return m.Index
+		return m.Offset
 	}
 	return 0
 }
@@ -81,28 +80,20 @@ func (m *Entry) GetValue() []byte {
 	return nil
 }
 
-func (m *Entry) GetTimestamp() int64 {
-	if m != nil {
-		return m.Timestamp
-	}
-	return 0
-}
-
 func init() {
 	proto.RegisterType((*Entry)(nil), "proto.Entry")
 }
 
-func init() { proto.RegisterFile("entry.proto", fileDescriptor_entry_4f5906245d08394f) }
+func init() { proto.RegisterFile("entry.proto", fileDescriptor_entry_3e91842c99935ae2) }
 
-var fileDescriptor_entry_4f5906245d08394f = []byte{
-	// 134 bytes of a gzipped FileDescriptorProto
+var fileDescriptor_entry_3e91842c99935ae2 = []byte{
+	// 126 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4e, 0xcd, 0x2b, 0x29,
-	0xaa, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x05, 0x53, 0x4a, 0xa5, 0x5c, 0xac, 0xae,
-	0x20, 0x51, 0x21, 0x01, 0x2e, 0x66, 0xe7, 0x20, 0x67, 0x09, 0x46, 0x05, 0x46, 0x0d, 0xde, 0x20,
-	0x10, 0x13, 0x24, 0xe2, 0x9d, 0x5a, 0x29, 0xc1, 0xa4, 0xc0, 0xa8, 0xc1, 0x19, 0x04, 0x62, 0x0a,
-	0x89, 0x70, 0xb1, 0x7a, 0xe6, 0xa5, 0xa4, 0x56, 0x48, 0x30, 0x2b, 0x30, 0x6a, 0x30, 0x07, 0x41,
-	0x38, 0x20, 0xd1, 0xb0, 0xc4, 0x9c, 0xd2, 0x54, 0x09, 0x16, 0x05, 0x46, 0x0d, 0x9e, 0x20, 0x08,
-	0x47, 0x48, 0x86, 0x8b, 0x33, 0x24, 0x33, 0x37, 0xb5, 0xb8, 0x24, 0x31, 0xb7, 0x40, 0x82, 0x15,
-	0xac, 0x1e, 0x21, 0x90, 0xc4, 0x06, 0xb6, 0xdd, 0x18, 0x10, 0x00, 0x00, 0xff, 0xff, 0x07, 0x99,
-	0x47, 0xb9, 0x93, 0x00, 0x00, 0x00,
+	0xaa, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x05, 0x53, 0x4a, 0xc9, 0x5c, 0xac, 0xae,
+	0x20, 0x51, 0x21, 0x29, 0x2e, 0x0e, 0xe7, 0x8c, 0xd4, 0xe4, 0xec, 0xe2, 0xd2, 0x5c, 0x09, 0x46,
+	0x05, 0x46, 0x0d, 0xde, 0x20, 0x38, 0x5f, 0x48, 0x80, 0x8b, 0xd9, 0x3b, 0xb5, 0x52, 0x82, 0x49,
+	0x81, 0x51, 0x83, 0x33, 0x08, 0xc4, 0x14, 0x12, 0xe3, 0x62, 0xf3, 0x4f, 0x4b, 0x2b, 0x4e, 0x2d,
+	0x91, 0x60, 0x56, 0x60, 0xd4, 0x60, 0x0e, 0x82, 0xf2, 0x84, 0x44, 0xb8, 0x58, 0xc3, 0x12, 0x73,
+	0x4a, 0x53, 0x25, 0x58, 0x14, 0x18, 0x35, 0x78, 0x82, 0x20, 0x9c, 0x24, 0x36, 0xb0, 0x5d, 0xc6,
+	0x80, 0x00, 0x00, 0x00, 0xff, 0xff, 0x76, 0xd2, 0x3e, 0x83, 0x81, 0x00, 0x00, 0x00,
 }

internal/proto/entry.proto

@@ -3,9 +3,8 @@ syntax = "proto3";
 package proto;
 
 message Entry {
-	uint32 CRC = 1;
+	uint32 Checksum = 1;
 	string Key = 2;
-	int64 Index = 3;
+	int64 Offset = 3;
 	bytes Value = 4;
-	int64 Timestamp = 5;
 }

internal/streampb/stream.go

@@ -1,6 +1,7 @@
 package streampb
 
 import (
+	"bufio"
 	"encoding/binary"
 	"io"
 
@@ -16,32 +17,40 @@ const (
 
 // NewEncoder creates a streaming protobuf encoder.
 func NewEncoder(w io.Writer) *Encoder {
-	return &Encoder{w}
+	return &Encoder{w: bufio.NewWriter(w)}
 }
 
 // Encoder wraps an underlying io.Writer and allows you to stream
 // proto encodings on it.
 type Encoder struct {
-	w io.Writer
+	w *bufio.Writer
 }
 
 // Encode takes any proto.Message and streams it to the underlying writer.
 // Messages are framed with a length prefix.
-func (e *Encoder) Encode(msg proto.Message) error {
+func (e *Encoder) Encode(msg proto.Message) (int64, error) {
 	prefixBuf := make([]byte, prefixSize)
 
 	buf, err := proto.Marshal(msg)
 	if err != nil {
-		return err
+		return 0, err
 	}
 	binary.BigEndian.PutUint64(prefixBuf, uint64(len(buf)))
 
 	if _, err := e.w.Write(prefixBuf); err != nil {
-		return errors.Wrap(err, "failed writing length prefix")
+		return 0, errors.Wrap(err, "failed writing length prefix")
 	}
 
-	_, err = e.w.Write(buf)
-	return errors.Wrap(err, "failed writing marshaled data")
+	n, err := e.w.Write(buf)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed writing marshaled data")
+	}
+
+	if err = e.w.Flush(); err != nil {
+		return 0, errors.Wrap(err, "failed flushing data")
+	}
+
+	return int64(n + prefixSize), nil
 }
 
 // NewDecoder creates a streaming protobuf decoder.

internal/utils.go

@@ -0,0 +1,36 @@
+package internal
+
+import (
+	"fmt"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+func GetDatafiles(path string) ([]string, error) {
+	fns, err := filepath.Glob(fmt.Sprintf("%s/*.data", path))
+	if err != nil {
+		return nil, err
+	}
+	sort.Strings(fns)
+	return fns, nil
+}
+
+func ParseIds(fns []string) ([]int, error) {
+	var ids []int
+	for _, fn := range fns {
+		fn = filepath.Base(fn)
+		ext := filepath.Ext(fn)
+		if ext != ".data" {
+			continue
+		}
+		id, err := strconv.ParseInt(strings.TrimSuffix(fn, ext), 10, 32)
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, int(id))
+	}
+	sort.Ints(ids)
+	return ids, nil
+}

internal/version.go

@@ -1,4 +1,4 @@
-package bitcask
+package internal
 
 import (
 	"fmt"

internal/version_test.go

@@ -1,4 +1,4 @@
-package bitcask
+package internal
 
 import (
 	"fmt"

options.go

@@ -0,0 +1,53 @@
+package bitcask
+
+const (
+	// DefaultMaxDatafileSize is the default maximum datafile size in bytes
+	DefaultMaxDatafileSize = 1 << 20 // 1MB
+
+	// DefaultMaxKeySize is the default maximum key size in bytes
+	DefaultMaxKeySize = 64 // 64 bytes
+
+	// DefaultMaxValueSize is the default value size in bytes
+	DefaultMaxValueSize = 1 << 16 // 65KB
+)
+
+// Option is a function that takes a config struct and modifies it
+type Option func(*config) error
+
+type config struct {
+	maxDatafileSize int
+	maxKeySize      int
+	maxValueSize    int
+}
+
+func newDefaultConfig() *config {
+	return &config{
+		maxDatafileSize: DefaultMaxDatafileSize,
+		maxKeySize:      DefaultMaxKeySize,
+		maxValueSize:    DefaultMaxValueSize,
+	}
+}
+
+// WithMaxDatafileSize sets the maximum datafile size option
+func WithMaxDatafileSize(size int) Option {
+	return func(cfg *config) error {
+		cfg.maxDatafileSize = size
+		return nil
+	}
+}
+
+// WithMaxKeySize sets the maximum key size option
+func WithMaxKeySize(size int) Option {
+	return func(cfg *config) error {
+		cfg.maxKeySize = size
+		return nil
+	}
+}
+
+// WithMaxValueSize sets the maximum value size option
+func WithMaxValueSize(size int) Option {
+	return func(cfg *config) error {
+		cfg.maxValueSize = size
+		return nil
+	}
+}