Fan out per-supplier RPCs with a context deadline and aggregate
Implement aggregateSearch which fans out one concurrent SearchRPC(ctx, supplier, hotels) per supplier and aggregates every result into a map[hotel][]price. Requirements: pass ctx into every RPC so a timeout cancels in-flight calls instead of waiting them out; collect results over a channel (no shared map under a mutex); close that channel only after all goroutines finish, never from a sender.
func aggregateSearch(ctx context.Context, md []SearchMeta) map[string][]float64 {
// your code here
return nil
}Write the implementation.
Launch one goroutine per supplier, each calling SearchRPC(ctx, …) and sending results to a buffered channel; a sync.WaitGroup plus a closer goroutine closes it after all finish. Aggregate by ranging the channel, and propagate ctx so a timeout cancels every in-flight RPC instead of waiting.
- ✗Creating the context with
context.WithTimeoutbut never calling itscancel— leaks the timer until the deadline - ✗Closing the results channel from a sender goroutine instead of after
wg.Wait, causing a send on a closed channel - ✗Not passing
ctxintoSearchRPC, so the deadline never actually cancels a slow supplier
- →Why must the channel
closehappen afterwg.Wait, and why in a separate goroutine? - →How would
errgroupwithWithContextshorten this fan-out-and-aggregate code?
Skeleton to implement
context.WithTimeout returns two values (ctx, cancel); a forgotten cancel leaks the timer. A separate goroutine closes the results channel after wg.Wait, never a sender. ctx is passed into every SearchRPC so the deadline cancels slow calls.
ctx, cancel := context.WithTimeout(context.Background(), callTimeout)
defer cancel()
inverted := make(map[string][]string) // supplier -> []hotel
for _, m := range md {
inverted[m.Supplier] = append(inverted[m.Supplier], m.Hotel)
}
out := make(chan []SearchResult, len(inverted)) // buffer sized to senders
var wg sync.WaitGroup
for sup, hotels := range inverted {
wg.Add(1)
go func(sup string, hotels []string) {
defer wg.Done()
res, err := SearchRPC(ctx, sup, hotels) // ctx cancels a slow RPC
if err != nil {
return
}
select {
case out <- res:
case <-ctx.Done(): // do not block if the receiver is gone
}
}(sup, hotels)
}
go func() { wg.Wait(); close(out) }() // close only after every Done
prices := make(map[string][]float64)
for batch := range out { // aggregate as results arrive
for _, r := range batch {
prices[r.Hotel] = append(prices[r.Hotel], r.Price)
}
}
