feat: reduce latency for reading AOF and flushing commands to destination

internal/reader/sync_standalone_reader.go

@@ -464,7 +464,7 @@ func (r *syncStandaloneReader) sendAOF(offset int64) {
 		iArgv, err := protoReader.ReadReply()
 		if err != nil {
 			if err == io.EOF {
-				time.Sleep(100 * time.Millisecond)
+				time.Sleep(10 * time.Millisecond)
 				continue
 			} else {
 				log.Panicf("[%s] read aof file failed. error=[%v]", r.stat.Name, err)

internal/writer/redis_standalone_writer.go

@@ -75,7 +75,7 @@ func (w *redisStandaloneWriter) Close() {
 func (w *redisStandaloneWriter) StartWrite(ctx context.Context) chan *entry.Entry {
 	w.chWg = sync.WaitGroup{}
 	w.chWg.Add(1)
-	timer := time.NewTicker(100 * time.Millisecond)
+	timer := time.NewTicker(10 * time.Millisecond)
 	go func() {
 		for {
 			select {

tests/cases/function.py

@@ -54,7 +54,12 @@ def split_mset_to_set():
     shake = h.Shake(opts)
     src.do("mset", "k1", "v1", "k2", "v2", "k3", "v3")
     # wait sync done
-    p.ASSERT_TRUE_TIMEOUT(lambda: shake.is_consistent(), timeout=10, interval=0.01)
+    try:
+        p.ASSERT_TRUE_TIMEOUT(lambda: shake.is_consistent(), timeout=10, interval=0.01)
+    except Exception as e:
+        with open(f"{shake.dir}/data/shake.log") as f:
+            p.log(f.read())
+        raise e
     dst.do("select", 1)
     p.ASSERT_EQ(dst.do("get", "k1"), b"v1")
     p.ASSERT_EQ(dst.do("get", "k2"), b"v2")