apps/redis-go
0
0
Fork 0
mirror of https://github.com/HDT3213/godis.git synced 2025-10-13 12:33:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

bug fix: aof rewrite

Browse Source
This commit is contained in:
hdt3213
2020-04-14 21:10:24 +08:00
parent 670638a279
commit ad4928a2a8
2 changed files with 18 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/db/aof.go
View File

@@ -60,7 +60,7 @@ func trim(msg []byte) string {
} }
// read aof file // read aof file
func (db *DB) loadAof() { func (db *DB) loadAof(maxBytes int) {
file, err := os.Open(db.aofFilename) file, err := os.Open(db.aofFilename)
if err != nil { if err != nil {
if _, ok := err.(*os.PathError); ok { if _, ok := err.(*os.PathError); ok {
@@ -78,7 +78,11 @@ func (db *DB) loadAof() {
var args [][]byte var args [][]byte
processing := false processing := false
var msg []byte var msg []byte
readBytes := 0
for { for {
if maxBytes != 0 && readBytes >= maxBytes {
break
}
if fixedLen == 0 { if fixedLen == 0 {
msg, err = reader.ReadBytes('\n') msg, err = reader.ReadBytes('\n')
if err == io.EOF { if err == io.EOF {
@@ -88,9 +92,10 @@ func (db *DB) loadAof() {
logger.Warn("invalid format: line should end with \\r\\n") logger.Warn("invalid format: line should end with \\r\\n")
return return
} }
readBytes += len(msg)
} else { } else {
msg = make([]byte, fixedLen+2) msg = make([]byte, fixedLen+2)
_, err = io.ReadFull(reader, msg) n, err := io.ReadFull(reader, msg)
if err == io.EOF { if err == io.EOF {
return return
} }
@@ -99,6 +104,7 @@ func (db *DB) loadAof() {
return return
} }
fixedLen = 0 fixedLen = 0
readBytes += n
} }
if err != nil { if err != nil {
logger.Warn(err) logger.Warn(err)
@@ -161,7 +167,7 @@ func (db *DB) loadAof() {
/*-- aof rewrite --*/ /*-- aof rewrite --*/
func (db *DB) aofRewrite() { func (db *DB) aofRewrite() {
file, err := db.startRewrite() file, fileSize, err := db.startRewrite()
if err != nil { if err != nil {
logger.Warn(err) logger.Warn(err)
return return
@@ -176,7 +182,7 @@ func (db *DB) aofRewrite() {
aofFilename: db.aofFilename, aofFilename: db.aofFilename,
} }
tmpDB.loadAof() tmpDB.loadAof(int(fileSize))
// rewrite aof file // rewrite aof file
tmpDB.Data.ForEach(func(key string, raw interface{}) bool { tmpDB.Data.ForEach(func(key string, raw interface{}) bool {
@@ -285,20 +291,24 @@ func persistZSet(key string, zset *SortedSet.SortedSet) *reply.MultiBulkReply {
return reply.MakeMultiBulkReply(args) return reply.MakeMultiBulkReply(args)
} }
func (db *DB) startRewrite() (*os.File, error) { func (db *DB) startRewrite() (*os.File, int64, error) {
db.pausingAof.Lock() // pausing aof db.pausingAof.Lock() // pausing aof
defer db.pausingAof.Unlock() defer db.pausingAof.Unlock()
// create rewrite channel // create rewrite channel
db.aofRewriteChan = make(chan *reply.MultiBulkReply, aofQueueSize) db.aofRewriteChan = make(chan *reply.MultiBulkReply, aofQueueSize)
// get current aof file size
fileInfo, _ := os.Stat(db.aofFilename)
filesize := fileInfo.Size()
// create tmp file // create tmp file
file, err := ioutil.TempFile("", "aof") file, err := ioutil.TempFile("", "aof")
if err != nil { if err != nil {
logger.Warn("tmp file create failed") logger.Warn("tmp file create failed")
return nil, err return nil, 0, err
} }
return file, nil return file, filesize, nil
} }
func (db *DB) finishRewrite(tmpFile *os.File) { func (db *DB) finishRewrite(tmpFile *os.File) {

2
src/db/db.go
View File

@@ -82,7 +82,7 @@ func MakeDB() *DB {
if config.Properties.AppendOnly { if config.Properties.AppendOnly {
db.aofFilename = config.Properties.AppendFilename db.aofFilename = config.Properties.AppendFilename
db.loadAof() db.loadAof(0)
aofFile, err := os.OpenFile(db.aofFilename, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600) aofFile, err := os.OpenFile(db.aofFilename, os.O_APPEND|os.O_CREATE|os.O_RDWR, 0600)
if err != nil { if err != nil {
logger.Warn(err) logger.Warn(err)