ledisdb/ledis/dump.go

202 lines
3.6 KiB
Go
Raw Normal View History

2014-05-28 10:20:45 +04:00
package ledis
import (
"bufio"
"bytes"
"encoding/binary"
2014-07-04 13:55:47 +04:00
"github.com/siddontang/go-snappy/snappy"
2014-06-19 13:19:40 +04:00
"github.com/siddontang/ledisdb/leveldb"
2014-05-28 10:20:45 +04:00
"io"
"os"
)
//dump format
2014-06-09 13:23:32 +04:00
// fileIndex(bigendian int64)|filePos(bigendian int64)
2014-05-28 10:20:45 +04:00
// |keylen(bigendian int32)|key|valuelen(bigendian int32)|value......
2014-07-04 13:55:47 +04:00
//
//key and value are both compressed for fast transfer dump on network using snappy
2014-07-09 05:33:44 +04:00
2014-06-09 13:23:32 +04:00
type MasterInfo struct {
LogFileIndex int64
LogPos int64
}
func (m *MasterInfo) WriteTo(w io.Writer) error {
if err := binary.Write(w, binary.BigEndian, m.LogFileIndex); err != nil {
return err
}
if err := binary.Write(w, binary.BigEndian, m.LogPos); err != nil {
return err
}
return nil
}
func (m *MasterInfo) ReadFrom(r io.Reader) error {
err := binary.Read(r, binary.BigEndian, &m.LogFileIndex)
if err != nil {
return err
}
err = binary.Read(r, binary.BigEndian, &m.LogPos)
if err != nil {
return err
}
return nil
2014-05-28 10:20:45 +04:00
}
func (l *Ledis) DumpFile(path string) error {
f, err := os.Create(path)
if err != nil {
return err
}
defer f.Close()
return l.Dump(f)
}
func (l *Ledis) Dump(w io.Writer) error {
var sp *leveldb.Snapshot
2014-06-09 13:23:32 +04:00
var m *MasterInfo = new(MasterInfo)
2014-05-28 10:20:45 +04:00
if l.binlog == nil {
sp = l.ldb.NewSnapshot()
} else {
2014-05-29 11:07:14 +04:00
l.Lock()
2014-05-28 10:20:45 +04:00
sp = l.ldb.NewSnapshot()
2014-06-09 13:23:32 +04:00
m.LogFileIndex = l.binlog.LogFileIndex()
m.LogPos = l.binlog.LogFilePos()
2014-05-29 11:07:14 +04:00
l.Unlock()
2014-05-28 10:20:45 +04:00
}
2014-06-09 13:23:32 +04:00
var err error
2014-05-28 10:20:45 +04:00
wb := bufio.NewWriterSize(w, 4096)
2014-06-09 13:23:32 +04:00
if err = m.WriteTo(wb); err != nil {
2014-05-28 10:20:45 +04:00
return err
}
2014-06-19 13:19:40 +04:00
it := sp.NewIterator()
it.SeekToFirst()
2014-07-04 13:55:47 +04:00
compressBuf := make([]byte, 4096)
2014-05-28 10:20:45 +04:00
var key []byte
var value []byte
for ; it.Valid(); it.Next() {
key = it.Key()
value = it.Value()
2014-07-04 13:55:47 +04:00
if key, err = snappy.Encode(compressBuf, key); err != nil {
return err
}
2014-05-28 10:20:45 +04:00
if err = binary.Write(wb, binary.BigEndian, uint16(len(key))); err != nil {
return err
}
if _, err = wb.Write(key); err != nil {
return err
}
2014-07-04 13:55:47 +04:00
if value, err = snappy.Encode(compressBuf, value); err != nil {
return err
}
2014-05-28 10:20:45 +04:00
if err = binary.Write(wb, binary.BigEndian, uint32(len(value))); err != nil {
return err
}
if _, err = wb.Write(value); err != nil {
return err
}
}
if err = wb.Flush(); err != nil {
return err
}
2014-07-04 13:55:47 +04:00
compressBuf = nil
2014-05-28 10:20:45 +04:00
return nil
}
2014-06-09 13:23:32 +04:00
func (l *Ledis) LoadDumpFile(path string) (*MasterInfo, error) {
2014-05-28 10:20:45 +04:00
f, err := os.Open(path)
if err != nil {
return nil, err
}
defer f.Close()
return l.LoadDump(f)
}
2014-06-09 13:23:32 +04:00
func (l *Ledis) LoadDump(r io.Reader) (*MasterInfo, error) {
2014-05-29 11:07:14 +04:00
l.Lock()
defer l.Unlock()
2014-06-09 13:23:32 +04:00
info := new(MasterInfo)
2014-05-28 10:20:45 +04:00
rb := bufio.NewReaderSize(r, 4096)
2014-06-09 13:23:32 +04:00
err := info.ReadFrom(rb)
2014-05-28 10:20:45 +04:00
if err != nil {
return nil, err
}
var keyLen uint16
var valueLen uint32
var keyBuf bytes.Buffer
var valueBuf bytes.Buffer
2014-07-04 13:55:47 +04:00
deKeyBuf := make([]byte, 4096)
deValueBuf := make([]byte, 4096)
var key, value []byte
2014-05-28 10:20:45 +04:00
for {
if err = binary.Read(rb, binary.BigEndian, &keyLen); err != nil && err != io.EOF {
return nil, err
} else if err == io.EOF {
break
}
if _, err = io.CopyN(&keyBuf, rb, int64(keyLen)); err != nil {
return nil, err
}
2014-07-04 13:55:47 +04:00
if key, err = snappy.Decode(deKeyBuf, keyBuf.Bytes()); err != nil {
return nil, err
}
2014-05-28 10:20:45 +04:00
if err = binary.Read(rb, binary.BigEndian, &valueLen); err != nil {
return nil, err
}
if _, err = io.CopyN(&valueBuf, rb, int64(valueLen)); err != nil {
return nil, err
}
2014-07-04 13:55:47 +04:00
if value, err = snappy.Decode(deValueBuf, valueBuf.Bytes()); err != nil {
return nil, err
}
if err = l.ldb.Put(key, value); err != nil {
2014-05-28 10:20:45 +04:00
return nil, err
}
2014-05-29 11:07:14 +04:00
if l.binlog != nil {
2014-07-04 13:55:47 +04:00
err = l.binlog.Log(encodeBinLogPut(key, value))
2014-05-29 11:07:14 +04:00
}
2014-05-28 10:20:45 +04:00
keyBuf.Reset()
valueBuf.Reset()
}
2014-07-04 13:55:47 +04:00
deKeyBuf = nil
deValueBuf = nil
2014-06-09 13:23:32 +04:00
return info, nil
2014-05-28 10:20:45 +04:00
}