Help us shape the future of Go! Take the 2026 Go Developer Survey

dag

package
v0.1.5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 14, 2026 License: MIT Imports: 27 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	EventDagStarted      = "dag.started"
	EventDagCompleted    = "dag.completed"
	EventDagFailed       = "dag.failed"
	EventNodeSubmitted   = "dag.node.submitted"
	EventNodeRunning     = "dag.node.running"
	EventNodeCompleted   = "dag.node.completed"
	EventNodeFailed      = "dag.node.failed"
	EventNodeStateChange = "dag.node.state_changed"
)
View Source 
const (
	StatusPending   = "pending"
	StatusReady     = "ready"
	StatusSubmitted = "submitted"
	StatusRunning   = "running"
	StatusStopping  = "stopping"
	StatusStopped   = "stopped"
	StatusDone      = "done"
	StatusFailed    = "failed"
	StatusCached    = "cached"
	StatusSkipped   = "skipped"
)
View Source 
const DefaultNodeOutputRole = "DEFAULT"

DefaultNodeOutputRole is the role an output_patterns entry is recorded under when it declares none. It is the analysis_result key the frontend reads for node output files that cannot be matched to any item's accept_formats (ComponentsRender resolves them as dataMap["DEFAULT"]).

View Source 
const ReasonStoppedByUser = "dag stopped by user"

ReasonStoppedByUser is the standard stop reason written to node error_message, shared so every scheduler reports the same text for a user-requested stop.

Variables

This section is empty.

Functions

func ApplyRequestParamParse added in v0.1.5 

func ApplyRequestParamParse(ctx context.Context, parser NodeRequestParser, target *types.AnalysisNode) bool

ApplyRequestParamParse parses target.request_param through the injected parser and overwrites the matching declared resolved_inputs. It returns true when the target was modified.

It is the request_param-source counterpart of ApplySelectAssayBindings: where that one needs a resolved upstream edge (and its recorded file id) to know what to rewrite, this one derives resolved_inputs straight from request_param. It exists so a node whose inputs were written as raw request values - for example a scatter(each) item's per-handle value, keyed by the input name - can be turned into the structured recipe the script form expects, without requiring any upstream graph snapshot. parser may be nil, in which case nothing changes.

func ApplySelectAssayBindings added in v0.1.5 

func ApplySelectAssayBindings(ctx context.Context, parser NodeRequestParser, target *types.AnalysisNode, bindings []UpstreamInputBinding) bool

ApplySelectAssayBindings rewrites target.request_param into the original input form ({"files": "<go_file id>"}) for every SelectAssay input that has a recorded upstream file id, then replaces the matching resolved_inputs with the structured recipe produced by the parser. It returns true when the target was modified.

target is mutated in place (RequestParam and ResolvedInputs). parser may be nil, in which case only request_param is rewritten and resolved_inputs keeps the merged raw value.

func CanTransition 

func CanTransition(from string, to string) bool

func EnsureTransition 

func EnsureTransition(from string, to string) error

func IsSuccessStatus 

func IsSuccessStatus(status string) bool

func IsTerminalStatus 

func IsTerminalStatus(status string) bool

func OutputFileIDFor added in v0.1.5 

func OutputFileIDFor(node *types.AnalysisNode, handle string) string

OutputFileIDFor returns the recorded go_file id of one output handle of a node, or an empty string when none was recorded.

func RecoverDag added in v0.1.4 

func RecoverDag(
	cfg *config.Config,
	repo interfaces.AnalysisRepository,
	schedulers *scheduler.Registry,
)

RecoverDag is the single recovery entry for every DAG scheduler.

It scans job_status running/stopping analyses in a background goroutine and hands each one back to the scheduler that owns it, resolved from analysis.scheduler_mode through the shared registry. Because ownership is a property of the analysis, one analysis can never be advanced by two schedulers at the same time.

Each orchestrator's RecoverRunningAnalyses handles a single analysis: running triggers a resume, stopping converges the stop request, so this loop does not need a separate cycle per job status.

func ResumeNodeStatusForRestart added in v0.1.4 

func ResumeNodeStatusForRestart(status string, hasContainer bool) (string, bool)

ResumeNodeStatusForRestart decides how an in-flight node should be treated after the owning process restarted.

It is shared by every scheduler that resumes work from the analysis_nodes table (legacy DagOrchestrator and the dynamic V2 orchestrator) so that restart semantics stay identical across scheduling paths.

Return values:

  • (StatusReady, true): the node lost its runner, it must be re-dispatched.
  • ("", false): keep the current status. This covers nodes that were never in-flight, and running nodes whose container is still alive — those are expected to be finalized by deferred reconciliation instead.

func SchedulerEventFilter added in v0.1.4 

func SchedulerEventFilter(evt RuntimeEvent) bool

SchedulerEventFilter is the delivery whitelist for dynamic scheduler sinks.

Only node terminal transitions can change dependency state, so only they can produce new schedulable work. Everything else the bus carries for the same analysis - node.submitted (the scheduler's own echo from pumpReadyQueue), node.running, node.state_changed, and the dag.* lifecycle events the scheduler publishes to itself - would only wake the loop for a no-op iteration that re-reads the runtime snapshot. Dropping them before the buffer also keeps the 256-entry window free for the events that actually matter.

This only narrows what a sink receives; the bus is untouched, so other subscribers (for example the realtime notifier) still observe every event.

func SortNodesByStartedAt 

func SortNodesByStartedAt(nodes []*types.AnalysisNode)

Types

type AnalysisEventSink added in v0.1.4 

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

AnalysisEventSink is the per-analysis runtime event mailbox owned by one running scheduler.

Its lifetime matches exactly one scheduler run: EventRouter.Register creates it when a run is accepted and EventRouter.Unregister retires it when the run goroutine exits.

Design notes:

  • The events channel is never closed. Retiring is signalled through the `closed` flag instead, which removes the classic "send on closed channel" race between the bus worker goroutine and the scheduler teardown.
  • Enqueue is non-blocking by contract because it runs on the shared bus worker goroutine.

func NewAnalysisEventSink added in v0.1.4 

func NewAnalysisEventSink() *AnalysisEventSink

NewAnalysisEventSink builds a detached sink that accepts every runtime event.

Schedulers normally obtain their sink from EventRouter.Register. This constructor exists for tests and for schedulers that run without a registered router.

func (*AnalysisEventSink) ConsumeDirty added in v0.1.4 

func (s *AnalysisEventSink) ConsumeDirty() bool

ConsumeDirty reports whether events may have been lost, clearing the flag.

func (*AnalysisEventSink) Enqueue added in v0.1.4 

func (s *AnalysisEventSink) Enqueue(evt RuntimeEvent)

Enqueue delivers one runtime event without ever blocking the caller.

Events rejected by the sink's keep predicate are dropped before they can take a buffer slot, so they neither displace a real event nor trip the overflow path.

func (*AnalysisEventSink) Events added in v0.1.4 

func (s *AnalysisEventSink) Events() <-chan RuntimeEvent

Events exposes the receive side of the event mailbox.

A nil sink returns a nil channel so a select arm on it blocks forever instead of panicking; callers are expected to have substituted a live sink.

func (*AnalysisEventSink) Wake added in v0.1.4 

func (s *AnalysisEventSink) Wake() <-chan struct{}

Wake exposes the overflow wake signal. It is closed-never and only ever carries a single pending notification.

type CacheDecision added in v0.1.4 

type CacheDecision struct {
	// Rerun requests re-materialization of the node as ready.
	Rerun bool
	// Reason is persisted on the node for observability.
	Reason string
}

CacheDecision tells the reconciler whether the existing node must rerun.

type CacheFacts added in v0.1.4 

type CacheFacts struct {
	// CacheType is the analysis cache_type that selected the policy.
	CacheType int
	// Existing is the node persisted by a previous run.
	Existing *types.AnalysisNode
	// ProbeCommandMD5 is the fingerprint of the run script that would run now.
	// It is only populated when the policy asks for a fingerprint.
	ProbeCommandMD5 string
	// ProbeParamsMD5 is the fingerprint of the params payload that would be used.
	ProbeParamsMD5 string
}

CacheFacts is the immutable input of a cache policy decision.

type CachePolicy added in v0.1.4 

type CachePolicy interface {
	// Name is a stable identifier used in logs and diagnostics.
	Name() string
	// ResetsGraph reports whether this cache type invalidates the whole persisted
	// graph before a run.
	//
	// It is the analysis-level half of a cache type, and the only place a reset may be
	// requested from: Decide below is always called with a node that survived that
	// reset, so a per-node rerun answer can never be mistaken for a graph-level one.
	ResetsGraph() bool
	// RequiresFingerprint reports whether Decide needs the probe fingerprints.
	RequiresFingerprint() bool
	// Decide returns the rerun decision for an already persisted node.
	Decide(facts CacheFacts) CacheDecision
}

CachePolicy is the Strategy implementation behind analysis.cache_type.

Implementations must be stateless (they may be shared across runs) and must not perform I/O: the reconciler prepares fingerprints and hands them over.

type CachePolicyRegistry added in v0.1.4 

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

CachePolicyRegistry resolves a CachePolicy per analysis cache type.

func NewCachePolicyRegistry added in v0.1.4 

func NewCachePolicyRegistry() *CachePolicyRegistry

NewCachePolicyRegistry builds the default registry covering every cache type declared in types.Analysis.

func (*CachePolicyRegistry) Register added in v0.1.4 

func (r *CachePolicyRegistry) Register(cacheType int, policy CachePolicy)

Register binds a policy to a cache type, replacing any previous entry.

func (*CachePolicyRegistry) Resolve added in v0.1.4 

func (r *CachePolicyRegistry) Resolve(cacheType int) CachePolicy

Resolve returns the policy for cacheType, falling back to the safe default.

func (*CachePolicyRegistry) ShouldResetGraph added in v0.1.4 

func (r *CachePolicyRegistry) ShouldResetGraph(cacheType int) bool

ShouldResetGraph reports whether cacheType invalidates the whole persisted graph before a run.

It is the analysis-level counterpart of the per-node policies: only rerun_all resets the graph, while every reuse policy keeps it and defers its decision to per-node reuse time.

type DagRuntimeEventNotifier 

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

DagRuntimeEventNotifier subscribes DAG runtime events and pushes UI action messages to project users through the unified realtime hub.

func NewDagRuntimeEventNotifier 

func NewDagRuntimeEventNotifier(db *gorm.DB, projectRepo interfaces.ProjectRepository, analysisRepo interfaces.AnalysisRepository, hub *realtime.Hub) *DagRuntimeEventNotifier

func (*DagRuntimeEventNotifier) Handle 

func (n *DagRuntimeEventNotifier) Handle(evt event.Event)

type DagScheduler 

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

func NewDagScheduler 

func NewDagScheduler(analysisID int64, runtime *RuntimeEngine, dispatcher *NodeDispatcher, bus event.Bus, cfg SchedulerConfig) *DagScheduler

func (*DagScheduler) Run 

func (s *DagScheduler) Run(ctx context.Context) (*SchedulerResult, error)

type EventRouter added in v0.1.4 

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

EventRouter is the single long-lived event.Bus subscriber for the whole process. It routes RuntimeEvent values to the sink of the owning analysis.

Why a router instead of per-run subscriptions:

  1. A per-run bus.Subscribe adds a permanent subscriber goroutine per run because event.Bus has no Unsubscribe.
  2. OrderedMemoryBus.Publish writes to every subscriber queue with a blocking send, so the blocking surface grows linearly with the number of concurrent runs and every subscriber is woken by every other analysis' events.

With the router there is exactly one subscriber and routing is an O(1) map lookup.

The instance is created once by the DI container (internal/container) and injected into every scheduler that needs runtime events.

func NewEventRouter added in v0.1.4 

func NewEventRouter() *EventRouter

NewEventRouter builds an empty router. The caller is responsible for subscribing it to the event bus exactly once.

func (*EventRouter) Handle added in v0.1.4 

func (r *EventRouter) Handle(evt event.Event)

Handle implements event.Handler.

It must stay O(1) and strictly non-blocking: a slow or wedged analysis must never stall the shared bus worker or any other subscriber.

func (*EventRouter) Register added in v0.1.4 

func (r *EventRouter) Register(analysisID int64) *AnalysisEventSink

Register creates the sink for analysisID, or returns the existing one when the analysis is already registered. The sink accepts every runtime event.

func (*EventRouter) RegisterWithFilter added in v0.1.4 

func (r *EventRouter) RegisterWithFilter(analysisID int64, keep EventSinkFilter) *AnalysisEventSink

RegisterWithFilter is Register plus a delivery filter: the sink only ever receives events for which keep returns true. A nil keep accepts everything, which is exactly Register's behaviour.

Filtering happens before the buffer, so filtered-out events never occupy a slot and the scheduler loop is never woken for them. The bus itself is not touched, so other subscribers still see every event.

func (*EventRouter) Unregister added in v0.1.4 

func (r *EventRouter) Unregister(analysisID int64)

Unregister retires and removes the sink for analysisID. It is safe to call more than once and safe to call for an unknown analysis.

The map entry is removed before retiring so a concurrent Register for the same analysis always obtains a fresh, live sink.

type EventSinkFilter added in v0.1.4 

type EventSinkFilter func(RuntimeEvent) bool

EventSinkFilter decides whether a runtime event is worth delivering to a sink.

It runs on the shared bus worker goroutine, so it must stay cheap and must never block.

type NodeArtifactFingerprinter added in v0.1.4 

type NodeArtifactFingerprinter interface {
	// Fingerprint materializes the node artifacts and fills in CommandMD5 / ParamsMD5.
	Fingerprint(ctx context.Context, node *types.AnalysisNode) error
}

NodeArtifactFingerprinter predicts a node's runtime artifacts and records their digests on the node.

It is deliberately narrower than prepare.NodeRuntimePreparer: a scheduler only needs the digests to decide whether a persisted node can be reused, it must not own execution-time preparation (output cleanup, script symlinks, ...) which stays with the dispatcher. Depending on the full preparer here is what made every materialized node look like it had to be prepared by the scheduler too.

It is shared by every scheduler that runs the fingerprint cache policies (dynamic V2 and dataflow V3), so a probe is always rendered the same way no matter which scheduler evaluates the cache.

type NodeCompletionCoordinator 

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

func NewNodeCompletionCoordinator 

func NewNodeCompletionCoordinator(
	analysisRepo interfaces.AnalysisRepository,
	containerRepo interfaces.ContainerRepository,
	containerOps NodeContainerOperator,
	dataRepo interfaces.DataRepository,
	bus event.Bus,
	cfg *config.Config,
	requestParser NodeRequestParser,
) *NodeCompletionCoordinator

func (*NodeCompletionCoordinator) Handle 

func (c *NodeCompletionCoordinator) Handle(evt event.Event)

type NodeContainerOperator 

type NodeContainerOperator interface {
	Delete(ctx context.Context, id int64) error
}

type NodeDispatcher 

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

func NewNodeDispatcher 

func NewNodeDispatcher(
	repo interfaces.AnalysisRepository,
	bus event.Bus,
	factory *executor.ExecuterFactory,
	preparer prepare.NodeRuntimePreparer,
	requestParser NodeRequestParser,

) *NodeDispatcher

func (*NodeDispatcher) Dispatch 

func (d *NodeDispatcher) Dispatch(ctx context.Context, analysisNodeID int64) error

func (*NodeDispatcher) Stop 

func (d *NodeDispatcher) Stop(ctx context.Context, node *types.AnalysisNode) (*executor.Result, error)

type NodeFailureCleanupFunc 

type NodeFailureCleanupFunc func(ctx context.Context, node *types.AnalysisNode)

type NodeRequestParser added in v0.1.5 

type NodeRequestParser interface {
	ParseRequestParam(ctx context.Context, scriptID int64, requestParam map[string]any) (map[string]any, error)
}

NodeRequestParser 把节点的 request_param（原始输入，例如 `{"handle": {"files": "<id>"}}`）依据脚本表单解析成结构化配方，与静态分析保存时 的解析结果一致（handler 侧复用 analysisform.BuildParseAnalysisResult 实现）。

由上层实现并通过 SetRequestParser 注入；未注入（nil）时跳过解析，resolved_inputs 保留原始值。

consumer-side 接口：只在 runtime 包定义，实现放在 handler 包，避免 dag -> handler 的循环依赖。

type NodeStopSweeper added in v0.1.4 

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

NodeStopSweeper owns "stop the nodes of one analysis", the single implementation shared by every scheduler (legacy dag, dynamic, dataflow).

Before this type existed each orchestrator carried its own copy of the stop sequence, and they had already drifted apart (one wrote failed, another stopped, and none of them stopped the containers). Keeping one implementation means the meaning of "a stopped analysis" cannot depend on which scheduler happened to run.

It is stateless apart from its timeout, so one instance can be shared across concurrent runs of different analyses.

func NewNodeStopSweeper added in v0.1.4 

func NewNodeStopSweeper(repo interfaces.AnalysisRepository, dispatcher *NodeDispatcher) *NodeStopSweeper

NewNodeStopSweeper builds the shared stop helper. A nil dispatcher is tolerated: the sweep then only latches node state instead of failing every call.

func (*NodeStopSweeper) MarkNodesStopped added in v0.1.4 

func (s *NodeStopSweeper) MarkNodesStopped(ctx context.Context, analysisID int64, reason string) error

MarkNodesStopped terminalizes every non-terminal node of the analysis.

It is the mandatory last step of a stop: the analysis is about to be written as a terminal stopped state, and recovery only scans running/stopping analyses, so a node left in "stopping" would never be looked at again. Without this the node would wait forever for a container stop event that may never arrive.

func (*NodeStopSweeper) StopActiveNodes added in v0.1.4 

func (s *NodeStopSweeper) StopActiveNodes(ctx context.Context, analysisID int64, reason string)

StopActiveNodes stops every node of the analysis that is still in flight: it first persists "stopping" (the stop latch), then asks the executor to stop the real process or container behind the node.

The ordering matters. Persisting "stopping" first is what makes the stop convergent even if something is still dispatching: the claim query only ever selects ready rows, and a dispatch that already claimed the node fails the running transition instead of creating a container. Only then is it meaningful to stop containers, because from here on nothing new can appear behind our back.

Only running/submitted nodes are handled: ready/pending nodes never reached an executor, so there is nothing to stop and they are left to MarkNodesStopped. A single failing node is logged, never propagated: the remaining nodes still have containers that must be told to stop.

type PreparerFingerprinter added in v0.1.4 

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

PreparerFingerprinter adapts the runtime preparer to the scheduler-side fingerprinting contract.

The preparer stays the single source of truth for run.sh / params.json rendering, so a scheduler borrows it to predict the artifacts instead of re-implementing the rendering rules (and drifting from the dispatcher).

func NewPreparerFingerprinter added in v0.1.4 

func NewPreparerFingerprinter(preparer prepare.NodeRuntimePreparer) *PreparerFingerprinter

NewPreparerFingerprinter wraps a NodeRuntimePreparer.

func (*PreparerFingerprinter) Fingerprint added in v0.1.4 

func (f *PreparerFingerprinter) Fingerprint(ctx context.Context, node *types.AnalysisNode) error

Fingerprint implements NodeArtifactFingerprinter.

Output cleanup is explicitly disabled: a probe runs for nodes that may well be reused as-is, and wiping their output directory would delete the results the downstream nodes still reference. The dispatcher cleans the output directory when it prepares the node for an actual execution.

type RerunAllPolicy added in v0.1.4 

type RerunAllPolicy struct{}

RerunAllPolicy implements types.CacheTypeRerunAll.

rerun_all is an analysis-level strategy: it is expressed by resetting the persisted graph before the run (ResetsGraph), so every node is materialized from scratch.

Per node there is therefore nothing left to disable, and Decide keeps whatever result is reusable - exactly like ReuseExistingPolicy. A node can only still be persisted when the reset was deliberately skipped, which is precisely the resume path, where a restart must not turn into a full rerun from scratch.

Answering "rerun" unconditionally here livelocks a run: the reconciler re-decides every persisted node on every pass, including the nodes this very run has just completed, so each success is immediately flipped back to ready and the graph never reaches a terminal state.

func (RerunAllPolicy) Decide added in v0.1.4 

func (RerunAllPolicy) Decide(facts CacheFacts) CacheDecision

Decide implements CachePolicy.

func (RerunAllPolicy) Name added in v0.1.4 

func (RerunAllPolicy) Name() string

Name implements CachePolicy.

func (RerunAllPolicy) RequiresFingerprint added in v0.1.4 

func (RerunAllPolicy) RequiresFingerprint() bool

RequiresFingerprint implements CachePolicy.

func (RerunAllPolicy) ResetsGraph added in v0.1.4 

func (RerunAllPolicy) ResetsGraph() bool

ResetsGraph implements CachePolicy.

type ReuseExistingPolicy added in v0.1.4 

type ReuseExistingPolicy struct{}

ReuseExistingPolicy keeps whatever is persisted (types.CacheTypeReuseExistingNode).

func (ReuseExistingPolicy) Decide added in v0.1.4 

func (ReuseExistingPolicy) Decide(facts CacheFacts) CacheDecision

Decide implements CachePolicy.

func (ReuseExistingPolicy) Name added in v0.1.4 

func (ReuseExistingPolicy) Name() string

Name implements CachePolicy.

func (ReuseExistingPolicy) RequiresFingerprint added in v0.1.4 

func (ReuseExistingPolicy) RequiresFingerprint() bool

RequiresFingerprint implements CachePolicy.

func (ReuseExistingPolicy) ResetsGraph added in v0.1.4 

func (ReuseExistingPolicy) ResetsGraph() bool

ResetsGraph implements CachePolicy.

type RunningEntry 

type RunningEntry struct {
	AnalysisID     int64
	TaskName       string
	StartedAt      time.Time
	UpdatedAt      time.Time
	MaxConcurrency int
	QueueSize      int
	PollIntervalMs int64
	TimeoutSeconds int64
	Status         string
	StopRequested  bool
	Cancel         context.CancelFunc
}

type RunningRegistry 

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

func NewRunningRegistry 

func NewRunningRegistry() *RunningRegistry

func (*RunningRegistry) Get added in v0.1.4 

func (r *RunningRegistry) Get(analysisID int64) *RunningEntry

Get returns a read-only copy of the running entry for the analysis. The copy has its Cancel func cleared so callers can only inspect state.

func (*RunningRegistry) IsRunning 

func (r *RunningRegistry) IsRunning(analysisID int64) bool

func (*RunningRegistry) IsStopping 

func (r *RunningRegistry) IsStopping(analysisID int64) bool

func (*RunningRegistry) MarkFinished 

func (r *RunningRegistry) MarkFinished(analysisID int64, status string)

func (*RunningRegistry) Register 

func (r *RunningRegistry) Register(entry *RunningEntry)

func (*RunningRegistry) RequestStop 

func (r *RunningRegistry) RequestStop(analysisID int64) bool

type RuntimeEngine 

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

func NewRuntimeEngine 

func NewRuntimeEngine(repo interfaces.AnalysisRepository) *RuntimeEngine

func (*RuntimeEngine) ClaimNextReadyNode 

func (e *RuntimeEngine) ClaimNextReadyNode(ctx context.Context, analysisID int64) (*types.AnalysisNode, error)

func (*RuntimeEngine) CompleteNode 

func (e *RuntimeEngine) CompleteNode(
	ctx context.Context,
	analysisNodeID int64,
	status string,
	resolvedOutputs map[string]any,
	exitCode int,
	errorMessage string,
	outputValidationErrors []string,
	outputFileIDs map[string]string,
) (*types.AnalysisNode, error)

func (*RuntimeEngine) GetSnapshot 

func (e *RuntimeEngine) GetSnapshot(ctx context.Context, analysisID int64) (*RuntimeSnapshot, error)

func (*RuntimeEngine) MarkNodeRunning 

func (e *RuntimeEngine) MarkNodeRunning(ctx context.Context, analysisNodeID int64) (*types.AnalysisNode, error)

func (*RuntimeEngine) RefreshReadyStatus 

func (e *RuntimeEngine) RefreshReadyStatus(ctx context.Context, analysisID int64) error

func (*RuntimeEngine) SetRequestParser added in v0.1.5 

func (e *RuntimeEngine) SetRequestParser(parser NodeRequestParser)

SetRequestParser 注入可选的 request_param 解析器，用于把 propagateOutputs 重写后的 原始输入还原成结构化 resolved_inputs。为 nil 时表示不解析。

type RuntimeEvent 

type RuntimeEvent struct {
	Name           string                 `json:"name"`
	AnalysisID     int64                  `json:"analysis_id,string"`
	AnalysisNodeID int64                  `json:"analysis_node_id,string,omitempty"`
	NodeID         string                 `json:"node_id,omitempty"`
	OccurredAt     time.Time              `json:"occurred_at"`
	Payload        map[string]interface{} `json:"payload,omitempty"`
}

type RuntimeSnapshot 

type RuntimeSnapshot struct {
	AnalysisID        int64          `json:"analysis_id,string"`
	TotalNodes        int            `json:"total_nodes"`
	StatusCount       map[string]int `json:"status_count"`
	CompletedCount    int            `json:"completed_count"`
	CompletionPercent int            `json:"completion_percent"`
	ReadyCount        int            `json:"ready_count"`
	RunningCount      int            `json:"running_count"`
	IsFinished        bool           `json:"is_finished"`
}

type SchedulerConfig 

type SchedulerConfig struct {
	MaxSteps       int
	MaxConcurrency int
	QueueSize      int
	PollInterval   time.Duration
	Timeout        time.Duration
}

type SchedulerResult 

type SchedulerResult struct {
	AnalysisID          int64            `json:"analysis_id"`
	SubmittedCount      int              `json:"submitted_count"`
	FailedToSubmitCount int              `json:"failed_to_submit_count"`
	TimedOut            bool             `json:"timed_out"`
	Snapshot            *RuntimeSnapshot `json:"snapshot"`
}

type ScriptAndParamsFingerprintPolicy added in v0.1.4 

type ScriptAndParamsFingerprintPolicy struct{}

ScriptAndParamsFingerprintPolicy reruns when either the generated run script or the params payload changed (types.CacheTypeReuseWhenScriptAndParamsUnchanged).

func (ScriptAndParamsFingerprintPolicy) Decide added in v0.1.4 

func (ScriptAndParamsFingerprintPolicy) Decide(facts CacheFacts) CacheDecision

Decide implements CachePolicy.

func (ScriptAndParamsFingerprintPolicy) Name added in v0.1.4 

func (ScriptAndParamsFingerprintPolicy) Name() string

Name implements CachePolicy.

func (ScriptAndParamsFingerprintPolicy) RequiresFingerprint added in v0.1.4 

func (ScriptAndParamsFingerprintPolicy) RequiresFingerprint() bool

RequiresFingerprint implements CachePolicy.

func (ScriptAndParamsFingerprintPolicy) ResetsGraph added in v0.1.4 

func (ScriptAndParamsFingerprintPolicy) ResetsGraph() bool

ResetsGraph implements CachePolicy.

type ScriptFingerprintPolicy added in v0.1.4 

type ScriptFingerprintPolicy struct{}

ScriptFingerprintPolicy reruns when the generated run script changed (types.CacheTypeReuseWhenScriptUnchanged).

func (ScriptFingerprintPolicy) Decide added in v0.1.4 

func (ScriptFingerprintPolicy) Decide(facts CacheFacts) CacheDecision

Decide implements CachePolicy.

func (ScriptFingerprintPolicy) Name added in v0.1.4 

func (ScriptFingerprintPolicy) Name() string

Name implements CachePolicy.

func (ScriptFingerprintPolicy) RequiresFingerprint added in v0.1.4 

func (ScriptFingerprintPolicy) RequiresFingerprint() bool

RequiresFingerprint implements CachePolicy.

func (ScriptFingerprintPolicy) ResetsGraph added in v0.1.4 

func (ScriptFingerprintPolicy) ResetsGraph() bool

ResetsGraph implements CachePolicy.

type UpstreamInputBinding added in v0.1.5 

type UpstreamInputBinding struct {
	// TargetHandle is the downstream input handle (the edge target_handle).
	TargetHandle string
	// SourceHandle is the upstream output handle (the edge source_handle).
	SourceHandle string
	// FileID is the go_file primary key recorded for the upstream output handle,
	// empty when the upstream output is not a mirrorable file.
	FileID string
}

UpstreamInputBinding is one resolved incoming edge feeding a downstream input.

func BuildUpstreamInputBindings added in v0.1.5 

func BuildUpstreamInputBindings(incoming []*types.AnalysisEdge, upstream map[string]*types.AnalysisNode) []UpstreamInputBinding

BuildUpstreamInputBindings turns the incoming edges of one node into handle bindings, reading each source node's recorded output file id. Edges without a resolvable source node are skipped; edges whose source produced no file id are kept but carry an empty FileID so callers can decide how to treat them.

type WorkerPool 

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

func NewWorkerPool 

func NewWorkerPool(dispatcher *NodeDispatcher, workers int, queueSize int) *WorkerPool

func (*WorkerPool) Cap added in v0.1.4 

func (p *WorkerPool) Cap() int

Cap reports the pool queue capacity, used by callers as their claim-ahead and backpressure limit.

func (*WorkerPool) Enqueue 

func (p *WorkerPool) Enqueue(nodeID int64) bool

Enqueue is the non-blocking variant: it reports false when the queue has no room left, leaving the retry/backpressure policy to the caller.

func (*WorkerPool) EnqueueWait added in v0.1.4 

func (p *WorkerPool) EnqueueWait(ctx context.Context, nodeID int64) bool

EnqueueWait blocks until nodeID is queued, ctx is cancelled, or the pool stops, and reports whether the node made it into the queue.

It must only be called from the goroutine that owns the pool lifecycle (the one that calls Stop). Stop closes the job channel, and a send on a closed channel panics in select even from the default branch, so a sender running on another goroutine could always race with Stop.

func (*WorkerPool) QueueLen 

func (p *WorkerPool) QueueLen() int

func (*WorkerPool) Start 

func (p *WorkerPool) Start(ctx context.Context)

func (*WorkerPool) Stop 

func (p *WorkerPool) Stop()

Source Files

View all Source files

Directories

Path Synopsis
Package orchestratorv2 implements the dynamic (Nextflow-like) DAG scheduler used by gobrave analyses.
 Package orchestratorv2 implements the dynamic (Nextflow-like) DAG scheduler used by gobrave analyses.
dag
dataflow
dynamic
Package trace provides the append-only, table-shaped run trace shared by the DAG schedulers.
 Package trace provides the append-only, table-shaped run trace shared by the DAG schedulers.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.