graph

Documentation

Index

• Constants
• func SemaSize(id string) int
• func TypeCmp(a, b reflect.Value) error
• type Engine
  • func (obj *Engine) Abort() error
  • func (obj *Engine) Apply(fn func(*pgraph.Graph) error) error
  • func (obj *Engine) AutoEdge() error
  • func (obj *Engine) AutoGroup(ag engine.AutoGrouper) error
  • func (obj *Engine) BadTimestamps(vertex pgraph.Vertex) []pgraph.Vertex
  • func (obj *Engine) Close() error
  • func (obj *Engine) Commit() error
  • func (obj *Engine) Graph() *pgraph.Graph
  • func (obj *Engine) Init() error
  • func (obj *Engine) Load(newGraph *pgraph.Graph) error
  • func (obj *Engine) OKTimestamp(vertex pgraph.Vertex) bool
  • func (obj *Engine) Pause(fastPause bool)
  • func (obj *Engine) Process(vertex pgraph.Vertex) error
  • func (obj *Engine) RefreshPending(vertex pgraph.Vertex) bool
  • func (obj *Engine) SendRecv(res engine.RecvableRes) (map[string]bool, error)
  • func (obj *Engine) SetDownstreamRefresh(vertex pgraph.Vertex, b bool)
  • func (obj *Engine) SetFastPause()
  • func (obj *Engine) SetUpstreamRefresh(vertex pgraph.Vertex, b bool)
  • func (obj *Engine) Start() error
  • func (obj *Engine) Validate() error
  • func (obj *Engine) Worker(vertex pgraph.Vertex) error
• type State
  • func (obj *State) Close() error
  • func (obj *State) Event(kind event.Kind)
  • func (obj *State) Init() error
  • func (obj *State) Poke()

Constants

const SemaSep = ":"

SemaSep is the trailing separator to split the semaphore id from the size.

Functions

func SemaSize

func SemaSize(id string) int

SemaSize returns the size integer associated with the semaphore id. It defaults to 1 if not found.

func TypeCmp

func TypeCmp(a, b reflect.Value) error

TypeCmp compares two reflect values to see if they are the same Kind. It can look into a ptr Kind to see if the underlying pair of ptr's can TypeCmp too!

Types

type Engine

type Engine struct {
	Program  string
	Hostname string
	World    engine.World

	// Prefix is a unique directory prefix which can be used. It should be
	// created if needed.
	Prefix    string
	Converger converger.Converger

	Debug bool
	Logf  func(format string, v ...interface{})
	// contains filtered or unexported fields
}

Engine encapsulates a generic graph and manages its operations.

func (*Engine) Abort

func (obj *Engine) Abort() error

Abort the pending graph and any work in progress on it. After this call you may Load a new graph.

func (*Engine) Apply

func (obj *Engine) Apply(fn func(*pgraph.Graph) error) error

Apply a function to the pending graph. You must pass in a function which will receive this graph as input, and return an error if it something does not succeed.

func (*Engine) AutoEdge

func (obj *Engine) AutoEdge() error

AutoEdge adds the automatic edges to the graph.

func (*Engine) AutoGroup

func (obj *Engine) AutoGroup(ag engine.AutoGrouper) error

AutoGroup runs the auto grouping on the loaded graph.

func (*Engine) BadTimestamps

func (obj *Engine) BadTimestamps(vertex pgraph.Vertex) []pgraph.Vertex

BadTimestamps returns the list of vertices that are causing our timestamp to be bad.

func (*Engine) Close

func (obj *Engine) Close() error

Close triggers a shutdown. Engine must be already paused before this is run.

func (*Engine) Commit

func (obj *Engine) Commit() error

Commit runs a graph sync and swaps the loaded graph with the current one. If it errors, then the running graph wasn't changed. It is recommended that you pause the engine before running this, and resume it after you're done.

func (*Engine) Graph

func (obj *Engine) Graph() *pgraph.Graph

Graph returns the running graph.

func (*Engine) Init

func (obj *Engine) Init() error

Init initializes the internal structures and starts this the graph running. If the struct does not validate, or it cannot initialize, then this errors. Initially it will contain an empty graph.

func (*Engine) Load

func (obj *Engine) Load(newGraph *pgraph.Graph) error

Load a new graph into the engine. Offline graph operations will be performed on this graph. To switch it to the active graph, and run it, use Commit.

func (*Engine) OKTimestamp

func (obj *Engine) OKTimestamp(vertex pgraph.Vertex) bool

OKTimestamp returns true if this vertex can run right now.

func (*Engine) Pause

func (obj *Engine) Pause(fastPause bool)

Pause the active, running graph. At the moment this cannot error.

func (*Engine) Process

func (obj *Engine) Process(vertex pgraph.Vertex) error

Process is the primary function to execute a particular vertex in the graph.

func (*Engine) RefreshPending

func (obj *Engine) RefreshPending(vertex pgraph.Vertex) bool

RefreshPending determines if any previous nodes have a refresh pending here. If this is true, it means I am expected to apply a refresh when I next run.

func (*Engine) SendRecv

func (obj *Engine) SendRecv(res engine.RecvableRes) (map[string]bool, error)

SendRecv pulls in the sent values into the receive slots. It is called by the receiver and must be given as input the full resource struct to receive on. It applies the loaded values to the resource.

func (*Engine) SetDownstreamRefresh

func (obj *Engine) SetDownstreamRefresh(vertex pgraph.Vertex, b bool)

SetDownstreamRefresh sets the refresh value to any downstream vertices.

func (*Engine) SetFastPause

func (obj *Engine) SetFastPause()

SetFastPause puts the graph into fast pause mode. This is usually done via the argument to the Pause command, but this method can be used if a pause was already started, and you'd like subsequent parts to pause quickly. Once in fast pause mode for a given pause action, you cannot switch to regular pause. This is because once you've started a fast pause, some dependencies might have been skipped when fast pausing, and future resources might have missed a poke. In general this is only called when you're trying to hurry up the exit.

func (*Engine) SetUpstreamRefresh

func (obj *Engine) SetUpstreamRefresh(vertex pgraph.Vertex, b bool)

SetUpstreamRefresh sets the refresh value to any upstream vertices.

func (*Engine) Start

func (obj *Engine) Start() error

Start runs the currently active graph. It also un-pauses the graph if it was paused.

func (*Engine) Validate

func (obj *Engine) Validate() error

Validate validates the pending graph to ensure it is appropriate for the engine. This should be called before Commit to avoid any surprises there! This prevents an error on Commit which could cause an engine shutdown.

func (*Engine) Worker

func (obj *Engine) Worker(vertex pgraph.Vertex) error

Worker is the common run frontend of the vertex. It handles all of the retry and retry delay common code, and ultimately returns the final status of this vertex execution.

type State

type State struct {

	// Vertex is the pointer in the graph that this state corresponds to. It
	// can be converted to a `Res` if necessary.
	// TODO: should this be passed in on Init instead?
	Vertex pgraph.Vertex

	Program  string
	Hostname string
	World    engine.World

	// Prefix is a unique directory prefix which can be used. It should be
	// created if needed.
	Prefix string

	// Debug turns on additional output and behaviours.
	Debug bool

	// Logf is the logging function that should be used to display messages.
	Logf func(format string, v ...interface{})
	// contains filtered or unexported fields
}

State stores some state about the resource it is mapped to.

func (*State) Close

func (obj *State) Close() error

Close shuts down and performs any cleanup. This is most akin to a "post" or cleanup command as the initiator for closing a vertex happens in graph sync.

func (*State) Event

func (obj *State) Event(kind event.Kind)

Event sends a Pause or Start event to the resource. It can also be used to send Poke events, but it's much more efficient to send them directly instead of passing them through the resource.

func (*State) Init

func (obj *State) Init() error

Init initializes structures like channels.

func (*State) Poke

func (obj *State) Poke()

Poke sends a nil message on the outputChan. This channel is used by the resource to signal a possible change. This will cause the Process loop to run if it can.