rpc: add buffering to the encode path.

This reduces the number of writes by 2 (1 client, 1 server) on each round trip.
A simple test shows 24% higher throughput.

R=rsc
CC=golang-dev
https://golang.org/cl/4279057

2 files changed, 31 insertions, 16 deletions

diff --git a/src/pkg/rpc/client.go b/src/pkg/rpc/client.go
index 6de6d1325b..94bbe83e40 100644
--- a/src/pkg/rpc/client.go
+++ b/src/pkg/rpc/client.go
@@ -148,8 +148,12 @@ func (call *Call) done() {
 
 // NewClient returns a new Client to handle requests to the
 // set of services at the other end of the connection.
+// It adds a buffer to the write side of the connection so
+// the header and payload are sent as a unit.
 func NewClient(conn io.ReadWriteCloser) *Client {
-	return NewClientWithCodec(&gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(conn)})
+	encBuf := bufio.NewWriter(conn)
+	client := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf}
+	return NewClientWithCodec(client)
 }
 
 // NewClientWithCodec is like NewClient but uses the specified
@@ -164,16 +168,20 @@ func NewClientWithCodec(codec ClientCodec) *Client {
 }
 
 type gobClientCodec struct {
-	rwc io.ReadWriteCloser
-	dec *gob.Decoder
-	enc *gob.Encoder
+	rwc    io.ReadWriteCloser
+	dec    *gob.Decoder
+	enc    *gob.Encoder
+	encBuf *bufio.Writer
 }
 
-func (c *gobClientCodec) WriteRequest(r *Request, body interface{}) os.Error {
-	if err := c.enc.Encode(r); err != nil {
-		return err
+func (c *gobClientCodec) WriteRequest(r *Request, body interface{}) (err os.Error) {
+	if err = c.enc.Encode(r); err != nil {
+		return
+	}
+	if err = c.enc.Encode(body); err != nil {
+		return
 	}
-	return c.enc.Encode(body)
+	return c.encBuf.Flush()
 }
 
 func (c *gobClientCodec) ReadResponseHeader(r *Response) os.Error {
diff --git a/src/pkg/rpc/server.go b/src/pkg/rpc/server.go
index 59ebaf4a80..d75db308b0 100644
--- a/src/pkg/rpc/server.go
+++ b/src/pkg/rpc/server.go
@@ -110,6 +110,7 @@
 package rpc
 
 import (
+	"bufio"
 	"gob"
 	"http"
 	"log"
@@ -336,9 +337,10 @@ func (s *service) call(sending *sync.Mutex, mtype *methodType, req *Request, arg
 }
 
 type gobServerCodec struct {
-	rwc io.ReadWriteCloser
-	dec *gob.Decoder
-	enc *gob.Encoder
+	rwc    io.ReadWriteCloser
+	dec    *gob.Decoder
+	enc    *gob.Encoder
+	encBuf *bufio.Writer
 }
 
 func (c *gobServerCodec) ReadRequestHeader(r *Request) os.Error {
@@ -349,11 +351,14 @@ func (c *gobServerCodec) ReadRequestBody(body interface{}) os.Error {
 	return c.dec.Decode(body)
 }
 
-func (c *gobServerCodec) WriteResponse(r *Response, body interface{}) os.Error {
-	if err := c.enc.Encode(r); err != nil {
-		return err
+func (c *gobServerCodec) WriteResponse(r *Response, body interface{}) (err os.Error) {
+	if err = c.enc.Encode(r); err != nil {
+		return
+	}
+	if err = c.enc.Encode(body); err != nil {
+		return
 	}
-	return c.enc.Encode(body)
+	return c.encBuf.Flush()
 }
 
 func (c *gobServerCodec) Close() os.Error {
@@ -367,7 +372,9 @@ func (c *gobServerCodec) Close() os.Error {
 // ServeConn uses the gob wire format (see package gob) on the
 // connection.  To use an alternate codec, use ServeCodec.
 func (server *Server) ServeConn(conn io.ReadWriteCloser) {
-	server.ServeCodec(&gobServerCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(conn)})
+	buf := bufio.NewWriter(conn)
+	srv := &gobServerCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(buf), buf}
+	server.ServeCodec(srv)
 }
 
 // ServeCodec is like ServeConn but uses the specified codec to