Improved read/write performance by another ~2x by not calling Stat() on every read/write

.gitignore

@@ -3,5 +3,6 @@
 
 /coverage.txt
 /bitcask
+/bitcaskd
 /tmp
 /dist

.goreleaser.yml

@@ -1,10 +1,18 @@
 builds:
   - 
+    binary: bitcask
     main: ./cmd/bitcask
     flags: -tags "static_build"
     ldflags: -w -X .Version={{.Version}} -X .Commit={{.Commit}}
     env:
       - CGO_ENABLED=0
+  - 
+    binary: bitcaskd
+    main: ./cmd/bitcaskd
+    flags: -tags "static_build"
+    ldflags: -w -X .Version={{.Version}} -X .Commit={{.Commit}}
+    env:
+      - CGO_ENABLED=0
 sign:
   artifacts: checksum
 archive:

Makefile

@@ -1,18 +1,24 @@
 .PHONY: dev build generate install image release profile bench test clean
 
 CGO_ENABLED=0
+VERSION=$(shell git describe --abbrev=0 --tags)
 COMMIT=$(shell git rev-parse --short HEAD)
 
 all: dev
 
 dev: build
 	@./bitcask --version
+	@./bitcaskd --version
 
 build: clean generate
 	@go build \
 		-tags "netgo static_build" -installsuffix netgo \
-		-ldflags "-w -X $(shell go list)/.Commit=$(COMMIT)" \
+		-ldflags "-w -X $(shell go list).Version=$(VERSION) -X $(shell go list).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)" \
+		./cmd/bitcaskd/...
 
 generate:
 	@go generate $(shell go list)/...

README.md

@@ -12,6 +12,10 @@ A Bitcask (LSM+WAL) Key/Value Store written in Go.
 
 * Embeddable
 * Builtin CLI
+* Builtin Redis-compatible server
+* Predictable read/write performance
+* Low latecny
+* High throughput (See: [Performance](README.md#Performance)
 
 ## Install
 
@@ -54,6 +58,38 @@ $ bitcask -p /tmp/db get Hello
 World
 ```
 
+## Usage (server)
+
+There is also a builtin very  simple Redis-compatible server called `bitcaskd`:
+
+```#!bash
+$ ./bitcaskd ./tmp
+INFO[0000] starting bitcaskd v0.0.7@146f777              bind=":6379" path=./tmp
+```
+
+Example session:
+
+```
+$ telnet localhost 6379
+Trying ::1...
+Connected to localhost.
+Escape character is '^]'.
+SET foo bar
++OK
+GET foo
+$3
+bar
+DEL foo
+:1
+GET foo
+$-1
+PING
++PONG
+QUIT
++OK
+Connection closed by foreign host.
+```
+
 ## Performance
 
 Benchmarks run on a 11" Macbook with a 1.4Ghz Intel Core i7:
@@ -61,13 +97,35 @@ Benchmarks run on a 11" Macbook with a 1.4Ghz Intel Core i7:
 ```
 $ make bench
 ...
-BenchmarkGet-4   	  300000	      5065 ns/op	     144 B/op	       4 allocs/op
-BenchmarkPut-4   	  100000	     14640 ns/op	     699 B/op	       7 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
 ```
 
-* ~30,000 reads/sec for non-active data
-* ~180,000 reads/sec for active data
-* ~60,000 writes/sec
+For 128B values:
+
+* ~200,000 reads/sec
+* ~130,000 writes/sec
+
+The full benchmark above shows linear performance as you increase key/value sizes.
 
 ## License
 

bitcask.go

@@ -2,35 +2,33 @@ package bitcask
 
 import (
 	"errors"
-	"fmt"
 	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
-	"sort"
-	"strconv"
 	"strings"
 	"time"
 
+	"github.com/derekparker/trie"
 	"github.com/gofrs/flock"
 )
 
-const (
-	DefaultMaxDatafileSize = 1 << 20 // 1MB
-)
-
 var (
-	ErrKeyNotFound       = errors.New("error: key not found")
-	ErrCannotAcquireLock = errors.New("error: cannot acquire lock")
+	ErrKeyNotFound    = errors.New("error: key not found")
+	ErrKeyTooLarge    = errors.New("error: key too large")
+	ErrValueTooLarge  = errors.New("error: value too large")
+	ErrDatabaseLocked = errors.New("error: database locked")
 )
 
 type Bitcask struct {
 	*flock.Flock
 
+	opts      Options
 	path      string
 	curr      *Datafile
 	keydir    *Keydir
 	datafiles []*Datafile
+	trie      *trie.Trie
 
 	maxDatafileSize int64
 }
@@ -38,6 +36,7 @@ type Bitcask struct {
 func (b *Bitcask) Close() error {
 	defer func() {
 		b.Flock.Unlock()
+		os.Remove(b.Flock.Path())
 	}()
 
 	for _, df := range b.datafiles {
@@ -73,12 +72,20 @@ func (b *Bitcask) Get(key string) ([]byte, error) {
 }
 
 func (b *Bitcask) Put(key string, value []byte) error {
+	if len(key) > b.opts.MaxKeySize {
+		return ErrKeyTooLarge
+	}
+	if len(value) > b.opts.MaxValueSize {
+		return ErrValueTooLarge
+	}
+
 	index, 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.id, index, time.Now().Unix())
+	b.trie.Add(key, item)
 
 	return nil
 }
@@ -90,10 +97,21 @@ func (b *Bitcask) Delete(key string) error {
 	}
 
 	b.keydir.Delete(key)
+	b.trie.Remove(key)
 
 	return nil
 }
 
+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
+}
+
 func (b *Bitcask) Fold(f func(key string) error) error {
 	for key := range b.keydir.Keys() {
 		if err := f(key); err != nil {
@@ -135,39 +153,6 @@ func (b *Bitcask) setMaxDatafileSize(size int64) error {
 	return nil
 }
 
-func MaxDatafileSize(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
-}
-
 func Merge(path string, force bool) error {
 	fns, err := getDatafiles(path)
 	if err != nil {
@@ -294,9 +279,11 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 		return nil, err
 	}
 
-	keydir := NewKeydir()
 	var datafiles []*Datafile
 
+	keydir := NewKeydir()
+	trie := trie.New()
+
 	for i, fn := range fns {
 		df, err := NewDatafile(path, ids[i], true)
 		if err != nil {
@@ -318,7 +305,8 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 
 			for key := range hint.Keys() {
 				item, _ := hint.Get(key)
-				keydir.Add(key, item.FileID, item.Index, item.Timestamp)
+				_ = keydir.Add(key, item.FileID, item.Index, item.Timestamp)
+				trie.Add(key, item)
 			}
 		} else {
 			for {
@@ -336,7 +324,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.Index, e.Timestamp)
+				trie.Add(e.Key, item)
 			}
 		}
 	}
@@ -353,10 +342,12 @@ func Open(path string, options ...func(*Bitcask) error) (*Bitcask, error) {
 
 	bitcask := &Bitcask{
 		Flock:     flock.New(filepath.Join(path, "lock")),
+		opts:      NewDefaultOptions(),
 		path:      path,
 		curr:      curr,
 		keydir:    keydir,
 		datafiles: datafiles,
+		trie:      trie,
 
 		maxDatafileSize: DefaultMaxDatafileSize,
 	}
@@ -374,7 +365,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,7 +3,10 @@ package bitcask
 import (
 	"fmt"
 	"io/ioutil"
+	"reflect"
+	"sort"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -127,6 +130,54 @@ func TestDeletedKeys(t *testing.T) {
 	})
 }
 
+func TestMaxKeySize(t *testing.T) {
+	assert := assert.New(t)
+
+	testdir, err := ioutil.TempDir("", "bitcask")
+	assert.NoError(err)
+
+	var db *Bitcask
+
+	size := 16
+
+	t.Run("Open", func(t *testing.T) {
+		db, err = Open(testdir, WithMaxKeySize(size))
+		assert.NoError(err)
+	})
+
+	t.Run("Put", func(t *testing.T) {
+		key := strings.Repeat(" ", size+1)
+		value := []byte("foobar")
+		err = db.Put(key, value)
+		assert.Error(err)
+		assert.Equal("error: key too large", err.Error())
+	})
+}
+
+func TestMaxValueSize(t *testing.T) {
+	assert := assert.New(t)
+
+	testdir, err := ioutil.TempDir("", "bitcask")
+	assert.NoError(err)
+
+	var db *Bitcask
+
+	size := 16
+
+	t.Run("Open", func(t *testing.T) {
+		db, err = Open(testdir, WithMaxValueSize(size))
+		assert.NoError(err)
+	})
+
+	t.Run("Put", func(t *testing.T) {
+		key := "foo"
+		value := []byte(strings.Repeat(" ", size+1))
+		err = db.Put(key, value)
+		assert.Error(err)
+		assert.Equal("error: value too large", err.Error())
+	})
+}
+
 func TestMerge(t *testing.T) {
 	assert := assert.New(t)
 
@@ -140,7 +191,7 @@ func TestMerge(t *testing.T) {
 		)
 
 		t.Run("Open", func(t *testing.T) {
-			db, err = Open(testdir, MaxDatafileSize(1024))
+			db, err = Open(testdir, WithMaxDatafileSize(1024))
 			assert.NoError(err)
 		})
 
@@ -198,6 +249,138 @@ func TestMerge(t *testing.T) {
 	})
 }
 
+func TestConcurrent(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)
+			assert.NoError(err)
+		})
+
+		t.Run("Put", func(t *testing.T) {
+			err = db.Put("foo", []byte("bar"))
+			assert.NoError(err)
+		})
+	})
+
+	t.Run("Concurrent", func(t *testing.T) {
+		t.Run("Put", func(t *testing.T) {
+			f := func(wg *sync.WaitGroup, x int) {
+				defer func() {
+					wg.Done()
+				}()
+				for i := 0; i <= 100; i++ {
+					if i%x == 0 {
+						key := fmt.Sprintf("k%d", i)
+						value := []byte(fmt.Sprintf("v%d", i))
+						err := db.Put(key, value)
+						assert.NoError(err)
+					}
+				}
+			}
+
+			wg := &sync.WaitGroup{}
+
+			go f(wg, 2)
+			wg.Add(1)
+
+			go f(wg, 3)
+			wg.Add(1)
+
+			wg.Wait()
+		})
+
+		t.Run("Get", func(t *testing.T) {
+			f := func(wg *sync.WaitGroup, N int) {
+				defer func() {
+					wg.Done()
+				}()
+				for i := 0; i <= N; i++ {
+					value, err := db.Get("foo")
+					assert.NoError(err)
+					assert.Equal([]byte("bar"), value)
+				}
+			}
+
+			wg := &sync.WaitGroup{}
+
+			go f(wg, 100)
+			wg.Add(1)
+
+			go f(wg, 100)
+			wg.Add(1)
+
+			wg.Wait()
+		})
+
+		t.Run("Close", func(t *testing.T) {
+			err = db.Close()
+			assert.NoError(err)
+		})
+	})
+}
+
+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)
 
@@ -210,7 +393,12 @@ func TestLocking(t *testing.T) {
 
 	_, err = Open(testdir)
 	assert.Error(err)
-	assert.Equal("error: cannot acquire lock", err.Error())
+	assert.Equal("error: database locked", err.Error())
+}
+
+type benchmarkTestCase struct {
+	name string
+	size int
 }
 
 func BenchmarkGet(b *testing.B) {
@@ -225,20 +413,39 @@ func BenchmarkGet(b *testing.B) {
 	}
 	defer db.Close()
 
-	err = db.Put("foo", []byte("bar"))
-	if err != nil {
-		b.Fatal(err)
+	tests := []benchmarkTestCase{
+		{"128B", 128},
+		{"256B", 256},
+		{"512B", 512},
+		{"1K", 1024},
+		{"2K", 2048},
+		{"4K", 4096},
+		{"8K", 8192},
+		{"16K", 16384},
+		{"32K", 32768},
 	}
 
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		val, err := db.Get("foo")
-		if err != nil {
-			b.Fatal(err)
-		}
-		if string(val) != "bar" {
-			b.Errorf("expected val=bar got=%s", val)
-		}
+	for _, tt := range tests {
+		b.Run(tt.name, func(b *testing.B) {
+			key := "foo"
+			value := []byte(strings.Repeat(" ", tt.size))
+
+			err = db.Put(key, value)
+			if err != nil {
+				b.Fatal(err)
+			}
+
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				val, err := db.Get(key)
+				if err != nil {
+					b.Fatal(err)
+				}
+				if string(val) != string(value) {
+					b.Errorf("unexpected value")
+				}
+			}
+		})
 	}
 }
 
@@ -254,11 +461,73 @@ func BenchmarkPut(b *testing.B) {
 	}
 	defer db.Close()
 
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		err := db.Put(fmt.Sprintf("key%d", i), []byte("bar"))
+	tests := []benchmarkTestCase{
+		{"128B", 128},
+		{"256B", 256},
+		{"512B", 512},
+		{"1K", 1024},
+		{"2K", 2048},
+		{"4K", 4096},
+		{"8K", 8192},
+		{"16K", 16384},
+		{"32K", 32768},
+	}
+
+	for _, tt := range tests {
+		b.Run(tt.name, func(b *testing.B) {
+			key := "foo"
+			value := []byte(strings.Repeat(" ", tt.size))
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				err := db.Put(key, value)
+				if err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}
+
+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/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

@@ -0,0 +1,123 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	log "github.com/sirupsen/logrus"
+	flag "github.com/spf13/pflag"
+	"github.com/tidwall/redcon"
+
+	"github.com/prologic/bitcask"
+)
+
+var (
+	bind            string
+	debug           bool
+	version         bool
+	maxDatafileSize int
+)
+
+func init() {
+	flag.Usage = func() {
+		fmt.Fprintf(os.Stderr, "Usage: %s [options] <path>\n", os.Args[0])
+		flag.PrintDefaults()
+	}
+
+	flag.BoolVarP(&version, "version", "v", false, "display version information")
+	flag.BoolVarP(&debug, "debug", "d", false, "enable debug logging")
+
+	flag.StringVarP(&bind, "bind", "b", ":6379", "interface and port to bind to")
+
+	flag.IntVar(&maxDatafileSize, "max-datafile-size", 1<<20, "maximum datafile size in bytes")
+}
+
+func main() {
+	flag.Parse()
+
+	if debug {
+		log.SetLevel(log.DebugLevel)
+	} else {
+		log.SetLevel(log.InfoLevel)
+	}
+
+	if version {
+		fmt.Printf("bitcaskd version %s", bitcask.FullVersion())
+		os.Exit(0)
+	}
+
+	if len(flag.Args()) < 1 {
+		flag.Usage()
+		os.Exit(1)
+	}
+
+	path := flag.Arg(0)
+
+	db, err := bitcask.Open(path, bitcask.WithMaxDatafileSize(maxDatafileSize))
+	if err != nil {
+		log.WithError(err).WithField("path", path).Error("error opening database")
+		os.Exit(1)
+	}
+
+	log.WithField("bind", bind).WithField("path", path).Infof("starting bitcaskd v%s", bitcask.FullVersion())
+
+	err = redcon.ListenAndServe(bind,
+		func(conn redcon.Conn, cmd redcon.Command) {
+			switch strings.ToLower(string(cmd.Args[0])) {
+			case "ping":
+				conn.WriteString("PONG")
+			case "quit":
+				conn.WriteString("OK")
+				conn.Close()
+			case "set":
+				if len(cmd.Args) != 3 {
+					conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
+					return
+				}
+				key := string(cmd.Args[1])
+				value := cmd.Args[2]
+				err = db.Put(key, value)
+				if err != nil {
+					conn.WriteString(fmt.Sprintf("ERR: %s", err))
+				} else {
+					conn.WriteString("OK")
+				}
+			case "get":
+				if len(cmd.Args) != 2 {
+					conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
+					return
+				}
+				key := string(cmd.Args[1])
+				value, err := db.Get(key)
+				if err != nil {
+					conn.WriteNull()
+				} else {
+					conn.WriteBulk(value)
+				}
+			case "del":
+				if len(cmd.Args) != 2 {
+					conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
+					return
+				}
+				key := string(cmd.Args[1])
+				err := db.Delete(key)
+				if err != nil {
+					conn.WriteInt(0)
+				} else {
+					conn.WriteInt(1)
+				}
+			default:
+				conn.WriteError("ERR unknown command '" + string(cmd.Args[0]) + "'")
+			}
+		},
+		func(conn redcon.Conn) bool {
+			return true
+		},
+		func(conn redcon.Conn, err error) {
+		},
+	)
+	if err != nil {
+		log.WithError(err).Fatal("oops")
+	}
+}

datafile.go

@@ -1,12 +1,14 @@
 package bitcask
 
 import (
-	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
+	"sync"
 	"time"
 
+	"github.com/pkg/errors"
+
 	pb "github.com/prologic/bitcask/proto"
 	"github.com/prologic/bitcask/streampb"
 )
@@ -20,11 +22,14 @@ var (
 )
 
 type Datafile struct {
-	id  int
-	r   *os.File
-	w   *os.File
-	dec *streampb.Decoder
-	enc *streampb.Encoder
+	sync.RWMutex
+
+	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) {
@@ -47,16 +52,23 @@ 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
 }
 
@@ -84,35 +96,26 @@ 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
+	return df.offset, nil
 }
 
-func (df *Datafile) Read() (pb.Entry, error) {
-	var e pb.Entry
+func (df *Datafile) Read() (e pb.Entry, 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)
 	if err != nil {
 		return
 	}
-	return df.Read()
+
+	return e, df.dec.Decode(&e)
 }
 
 func (df *Datafile) Write(e pb.Entry) (int64, error) {
@@ -120,20 +123,17 @@ 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.Index = df.offset
 	e.Timestamp = time.Now().Unix()
 
-	err = df.enc.Encode(&e)
+	df.Lock()
+	n, err := df.enc.Encode(&e)
+	df.Unlock()
+
 	if err != nil {
 		return -1, err
 	}
+	df.offset += n
 
-	return index, nil
+	return e.Index, nil
 }

go.mod

@@ -1,6 +1,7 @@
 module github.com/prologic/bitcask
 
 require (
+	github.com/derekparker/trie v0.0.0-20180212171413-e608c2733dc7
 	github.com/gofrs/flock v0.7.1
 	github.com/gogo/protobuf v1.2.1
 	github.com/golang/protobuf v1.2.0
@@ -12,7 +13,9 @@ require (
 	github.com/prometheus/client_golang v0.9.2 // indirect
 	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
 )

go.sum

@@ -8,6 +8,8 @@ github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/derekparker/trie v0.0.0-20180212171413-e608c2733dc7 h1:Cab9yoTQh1TxObKfis1DzZ6vFLK5kbeenMjRES/UE3o=
+github.com/derekparker/trie v0.0.0-20180212171413-e608c2733dc7/go.mod h1:D6ICZm05D9VN1n/8iOtBxLpXtoGp6HDFUJ1RNVieOSE=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/gofrs/flock v0.7.1 h1:DP+LD/t0njgoPBvT5MJLeliUIVQR03hiKR6vezdwHlc=
@@ -67,6 +69,8 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/tidwall/redcon v0.9.0 h1:tiT9DLAoohsdNaFg9Si5dRsv9+FjvZYnhMOEtSFwBqA=
+github.com/tidwall/redcon v0.9.0/go.mod h1:bdYBm4rlcWpst2XMwKVzWDF9CoUxEbUmM7CQrKeOZas=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

keydir.go

@@ -25,15 +25,18 @@ func NewKeydir() *Keydir {
 	}
 }
 
-func (k *Keydir) Add(key string, fileid int, index, timestamp int64) {
-	k.Lock()
-	defer k.Unlock()
-
-	k.kv[key] = Item{
+func (k *Keydir) Add(key string, fileid int, index, timestamp int64) Item {
+	item := Item{
 		FileID:    fileid,
 		Index:     index,
 		Timestamp: timestamp,
 	}
+
+	k.Lock()
+	k.kv[key] = item
+	k.Unlock()
+
+	return item
 }
 
 func (k *Keydir) Get(key string) (Item, bool) {

options.go

@@ -0,0 +1,42 @@
+package bitcask
+
+const (
+	DefaultMaxDatafileSize = 1 << 20 // 1MB
+	DefaultMaxKeySize      = 64      // 64 bytes
+	DefaultMaxValueSize    = 1 << 16 // 65KB
+)
+
+type Options struct {
+	MaxDatafileSize int
+	MaxKeySize      int
+	MaxValueSize    int
+}
+
+func NewDefaultOptions() Options {
+	return Options{
+		MaxDatafileSize: DefaultMaxDatafileSize,
+		MaxKeySize:      DefaultMaxKeySize,
+		MaxValueSize:    DefaultMaxValueSize,
+	}
+}
+
+func WithMaxDatafileSize(size int) func(*Bitcask) error {
+	return func(b *Bitcask) error {
+		b.opts.MaxDatafileSize = size
+		return nil
+	}
+}
+
+func WithMaxKeySize(size int) func(*Bitcask) error {
+	return func(b *Bitcask) error {
+		b.opts.MaxKeySize = size
+		return nil
+	}
+}
+
+func WithMaxValueSize(size int) func(*Bitcask) error {
+	return func(b *Bitcask) error {
+		b.opts.MaxValueSize = size
+		return nil
+	}
+}

streampb/stream.go

@@ -1,6 +1,7 @@
 package streampb
 
 import (
+	"bufio"
 	"encoding/binary"
 	"io"
 
@@ -16,57 +17,64 @@ const (
 
 // NewEncoder creates a streaming protobuf encoder.
 func NewEncoder(w io.Writer) *Encoder {
-	return &Encoder{w: w, prefixBuf: make([]byte, prefixSize)}
+	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
-	prefixBuf []byte
+	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(e.prefixBuf, uint64(len(buf)))
+	binary.BigEndian.PutUint64(prefixBuf, uint64(len(buf)))
 
-	if _, err := e.w.Write(e.prefixBuf); err != nil {
-		return errors.Wrap(err, "failed writing length prefix")
+	if _, err := e.w.Write(prefixBuf); err != nil {
+		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.
 func NewDecoder(r io.Reader) *Decoder {
-	return &Decoder{
-		r:         r,
-		prefixBuf: make([]byte, prefixSize),
-	}
+	return &Decoder{r: r}
 }
 
 // Decoder wraps an underlying io.Reader and allows you to stream
 // proto decodings on it.
 type Decoder struct {
-	r         io.Reader
-	prefixBuf []byte
+	r io.Reader
 }
 
 // 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 {
-	_, err := io.ReadFull(d.r, d.prefixBuf)
+	prefixBuf := make([]byte, prefixSize)
+
+	_, err := io.ReadFull(d.r, prefixBuf)
 	if err != nil {
 		return err
 	}
 
-	n := binary.BigEndian.Uint64(d.prefixBuf)
+	n := binary.BigEndian.Uint64(prefixBuf)
 
 	buf := make([]byte, n)
 

utils.go

@@ -0,0 +1,36 @@
+package bitcask
+
+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
+}