blob: 8948ac676d9c6d9953fc621077b61665a9c3402c [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"
"fmt"
"os"
"os/exec"
"strings"
"syscall"
"time"
)
type job struct {
n *DepNode
ex *Executor
parents []*job
outputTs int64
numDeps int
depsTs int64
id int
runners []runner
}
type runner struct {
output string
cmd string
echo bool
ignoreError bool
shell string
}
type jobResult struct {
j *job
w *worker
}
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:
j.build()
w.wm.ReportResult(w, j)
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 evalCmd(ev *Evaluator, r runner, s string) []runner {
r = newRunner(r, s)
if strings.IndexByte(r.cmd, '$') < 0 {
// fast path
return []runner{r}
}
// TODO(ukai): parse once more earlier?
expr, _, err := parseExpr([]byte(r.cmd), nil, false)
if err != nil {
panic(fmt.Errorf("parse cmd %q: %v", r.cmd, err))
}
buf := newBuf()
expr.Eval(buf, ev)
cmds := buf.String()
freeBuf(buf)
var runners []runner
for _, cmd := range strings.Split(cmds, "\n") {
if len(runners) > 0 && strings.HasSuffix(runners[len(runners)-1].cmd, "\\") {
runners[len(runners)-1].cmd += "\n"
runners[len(runners)-1].cmd += cmd
} else {
runners = append(runners, newRunner(r, cmd))
}
}
return runners
}
func newRunner(r runner, s string) runner {
for {
s = trimLeftSpace(s)
if s == "" {
return runner{}
}
switch s[0] {
case '@':
if !DryRunFlag {
r.echo = false
}
s = s[1:]
continue
case '-':
r.ignoreError = true
s = s[1:]
continue
}
break
}
r.cmd = s
return r
}
func (r runner) run(output string) error {
if r.echo || DryRunFlag {
fmt.Printf("%s\n", r.cmd)
}
if DryRunFlag {
return nil
}
args := []string{r.shell, "-c", r.cmd}
cmd := exec.Cmd{
Path: args[0],
Args: args,
}
out, err := cmd.CombinedOutput()
fmt.Printf("%s", out)
exit := exitStatus(err)
if r.ignoreError && exit != 0 {
fmt.Printf("[%s] Error %d (ignored)\n", output, exit)
err = nil
}
return err
}
func (j *job) createRunners() []runner {
runners, _ := j.ex.createRunners(j.n, false)
return runners
}
// 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() {
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
}
if len(j.parents) == 0 {
errorNoLocationExit("*** No rule to make target %q.", j.n.Output)
} else {
errorNoLocationExit("*** No rule to make target %q, needed by %q.", j.n.Output, j.parents[0].n.Output)
}
errorNoLocationExit("no rule to make target %q", j.n.Output)
}
if j.outputTs >= j.depsTs {
// TODO: stats.
return
}
for _, r := range j.createRunners() {
err := r.run(j.n.Output)
if err != nil {
exit := exitStatus(err)
errorNoLocationExit("[%s] Error %d: %v", j.n.Output, exit, err)
}
}
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()
}
}
}
func (wm *workerManager) handleJobs() {
for {
if wm.para == nil && len(wm.freeWorkers) == 0 {
return
}
if wm.readyQueue.Len() == 0 {
return
}
j := heap.Pop(&wm.readyQueue).(*job)
logf("run: %s", j.n.Output)
if wm.para != nil {
j.runners = j.createRunners()
if len(j.runners) == 0 {
wm.updateParents(j)
wm.finishCnt++
} else {
wm.runnings[j.n.Output] = j
wm.para.RunCommand(j.runners)
}
} else {
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--
logf("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
waitChan chan bool
doneChan chan bool
freeWorkers []*worker
busyWorkers map[*worker]bool
ex *Executor
para *paraWorker
paraChan chan *paraResult
runnings map[string]*job
finishCnt int
}
func newWorkerManager(numJobs int, paraPath string) *workerManager {
wm := &workerManager{
maxJobs: numJobs,
jobChan: make(chan *job),
resultChan: make(chan jobResult),
newDepChan: make(chan newDep),
waitChan: make(chan bool),
doneChan: make(chan bool),
busyWorkers: make(map[*worker]bool),
}
if paraPath != "" {
wm.runnings = make(map[string]*job)
wm.paraChan = make(chan *paraResult)
wm.para = newParaWorker(wm.paraChan, numJobs, paraPath)
go wm.para.Run()
} else {
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
}
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)
logf("ready: %s", j.n.Output)
}
func (wm *workerManager) handleNewDep(j *job, neededBy *job) {
if j.numDeps < 0 {
neededBy.numDeps--
if neededBy.id > 0 {
panic("already in WM... can this happen?")
}
} else {
j.parents = append(j.parents, neededBy)
}
}
func (wm *workerManager) Run() {
done := false
for wm.hasTodo() || len(wm.busyWorkers) > 0 || len(wm.runnings) > 0 || !done {
select {
case j := <-wm.jobChan:
logf("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:
logf("done: %s", jr.j.n.Output)
delete(wm.busyWorkers, jr.w)
wm.freeWorkers = append(wm.freeWorkers, jr.w)
wm.updateParents(jr.j)
wm.finishCnt++
case af := <-wm.newDepChan:
wm.handleNewDep(af.j, af.neededBy)
logf("dep: %s (%d) %s", af.neededBy.n.Output, af.neededBy.numDeps, af.j.n.Output)
case pr := <-wm.paraChan:
if pr.status < 0 && pr.signal < 0 {
j := wm.runnings[pr.output]
for _, r := range j.runners {
if r.echo || DryRunFlag {
fmt.Printf("%s\n", r.cmd)
}
}
} else {
fmt.Fprint(os.Stdout, pr.stdout)
fmt.Fprint(os.Stderr, pr.stderr)
j := wm.runnings[pr.output]
wm.updateParents(j)
delete(wm.runnings, pr.output)
wm.finishCnt++
}
case done = <-wm.waitChan:
}
wm.handleJobs()
if wm.para != nil {
numBusy := len(wm.runnings)
if numBusy > wm.maxJobs {
numBusy = wm.maxJobs
}
logf("job=%d ready=%d free=%d busy=%d", len(wm.jobs)-wm.finishCnt, wm.readyQueue.Len(), wm.maxJobs-numBusy, numBusy)
} else {
logf("job=%d ready=%d free=%d busy=%d", len(wm.jobs)-wm.finishCnt, wm.readyQueue.Len(), len(wm.freeWorkers), len(wm.busyWorkers))
}
}
if wm.para != nil {
logf("Wait for para to finish")
wm.para.Wait()
} else {
for _, w := range wm.freeWorkers {
w.Wait()
}
for w := range wm.busyWorkers {
w.Wait()
}
}
wm.doneChan <- true
}
func (wm *workerManager) PostJob(j *job) {
wm.jobChan <- j
}
func (wm *workerManager) ReportResult(w *worker, j *job) {
wm.resultChan <- jobResult{w: w, j: j}
}
func (wm *workerManager) ReportNewDep(j *job, neededBy *job) {
wm.newDepChan <- newDep{j: j, neededBy: neededBy}
}
func (wm *workerManager) Wait() {
wm.waitChan <- true
<-wm.doneChan
}