code: func (e *executor) mapperLocal(ctx context.Context, shards []uint64, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
  	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.mapperLocal")
  	defer span.Finish()
  
  	ch := make(chan mapResponse, len(shards))
  
  	for _, shard := range shards {
  		go func(shard uint64) {
  			result, err := mapFn(shard)
  
  			// Return response to the channel.
  			select {
  			case <-ctx.Done():
  			case ch <- mapResponse{result: result, err: err}:
  			}
  		}(shard)
  	}
  
  	// Reduce results
  	var maxShard int
  	var result interface{}
  	for {
  		select {
  		case <-ctx.Done():
  			return nil, ctx.Err()
  		case resp := <-ch:
  			if resp.err != nil {
  				return nil, resp.err
  			}
  			result = reduceFn(result, resp.result)
  			maxShard++
  		}
  
  		// Exit once all shards are processed.
  		if maxShard == len(shards) {
  			return result, nil
  		}
  	}
  }
docs: // mapperLocal performs map & reduce entirely on the local node.