work

Documentation

Index

• func CmdScheduleNextJob() loop.Cmd
• func NewRequest(ctx context.Context, req *reqctx.RequestDetails, stageIndex int, ...) *pbssinternal.ProcessRangeRequest
• type MsgJobFailed
• type MsgJobSucceeded
• type MsgScheduleNextJob
• type RemoteWorker
  • func NewRemoteWorker(clientFactory client.InternalClientFactory, logger *zap.Logger) *RemoteWorker
  • func (w *RemoteWorker) ID() string
  • func (w *RemoteWorker) Work(ctx context.Context, unit stage.Unit, startBlock uint64, moduleNames []string, ...) loop.Cmd
• type Result
• type RetryableErr
  • func NewRetryableErr(cause error) *RetryableErr
  • func (r *RetryableErr) Error() string
• type SimpleWorkerFactory
  • func NewWorkerFactoryFromFunc(...) *SimpleWorkerFactory
  • func (f SimpleWorkerFactory) ID() string
  • func (f SimpleWorkerFactory) Work(ctx context.Context, unit stage.Unit, startBlock uint64, moduleNames []string, ...) loop.Cmd
• type Worker
• type WorkerFactory
• type WorkerPool
  • func NewWorkerPool(ctx context.Context, workerCount int, workerFactory WorkerFactory) *WorkerPool
  • func (p *WorkerPool) Borrow() Worker
  • func (p *WorkerPool) Return(worker Worker)
  • func (p *WorkerPool) WorkerAvailable() (avail bool, shouldRetry bool)
• type WorkerState
• type WorkerStatus

Functions

func CmdScheduleNextJob

func CmdScheduleNextJob() loop.Cmd

func NewRequest

func NewRequest(ctx context.Context, req *reqctx.RequestDetails, stageIndex int, startBlock uint64) *pbssinternal.ProcessRangeRequest

Types

type MsgJobFailed

type MsgJobFailed struct {
	Unit  stage.Unit
	Error error
}

type MsgJobSucceeded

type MsgJobSucceeded struct {
	Unit   stage.Unit
	Worker Worker
}

type MsgScheduleNextJob

type MsgScheduleNextJob struct{}

type RemoteWorker

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

func NewRemoteWorker

func NewRemoteWorker(clientFactory client.InternalClientFactory, logger *zap.Logger) *RemoteWorker

func (*RemoteWorker) ID

func (w *RemoteWorker) ID() string

func (*RemoteWorker) Work

func (w *RemoteWorker) Work(ctx context.Context, unit stage.Unit, startBlock uint64, moduleNames []string, upstream *response.Stream) loop.Cmd

type Result

type Result struct {
	PartialFilesWritten store.FileInfos
	Error               error
}

type RetryableErr

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

func NewRetryableErr

func NewRetryableErr(cause error) *RetryableErr

func (*RetryableErr) Error

func (r *RetryableErr) Error() string

type SimpleWorkerFactory

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

func NewWorkerFactoryFromFunc

func NewWorkerFactoryFromFunc(f func(ctx context.Context, unit stage.Unit, startBlock uint64, moduleNames []string, upstream *response.Stream) loop.Cmd) *SimpleWorkerFactory

func (SimpleWorkerFactory) ID

func (f SimpleWorkerFactory) ID() string

func (SimpleWorkerFactory) Work

func (f SimpleWorkerFactory) Work(ctx context.Context, unit stage.Unit, startBlock uint64, moduleNames []string, upstream *response.Stream) loop.Cmd

type Worker

type Worker interface {
	ID() string
	Work(ctx context.Context, unit stage.Unit, startBlock uint64, moduleNames []string, upstream *response.Stream) loop.Cmd // *Result
}

type WorkerFactory

type WorkerFactory = func(logger *zap.Logger) Worker

The tracer will be provided by the worker pool, on worker creation

type WorkerPool

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

func NewWorkerPool

func NewWorkerPool(ctx context.Context, workerCount int, workerFactory WorkerFactory) *WorkerPool

func (*WorkerPool) Borrow

func (p *WorkerPool) Borrow() Worker

func (*WorkerPool) Return

func (p *WorkerPool) Return(worker Worker)

func (*WorkerPool) WorkerAvailable

func (p *WorkerPool) WorkerAvailable() (avail bool, shouldRetry bool)

type WorkerState

type WorkerState int

const (
	WorkerFree WorkerState = iota
	WorkerWorking
	WorkerInitialWait
)

type WorkerStatus

type WorkerStatus struct {
	State  WorkerState
	Worker Worker
}