codec: fix data race when reading/writing (*rpcCodec).cls variable.

Only access (*rpcCodec).cls variable via isClosed() and Close() methods.
Within those methods, use a RWMutex to guard access.

Fixes #84

codec/msgpack.go

@@ -781,7 +781,7 @@ func (c *msgpackSpecRpcCodec) ReadRequestBody(body interface{}) error {
 
 func (c *msgpackSpecRpcCodec) parseCustomHeader(expectTypeByte byte, msgid *uint64, methodOrError *string) (err error) {
 
-	if c.cls {
+	if c.isClosed() {
 		return io.EOF
 	}
 

codec/rpc.go

@@ -43,8 +43,10 @@ type rpcCodec struct {
 	bw  *bufio.Writer
 	br  *bufio.Reader
 	mu  sync.Mutex
-	cls bool
 	h   Handle
+
+	cls   bool
+	clsmu sync.RWMutex
 }
 
 func newRPCCodec(conn io.ReadWriteCloser, h Handle) rpcCodec {
@@ -69,7 +71,7 @@ func (c *rpcCodec) BufferedWriter() *bufio.Writer {
 }
 
 func (c *rpcCodec) write(obj1, obj2 interface{}, writeObj2, doFlush bool) (err error) {
-	if c.cls {
+	if c.isClosed() {
 		return io.EOF
 	}
 	if err = c.enc.Encode(obj1); err != nil {
@@ -94,7 +96,7 @@ func (c *rpcCodec) write(obj1, obj2 interface{}, writeObj2, doFlush bool) (err e
 }
 
 func (c *rpcCodec) read(obj interface{}) (err error) {
-	if c.cls {
+	if c.isClosed() {
 		return io.EOF
 	}
 	//If nil is passed in, we should still attempt to read content to nowhere.
@@ -105,11 +107,20 @@ func (c *rpcCodec) read(obj interface{}) (err error) {
 	return c.dec.Decode(obj)
 }
 
+func (c *rpcCodec) isClosed() bool {
+	c.clsmu.RLock()
+	x := c.cls
+	c.clsmu.RUnlock()
+	return x
+}
+
 func (c *rpcCodec) Close() error {
-	if c.cls {
+	if c.isClosed() {
 		return io.EOF
 	}
+	c.clsmu.Lock()
 	c.cls = true
+	c.clsmu.Unlock()
 	return c.rwc.Close()
 }