fix: fix "read aof file failed" error (#917)

This commit is contained in:
suxb201 2025-01-14 19:03:13 +08:00 committed by GitHub
parent 0046b8439e
commit 45ce483ab6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 25 additions and 13 deletions

View File

@ -4,6 +4,7 @@ import (
"bufio"
"bytes"
"context"
"errors"
"encoding/json"
"fmt"
"io"
@ -477,7 +478,7 @@ func (r *syncStandaloneReader) sendRDB(rdbFilePath string) {
}
func (r *syncStandaloneReader) sendAOF(offset int64) {
aofReader := rotate.NewAOFReader(r.stat.Name, r.stat.Dir, offset)
aofReader := rotate.NewAOFReader(r.ctx, r.stat.Name, r.stat.Dir, offset)
defer aofReader.Close()
protoReader := proto.NewReader(bufio.NewReader(aofReader))
for {
@ -488,12 +489,10 @@ func (r *syncStandaloneReader) sendAOF(offset int64) {
iArgv, err := protoReader.ReadReply()
if err != nil {
if err == io.EOF {
time.Sleep(10 * time.Millisecond)
continue
} else {
log.Panicf("[%s] read aof file failed. error=[%v]", r.stat.Name, err)
if errors.Is(err, context.Canceled) {
return
}
log.Panicf("[%s] read aof file failed. error=[%v]", r.stat.Name, err)
}
argv := client.ArrayString(iArgv, nil)

View File

@ -3,6 +3,7 @@ package rotate
import (
"RedisShake/internal/log"
"RedisShake/internal/utils"
"context"
"fmt"
"io"
"os"
@ -10,6 +11,7 @@ import (
)
type AOFReader struct {
ctx context.Context
name string
dir string
file *os.File
@ -18,8 +20,9 @@ type AOFReader struct {
filepath string
}
func NewAOFReader(name string, dir string, offset int64) *AOFReader {
func NewAOFReader(ctx context.Context, name string, dir string, offset int64) *AOFReader {
r := new(AOFReader)
r.ctx = ctx
r.name = name
r.dir = dir
@ -69,17 +72,27 @@ func (r *AOFReader) readNextFile(offset int64) bool {
func (r *AOFReader) Read(buf []byte) (n int, err error) {
n, err = r.file.Read(buf)
if err == io.EOF {
if !r.readNextFile(r.offset) {
return n, io.EOF
for err == io.EOF {
// sleep or context
timer := time.NewTimer(1 * time.Millisecond)
select {
case <-r.ctx.Done():
return n, r.ctx.Err()
case <-timer.C:
}
r.readNextFile(r.offset) // try to read next file
_, err = r.file.Seek(0, 1)
if err != nil {
log.Panicf(err.Error())
}
n, err = r.file.Read(buf)
if err != nil {
return n, err
if err == nil {
break
} else if err == io.EOF {
continue
} else {
log.Panicf("[%s] read file failed. filename=[%s], err=[%v]", r.name, r.filepath, err)
}
}
if err != nil {

View File

@ -90,7 +90,7 @@ func (w *fileWriter) processWrite(ctx context.Context) {
defer ticker.Stop()
file, err := os.Create(w.path)
if err != nil {
log.Panicf("create file failed:", err)
log.Panicf("create file failed: %v", err)
return
}
defer file.Close()