Fix RPC deadlock

go.mod

@@ -6,4 +6,5 @@ require (
 	github.com/kylelemons/godebug v1.1.0
 	github.com/stretchr/testify v1.7.0 // indirect
 	github.com/tinylib/msgp v1.1.5
+	golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9
 )

go.sum

@@ -21,6 +21,7 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9 h1:SQFwaSi55rU7vdNs9Yr0Z324VNlrF+0wMqRXT4St8ck=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

internal/errors/errors.go

@@ -32,7 +32,7 @@ func (e *capnpError) GoString() string {
 // The returned error's type will match err's type.
 func Annotate(prefix, msg string, err error) error {
 	if err == nil {
-		panic("Annotate on nil error")
+		return nil
 	}
 	ce, ok := err.(*capnpError)
 	if !ok {

internal/errors/errors_test.go

@@ -3,6 +3,8 @@ package errors
 import (
 	"errors"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestErrorString(t *testing.T) {
@@ -52,6 +54,11 @@ func TestAnnotate(t *testing.T) {
 		want     string
 		wantType Type
 	}{
+		{
+			prefix: "prefix",
+			msg:    "context",
+			err:    nil,
+		},
 		{
 			msg:      "context",
 			err:      errors.New("goofed"),
@@ -113,6 +120,11 @@ func TestAnnotate(t *testing.T) {
 	}
 	for _, test := range tests {
 		got := Annotate(test.prefix, test.msg, test.err)
+		if test.err == nil {
+			assert.Nil(t, got)
+			continue
+		}
+
 		if got.Error() != test.want {
 			t.Errorf("Annotate(%q, %q, %#v).Error() = %q; %q", test.prefix, test.msg, test.err, got.Error(), test.want)
 		}

internal/mpsc/mpsc_test.go

@@ -22,7 +22,8 @@ func TestRecvEmpty(t *testing.T) {
 	q := New()
 
 	// Recv() on an empty queue should block until the context is canceled.
-	ctx, _ := context.WithTimeout(context.Background(), time.Millisecond*10)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*10)
+	defer cancel()
 
 	v, err := q.Recv(ctx)
 	assert.Equal(t, ctx.Err(), err, "Returned error is not ctx.Err()")

rpc/answer.go

@@ -181,7 +181,7 @@ func (ans *answer) Return(e error) {
 		default:
 			ans.c.tasks.Done() // added by handleCall
 			if err := ans.c.shutdown(err); err != nil {
-				ans.c.report(err)
+				ans.c.er.ReportError(err)
 			}
 			// shutdown released c.mu
 			rl.release()
@@ -210,7 +210,7 @@ func (ans *answer) sendReturn(cstates []capnp.ClientState) (releaseList, error)
 	var err error
 	ans.exportRefs, err = ans.c.fillPayloadCapTable(ans.results, ans.resultCapTable, cstates)
 	if err != nil {
-		ans.c.report(annotate(err).errorf("send return"))
+		ans.c.er.annotatef(err, "send return")
 		// Continue.  Don't fail to send return if cap table isn't fully filled.
 	}
 
@@ -220,7 +220,7 @@ func (ans *answer) sendReturn(cstates []capnp.ClientState) (releaseList, error)
 		fin := ans.flags&finishReceived != 0
 		ans.c.mu.Unlock()
 		if err := ans.sendMsg(); err != nil {
-			ans.c.reportf("send return: %v", err)
+			ans.c.er.reportf("send return: %v", err)
 		}
 		if fin {
 			ans.releaseMsg()
@@ -258,13 +258,13 @@ func (ans *answer) sendException(e error) releaseList {
 		fin := ans.flags&finishReceived != 0
 		ans.c.mu.Unlock()
 		if exc, err := ans.ret.NewException(); err != nil {
-			ans.c.reportf("send exception: %v", err)
+			ans.c.er.reportf("send exception: %v", err)
 		} else {
 			exc.SetType(rpccp.Exception_Type(errors.TypeOf(e)))
 			if err := exc.SetReason(e.Error()); err != nil {
-				ans.c.reportf("send exception: %v", err)
+				ans.c.er.reportf("send exception: %v", err)
 			} else if err := ans.sendMsg(); err != nil {
-				ans.c.reportf("send return: %v", err)
+				ans.c.er.reportf("send return: %v", err)
 			}
 		}
 		if fin {

rpc/import.go

@@ -255,30 +255,29 @@ func (ic *importClient) Brand() capnp.Brand {
 
 func (ic *importClient) Shutdown() {
 	ic.c.mu.Lock()
+	defer ic.c.mu.Unlock()
+
 	if !ic.c.startTask() {
-		ic.c.mu.Unlock()
 		return
 	}
 	defer ic.c.tasks.Done()
+
 	ent := ic.c.imports[ic.id]
 	if ic.generation != ent.generation {
 		// A new reference was added concurrently with the Shutdown.  See
 		// impent.generation documentation for an explanation.
-		ic.c.mu.Unlock()
 		return
 	}
+
 	delete(ic.c.imports, ic.id)
-	err := ic.c.sendMessage(context.Background(), func(msg rpccp.Message) error {
-		rel, err := msg.NewRelease()
-		if err != nil {
+
+	ic.c.sendq.SendAsync(context.Background(),
+		prepFunc(func(m rpccp.Message) error {
+			rel, err := m.NewRelease()
+			if err == nil {
+				rel.SetId(uint32(ic.id))
+				rel.SetReferenceCount(uint32(ent.wireRefs))
+			}
 			return err
-		}
-		rel.SetId(uint32(ic.id))
-		rel.SetReferenceCount(uint32(ent.wireRefs))
-		return nil
-	})
-	ic.c.mu.Unlock()
-	if err != nil {
-		ic.c.report(annotate(err).errorf("send release"))
-	}
+		}), ic.c.er.annotater("send releases"))
 }

rpc/question.go

@@ -75,39 +75,42 @@ func (q *question) handleCancel(ctx context.Context) {
 	}
 
 	q.c.mu.Lock()
+	defer q.c.mu.Unlock()
+
+	// Promise already fulfilled?
 	if q.flags&finished != 0 {
-		// Promise already fulfilled.
-		q.c.mu.Unlock()
 		return
 	}
+
 	q.flags |= finished
 	q.release = func() {}
-	err := q.c.sendMessage(q.c.bgctx, func(msg rpccp.Message) error {
-		fin, err := msg.NewFinish()
-		if err != nil {
-			return err
+	q.c.sendq.SendAsync(q.c.bgctx, prepFunc(func(m rpccp.Message) error {
+		fin, err := m.NewFinish()
+		if err == nil {
+			fin.SetQuestionId(uint32(q.id))
+			fin.SetReleaseResultCaps(true)
 		}
-		fin.SetQuestionId(uint32(q.id))
-		fin.SetReleaseResultCaps(true)
-		return nil
-	})
-	if err == nil {
-		q.flags |= finishSent
-	} else {
-		select {
-		case <-q.c.bgctx.Done():
-		default:
-			q.c.report(annotate(err).errorf("send finish"))
+		return err
+	}), errorReporterFunc(func(err error) {
+		// q.c.mu will be held by c.sendMessage when callback is invoked
+		if err == nil {
+			q.flags |= finishSent
+		} else {
+			select {
+			case <-q.c.bgctx.Done():
+			default:
+				q.c.er.annotatef(err, "send finish")
+			}
 		}
-	}
-	close(q.finishMsgSend)
-	q.c.mu.Unlock()
+		close(q.finishMsgSend)
+		q.c.mu.Unlock()
 
-	q.p.Reject(rejectErr)
-	if q.bootstrapPromise != nil {
-		q.bootstrapPromise.Fulfill(q.p.Answer().Client())
-		q.p.ReleaseClients()
-	}
+		q.p.Reject(rejectErr)
+		if q.bootstrapPromise != nil {
+			q.bootstrapPromise.Fulfill(q.p.Answer().Client())
+			q.p.ReleaseClients()
+		}
+	}))
 }
 
 func (q *question) PipelineSend(ctx context.Context, transform []capnp.PipelineOp, s capnp.Send) (*capnp.Answer, capnp.ReleaseFunc) {