func (self *Pipestance) queryQueue(outerCtx context.Context) {
	prepDone := false
	ctx, task := trace.NewTask(outerCtx, "queryQueue")
	defer func() {
		if !prepDone {
			task.End()
		}
	}()
	if self.node == nil || self.node.top == nil || self.node.top.rt == nil ||
		self.node.top.rt.JobManager == nil ||
		!self.node.top.rt.JobManager.hasQueueCheck() {
		return
	}
	QUEUE_CHECK_LIMIT := 5 * time.Minute
	self.queueCheckLock.Lock()
	if self.queueCheckActive || time.Since(self.lastQueueCheck) < QUEUE_CHECK_LIMIT {
		self.queueCheckLock.Unlock()
		return
	} else {
		self.queueCheckActive = true
		self.queueCheckLock.Unlock()
	}
	needsQuery := make(map[string]*Metadata)
	{
		metas := make(map[*Metadata]bool)  
		nodes := self.node.getFrontierNodes()
		for _, node := range nodes {
			for _, m := range node.collectMetadatas() {
				if !metas[m] {
					if st, ok := m.getState(); ok &&
						(st == Queued || st == Running) &&
						m.exists(JobId) {
						metas[m] = true
						id := m.readRaw(JobId)
						if id != "" {
							needsQuery[id] = m
						}
					}
				}
			}
		}
	}
	if len(needsQuery) == 0 {
		self.queueCheckLock.Lock()
		self.queueCheckActive = false
		self.queueCheckLock.Unlock()
		return
	}
	jobsIn := make([]string, 0, len(needsQuery))
	for id := range needsQuery {
		jobsIn = append(jobsIn, id)
	}
	prepDone = true
	go func(ctx context.Context, task *trace.Task) {
		defer task.End()
		queued, raw := self.node.top.rt.JobManager.checkQueue(jobsIn, ctx)
		for _, id := range queued {
			delete(needsQuery, id)
		}
		if len(needsQuery) > 0 && raw != "" {
			util.LogInfo("runtime",
				"Some jobs thought to be queued were unknown to the job manager.  Raw output:\n%s\n",
				raw)
		}
		if !self.readOnly() {
			for id, m := range needsQuery {
				if m != nil {
					m.failNotRunning(id)
				}
			}
		}
		self.queueCheckLock.Lock()
		self.queueCheckActive = false
		self.lastQueueCheck = time.Now()
		self.queueCheckLock.Unlock()
	}(ctx, task)
}
Check that the queued jobs are actually queued.
package core
import (
	"context"
	"fmt"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"runtime/trace"
	"sync"
	"syscall"
	"time"
	"github.com/martian-lang/martian/martian/syntax"
	"github.com/martian-lang/martian/martian/util"
)
type Stagestance struct {
	node *Node
}
func NewStagestance(parent Nodable, call *syntax.CallGraphStage, srcPaths []string) (*Stagestance, error) {
	self := &Stagestance{}
	self.node = NewNode(parent, call)
	stage := call.Callable().(*syntax.Stage)
	stagecodePath, err := stage.Src.FindPath(srcPaths)
	if err != nil && len(srcPaths) > 0 {
		util.PrintError(err, "runtime", "WARNING: stage code not found")
	}
	self.node.resolvedCmd = stagecodePath
	if self.node.top.rt.Config.StressTest {
		switch self.node.stagecode.Type {
		case syntax.PythonStage:
			self.node.stagecode.Path = util.RelPath(path.Join("..", "adapters", "python", "tester"))
		default:
			return self, fmt.Errorf("Unsupported stress test language: %v", stage.Src.Lang)
		}
	}
	if stage.Resources != nil {
		self.node.resources = &JobResources{
			Threads: float64(stage.Resources.Threads),
			MemGB:   float64(stage.Resources.MemGB),
			VMemGB:  float64(stage.Resources.VMemGB),
			Special: stage.Resources.Special,
		}
	}
	if splits := call.Forks; len(splits) > 0 {
		exps := make([]*syntax.CallStm, len(splits))
		for i, s := range splits {
			exps[i] = s.Call()
		}
		self.node.forkRoots = exps
	}
	return self, self.buildForks()
}
func (self *Stagestance) buildForks() error {
	self.node.buildForks()
	return setupRetains(
		self.node.call.Callable().(*syntax.Stage).Retain,
		self.node.forks)
}
func setupRetains(retain *syntax.RetainParams, forks []*Fork) error {
	if retain != nil {
		for _, param := range retain.Params {
			for _, fork := range forks {
				if fork.fileArgs == nil {
					fork.fileArgs = make(
						map[string]map[Nodable]struct{},
						len(retain.Params))
				}
				if arg := fork.fileArgs[param.Id]; arg == nil {
					fork.fileArgs[param.Id] = map[Nodable]struct{}{
						nil: {},
					}
				} else {
					arg[nil] = struct{}{}
				}
			}
		}
	}
	return nil
}
func (self *Stagestance) getNode() *Node    { return self.node }
func (self *Stagestance) GetFQName() string { return self.node.GetFQName() }
func (self *Stagestance) GetPrenodes() map[string]Nodable {
	return self.node.GetPrenodes()
}
func (self *Stagestance) GetPostNodes() map[string]Nodable {
	return self.node.GetPostNodes()
}
func (self *Stagestance) matchForks(id ForkId) []*Fork {
	return self.node.matchForks(id)
}
func (self *Stagestance) Callable() syntax.Callable {
	return self.node.Callable()
}
type Pipestance struct {
	node     *Node
	metadata *Metadata
	uuid     string
	allNodesCache    []*Node
	queueCheckLock   sync.Mutex
	queueCheckActive bool
	lastQueueCheck   time.Time
}
func (self *Pipestance) OnFinishHook(outerCtx context.Context) {
	if exec_path := self.getNode().top.rt.Config.OnFinishHandler; exec_path != "" {
		ctx, task := trace.NewTask(outerCtx, "onfinish")
		defer task.End()
		util.Println("\nRunning onfinish handler...")
		args := []string{self.GetPath(), string(self.GetState(ctx)), self.node.top.GetPsid()}
		if self.GetState(ctx) == Failed {
			_, _, _, _, _, err_paths := self.GetFatalError()
			if len(err_paths) > 0 {
				err_path, _ := filepath.Rel(filepath.Dir(self.GetPath()), err_paths[0])
				args = append(args, err_path)
			}
		}
		real_path, err := exec.LookPath(exec_path)
		if err != nil {
			util.LogInfo("finishr", "Could not find %v: %v", exec_path, err)
			return
		}
		ectx, cancel := context.WithTimeout(ctx, time.Hour)
		defer cancel()
		cmd := exec.CommandContext(ectx, real_path, args...)
		cmd.Stdin = os.Stdin
		cmd.Stdout = os.Stdout
		cmd.Stderr = os.Stderr
		cmd.SysProcAttr = util.Pdeathsig(
			new(syscall.SysProcAttr),
			syscall.SIGINT)
		if err := cmd.Run(); err != nil {
			if ee, ok := err.(*exec.ExitError); ok &&
				ee.ProcessState != nil && ee.ProcessState.Sys() != nil {
				if ws, ok := ee.ProcessState.Sys().(*syscall.WaitStatus); ok && ws.Signaled() {
					util.LogError(err, "finishr", "%s died with signal %v",
						real_path, ws.Signal())
				} else {
					util.LogError(err, "finishr", "%v failed", real_path)
				}
			} else {
				util.LogInfo("finishr", "Error running %v: %v",
					real_path, err)
			}
		}
	}
}
func NewPipestance(parent Nodable, call *syntax.CallGraphPipeline, srcPaths []string) (*Pipestance, error) {
	self := &Pipestance{}
	self.node = NewNode(parent, call)
	self.metadata = NewMetadata(self.node.parent.GetFQName(), self.GetPath())
	pipeline := call.Callable().(*syntax.Pipeline)
	preflightNodes := []Nodable{}
	for _, subcall := range call.Children {
		switch subcall := subcall.(type) {
		case *syntax.CallGraphStage:
			if s, err := NewStagestance(self.node, subcall, srcPaths); err != nil {
				return nil, err
			} else {
				self.node.subnodes[subcall.Call().Id] = s
				if subcall.Call().Modifiers.Preflight {
					preflightNodes = append(preflightNodes, s)
				}
			}
		case *syntax.CallGraphPipeline:
			if p, err := NewPipestance(self.node, subcall, srcPaths); err != nil {
				return nil, err
			} else {
				self.node.subnodes[subcall.Call().Id] = p
			}
		default:
			return nil, fmt.Errorf("Unsupported callable type %v", subcall)
		}
	}
	if r := pipeline.Ret; r != nil && r.Bindings != nil && len(r.Bindings.List) > 0 {
		self.node.makeReturnBindings(r.Bindings.List)
	}
	for _, preflightNode := range preflightNodes {
		for _, subnode := range self.node.subnodes {
			if !subnode.getNode().call.Call().Modifiers.Preflight {
				subnode.getNode().setPrenode(preflightNode)
			}
		}
	}
	if splits := call.Forks; len(splits) > 0 {
		exps := make([]*syntax.CallStm, len(splits))
		for i, s := range splits {
			exps[i] = s.Call()
		}
		self.node.forkRoots = exps
	}
	return self, self.buildForks()
}
func (self *Pipestance) buildForks() error {
	self.node.buildForks()
	if rs := self.node.call.Retained(); len(rs) > 0 {
		for _, r := range rs {
			node := self.node.top.allNodes[r.Id]
			if node == nil {
				return fmt.Errorf("Retaining unknown node %s", r.Id)
			}
			for _, fork := range node.forks {
				if fork.fileArgs == nil {
					fork.fileArgs = make(
						map[string]map[Nodable]struct{},
						len(rs))
				}
				if arg := fork.fileArgs[r.OutputId]; arg == nil {
					fork.fileArgs[r.OutputId] = map[Nodable]struct{}{
						nil: {},
					}
				} else {
					arg[nil] = struct{}{}
				}
			}
		}
	}
	return nil
}
func (self *Pipestance) getNode() *Node    { return self.node }
func (self *Pipestance) GetPname() string  { return self.node.call.Call().Id }
func (self *Pipestance) GetPsid() string   { return self.node.top.GetPsid() }
func (self *Pipestance) GetFQName() string { return self.node.GetFQName() }
func (self *Pipestance) RefreshState(ctx context.Context) {
	r := trace.StartRegion(ctx, "refresh")
	defer r.End()
	self.node.refreshState(self.readOnly())
}
func (self *Pipestance) readOnly() bool { return !self.metadata.exists(Lock) }
func (self *Pipestance) GetPrenodes() map[string]Nodable {
	return self.node.GetPrenodes()
}
func (self *Pipestance) GetPostNodes() map[string]Nodable {
	return self.node.GetPostNodes()
}
func (self *Pipestance) matchForks(id ForkId) []*Fork {
	return self.node.matchForks(id)
}
func (self *Pipestance) Callable() syntax.Callable {
	return self.node.Callable()
}
func (self *Pipestance) allNodes() []*Node {
	if self.allNodesCache == nil {
		self.allNodesCache = self.node.allNodes()
	}
	return self.allNodesCache
}
func (self *Pipestance) LoadMetadata(ctx context.Context) {
	r := trace.StartRegion(ctx, "LoadMetadata")
	defer r.End()
	for _, node := range self.allNodes() {
		node.loadMetadata()
	}
	for _, node := range self.allNodes() {
		node.state = node.getState()
		if node.state == Running && !self.readOnly() {
			if err := node.mkdirs(); err != nil {
				util.LogError(err, "runtime",
					"Error creating pipestance directories.")
			}
		}
	}
}
func (self *Pipestance) GetState(ctx context.Context) MetadataState {
	r := trace.StartRegion(ctx, "pipestance.GetState")
	defer r.End()
	nodes := self.node.getFrontierNodes()
	for _, node := range nodes {
		if node.state == Failed {
			return Failed
		}
	}
	for _, node := range nodes {
		if node.state == Running {
			return Running
		}
	}
	every := true
	for _, node := range self.allNodes() {
		if node.state != DisabledState {
			every = false
			break
		}
	}
	if every {
		return DisabledState
	}
	every = true
	for _, node := range self.allNodes() {
		if node.state != Complete && node.state != DisabledState {
			every = false
			break
		}
	}
	if every {
		return Complete
	}
	return ForkWaiting
}
func (self *Pipestance) Kill() {
	self.KillWithMessage("Job was killed by Martian.")
}
func (self *Pipestance) KillWithMessage(message string) {
	if self.readOnly() {
		return
	}
	nodes := self.node.getFrontierNodes()
	for _, node := range nodes {
		node.kill(message)
	}
}
func (self *Pipestance) RestoreForks(ctx context.Context) {
	defer trace.StartRegion(ctx, "restoreForks").End()
	for _, node := range self.allNodes() {
		node.expandForks(false)
		for _, fork := range node.forks {
			fork.metadatasCache = nil
		}
	}
}
func (self *Pipestance) RestartRunningNodes(jobMode string, outerCtx context.Context) error {
	ctx, task := trace.NewTask(outerCtx, "restartNodes")
	defer task.End()
	if self.readOnly() {
		return &RuntimeError{"Pipestance is in read only mode."}
	}
	self.LoadMetadata(ctx)
	nodes := self.node.getFrontierNodes()
	var errs syntax.ErrorList
	for _, node := range nodes {
		if node.state == Running {
			util.PrintInfo("runtime", "Found orphaned stage: %s", node.GetFQName())
			if jobMode == localMode || node.local {
				if err := node.reset(); err != nil {
					errs = append(errs, err)
				}
			}
		}
	}
	return errs.If()
}
func (self *Pipestance) RestartLocalJobs(jobMode string) error {
	if self.readOnly() {
		return &RuntimeError{"Pipestance is in read only mode."}
	}
	for _, node := range self.node.getFrontierNodes() {
		if node.state == Running {
			if err := node.restartLocallyQueuedJobs(); err != nil {
				return err
			}
		}
		if node.state == Running && (jobMode == localMode || node.local) {
			util.PrintInfo("runtime", "Found orphaned local stage: %s", node.GetFQName())
			if err := node.restartLocalJobs(); err != nil {
				return err
			}
		}
	}
	return nil
}
func (self *Pipestance) CheckHeartbeats(ctx context.Context) {
	r := trace.StartRegion(ctx, "heartbeat")
	defer r.End()
	if self.readOnly() {
		return
	}
	self.queryQueue(ctx)
	nodes := self.node.getFrontierNodes()
	for _, node := range nodes {
		node.checkHeartbeats()
	}
}