Merge remote-tracking branch 'origin/main' into leak-stacks

Author: zenhack

answer.go

@@ -156,28 +156,29 @@ func (p *Promise) Resolve(r Ptr, e error) {
 		}
 		p.caller = nil
 
-		if len(p.clients) > 0 || p.ongoingCalls > 0 {
-			// Pending resolution state: wait for clients to be fulfilled
-			// and calls to have answers.  p.clients cannot be touched in the
-			// pending resolution state, so we have exclusive access to the
-			// variable.
-			if p.ongoingCalls > 0 {
-				p.callsStopped = make(chan struct{})
-			}
-			syncutil.Without(&p.mu, func() {
-				res := resolution{p.method, r, e}
-				for path, cp := range p.clients {
-					t := path.transform()
-					cp.promise.fulfill(res.client(t))
-					shutdownPromises = append(shutdownPromises, cp.promise)
-					cp.promise = nil
-				}
-				if p.callsStopped != nil {
-					<-p.callsStopped
-				}
-			})
+		if p.ongoingCalls > 0 {
+			p.callsStopped = make(chan struct{})
 		}
+	})
 
+	if len(p.clients) > 0 || p.ongoingCalls > 0 {
+		// Pending resolution state: wait for clients to be fulfilled
+		// and calls to have answers.  p.clients cannot be touched in the
+		// pending resolution state, so we have exclusive access to the
+		// variable.
+		res := resolution{p.method, r, e}
+		for path, cp := range p.clients {
+			t := path.transform()
+			cp.promise.fulfill(res.client(t))
+			shutdownPromises = append(shutdownPromises, cp.promise)
+			cp.promise = nil
+		}
+		if p.callsStopped != nil {
+			<-p.callsStopped
+		}
+	}
+
+	syncutil.With(&p.mu, func() {
 		// Move p into resolved state.
 		p.callsStopped = nil
 		p.result, p.err = r, e

rpc/answer.go

@@ -30,11 +30,11 @@ type answer struct {
 	// entry is a placeholder until the remote vat cancels the call.
 	ret rpccp.Return
 
-	// sendMsg sends the return message.  The caller MUST NOT hold ans.c.mu.
+	// sendMsg sends the return message.  The caller MUST NOT hold ans.c.lk.
 	sendMsg func()
 
 	// msgReleaser releases the return message when its refcount hits zero.
-	// The caller MUST NOT hold ans.c.mu.
+	// The caller MUST NOT hold ans.c.lk.
 	msgReleaser *rc.Releaser
 
 	// results is the memoized answer to ret.Results().
@@ -99,27 +99,27 @@ func errorAnswer(c *Conn, id answerID, err error) *answer {
 // all references to it are dropped; the caller is responsible for one reference. This will not
 // happen before the message is sent, as the returned send function retains a reference.
 func (c *Conn) newReturn(ctx context.Context) (_ rpccp.Return, sendMsg func(), _ *rc.Releaser, _ error) {
-	msg, send, releaseMsg, err := c.transport.NewMessage()
+	outMsg, err := c.transport.NewMessage()
 	if err != nil {
 		return rpccp.Return{}, nil, nil, rpcerr.Failedf("create return: %w", err)
 	}
-	ret, err := msg.NewReturn()
+	ret, err := outMsg.Message.NewReturn()
 	if err != nil {
-		releaseMsg()
+		outMsg.Release()
 		return rpccp.Return{}, nil, nil, rpcerr.Failedf("create return: %w", err)
 	}
 
 	// Before releasing the message, we need to wait both until it is sent and
 	// until the local vat is done with it.  We therefore implement a simple
 	// ref-counting mechanism such that 'release' must be called twice before
 	// 'releaseMsg' is called.
-	releaser := rc.NewReleaser(2, releaseMsg)
+	releaser := rc.NewReleaser(2, outMsg.Release)
 
 	return ret, func() {
 		c.sender.Send(asyncSend{
-			send:    send,
+			send:    outMsg.Send,
 			release: releaser.Decr,
-			callback: func(err error) {
+			onSent: func(err error) {
 				if err != nil {
 					c.er.ReportError(fmt.Errorf("send return: %w", err))
 				}
@@ -129,11 +129,11 @@ func (c *Conn) newReturn(ctx context.Context) (_ rpccp.Return, sendMsg func(), _
 }
 
 // setPipelineCaller sets ans.pcall to pcall if the answer has not
-// already returned.  The caller MUST NOT hold ans.c.mu.
+// already returned.  The caller MUST NOT hold ans.c.lk.
 //
 // This also sets ans.promise to a new promise, wrapping pcall.
 func (ans *answer) setPipelineCaller(m capnp.Method, pcall capnp.PipelineCaller) {
-	syncutil.With(&ans.c.mu, func() {
+	syncutil.With(&ans.c.lk, func() {
 		if !ans.flags.Contains(resultsReady) {
 			ans.pcall = pcall
 			ans.promise = capnp.NewPromise(m, pcall)
@@ -181,12 +181,12 @@ func (ans *answer) setBootstrap(c capnp.Client) error {
 
 // Return sends the return message.
 //
-// The caller MUST NOT hold ans.c.mu.
+// The caller MUST NOT hold ans.c.lk.
 func (ans *answer) Return(e error) {
-	ans.c.mu.Lock()
+	ans.c.lk.Lock()
 	if e != nil {
 		rl := ans.sendException(e)
-		ans.c.mu.Unlock()
+		ans.c.lk.Unlock()
 		rl.release()
 		ans.pcalls.Wait()
 		ans.c.tasks.Done() // added by handleCall
@@ -202,13 +202,13 @@ func (ans *answer) Return(e error) {
 				ans.c.er.ReportError(err)
 			}
 
-			ans.c.mu.Unlock()
+			ans.c.lk.Unlock()
 			rl.release()
 			ans.pcalls.Wait()
 			return
 		}
 	}
-	ans.c.mu.Unlock()
+	ans.c.lk.Unlock()
 	rl.release()
 	ans.pcalls.Wait()
 	ans.c.tasks.Done() // added by handleCall
@@ -219,7 +219,7 @@ func (ans *answer) Return(e error) {
 // Finish with releaseResultCaps set to true, then sendReturn returns
 // the number of references to be subtracted from each export.
 //
-// The caller MUST be holding onto ans.c.mu. sendReturn MUST NOT be
+// The caller MUST be holding onto ans.c.lk. sendReturn MUST NOT be
 // called if sendException was previously called.
 func (ans *answer) sendReturn() (releaseList, error) {
 	ans.pcall = nil
@@ -252,13 +252,13 @@ func (ans *answer) sendReturn() (releaseList, error) {
 			}
 			ans.promise = nil
 		}
-		ans.c.mu.Unlock()
+		ans.c.lk.Unlock()
 		ans.sendMsg()
 		if fin {
-			ans.c.mu.Lock()
+			ans.c.lk.Lock()
 			return ans.destroy()
 		}
-		ans.c.mu.Lock()
+		ans.c.lk.Lock()
 	}
 	ans.flags |= returnSent
 	if !ans.flags.Contains(finishReceived) {
@@ -269,7 +269,7 @@ func (ans *answer) sendReturn() (releaseList, error) {
 
 // sendException sends an exception on the answer's return message.
 //
-// The caller MUST be holding onto ans.c.mu. sendException MUST NOT
+// The caller MUST be holding onto ans.c.lk. sendException MUST NOT
 // be called if sendReturn was previously called.
 func (ans *answer) sendException(ex error) releaseList {
 	ans.err = ex
@@ -286,7 +286,7 @@ func (ans *answer) sendException(ex error) releaseList {
 	default:
 		// Send exception.
 		fin := ans.flags.Contains(finishReceived)
-		ans.c.mu.Unlock()
+		ans.c.lk.Unlock()
 		if e, err := ans.ret.NewException(); err != nil {
 			ans.c.er.ReportError(fmt.Errorf("send exception: %w", err))
 		} else {
@@ -298,11 +298,11 @@ func (ans *answer) sendException(ex error) releaseList {
 			}
 		}
 		if fin {
-			ans.c.mu.Lock()
+			ans.c.lk.Lock()
 			rl, _ := ans.destroy()
 			return rl
 		}
-		ans.c.mu.Lock()
+		ans.c.lk.Lock()
 	}
 	ans.flags |= returnSent
 	if !ans.flags.Contains(finishReceived) {
@@ -314,17 +314,19 @@ func (ans *answer) sendException(ex error) releaseList {
 	return rl
 }
 
-// destroy removes the answer from the table and returns the clients to
-// release.  The answer must have sent a return and received a finish.
-// The caller must be holding onto ans.c.mu.
+// destroy removes the answer from the table and returns ReleaseFuncs to
+// run. The answer must have sent a return and received a finish.
+// The caller must be holding onto ans.c.lk.
 //
 // shutdown has its own strategy for cleaning up an answer.
 func (ans *answer) destroy() (releaseList, error) {
-	defer syncutil.Without(&ans.c.mu, ans.msgReleaser.Decr)
-	delete(ans.c.answers, ans.id)
+	rl := releaseList{ans.msgReleaser.Decr}
+	delete(ans.c.lk.answers, ans.id)
 	if !ans.flags.Contains(releaseResultCapsFlag) || len(ans.exportRefs) == 0 {
-		return nil, nil
+		return rl, nil
 
 	}
-	return ans.c.releaseExportRefs(ans.exportRefs)
+	ret, err := ans.c.releaseExportRefs(ans.exportRefs)
+	ret = append(ret, rl...)
+	return ret, err
 }

rpc/export.go

@@ -41,12 +41,12 @@ func (c *Conn) clearExportID(m *capnp.Metadata) {
 }
 
 // findExport returns the export entry with the given ID or nil if
-// couldn't be found.
+// couldn't be found. The caller must be holding c.mu
 func (c *Conn) findExport(id exportID) *expent {
-	if int64(id) >= int64(len(c.exports)) {
+	if int64(id) >= int64(len(c.lk.exports)) {
 		return nil
 	}
-	return c.exports[id] // might be nil
+	return c.lk.exports[id] // might be nil
 }
 
 // releaseExport decreases the number of wire references to an export
@@ -63,8 +63,8 @@ func (c *Conn) releaseExport(id exportID, count uint32) (capnp.Client, error) {
 	switch {
 	case count == ent.wireRefs:
 		client := ent.client
-		c.exports[id] = nil
-		c.exportID.remove(uint32(id))
+		c.lk.exports[id] = nil
+		c.lk.exportID.remove(uint32(id))
 		metadata := client.State().Metadata
 		syncutil.With(metadata, func() {
 			c.clearExportID(metadata)
@@ -98,7 +98,7 @@ func (c *Conn) releaseExportRefs(refs map[exportID]uint32) (releaseList, error)
 		if rl == nil {
 			rl = make(releaseList, 0, n)
 		}
-		rl = append(rl, client)
+		rl = append(rl, client.Release)
 		n--
 	}
 	return rl, firstErr
@@ -116,7 +116,7 @@ func (c *Conn) sendCap(d rpccp.CapDescriptor, client capnp.Client) (_ exportID,
 	state := client.State()
 	bv := state.Brand.Value
 	if ic, ok := bv.(*importClient); ok && ic.c == c {
-		if ent := c.imports[ic.id]; ent != nil && ent.generation == ic.generation {
+		if ent := c.lk.imports[ic.id]; ent != nil && ent.generation == ic.generation {
 			d.SetReceiverHosted(uint32(ic.id))
 			return 0, false, nil
 		}
@@ -149,7 +149,7 @@ func (c *Conn) sendCap(d rpccp.CapDescriptor, client capnp.Client) (_ exportID,
 	defer state.Metadata.Unlock()
 	id, ok := c.findExportID(state.Metadata)
 	if ok {
-		ent := c.exports[id]
+		ent := c.lk.exports[id]
 		ent.wireRefs++
 		d.SetSenderHosted(uint32(id))
 		return id, true, nil
@@ -160,11 +160,11 @@ func (c *Conn) sendCap(d rpccp.CapDescriptor, client capnp.Client) (_ exportID,
 		client:   client.AddRef(),
 		wireRefs: 1,
 	}
-	id = exportID(c.exportID.next())
-	if int64(id) == int64(len(c.exports)) {
-		c.exports = append(c.exports, ee)
+	id = exportID(c.lk.exportID.next())
+	if int64(id) == int64(len(c.lk.exports)) {
+		c.lk.exports = append(c.lk.exports, ee)
 	} else {
-		c.exports[id] = ee
+		c.lk.exports[id] = ee
 	}
 	c.setExportID(state.Metadata, id)
 	d.SetSenderHosted(uint32(id))
@@ -217,28 +217,28 @@ type embargo struct {
 //
 // The caller must be holding onto c.mu.
 func (c *Conn) embargo(client capnp.Client) (embargoID, capnp.Client) {
-	id := embargoID(c.embargoID.next())
+	id := embargoID(c.lk.embargoID.next())
 	e := &embargo{
 		c:      client,
 		lifted: make(chan struct{}),
 	}
-	if int64(id) == int64(len(c.embargoes)) {
-		c.embargoes = append(c.embargoes, e)
+	if int64(id) == int64(len(c.lk.embargoes)) {
+		c.lk.embargoes = append(c.lk.embargoes, e)
 	} else {
-		c.embargoes[id] = e
+		c.lk.embargoes[id] = e
 	}
 	var c2 capnp.Client
-	c2, c.embargoes[id].p = capnp.NewPromisedClient(c.embargoes[id])
+	c2, c.lk.embargoes[id].p = capnp.NewPromisedClient(c.lk.embargoes[id])
 	return id, c2
 }
 
 // findEmbargo returns the embargo entry with the given ID or nil if
-// couldn't be found.
+// couldn't be found. Must be holding c.mu
 func (c *Conn) findEmbargo(id embargoID) *embargo {
-	if int64(id) >= int64(len(c.embargoes)) {
+	if int64(id) >= int64(len(c.lk.embargoes)) {
 		return nil
 	}
-	return c.embargoes[id] // might be nil
+	return c.lk.embargoes[id] // might be nil
 }
 
 // lift disembargoes the client.  It must be called only once.
@@ -308,13 +308,13 @@ func (sl *senderLoopback) buildDisembargo(msg rpccp.Message) error {
 	return nil
 }
 
-type releaseList []capnp.Client
+type releaseList []capnp.ReleaseFunc
 
 func (rl releaseList) release() {
-	for _, c := range rl {
-		c.Release()
+	for _, r := range rl {
+		r()
 	}
 	for i := range rl {
-		rl[i] = capnp.Client{}
+		rl[i] = func() {}
 	}
 }