fix RESP buffer read

handler.go

@@ -315,9 +315,14 @@ func client(n *Nodis, conn *redis.Conn, cmd redis.Command) {
 	execCommand(conn, func() {
 		switch strings.ToUpper(cmd.Args[0]) {
 		case "LIST":
-			conn.WriteString("id=1 addr=" + conn.Client.RemoteAddr().String() +
-				" fd=" + strconv.Itoa(conn.Fd) +
-				" name=" + conn.Name + " age=0 idle=0 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 events=r cmd=client ")
+			redis.ClientLocker.RLock()
+			conn.WriteArray(len(redis.Clients))
+			for _, c := range redis.Clients {
+				conn.WriteString("id=" + strconv.Itoa(c.Fd) + " addr=" + c.Client.RemoteAddr().String() +
+					" fd=" + strconv.Itoa(c.Fd) +
+					" name=" + c.Name + " age=0 idle=0 flags=N db=0 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 events=r cmd=client")
+			}
+			redis.ClientLocker.RUnlock()
 		case "SETNAME":
 			conn.Name = cmd.Args[1]
 			conn.WriteString("OK")

redis/resp.go

@@ -64,14 +64,15 @@ READ:
 		return err
 	}
 	r.l += rn
-	if r.l < n {
+	if rn < n {
+		n -= rn
 		goto READ
 	}
 	return nil
 }
 
-// peekBufferByte returns the byte at the specified index in the buffer.
-func (r *Reader) peekBufferByte(i int) byte {
+// peekByte returns the byte at the specified index in the buffer.
+func (r *Reader) peekByte(i int) byte {
 	var b = r.buf[r.r : r.r+r.l]
 	return b[len(b)+i-1]
 }
@@ -104,7 +105,7 @@ func (r *Reader) readLine() error {
 			r.discard()
 			return err
 		}
-		if r.l > 1 && r.peekBufferByte(0) == '\n' {
+		if r.l > 1 && r.peekByte(0) == '\n' {
 			r.l -= 2
 			break
 		}
@@ -137,7 +138,7 @@ func (r *Reader) ReadCommand() error {
 	if err != nil {
 		return err
 	}
-	if r.peekBufferByte(0) != ArrayType {
+	if r.peekByte(0) != ArrayType {
 		return r.ReadInlineCommand()
 	}
 	r.discard()
@@ -242,7 +243,7 @@ func (r *Reader) readBulk() (string, error) {
 	if err != nil {
 		return "", err
 	}
-	if r.peekBufferByte(0) != BulkType {
+	if r.peekByte(0) != BulkType {
 		return "", ErrInvalidRequestExceptedBulk
 	}
 	r.discard()
@@ -273,17 +274,17 @@ func (r *Reader) readUtil(end byte) (bool, error) {
 			return lineEnd, err
 		}
 		if end == ' ' {
-			if r.peekBufferByte(0) == '\r' {
+			if r.peekByte(0) == '\r' {
 				r.l--
 				continue
 			}
-			if r.peekBufferByte(0) == '\n' {
+			if r.peekByte(0) == '\n' {
 				lineEnd = true
 				r.l--
 				break
 			}
 		}
-		if r.peekBufferByte(0) == end && (r.l > 1 && r.peekBufferByte(-1) != '\\') {
+		if r.peekByte(0) == end && (r.l > 1 && r.peekByte(-1) != '\\') {
 			r.l--
 			break
 		}
@@ -296,7 +297,7 @@ func (r *Reader) ReadInlineCommand() error {
 	var lineEnd = false
 	var err error
 	for {
-		first := r.peekBufferByte(0)
+		first := r.peekByte(0)
 		if first == ' ' || first == '\t' || first == '\r' {
 			r.discard()
 			err = r.readByte()

redis/resp_test.go

@@ -382,3 +382,16 @@ func TestReadByteN(t *testing.T) {
 		}
 	}
 }
+
+func BenchmarkReadCommand(b *testing.B) {
+	doc := "set foo bar\r\n"
+	for i := 0; i < 1000; i++ {
+		doc += "set foo bar\r\n"
+	}
+	r := NewReader(strings.NewReader(doc))
+	b.ResetTimer()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		_ = r.ReadCommand()
+	}
+}

redis/server.go

@@ -16,7 +16,7 @@ const (
 )
 
 var (
-	clientLocker sync.RWMutex
+	ClientLocker sync.RWMutex
 	Clients      = make(map[int]*Conn)
 )
 
@@ -50,7 +50,7 @@ func Serve(addr string, handler HandlerFunc) error {
 }
 
 func handleConn(conn net.Conn, handler HandlerFunc) {
-	clientLocker.Lock()
+	ClientLocker.Lock()
 	c := &Conn{
 		Fd:       len(Clients) + 1,
 		Reader:   NewReader(conn),
@@ -59,7 +59,7 @@ func handleConn(conn net.Conn, handler HandlerFunc) {
 		Commands: make([]func(), 0),
 	}
 	Clients[c.Fd] = c
-	clientLocker.Unlock()
+	ClientLocker.Unlock()
 	for {
 		err := c.Reader.ReadCommand()
 		if err != nil {
@@ -75,7 +75,7 @@ func handleConn(conn net.Conn, handler HandlerFunc) {
 		_ = c.Push()
 	}
 	_ = conn.Close()
-	clientLocker.Lock()
+	ClientLocker.Lock()
 	delete(Clients, c.Fd)
-	clientLocker.Unlock()
+	ClientLocker.Unlock()
 }