update table test

rpl/file_table.go

@@ -18,7 +18,7 @@ import (
 
 var (
 	magic             = []byte("\x1c\x1d\xb8\x88\xff\x9e\x45\x55\x40\xf0\x4c\xda\xe0\xce\x47\xde\x65\x48\x71\x17")
-	log0Data          = []byte("00000000000000000")
+	log0Data          = make([]byte, 17)
 	errTableNeedFlush = errors.New("write table need flush")
 	errTableFrozen    = errors.New("write table is frozen")
 )
@@ -26,7 +26,7 @@ var (
 const tableReaderKeepaliveInterval int64 = 30
 
 func fmtTableName(index int64) string {
-	return fmt.Sprintf("%08d", index)
+	return fmt.Sprintf("%08d.ldb", index)
 }
 
 type tableReader struct {
@@ -184,13 +184,15 @@ func (t *tableReader) repair() error {
 
 	defer t.close()
 
-	tw := newTableWriter(path.Base(t.name), t.index, maxLogFileSize)
+	tw := newTableWriter(path.Dir(t.name), t.index, maxLogFileSize)
-	tw.name = tw.name + ".tmp"
+
-	os.Remove(tw.name)
+	tmpName := tw.name + ".tmp"
+	tw.name = tmpName
+	os.Remove(tmpName)
 
 	defer func() {
 		tw.Close()
-		os.Remove(tw.name)
+		os.Remove(tmpName)
 	}()
 
 	var l Log
@@ -221,9 +223,11 @@ func (t *tableReader) repair() error {
 	t.offsetStartPos = tr.offsetStartPos
 	t.offsetLen = tr.offsetLen
 
+	defer tr.Close()
+
 	os.Remove(t.name)
 
-	if err := os.Rename(tw.name, t.name); err != nil {
+	if err := os.Rename(tmpName, t.name); err != nil {
 		return err
 	}
 
@@ -244,7 +248,7 @@ func (t *tableReader) decodeLogHead(l *Log, pos int64) (int64, error) {
 	return pos + int64(l.HeadSize()) + int64(dataLen), nil
 }
 
-func (t *tableReader) ReadLog(id uint64, l *Log) error {
+func (t *tableReader) GetLog(id uint64, l *Log) error {
 	if id < t.first || id > t.last {
 		return fmt.Errorf("log %d not in [%d=%d]", id, t.first, t.last)
 	}
@@ -373,7 +377,7 @@ func (t *tableWriter) Flush() (*tableReader, error) {
 	st, _ := t.wf.Stat()
 
 	tr.offsetStartPos = st.Size() + int64(len(log0Data))
-	tr.offsetLen = uint32(len(t.offsetBuf) / 4)
+	tr.offsetLen = uint32(len(t.offsetBuf))
 
 	tr.first = t.first
 	tr.last = t.last
@@ -416,13 +420,13 @@ func (t *tableWriter) Flush() (*tableReader, error) {
 }
 
 func (t *tableWriter) StoreLog(l *Log) error {
+	if l.ID == 0 {
+		return ErrStoreLogID
+	}
+
 	t.Lock()
 	defer t.Unlock()
 
-	if t.frozen {
-		return errTableFrozen
-	}
-
 	if t.last > 0 && l.ID != t.last+1 {
 		return ErrStoreLogID
 	}
@@ -433,7 +437,7 @@ func (t *tableWriter) StoreLog(l *Log) error {
 
 	var err error
 	if t.wf == nil {
-		if t.wf, err = os.OpenFile(t.name, os.O_CREATE|os.O_APPEND, 0644); err != nil {
+		if t.wf, err = os.OpenFile(t.name, os.O_CREATE|os.O_WRONLY, 0644); err != nil {
 			return err
 		}
 	}
@@ -473,7 +477,7 @@ func (t *tableWriter) GetLog(id uint64, l *Log) error {
 		return errTableFrozen
 	}
 
-	if id < t.first && id > t.last {
+	if id < t.first || id > t.last {
 		return fmt.Errorf("log %d not in [%d=%d]", id, t.first, t.last)
 	}
 

rpl/file_table_test.go

@@ -1,17 +1,198 @@
 package rpl
 
 import (
+	"github.com/siddontang/go/log"
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 )
 
 func TestFileTable(t *testing.T) {
-	base, err := ioutil.TempDir("./", "test_table")
+	log.SetLevel(log.LevelFatal)
+
+	base, err := ioutil.TempDir("", "test_table")
 	if err != nil {
 		t.Fatal(err)
 	}
 
+	os.MkdirAll(base, 0755)
+
 	defer os.RemoveAll(base)
 
+	l := new(Log)
+	l.Compression = 0
+	l.Data = make([]byte, 4096)
+
+	w := newTableWriter(base, 1, 1024*1024)
+	defer w.Close()
+
+	for i := 0; i < 10; i++ {
+		l.ID = uint64(i + 1)
+		l.CreateTime = uint32(time.Now().Unix())
+
+		l.Data[0] = byte(i + 1)
+
+		if err := w.StoreLog(l); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if w.first != 1 {
+		t.Fatal(w.first)
+	} else if w.last != 10 {
+		t.Fatal(w.last)
+	}
+
+	l.ID = 10
+	if err := w.StoreLog(l); err == nil {
+		t.Fatal("must err")
+	}
+
+	var ll Log
+
+	if err = ll.Unmarshal(log0Data); err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 0; i < 10; i++ {
+		if err := w.GetLog(uint64(i+1), &ll); err != nil {
+			t.Fatal(err)
+		} else if len(ll.Data) != 4096 {
+			t.Fatal(len(ll.Data))
+		} else if ll.Data[0] != byte(i+1) {
+			t.Fatal(ll.Data[0])
+		}
+	}
+
+	if err := w.GetLog(12, &ll); err == nil {
+		t.Fatal("must nil")
+	}
+
+	for i := 0; i < 10; i++ {
+		if err := w.GetLog(uint64(i+1), &ll); err != nil {
+			t.Fatal(err)
+		} else if len(ll.Data) != 4096 {
+			t.Fatal(len(ll.Data))
+		} else if ll.Data[0] != byte(i+1) {
+			t.Fatal(ll.Data[0])
+		}
+	}
+
+	var r *tableReader
+	name := w.name
+
+	if r, err = w.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 10; i < 20; i++ {
+		l.ID = uint64(i + 1)
+		l.CreateTime = uint32(time.Now().Unix())
+
+		l.Data[0] = byte(i + 1)
+
+		if err := w.StoreLog(l); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if w.first != 11 {
+		t.Fatal(w.first)
+	} else if w.last != 20 {
+		t.Fatal(w.last)
+	}
+
+	defer r.Close()
+
+	if err := r.GetLog(12, &ll); err == nil {
+		t.Fatal("must nil")
+	}
+
+	r.Close()
+
+	if r, err = newTableReader(name); err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	for i := 0; i < 10; i++ {
+		if err := r.GetLog(uint64(i+1), &ll); err != nil {
+			t.Fatal(err)
+		} else if len(ll.Data) != 4096 {
+			t.Fatal(len(ll.Data))
+		} else if ll.Data[0] != byte(i+1) {
+			t.Fatal(ll.Data[0])
+		}
+	}
+
+	if err := r.GetLog(12, &ll); err == nil {
+		t.Fatal("must nil")
+	}
+
+	st, _ := r.f.Stat()
+	s := st.Size()
+
+	r.Close()
+
+	testRepair(t, name, s, 11)
+	testRepair(t, name, s, 32)
+	testRepair(t, name, s, 42)
+	testRepair(t, name, s, 72)
+
+	if err := os.Truncate(name, s-73); err != nil {
+		t.Fatal(err)
+	}
+
+	if r, err = newTableReader(name); err == nil {
+		t.Fatal("can not repair")
+	}
+
+	name = w.name
+
+	if r, err := w.Flush(); err != nil {
+		t.Fatal(err)
+	} else {
+		r.Close()
+	}
+
+	if r, err = newTableReader(name); err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+}
+
+func testRepair(t *testing.T, name string, s int64, cutSize int64) {
+	var r *tableReader
+	var err error
+
+	if err := os.Truncate(name, s-cutSize); err != nil {
+		t.Fatal(err)
+	}
+
+	if r, err = newTableReader(name); err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	var ll Log
+	for i := 0; i < 10; i++ {
+		if err := r.GetLog(uint64(i+1), &ll); err != nil {
+			t.Fatal(err)
+		} else if len(ll.Data) != 4096 {
+			t.Fatal(len(ll.Data))
+		} else if ll.Data[0] != byte(i+1) {
+			t.Fatal(ll.Data[0])
+		}
+	}
+
+	if err := r.GetLog(12, &ll); err == nil {
+		t.Fatal("must nil")
+	}
+
+	st, _ := r.f.Stat()
+	if s != st.Size() {
+		t.Fatalf("repair error size %d != %d", s, st.Size())
+	}
+
 }