scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-11 15:21:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

don't cancel the context if replicas should be attempted

Browse source
This commit is contained in:
Travis 2021-02-16 15:10:24 -06:00
parent 42b8522e80
commit d27afc42f1
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
1 changed files with 16 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
executor.go
View file

@ -46,6 +46,8 @@ const (
columnLabel = "col"
rowLabel = "row"
errConnectionRefused = "connect: connection refused"
)
// executor recursively executes calls in a PQL query across all shards.
@ -3501,7 +3503,6 @@ func (e *executor) executeGroupByShard(ctx context.Context, qcx *Qcx, index stri
}
func (e *executor) executeRows(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (RowIDs, error) {
// Fetch field name from argument.
// Check "field" first for backwards compatibility.
// TODO: remove at Pilosa 2.0
@ -5579,7 +5580,12 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
// On error retry against remaining nodes. If an error returns then
// the context will cancel and cause all open goroutines to return.
if resp.err != nil {
// We distinguish here between an error which indicates that the
// node is not available (and therefore we need to failover to a
// replica) and a valid error from a healthy node. In the case of
// the latter, there's no need to retry a replica, we should trust
// the error from the healthy node and return that immediately.
if resp.err != nil && strings.Contains(resp.err.Error(), errConnectionRefused) {
// Filter out unavailable nodes.
nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
@ -5587,15 +5593,16 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
return nil, resp.err
} else if err != nil {
return nil, errors.Wrap(err, "calling mapper")
return nil, errors.Wrap(err, "mapping on secondary node")
}
continue
} else if resp.err != nil {
return nil, errors.Wrap(resp.err, "mapping on primary node")
}
// Reduce value.
result = reduceFn(ctx, result, resp.result)
if err, ok := result.(error); ok {
cancel()
return nil, err
}
@ -5689,7 +5696,11 @@ func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch cha
// The cancel coming after the above send is intentional.
// We want to report the actual error that happened
// before we cause anything to return "context canceled".
if resp.err != nil {
// Also, we only want to call cancel if the error is from a
// healthy node. If the error is "connection refused" because
// the node is unavailable, we don't call cancel, and instead
// let the mapper continue trying replica nodes.
if resp.err != nil && !strings.Contains(resp.err.Error(), errConnectionRefused) {
cancel()
}
}