Designing ASICs, or Application Specific Integrated Circuits, is a process that combines computer science and blends it with physics. Like algorithms coded in a language like C++ or Java, there's a corollary in ASIC design that determines how fast it will run. To find this out, however, requires quite a lot of compute time. ASIC design teams are known for assembling clusters with the latest Intel Xeons, the fastest DRAM that reaches capacities per server of 512 GB, and a wickely-fast disk array that handle a few hundred disk-intensive jobs simultaneously.
There are some compute jobs that are parallel. No job is linked to the success or failure (job runs, but the exercise does not produce a satisfactory result) of any other job. This is what many people in the HPC community call "embarrassingly parallel".
Of course, there are other embarrassingly parallel jobs in the computer science realm as well. When I modify a line of code, I can rerun the suite of unit tests to ensure I did not suffer a regression of functionality. All tests can be run independent of one another.
I've often needed the ability to run many jobs, but limit the number of active jobs to some fixed number. For instance, I may have a quad-core laptop and I'd like to keep 1 core available for Safari and Mail while the other 3 chug their way through the tasks. Or, a more likely scenario is to submit 1000 jobs to a compute cluster, but run no more than 50 at a time. I don't want to launch 50, wait for them all to finish, and then launch another 50. I want to launch job 51 when one of the first 50 jobs finish.
In the past, I've turned to Ruby for this. However, I last wrote that code a couple of employers ago and no longer have legal access to use it. No worries, it's a common pattern, I can write it again. Further, I've been having a lot of fun writing Go over the past 2 years, so I'll redo it in style.
First, I'm leveraging the notion of an Object pool pattern, but these objects will instead be goroutines. Because I'm being a little generic, I'll assume that a file exists that contains every job to be run, one per line. For instance:
sleep 1 sleep 2 sleep 3 sleep 4
This file consists of 4 jobs. Each job will sleep a specific number of seconds before returning. I'll first create a task that reads a file and returns a slice of strings, each index containing one command-line job.
// getTasks reads file containing tasks, formatted one per line, and returns the array of tasks. func getTasks(f string) ([]string, error) { rval := []string{} contents, err := ioutil.ReadFile(f) if err != nil { return []string{}, err } sp := strings.Split(string(contents), "\n") for _, s := range sp { trim := strings.Trim(string(s), " ") if trim != "" { rval = append(rval, trim) } } return rval, nil }
Now that we have the tasks, we can look at launching a number of goroutines equal to a command-line flag.
// Start workers wg := new(sync.WaitGroup) // allocate a wait group for i := 0; i < \*workers; i++ { wg.Add(1) // Increment waitgroup for every concurrent goroutine. go func(pipeline chan Task, wg \*sync.WaitGroup) { defer wg.Done() // defer the waitgroup decrement until the goroutine exits for task := range pipeline { // range the channel until the channel is closed. args := []string(task) cmd := exec.Command(args[0], args[1:]...) // Run the command if err := cmd.Run(); err != nil { log.Fatal(err) } } }(ch, wg) }
There a lot of work going on here. First, workers contains a pointer to a flag variable indicating how many goroutines are being requested. The sync.WaitGroup is a threadsafe object that can be atomically incremented (wg.Add(1)) or decremented (wg.Done()). In this case, the main thread is incrementing the count for every goroutine created, and each goroutine will decrement when it exits.
Once the workers are started, the commands contained in the file are sent into the channel. When the last one is sent, the channel is closed.
ch := make(chan Task) for _, t := range tasks { fmt.Printf("scheduling '%s'\n", t) ch <- Task(strings.Split(t, " ")) } close(ch)
Closing the channel while goroutines are running is quite alright. Close does not deallocate the channel, it is simply waiting to be closed when it empties.
The main thread must now block on the waitgroup. We know that the goroutines will continue to exist until it loops back to a closed channel, which happens when all work has been removed. All goroutines will decrement the waitgroup just prior to exit, which will return control to the main loop.
fmt.Println("Waiting for tasks to finish...") wg.Wait()
I hope you see how powerful the concurrency features are in Go, and can envision some way for you to take advantage of goroutines and the built-in synchronization to speed up your applications!
The full program:
package main import ( "flag" "fmt" "io/ioutil" "log" "os/exec" "strings" "sync" ) type Task []string func main() { var file = flag.String("f", "file", "Input file containing tasks, one per line") var workers = flag.Int("w", 1, "Number of worker threads") flag.Parse() ch := make(chan Task) tasks, err := getTasks(\*file) if err != nil { log.Fatal(err) } wg := new(sync.WaitGroup) // Start workers for i := 0; i < \*workers; i++ { wg.Add(1) go func(pipeline chan Task, wg \*sync.WaitGroup) { defer wg.Done() for task := range pipeline { args := []string(task) fmt.Printf("executing '%s'\n", strings.Join(args, " ")) cmd := exec.Command(args[0], args[1:]...) if err := cmd.Run(); err != nil { log.Fatal(err) } } }(ch, wg) } fmt.Printf("Read %d tasks\n", len(tasks)) for _, t := range tasks { fmt.Printf("scheduling '%s'\n", t) ch <- Task(strings.Split(t, " ")) } close(ch) fmt.Println("Waiting for tasks to finish...") wg.Wait() } // getTasks reads file containing tasks, formatted one per line, and returns the array of tasks. func getTasks(f string) ([]string, error) { rval := []string{} contents, err := ioutil.ReadFile(f) if err != nil { return []string{}, err } sp := strings.Split(string(contents), "\n") for _, s := range sp { trim := strings.Trim(string(s), " ") if trim != "" { rval = append(rval, trim) } } return rval, nil }
