go/src/v.io/x/playground/compilerd/jobqueue/jobqueue_test.go - release.projects.playground - Git at Google

 // Copyright 2015 The Vanadium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 package jobqueue

 import (
 	"bytes"
 	"fmt"
 	"os"
 	"os/exec"
 	"path"
 	"strconv"
 	"strings"
 	"sync"
 	"testing"
 	"time"

 	"v.io/x/playground/lib/event"
 )

 var (
 	defaultMaxTime  = 10 * time.Second
 	defaultMaxSize  = 1 << 16
 	defaultMemLimit = 100
 )

 func init() {
 	// Compile builder binary and put in path.
 	pgDir := os.ExpandEnv("${V23_ROOT}/release/projects/playground/go")

 	cmd := exec.Command("make", "builder")
 	cmd.Dir = path.Join(pgDir, "src", "v.io", "x", "playground")
 	if out, err := cmd.CombinedOutput(); err != nil {
 		fmt.Println("Error running 'make builder'")
 		fmt.Println(string(out))
 		panic(err)
 	}

 	pgBinDir := path.Join(pgDir, "bin")
 	if err := os.Setenv("PATH", pgBinDir+":"+os.Getenv("PATH")); err != nil {
 		panic(err)
 	}
 }

 // mockTestFile is a simple Go progam that will be sent to the dispatcher in the
 // test runs.
 var mockTestFile = `
 package main

 import(
 	"fmt"
 	"time"
 )

 func main() {
 	fmt.Printf("PROGRAM START")
 	time.Sleep(1 * time.Second)
 	fmt.Printf("PROGRAM END")
 }
 `

 // mockTestBody is a bundle containing the mockTestFile, as it would look if it
 // came from an http request from the playground client.
 var mockTestBody = []byte(fmt.Sprintf(`{
 	"files": [ {
 		"name": "src/main/main.go",
 		"body": %v
 	} ]
 }`, strconv.Quote(mockTestFile)))

 // testConfig encapsulates all the parameters and expectations for a single
 // test run.
 type testConfig struct {
 	// Number of jobs to enqueue.
 	jobs int

 	// Dispatcher configuration.
 	workers int
 	jobCap  int

 	// Job configuration.
 	useDocker bool
 	maxSize   int
 	maxTime   time.Duration
 	memLimit  int

 	// Test expectations. Default is to expect success.
 	expectEnqueueFail    bool
 	expectOutputTooLarge bool
 	expectJobFail        bool
 }

 func newMockResponseEventSink() *event.ResponseEventSink {
 	var b bytes.Buffer
 	return event.NewResponseEventSink(&b, false)
 }

 // eventsMatch returns true iff there is an event whose message matches the
 // given string.
 func eventsMatch(events []event.Event, match string) bool {
 	for _, e := range events {
 		if strings.Contains(e.Message, match) {
 			return true
 		}
 	}
 	return false
 }

 // assertExpectedResult asserts that the result matches the test expectations
 // in the test config.
 func assertExpectedResult(t *testing.T, c testConfig, r Result) {
 	expectSuccess := !c.expectJobFail
 	if expectSuccess != r.Success {
 		t.Errorf("Expected result.Success to be %v but was %v. Test config: %#v", expectSuccess, r.Success, c)
 	}

 	if !r.Success {
 		return
 	}

 	if c.expectOutputTooLarge {
 		want := "Program output too large, killed."
 		if !eventsMatch(r.Events, want) {
 			t.Errorf("Event message %v not found in %#v. Test config: %#v", want, r.Events, c)
 		}
 		return
 	}

 	// Expect normal program output.
 	want := "PROGRAM START"
 	if !eventsMatch(r.Events, want) {
 		t.Errorf("Event message %v not found in %#v. Test config: %#v", want, r.Events, c)
 	}

 	want = "PROGRAM END"
 	if !eventsMatch(r.Events, want) {
 		t.Errorf("Event message %v not found in %#v. Test config: %#v", want, r.Events, c)
 	}
 }

 // runTest runs a particular test configuration. It starts a dispatcher, queues
 // jobs and waits for them to finish, and asserts that they match the
 // expectations in the testConfig.
 func runTest(t *testing.T, c testConfig) {
 	fmt.Printf("Testing %v jobs on %v workers with jobCap of %v and useDocker %v\n", c.jobs, c.workers, c.jobCap, c.useDocker)
 	d := NewDispatcher(c.workers, c.jobCap)

 	var enqueueError error

 	// wg waits for all the jobs to finish.
 	var wg sync.WaitGroup

 	// Start all the jobs.
 	for i := 0; i < c.jobs; i++ {
 		res := newMockResponseEventSink()
 		job := NewJob(mockTestBody, res, c.maxSize, c.maxTime, c.useDocker, c.memLimit)

 		resultChan, err := d.Enqueue(job)
 		if err != nil {
 			enqueueError = err
 			break
 		}

 		wg.Add(1)

 		go func() {
 			r := <-resultChan
 			assertExpectedResult(t, c, r)
 			wg.Done()
 		}()
 	}

 	if !c.expectEnqueueFail && enqueueError != nil {
 		t.Fatalf("Enqueue failed: %v. Test config: %#v", enqueueError, c)
 	}

 	if c.expectEnqueueFail && enqueueError == nil {
 		t.Fatalf("Expected Enqueue to fail but it did not. Test config: %#v", c)
 	}

 	// Test must finish in 30 seconds.
 	timeout := time.After(30 * time.Second)

 	jobsFinished := make(chan bool)

 	go func() {
 		wg.Wait()
 		jobsFinished <- true
 	}()

 	// Wait for timeout or all jobs to finish.
 	select {
 	case <-timeout:
 		t.Fatalf("Expected jobs to complete but got timeout. Test config: %#v", c)
 	case <-jobsFinished:
 	}

 	d.Stop()
 }

 func TestJobQueue(t *testing.T) {

 	// Test success cases without docker.
 	runTest(t, testConfig{
 		jobs:      1,
 		workers:   1,
 		jobCap:    1,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		useDocker: false,
 	})

 	runTest(t, testConfig{
 		jobs:      3,
 		workers:   1,
 		jobCap:    3,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		useDocker: false,
 	})

 	runTest(t, testConfig{
 		jobs:      3,
 		workers:   3,
 		jobCap:    3,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		useDocker: false,
 	})

 	runTest(t, testConfig{
 		jobs:      6,
 		workers:   3,
 		jobCap:    10,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		useDocker: false,
 	})

 	// Test success cases with docker.
 	runTest(t, testConfig{
 		jobs:      1,
 		workers:   1,
 		jobCap:    1,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		memLimit:  defaultMemLimit,
 		useDocker: true,
 	})

 	runTest(t, testConfig{
 		jobs:      3,
 		workers:   1,
 		jobCap:    3,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		memLimit:  defaultMemLimit,
 		useDocker: true,
 	})

 	runTest(t, testConfig{
 		jobs:      3,
 		workers:   3,
 		jobCap:    3,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		memLimit:  defaultMemLimit,
 		useDocker: true,
 	})

 	runTest(t, testConfig{
 		jobs:      6,
 		workers:   3,
 		jobCap:    10,
 		maxSize:   defaultMaxSize,
 		maxTime:   defaultMaxTime,
 		memLimit:  defaultMemLimit,
 		useDocker: true,
 	})

 	// Test Enqueue should fail when job capacity is exceeded.
 	runTest(t, testConfig{
 		jobs:              5,
 		workers:           2,
 		jobCap:            2,
 		maxSize:           defaultMaxSize,
 		maxTime:           defaultMaxTime,
 		useDocker:         false,
 		expectEnqueueFail: true,
 	})

 	runTest(t, testConfig{
 		jobs:              5,
 		workers:           2,
 		jobCap:            2,
 		maxSize:           defaultMaxSize,
 		maxTime:           defaultMaxTime,
 		memLimit:          defaultMemLimit,
 		useDocker:         true,
 		expectEnqueueFail: true,
 	})

 	// Test job should fail if it exceeds size limit.
 	runTest(t, testConfig{
 		jobs:                 1,
 		workers:              1,
 		jobCap:               1,
 		maxSize:              100,
 		maxTime:              defaultMaxTime,
 		useDocker:            false,
 		expectOutputTooLarge: true,
 	})

 	runTest(t, testConfig{
 		jobs:                 1,
 		workers:              1,
 		jobCap:               1,
 		maxSize:              100,
 		maxTime:              defaultMaxTime,
 		memLimit:             defaultMemLimit,
 		useDocker:            true,
 		expectOutputTooLarge: true,
 	})

 	// Test job should fail if it exceeds max time.
 	runTest(t, testConfig{
 		jobs:          1,
 		workers:       1,
 		jobCap:        1,
 		maxSize:       defaultMaxSize,
 		maxTime:       1 * time.Second,
 		useDocker:     false,
 		expectJobFail: true,
 	})

 	runTest(t, testConfig{
 		jobs:          1,
 		workers:       1,
 		jobCap:        1,
 		maxSize:       defaultMaxSize,
 		maxTime:       1 * time.Second,
 		useDocker:     true,
 		expectJobFail: true,
 	})

 	// Test job should fail if it exceeds memory limit (docker only).
 	runTest(t, testConfig{
 		jobs:          1,
 		workers:       1,
 		jobCap:        1,
 		maxSize:       defaultMaxSize,
 		maxTime:       defaultMaxTime,
 		memLimit:      5,
 		useDocker:     true,
 		expectJobFail: true,
 	})
 }

 func TestJobCancel(t *testing.T) {
 	d := NewDispatcher(1, 10)

 	// Create five jobs.
 	res1 := newMockResponseEventSink()
 	job1 := NewJob(mockTestBody, res1, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

 	res2 := newMockResponseEventSink()
 	job2 := NewJob(mockTestBody, res2, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

 	res3 := newMockResponseEventSink()
 	job3 := NewJob(mockTestBody, res3, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

 	res4 := newMockResponseEventSink()
 	job4 := NewJob(mockTestBody, res4, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

 	res5 := newMockResponseEventSink()
 	job5 := NewJob(mockTestBody, res5, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

 	// Cancel first job right away.
 	job1.Cancel()

 	// Queue all jobs.
 	resultChan1, err := d.Enqueue(job1)
 	if err != nil {
 		t.Fatalf("Enqueue failed: %v", err)
 	}
 	resultChan2, err := d.Enqueue(job2)
 	if err != nil {
 		t.Fatalf("Enqueue failed: %v", err)
 	}
 	resultChan3, err := d.Enqueue(job3)
 	if err != nil {
 		t.Fatalf("Enqueue failed: %v", err)
 	}
 	resultChan4, err := d.Enqueue(job4)
 	if err != nil {
 		t.Fatalf("Enqueue failed: %v", err)
 	}
 	resultChan5, err := d.Enqueue(job5)
 	if err != nil {
 		t.Fatalf("Enqueue failed: %v", err)
 	}

 	r1 := <-resultChan1
 	r2 := <-resultChan2

 	// Cancel second job after it has already run.
 	job2.Cancel()

 	// Cancel fourth job after it has been queued, but before it has run.
 	job4.Cancel()

 	r3 := <-resultChan3
 	r4 := <-resultChan4

 	// Wait half a second to give job 5 a chance to start running, then cancel it.
 	time.Sleep(500 * time.Millisecond)
 	job5.Cancel()

 	r5 := <-resultChan5

 	// Check that jobs 1 and 4 failed.
 	if r1.Success {
 		t.Errorf("Expected job 1 to fail but it succeeded.")
 	}
 	if r4.Success {
 		t.Errorf("Expected job 4 to fail but it succeeded.")
 	}

 	// Check that jobs 2, 3, and 5 succeeded.
 	if !r2.Success {
 		t.Errorf("Expected job 2 to succeed but it failed.")
 	}
 	if !r3.Success {
 		t.Errorf("Expected job 3 to succeed but it failed.")
 	}
 	if !r5.Success {
 		t.Errorf("Expected job 5 to succeed but it failed.")
 	}
 }
	// Copyright 2015 The Vanadium Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	package jobqueue

	import (
	"bytes"
	"fmt"
	"os"
	"os/exec"
	"path"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"v.io/x/playground/lib/event"
	)

	var (
	defaultMaxTime = 10 * time.Second
	defaultMaxSize = 1 << 16
	defaultMemLimit = 100
	)

	func init() {
	// Compile builder binary and put in path.
	pgDir := os.ExpandEnv("${V23_ROOT}/release/projects/playground/go")

	cmd := exec.Command("make", "builder")
	cmd.Dir = path.Join(pgDir, "src", "v.io", "x", "playground")
	if out, err := cmd.CombinedOutput(); err != nil {
	fmt.Println("Error running 'make builder'")
	fmt.Println(string(out))
	panic(err)
	}

	pgBinDir := path.Join(pgDir, "bin")
	if err := os.Setenv("PATH", pgBinDir+":"+os.Getenv("PATH")); err != nil {
	panic(err)
	}
	}

	// mockTestFile is a simple Go progam that will be sent to the dispatcher in the
	// test runs.
	var mockTestFile = `
	package main

	import(
	"fmt"
	"time"
	)

	func main() {
	fmt.Printf("PROGRAM START")
	time.Sleep(1 * time.Second)
	fmt.Printf("PROGRAM END")
	}
	`

	// mockTestBody is a bundle containing the mockTestFile, as it would look if it
	// came from an http request from the playground client.
	var mockTestBody = []byte(fmt.Sprintf(`{
	"files": [ {
	"name": "src/main/main.go",
	"body": %v
	} ]
	}`, strconv.Quote(mockTestFile)))

	// testConfig encapsulates all the parameters and expectations for a single
	// test run.
	type testConfig struct {
	// Number of jobs to enqueue.
	jobs int

	// Dispatcher configuration.
	workers int
	jobCap int

	// Job configuration.
	useDocker bool
	maxSize int
	maxTime time.Duration
	memLimit int

	// Test expectations. Default is to expect success.
	expectEnqueueFail bool
	expectOutputTooLarge bool
	expectJobFail bool
	}

	func newMockResponseEventSink() *event.ResponseEventSink {
	var b bytes.Buffer
	return event.NewResponseEventSink(&b, false)
	}

	// eventsMatch returns true iff there is an event whose message matches the
	// given string.
	func eventsMatch(events []event.Event, match string) bool {
	for _, e := range events {
	if strings.Contains(e.Message, match) {
	return true
	}
	}
	return false
	}

	// assertExpectedResult asserts that the result matches the test expectations
	// in the test config.
	func assertExpectedResult(t *testing.T, c testConfig, r Result) {
	expectSuccess := !c.expectJobFail
	if expectSuccess != r.Success {
	t.Errorf("Expected result.Success to be %v but was %v. Test config: %#v", expectSuccess, r.Success, c)
	}

	if !r.Success {
	return
	}

	if c.expectOutputTooLarge {
	want := "Program output too large, killed."
	if !eventsMatch(r.Events, want) {
	t.Errorf("Event message %v not found in %#v. Test config: %#v", want, r.Events, c)
	}
	return
	}

	// Expect normal program output.
	want := "PROGRAM START"
	if !eventsMatch(r.Events, want) {
	t.Errorf("Event message %v not found in %#v. Test config: %#v", want, r.Events, c)
	}

	want = "PROGRAM END"
	if !eventsMatch(r.Events, want) {
	t.Errorf("Event message %v not found in %#v. Test config: %#v", want, r.Events, c)
	}
	}

	// runTest runs a particular test configuration. It starts a dispatcher, queues
	// jobs and waits for them to finish, and asserts that they match the
	// expectations in the testConfig.
	func runTest(t *testing.T, c testConfig) {
	fmt.Printf("Testing %v jobs on %v workers with jobCap of %v and useDocker %v\n", c.jobs, c.workers, c.jobCap, c.useDocker)
	d := NewDispatcher(c.workers, c.jobCap)

	var enqueueError error

	// wg waits for all the jobs to finish.
	var wg sync.WaitGroup

	// Start all the jobs.
	for i := 0; i < c.jobs; i++ {
	res := newMockResponseEventSink()
	job := NewJob(mockTestBody, res, c.maxSize, c.maxTime, c.useDocker, c.memLimit)

	resultChan, err := d.Enqueue(job)
	if err != nil {
	enqueueError = err
	break
	}

	wg.Add(1)

	go func() {
	r := <-resultChan
	assertExpectedResult(t, c, r)
	wg.Done()
	}()
	}

	if !c.expectEnqueueFail && enqueueError != nil {
	t.Fatalf("Enqueue failed: %v. Test config: %#v", enqueueError, c)
	}

	if c.expectEnqueueFail && enqueueError == nil {
	t.Fatalf("Expected Enqueue to fail but it did not. Test config: %#v", c)
	}

	// Test must finish in 30 seconds.
	timeout := time.After(30 * time.Second)

	jobsFinished := make(chan bool)

	go func() {
	wg.Wait()
	jobsFinished <- true
	}()

	// Wait for timeout or all jobs to finish.
	select {
	case <-timeout:
	t.Fatalf("Expected jobs to complete but got timeout. Test config: %#v", c)
	case <-jobsFinished:
	}

	d.Stop()
	}

	func TestJobQueue(t *testing.T) {

	// Test success cases without docker.
	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	useDocker: false,
	})

	runTest(t, testConfig{
	jobs: 3,
	workers: 1,
	jobCap: 3,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	useDocker: false,
	})

	runTest(t, testConfig{
	jobs: 3,
	workers: 3,
	jobCap: 3,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	useDocker: false,
	})

	runTest(t, testConfig{
	jobs: 6,
	workers: 3,
	jobCap: 10,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	useDocker: false,
	})

	// Test success cases with docker.
	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	memLimit: defaultMemLimit,
	useDocker: true,
	})

	runTest(t, testConfig{
	jobs: 3,
	workers: 1,
	jobCap: 3,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	memLimit: defaultMemLimit,
	useDocker: true,
	})

	runTest(t, testConfig{
	jobs: 3,
	workers: 3,
	jobCap: 3,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	memLimit: defaultMemLimit,
	useDocker: true,
	})

	runTest(t, testConfig{
	jobs: 6,
	workers: 3,
	jobCap: 10,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	memLimit: defaultMemLimit,
	useDocker: true,
	})

	// Test Enqueue should fail when job capacity is exceeded.
	runTest(t, testConfig{
	jobs: 5,
	workers: 2,
	jobCap: 2,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	useDocker: false,
	expectEnqueueFail: true,
	})

	runTest(t, testConfig{
	jobs: 5,
	workers: 2,
	jobCap: 2,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	memLimit: defaultMemLimit,
	useDocker: true,
	expectEnqueueFail: true,
	})

	// Test job should fail if it exceeds size limit.
	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: 100,
	maxTime: defaultMaxTime,
	useDocker: false,
	expectOutputTooLarge: true,
	})

	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: 100,
	maxTime: defaultMaxTime,
	memLimit: defaultMemLimit,
	useDocker: true,
	expectOutputTooLarge: true,
	})

	// Test job should fail if it exceeds max time.
	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: defaultMaxSize,
	maxTime: 1 * time.Second,
	useDocker: false,
	expectJobFail: true,
	})

	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: defaultMaxSize,
	maxTime: 1 * time.Second,
	useDocker: true,
	expectJobFail: true,
	})

	// Test job should fail if it exceeds memory limit (docker only).
	runTest(t, testConfig{
	jobs: 1,
	workers: 1,
	jobCap: 1,
	maxSize: defaultMaxSize,
	maxTime: defaultMaxTime,
	memLimit: 5,
	useDocker: true,
	expectJobFail: true,
	})
	}

	func TestJobCancel(t *testing.T) {
	d := NewDispatcher(1, 10)

	// Create five jobs.
	res1 := newMockResponseEventSink()
	job1 := NewJob(mockTestBody, res1, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

	res2 := newMockResponseEventSink()
	job2 := NewJob(mockTestBody, res2, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

	res3 := newMockResponseEventSink()
	job3 := NewJob(mockTestBody, res3, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

	res4 := newMockResponseEventSink()
	job4 := NewJob(mockTestBody, res4, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

	res5 := newMockResponseEventSink()
	job5 := NewJob(mockTestBody, res5, defaultMaxSize, defaultMaxTime, false, defaultMemLimit)

	// Cancel first job right away.
	job1.Cancel()

	// Queue all jobs.
	resultChan1, err := d.Enqueue(job1)
	if err != nil {
	t.Fatalf("Enqueue failed: %v", err)
	}
	resultChan2, err := d.Enqueue(job2)
	if err != nil {
	t.Fatalf("Enqueue failed: %v", err)
	}
	resultChan3, err := d.Enqueue(job3)
	if err != nil {
	t.Fatalf("Enqueue failed: %v", err)
	}
	resultChan4, err := d.Enqueue(job4)
	if err != nil {
	t.Fatalf("Enqueue failed: %v", err)
	}
	resultChan5, err := d.Enqueue(job5)
	if err != nil {
	t.Fatalf("Enqueue failed: %v", err)
	}

	r1 := <-resultChan1
	r2 := <-resultChan2

	// Cancel second job after it has already run.
	job2.Cancel()

	// Cancel fourth job after it has been queued, but before it has run.
	job4.Cancel()

	r3 := <-resultChan3
	r4 := <-resultChan4

	// Wait half a second to give job 5 a chance to start running, then cancel it.
	time.Sleep(500 * time.Millisecond)
	job5.Cancel()

	r5 := <-resultChan5

	// Check that jobs 1 and 4 failed.
	if r1.Success {
	t.Errorf("Expected job 1 to fail but it succeeded.")
	}
	if r4.Success {
	t.Errorf("Expected job 4 to fail but it succeeded.")
	}

	// Check that jobs 2, 3, and 5 succeeded.
	if !r2.Success {
	t.Errorf("Expected job 2 to succeed but it failed.")
	}
	if !r3.Success {
	t.Errorf("Expected job 3 to succeed but it failed.")
	}
	if !r5.Success {
	t.Errorf("Expected job 5 to succeed but it failed.")
	}
	}