ugorji/go — pull request #27 by johnniedoe (2013-12-15 19:32:59+00:00)
Title: Use the ioDecReader when reading the RPC Array header byte as well, to allow pipelined/async requests to function better.

It looks like the initial array byte in a messagepacked rpc request is currently read directly from the connection, while the rest of the message is read through a buffered reader, for efficiency. When multiple requests are pipelined together in a series of async requests though, the header byte doesn't seem to get read properly in the pipelined requests.

This fix uses the ioDecReader for the header byte as well, allowing pipelined requests to function again.

sample client/server which triggers the condition:
=============================================

package main

import (
"github.com/ugorji/go/codec"
"log"
"net"
"net/rpc"
"os"
"time"
)

type F struct{}

func (t *F) Square(a int, reply *int) error {
time.Sleep(1e9)
*reply = a * a
return nil
}

func main() {
var mh codec.MsgpackHandle

if len(os.Args) > 1 && os.Args[1] == "s" { // run server

testServer := &F{}
rpc.Register(testServer)

listener, err := net.Listen("tcp", "127.0.0.1:9090")
if err != nil {
panic(err)
}
defer listener.Close()

for {
conn, err := listener.Accept()
if err != nil {
panic(err)
}
rpcCodec := codec.MsgpackSpecRpc.ServerCodec(conn, &mh)
go rpc.ServeCodec(rpcCodec)
}
} else { // run client

conn, err := net.Dial("tcp", "localhost:9090")
if err != nil {
log.Fatal("dialing:", err)
}
client := rpc.NewClientWithCodec(codec.MsgpackSpecRpc.ClientCodec(conn, &mh))

// with delay, client succeeds
makeAsyncCalls(client, 3, 1e7)
// without, client fails
makeAsyncCalls(client, 10, 0)

}
}

func makeAsyncCalls(client *rpc.Client, numCalls int, delay time.Duration) {
replies := make([]int, numCalls)
calls := make([]*rpc.Call, numCalls)
for i, _ := range replies {
calls[i] = client.Go("F.Square", i, &replies[i], nil)
time.Sleep(delay)
}
for i, _ := range replies {
done := <-calls[i].Done
log.Printf("Reply(%v), Channel Value(%v)\n", replies[i], done)
}
}