Merge remote-tracking branch 'origin/main' into rc-pkg

Author: zenhack

answer.go

@@ -53,17 +53,6 @@ type promiseState struct {
 	// the promise leaves the unresolved state.
 	caller PipelineCaller
 
-	// ongoingCalls counts the number of calls to caller that have not
-	// yielded an Answer yet (but not necessarily finished).
-	ongoingCalls int
-	// If callsStopped is non-nil, then the promise has entered into
-	// the pending state and is waiting for ongoingCalls to drop to zero.
-	// After decrementing ongoingCalls, callsStopped should be closed if
-	// ongoingCalls is zero to wake up the goroutine.
-	//
-	// Only Fulfill or Reject will set callsStopped.
-	callsStopped chan struct{}
-
 	// clients is a table of promised clients created to proxy the eventual
 	// result's clients.  Even after resolution, this table may still have
 	// entries until the clients are released. Cannot be read or written
@@ -161,30 +150,17 @@ func (p *Promise) Resolve(r Ptr, e error) {
 	dq := &deferred.Queue{}
 	defer dq.Run()
 
-	var (
-		// We need to access some of these fields from p.state while
-		// not holding the lock, so we store them here while holding it.
-		// p.clients cannot be touched in the pending resolution state,
-		// so we have exclusive access to the variable anyway.
-		clients      map[clientPath]*clientAndPromise
-		callsStopped chan struct{}
-	)
-
-	p.state.With(func(p *promiseState) {
+	// It's ok to extract p.clients and use it while not holding the lock:
+	// it may not be accessed in the pending resolution state, so we have
+	// exclusive access to the variable anyway.
+	clients := mutex.With1(&p.state, func(p *promiseState) map[clientPath]*clientAndPromise {
 		if e != nil {
 			p.requireUnresolved("Reject")
 		} else {
 			p.requireUnresolved("Fulfill")
 		}
 		p.caller = nil
-
-		if p.ongoingCalls > 0 {
-			p.callsStopped = make(chan struct{})
-		}
-
-		if len(p.clients) > 0 || p.ongoingCalls > 0 {
-			clients = p.clients
-		}
+		return p.clients
 	})
 
 	// Pending resolution state: wait for clients to be fulfilled
@@ -195,13 +171,9 @@ func (p *Promise) Resolve(r Ptr, e error) {
 		cp.promise.fulfill(dq, res.client(t))
 		cp.promise = nil
 	}
-	if callsStopped != nil {
-		<-callsStopped
-	}
 
 	p.state.With(func(p *promiseState) {
 		// Move p into resolved state.
-		p.callsStopped = nil
 		p.result, p.err = r, e
 		for _, f := range p.signals {
 			f()
@@ -351,17 +323,9 @@ func (ans *Answer) PipelineSend(ctx context.Context, transform []PipelineOp, s S
 	l := p.state.Lock()
 	switch {
 	case l.Value().isUnresolved():
-		l.Value().ongoingCalls++
 		caller := l.Value().caller
 		l.Unlock()
-		ans, release := caller.PipelineSend(ctx, transform, s)
-		p.state.With(func(p *promiseState) {
-			p.ongoingCalls--
-			if p.ongoingCalls == 0 && p.callsStopped != nil {
-				close(p.callsStopped)
-			}
-		})
-		return ans, release
+		return caller.PipelineSend(ctx, transform, s)
 	case l.Value().isPendingResolution():
 		// Block new calls until resolved.
 		l.Unlock()
@@ -387,17 +351,9 @@ func (ans *Answer) PipelineRecv(ctx context.Context, transform []PipelineOp, r R
 	l := p.state.Lock()
 	switch {
 	case l.Value().isUnresolved():
-		l.Value().ongoingCalls++
 		caller := l.Value().caller
 		l.Unlock()
-		pcall := caller.PipelineRecv(ctx, transform, r)
-		p.state.With(func(p *promiseState) {
-			p.ongoingCalls--
-			if p.ongoingCalls == 0 && p.callsStopped != nil {
-				close(p.callsStopped)
-			}
-		})
-		return pcall
+		return caller.PipelineRecv(ctx, transform, r)
 	case l.Value().isPendingResolution():
 		// Block new calls until resolved.
 		l.Unlock()