Stop parallel search after first successful chan closes

[magik6k]

See ipfs/kubo#5404 (comment) / libp2p/go-libp2p-routing#29

[Stebalien]

Can't we just cancel the context? Something like:

diff --git a/parallel.go b/parallel.go
index f7c5015..fe1d976 100644
--- a/parallel.go
+++ b/parallel.go
@@ -164,6 +164,8 @@ func (r Parallel) search(ctx context.Context, do func(routing.IpfsRouting) (<-ch
 	var errs []error
 	var wg sync.WaitGroup
 
+	ctx, cancel := context.WithCancel(ctx)
+
 	for _, ri := range r.Routers {
 		vchan, err := do(ri)
 		switch err {
@@ -177,15 +179,20 @@ func (r Parallel) search(ctx context.Context, do func(routing.IpfsRouting) (<-ch
 
 		go func() {
 			defer wg.Done()
+			sent := 0
 			for {
 				select {
 				case v, ok := <-vchan:
 					if !ok {
+						if sent > 0 {
+							cancel()
+						}
 						return
 					}
 
 					select {
 					case out <- v:
+						sent++
 					case <-ctx.Done():
 						return
 					}
@@ -199,6 +206,7 @@ func (r Parallel) search(ctx context.Context, do func(routing.IpfsRouting) (<-ch
 	go func() {
 		wg.Wait()
 		close(out)
+		cancel()
 	}()
 
 	return out, nil

[Stebalien]

(same thing really, just uses the context instead of creating and managing a new channel)

[magik6k]

Nope, already tried that, go vet complains about leaking close

[Stebalien]

@magik6k you just need that final cancel in after the close(out) (go vet looks happy with that patch).

[magik6k]

oh.

Looks far better. I squashed all the random the commits, and ran go test ./... -count=400 successfully. Will merge once pubsub PR is ready