func (b *Batch) Run(ctx context.Context) error {
	done := make(chan *Run)
	var wg sync.WaitGroup
	for _, run := range b.Runs {
		if run.State.Phase == runner.Eval {
			wg.Add(1)
		}
	}
	b.Status.Printf("remaining: %d", len(b.Runs))
	for _, run := range b.Runs {
		go func(run *Run) {
			err := run.Go(ctx, &wg)
			switch {
			case ctx.Err() != nil:
			case err != nil:
				b.Log.Errorf("run %v: error: %v", run.ID, err)
			default:
				b.Log.Printf("run %v: done: %v", run.ID, run.State.Result)
			}
			done <- run
		}(run)
	}
	n := 0
	for n < len(b.Runs) {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-done:
			n++
			b.Status.Printf("remaining: %d", len(b.Runs)-n)
		}
	}
	return nil
}
Run runs the batch until completion, too many errors, or context
completion. Run reports batch progress to the batch's logger every
10 seconds.
package batch
import (
	"context"
	"encoding/csv"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	golog "log"
	"os"
	"path/filepath"
	"sync"
	"github.com/grailbio/base/digest"
	"github.com/grailbio/base/limiter"
	"github.com/grailbio/base/state"
	"github.com/grailbio/base/status"
	"github.com/grailbio/reflow"
	"github.com/grailbio/reflow/errors"
	"github.com/grailbio/reflow/flow"
	"github.com/grailbio/reflow/lang"
	"github.com/grailbio/reflow/log"
	"github.com/grailbio/reflow/pool"
	"github.com/grailbio/reflow/runner"
	"github.com/grailbio/reflow/syntax"
	"github.com/grailbio/reflow/taskdb"
	"github.com/grailbio/reflow/types"
	"github.com/grailbio/reflow/values"
)
const (
	statePrefix = "state"
)
type State int
const (
	StateInit State = iota
	StateRunning
	StateDone
	StateError
	stateMax
)
func (s State) Name() string {
	switch s {
	case StateInit:
		return "init"
	case StateRunning:
		return "running"
	case StateDone:
		return "done"
	case StateError:
		return "error"
	default:
		panic("bad state")
	}
}
type Run struct {
	ID string
	Program string
	Args map[string]string
	Argv []string
	RunID taskdb.RunID
	State runner.State `json:"-"`
	Status *status.Task
	batch *Batch
	log   *log.Logger
}
func (r *Run) Go(ctx context.Context, initWG *sync.WaitGroup) error {
	defer func() {
		r.Status.Done()
		r.batch.Limiter.Release(1)
	}()
	flow, typ, err := r.flow()
	if err != nil {
		return err
	}
	path := r.batch.path(fmt.Sprintf("log.%s", r.ID))
	batchLogFile, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600)
	if err != nil {
		return err
	}
	defer batchLogFile.Close()
	var w io.Writer = batchLogFile
	runLogPath := filepath.Join(r.batch.Rundir, digest.Digest(r.RunID).Hex()+".execlog")
	os.MkdirAll(filepath.Dir(runLogPath), 0777)
	runLogFile, err := os.Create(runLogPath)
	if err != nil {
		r.log.Errorf("create %s: %v", runLogPath, err)
	} else {
		defer runLogFile.Close()
		w = io.MultiWriter(w, runLogFile)
	}
	r.log = log.New(
		golog.New(w, "", golog.LstdFlags),
		log.DebugLevel,
	)
	evalConfig := r.batch.EvalConfig
	evalConfig.Log = r.log
	evalConfig.RunID = r.RunID
	run := &runner.Runner{
		State:      r.State,
		Flow:       flow,
		EvalConfig: evalConfig,
		Type:       typ,
		Labels:     pool.Labels{"program": r.Program},
	}
	run.Program = r.Program
	run.Params = r.Args
	run.Args = r.Argv
	switch run.Phase {
	default:
		initWG.Wait()
		if err := r.batch.Limiter.Acquire(ctx, 1); err != nil {
			return err
		}
	case runner.Eval:
		initWG.Done()
	}
	for ok := true; ok; {
		ok = run.Do(ctx)
		r.State = run.State
		r.Status.Print(run.State)
		r.log.Debugf("run %s: state: %v", r.ID, run.State)
		r.batch.commit(r)
	}
	if run.Err != nil {
		r.Status.Printf("error %v", run.Err)
		return run.Err
	}
	return nil
}
func (r *Run) flow() (*flow.Flow, *types.T, error) {
	switch ext := filepath.Ext(r.Program); ext {
	default:
		return nil, nil, fmt.Errorf("unrecognized file extension %s", ext)
	case ".reflow":
		f, err := os.Open(r.batch.path(r.Program))
		if err != nil {
			return nil, nil, err
		}
		file := f.Name()
		prog := &lang.Program{File: file, Errors: os.Stderr, Args: r.Argv}
		if err := prog.ParseAndTypecheck(f); err != nil {
			return nil, nil, err
		}
		flags := prog.Flags()
		err = parseFlags(flags, r.Args, r.batch.Args)
		if err != nil {
			return nil, nil, err
		}
		flags.VisitAll(func(f *flag.Flag) {
			if f.Value.String() == "" {
				err = fmt.Errorf("argument %q is undefined", f.Name)
			}
		})
		return prog.Eval(), nil, err
	case ".rf", ".rfx":
		sess := syntax.NewSession(nil)
		m, err := sess.Open(r.Program)
		if err != nil {
			return nil, nil, err
		}
		var maintyp *types.T
		for _, f := range m.Type(nil).Fields {
			if f.Name == "Main" {
				maintyp = f.T
				break
			}
		}
		if maintyp == nil {
			return nil, nil, fmt.Errorf("module %v does not define symbol Main", r.Program)
		}
		flags, err := m.Flags(sess, sess.Values)
		if err != nil {
			return nil, nil, err
		}
		err = parseFlags(flags, r.Args, r.batch.Args)
		if err != nil {
			return nil, nil, err
		}
		env := sess.Values.Push()
		if err := m.FlagEnv(flags, env, types.NewEnv()); err != nil {
			return nil, nil, err
		}
		v, err := m.Make(sess, env)
		if err != nil {
			return nil, nil, err
		}
		v = v.(values.Module)["Main"]
		v = syntax.Force(v, maintyp)
		switch v := v.(type) {
		case *flow.Flow:
			return v, maintyp, nil
		default:
			return &flow.Flow{Op: flow.Val, Value: v}, maintyp, nil
		}
	}
}
type config struct {
	Program  string `json:"program"`
	RunsFile string `json:"runs_file"`
}
type Batch struct {
	Dir string
	ConfigFilename string
	Rundir string
	User string
	Args []string
	Status *status.Group
	BatchState
	flow.EvalConfig
	Limiter *limiter.Limiter
	file   *state.File
	states map[string]*state.File
	config config
	flow   *flow.Flow
}
type BatchState struct {
	ID digest.Digest
	Runs map[string]*Run
}
func (b *Batch) Init(reset bool, retry bool) error {
	var (
		err error
		f   *os.File
	)
	if f, err = os.Open(filepath.Join(b.Dir, b.ConfigFilename)); err != nil {
		return err
	}
	defer f.Close()
	if err := json.NewDecoder(f).Decode(&b.config); err != nil {
		return err
	}
	dw := reflow.Digester.NewWriter()
	if _, err = dw.Write([]byte(b.config.Program)); err != nil {
		return err
	}
	if _, err = dw.Write([]byte(b.config.RunsFile)); err != nil {
		return err
	}
	var rf *os.File
	if rf, err = os.Open(b.path(b.config.RunsFile)); err != nil {
		return err
	}
	var rfb []byte
	if rfb, err = ioutil.ReadAll(rf); err != nil {
		rf.Close()
		return err
	}
	rf.Close()
	if _, err = dw.Write(rfb); err != nil {
		return err
	}
	batchID := dw.Digest()
	if b.file, err = state.Open(filepath.Join(b.Dir, statePrefix)); err != nil {
		return err
	}
	b.states = map[string]*state.File{}
	b.Runs = map[string]*Run{}
	if reset {
		b.ID = batchID
		if err := b.file.Marshal(b.BatchState); err != nil {
			b.Log.Errorf("marshal: %v", err)
		}
	} else {
		if err := b.file.Unmarshal(&b.BatchState); err != nil && err != state.ErrNoState {
			return err
		}
		if b.ID != batchID {
			b.ID = batchID
			if err := b.file.Marshal(b.BatchState); err != nil {
				b.Log.Errorf("marshal: %v", err)
			}
		}
	}
	b.commit(nil)
	b.Log.Printf("batch program %v runsfile %v", b.config.Program, b.config.RunsFile)
	return b.read(retry)
}
func (b *Batch) Close() {
	for _, file := range b.states {
		file.Close()
	}
	if b.file != nil {
		b.file.Close()
	}
}