Method Sort

worker/sort.go:89–128

Sort is used to sort given UID matrix.

(ctx context.Context, s *pb.SortMessage)

// Sort is used to sort given UID matrix.
func (w *grpcWorker) Sort(ctx context.Context, s *pb.SortMessage) (*pb.SortResult, error) {
	if ctx.Err() != nil {
		return &emptySortResult, ctx.Err()
	}

	// Manually extract trace context from gRPC metadata for cross-alpha tracing
	ctx = x.ExtractTraceContext(ctx)

	ctx, span := otel.Tracer("").Start(ctx, "worker.Sort")
	defer span.End()

	gid, err := groups().BelongsToReadOnly(s.Order[0].Attr, s.ReadTs)
	if err != nil {
		return &emptySortResult, err
	}

	span.AddEvent("Sorting", trace.WithAttributes(
		attribute.String("attribute", s.Order[0].Attr),
		attribute.Int("groupId", int(gid))))

	if gid != groups().groupId() {
		return nil, errors.Errorf("attr: %q groupId: %v Request sent to wrong server.",
			s.Order[0].Attr, gid)
	}

	var reply *pb.SortResult
	c := make(chan error, 1)
	go func() {
		var err error
		reply, err = processSort(ctx, s)
		c <- err
	}()

	select {
	case <-ctx.Done():
		return &emptySortResult, ctx.Err()
	case err := <-c:
		return reply, err
	}
}

var (
	errContinue = errors.Errorf("Continue processing buckets")