mirror
/
redis
mirror of https://github.com/go-redis/redis.git
1
0
Fork 1
Code Issues Packages Projects Releases Wiki Activity

Separate read and write buffers for PubSub.

This commit is contained in:
Vladimir Mihailenco 2017-02-19 09:42:45 +02:00
parent a40c8d17fe
commit 6b6f5ca133
5 changed files with 12 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/pool/conn.go
View File

@ -21,12 +21,11 @@ type Conn struct {
} }
func NewConn(netConn net.Conn) *Conn { func NewConn(netConn net.Conn) *Conn {
buf := make([]byte, 4096)
cn := &Conn{ cn := &Conn{
netConn: netConn, netConn: netConn,
Wb: proto.NewWriteBuffer(buf), Wb: proto.NewWriteBuffer(),
} }
cn.Rd = proto.NewReader(cn.netConn, buf) cn.Rd = proto.NewReader(cn.netConn)
cn.SetUsedAt(time.Now()) cn.SetUsedAt(time.Now())
return cn return cn
} }

4
internal/proto/reader.go
View File

@ -26,10 +26,10 @@ type Reader struct {
buf []byte buf []byte
} }
func NewReader(rd io.Reader, buf []byte) *Reader { func NewReader(rd io.Reader) *Reader {
return &Reader{ return &Reader{
src: bufio.NewReader(rd), src: bufio.NewReader(rd),
buf: buf, buf: make([]byte, 4096),
} }
} }

8
internal/proto/reader_test.go
View File

@ -14,18 +14,18 @@ import (
var _ = Describe("Reader", func() { var _ = Describe("Reader", func() {
It("should read n bytes", func() { It("should read n bytes", func() {
data, err := proto.NewReader(strings.NewReader("ABCDEFGHIJKLMNO"), nil).ReadN(10) data, err := proto.NewReader(strings.NewReader("ABCDEFGHIJKLMNO")).ReadN(10)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
Expect(len(data)).To(Equal(10)) Expect(len(data)).To(Equal(10))
Expect(string(data)).To(Equal("ABCDEFGHIJ")) Expect(string(data)).To(Equal("ABCDEFGHIJ"))
data, err = proto.NewReader(strings.NewReader(strings.Repeat("x", 8192)), nil).ReadN(6000) data, err = proto.NewReader(strings.NewReader(strings.Repeat("x", 8192))).ReadN(6000)
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
Expect(len(data)).To(Equal(6000)) Expect(len(data)).To(Equal(6000))
}) })
It("should read lines", func() { It("should read lines", func() {
p := proto.NewReader(strings.NewReader("$5\r\nhello\r\n"), nil) p := proto.NewReader(strings.NewReader("$5\r\nhello\r\n"))
data, err := p.ReadLine() data, err := p.ReadLine()
Expect(err).NotTo(HaveOccurred()) Expect(err).NotTo(HaveOccurred())
@ -63,7 +63,7 @@ func benchmarkParseReply(b *testing.B, reply string, m proto.MultiBulkParse, wan
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {
buf.WriteString(reply) buf.WriteString(reply)
} }
p := proto.NewReader(buf, nil) p := proto.NewReader(buf)
b.ResetTimer() b.ResetTimer()
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {

4
internal/proto/write_buffer.go
View File

@ -12,9 +12,9 @@ type WriteBuffer struct {
b []byte b []byte
} }
func NewWriteBuffer(b []byte) *WriteBuffer { func NewWriteBuffer() *WriteBuffer {
return &WriteBuffer{ return &WriteBuffer{
b: b, b: make([]byte, 0, 4096),
} }
} }

4
internal/proto/write_buffer_test.go
View File

@ -14,7 +14,7 @@ var _ = Describe("WriteBuffer", func() {
var buf *proto.WriteBuffer var buf *proto.WriteBuffer
BeforeEach(func() { BeforeEach(func() {
buf = proto.NewWriteBuffer(nil) buf = proto.NewWriteBuffer()
}) })
It("should reset", func() { It("should reset", func() {
@ -53,7 +53,7 @@ var _ = Describe("WriteBuffer", func() {
}) })
func BenchmarkWriteBuffer_Append(b *testing.B) { func BenchmarkWriteBuffer_Append(b *testing.B) {
buf := proto.NewWriteBuffer(nil) buf := proto.NewWriteBuffer()
args := []interface{}{"hello", "world", "foo", "bar"} args := []interface{}{"hello", "world", "foo", "bar"}
for i := 0; i < b.N; i++ { for i := 0; i < b.N; i++ {