From e7ad65dd4e80da36fe4499df020aabcc02827861 Mon Sep 17 00:00:00 2001 From: Ian Denhardt Date: Tue, 20 Dec 2022 05:11:06 -0500 Subject: [PATCH 1/2] Clean up locking in handleReturn This is a little less convoluted, and avoids some uses of syncutil.Without. No functional change. --- rpc/rpc.go | 44 +++++++++++++++++--------------------------- 1 file changed, 17 insertions(+), 27 deletions(-) diff --git a/rpc/rpc.go b/rpc/rpc.go index 2c672925..d31fe5d3 100644 --- a/rpc/rpc.go +++ b/rpc/rpc.go @@ -978,37 +978,27 @@ func (c *Conn) handleReturn(ctx context.Context, ret rpccp.Return, release capnp c.er.ReportError(rpcerr.Annotate(pr.err, "incoming return")) } + if q.bootstrapPromise == nil && pr.err == nil { + // The result of the message contains actual data (not just a + // client or an error), so we save the ReleaseFunc for later: + q.release = release + } + c.lk.Unlock() // We're going to potentially block fulfilling some promises so fork // off a goroutine to avoid blocking the receive loop. - // - // TODO(cleanup): This is a bit weird in that we hold the lock across - // the go statement, and do the unlock in the new goroutine, but before - // we actually block. This was less weird when the go statement wasn't - // there, and we should rework this so it's easier to understand what's - // going on. go func() { - switch { - case q.bootstrapPromise != nil: - syncutil.Without(&c.lk, func() { - q.p.Resolve(pr.result, pr.err) - q.bootstrapPromise.Fulfill(q.p.Answer().Client()) - q.p.ReleaseClients() - release() - }) - case pr.err != nil: - // TODO(someday): send unimplemented message back to remote if - // pr.unimplemented == true. - syncutil.Without(&c.lk, func() { - q.p.Reject(pr.err) - release() - }) - default: - q.release = release - syncutil.Without(&c.lk, func() { - q.p.Fulfill(pr.result) - }) + q.p.Resolve(pr.result, pr.err) + if q.bootstrapPromise != nil { + q.bootstrapPromise.Fulfill(q.p.Answer().Client()) + q.p.ReleaseClients() + // We can release now; root pointer of the result is a client, so the + // message won't be accessed: + release() + } else if pr.err != nil { + // We can release now; the result is an error, so data from the message + // won't be accessed: + release() } - c.lk.Unlock() // Send disembargoes. Failing to send one of these just never lifts // the embargo on our side, but doesn't cause a leak. From eeb78623e1c6513420188d96d45713d788f7e740 Mon Sep 17 00:00:00 2001 From: Ian Denhardt Date: Tue, 20 Dec 2022 05:13:11 -0500 Subject: [PATCH 2/2] Hold c.lk while calling sendMessage This is a requirement per the contract, but we weren't doing it here. --- rpc/rpc.go | 60 ++++++++++++++++++++++++++++-------------------------- 1 file changed, 31 insertions(+), 29 deletions(-) diff --git a/rpc/rpc.go b/rpc/rpc.go index d31fe5d3..72805baa 100644 --- a/rpc/rpc.go +++ b/rpc/rpc.go @@ -1000,39 +1000,41 @@ func (c *Conn) handleReturn(ctx context.Context, ret rpccp.Return, release capnp release() } - // Send disembargoes. Failing to send one of these just never lifts - // the embargo on our side, but doesn't cause a leak. - // - // TODO(soon): make embargo resolve to error client. - for _, s := range pr.disembargoes { - c.sendMessage(ctx, s.buildDisembargo, func(err error) { + syncutil.With(&c.lk, func() { + // Send disembargoes. Failing to send one of these just never lifts + // the embargo on our side, but doesn't cause a leak. + // + // TODO(soon): make embargo resolve to error client. + for _, s := range pr.disembargoes { + c.sendMessage(ctx, s.buildDisembargo, func(err error) { + if err != nil { + err = fmt.Errorf("incoming return: send disembargo: %w", err) + c.er.ReportError(err) + } + }) + } + + // Send finish. + c.sendMessage(ctx, func(m rpccp.Message) error { + fin, err := m.NewFinish() + if err == nil { + fin.SetQuestionId(uint32(qid)) + fin.SetReleaseResultCaps(false) + } + return err + }, func(err error) { + c.lk.Lock() + defer c.lk.Unlock() + defer close(q.finishMsgSend) + if err != nil { - err = fmt.Errorf("incoming return: send disembargo: %w", err) + err = fmt.Errorf("incoming return: send finish: build message: %w", err) c.er.ReportError(err) + } else { + q.flags |= finishSent + c.lk.questionID.remove(uint32(qid)) } }) - } - - // Send finish. - c.sendMessage(ctx, func(m rpccp.Message) error { - fin, err := m.NewFinish() - if err == nil { - fin.SetQuestionId(uint32(qid)) - fin.SetReleaseResultCaps(false) - } - return err - }, func(err error) { - c.lk.Lock() - defer c.lk.Unlock() - defer close(q.finishMsgSend) - - if err != nil { - err = fmt.Errorf("incoming return: send finish: build message: %w", err) - c.er.ReportError(err) - } else { - q.flags |= finishSent - c.lk.questionID.remove(uint32(qid)) - } }) }()