runner

Documentation

Index

• type Cluster
• type Phase
  • func (i Phase) String() string
• type Runner
  • func (r *Runner) Allocate(ctx context.Context) error
  • func (r *Runner) Do(ctx context.Context) bool
  • func (r *Runner) Eval(ctx context.Context) (string, error)
• type State
  • func (s *State) Reset()
  • func (s State) String() string
• type StaticCluster
  • func (s *StaticCluster) Allocate(ctx context.Context, req reflow.Requirements, labels pool.Labels) (pool.Alloc, error)
• type Stealer
  • func (s *Stealer) Go(ctx context.Context, e *reflow.Eval)
• type TracingCluster

Types

type Cluster

type Cluster interface {
	pool.Pool

	// Allocate reserves an alloc of at least min, and at most max resources.
	// The cluster may scale elastically in order to meet this demand.
	// Labels are passed down to the underlying pool.
	// Progress is reported to the provided status object.
	Allocate(ctx context.Context, req reflow.Requirements, labels pool.Labels) (pool.Alloc, error)
}

Cluster is a kind of pool.Pool that also allows the user to directly reserve an alloc. This way, the cluster can be responsive to demand.

type Phase

type Phase int

Phase enumerates the possible phases of a run.

const (
	// Init indicates the run is needs initialization.
	Init Phase = iota
	// Eval indicates the run needs evaluation.
	Eval
	// Retry indicates the run needs to be considered for retrying.
	Retry
	// Done indicates the run is complete.
	Done

	// MaxPhase is the maximum value of Phase.
	MaxPhase
)

func (Phase) String

func (i Phase) String() string

type Runner

type Runner struct {
	// State contains the state of the run. The user can serialize
	// this in order to resume runs.
	State

	reflow.EvalConfig

	// Cluster is the main cluster from which Allocs are allocated.
	Cluster Cluster

	// ClusterAux defines the cluster from which capacity
	// for auxilliary workers is allocated. If nil, Cluster is used
	// instead.
	ClusterAux Cluster

	// Flow is the flow to be evaluated.
	Flow *reflow.Flow

	// Type is the type of output. When Type is nil, it is taken to be
	// (legacy) reflow.Fileset.
	Type *types.T

	// Transferer is the transfer manager used for node-to-node data
	// transfers.
	Transferer reflow.Transferer

	// Retain is the amount of time the primary alloc should be retained
	// after failure.
	Retain time.Duration

	// Alloc is the primary alloc in which the flow is evaluated.
	Alloc pool.Alloc

	// Labels are the set of labels affiliated with this run.
	Labels pool.Labels
}

A Runner is responsible for evaluating a reflow.Flow on a cluster. Runners also launch and maintain auxilliary work-stealing allocs, and manages data transfer and failure handling between the primary evaluation alloc and the auxilliary workers.

TODO(marius): introduce a "stealer-only" mode where there is no primary alloc, but with a shared repository (e.g., S3) attached to the Eval.

func (*Runner) Allocate

func (r *Runner) Allocate(ctx context.Context) error

Allocate reserves a new alloc from r.Cluster when r.Alloc is nil.

func (*Runner) Do

func (r *Runner) Do(ctx context.Context) bool

Do steps the runner state machine. Do returns true whenever it can make more progress, thus a caller should call Do in a loop:

for r.Do(ctx) {
	// report progress, save state, etc.
}

func (*Runner) Eval

func (r *Runner) Eval(ctx context.Context) (string, error)

Eval evaluates the flow, returning the resulting Value. In the case of failure, r.Alloc is kept-alive for an additional r.Retain duration.

type State

type State struct {
	// ID is this run's global ID.
	ID digest.Digest
	// Program stores the reflow program name.
	Program string
	// Params is the run parameters
	Params map[string]string
	// Args stores the run arguments
	Args []string
	// Phase holds the current phase of the run.
	Phase Phase
	// AllocID is the full URI for the run's alloc.
	AllocID string
	// AllocInspect is the alloc's inspect output.
	AllocInspect pool.AllocInspect
	// Value contains the result of the evaluation,
	// rendered as a string.
	// TODO(marius): serialize the value into JSON.
	Result string
	// Err contains runtime errors.
	Err *errors.Error
	// NumTries is the number of evaluation attempts
	// that have been made.
	NumTries int
	// LastTry is the timestamp of the last evaluation attempt.
	LastTry time.Time
	// Created is the time of the run's creation.
	Created time.Time
	// Completion is the time of the run's completion.
	Completion time.Time

	// TotalResources stores the total amount of resources used
	// by this run. Note that the resources are in resource-minutes.
	TotalResources reflow.Resources
}

State contains the full state of a run. A State can be serialized and later recovered in order to resume a run.

func (*State) Reset

func (s *State) Reset()

Reset resets the state so that it will reinitialize if run. Run metadata (including its name) are preserved.

func (State) String

func (s State) String() string

String returns a string representation of the state.

type StaticCluster

type StaticCluster struct {
	pool.Pool
}

A StaticCluster implements a pass-through Cluster on top of a pool.Pool.

func (*StaticCluster) Allocate

func (s *StaticCluster) Allocate(ctx context.Context, req reflow.Requirements, labels pool.Labels) (pool.Alloc, error)

Allocate reserves an alloc from the underlying static pool.

type Stealer

type Stealer struct {
	Cluster Cluster
	Log     *log.Logger
	Labels  pool.Labels
}

Stealer is a work-stealer. It periodically queries additional resource requirements from an Eval, attempts to allocate additional allocs from a cluster, and then launches workers that steal work from the same Eval. Work stealers free their allocs when there is no more work to be stolen.

func (*Stealer) Go

func (s *Stealer) Go(ctx context.Context, e *reflow.Eval)

Go polls the Eval e for required resources, allocates new allocs and spins up workers as needed. Go returns when the provided context is complete.

type TracingCluster

type TracingCluster struct {
	Cluster
}

TracingCluster is a cluster that traces the actions of an underlying cluster manager.