func (c *QueueController) Abort(run runner.RunID) (runner.RunStatus, error) {
	resultCh := make(chan result)
	c.reqCh <- abortReq{run, resultCh}
	result := <-resultCh
	return result.st, result.err
}
Abort kills the given run, returning its final status.
package runners
import (
	"fmt"
	"os"
	"sync"
	"time"
	log "github.com/sirupsen/logrus"
	"github.com/twitter/scoot/common/errors"
	"github.com/twitter/scoot/common/log/hooks"
	"github.com/twitter/scoot/common/log/tags"
	"github.com/twitter/scoot/common/stats"
	"github.com/twitter/scoot/runner"
	"github.com/twitter/scoot/runner/execer"
	"github.com/twitter/scoot/snapshot"
)
func init() {
	if loglevel := os.Getenv("SCOOT_LOGLEVEL"); loglevel != "" {
		level, err := log.ParseLevel(loglevel)
		if err != nil {
			log.Error(err)
			return
		}
		log.SetLevel(level)
		log.AddHook(hooks.NewContextHook())
	} else {
		log.SetLevel(log.ErrorLevel)
	}
}
const QueueFullMsg = "No resources available. Please try later."
const QueueInitingMsg = "Queue is still initializing. Please try later."
const QueueInvalidMsg = "Failed initialization, queue permanently broken."
type result struct {
	st  runner.RunStatus
	err error
}
type runReq struct {
	cmd      *runner.Command
	resultCh chan result
}
type abortReq struct {
	runID    runner.RunID
	resultCh chan result
}
type cmdAndID struct {
	cmd *runner.Command
	id  runner.RunID
}
func NewQueueRunner(
	exec execer.Execer,
	filerMap runner.RunTypeMap,
	output runner.OutputCreator,
	capacity int,
	stat stats.StatsReceiver,
	dirMonitor *stats.DirsMonitor,
	rID runner.RunnerID,
) runner.Service {
	if stat == nil {
		stat = stats.NilStatsReceiver()
	}
	history := 1
	if capacity > 0 {
		history = 0  
	} else if capacity == 0 {
		capacity = 1  
	}
	statusManager := NewStatusManager(history)
	inv := NewInvoker(exec, filerMap, output, stat, dirMonitor, rID)
	controller := &QueueController{
		statusManager: statusManager,
		inv:           inv,
		filerMap:      filerMap,
		updateReq:     make(map[runner.RunType]bool),
		capacity:      capacity,
		reqCh:         make(chan interface{}),
		updateCh:      make(chan interface{}),
		cancelTimerCh: make(chan interface{}),
	}
	run := &Service{controller, statusManager}
	var wg sync.WaitGroup
	var initErr error = nil
	wait := false
	for t, f := range filerMap {
		if f.IDC != nil {
			log.Infof("Starting goroutine to wait on init for filer %v", t)
			wg.Add(1)
			wait = true
			go func(rt runner.RunType, idc snapshot.InitDoneCh) {
				err := <-idc
				if err != nil {
					initErr = err
					log.Errorf("Init channel for filer %v returned error: %s", rt, err)
				}
				wg.Done()
			}(t, f.IDC)
		}
	}
	if wait {
		go func() {
			wg.Wait()
			if initErr != nil {
				stat.Counter(stats.WorkerDownloadInitFailure).Inc(1)
				statusManager.UpdateService(runner.ServiceStatus{Initialized: false, Error: initErr})
			} else {
				statusManager.UpdateService(runner.ServiceStatus{Initialized: true})
				controller.startUpdateTickers()
			}
		}()
	} else {
		statusManager.UpdateService(runner.ServiceStatus{Initialized: true})
		controller.startUpdateTickers()
	}
	go controller.loop()
	return run
}
func NewSingleRunner(
	exec execer.Execer,
	filerMap runner.RunTypeMap,
	output runner.OutputCreator,
	stat stats.StatsReceiver,
	dirMonitor *stats.DirsMonitor,
	rID runner.RunnerID,
) runner.Service {
	return NewQueueRunner(exec, filerMap, output, 0, stat, dirMonitor, rID)
}
type QueueController struct {
	inv           *Invoker
	filerMap      runner.RunTypeMap
	updateReq     map[runner.RunType]bool
	updateLock    sync.Mutex
	statusManager *StatusManager
	capacity      int
	queue        []cmdAndID
	runningID    runner.RunID
	runningCmd   *runner.Command
	runningAbort chan<- struct{}
	lastExitCode errors.ExitCode
	reqCh chan interface{}
	updateCh chan interface{}
	cancelTimerCh chan interface{}
}
func (c *QueueController) startUpdateTickers() {
	if c.updateCh == nil {
		return
	}
	for t, f := range c.filerMap {
		d := f.Filer.UpdateInterval()
		if d == snapshot.NoDuration {
			continue
		}
		go func(rt runner.RunType, td time.Duration) {
			ticker := time.NewTicker(td)
		Loop:
			for {
				select {
				case <-ticker.C:
					c.updateLock.Lock()
					c.updateReq[rt] = true
					c.updateLock.Unlock()
					c.updateCh <- nil
				case <-c.cancelTimerCh:
					ticker.Stop()
					break Loop
				}
			}
		}(t, d)
	}
}
func (c *QueueController) Run(cmd *runner.Command) (runner.RunStatus, error) {
	resultCh := make(chan result)
	c.reqCh <- runReq{cmd, resultCh}
	result := <-resultCh
	return result.st, result.err
}
func (c *QueueController) enqueue(cmd *runner.Command) (runner.RunStatus, error) {
	_, svcStatus, _ := c.statusManager.StatusAll()
	log.WithFields(
		log.Fields{
			"ready":          svcStatus.Initialized,
			"err":            svcStatus.Error,
			"availableSlots": c.capacity - len(c.queue),
			"totalSlots":     c.capacity,
			"currentRun":     c.runningID,
			"jobID":          cmd.JobID,
			"taskID":         cmd.TaskID,
			"tag":            cmd.Tag,
		}).Info("Trying to run")
	if !svcStatus.Initialized {
		errStr := QueueInitingMsg
		if svcStatus.Error != nil {
			errStr = svcStatus.Error.Error()
		}
		return runner.RunStatus{Error: errStr}, fmt.Errorf(QueueInitingMsg)
	}
	if len(c.queue) >= c.capacity {
		return runner.RunStatus{}, fmt.Errorf(QueueFullMsg)
	}
	st, err := c.statusManager.NewRun()
	if err != nil {
		return st, err
	}
	c.queue = append(c.queue, cmdAndID{cmd, st.RunID})
	return st, nil
}