sts.go

package main

import (
	"fmt"
	"sort"
	"time"
)

const (
	workers = 10  // how many worker goroutines to spawn
	iters   = 100 // how many iterations of work each worker does
)

var (
	workTime = 20 * time.Millisecond // the amount of time that simulated work takes
)

// struct that holds a counter
type st struct {
	counter int
}

func (s *st) String() string {
	return fmt.Sprintf("%v", s.counter)
}

// an update to the counter of an st
type update struct {
	s *st // the st to be updated
	c int // the amount by which to increment the count
}

func main() {
	// Create a channel to collect updates
	// It is buffered because that seems to allow the logic as written to work,
	// and in particular setting the buffer depth to equal the number of workers
	// seems to allow us to process updates from all workers without starving
	// any individual ones.
	// Try experimenting with different channel sizes to see how the behavior
	// changes.
	updates := make(chan update, workers)

	// Make a bunch of structs and spawn workers to work on them
	sts := make([]*st, 0, workers)
	for i := 0; i < workers; i++ {
		s := &st{}
		sts = append(sts, s)
		go work(s, updates)
	}

	// Here's the crux of the solution - use a single goroutine to both handle
	// updating the counts as well as doing the "scheduling" work.
	for {
		// Process the next update, but only if available.  That's what select
		// with a default clause accomplishes.
		select {
		case u := <-updates:
			// update counter
			u.s.counter += u.c
		default:
			// no update available, just continue
		}
		sort.Sort(ByCounter(sts))
		fmt.Printf("Scheduling with: %v\n", sts)

		if sts[0].counter == iters && sts[workers-1].counter == iters {
			fmt.Println("Finished")
			break
		}
	}
}

// work simulates a worker by looping, sleeping in each loop and submitting an
// updated count. The key to this is that it uses the update struct as an
// accumulator.
func work(s *st, updates chan update) {
	u := update{s, 0}
	for i := 0; i < iters; i++ {
		// do some work
		fmt.Println("Working")
		time.Sleep(workTime)
		u.c += 1

		// Submit the update, but only if there's room on the updates channel.
		// Otherwise, just continue accumulating.
		select {
		case updates <- u:
			// successfully submitted update, reset
			u = update{s, 0}
		default:
			last := i == iters-1
			if last {
				// wait for the last update to submit
				updates <- u
			}
		}
	}
}

type ByCounter []*st

func (a ByCounter) Len() int           { return len(a) }
func (a ByCounter) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByCounter) Less(i, j int) bool { return a[i].counter < a[j].counter }