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)
			}
		}
	}
}
Run a script whenever a pipestance finishes.
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
}