sharding requests: fix race condition
rereqs was shared in goroutines; dual appends was racey.
There is no reason for rereqs to exist.
@@ -1201,35 +1201,23 @@ func (cl *Client) handleShardedReq(ctx context.Context, req kmsg.Request) ([]Res
retryTimeout = cl.cfg.retryTimeout(req.Key())
wg sync.WaitGroup
- issue func(reqs []reqTry)
+ issue func(reqTry)
)
// issue is called to progressively split and issue requests.
//
// This recursively calls itself if a request fails and can be retried.
- issue = func(reqTries []reqTry) {
- defer wg.Done()
-
- var rereqs []reqTry
- defer func() {
- if len(rereqs) > 0 {
- wg.Add(1)
- issue(rereqs)
- }
- }()
-
- for i := range reqTries {
- try := reqTries[i]
+ issue = func(try reqTry) {
issues, reshardable, err := sharder.shard(ctx, try.req)
if err != nil {
addShard(shard(nil, try.req, nil, err)) // failure to shard means data loading failed; this request is failed
- continue
+ return
}
for i := range issues {
- wg.Add(1)
tries := try.tries
- issue := issues[i]
+ myIssue := issues[i]
+ wg.Add(1)
go func() {
defer wg.Done()
start:
@@ -1237,21 +1225,21 @@ func (cl *Client) handleShardedReq(ctx context.Context, req kmsg.Request) ([]Res
broker := cl.broker()
var err error
- if !issue.any {
- broker, err = cl.brokerOrErr(ctx, issue.broker, ErrUnknownBroker)
+ if !myIssue.any {
+ broker, err = cl.brokerOrErr(ctx, myIssue.broker, ErrUnknownBroker)
}
if err != nil {
- addShard(shard(nil, issue.req, nil, err)) // failure to load a broker is a failure to issue a request
+ addShard(shard(nil, myIssue.req, nil, err)) // failure to load a broker is a failure to issue a request
return
}
- resp, err := broker.waitResp(ctx, issue.req)
+ resp, err := broker.waitResp(ctx, myIssue.req)
if err == nil {
// Successful responses may need to perform some
// response internal error checking cleanup.
// So, we call onResp, then keep the response.
sharder.onResp(resp)
- addShard(shard(broker, issue.req, resp, nil))
+ addShard(shard(broker, myIssue.req, resp, nil))
return
}
@@ -1266,18 +1254,16 @@ func (cl *Client) handleShardedReq(ctx context.Context, req kmsg.Request) ([]Res
if !reshardable {
goto start
}
- rereqs = append(rereqs, reqTry{tries, issue.req})
+ issue(reqTry{tries, myIssue.req})
return
}
- addShard(shard(broker, issue.req, nil, err)) // the error was not retriable
+ addShard(shard(broker, myIssue.req, nil, err)) // the error was not retriable
}()
}
}
- }
- wg.Add(1)
- issue([]reqTry{{0, req}})
+ issue(reqTry{0, req})
wg.Wait()
return shards, sharder.merge
