Replace LevelDB with BoltDB. Also, fix unit tests. Also make unit tes…

.travis.yml

@@ -1,11 +1,19 @@
 language: go
 go: 
- - 1.2
+ - 1.4
 
-script:
- - go get
+install:
  - go get github.com/stretchr/testify/assert
  - go get github.com/golang/glog
- - go get code.google.com/p/go.tools/cmd/cover
- - cd replica
- - go test -v -cover
+ - go get golang.org/x/tools/cmd/cover
+ - go get github.com/boltdb/bolt/...
+ - go get github.com/gogo/protobuf/proto
+ - go get github.com/gogo/protobuf/protoc-gen-gogo
+ - go get github.com/gogo/protobuf/gogoproto
+ - go get github.com/gogo/protobuf/protoc-gen-gofast
+
+script:
+ - go test -v -cover ./...
+
+sudo: false
+

build.sh

@@ -1,5 +1,5 @@
 #!/bin/bash
 
 go get github.com/stretchr/testify
-go get code.google.com/p/leveldb-go/leveldb
+go get github.com/boltdb/bolt/...
 go get github.com/golang/glog

demo/server.go

@@ -8,9 +8,9 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/go-distributed/epaxos/message"
-	"github.com/go-distributed/epaxos/replica"
-	"github.com/go-distributed/epaxos/transporter"
+	"github.com/sargun/epaxos/message"
+	"github.com/sargun/epaxos/replica"
+	"github.com/sargun/epaxos/transporter"
 	"github.com/golang/glog"
 )
 

livetest/livereplica_test.go

@@ -7,11 +7,13 @@ import (
 	"testing"
 	"time"
 
-	"github.com/go-distributed/epaxos/message"
-	"github.com/go-distributed/epaxos/replica"
-	"github.com/go-distributed/epaxos/test"
-	"github.com/go-distributed/epaxos/transporter"
+	"github.com/sargun/epaxos/message"
+	"github.com/sargun/epaxos/replica"
+	"github.com/sargun/epaxos/test"
+	"github.com/sargun/epaxos/transporter"
 	"github.com/stretchr/testify/assert"
+	"io/ioutil"
+	"os"
 )
 
 var _ = fmt.Printf
@@ -45,6 +47,7 @@ func livetestlibSetupCluster(clusterSize int) []*replica.Replica {
 			Size:            uint8(clusterSize),
 			StateMachine:    new(test.DummySM),
 			Transporter:     transporter.NewDummyTR(uint8(i), clusterSize),
+			PersistentPath:  tempfile(),
 		}
 		nodes[i], _ = replica.New(param)
 	}
@@ -73,6 +76,7 @@ func livetestlibSetupEasyTimeoutCluster(clusterSize int) []*replica.Replica {
 			Size:            uint8(clusterSize),
 			StateMachine:    new(test.DummySM),
 			Transporter:     transporter.NewDummyTR(uint8(i), clusterSize),
+			PersistentPath:  tempfile(),
 		}
 		nodes[i], _ = replica.New(param)
 	}
@@ -360,3 +364,12 @@ func Test3ProposerConflictTimeout(t *testing.T) {
 		assert.True(t, liveTestlibVerifyDependency(nodes[0], pos))
 	}
 }
+
+
+// Borrowed from: https://github.com/boltdb/bolt/blob/2f4ba1c5331c044ed8c2743b791d5bedf0efa54b/db_test.go#L30-L38
+func tempfile() string {
+	f, _ := ioutil.TempFile("", "bolt-")
+	f.Close()
+	os.Remove(f.Name())
+	return f.Name()
+}

message/ballot_test.go

@@ -117,7 +117,7 @@ func TestBallotClone(t *testing.T) {
 
 func TestBallotEpoch(t *testing.T) {
 	b := NewBallot(2, 33, 4)
-	assert.Equal(t, b.GetEpoch(), uint8(2))
+	assert.Equal(t, b.GetEpoch(), uint32(2))
 }
 
 func TestBallotIsInitialBallot(t *testing.T) {

persistent/db.go

@@ -0,0 +1,129 @@
+package persistent
+
+import (
+	"os"
+	"path/filepath"
+
+	"github.com/boltdb/bolt"
+	"github.com/sargun/epaxos"
+)
+
+var BUCKET_NAME []byte = []byte{'e', 'p', 'a', 'x' , 'o', 's'}
+type DB interface {
+	GetPath()	string
+	Put(key string, value []byte) error
+	Get(key string) ([]byte, error)
+	Delete(key string) error
+	BatchPut(kvs []*epaxos.KVpair) error
+	Close() error
+	Drop() error
+}
+type BoltDB struct {
+	db		*bolt.DB
+	fpath	string
+}
+/*
+type LevelDB struct {
+	fpath string
+	ldb   *leveldb.DB
+	wsync *db.WriteOptions
+}
+*/
+
+func NewBoltDB(path string, restore bool) (*BoltDB, error) {
+	fpath, err := filepath.Abs(path)
+	if err != nil {
+		return nil, err
+	}
+
+	if !restore {
+		err = os.Remove(fpath)
+		if err != nil && os.IsExist(err) {
+				return nil, err
+		}
+	}
+
+	db, err := bolt.Open(fpath, 0600, nil) // TODO: tune the option
+	if err != nil {
+		return nil, err
+	}
+
+	err2 := db.Update(func(tx *bolt.Tx) error {
+		tx.CreateBucketIfNotExists(BUCKET_NAME)
+		return nil
+	})
+	if err2 != nil {
+		db.Close()
+		return nil, err2
+	}
+
+	ret := &BoltDB{
+		db:	db,
+		fpath: fpath,
+	}
+	return ret, nil
+}
+
+func (d *BoltDB) Put(key string, value []byte) error {
+	err := d.db.Update(func(tx *bolt.Tx) error {
+		tx.Bucket(BUCKET_NAME).Put([]byte(key), value)
+		return nil
+	})
+	return err
+}
+func (d *BoltDB) Get(key string) ([]byte, error) {
+	var ret []byte
+	found := false
+	err := d.db.View(func(tx *bolt.Tx) error {
+		value := tx.Bucket(BUCKET_NAME).Get([]byte(key))
+		if value != nil {
+			found = true
+			ret = make([]byte, len(value))
+			copy(ret, value)
+		}
+		return nil
+	})
+	if found {
+		return ret, err
+	} else {
+		return ret, epaxos.ErrorNotFound
+	}
+}
+func (d *BoltDB) Delete(key string) error {
+	err := d.db.Update(func(tx *bolt.Tx) error {
+		tx.Bucket(BUCKET_NAME).Delete([]byte(key))
+		return nil
+	})
+	return err
+}
+
+func (d *BoltDB) Close() error {
+	return d.db.Close()
+}
+
+func (d *BoltDB) Drop() error {
+	err := os.Remove(d.fpath)
+	if os.IsNotExist(err) {
+		return nil
+	} else {
+		return err
+	}
+}
+
+
+func (d *BoltDB) BatchPut(kvs []*epaxos.KVpair) error {
+	err := d.db.Update(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket(BUCKET_NAME)
+		for i := range kvs {
+			bucket.Put([]byte(kvs[i].Key), kvs[i].Value)
+		}
+		return nil
+	})
+	return err
+
+}
+
+func (d *BoltDB) GetPath() string {
+
+	return d.db.Path()
+}

persistent/leveldb_test.go → persistent/db_test.go

@@ -3,26 +3,25 @@ package persistent
 import (
 	"testing"
 
-	"code.google.com/p/leveldb-go/leveldb/db"
-	"github.com/go-distributed/epaxos"
+	"github.com/sargun/epaxos"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestNewCloseAndDrop(t *testing.T) {
-	l, err := NewLevelDB("/tmp/test", false)
+	l, err := NewBoltDB("/tmp/test", false)
 	defer func() {
 		assert.NoError(t, l.Close())
 		assert.NoError(t, l.Drop())
 	}()
 
 	assert.NoError(t, err)
 	assert.NotNil(t, l)
-	assert.NotNil(t, l.ldb)
-	assert.Equal(t, l.wsync, &db.WriteOptions{Sync: true})
+	assert.NotNil(t, l.db)
+	assert.Equal(t, l.db.NoSync, false)
 }
 
 func TestPutAndGet(t *testing.T) {
-	l, err := NewLevelDB("/tmp/test", false)
+	l, err := NewBoltDB("/tmp/test", false)
 	assert.NoError(t, err)
 
 	defer func() {
@@ -41,7 +40,7 @@ func TestPutAndGet(t *testing.T) {
 }
 
 func TestBatchPut(t *testing.T) {
-	l, err := NewLevelDB("/tmp/test", false)
+	l, err := NewBoltDB("/tmp/test", false)
 	assert.NoError(t, err)
 
 	defer func() {

protobuf/Makefile

@@ -1,2 +1,9 @@
-all: message.proto
-	protoc --proto_path=${GOPATH}/src:. --gogo_out=. message.proto
+.PHONY: all clean
+
+all: message.pb.go
+
+message.pb.go: message.proto
+	protoc --proto_path=${GOPATH}/src:${GOPATH}/src/github.com/gogo/protobuf/protobuf:. --gogo_out=. *.proto
+
+clean:
+	rm -f message.pb.go