rpc: fix two concurrency issues in client

Read operations could be dropped after reconnect because the new reader
goroutine would be started before the previous one exited. Start it later.

For short-lived connections performing a single call, the result would
sometimes be dropped because we closed the connection before waiting for
handler shutdown. Reverse shutdown order to fix this problem.
This commit is contained in:
Felix Lange 2019-01-24 13:42:25 +01:00
parent 5f7693efa6
commit a0ecea3fa7
2 changed files with 43 additions and 2 deletions

View file

@ -117,8 +117,8 @@ func (c *Client) newClientConn(conn ServerCodec) *clientConn {
} }
func (cc *clientConn) close(err error, inflightReq *requestOp) { func (cc *clientConn) close(err error, inflightReq *requestOp) {
cc.codec.Close()
cc.handler.close(err, inflightReq) cc.handler.close(err, inflightReq)
cc.codec.Close()
} }
type readOp struct { type readOp struct {
@ -558,7 +558,6 @@ func (c *Client) dispatch(codec ServerCodec) {
// Reconnect: // Reconnect:
case newcodec := <-c.reconnected: case newcodec := <-c.reconnected:
log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.RemoteAddr()) log.Debug("RPC client reconnected", "reading", reading, "conn", newcodec.RemoteAddr())
go c.read(newcodec)
if reading { if reading {
// Wait for the previous read loop to exit. This is a rare case which // Wait for the previous read loop to exit. This is a rare case which
// happens if this loop isn't notified in time after the connection breaks. // happens if this loop isn't notified in time after the connection breaks.
@ -568,6 +567,7 @@ func (c *Client) dispatch(codec ServerCodec) {
conn.close(errClientReconnected, lastOp) conn.close(errClientReconnected, lastOp)
c.drainRead() c.drainRead()
} }
go c.read(newcodec)
reading = true reading = true
conn = c.newClientConn(newcodec) conn = c.newClientConn(newcodec)
// Re-register the in-flight request on the new handler // Re-register the in-flight request on the new handler

View file

@ -18,6 +18,7 @@ package rpc
import ( import (
"bufio" "bufio"
"bytes"
"io" "io"
"io/ioutil" "io/ioutil"
"net" "net"
@ -109,3 +110,43 @@ func runTestScript(t *testing.T, file string) {
} }
} }
} }
// This test checks that responses are delivered for very short-lived connections that
// only carry a single request.
func TestServerShortLivedConn(t *testing.T) {
server := newTestServer()
defer server.Stop()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal("can't listen:", err)
}
defer listener.Close()
go server.ServeListener(listener)
var (
request = `{"jsonrpc":"2.0","id":1,"method":"rpc_modules"}` + "\n"
wantResp = `{"jsonrpc":"2.0","id":1,"result":{"nftest":"1.0","rpc":"1.0","test":"1.0"}}` + "\n"
deadline = time.Now().Add(10 * time.Second)
)
for i := 0; i < 20; i++ {
conn, err := net.Dial("tcp", listener.Addr().String())
if err != nil {
t.Fatal("can't dial:", err)
}
defer conn.Close()
conn.SetDeadline(deadline)
// Write the request, then half-close the connection so the server stops reading.
conn.Write([]byte(request))
conn.(*net.TCPConn).CloseWrite()
// Now try to get the response.
buf := make([]byte, 2000)
n, err := conn.Read(buf)
if err != nil {
t.Fatal("read error:", err)
}
if !bytes.Equal(buf[:n], []byte(wantResp)) {
t.Fatalf("wrong response: %s", buf[:n])
}
}
}