generate rdb file

README.md

@@ -0,0 +1,3 @@
+* 简单的基于 epoll 的事件处理；
+* set/get 命令
+* 生成 rdb 文件（save 命令）

dict/dict.go

@@ -18,6 +18,10 @@ func (de *DictEntry) GetVal() interface{} {
 	return de.v
 }
 
+func (de *DictEntry) GetKey() interface{} {
+	return de.key
+}
+
 type DictType struct {
 	HashFunction func(key interface{}) uint64
 	KeyDup       func(privdata interface{}, key interface{}) interface{}
@@ -235,6 +239,10 @@ func (d *Dict) SetEntryVal(de *DictEntry, val interface{}) {
 	de.v = val
 }
 
+func (d *Dict) Size() int {
+	return int(d.ht[0].used + d.ht[1].used)
+}
+
 func (d *Dict) Add(key interface{}, val interface{}) bool {
 	entry := d.addRaw(key, nil)
 
@@ -337,3 +345,17 @@ func (d *Dict) genericDelete(key interface{}) *DictEntry {
 func (d *Dict) Delete(key interface{}) bool {
 	return d.genericDelete(key) != nil
 }
+
+func (d *Dict) GetSafeInterator() *DictIterator {
+	iter := &DictIterator{
+		d:         d,
+		table:     0,
+		index:     -1,
+		safe:      false,
+		entry:     nil,
+		nextEntry: nil,
+	}
+
+	iter.safe = true
+	return iter
+}

dict/dict_iter.go

@@ -0,0 +1,44 @@
+package dict
+
+type DictIterator struct {
+	d                *Dict
+	index            int
+	table            int
+	safe             bool
+	entry, nextEntry *DictEntry
+	// fingerprint      int64
+}
+
+func (di *DictIterator) Next() *DictEntry {
+	for {
+		if di.entry == nil {
+			ht := &di.d.ht[di.table]
+			if di.index == -1 && di.table == 0 {
+				if di.safe {
+					di.d.iterators++
+				}
+				// else {
+				// di.fingerprint =
+				// }
+			}
+			di.index++
+			if di.index >= int(ht.size) {
+				if di.d.isRehashing() && di.table == 0 {
+					di.table++
+					di.index = 0
+					ht = &di.d.ht[1]
+				} else {
+					break
+				}
+			}
+			di.entry = ht.table[di.index]
+		} else {
+			di.entry = di.nextEntry
+		}
+		if di.entry != nil {
+			di.nextEntry = di.entry.next
+			return di.entry
+		}
+	}
+	return nil
+}

rdb.go

@@ -9,9 +9,22 @@ import (
 	"time"
 	"unsafe"
 
+	"github.com/BitInit/fake-redis/dict"
 	"github.com/BitInit/fake-redis/rdb"
 )
 
+const (
+	rdbOpCodeAux          = 250
+	rdbOpCodeResizeDB     = 251
+	rdbOpCodeExpireTimeMS = 252
+	rdbOpCodeSelectDB     = 254
+	rdbOpCodeEOF          = 255
+)
+
+const (
+	rdbTypeString = 0
+)
+
 const RDB_VERSION = 9
 const RDB_OPCODE_AUX = 250
 
@@ -63,13 +76,76 @@ func rdbSaveRio(r *rdb.Rdb) error {
 		return err
 	}
 
+	for i := 0; i < server.dbnum; i++ {
+		db := server.db[i]
+		saveDb(r, i, db.dict, db.expires)
+	}
+	return nil
+}
+
+func saveDb(r *rdb.Rdb, i int, kv *dict.Dict, expires *dict.Dict) error {
+	dbSize := kv.Size()
+	expiresSize := expires.Size()
+	if dbSize == 0 {
+		return nil
+	}
+	if err := r.SaveType(rdbOpCodeSelectDB); err != nil {
+		return err
+	}
+	if _, err := r.SaveLen(uint64(i)); err != nil {
+		return err
+	}
+
+	if err := r.SaveType(rdbOpCodeResizeDB); err != nil {
+		return err
+	}
+	if _, err := r.SaveLen(uint64(dbSize)); err != nil {
+		return err
+	}
+	if _, err := r.SaveLen(uint64(expiresSize)); err != nil {
+		return err
+	}
+
+	di := kv.GetSafeInterator()
+	for de := di.Next(); de != nil; de = di.Next() {
+		key := de.GetKey().([]byte)
+		val := de.GetVal().(*robj)
+		expire := expires.GetVal(key)
+
+		if expire != nil {
+			if err := r.SaveType(rdbOpCodeExpireTimeMS); err != nil {
+				return err
+			}
+			et := expire.(int64)
+			if _, err := r.SaveLen(uint64(et)); err != nil {
+				return err
+			}
+		}
+		if err := saveObjectType(r, val); err != nil {
+			return err
+		}
+		if err := r.SaveRawString(string(key)); err != nil {
+			return err
+		}
+		if err := saveObject(r, val); err != nil {
+			return err
+		}
+
+	}
+
+	if err := r.SaveType(rdbOpCodeEOF); err != nil {
+		return err
+	}
+	if err := r.SaveCheckSum(); err != nil {
+		return err
+	}
 	return nil
 }
 
 func rdbSaveInfoAuxFields(r *rdb.Rdb) error {
 	redisBits := unsafe.Sizeof(r)
 
-	if err := r.SaveAuxField("redis-ver", VERSION); err != nil {
+	if err := saveAuxField(r, "redis-ver", VERSION); err != nil {
 		return err
 	}
 	if err := rdbSaveAuxFieldStrInt(r, "redis-bits", int(redisBits)); err != nil {
@@ -83,5 +159,34 @@ func rdbSaveInfoAuxFields(r *rdb.Rdb) error {
 
 func rdbSaveAuxFieldStrInt(r *rdb.Rdb, key string, val int) error {
 	v := strconv.Itoa(val)
-	return r.SaveAuxField(key, v)
+	return saveAuxField(r, key, v)
+}
+
+func saveAuxField(r *rdb.Rdb, key string, val string) error {
+	if err := r.SaveType(rdbOpCodeAux); err != nil {
+		return err
+	}
+	if err := r.SaveRawString(key); err != nil {
+		return err
+	}
+	if err := r.SaveRawString(val); err != nil {
+		return err
+	}
+	return nil
+}
+
+func saveObjectType(r *rdb.Rdb, o *robj) error {
+	switch o.tp {
+	case OBJ_STRING:
+		return r.SaveType(rdbTypeString)
+		// other type
+	}
+	return nil
+}
+
+func saveObject(r *rdb.Rdb, val *robj) error {
+	if val.tp == OBJ_STRING {
+		return r.SaveRawString(string(val.ptr.([]byte)))
+	}
+	return fmt.Errorf("no support type")
 }

rdb/rdb.go

@@ -7,10 +7,6 @@ import (
 	"strconv"
 )
 
-const (
-	rdbOpCodeAux = 250
-)
-
 const (
 	rdbEncVal   = 3
 	rdb6BitLen  = 0
@@ -64,24 +60,33 @@ func InitWithFile(fp *os.File) *Rdb {
 	return &Rdb{
 		_rio: &fileRio{
 			fp:       fp,
+			chckSum:  0,
 			buffered: 0,
 			autosync: 0,
 		},
 	}
 }
 
 func (r *Rdb) WriteRaw(buf []byte) (int, error) {
-	return r._rio.write(buf)
+	if err := r._rio.updateCkSum(buf); err != nil {
+		return 0, err
+	}
+	n, err := r._rio.write(buf)
+	if err != nil {
+		return n, err
+	}
+
+	return n, nil
 }
 
-func (r *Rdb) saveType(tp byte) error {
+func (r *Rdb) SaveType(tp byte) error {
 	if _, err := r.WriteRaw([]byte{tp}); err != nil {
 		return err
 	}
 	return nil
 }
 
-func (r *Rdb) saveRawString(s string) error {
+func (r *Rdb) SaveRawString(s string) error {
 	l := len(s)
 	if l <= 11 {
 		buf := tryIntegerEncoding(s)
@@ -92,7 +97,7 @@ func (r *Rdb) saveRawString(s string) error {
 		}
 	}
 
-	if _, err := r.saveLen(uint64(l)); err != nil {
+	if _, err := r.SaveLen(uint64(l)); err != nil {
 		return err
 	}
 	if l > 0 {
@@ -103,7 +108,7 @@ func (r *Rdb) saveRawString(s string) error {
 	return nil
 }
 
-func (r *Rdb) saveLen(l uint64) (int, error) {
+func (r *Rdb) SaveLen(l uint64) (int, error) {
 	var buf []byte
 	if l < (1 << 6) {
 		buf = []byte{
@@ -124,14 +129,9 @@ func (r *Rdb) saveLen(l uint64) (int, error) {
 	return r.WriteRaw(buf)
 }
 
-func (r *Rdb) SaveAuxField(key string, val string) error {
-	if err := r.saveType(rdbOpCodeAux); err != nil {
-		return err
-	}
-	if err := r.saveRawString(key); err != nil {
-		return err
-	}
-	if err := r.saveRawString(val); err != nil {
+func (r *Rdb) SaveCheckSum() error {
+	chSum := binary.LittleEndian.AppendUint64([]byte{}, r._rio.getCkSum())
+	if _, err := r.WriteRaw(chSum); err != nil {
 		return err
 	}
 	return nil

rdb/rdb_test.go

@@ -12,7 +12,11 @@ type memRio struct {
 	pos int
 }
 
-func (mr *memRio) updateCkSum(buf []byte) {}
+func (mr *memRio) updateCkSum(buf []byte) error { return nil }
+
+func (mr *memRio) getCkSum() uint64 {
+	return 0
+}
 
 func (mr *memRio) read(buf []byte) int {
 	l := len(buf)
@@ -37,17 +41,17 @@ func (mr *memRio) clean() {
 func TestSaveLen(t *testing.T) {
 	mr := &memRio{pos: 0}
 	r := &Rdb{_rio: mr}
-	r.saveLen(31)
+	r.SaveLen(31)
 	assert.Equal(t, 1, len(mr.buf))
 	assert.Equal(t, int(31), int(mr.buf[0]))
 
 	mr.clean()
-	r.saveLen(64)
+	r.SaveLen(64)
 	assert.Equal(t, 2, len(mr.buf))
 	assert.Equal(t, int(1<<6), int(mr.buf[0]))
 
 	mr.clean()
-	r.saveLen(math.MaxUint32 + 1)
+	r.SaveLen(math.MaxUint32 + 1)
 	assert.Equal(t, 9, len(mr.buf))
 	assert.Equal(t, int(0x01), int(mr.buf[4]))
 }

rdb/rio.go

@@ -1,20 +1,33 @@
 package rdb
 
-import "os"
+import (
+	"os"
+
+	"github.com/BitInit/fake-redis/util"
+)
 
 type rio interface {
 	read([]byte) int
 	write([]byte) (int, error)
-	updateCkSum([]byte)
+	updateCkSum([]byte) error
+	getCkSum() uint64
 }
 
 type fileRio struct {
 	fp       *os.File
+	chckSum  uint64
 	buffered int
 	autosync int
 }
 
-func (r *fileRio) updateCkSum(buf []byte) {}
+func (r *fileRio) updateCkSum(buf []byte) error {
+	r.chckSum = util.Crc64(r.chckSum, buf)
+	return nil
+}
+
+func (r *fileRio) getCkSum() uint64 {
+	return r.chckSum
+}
 
 func (fr *fileRio) read(buf []byte) int {
 	return 0