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

only cancel() in mapper on a secondary, replica error

Browse source 
There is another case where cancelling here might be useful,
and that's if the query is on a primary node and the replication
factor is 1, meaning there are no secondary nodes to fail over to.
That case is handled here as well.
This commit is contained in:
Travis 2021-02-16 16:23:22 -06:00
parent d27afc42f1
commit a7d4226326
No known key found for this signature in database
GPG key ID: 37080CC2042BA34E
1 changed files with 7 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
executor.go
View file

@ -5565,7 +5565,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
}
// Start mapping across all primary owners.
if err := e.mapper(ctx, cancel, ch, nodes, index, shards, c, opt, mapFn, reduceFn); err != nil {
if err := e.mapper(ctx, cancel, ch, nodes, index, shards, c, opt, e.Cluster.ReplicaN == 1, mapFn, reduceFn); err != nil {
return nil, errors.Wrap(err, "starting mapper")
}
@ -5590,7 +5590,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
nodes = topology.Nodes(nodes).FilterID(resp.node.ID)
// Begin mapper against secondary nodes.
if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
if err := e.mapper(ctx, cancel, ch, nodes, index, resp.shards, c, opt, true, mapFn, reduceFn); errors.Cause(err) == errShardUnavailable {
return nil, resp.err
} else if err != nil {
return nil, errors.Wrap(err, "mapping on secondary node")
@ -5659,7 +5659,7 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
return newRows
}
func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
func (e *executor) mapper(ctx context.Context, cancel context.CancelFunc, ch chan mapResponse, nodes []*topology.Node, index string, shards []uint64, c *pql.Call, opt *execOptions, lastAttempt bool, mapFn mapFunc, reduceFn reduceFunc) error {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapper")
defer span.Finish()
done := ctx.Done()
@ -5696,11 +5696,10 @@ 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".
// 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) {
// Also, we only want to call cancel if the error occurs on a
// secondary node (or a primary node with no replicas), meaning
// there are no other nodes remaining to which we can fail over.
if resp.err != nil && lastAttempt {
cancel()
}
}