tasks

Documentation

Index

• type MergeTask
• type QueryStreamTask
  • func NewQueryStreamTask(ctx context.Context, collection *segments.Collection, ...) *QueryStreamTask
  • func (t *QueryStreamTask) Canceled() error
  • func (t *QueryStreamTask) Done(err error)
  • func (t *QueryStreamTask) Execute() error
  • func (t *QueryStreamTask) IsGpuIndex() bool
  • func (t *QueryStreamTask) NQ() int64
  • func (t *QueryStreamTask) PreExecute() error
  • func (t *QueryStreamTask) SearchResult() *internalpb.SearchResults
  • func (t *QueryStreamTask) Username() string
  • func (t *QueryStreamTask) Wait() error
• type QueryTask
  • func NewQueryTask(ctx context.Context, collection *segments.Collection, ...) *QueryTask
  • func (t *QueryTask) Canceled() error
  • func (t *QueryTask) Done(err error)
  • func (t *QueryTask) Execute() error
  • func (t *QueryTask) IsGpuIndex() bool
  • func (t *QueryTask) NQ() int64
  • func (t *QueryTask) PreExecute() error
  • func (t *QueryTask) Result() *internalpb.RetrieveResults
  • func (t *QueryTask) SearchResult() *internalpb.SearchResults
  • func (t *QueryTask) Username() string
  • func (t *QueryTask) Wait() error
• type Scheduler
  • func NewScheduler(policyName string) Scheduler
• type SearchTask
  • func NewSearchTask(ctx context.Context, collection *segments.Collection, ...) *SearchTask
  • func (t *SearchTask) Canceled() error
  • func (t *SearchTask) Done(err error)
  • func (t *SearchTask) Execute() error
  • func (t *SearchTask) GetNodeID() int64
  • func (t *SearchTask) IsGpuIndex() bool
  • func (t *SearchTask) Merge(other *SearchTask) bool
  • func (t *SearchTask) MergeWith(other Task) bool
  • func (t *SearchTask) NQ() int64
  • func (t *SearchTask) PreExecute() error
  • func (t *SearchTask) SearchResult() *internalpb.SearchResults
  • func (t *SearchTask) Username() string
  • func (t *SearchTask) Wait() error
• type StreamingSearchTask
  • func NewStreamingSearchTask(ctx context.Context, collection *segments.Collection, ...) *StreamingSearchTask
  • func (t *StreamingSearchTask) Execute() error
  • func (t *StreamingSearchTask) MergeWith(other Task) bool
• type Task

Types

type MergeTask

type MergeTask interface {
	Task

	// MergeWith other task, return true if merge success.
	// After success, the task merged should be dropped.
	MergeWith(Task) bool
}

MergeTask is a Task which can be merged with other task

type QueryStreamTask

type QueryStreamTask struct {
	// contains filtered or unexported fields
}

func NewQueryStreamTask

func NewQueryStreamTask(ctx context.Context,
	collection *segments.Collection,
	manager *segments.Manager,
	req *querypb.QueryRequest,
	srv streamrpc.QueryStreamServer,
	streamBatchSize int,
) *QueryStreamTask

func (*QueryStreamTask) Canceled

func (t *QueryStreamTask) Canceled() error

func (*QueryStreamTask) Done

func (t *QueryStreamTask) Done(err error)

func (*QueryStreamTask) Execute

func (t *QueryStreamTask) Execute() error

func (*QueryStreamTask) IsGpuIndex

func (t *QueryStreamTask) IsGpuIndex() bool

func (*QueryStreamTask) NQ

func (t *QueryStreamTask) NQ() int64

func (*QueryStreamTask) PreExecute

func (t *QueryStreamTask) PreExecute() error

PreExecute the task, only call once.

func (*QueryStreamTask) SearchResult

func (t *QueryStreamTask) SearchResult() *internalpb.SearchResults

func (*QueryStreamTask) Username

func (t *QueryStreamTask) Username() string

Return the username which task is belong to. Return "" if the task do not contain any user info.

func (*QueryStreamTask) Wait

func (t *QueryStreamTask) Wait() error

type QueryTask

type QueryTask struct {
	// contains filtered or unexported fields
}

func NewQueryTask

func NewQueryTask(ctx context.Context,
	collection *segments.Collection,
	manager *segments.Manager,
	req *querypb.QueryRequest,
) *QueryTask

func (*QueryTask) Canceled

func (t *QueryTask) Canceled() error

func (*QueryTask) Done

func (t *QueryTask) Done(err error)

func (*QueryTask) Execute

func (t *QueryTask) Execute() error

Execute the task, only call once.

func (*QueryTask) IsGpuIndex

func (t *QueryTask) IsGpuIndex() bool

func (*QueryTask) NQ

func (t *QueryTask) NQ() int64

func (*QueryTask) PreExecute

func (t *QueryTask) PreExecute() error

PreExecute the task, only call once.

func (*QueryTask) Result

func (t *QueryTask) Result() *internalpb.RetrieveResults

func (*QueryTask) SearchResult

func (t *QueryTask) SearchResult() *internalpb.SearchResults

func (*QueryTask) Username

func (t *QueryTask) Username() string

Return the username which task is belong to. Return "" if the task do not contain any user info.

func (*QueryTask) Wait

func (t *QueryTask) Wait() error

type Scheduler

type Scheduler interface {
	// Add a new task into scheduler, follow some constraints.
	// 1. It's a non-block operation.
	// 2. Error will be returned if scheduler reaches some limit.
	// 3. Concurrent safe.
	Add(task Task) error

	// Start schedule the owned task asynchronously and continuously.
	// Shall be called only once
	Start()

	// Stop make scheduler deny all incoming tasks
	// and cleans up all related resources
	Stop()

	// GetWaitingTaskTotalNQ
	GetWaitingTaskTotalNQ() int64

	// GetWaitingTaskTotal
	GetWaitingTaskTotal() int64
}

func NewScheduler

func NewScheduler(policyName string) Scheduler

NewScheduler create a scheduler by policyName.

type SearchTask

type SearchTask struct {
	// contains filtered or unexported fields
}

func NewSearchTask

func NewSearchTask(ctx context.Context,
	collection *segments.Collection,
	manager *segments.Manager,
	req *querypb.SearchRequest,
	serverID int64,
) *SearchTask

func (*SearchTask) Canceled

func (t *SearchTask) Canceled() error

func (*SearchTask) Done

func (t *SearchTask) Done(err error)

func (*SearchTask) Execute

func (t *SearchTask) Execute() error

func (*SearchTask) GetNodeID

func (t *SearchTask) GetNodeID() int64

func (*SearchTask) IsGpuIndex

func (t *SearchTask) IsGpuIndex() bool

func (*SearchTask) Merge

func (t *SearchTask) Merge(other *SearchTask) bool

func (*SearchTask) MergeWith

func (t *SearchTask) MergeWith(other Task) bool

func (*SearchTask) NQ

func (t *SearchTask) NQ() int64

func (*SearchTask) PreExecute

func (t *SearchTask) PreExecute() error

func (*SearchTask) SearchResult

func (t *SearchTask) SearchResult() *internalpb.SearchResults

func (*SearchTask) Username

func (t *SearchTask) Username() string

Return the username which task is belong to. Return "" if the task do not contain any user info.

func (*SearchTask) Wait

func (t *SearchTask) Wait() error

type StreamingSearchTask

type StreamingSearchTask struct {
	SearchTask
	// contains filtered or unexported fields
}

func NewStreamingSearchTask

func NewStreamingSearchTask(ctx context.Context,
	collection *segments.Collection,
	manager *segments.Manager,
	req *querypb.SearchRequest,
	serverID int64,
) *StreamingSearchTask

func (*StreamingSearchTask) Execute

func (t *StreamingSearchTask) Execute() error

func (*StreamingSearchTask) MergeWith

func (t *StreamingSearchTask) MergeWith(other Task) bool

type Task

type Task interface {
	// Return the username which task is belong to.
	// Return "" if the task do not contain any user info.
	Username() string

	// Return whether the task would be running on GPU.
	IsGpuIndex() bool

	// PreExecute the task, only call once.
	PreExecute() error

	// Execute the task, only call once.
	Execute() error

	// Done notify the task finished.
	Done(err error)

	// Check if the Task is canceled.
	// Concurrent safe.
	Canceled() error

	// Wait for task finish.
	// Concurrent safe.
	Wait() error

	// Return the NQ of task.
	NQ() int64

	SearchResult() *internalpb.SearchResults
}

A task is execute unit of scheduler.