Go Project: Job Runner

A job runner is a program that gives pieces of work to worker goroutines and collects their results. This runner runs five jobs on three workers and prints a report when the work ends.

Count the finished jobs

countIf is a generic helper that counts the items a check keeps. Here the check keeps results marked done, and the count prints.

package main

import "fmt"

type Result struct {
	ID    int
	Value int
	Done  bool
}

func countIf[T any](items []T, keep func(T) bool) int {
	count := 0
	for _, item := range items {
		if keep(item) {
			count++
		}
	}
	return count
}

func main() {
	results := []Result{
		{ID: 1, Value: 9, Done: true},
		{ID: 2, Value: 0, Done: false},
		{ID: 3, Value: 49, Done: true},
	}
	finished := countIf(results, func(r Result) bool {
		return r.Done
	})
	fmt.Println("Finished:", finished)
}

Workers fill indexed slots

Two workers pull jobs from one channel. Each result goes into the slot for its job ID, so the slice prints in job order.

package main

import (
	"fmt"
	"sync"
)

type Job struct {
	ID   int
	Work int
	Fail bool
}

type Result struct {
	ID    int
	Value int
	Done  bool
}

func main() {
	jobs := []Job{
		{ID: 1, Work: 3, Fail: false},
		{ID: 2, Work: 5, Fail: false},
		{ID: 3, Work: 7, Fail: false},
	}
	results := make([]Result, len(jobs))
	jobCh := make(chan Job)
	var wg sync.WaitGroup
	for w := 0; w < 2; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobCh {
				results[job.ID-1] = Result{ID: job.ID, Value: job.Work * job.Work, Done: true}
			}
		}()
	}
	for _, job := range jobs {
		jobCh <- job
	}
	close(jobCh)
	wg.Wait()
	fmt.Println(results)
}

A failure travels on an error channel

One job is marked to fail, and its worker sends an error instead of a finished value. Main collects the errors after the workers finish and prints the failed count.

package main

import (
	"fmt"
	"sync"
)

type Job struct {
	ID   int
	Work int
	Fail bool
}

type Result struct {
	ID    int
	Value int
	Done  bool
}

func main() {
	jobs := []Job{
		{ID: 1, Work: 3, Fail: false},
		{ID: 2, Work: 5, Fail: true},
		{ID: 3, Work: 7, Fail: false},
	}
	results := make([]Result, len(jobs))
	jobCh := make(chan Job)
	errCh := make(chan error, len(jobs))
	var wg sync.WaitGroup
	for w := 0; w < 2; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobCh {
				if job.Fail {
					results[job.ID-1] = Result{ID: job.ID, Done: false}
					errCh <- fmt.Errorf("job %d failed", job.ID)
				} else {
					results[job.ID-1] = Result{ID: job.ID, Value: job.Work * job.Work, Done: true}
				}
			}
		}()
	}
	for _, job := range jobs {
		jobCh <- job
	}
	close(jobCh)
	wg.Wait()
	close(errCh)
	failed := 0
	for range errCh {
		failed++
	}
	fmt.Println(results)
	fmt.Println("Failed:", failed)
}

Example

package main

import (
	"encoding/json"
	"fmt"
	"sync"
)

type Job struct {
	ID   int
	Work int
	Fail bool
}

type Result struct {
	ID    int
	Value int
	Done  bool
}

func countIf[T any](items []T, keep func(T) bool) int {
	count := 0
	for _, item := range items {
		if keep(item) {
			count++
		}
	}
	return count
}

func main() {
	jobs := []Job{
		{ID: 1, Work: 3, Fail: false},
		{ID: 2, Work: 5, Fail: true},
		{ID: 3, Work: 7, Fail: false},
		{ID: 4, Work: 2, Fail: false},
		{ID: 5, Work: 4, Fail: false},
	}
	results := make([]Result, len(jobs))
	jobCh := make(chan Job)
	errCh := make(chan error, len(jobs))
	var wg sync.WaitGroup
	for w := 0; w < 3; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobCh {
				if job.Fail {
					results[job.ID-1] = Result{ID: job.ID, Done: false}
					errCh <- fmt.Errorf("job %d failed", job.ID)
				} else {
					results[job.ID-1] = Result{ID: job.ID, Value: job.Work * job.Work, Done: true}
				}
			}
		}()
	}
	for _, job := range jobs {
		jobCh <- job
	}
	close(jobCh)
	wg.Wait()
	close(errCh)
	failed := 0
	for range errCh {
		failed++
	}
	finished := countIf(results, func(r Result) bool {
		return r.Done
	})
	data, err := json.Marshal(results)
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println(string(data))
	fmt.Printf("Ran: %d Failed: %d\n", finished+failed, failed)
}

Key points

Three workers pull jobs from the jobs channel until main closes it.
Each worker writes into the slot that matches the job ID, so results stay in ID order.
A job marked Fail sends an error value on the error channel and leaves its slot marked not done.
The error channel is buffered with room for every job, so sending an error never blocks a worker.
Main closes the jobs channel, joins the workers with the WaitGroup, and then closes and drains the error channel.
countIf counts the finished results, and the failed count comes from the collected errors.
The report marshals the results slice with encoding/json, so field order follows the Result struct.

In this example, five jobs go into the jobs channel and three workers pull them out. Each worker squares the Work value and stores it in the slot for that job ID. Job 2 is marked to fail, so its worker marks the slot not done and sends an error. Main closes the channel, waits for the workers, and collects one error. countIf counts four finished results, and the results print as JSON in ID order. The last line prints Ran: 5 Failed: 1.

Try it Yourself

Edit the code, then press Run.

Running is simulated · Go Playground

package main

import (
	"encoding/json"
	"fmt"
	"sync"
)

type Job struct {
	ID   int
	Work int
	Fail bool
}

type Result struct {
	ID    int
	Value int
	Done  bool
}

func countIf[T any](items []T, keep func(T) bool) int {
	count := 0
	for _, item := range items {
		if keep(item) {
			count++
		}
	}
	return count
}

func main() {
	jobs := []Job{
		{ID: 1, Work: 3, Fail: false},
		{ID: 2, Work: 5, Fail: true},
		{ID: 3, Work: 7, Fail: false},
		{ID: 4, Work: 2, Fail: false},
		{ID: 5, Work: 4, Fail: false},
	}
	results := make([]Result, len(jobs))
	jobCh := make(chan Job)
	errCh := make(chan error, len(jobs))
	var wg sync.WaitGroup
	for w := 0; w < 3; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobCh {
				if job.Fail {
					results[job.ID-1] = Result{ID: job.ID, Done: false}
					errCh <- fmt.Errorf("job %d failed", job.ID)
				} else {
					results[job.ID-1] = Result{ID: job.ID, Value: job.Work * job.Work, Done: true}
				}
			}
		}()
	}
	for _, job := range jobs {
		jobCh <- job
	}
	close(jobCh)
	wg.Wait()
	close(errCh)
	failed := 0
	for range errCh {
		failed++
	}
	finished := countIf(results, func(r Result) bool {
		return r.Done
	})
	data, err := json.Marshal(results)
	if err != nil {
		fmt.Println(err)
		return
	}
	fmt.Println(string(data))
	fmt.Printf("Ran: %d Failed: %d\n", finished+failed, failed)
}

Exercise

Why does the report list results in job ID order when workers finish jobs at different times?

The jobs channel sorts the jobs by ID. JSON sorts struct fields by name. Each worker writes into the slot that matches its job ID. The WaitGroup prints the results itself.