*: use per-stream Finished signal instead of shared sfin channel

Replace the shared sfin channel with stream.Finished(), giving each
stream its own completion signal. The shared channel worked for
single-stream-at-a-time. Per-stream signals are required for
multiplexing where multiple streams finish independently.

Author: shubhamdhama

drpcmanager/manager.go

@@ -80,7 +80,6 @@ type Manager struct {
 
 	sem  drpcsignal.Chan // held by the active stream
 	sbuf streamBuffer    // largest stream id created
-	sfin chan struct{}   // shared signal for stream finished
 
 	pdone   drpcsignal.Chan // signals when NewServerStream has registered the new stream
 	invokes chan invokeInfo // completed invoke info from manageReader to NewServerStream
@@ -119,8 +118,6 @@ func NewWithOptions(tr drpc.Transport, opts Options) *Manager {
 		opts: opts,
 
 		invokes: make(chan invokeInfo),
-
-		sfin: make(chan struct{}, 1),
 	}
 
 	// initialize the stream buffer
@@ -137,7 +134,6 @@ func NewWithOptions(tr drpc.Transport, opts Options) *Manager {
 
 	// set the internal stream options
 	drpcopts.SetStreamTransport(&m.opts.Stream.Internal, m.tr)
-	drpcopts.SetStreamFin(&m.opts.Stream.Internal, m.sfin)
 
 	go m.manageReader()
 
@@ -370,10 +366,10 @@ func (m *Manager) manageStream(ctx context.Context, stream *drpcstream.Stream) {
 			}
 		}
 		stream.Cancel(err)
-		<-m.sfin
+		<-stream.Finished()
 		m.sem.Recv()
 
-	case <-m.sfin:
+	case <-stream.Finished():
 		m.sem.Recv()
 
 	case <-ctx.Done():
@@ -394,7 +390,7 @@ func (m *Manager) manageStream(ctx context.Context, stream *drpcstream.Stream) {
 			stream.Cancel(ctx.Err())
 
 			// wait for the stream to signal that it is finished.
-			<-m.sfin
+			<-stream.Finished()
 		} else {
 			// If the stream isn't already finished, we have to terminate the
 			// transport to do an active cancel. If it is already finished,
@@ -407,7 +403,7 @@ func (m *Manager) manageStream(ctx context.Context, stream *drpcstream.Stream) {
 			}
 
 			// wait for the stream to signal that it is finished.
-			<-m.sfin
+			<-stream.Finished()
 
 			// allow a new stream to begin.
 			m.sem.Recv()

drpcstream/stream.go

@@ -45,7 +45,6 @@ type Options struct {
 type Stream struct {
 	ctx  streamCtx
 	opts Options
-	fin  chan<- struct{}
 	task *trace.Task
 
 	write inspectMutex
@@ -100,7 +99,6 @@ func NewWithOptions(ctx context.Context, sid uint64, wr *drpcwire.Writer, opts O
 			tr:      drpcopts.GetStreamTransport(&opts.Internal),
 		},
 		opts: opts,
-		fin:  drpcopts.GetStreamFin(&opts.Internal),
 		task: task,
 
 		pa: pa,
@@ -309,9 +307,6 @@ func (s *Stream) checkFinished() {
 		if s.sigs.fin.Set(nil) {
 			s.log("FIN", func() string { return "" })
 			s.ctx.sig.Set(context.Canceled)
-			if s.fin != nil {
-				s.fin <- struct{}{}
-			}
 			if s.task != nil {
 				s.task.End()
 			}

internal/drpcopts/stream.go

@@ -23,12 +23,6 @@ func GetStreamTransport(opts *Stream) drpc.Transport { return opts.transport }
 // SetStreamTransport sets the drpc.Transport stored in the options.
 func SetStreamTransport(opts *Stream, tr drpc.Transport) { opts.transport = tr }
 
-// GetStreamFin returns the chan<- struct{} stored in the options.
-func GetStreamFin(opts *Stream) chan<- struct{} { return opts.fin }
-
-// SetStreamFin sets the chan<- struct{} stored in the options.
-func SetStreamFin(opts *Stream, fin chan<- struct{}) { opts.fin = fin }
-
 // GetStreamKind returns the StreamKind stored in the options.
 func GetStreamKind(opts *Stream) drpc.StreamKind { return opts.kind }