scheduler

Documentation

Overview

Package scheduler contains the periodic feed update scheduler.

The scheduler reads the database configuration for all of the feeds to determines how often to update each one. Periodic updates are scheduled on different goroutines. The scheduler guarantees that for each feed there is only one update happening at a time.

The scheduler is a highly concurrent component. Not only is it scheduling multiple feed updates at the same time, but can also receive concurrent reset requests to update the periodic update configuration it is using. For example, after a system is installed the scheduler is instructed to reset the periodic update configuration for the feeds in that system.

To satisfy my own intellectual itch, the scheduler is implemented solely using the "concurrent sequential processes" model. The only concurrency primitive used is the Go channel. In the package there are actually three types of schedulers: a single root scheduler (type `Scheduler`) which contains 0 or more system schedulers (one per system), each of which in turns contains 1 or more "feed schedulers" (one per feed in the system). Each scheduler has its own `run` function which loops indefinitely reading messages on different channels. The scheduler will do different things depending on the message it receives: "things" include triggering a new feed update, shutting down, or resetting the periodic update configuration.

As an example consider the process of resetting the configuration for a particular system after the system is updated. This process is triggered using the `Reset` function. That function sends a message on the right channel to the main run loop of the root scheduler. The root scheduler reads the message and creates the system scheduler if needed. It thens send a message to the system scheduler run loop with the new configuration. That run loop reads the message and propagates it to the feed schedulers contained inside.

Index

• func NormalizeSchedulingPolicy(logger *slog.Logger, feedConfig *api.FeedConfig, position int, ...)
• type DefaultScheduler
  • func NewDefaultScheduler() *DefaultScheduler
  • func (s *DefaultScheduler) Reset(ctx context.Context) error
  • func (s *DefaultScheduler) ResetSystem(ctx context.Context, systemID string) error
  • func (s *DefaultScheduler) Run(ctx context.Context, public api.PublicServer, admin api.AdminServer, ...)
  • func (s *DefaultScheduler) Status(ctx context.Context) ([]FeedStatus, error)
• type FeedStatus
• type Scheduler
  • func NoOpScheduler(logger *slog.Logger) Scheduler

Functions

func NormalizeSchedulingPolicy

func NormalizeSchedulingPolicy(logger *slog.Logger, feedConfig *api.FeedConfig, position int, agencies []*api.Agency)

NormalizeSchedulingPolicy normalizes the scheduling policy fields in the feed config.

After this function runs the following guarantees are in place:

• The scheduling policy is NONE, PERIODIC or DAILY. In particular, the DEFAULT strategy does not appear.

• If the scheduling policy is PERIODIC, the `periodic_update_period_ms` field has a valid value.

• If the scheduling policy is DAILY, the `daily_update_time` and `daily_update_timezone` fields have valid values. For the timezone field, this means the field contains a timezone that the current OS recognizes.

Types

type DefaultScheduler

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

DefaultScheduler is Transiter's default scheduler.

func NewDefaultScheduler

func NewDefaultScheduler() *DefaultScheduler

NewDefaultScheduler creates a new instance of the default scheduler.

This scheduler must be run using the `Run` method.

func (*DefaultScheduler) Reset

func (s *DefaultScheduler) Reset(ctx context.Context) error

func (*DefaultScheduler) ResetSystem

func (s *DefaultScheduler) ResetSystem(ctx context.Context, systemID string) error

func (*DefaultScheduler) Run

func (s *DefaultScheduler) Run(ctx context.Context, public api.PublicServer, admin api.AdminServer, logger *slog.Logger)

Run runs the scheduler until the provided context is cancelled.

func (*DefaultScheduler) Status

func (s *DefaultScheduler) Status(ctx context.Context) ([]FeedStatus, error)

type FeedStatus

type FeedStatus struct {
	// ID of the system.
	SystemID string
	// The feed configuration being used.
	FeedConfig *api.FeedConfig
	// Update period.
	Period time.Duration
	// Time of the last update that finished.
	LastFinishedUpdate time.Time
	// Time of the last update that finished successfully.
	LastSuccessfulUpdate time.Time
	// Whether an update for this feed is currently running.
	CurrentlyRunning bool
}

FeedStatus describes the status of one feed within a scheduler.

type Scheduler

type Scheduler interface {
	// Reset resets the scheduler with the current periodic update configuration for all systems.
	Reset(ctx context.Context) error
	// ResetSystem resets the scheduler for the provided system.
	ResetSystem(ctx context.Context, systemID string) error
	// Status returns a list of feed statuses for all feeds being periodically updated by the scheduler.
	Status(ctx context.Context) ([]FeedStatus, error)
}

Scheduler defines the API of a Transiter scheduler.

The methods of this interface are invoked in the Transiter admin service.

func NoOpScheduler

func NoOpScheduler(logger *slog.Logger) Scheduler

NoOpScheduler returns a scheduler that does nothing.

It is used when Transiter is running without a scheduler.