From 0b0a666e4910740f6a22f54be4a69f3059a344c7 Mon Sep 17 00:00:00 2001 From: janson Date: Tue, 12 Feb 2019 10:00:08 +0800 Subject: [PATCH 1/7] fix: writeFrame may block forever in keepalive function, then it never timeout --- session.go | 41 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/session.go b/session.go index c29634e..71219e6 100644 --- a/session.go +++ b/session.go @@ -289,7 +289,13 @@ func (s *Session) keepalive() { for { select { case <-tickerPing.C: - s.writeFrame(newFrame(cmdNOP, 0)) + _, err := s.writeFrameWithDeadline(newFrame(cmdNOP, 0), tickerTimeout.C) + if err == errTimeout { + if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) { + s.Close() + return + } + } s.notifyBucket() // force a signal to the recvLoop case <-tickerTimeout.C: if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) { @@ -345,6 +351,35 @@ func (s *Session) writeFrame(f Frame) (n int, err error) { case s.writes <- req: } - result := <-req.result - return result.n, result.err + select { + case <-s.die: + return 0, errors.New(errBrokenPipe) + case result := <-req.result: + return result.n, result.err + } +} + +//writeFrame may block forever in keepalive function, then it never timeout +//so set a deadline to writeFrame that used in keepalive +func (s *Session) writeFrameWithDeadline(f Frame, deadline <-chan time.Time) (int, error) { + req := writeRequest{ + frame: f, + result: make(chan writeResult, 1), + } + select { + case <-s.die: + return 0, errors.New(errBrokenPipe) + case s.writes <- req: + case <-deadline: + return 0, errTimeout + } + + select { + case result := <-req.result: + return result.n, result.err + case <-deadline: + return 0, errTimeout + case <-s.die: + return 0, errors.New(errBrokenPipe) + } } From 2775ecb3f84cfdf914b2d36e6b25684ab811b2a1 Mon Sep 17 00:00:00 2001 From: janson Date: Tue, 12 Feb 2019 14:31:33 +0800 Subject: [PATCH 2/7] add TestKeepAliveBlockWriteTimeout and TestDeadlineFrame --- session.go | 4 +-- session_test.go | 77 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/session.go b/session.go index 71219e6..dd6976c 100644 --- a/session.go +++ b/session.go @@ -359,8 +359,8 @@ func (s *Session) writeFrame(f Frame) (n int, err error) { } } -//writeFrame may block forever in keepalive function, then it never timeout -//so set a deadline to writeFrame that used in keepalive +// writeFrame may block forever in keepalive function, then it never timeout +// so set a deadline to writeFrame only used in keepalive func (s *Session) writeFrameWithDeadline(f Frame, deadline <-chan time.Time) (int, error) { req := writeRequest{ frame: f, diff --git a/session_test.go b/session_test.go index 9aca780..972718c 100644 --- a/session_test.go +++ b/session_test.go @@ -278,6 +278,16 @@ func TestIsClose(t *testing.T) { } } +type blockWriteConn struct { + net.Conn +} + +func (c *blockWriteConn) Write(b []byte) (n int, err error) { + forever := time.Hour * 24 + time.Sleep(forever) + return c.Conn.Write(b) +} + func TestKeepAliveTimeout(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") if err != nil { @@ -304,6 +314,34 @@ func TestKeepAliveTimeout(t *testing.T) { } } +func TestKeepAliveBlockWriteTimeout(t *testing.T) { + ln, err := net.Listen("tcp", "localhost:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { + ln.Accept() + }() + + cli, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer cli.Close() + //when writeFrame block, keepalive in old version never timeout + blockWriteCli := &blockWriteConn{cli} + + config := DefaultConfig() + config.KeepAliveInterval = time.Second + config.KeepAliveTimeout = 2 * time.Second + session, _ := Client(blockWriteCli, config) + time.Sleep(3 * time.Second) + if !session.IsClosed() { + t.Fatal("keepalive-timeout failed") + } +} + func TestServerEcho(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") if err != nil { @@ -543,6 +581,45 @@ func TestRandomFrame(t *testing.T) { cli.Close() } +func TestDeadlineFrame(t *testing.T) { + addr, stop, cli, err := setupServer(t) + if err != nil { + t.Fatal(err) + } + defer stop() + // pure random + session, _ := Client(cli, nil) + for i := 0; i < 100; i++ { + rnd := make([]byte, rand.Uint32()%1024) + io.ReadFull(crand.Reader, rnd) + session.conn.Write(rnd) + } + cli.Close() + + // random cmds, writeFrameWithDeadline only used in keepalive with cmd of cmdNOP + cli, err = net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + allcmds := []byte{cmdSYN, cmdFIN, cmdPSH, cmdNOP} + session, _ = Client(cli, nil) + for i := 0; i < 100; i++ { + f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) + session.writeFrameWithDeadline(f, time.After(session.config.KeepAliveTimeout)) + } + //deadline occur + { + c := make(chan time.Time) + close(c) + f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) + _, err := session.writeFrameWithDeadline(f, c) + if err != errTimeout { + t.Fatal("write frame with deadline failed", err) + } + } + cli.Close() +} + func TestReadDeadline(t *testing.T) { _, stop, cli, err := setupServer(t) if err != nil { From 2c51c16a6973771f10b1942ab01688e035f9c638 Mon Sep 17 00:00:00 2001 From: janson Date: Tue, 12 Feb 2019 19:00:43 +0800 Subject: [PATCH 3/7] do better coverage --- session_test.go | 47 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 37 insertions(+), 10 deletions(-) diff --git a/session_test.go b/session_test.go index 972718c..7439b08 100644 --- a/session_test.go +++ b/session_test.go @@ -278,16 +278,6 @@ func TestIsClose(t *testing.T) { } } -type blockWriteConn struct { - net.Conn -} - -func (c *blockWriteConn) Write(b []byte) (n int, err error) { - forever := time.Hour * 24 - time.Sleep(forever) - return c.Conn.Write(b) -} - func TestKeepAliveTimeout(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") if err != nil { @@ -314,6 +304,16 @@ func TestKeepAliveTimeout(t *testing.T) { } } +type blockWriteConn struct { + net.Conn +} + +func (c *blockWriteConn) Write(b []byte) (n int, err error) { + forever := time.Hour * 24 + time.Sleep(forever) + return c.Conn.Write(b) +} + func TestKeepAliveBlockWriteTimeout(t *testing.T) { ln, err := net.Listen("tcp", "localhost:0") if err != nil { @@ -579,6 +579,19 @@ func TestRandomFrame(t *testing.T) { session.conn.Write(buf) cli.Close() + + // writeFrame after die + cli, err = net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + session, _ = Client(cli, nil) + //close first + session.Close() + for i := 0; i < 100; i++ { + f := newFrame(byte(rand.Uint32()), rand.Uint32()) + session.writeFrame(f) + } } func TestDeadlineFrame(t *testing.T) { @@ -596,6 +609,19 @@ func TestDeadlineFrame(t *testing.T) { } cli.Close() + // writeFrame after die + cli, err = net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + session, _ = Client(cli, nil) + //close first + session.Close() + for i := 0; i < 100; i++ { + f := newFrame(byte(rand.Uint32()), rand.Uint32()) + session.writeFrameWithDeadline(f, time.After(session.config.KeepAliveTimeout)) + } + // random cmds, writeFrameWithDeadline only used in keepalive with cmd of cmdNOP cli, err = net.Dial("tcp", addr) if err != nil { @@ -618,6 +644,7 @@ func TestDeadlineFrame(t *testing.T) { } } cli.Close() + } func TestReadDeadline(t *testing.T) { From d7d8337fbf7a9e2f7f9c4cf57856a4d2b3a85dbb Mon Sep 17 00:00:00 2001 From: jannson Date: Tue, 12 Feb 2019 21:48:52 +0800 Subject: [PATCH 4/7] all cover for writeFrameWithDeadline --- session_test.go | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/session_test.go b/session_test.go index 7439b08..3a96f27 100644 --- a/session_test.go +++ b/session_test.go @@ -645,6 +645,29 @@ func TestDeadlineFrame(t *testing.T) { } cli.Close() + { + cli, err = net.Dial("tcp", addr) + if err != nil { + t.Fatal(err) + } + config := DefaultConfig() + config.KeepAliveInterval = time.Second + config.KeepAliveTimeout = 2 * time.Second + session, _ = Client(&blockWriteConn{cli}, config) + f := newFrame(byte(rand.Uint32()), rand.Uint32()) + c := make(chan time.Time) + go func() { + //die first, deadline second, better for coverage + time.Sleep(time.Second) + session.Close() + time.Sleep(time.Second) + close(c) + }() + _, err = session.writeFrameWithDeadline(f, c) + if err.Error() != errBrokenPipe { + t.Fatal("write frame with deadline failed", err) + } + } } func TestReadDeadline(t *testing.T) { From 5bef4b83d60f471246e73bd8efd67da583ef18e0 Mon Sep 17 00:00:00 2001 From: jannson Date: Tue, 12 Feb 2019 23:50:01 +0800 Subject: [PATCH 5/7] calling writeFrameWithDeadline in writeFrame for better reading --- session.go | 17 +---------------- 1 file changed, 1 insertion(+), 16 deletions(-) diff --git a/session.go b/session.go index dd6976c..7d04e52 100644 --- a/session.go +++ b/session.go @@ -341,22 +341,7 @@ func (s *Session) sendLoop() { // writeFrame writes the frame to the underlying connection // and returns the number of bytes written if successful func (s *Session) writeFrame(f Frame) (n int, err error) { - req := writeRequest{ - frame: f, - result: make(chan writeResult, 1), - } - select { - case <-s.die: - return 0, errors.New(errBrokenPipe) - case s.writes <- req: - } - - select { - case <-s.die: - return 0, errors.New(errBrokenPipe) - case result := <-req.result: - return result.n, result.err - } + return s.writeFrameWithDeadline(f, nil) } // writeFrame may block forever in keepalive function, then it never timeout From 0bafab3e4427cfe5f57eee021aca53ecffdced79 Mon Sep 17 00:00:00 2001 From: jannson Date: Tue, 12 Feb 2019 23:58:45 +0800 Subject: [PATCH 6/7] use writeFrameInternal instead of writeFrameWithDeadline --- session.go | 9 ++++----- session_test.go | 12 ++++++------ 2 files changed, 10 insertions(+), 11 deletions(-) diff --git a/session.go b/session.go index 7d04e52..8e3cc1f 100644 --- a/session.go +++ b/session.go @@ -289,7 +289,7 @@ func (s *Session) keepalive() { for { select { case <-tickerPing.C: - _, err := s.writeFrameWithDeadline(newFrame(cmdNOP, 0), tickerTimeout.C) + _, err := s.writeFrameInternal(newFrame(cmdNOP, 0), tickerTimeout.C) if err == errTimeout { if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) { s.Close() @@ -341,12 +341,11 @@ func (s *Session) sendLoop() { // writeFrame writes the frame to the underlying connection // and returns the number of bytes written if successful func (s *Session) writeFrame(f Frame) (n int, err error) { - return s.writeFrameWithDeadline(f, nil) + return s.writeFrameInternal(f, nil) } -// writeFrame may block forever in keepalive function, then it never timeout -// so set a deadline to writeFrame only used in keepalive -func (s *Session) writeFrameWithDeadline(f Frame, deadline <-chan time.Time) (int, error) { +// internal writeFrame version to support deadline used in keepalive +func (s *Session) writeFrameInternal(f Frame, deadline <-chan time.Time) (int, error) { req := writeRequest{ frame: f, result: make(chan writeResult, 1), diff --git a/session_test.go b/session_test.go index 3a96f27..32fd20b 100644 --- a/session_test.go +++ b/session_test.go @@ -594,7 +594,7 @@ func TestRandomFrame(t *testing.T) { } } -func TestDeadlineFrame(t *testing.T) { +func TestWriteFrameInternal(t *testing.T) { addr, stop, cli, err := setupServer(t) if err != nil { t.Fatal(err) @@ -619,10 +619,10 @@ func TestDeadlineFrame(t *testing.T) { session.Close() for i := 0; i < 100; i++ { f := newFrame(byte(rand.Uint32()), rand.Uint32()) - session.writeFrameWithDeadline(f, time.After(session.config.KeepAliveTimeout)) + session.writeFrameInternal(f, time.After(session.config.KeepAliveTimeout)) } - // random cmds, writeFrameWithDeadline only used in keepalive with cmd of cmdNOP + // random cmds cli, err = net.Dial("tcp", addr) if err != nil { t.Fatal(err) @@ -631,14 +631,14 @@ func TestDeadlineFrame(t *testing.T) { session, _ = Client(cli, nil) for i := 0; i < 100; i++ { f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) - session.writeFrameWithDeadline(f, time.After(session.config.KeepAliveTimeout)) + session.writeFrameInternal(f, time.After(session.config.KeepAliveTimeout)) } //deadline occur { c := make(chan time.Time) close(c) f := newFrame(allcmds[rand.Int()%len(allcmds)], rand.Uint32()) - _, err := session.writeFrameWithDeadline(f, c) + _, err := session.writeFrameInternal(f, c) if err != errTimeout { t.Fatal("write frame with deadline failed", err) } @@ -663,7 +663,7 @@ func TestDeadlineFrame(t *testing.T) { time.Sleep(time.Second) close(c) }() - _, err = session.writeFrameWithDeadline(f, c) + _, err = session.writeFrameInternal(f, c) if err.Error() != errBrokenPipe { t.Fatal("write frame with deadline failed", err) } From d48c64181ce88447e0edb027e2e564379bd59ba4 Mon Sep 17 00:00:00 2001 From: jannson Date: Wed, 13 Feb 2019 00:36:40 +0800 Subject: [PATCH 7/7] no errTimeout checking for writeFrameInternal in keepalive, do it simple, but better --- session.go | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/session.go b/session.go index 8e3cc1f..d0c3a13 100644 --- a/session.go +++ b/session.go @@ -289,13 +289,7 @@ func (s *Session) keepalive() { for { select { case <-tickerPing.C: - _, err := s.writeFrameInternal(newFrame(cmdNOP, 0), tickerTimeout.C) - if err == errTimeout { - if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) { - s.Close() - return - } - } + s.writeFrameInternal(newFrame(cmdNOP, 0), tickerPing.C) s.notifyBucket() // force a signal to the recvLoop case <-tickerTimeout.C: if !atomic.CompareAndSwapInt32(&s.dataReady, 1, 0) {