blob: a339d2312e12e7fe5960f1fef8d721a65c8dd954 [file] [log] [blame]
// Copyright 2015 Google Inc. All rights reserved
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package kati
import (
"container/heap"
"errors"
"fmt"
"os"
"os/exec"
"syscall"
"time"
"github.com/golang/glog"
)
var (
errNothingDone = errors.New("nothing done")
)
type job struct {
n *DepNode
ex *Executor
parents []*job
outputTs int64
numDeps int
depsTs int64
id int
runners []runner
}
type jobResult struct {
j *job
w *worker
err error
}
type newDep struct {
j *job
neededBy *job
}
type worker struct {
wm *workerManager
jobChan chan *job
waitChan chan bool
doneChan chan bool
}
type jobQueue []*job
func (jq jobQueue) Len() int { return len(jq) }
func (jq jobQueue) Swap(i, j int) { jq[i], jq[j] = jq[j], jq[i] }
func (jq jobQueue) Less(i, j int) bool {
// First come, first serve, for GNU make compatibility.
return jq[i].id < jq[j].id
}
func (jq *jobQueue) Push(x interface{}) {
item := x.(*job)
*jq = append(*jq, item)
}
func (jq *jobQueue) Pop() interface{} {
old := *jq
n := len(old)
item := old[n-1]
*jq = old[0 : n-1]
return item
}
func newWorker(wm *workerManager) *worker {
w := &worker{
wm: wm,
jobChan: make(chan *job),
waitChan: make(chan bool),
doneChan: make(chan bool),
}
return w
}
func (w *worker) Run() {
done := false
for !done {
select {
case j := <-w.jobChan:
err := j.build()
w.wm.ReportResult(w, j, err)
case done = <-w.waitChan:
}
}
w.doneChan <- true
}
func (w *worker) PostJob(j *job) {
w.jobChan <- j
}
func (w *worker) Wait() {
w.waitChan <- true
<-w.doneChan
}
func (j *job) createRunners() ([]runner, error) {
runners, _, err := createRunners(j.ex.ctx, j.n)
return runners, err
}
// TODO(ukai): use time.Time?
func getTimestamp(filename string) int64 {
st, err := os.Stat(filename)
if err != nil {
return -2
}
return st.ModTime().Unix()
}
func (j *job) build() error {
if j.n.IsPhony {
j.outputTs = -2 // trigger cmd even if all inputs don't exist.
} else {
j.outputTs = getTimestamp(j.n.Output)
}
if !j.n.HasRule {
if j.outputTs >= 0 || j.n.IsPhony {
return errNothingDone
}
if len(j.parents) == 0 {
return fmt.Errorf("*** No rule to make target %q.", j.n.Output)
}
return fmt.Errorf("*** No rule to make target %q, needed by %q.", j.n.Output, j.parents[0].n.Output)
}
if j.outputTs >= j.depsTs {
// TODO: stats.
return errNothingDone
}
rr, err := j.createRunners()
if err != nil {
return err
}
if len(rr) == 0 {
return errNothingDone
}
for _, r := range rr {
err := r.run(j.n.Output)
glog.Warningf("cmd result for %q: %v", j.n.Output, err)
if err != nil {
exit := exitStatus(err)
return fmt.Errorf("*** [%s] Error %d", j.n.Output, exit)
}
}
if j.n.IsPhony {
j.outputTs = time.Now().Unix()
} else {
j.outputTs = getTimestamp(j.n.Output)
if j.outputTs < 0 {
j.outputTs = time.Now().Unix()
}
}
return nil
}
func (wm *workerManager) handleJobs() error {
for {
if len(wm.freeWorkers) == 0 {
return nil
}
if wm.readyQueue.Len() == 0 {
return nil
}
j := heap.Pop(&wm.readyQueue).(*job)
glog.V(1).Infof("run: %s", j.n.Output)
j.numDeps = -1 // Do not let other workers pick this.
w := wm.freeWorkers[0]
wm.freeWorkers = wm.freeWorkers[1:]
wm.busyWorkers[w] = true
w.jobChan <- j
}
}
func (wm *workerManager) updateParents(j *job) {
for _, p := range j.parents {
p.numDeps--
glog.V(1).Infof("child: %s (%d)", p.n.Output, p.numDeps)
if p.depsTs < j.outputTs {
p.depsTs = j.outputTs
}
wm.maybePushToReadyQueue(p)
}
}
type workerManager struct {
maxJobs int
jobs []*job
readyQueue jobQueue
jobChan chan *job
resultChan chan jobResult
newDepChan chan newDep
stopChan chan bool
waitChan chan bool
doneChan chan error
freeWorkers []*worker
busyWorkers map[*worker]bool
ex *Executor
runnings map[string]*job
finishCnt int
skipCnt int
}
func newWorkerManager(numJobs int) (*workerManager, error) {
wm := &workerManager{
maxJobs: numJobs,
jobChan: make(chan *job),
resultChan: make(chan jobResult),
newDepChan: make(chan newDep),
stopChan: make(chan bool),
waitChan: make(chan bool),
doneChan: make(chan error),
busyWorkers: make(map[*worker]bool),
}
wm.busyWorkers = make(map[*worker]bool)
for i := 0; i < numJobs; i++ {
w := newWorker(wm)
wm.freeWorkers = append(wm.freeWorkers, w)
go w.Run()
}
heap.Init(&wm.readyQueue)
go wm.Run()
return wm, nil
}
func exitStatus(err error) int {
if err == nil {
return 0
}
exit := 1
if err, ok := err.(*exec.ExitError); ok {
if w, ok := err.ProcessState.Sys().(syscall.WaitStatus); ok {
return w.ExitStatus()
}
}
return exit
}
func (wm *workerManager) hasTodo() bool {
return wm.finishCnt != len(wm.jobs)
}
func (wm *workerManager) maybePushToReadyQueue(j *job) {
if j.numDeps != 0 {
return
}
heap.Push(&wm.readyQueue, j)
glog.V(1).Infof("ready: %s", j.n.Output)
}
func (wm *workerManager) handleNewDep(j *job, neededBy *job) {
if j.numDeps < 0 {
neededBy.numDeps--
if neededBy.id > 0 {
panic("FIXME: already in WM... can this happen?")
}
} else {
j.parents = append(j.parents, neededBy)
}
}
func (wm *workerManager) Run() {
done := false
var err error
Loop:
for wm.hasTodo() || len(wm.busyWorkers) > 0 || len(wm.runnings) > 0 || !done {
select {
case j := <-wm.jobChan:
glog.V(1).Infof("wait: %s (%d)", j.n.Output, j.numDeps)
j.id = len(wm.jobs) + 1
wm.jobs = append(wm.jobs, j)
wm.maybePushToReadyQueue(j)
case jr := <-wm.resultChan:
glog.V(1).Infof("done: %s", jr.j.n.Output)
delete(wm.busyWorkers, jr.w)
wm.freeWorkers = append(wm.freeWorkers, jr.w)
wm.updateParents(jr.j)
wm.finishCnt++
if jr.err == errNothingDone {
wm.skipCnt++
jr.err = nil
}
if jr.err != nil {
err = jr.err
close(wm.stopChan)
break Loop
}
case af := <-wm.newDepChan:
wm.handleNewDep(af.j, af.neededBy)
glog.V(1).Infof("dep: %s (%d) %s", af.neededBy.n.Output, af.neededBy.numDeps, af.j.n.Output)
case done = <-wm.waitChan:
}
err = wm.handleJobs()
if err != nil {
break Loop
}
glog.V(1).Infof("job=%d ready=%d free=%d busy=%d", len(wm.jobs)-wm.finishCnt, wm.readyQueue.Len(), len(wm.freeWorkers), len(wm.busyWorkers))
}
if !done {
<-wm.waitChan
}
for _, w := range wm.freeWorkers {
w.Wait()
}
for w := range wm.busyWorkers {
w.Wait()
}
wm.doneChan <- err
}
func (wm *workerManager) PostJob(j *job) error {
select {
case wm.jobChan <- j:
return nil
case <-wm.stopChan:
return errors.New("worker manager stopped")
}
}
func (wm *workerManager) ReportResult(w *worker, j *job, err error) {
select {
case wm.resultChan <- jobResult{w: w, j: j, err: err}:
case <-wm.stopChan:
}
}
func (wm *workerManager) ReportNewDep(j *job, neededBy *job) {
select {
case wm.newDepChan <- newDep{j: j, neededBy: neededBy}:
case <-wm.stopChan:
}
}
func (wm *workerManager) Wait() (int, error) {
wm.waitChan <- true
err := <-wm.doneChan
glog.V(2).Infof("finish %d skip %d", wm.finishCnt, wm.skipCnt)
return wm.finishCnt - wm.skipCnt, err
}