Documentation
¶
Overview ¶
Package graph contains the actual implementation of the resource graph engine that runs the graph of resources in real-time. This package has the algorithm that runs all the graph transitions.
Index ¶
- Constants
- func ClearRecv(res engine.Res)
- func SemaSize(id string) int
- func TypeCmp(a, b reflect.Value) error
- func UpdatedStrings(updated map[engine.RecvableRes]map[string]*engine.Send) []string
- type Engine
- func (obj *Engine) Abort() error
- func (obj *Engine) Apply(fn func(*pgraph.Graph) error) error
- func (obj *Engine) AutoEdge(ctx context.Context) error
- func (obj *Engine) AutoGroup(ctx context.Context, ag engine.AutoGrouper) error
- func (obj *Engine) BadTimestamps(vertex pgraph.Vertex) []pgraph.Vertex
- func (obj *Engine) Commit(ctx context.Context) error
- func (obj *Engine) Graph() *pgraph.Graph
- func (obj *Engine) HardInterrupt() error
- func (obj *Engine) Init() error
- func (obj *Engine) IsClosing() bool
- func (obj *Engine) Load(newGraph *pgraph.Graph) error
- func (obj *Engine) OKTimestamp(vertex pgraph.Vertex) bool
- func (obj *Engine) Pause() error
- func (obj *Engine) Process(ctx context.Context, vertex pgraph.Vertex) error
- func (obj *Engine) RefreshPending(vertex pgraph.Vertex) bool
- func (obj *Engine) Resume() error
- func (obj *Engine) ReversalList() (map[string]string, error)
- func (obj *Engine) Reversals(ctx context.Context) error
- func (obj *Engine) SendRecv() error
- func (obj *Engine) SetDownstreamRefresh(vertex pgraph.Vertex, b bool)
- func (obj *Engine) SetFastPause(b bool)
- func (obj *Engine) SetUpstreamRefresh(vertex pgraph.Vertex, b bool)
- func (obj *Engine) Shutdown() error
- func (obj *Engine) SoftInterrupt()
- func (obj *Engine) StartBackground(ctx context.Context, kind string) error
- func (obj *Engine) StopBackground(kind string) error
- func (obj *Engine) Validate() error
- func (obj *Engine) Worker(vertex pgraph.Vertex) error
- type Exporter
- type RecvFn
- type State
- func (obj *State) Cleanup() error
- func (obj *State) Init() error
- func (obj *State) Pause() error
- func (obj *State) Poke()
- func (obj *State) Resume() error
- func (obj *State) ReversalCleanup() error
- func (obj *State) ReversalDelete() error
- func (obj *State) ReversalInit() error
- func (obj *State) ReversalWrite(str string, overwrite bool) error
Constants ¶
const ( // ReverseFile is the file name in the resource state dir where any // reversal information is stored. ReverseFile = "reverse" // ReversePerm is the permissions mode used to create the ReverseFile. ReversePerm = 0600 )
const SemaSep = ":"
SemaSep is the trailing separator to split the semaphore id from the size.
const ( // StateDir is the name of the sub directory where all the local // resource state is stored. StateDir = "state" )
Variables ¶
This section is empty.
Functions ¶
func ClearRecv ¶
ClearRecv turns off the changed flags on the receive keys of this resource, and of any grouped resources found within it. The engine runs this after a CheckApply which returned without erroring, since that resource has now had its chance to consume those notifications. If CheckApply is skipped or if it errors and gets retried, then we must not run this, or the notification would be lost, because SendRecv won't produce it a second time. See the Send struct for more information on why these flags are sticky.
func SemaSize ¶
SemaSize returns the size integer associated with the semaphore id. It defaults to 1 if not found.
func TypeCmp ¶
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!
func UpdatedStrings ¶
UpdatedStrings returns a list of strings showing what was updated after a Send/Recv run returned the updated datastructure. This is useful for logs.
Types ¶
type Engine ¶
type Engine struct {
Program string
Version string
Hostname string
// Break off separate logical pieces into chunks where possible.
Converger *converger.Coordinator
Exporter *Exporter
Local *local.API
World engine.World
// TODO: remove Cancel from here since it's part of Local now?
Cancel context.CancelCauseFunc
// Prefix is a unique directory prefix which can be used. It should be
// created if needed.
Prefix string
Debug bool
Logf func(format string, v ...interface{})
// contains filtered or unexported fields
}
Engine encapsulates a generic graph and manages its operations. The engine starts off running (unpaused) but with an empty graph. It then cycles through being paused, gets vertices added, and then (unpause) resuming. The vertices have state which includes the running Worker (Watch for Res) and those start paused, which resume and start up after Commit. Getting the architectural dance right and knowing what starts paused vs running was tricky, but I think it's correct now.
func (*Engine) Abort ¶
Abort the pending graph and any work in progress on it. After this call you may Load a new graph.
func (*Engine) Apply ¶
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 something does not succeed.
func (*Engine) BadTimestamps ¶
BadTimestamps returns the list of vertices that are causing our timestamp to be bad. A prerequisite is bad if it hasn't completed a Process since we last ran. A prerequisite with a Process in flight zeroes its own timestamp for the duration, which makes it bad for everyone, including a vertex which has never run, such as one just added by a graph swap. Otherwise the zero timestamp of that vertex would let it start while a prerequisite is in the middle of a CheckApply, and before that work is done.
func (*Engine) Commit ¶
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) HardInterrupt ¶
Interrupt asks every resource which supports it to unblock whatever long running operation it might be in the middle of. It is the last resort before a kill -9, and it is usually run on the third ^C, after SoftInterrupt already cancelled the contexts and something still hasn't let go. It only helps for resources which block on something their context can't reach, such as a child process, so most resources don't implement it. It may leave those which do in a partial or unknown state, which is why it's the final rung and not an earlier one. This never blocks: a resource which can't be interrupted returns an error, which we log and step past.
func (*Engine) Init ¶
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) IsClosing ¶
IsClosing tells the caller if a Shutdown() was run. This is helpful so that the graph can behave slightly differently when receiving the final empty graph. This is because it's empty because we passed one to unload everything, not because the user actually removed all resources. We may want to preserve the exported state for example, and not purge it.
func (*Engine) Load ¶
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 ¶
OKTimestamp returns true if this vertex can run right now.
func (*Engine) Process ¶
Process is the primary function to execute a particular vertex in the graph.
func (*Engine) RefreshPending ¶
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) Resume ¶
Resume un-pauses the active graph. Very little that is interesting should happen here. It all happens in the Commit method. During Commit, the Worker method starts which in turn causes Watch to start, however the main body of the Worker starts in a paused mode. It waits for the resume signal before it lets CheckApply run for the first time. No CheckApply methods may run when we are paused. This also needs to Resume any pre-existing resources. Do not call this concurrently with the Pause method.
func (*Engine) ReversalList ¶
ReversalList returns all the available pending reversal data on this host. It can then be decoded by whatever method is appropriate for.
func (*Engine) Reversals ¶
Reversals adds the reversals onto the loaded graph. This should happen last, and before Commit. TODO: use the ctx? maybe one day wrap the file i/o stuff
func (*Engine) SendRecv ¶
SendRecv runs the engine invocation during graph swap on the new graph with data from the old graph, so that we won't need to unnecessarily re-make a resource that had previously received some data and is now different than the equivalent resource in this new incoming graph!
func (*Engine) SetDownstreamRefresh ¶
SetDownstreamRefresh sets the refresh value to any downstream vertices.
func (*Engine) SetFastPause ¶
SetFastPause puts the graph into fast pause mode. 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. XXX: consider moving this into SoftInterrupt and nuking it from the API.
func (*Engine) SetUpstreamRefresh ¶
SetUpstreamRefresh sets the refresh value to any upstream vertices.
func (*Engine) Shutdown ¶
Shutdown the engine. Engine must be already paused before this is run. It is actually just a Load of an empty graph and a Commit. It waits for all the resources to exit before returning.
func (*Engine) SoftInterrupt ¶
func (obj *Engine) SoftInterrupt()
SoftInterrupt cancels the context of every Process which is currently running, and stops any new ones from starting. It is usually run on the second ^C, once a pause has been requested and is taking longer than the user is willing to wait for. A pause can only complete between two Process runs, so without this the exit is at the mercy of whichever CheckApply happens to be in flight, and the context that one received is not otherwise cancelled until after that same pause has finished. Every resource is required to return promptly when its context is cancelled, so this is enough for all of the well behaved ones. See the Interrupt method for those which aren't, or which block on something a context can't reach. This also flips the fast pause to true which should hopefully prevent poke's from propagating.
This is a one-way latch, since it is only ever used on the way out.
func (*Engine) StartBackground ¶
StartBackground starts up the background function for this kind, if one isn't already running. If you cancel this context, it's because you want to abort the startup and exit everything early. That function may error after startup. This function isn't thread-safe, because it's currently only called linearly from inside the Commit function, which uses it sequentially.
func (*Engine) StopBackground ¶
StopBackground stops the background function for this kind if one is running. This doesn't take a context, because it's a shutdown procedure, and cancelling this kind of scenario is something we'd do on shutdown anyways... This function isn't thread-safe, because it's currently only called linearly from inside the Commit function, which uses it sequentially.
func (*Engine) Validate ¶
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 ¶
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. This function cannot be "re-run" for the same vertex. The retry mechanism stuff happens inside of this. To actually "re-run" you need to remove the vertex and build a new one. The engine guarantees that we do not allow CheckApply to run while we are paused. That is enforced here. The one exception is an async resource, whose CheckApply may keep running across a pause. See asyncCheckApply for how that stays safe.
type Exporter ¶
type Exporter struct {
// Watch specifies if we want to enable the additional watch feature. It
// should probably be left off unless we're debugging something or using
// weird environments where we expect someone to mess with our res data.
Watch bool
World engine.World
Debug bool
Logf func(format string, v ...interface{})
// contains filtered or unexported fields
}
Exporter is the main engine mechanism that sends the exported resource data to the World database. The code is relatively succinct, but slightly subtle.
func (*Exporter) Export ¶
Export performs the worldly export, and then stores the resource unique ID in our in-memory data store. Exported resources use this tracking to know when to run their cleanups. If this function encounters an error, it returns (false, err). If it does nothing it returns (true, nil). If it does work it return (false, nil). These return codes match how CheckApply returns. This may run concurrently by multiple different resources, so as a result it must stay thread safe.
func (*Exporter) Prune ¶
Prune removes any exports which are no longer actively being presented in the resource graph. This cleans things up between graph swaps. This should NOT run if we're shutting down cleanly. Keep in mind that this must act on the new graph which is available by "Commit", not before we're ready to "Commit".
type RecvFn ¶
RecvFn represents a custom Recv function which can be used in place of the stock, built-in one. This is needed if we want to receive from a different resource data source than our own. (Only for special occasions of course!)
type State ¶
type State struct {
// Graph is a pointer to the graph that this vertex is part of.
Graph *pgraph.Graph
// 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
Version string
Hostname string
Converger *converger.Coordinator
Local *local.API
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) Cleanup ¶
Cleanup 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) Pause ¶
Pause pauses this resource. It must not be called on any already paused live resource. It will block until the resource pauses with an acknowledgment, or until an exit for that resource is seen. If the latter happens it will error, regardless of the last pause state. It must not be called concurrently with either the Resume() method or itself, so only call these one at a time and alternate between the two.
func (*State) Poke ¶
func (obj *State) Poke()
Poke sends a notification on the poke channel. This channel is used to notify the Worker to run the Process/CheckApply when it can. This is used when there is a need to schedule or reschedule some work which got postponed or dropped. This doesn't contain any internal synchronization primitives or wait groups, callers are expected to make sure that they don't leave any of these running by the time the Worker() shuts down.
func (*State) Resume ¶
Resume unpauses this resource. It can and must be called once on a brand-new resource that has just started running as they start paused. It must not be called concurrently with either the Pause() method or itself, so only call these one at a time and alternate between the two.
func (*State) ReversalCleanup ¶
ReversalCleanup performs the reversal shutdown steps if necessary for this resource.
func (*State) ReversalDelete ¶
ReversalDelete removes the reversal state information for this resource.
func (*State) ReversalInit ¶
ReversalInit performs the reversal initialization steps if necessary for this resource.
Source Files
Directories