Run commands concurrently. Inspired by xargs.

parallel-exec.go

package main

import (
	"bufio"
	"flag"
	"fmt"
	"os"
	"os/exec"
	"strings"
	"sync"
)

type Job struct {
	workers int
	tasks   chan *exec.Cmd
	wg      sync.WaitGroup
}

func NewJob(n int) (*Job, error) {
	w := &Job{
		workers: n,
		tasks:   make(chan *exec.Cmd),
	}
	return w, nil
}

func (w *Job) Start() {
	for i := 1; i <= w.workers; i++ {
		go w.start(w.tasks)
	}
}

func (w *Job) start(tasks chan *exec.Cmd) {
	for task := range tasks {
		w.run(task)
	}
}

func (w *Job) Close() {
	close(w.tasks)
}

func (w *Job) Wait() {
	w.wg.Wait()
}

func (w *Job) Send(task *exec.Cmd) {
	w.tasks <- task
}

func (w *Job) run(cmd *exec.Cmd) {
	w.wg.Add(1)
	fmt.Fprintf(os.Stdout, "Running %s command\n", cmd.Args)
	err := cmd.Run()
	if err != nil {
		fmt.Fprintf(os.Stderr, "Error while running %s command: %s\n", cmd.Args, err.Error())
	}
	w.wg.Done()
}

func main() {
	var nWorker = flag.Int("w", 3, "")
	flag.Parse()

	in := bufio.NewReader(os.Stdin)
	sc := bufio.NewScanner(in)

	worker, err := NewJob(*nWorker)
	if err != nil {
		fmt.Fprintf(os.Stderr, "Error while creating new worker: %s\n", err.Error())
		os.Exit(1)
	}
	go worker.Start()
	for sc.Scan() {
		cmdStr := strings.Fields(sc.Text())
		cmd := exec.Command("/bin/bash", cmdStr...)
		worker.Send(cmd)
	}
	if sc.Err() != nil {
		fmt.Fprintf(os.Stderr, "Error while scanning stdin: %s\n", err.Error())
		os.Exit(1)
	}
	worker.Close()
	worker.Wait()
}