DoBatch request against a set of keys in the ring, handling replication and failures. For example if we want to write N items where they may all hit different instances, and we want them all replicated R ways with quorum writes, we track the relationship between batch RPCs and the items within them.
(ctx context.Context, op Operation, r ReadRing, e util.AsyncExecutor, keys []uint32, callback func(InstanceDesc, []int) error, cleanup func())
| 72 | // |
| 73 | // Not implemented as a method on Ring so we can test separately. |
| 74 | func DoBatch(ctx context.Context, op Operation, r ReadRing, e util.AsyncExecutor, keys []uint32, callback func(InstanceDesc, []int) error, cleanup func()) error { |
| 75 | if r.InstancesCount() <= 0 { |
| 76 | cleanup() |
| 77 | return fmt.Errorf("DoBatch: InstancesCount <= 0") |
| 78 | } |
| 79 | |
| 80 | if e == nil { |
| 81 | e = noOpExecutor |
| 82 | } |
| 83 | |
| 84 | expectedTrackers := len(keys) * (r.ReplicationFactor() + 1) / r.InstancesCount() |
| 85 | itemTrackers := make([]itemTracker, len(keys)) |
| 86 | instances := make(map[string]instance, r.InstancesCount()) |
| 87 | |
| 88 | var ( |
| 89 | bufDescs [GetBufferSize]InstanceDesc |
| 90 | bufHosts [GetBufferSize]string |
| 91 | bufZones = make(map[string]int, GetZoneSize) |
| 92 | ) |
| 93 | for i, key := range keys { |
| 94 | replicationSet, err := r.Get(key, op, bufDescs[:0], bufHosts[:0], bufZones) |
| 95 | if err != nil { |
| 96 | cleanup() |
| 97 | return err |
| 98 | } |
| 99 | itemTrackers[i].minSuccess = len(replicationSet.Instances) - replicationSet.MaxErrors |
| 100 | itemTrackers[i].maxFailures = replicationSet.MaxErrors |
| 101 | itemTrackers[i].remaining.Store(int32(len(replicationSet.Instances))) |
| 102 | |
| 103 | for _, desc := range replicationSet.Instances { |
| 104 | curr, found := instances[desc.Addr] |
| 105 | if !found { |
| 106 | curr.itemTrackers = make([]*itemTracker, 0, expectedTrackers) |
| 107 | curr.indexes = make([]int, 0, expectedTrackers) |
| 108 | } |
| 109 | instances[desc.Addr] = instance{ |
| 110 | desc: desc, |
| 111 | itemTrackers: append(curr.itemTrackers, &itemTrackers[i]), |
| 112 | indexes: append(curr.indexes, i), |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | tracker := batchTracker{ |
| 118 | done: make(chan struct{}, 1), |
| 119 | err: make(chan error, 1), |
| 120 | } |
| 121 | tracker.rpcsPending.Store(int32(len(itemTrackers))) |
| 122 | |
| 123 | var wg sync.WaitGroup |
| 124 | |
| 125 | wg.Add(len(instances)) |
| 126 | for _, i := range instances { |
| 127 | e.Submit(func() { |
| 128 | err := callback(i.desc, i.indexes) |
| 129 | tracker.record(i, err) |
| 130 | wg.Done() |
| 131 | }) |