fix: resolve timeout and performance issues in buffer write mode and AOF writes

internal/client/redis.go

@@ -22,9 +22,9 @@ type Redis struct {
 	writer      *bufio.Writer
 	protoReader *proto.Reader
 	protoWriter *proto.Writer
-	timer     *time.Timer
-	sendCount uint64
-	mu        sync.Mutex
+	timer       *time.Timer
+	sendBytes   uint64
+	mu          sync.Mutex
 }
 
 func NewSentinelMasterClient(ctx context.Context, address string, username string, password string, Tls bool) *Redis {
@@ -200,29 +200,35 @@ func (r *Redis) SendBytesBuff(buf []byte) {
 	if err != nil {
 		log.Panicf(err.Error())
 	}
-	r.flushBuff()
+	r.flushBuff(len(buf))
 }
 
-func (r *Redis) flushBuff() {
-	if atomic.AddUint64(&r.sendCount, 1)%100 != 0 {
-		return
-	}
+func (r *Redis) resetTimer() {
 	if !r.timer.Stop() {
 		select {
 		case <-r.timer.C:
 		default:
 		}
 	}
 	r.timer.Reset(time.Second)
-	r.flush()
+}
+
+func (r *Redis) flushBuff(l int) {
+	// if the data size is too small, no need to flush
+	if atomic.AddUint64(&r.sendBytes, uint64(l)) > 64*1024 {
+		r.flush()
+		r.resetTimer()
+		return
+	}
+	r.resetTimer()
 }
 
 func (r *Redis) flush() {
 	err := r.writer.Flush()
 	if err != nil {
 		log.Panicf(err.Error())
 	}
-	atomic.StoreUint64(&r.sendCount, 0)
+	atomic.StoreUint64(&r.sendBytes, 0)
 }
 
 func (r *Redis) autoFlush(ctx context.Context) {
@@ -231,7 +237,7 @@ func (r *Redis) autoFlush(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-r.timer.C:
-			if atomic.LoadUint64(&r.sendCount) > 0 {
+			if atomic.LoadUint64(&r.sendBytes) > 0 {
 				r.mu.Lock()
 				err := r.writer.Flush()
 				r.mu.Unlock()

internal/reader/sync_standalone_reader.go

@@ -80,7 +80,8 @@ type syncStandaloneReader struct {
 	}
 
 	// version info
-	SupportPSYNC  bool
+	SupportPSYNC bool
+	isDiskless   bool
 }
 
 func NewSyncStandaloneReader(ctx context.Context, opts *SyncReaderOptions) Reader {
@@ -94,34 +95,33 @@ func NewSyncStandaloneReader(ctx context.Context, opts *SyncReaderOptions) Reade
 	r.stat.Dir = utils.GetAbsPath(r.stat.Name)
 	utils.CreateEmptyDir(r.stat.Dir)
 
-	r.SupportPSYNC = r.supportPSYNC();
+	r.SupportPSYNC = r.supportPSYNC()
 	return r
 }
 
-
 func (r *syncStandaloneReader) supportPSYNC() bool {
 	reply := r.client.DoWithStringReply("info", "server")
 	for _, line := range strings.Split(reply, "\n") {
 		if strings.HasPrefix(line, "redis_version:") {
 			version := strings.Split(line, ":")[1]
-			parts := strings.Split(version,".");
-			if len(parts) > 2{
-				v1,_ := strconv.Atoi(parts[0]);
-				v2,_ := strconv.Atoi(parts[1]);
-				if v1 * 1000  + v2 < 2008{
+			parts := strings.Split(version, ".")
+			if len(parts) > 2 {
+				v1, _ := strconv.Atoi(parts[0])
+				v2, _ := strconv.Atoi(parts[1])
+				if v1*1000+v2 < 2008 {
 					return false
 				}
 			}
 
 		}
 	}
 
-	return true;
+	return true
 }
 
 func (r *syncStandaloneReader) StartRead(ctx context.Context) []chan *entry.Entry {
-	if !r.SupportPSYNC{
-		return r.StartReadWithSync(ctx);
+	if !r.SupportPSYNC {
+		return r.StartReadWithSync(ctx)
 	}
 	r.ctx = ctx
 	r.ch = make(chan *entry.Entry, 1024)
@@ -206,6 +206,8 @@ func (r *syncStandaloneReader) sendPSync() {
 		reply := r.client.DoWithStringReply(argv...)
 		if reply != "OK" {
 			log.Warnf("[%s] send replconf capa eof to redis server failed. reply=[%v]", r.stat.Name, reply)
+		} else {
+			r.isDiskless = true
 		}
 	}
 	r.checkBgsaveInProgress()
@@ -518,6 +520,12 @@ func (r *syncStandaloneReader) StatusString() string {
 	if r.stat.Status == kSyncAof {
 		return fmt.Sprintf("%s, diff=[%v]", r.stat.Status, -r.stat.AofSentOffset+r.stat.AofReceivedOffset)
 	}
+	if r.stat.Status == kReceiveRdb {
+		if r.isDiskless {
+			return fmt.Sprintf("%s diskless, size=[%s]", r.stat.Status, r.stat.RdbReceivedHuman)
+		}
+		return fmt.Sprintf("%s, size=[%s/%s]", r.stat.Status, r.stat.RdbReceivedHuman, r.stat.RdbFileSizeHuman)
+	}
 	return string(r.stat.Status)
 }
 

internal/utils/file_rotate/aof_writer.go

@@ -50,10 +50,6 @@ func (w *AOFWriter) Write(buf []byte) {
 		w.Close()
 		w.openFile(w.offset)
 	}
-	err = w.file.Sync()
-	if err != nil {
-		log.Panicf(err.Error())
-	}
 }
 
 func (w *AOFWriter) Close() {