Squashed 'tools/' changes from 7a66090..c407250

c407250 Thingdoer
e9749a5 Make scheduler aware of test parallelisation

git-subtree-dir: tools
git-subtree-split: c40725021f65889dbaad38bd91a581452a6a2b6f

.gitignore

@@ -2,3 +2,6 @@ cover/cover
 socks/proxy
 socks/image.tar
 runner/runner
+*.pyc
+*~
+thingdoer/thingdoer

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 <cmd> <args..>" % sys.argv[0]
+  print "%s (--target=...) <cmd> <args..>" % sys.argv[0]
   print "   time <test name> <run time>"
   print "   sched <test run> <num shards> <shard id>"
 
 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()
 

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)

thingdoer/example.yaml

@@ -0,0 +1,5 @@
+- name: one
+  command: echo one
+- name: two
+  after: [one]
+  command: echo two

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(&parallelism, "p", 2, "level of parallelism")
+	flag.Parse()
+
+	args := flag.Args()
+	if len(args) <= 0 {
+		log.Fatalf("Usage: thingdoer <file>")
+	}
+
+	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()
+}