feat: set buff_send as default

internal/client/redis.go

@@ -8,8 +8,6 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"sync"
-	"sync/atomic"
 	"time"
 
 	"RedisShake/internal/client/proto"
@@ -22,13 +20,6 @@ type Redis struct {
 	writer      *bufio.Writer
 	protoReader *proto.Reader
 	protoWriter *proto.Writer
-	timer       *time.Timer
-	sendBytes   uint64
-	mu          sync.Mutex
-}
-
-func NewSentinelMasterClient(ctx context.Context, address string, username string, password string, Tls bool) *Redis {
-	return NewRedisClient(ctx, address, username, password, Tls, false)
 }
 
 func NewRedisClient(ctx context.Context, address string, username string, password string, Tls bool, replica bool) *Redis {
@@ -56,7 +47,7 @@ func NewRedisClient(ctx context.Context, address string, username string, passwo
 
 	r.conn = conn
 	r.reader = bufio.NewReader(conn)
-	r.writer = bufio.NewWriter(conn)
+	r.writer = bufio.NewWriterSize(conn, 16*1024*1024) // size is 16MB
 	r.protoReader = proto.NewReader(r.reader)
 	r.protoWriter = proto.NewWriter(r.writer)
 
@@ -86,9 +77,6 @@ func NewRedisClient(ctx context.Context, address string, username string, passwo
 		r = NewRedisClient(ctx, replicaInfo.BestReplica, username, password, Tls, false)
 	}
 
-	r.timer = time.NewTimer(time.Second)
-	go r.autoFlush(ctx)
-
 	return r
 }
 
@@ -182,71 +170,22 @@ func (r *Redis) Send(args ...interface{}) {
 	if err != nil {
 		log.Panicf(err.Error())
 	}
-	r.flush()
-}
-
-func (r *Redis) SendBytes(buf []byte) {
-	_, err := r.writer.Write(buf)
-	if err != nil {
-		log.Panicf(err.Error())
-	}
-	r.flush()
+	r.Flush()
 }
 
+// SendBytesBuff send bytes to buffer, need to call Flush() to send the buffer
 func (r *Redis) SendBytesBuff(buf []byte) {
-	r.mu.Lock()
-	defer r.mu.Unlock()
 	_, err := r.writer.Write(buf)
 	if err != nil {
 		log.Panicf(err.Error())
 	}
-	r.flushBuff(len(buf))
-}
-
-func (r *Redis) resetTimer() {
-	if !r.timer.Stop() {
-		select {
-		case <-r.timer.C:
-		default:
-		}
-	}
-	r.timer.Reset(time.Second)
-}
-
-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() {
+func (r *Redis) Flush() {
 	err := r.writer.Flush()
 	if err != nil {
 		log.Panicf(err.Error())
 	}
-	atomic.StoreUint64(&r.sendBytes, 0)
-}
-
-func (r *Redis) autoFlush(ctx context.Context) {
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-r.timer.C:
-			if atomic.LoadUint64(&r.sendBytes) > 0 {
-				r.mu.Lock()
-				err := r.writer.Flush()
-				r.mu.Unlock()
-				if err != nil {
-					log.Panicf(err.Error())
-				}
-			}
-		}
-	}
 }
 
 func (r *Redis) Receive() (interface{}, error) {
@@ -285,13 +224,6 @@ func (r *Redis) Close() {
 	if err := r.conn.Close(); err != nil {
 		log.Infof("close redis conn err: %s\n", err.Error())
 	}
-	// release the timer
-	if !r.timer.Stop() {
-		select {
-		case <-r.timer.C:
-		default:
-		}
-	}
 }
 
 /* Commands */

internal/writer/redis_standalone_writer.go

@@ -24,7 +24,6 @@ type RedisWriterOptions struct {
 	Password string                 `mapstructure:"password" default:""`
 	Tls      bool                   `mapstructure:"tls" default:"false"`
 	OffReply bool                   `mapstructure:"off_reply" default:"false"`
-	BuffSend bool                   `mapstructure:"buff_send" default:"false"`
 	Sentinel client.SentinelOptions `mapstructure:"sentinel"`
 }
 
@@ -39,8 +38,6 @@ type redisStandaloneWriter struct {
 	ch          chan *entry.Entry
 	chWg        sync.WaitGroup
 
-	buffSend bool
-
 	stat struct {
 		Name              string `json:"name"`
 		UnansweredBytes   int64  `json:"unanswered_bytes"`
@@ -54,7 +51,6 @@ func NewRedisStandaloneWriter(ctx context.Context, opts *RedisWriterOptions) Wri
 	rw.stat.Name = "writer_" + strings.Replace(opts.Address, ":", "_", -1)
 	rw.client = client.NewRedisClient(ctx, opts.Address, opts.Username, opts.Password, opts.Tls, false)
 	rw.ch = make(chan *entry.Entry, 1024)
-	rw.buffSend = opts.BuffSend
 	if opts.OffReply {
 		log.Infof("turn off the reply of write")
 		rw.offReply = true
@@ -79,31 +75,38 @@ 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)
 	go func() {
-		for e := range w.ch {
-			// switch db if we need
-			if w.DbId != e.DbId {
-				w.switchDbTo(e.DbId)
-			}
-			// send
-			bytes := e.Serialize()
-			for e.SerializedSize+atomic.LoadInt64(&w.stat.UnansweredBytes) > config.Opt.Advanced.TargetRedisClientMaxQuerybufLen {
-				time.Sleep(1 * time.Nanosecond)
-			}
-			log.Debugf("[%s] send cmd. cmd=[%s]", w.stat.Name, e.String())
-			if !w.offReply {
-				w.chWaitReply <- e
-				atomic.AddInt64(&w.stat.UnansweredBytes, e.SerializedSize)
-				atomic.AddInt64(&w.stat.UnansweredEntries, 1)
-			}
-			if w.buffSend {
+		for {
+			select {
+			case <-ctx.Done():
+				// do nothing until w.ch is closed
+			case <-timer.C:
+				w.client.Flush()
+			case e, ok := <-w.ch:
+				if !ok {
+					w.client.Flush()
+					w.chWg.Done()
+					return
+				}
+				// switch db if we need
+				if w.DbId != e.DbId {
+					w.switchDbTo(e.DbId)
+				}
+				// send
+				bytes := e.Serialize()
+				for e.SerializedSize+atomic.LoadInt64(&w.stat.UnansweredBytes) > config.Opt.Advanced.TargetRedisClientMaxQuerybufLen {
+					time.Sleep(1 * time.Nanosecond)
+				}
+				log.Debugf("[%s] send cmd. cmd=[%s]", w.stat.Name, e.String())
+				if !w.offReply {
+					w.chWaitReply <- e
+					atomic.AddInt64(&w.stat.UnansweredBytes, e.SerializedSize)
+					atomic.AddInt64(&w.stat.UnansweredEntries, 1)
+				}
 				w.client.SendBytesBuff(bytes)
-			} else {
-				w.client.SendBytes(bytes)
 			}
-
 		}
-		w.chWg.Done()
 	}()
 
 	return w.ch

shake.toml

@@ -34,7 +34,6 @@ username = ""              # keep empty if not using ACL
 password = ""              # keep empty if no authentication is required
 tls = false
 off_reply = false          # turn off the server reply
-buff_send = false          # buffer send, default false. may be a sync delay when true, but it can greatly improve the speed
 
 [filter]
 # Allow keys with specific prefixes or suffixes