From f45e265f712dc5742368ab7f8bdd0023daa1fbf1 Mon Sep 17 00:00:00 2001 From: Tom Wilkie Date: Thu, 19 May 2016 10:20:28 +0100 Subject: [PATCH] Squashed 'tools/' changes from 7a66090..c407250 c407250 Thingdoer e9749a5 Make scheduler aware of test parallelisation git-subtree-dir: tools git-subtree-split: c40725021f65889dbaad38bd91a581452a6a2b6f --- .gitignore | 3 + sched | 26 ++++---- scheduler/main.py | 15 ++++- thingdoer/example.yaml | 5 ++ thingdoer/main.go | 141 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 177 insertions(+), 13 deletions(-) create mode 100644 thingdoer/example.yaml create mode 100644 thingdoer/main.go diff --git a/.gitignore b/.gitignore index f5c83426c..b0f60f376 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ cover/cover socks/proxy socks/image.tar runner/runner +*.pyc +*~ +thingdoer/thingdoer diff --git a/sched b/sched index e94e8af8f..cf47773e5 100755 --- a/sched +++ b/sched @@ -1,36 +1,38 @@ #!/usr/bin/python import sys, string, json, urllib import requests +import optparse -BASE_URL="http://positive-cocoa-90213.appspot.com" - -def test_time(test_name, runtime): - r = requests.post(BASE_URL + "/record/%s/%f" % (urllib.quote(test_name, safe=""), runtime)) +def test_time(target, test_name, runtime): + r = requests.post(target + "/record/%s/%f" % (urllib.quote(test_name, safe=""), runtime)) print r.text assert r.status_code == 204 -def test_sched(test_run, shard_count, shard_id): +def test_sched(target, test_run, shard_count, shard_id): tests = json.dumps({'tests': string.split(sys.stdin.read())}) - r = requests.post(BASE_URL + "/schedule/%s/%d/%d" % (test_run, shard_count, shard_id), data=tests) + r = requests.post(target + "/schedule/%s/%d/%d" % (test_run, shard_count, shard_id), data=tests) assert r.status_code == 200 result = r.json() for test in sorted(result['tests']): print test def usage(): - print "%s " % sys.argv[0] + print "%s (--target=...) " % sys.argv[0] print " time " print " sched " def main(): - if len(sys.argv) < 4: + parser = optparse.OptionParser() + parser.add_option('--target', default="http://positive-cocoa-90213.appspot.com") + options, args = parser.parse_args() + if len(args) < 3: usage() sys.exit(1) - if sys.argv[1] == "time": - test_time(sys.argv[2], float(sys.argv[3])) - elif sys.argv[1] == "sched": - test_sched(sys.argv[2], int(sys.argv[3]), int(sys.argv[4])) + if args[0] == "time": + test_time(options.target, args[1], float(args[2])) + elif args[0] == "sched": + test_sched(options.target, args[1], int(args[2]), int(args[3])) else: usage() diff --git a/scheduler/main.py b/scheduler/main.py index f509f0e1f..8907e202d 100644 --- a/scheduler/main.py +++ b/scheduler/main.py @@ -23,6 +23,19 @@ class Test(ndb.Model): total_run_time = ndb.FloatProperty(default=0.) # Not total, but a EWMA total_runs = ndb.IntegerProperty(default=0) + def parallelism(self): + name = self.key.string_id() + m = re.search('(\d+)_test.sh$', name) + if m is None: + return 1 + else: + return int(m.group(1)) + + def cost(self): + p = self.parallelism() + logging.info("Test %s has parallelism %d and avg run time %s", self.key.string_id(), p, self.total_run_time) + return self.parallelism() * self.total_run_time + class Schedule(ndb.Model): shards = ndb.JsonProperty() @@ -52,7 +65,7 @@ def schedule(test_run, shard_count, shard): test_times = ndb.get_multi(ndb.Key(Test, test_name) for test_name in test_names) def avg(test): if test is not None: - return test.total_run_time + return test.cost() return 1 test_times = [(test_name, avg(test)) for test_name, test in zip(test_names, test_times)] test_times_dict = dict(test_times) diff --git a/thingdoer/example.yaml b/thingdoer/example.yaml new file mode 100644 index 000000000..a5b7cf059 --- /dev/null +++ b/thingdoer/example.yaml @@ -0,0 +1,5 @@ +- name: one + command: echo one +- name: two + after: [one] + command: echo two diff --git a/thingdoer/main.go b/thingdoer/main.go new file mode 100644 index 000000000..c2bceead7 --- /dev/null +++ b/thingdoer/main.go @@ -0,0 +1,141 @@ +package main + +import ( + "bytes" + "flag" + "io/ioutil" + "log" + "os" + "os/exec" + "sync" + "time" + + "gopkg.in/yaml.v2" +) + +type thing struct { + Name string `yaml:"name"` + After []string `yaml:"after"` + Command string + success bool +} + +func (t *thing) run() { + var out bytes.Buffer + + cmd := exec.Command("/bin/sh", "-c", t.Command) + cmd.Env = os.Environ() + cmd.Stdout = &out + cmd.Stderr = &out + + start := time.Now() + err := cmd.Run() + duration := float64(time.Now().Sub(start)) / float64(time.Second) + + if err != nil { + log.Printf(">>> %s finished after %0.1f secs with error: %v\n", t.Name, duration, err) + } else { + log.Printf(">>> %s finished with success after %0.1f secs\n", t.Name, duration) + } + if err != nil { + log.Print(out.String()) + log.Println() + } + + t.success = err == nil +} + +func remove(s string, ts []string) []string { + for i, t := range ts { + if t == s { + return ts[:i+copy(ts[i:], ts[i+1:])] + } + } + return ts +} + +func main() { + var parallelism int + flag.IntVar(¶llelism, "p", 2, "level of parallelism") + flag.Parse() + + args := flag.Args() + if len(args) <= 0 { + log.Fatalf("Usage: thingdoer ") + } + + contents, err := ioutil.ReadFile(args[0]) + if err != nil { + log.Fatalf("Error reading file: %v", err) + } + + things := []*thing{} + if err := yaml.Unmarshal(contents, &things); err != nil { + log.Fatalf("Error parsing file: %v", err) + } + + log.Printf("things: %#v", things) + + next := make(chan *thing, parallelism) + done := make(chan *thing) + wg := sync.WaitGroup{} + wg.Add(parallelism) + + for i := 0; i < parallelism; i++ { + go func() { + defer wg.Done() + + for thing := range next { + thing.run() + done <- thing + } + }() + } + + sched := func() { + var thing *thing + for i, t := range things { + if len(t.After) == 0 { + thing = t + things = things[:i+copy(things[i:], things[i+1:])] + break + } + } + if thing != nil { + log.Printf("Doing %s", thing.Name) + next <- thing + } else { + log.Printf("Nothing to do, waiting for something to finish") + } + } + + // schedule p things if possible + for i := 0; i < parallelism; i++ { + sched() + } + + for { + // next, wait for something to finish + thing := <-done + if !thing.success { + log.Printf("%s failed, stopping", thing.Name) + break + } + + // remove it as a dependancy from + // other things + for _, t := range things { + t.After = remove(thing.Name, t.After) + } + + // (potentially) schedule something else to run + if len(things) > 0 { + sched() + } else { + break + } + } + + close(next) + wg.Wait() +}