Do function f in parallel for all replicas in the set, erroring is we exceed MaxErrors and returning early otherwise. zoneResultsQuorum allows only include results from zones that already reach quorum to improve performance.
(ctx context.Context, delay time.Duration, zoneResultsQuorum bool, partialDataEnabled bool, f func(context.Context, *InstanceDesc) (any, error))
| 28 | // MaxErrors and returning early otherwise. zoneResultsQuorum allows only include |
| 29 | // results from zones that already reach quorum to improve performance. |
| 30 | func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, zoneResultsQuorum bool, partialDataEnabled bool, f func(context.Context, *InstanceDesc) (any, error)) ([]any, error) { |
| 31 | type instanceResult struct { |
| 32 | res any |
| 33 | err error |
| 34 | instance *InstanceDesc |
| 35 | } |
| 36 | |
| 37 | // Initialise the result tracker, which is used to keep track of successes and failures. |
| 38 | var tracker replicationSetResultTracker |
| 39 | if r.MaxUnavailableZones > 0 { |
| 40 | tracker = newZoneAwareResultTracker(r.Instances, r.MaxUnavailableZones, zoneResultsQuorum) |
| 41 | } else { |
| 42 | tracker = newDefaultResultTracker(r.Instances, r.MaxErrors) |
| 43 | } |
| 44 | |
| 45 | var ( |
| 46 | ch = make(chan instanceResult, len(r.Instances)) |
| 47 | forceStart = make(chan struct{}, r.MaxErrors) |
| 48 | ) |
| 49 | ctx, cancel := context.WithCancel(ctx) |
| 50 | defer cancel() |
| 51 | |
| 52 | // Spawn a goroutine for each instance. |
| 53 | for i := range r.Instances { |
| 54 | go func(i int, ing *InstanceDesc) { |
| 55 | // Wait to send extra requests. Works only when zone-awareness is disabled. |
| 56 | if delay > 0 && r.MaxUnavailableZones == 0 && i >= len(r.Instances)-r.MaxErrors { |
| 57 | after := time.NewTimer(delay) |
| 58 | defer after.Stop() |
| 59 | select { |
| 60 | case <-ctx.Done(): |
| 61 | return |
| 62 | case <-forceStart: |
| 63 | case <-after.C: |
| 64 | } |
| 65 | } |
| 66 | result, err := f(ctx, ing) |
| 67 | ch <- instanceResult{ |
| 68 | res: result, |
| 69 | err: err, |
| 70 | instance: ing, |
| 71 | } |
| 72 | }(i, &r.Instances[i]) |
| 73 | } |
| 74 | |
| 75 | for !tracker.succeeded() && !tracker.finished() { |
| 76 | select { |
| 77 | case res := <-ch: |
| 78 | tracker.done(res.instance, res.res, res.err) |
| 79 | if res.err != nil { |
| 80 | if tracker.failed() && (!partialDataEnabled || tracker.failedCompletely()) { |
| 81 | return nil, res.err |
| 82 | } |
| 83 | |
| 84 | if validation.IsLimitError(res.err) { |
| 85 | return nil, res.err |
| 86 | } |
| 87 |