infprocessor

Documentation

Index

• type P
  • func NewP(queue *TaskQueue, engineRouter engineRouter) *P
  • func (p *P) AddOrUpdateEngineStatus(srv engineCommunicator, engineStatus *v1.EngineStatus, ...)
  • func (p *P) MaxInProgressTaskDuration() time.Duration
  • func (p *P) NumEnginesByTenantID() map[string]int
  • func (p *P) NumInProgressTasks() int
  • func (p *P) NumQueuedTasks() int32
  • func (p *P) ProcessTaskResult(taskResult *v1.TaskResult, clusterInfo *auth.ClusterInfo) error
  • func (p *P) RemoveEngine(engineID string, clusterInfo *auth.ClusterInfo)
  • func (p *P) Run(ctx context.Context) error
• type Task
  • func (t *Task) WaitForCompletion(ctx context.Context) (*http.Response, error)
• type TaskQueue
  • func NewTaskQueue() *TaskQueue
  • func (q *TaskQueue) Dequeue(ctx context.Context) (*Task, error)
  • func (q *TaskQueue) Enqueue(t *Task)

Types

type P

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

P processes inference tasks.

func NewP

func NewP(queue *TaskQueue, engineRouter engineRouter) *P

NewP creates a new processor.

func (*P) AddOrUpdateEngineStatus

func (p *P) AddOrUpdateEngineStatus(
	srv engineCommunicator,
	engineStatus *v1.EngineStatus,
	clusterInfo *auth.ClusterInfo,
)

AddOrUpdateEngineStatus adds or updates the engine status.

func (*P) MaxInProgressTaskDuration

func (p *P) MaxInProgressTaskDuration() time.Duration

MaxInProgressTaskDuration returns the maximum duration of in-progress tasks.

func (*P) NumEnginesByTenantID

func (p *P) NumEnginesByTenantID() map[string]int

NumEnginesByTenantID returns the number of engines by tenant ID.

func (*P) NumInProgressTasks

func (p *P) NumInProgressTasks() int

NumInProgressTasks returns the number of in-progress tasks.

func (*P) NumQueuedTasks

func (p *P) NumQueuedTasks() int32

NumQueuedTasks returns the number of queued tasks.

func (*P) ProcessTaskResult

func (p *P) ProcessTaskResult(
	taskResult *v1.TaskResult,
	clusterInfo *auth.ClusterInfo,
) error

ProcessTaskResult processes the task result.

func (*P) RemoveEngine

func (p *P) RemoveEngine(engineID string, clusterInfo *auth.ClusterInfo)

RemoveEngine removes the engine.

func (*P) Run

func (p *P) Run(ctx context.Context) error

Run runs the processor.

type Task

type Task struct {
	ID       string
	TenantID string

	Req    *v1.CreateChatCompletionRequest
	Header http.Header
	RespCh chan *http.Response
	ErrCh  chan error

	EngineID string

	CreatedAt time.Time
	// contains filtered or unexported fields
}

Task is an inference task. TODO(kenji): Consider preserving the request context as well.

func (*Task) WaitForCompletion

func (t *Task) WaitForCompletion(ctx context.Context) (*http.Response, error)

WaitForCompletion waits for the completion of the task.

type TaskQueue

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

TaskQueue is a queue for inference tasks.

func NewTaskQueue

func NewTaskQueue() *TaskQueue

NewTaskQueue creates a new task queue.

func (*TaskQueue) Dequeue

func (q *TaskQueue) Dequeue(ctx context.Context) (*Task, error)

Dequeue removes a task from the queue.

func (*TaskQueue) Enqueue

func (q *TaskQueue) Enqueue(t *Task)

Enqueue inserts a task into the queue.