add log init and restore

Author: Hongchao Deng

db/backend.go

@@ -1,35 +1,61 @@
 package db
 
 import (
+	"fmt"
+	"io/ioutil"
+	"os"
 	"strings"
 
-	"github.com/go-distributed/xtree/db/log"
+	dblog "github.com/go-distributed/xtree/db/log"
 	"github.com/go-distributed/xtree/db/message"
 	"github.com/go-distributed/xtree/third-party/github.com/google/btree"
 )
 
 type backend struct {
-	bt    *btree.BTree
-	cache *cache
-	rev   int
-	log   *log.Log
+	bt     *btree.BTree
+	cache  *cache
+	rev    int
+	dblog  *dblog.DBLog
+	config *DBConfig
 }
 
 func newBackend() *backend {
-	bt := btree.New(10)
-	log, err := log.Create()
+	dataDir, err := ioutil.TempDir("", "backend")
+	if err != nil {
+		panic("not implemented")
+	}
+
+	config := &DBConfig{
+		DataDir: dataDir,
+	}
+	b, err := newBackendWithConfig(config)
 	if err != nil {
-		panic("Not implemented")
+		panic("not implemented")
+	}
+	return b
+}
+
+func newBackendWithConfig(config *DBConfig) (b *backend, err error) {
+	bt := btree.New(10)
+	b = &backend{
+		bt:     bt,
+		cache:  newCache(),
+		config: config,
 	}
-	return &backend{
-		bt:    bt,
-		cache: newCache(),
-		log:   log,
+	haveLog := dblog.Exist(config.DataDir)
+	switch haveLog {
+	case false:
+		fmt.Println("didn't have log file. Init...")
+		err = b.init(config)
+	case true:
+		fmt.Println("had log file. Restore...")
+		err = b.restore(config)
 	}
+	return
 }
 
 func (b *backend) getData(offset int64) []byte {
-	rec, err := b.log.GetRecord(offset)
+	rec, err := b.dblog.GetRecord(offset)
 	if err != nil {
 		panic("unimplemented")
 	}
@@ -77,7 +103,7 @@ func (b *backend) Put(rev int, path Path, data []byte) {
 	}
 
 	b.rev++
-	offset, err := b.log.Append(&message.Record{
+	offset, err := b.dblog.Append(&message.Record{
 		Key:  path.p,
 		Data: data,
 	})
@@ -102,5 +128,33 @@ func (b *backend) Ls(pathname string) (paths []Path) {
 		paths = append(paths, *p)
 		return true
 	})
+
+	return
+}
+
+// init() creates a new log file
+func (b *backend) init(config *DBConfig) (err error) {
+	b.dblog, err = dblog.Create(config.DataDir)
 	return
 }
+
+// restore() restores database from the log file.
+func (b *backend) restore(config *DBConfig) (err error) {
+	rev := 0
+	return dblog.Reuse(config.DataDir,
+		func(l *dblog.DBLog) {
+			b.dblog = l
+		},
+		func(r *message.Record) (err error) {
+			rev++
+			p := newPath(r.Key)
+			b.Put(rev, *p, r.Data)
+			return
+		})
+}
+
+// clean up resource after testing
+func (b *backend) testableCleanupResource() (err error) {
+	b.dblog.Close()
+	return os.RemoveAll(b.config.DataDir)
+}

db/backend_test.go

@@ -17,6 +17,7 @@ func TestPut(t *testing.T) {
 	}
 
 	b := newBackend()
+	defer b.testableCleanupResource()
 	for i, tt := range tests {
 		b.Put(tt.rev, tt.path, tt.data)
 		v := b.Get(tt.rev, tt.path)
@@ -27,6 +28,7 @@ func TestPut(t *testing.T) {
 			t.Errorf("#%d: data = %s, want %s", i, v.data, tt.data)
 		}
 	}
+
 }
 
 func TestPutOnExistingPath(t *testing.T) {
@@ -40,6 +42,7 @@ func TestPutOnExistingPath(t *testing.T) {
 	}
 
 	b := newBackend()
+	defer b.testableCleanupResource()
 	for i, tt := range tests {
 		b.Put(2*i+1, tt.path, tt.data1)
 		v := b.Get(2*i+1, tt.path)
@@ -65,6 +68,8 @@ func TestPutOnExistingPath(t *testing.T) {
 
 func TestGetMVCC(t *testing.T) {
 	b := newBackend()
+	defer b.testableCleanupResource()
+
 	b.Put(1, *newPath("/a"), []byte("1"))
 	b.Put(2, *newPath("/b"), []byte("2"))
 	b.Put(3, *newPath("/a"), []byte("3"))
@@ -99,12 +104,15 @@ func TestGetMVCC(t *testing.T) {
 }
 
 func TestLs(t *testing.T) {
-	back := newBackend()
 	d := []byte("somedata")
-	back.Put(1, *newPath("/a"), d)
-	back.Put(2, *newPath("/a/b"), d)
-	back.Put(3, *newPath("/a/c"), d)
-	back.Put(4, *newPath("/b"), d)
+
+	b := newBackend()
+	defer b.testableCleanupResource()
+
+	b.Put(1, *newPath("/a"), d)
+	b.Put(2, *newPath("/a/b"), d)
+	b.Put(3, *newPath("/a/c"), d)
+	b.Put(4, *newPath("/b"), d)
 
 	tests := []struct {
 		p   string
@@ -118,7 +126,7 @@ func TestLs(t *testing.T) {
 		{"/c", []string{}},
 	}
 	for i, tt := range tests {
-		ps := back.Ls(tt.p)
+		ps := b.Ls(tt.p)
 		if len(ps) != len(tt.wps) {
 			t.Fatalf("#%d: len(ps) = %d, want %d", i, len(ps), len(tt.wps))
 		}
@@ -130,9 +138,44 @@ func TestLs(t *testing.T) {
 	}
 }
 
+func TestRestore(t *testing.T) {
+	tests := []struct {
+		rev  int
+		path Path
+		data []byte
+	}{
+		{1, *newPath("/foo/bar"), []byte("somedata")},
+		{2, *newPath("/bar/foo"), []byte("datasome")},
+	}
+
+	b := newBackend()
+	for _, tt := range tests {
+		// append records to the log
+		b.Put(tt.rev, tt.path, tt.data)
+	}
+	b.dblog.Close()
+
+	// simulate restoring log in another backend
+	b2, err := newBackendWithConfig(b.config)
+	defer b2.testableCleanupResource()
+	if err != nil {
+		t.Errorf("newBackendWithConfig failed: %v", err)
+	}
+	for i, tt := range tests {
+		v := b2.Get(tt.rev, tt.path)
+		if v.rev != tt.rev {
+			t.Errorf("#%d: rev = %d, want %d", i, v.rev, tt.rev)
+		}
+		if !reflect.DeepEqual(v.data, tt.data) {
+			t.Errorf("#%d: data = %s, want %s", i, v.data, tt.data)
+		}
+	}
+}
+
 func BenchmarkPut(b *testing.B) {
 	b.StopTimer()
 	back := newBackend()
+	defer back.testableCleanupResource()
 	d := []byte("somedata")
 	path := make([]Path, b.N)
 	for i := range path {
@@ -148,6 +191,8 @@ func BenchmarkPut(b *testing.B) {
 func BenchmarkGetWithCache(b *testing.B) {
 	b.StopTimer()
 	back := newBackend()
+	defer back.testableCleanupResource()
+
 	d := []byte("somedata")
 	path := make([]Path, b.N)
 	for i := range path {
@@ -168,6 +213,7 @@ func BenchmarkGetWithCache(b *testing.B) {
 func BenchmarkGetWithOutCache(b *testing.B) {
 	b.StopTimer()
 	back := newBackend()
+	defer back.testableCleanupResource()
 	back.cache = nil
 	d := []byte("somedata")
 	path := make([]Path, b.N)

db/config.go

@@ -0,0 +1,5 @@
+package db
+
+type DBConfig struct {
+	DataDir string
+}

db/log/encoder_test.go

@@ -30,7 +30,7 @@ func TestEncoderDecoder(t *testing.T) {
 			t.Fatalf("#%d: cannot flush encode, err: %v", i, err)
 		}
 
-		rec := &message.Record{}
+		rec := new(message.Record)
 		dBuf := bytes.NewBuffer(eBuf.Bytes())
 		decoder := newDecoder(dBuf)
 

db/log/log.go

@@ -1,49 +1,112 @@
 package log
 
 import (
-	"io/ioutil"
+	"fmt"
+	"io"
 	"os"
+	"path"
 
 	"github.com/go-distributed/xtree/db/message"
 )
 
-type Log struct {
-	f *os.File
+const (
+	logFilename = "records.log"
+)
+
+type DBLog struct {
+	writeFile, readFile *os.File
+	encoder             *encoder
+}
+
+func Create(dataDir string) (l *DBLog, err error) {
+	l, err = newDBLog(path.Join(dataDir, logFilename), true)
+	return
 }
 
-func Create() (*Log, error) {
-	f, err := ioutil.TempFile("", "backend")
-	if err != nil {
-		return nil, err
+func newDBLog(logPath string, needCreate bool) (l *DBLog, err error) {
+	var writeFile, readFile *os.File
+	flag := os.O_WRONLY | os.O_APPEND | os.O_SYNC
+	if needCreate {
+		flag |= os.O_CREATE
+	}
+	if writeFile, err = os.OpenFile(logPath, flag, 0600); err != nil {
+		return
+	}
+	if !needCreate {
+		writeFile.Seek(0, os.SEEK_END)
+	}
+	if readFile, err = os.Open(logPath); err != nil {
+		writeFile.Close()
+		return
+	}
+	l = &DBLog{
+		writeFile: writeFile,
+		readFile:  readFile,
+		encoder:   newEncoder(writeFile),
 	}
-	return &Log{f}, nil
+	return
 }
 
-func (l *Log) Destroy() error {
-	if err := l.f.Close(); err != nil {
-		return err
+func Reuse(dataDir string,
+	setLog func(*DBLog),
+	replayRecord func(*message.Record) error) (err error) {
+	var l *DBLog
+	if l, err = newDBLog(path.Join(dataDir, logFilename),
+		false); err != nil {
+		return
 	}
-	return os.Remove(l.f.Name())
+	setLog(l)
+	decoder := newDecoder(l.readFile)
+	// TODO: parallel?
+	fi, _ := os.Stat(l.readFile.Name())
+	fmt.Println(fi.Size())
+	return
+	for {
+		r := new(message.Record)
+		if err = decoder.decode(r); err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return
+		}
+		ret, _ := l.readFile.Seek(0, os.SEEK_CUR)
+		fmt.Printf("%v %#v\n", ret, r)
+		if err = replayRecord(r); err != nil {
+			return
+		}
+	}
+}
+
+func Exist(dataDir string) bool {
+	p := path.Join(dataDir, logFilename)
+	_, err := os.Stat(p)
+	return err == nil
 }
 
-func (l *Log) GetRecord(offset int64) (r *message.Record, err error) {
-	if _, err = l.f.Seek(offset, 0); err != nil {
+func (l *DBLog) GetRecord(offset int64) (r *message.Record, err error) {
+	if _, err = l.readFile.Seek(offset, 0); err != nil {
 		return
 	}
-	decoder := newDecoder(l.f)
-	r = &message.Record{}
+	decoder := newDecoder(l.readFile)
+	r = new(message.Record)
 	err = decoder.decode(r)
 	return
 }
 
-func (l *Log) Append(r *message.Record) (offset int64, err error) {
-	if offset, err = l.f.Seek(0, 2); err != nil {
+func (l *DBLog) Append(r *message.Record) (offset int64, err error) {
+	if offset, err = l.writeFile.Seek(0, os.SEEK_CUR); err != nil {
 		return
 	}
-	encoder := newEncoder(l.f)
-	if err = encoder.encode(r); err != nil {
+	if err = l.encoder.encode(r); err != nil {
 		return
 	}
-	err = encoder.flush()
+	err = l.encoder.flush()
 	return offset, err
 }
+
+func (l *DBLog) Close() (err error) {
+	if err = l.readFile.Close(); err != nil {
+		return
+	}
+	return l.writeFile.Close()
+}