queuing

Documentation

Index

• Variables
• func New(baseURL string, config *Config) (aio.Subsystem, error)
• func NewDST(config *ConfigDST, r *rand.Rand) (aio.Subsystem, error)
• type Config
• type ConfigDST
• type MatchResult
• type QueuingSubsystem
  • func (t *QueuingSubsystem) NewWorker(i int) aio.Worker
  • func (t *QueuingSubsystem) Reset() error
  • func (t *QueuingSubsystem) Start() error
  • func (t *QueuingSubsystem) Stop() error
  • func (t *QueuingSubsystem) String() string
• type QueuingSubsystemDST
  • func (q *QueuingSubsystemDST) NewWorker(int) aio.Worker
  • func (q *QueuingSubsystemDST) Reset() error
  • func (q *QueuingSubsystemDST) Start() error
  • func (q *QueuingSubsystemDST) Stop() error
  • func (q *QueuingSubsystemDST) String() string
• type QueuingWorker
  • func (w *QueuingWorker) Process(sqes []*bus.SQE[t_aio.Submission, t_aio.Completion]) []*bus.CQE[t_aio.Submission, t_aio.Completion]
• type QueuingWorkerDST
  • func (w *QueuingWorkerDST) Process(sqes []*bus.SQE[t_aio.Submission, t_aio.Completion]) []*bus.CQE[t_aio.Submission, t_aio.Completion]
• type RouteHandler
• type Router
  • func CoroutineRouter() Router
  • func NewRouter() Router
• type RouterImpl
  • func (r *RouterImpl) Handle(pattern string, handler *RouteHandler)
  • func (r *RouterImpl) Match(route string) (*MatchResult, error)

Variables

var (
	ErrConnectionNotFound = errors.New("connection not found")
)

Functions

func New

func New(baseURL string, config *Config) (aio.Subsystem, error)

New creates a new queuing subsystem with the given config.

func NewDST

func NewDST(config *ConfigDST, r *rand.Rand) (aio.Subsystem, error)

NewDST is a simple helper functions that wraps New and returns a pre-configured QueuingSubsystem. This configurations aligns with the DST tests. Search for: 'id = fmt.Sprintf("/gpu/summarize/%s", id)'

Types

type Config

type Config struct {
	Connections []*t_conn.ConnectionConfig
	Routes      []*t_route.RoutingConfig
}

Config is the configuration for the queuing subsystem.

type ConfigDST

type ConfigDST struct {
	P float32
}

type MatchResult

type MatchResult struct {
	Route        string
	RoutePattern string
	Connection   string
	Queue        string
}

Resonate chi router wrapper.

type QueuingSubsystem

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

QueuingSubsystem is a subsystem that dispatches tasks to user defined connections.

func (*QueuingSubsystem) NewWorker

func (t *QueuingSubsystem) NewWorker(i int) aio.Worker

NewWorker creates a new worker for the queuing subsystem with the submission queues for each connection.

func (*QueuingSubsystem) Reset

func (t *QueuingSubsystem) Reset() error

Reset resets the subsystems state.

func (*QueuingSubsystem) Start

func (t *QueuingSubsystem) Start() error

Start dispatches all connection to their own goroutines.

func (*QueuingSubsystem) Stop

func (t *QueuingSubsystem) Stop() error

Stop cancels the subsystems context and waits for all connections to finish.

func (*QueuingSubsystem) String

func (t *QueuingSubsystem) String() string

String returns the name of the subsystem.

type QueuingSubsystemDST

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

func (*QueuingSubsystemDST) NewWorker

func (q *QueuingSubsystemDST) NewWorker(int) aio.Worker

func (*QueuingSubsystemDST) Reset

func (q *QueuingSubsystemDST) Reset() error

func (*QueuingSubsystemDST) Start

func (q *QueuingSubsystemDST) Start() error

func (*QueuingSubsystemDST) Stop

func (q *QueuingSubsystemDST) Stop() error

func (*QueuingSubsystemDST) String

func (q *QueuingSubsystemDST) String() string

type QueuingWorker

type QueuingWorker struct {
	// BaseURL is the base URL for the API.
	BaseURL string

	// ConnectionRouter is the router to route requests to the appropriate connections.
	ConnectionRouter Router

	// ConnectionsSQ is a map of connection names to their submission queues.
	ConnectionsSQ map[string]chan *t_conn.ConnectionSubmission
	// contains filtered or unexported fields
}

QueuingWorker is a worker that dispatches submissions to the appropriate connections.

func (*QueuingWorker) Process

func (w *QueuingWorker) Process(sqes []*bus.SQE[t_aio.Submission, t_aio.Completion]) []*bus.CQE[t_aio.Submission, t_aio.Completion]

Process dispatches the given submissions to the appropriate connections.

type QueuingWorkerDST

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

func (*QueuingWorkerDST) Process

func (w *QueuingWorkerDST) Process(sqes []*bus.SQE[t_aio.Submission, t_aio.Completion]) []*bus.CQE[t_aio.Submission, t_aio.Completion]

type RouteHandler

type RouteHandler struct {
	Connection string
	Queue      string
}

Resonate chi router wrapper.

type Router

type Router interface {
	Handle(pattern string, handler *RouteHandler)
	Match(pattern string) (*MatchResult, error)
}

Resonate chi router wrapper.

var (
	ErrRouteDoesNotMatchAnyPattern = errors.New("route does not match any pattern")

	// Workaround for no dependency injection in coroutines. (TODO: consider alternatives)
	CRouter Router = NewRouter()
)

func CoroutineRouter

func CoroutineRouter() Router

func NewRouter

func NewRouter() Router

type RouterImpl

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

Resonate chi router wrapper.

func (*RouterImpl) Handle

func (r *RouterImpl) Handle(pattern string, handler *RouteHandler)

func (*RouterImpl) Match

func (r *RouterImpl) Match(route string) (*MatchResult, error)