go · level 3 check

a concurrent job runner.

Time to ship a real tool! `jobrun` reads a list of jobs, runs them on a pool of workers and prints a tidy report. The first line of input says how many workers to use. Every other line is a job: a name and how many milliseconds it takes, such as `build 30`.

Inside, `NewRunner` takes functional options, `Run` feeds the jobs to the workers through a channel, and a closer goroutine closes the results once every worker is done. Each job gets its own context with a timeout of 100 milliseconds, so a job that takes too long is stopped and reported as `timed out`.

A shipped tool is polite about bad input: it skips blank lines, explains every line it cannot read, and refuses a worker count that is not a positive number. Workers finish in any order, so the report is sorted by name. Fill in each `____`, then try your own job lists.

Your program

package main

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
)

type Job struct {
	Line   int
	Name   string
	Millis int
}

type Result struct {
	Job Job
	Err error
}

type Runner struct {
	workers int
	timeout time.Duration
}

type Option func(*Runner)

func WithWorkers(n int) Option {
	return func(r *Runner) {
		if n > 0 {
			r.workers = ____
		}
	}
}

func WithTimeout(d time.Duration) Option {
	return func(r *Runner) {
		if d > 0 {
			r.timeout = d
		}
	}
}

func NewRunner(opts ...Option) *Runner {
	r := &Runner{workers: 2, timeout: 100 * time.Millisecond}
	for _, opt := range opts {
		opt(r)
	}
	return r
}

func work(ctx context.Context, job Job) error {
	select {
	case <-time.After(time.Duration(job.Millis) * time.Millisecond):
		return nil
	case <-ctx.____():
		return ctx.Err()
	}
}

func (r *Runner) Run(jobs []Job) []Result {
	queue := make(chan Job)
	out := make(chan Result)
	var wg sync.WaitGroup
	for w := 0; w < r.workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range queue {
				ctx, cancel := context.WithTimeout(context.Background(), r.timeout)
				err := work(ctx, job)
				____()
				if errors.Is(err, context.DeadlineExceeded) {
					err = errors.New("timed out")
				}
				out <- Result{Job: job, Err: err}
			}
		}()
	}
	go func() {
		for _, job := range jobs {
			queue <- job
		}
		close(____)
	}()
	go func() {
		wg.____()
		close(out)
	}()
	var results []Result
	for res := range out {
		results = append(results, res)
	}
	sort.Slice(results, func(i, j int) bool {
		a, b := results[i].Job, results[j].Job
		if a.Name != b.Name {
			return a.Name ____ b.Name
		}
		return a.Line < b.Line
	})
	return results
}

func parse(line string, n int) (Job, error) {
	fields := strings.Fields(line)
	if len(fields) != ____ {
		return Job{}, fmt.Errorf("line %d: want a name and a time", n)
	}
	ms, err := strconv.Atoi(fields[1])
	if err != nil || ms < 0 {
		return Job{}, fmt.Errorf("line %d: bad time %q", n, fields[1])
	}
	return Job{Line: n, Name: fields[0], Millis: ms}, nil
}

func main() {
	scanner := bufio.NewScanner(os.Stdin)
	if !scanner.Scan() {
		fmt.Println("no input")
		return
	}
	workers, err := strconv.Atoi(strings.TrimSpace(scanner.Text()))
	if err != nil || workers < 1 {
		fmt.Println("first line must be the number of workers")
		return
	}
	var jobs []Job
	skipped := 0
	for n := 2; scanner.Scan(); n++ {
		line := strings.TrimSpace(scanner.Text())
		if line == "" {
			continue
		}
		job, err := parse(line, n)
		if err != nil {
			fmt.Println("skipped", err)
			skipped++
			continue
		}
		jobs = append(jobs, job)
	}
	ok, failed := 0, 0
	for _, res := range NewRunner(WithWorkers(workers)).Run(jobs) {
		if res.Err != nil {
			failed++
			fmt.Printf("%-8s failed: %v\n", res.Job.Name, res.Err)
			continue
		}
		ok++
		fmt.Printf("%-8s ok\n", res.Job.Name)
	}
	fmt.Printf("%d ok, %d failed, %d skipped\n", ok, failed, skipped)
}

Fill in every ____ then run the checks.

1 It runs every job and sorts the reportuses: worker pools
2 It stops a slow job on timeuses: context timeouts
3 It skips bad lines and says whyuses: handling bad input
4 It refuses a bad worker countuses: handling bad input
5 It copes with no jobs at alluses: closing channels
6 It keeps the order fixed with one worker and repeated namesuses: sorting results

Take as long as you like and try as many times as you want.

a concurrent job runner.

level up.

playground