scheduler

Documentation

Index

• Constants
• type BaseScheduleDispatcher
  • func NewBaseScheduleDispatcher(changeFeedID model.ChangeFeedID, communicator ScheduleDispatcherCommunicator, ...) *BaseScheduleDispatcher
  • func (s *BaseScheduleDispatcher) GetTaskPositions() (map[model.CaptureID]*model.TaskPosition, error)
  • func (s *BaseScheduleDispatcher) GetTaskStatuses() (map[model.CaptureID]*model.TaskStatus, error)
  • func (s *BaseScheduleDispatcher) MoveKeySpan(keyspanID model.KeySpanID, target model.CaptureID)
  • func (s *BaseScheduleDispatcher) OnAgentCheckpoint(captureID model.CaptureID, checkpointTs model.Ts, resolvedTs model.Ts)
  • func (s *BaseScheduleDispatcher) OnAgentFinishedKeySpanOperation(captureID model.CaptureID, keyspanID model.KeySpanID)
  • func (s *BaseScheduleDispatcher) OnAgentSyncTaskStatuses(captureID model.CaptureID, running, adding, removing []model.KeySpanID)
  • func (s *BaseScheduleDispatcher) Rebalance()
  • func (s *BaseScheduleDispatcher) Tick(ctx context.Context, checkpointTs model.Ts, currentKeySpans []model.KeySpanID, ...) (newCheckpointTs, resolvedTs model.Ts, err error)
• type InfoProvider
• type ScheduleDispatcher
• type ScheduleDispatcherCommunicator

Constants

const (
	// CheckpointCannotProceed is a placeholder indicating that the
	// Owner should not advance the global checkpoint TS just yet.
	CheckpointCannotProceed = model.Ts(0)
)

Types

type BaseScheduleDispatcher

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

BaseScheduleDispatcher implements the basic logic of a ScheduleDispatcher. For it to be directly useful to the Owner, the Owner should implement it own ScheduleDispatcherCommunicator.

func NewBaseScheduleDispatcher

func NewBaseScheduleDispatcher(
	changeFeedID model.ChangeFeedID,
	communicator ScheduleDispatcherCommunicator,
	checkpointTs model.Ts,
) *BaseScheduleDispatcher

NewBaseScheduleDispatcher creates a new BaseScheduleDispatcher.

func (*BaseScheduleDispatcher) GetTaskPositions

func (s *BaseScheduleDispatcher) GetTaskPositions() (map[model.CaptureID]*model.TaskPosition, error)

GetTaskPositions implements InfoProvider for BaseScheduleDispatcher.

func (*BaseScheduleDispatcher) GetTaskStatuses

func (s *BaseScheduleDispatcher) GetTaskStatuses() (map[model.CaptureID]*model.TaskStatus, error)

GetTaskStatuses implements InfoProvider for BaseScheduleDispatcher.

func (*BaseScheduleDispatcher) MoveKeySpan

func (s *BaseScheduleDispatcher) MoveKeySpan(keyspanID model.KeySpanID, target model.CaptureID)

MoveKeySpan implements the interface SchedulerDispatcher.

func (*BaseScheduleDispatcher) OnAgentCheckpoint

func (s *BaseScheduleDispatcher) OnAgentCheckpoint(captureID model.CaptureID, checkpointTs model.Ts, resolvedTs model.Ts)

OnAgentCheckpoint is called when the processor sends a checkpoint.

func (*BaseScheduleDispatcher) OnAgentFinishedKeySpanOperation

func (s *BaseScheduleDispatcher) OnAgentFinishedKeySpanOperation(captureID model.CaptureID, keyspanID model.KeySpanID)

OnAgentFinishedKeySpanOperation is called when a keyspan operation has been finished by the processor.

func (*BaseScheduleDispatcher) OnAgentSyncTaskStatuses

func (s *BaseScheduleDispatcher) OnAgentSyncTaskStatuses(captureID model.CaptureID, running, adding, removing []model.KeySpanID)

OnAgentSyncTaskStatuses is called when the processor sends its complete current state.

func (*BaseScheduleDispatcher) Rebalance

func (s *BaseScheduleDispatcher) Rebalance()

Rebalance implements the interface ScheduleDispatcher.

func (*BaseScheduleDispatcher) Tick

func (s *BaseScheduleDispatcher) Tick(
	ctx context.Context,
	checkpointTs model.Ts,

	currentKeySpans []model.KeySpanID,
	captures map[model.CaptureID]*model.CaptureInfo,
) (newCheckpointTs, resolvedTs model.Ts, err error)

Tick implements the interface ScheduleDispatcher.

type InfoProvider

type InfoProvider interface {
	// GetTaskStatuses returns the task statuses.
	GetTaskStatuses() (map[model.CaptureID]*model.TaskStatus, error)

	// GetTaskPositions returns the task positions.
	GetTaskPositions() (map[model.CaptureID]*model.TaskPosition, error)
}

InfoProvider is the interface to get information about the internal states of the scheduler. We need this interface so that we can provide the information through HTTP API.

type ScheduleDispatcher

type ScheduleDispatcher interface {
	// Tick is called periodically to update the SchedulerDispatcher on the latest state of replication.
	// This function should NOT be assumed to be thread-safe. No concurrent calls allowed.
	Tick(
		ctx context.Context,
		checkpointTs model.Ts,
		currentKeySpans []model.KeySpanID,
		captures map[model.CaptureID]*model.CaptureInfo,
	) (newCheckpointTs, newResolvedTs model.Ts, err error)

	// MoveKeySpan requests that a keyspan be moved to target.
	// It should be thread-safe.
	MoveKeySpan(keyspanID model.KeySpanID, target model.CaptureID)

	// Rebalance triggers a rebalance operation.
	// It should be thread-safe
	Rebalance()
}

ScheduleDispatcher is an interface for a keyspan scheduler used in Owner.

type ScheduleDispatcherCommunicator

type ScheduleDispatcherCommunicator interface {
	// DispatchKeySpan should send a dispatch command to the Processor.
	DispatchKeySpan(ctx context.Context,
		changeFeedID model.ChangeFeedID,
		keyspanID model.KeySpanID,
		captureID model.CaptureID,
		isDelete bool,
	) (done bool, err error)

	// Announce announces to the specified capture that the current node has become the Owner.
	Announce(ctx context.Context,
		changeFeedID model.ChangeFeedID,
		captureID model.CaptureID) (done bool, err error)
}

ScheduleDispatcherCommunicator is an interface for the BaseScheduleDispatcher to send commands to Processors. The owner of a BaseScheduleDispatcher should provide an implementation of ScheduleDispatcherCommunicator to supply BaseScheduleDispatcher some methods to specify its behavior.