Merge pull request #10 from bvobart/parallel-linters

Make `mllint`s linters run in parallel instead of sequentially

commands/mllint/collector.go

@@ -0,0 +1,54 @@
+package mllint
+
+func ForEachTask(tasks chan *RunnerTask, f func(task *RunnerTask, result LinterResult)) {
+	for {
+		task, open := <-tasks
+		if !open {
+			return
+		}
+
+		f(task, <-task.Result)
+	}
+}
+
+func CollectTasks(tasks ...*RunnerTask) chan *RunnerTask {
+	c := collector{
+		total:  len(tasks),
+		done:   make(chan struct{}),
+		funnel: make(chan *RunnerTask),
+	}
+
+	for _, task := range tasks {
+		go c.awaitResult(task)
+	}
+
+	go c.awaitDone()
+	return c.funnel
+}
+
+type collector struct {
+	total  int
+	done   chan struct{}
+	funnel chan *RunnerTask
+}
+
+func (c *collector) awaitResult(task *RunnerTask) {
+	result := <-task.Result
+	task.Result <- result
+	c.funnel <- task
+	c.done <- struct{}{}
+}
+
+func (c *collector) awaitDone() {
+	nDone := 0
+	for {
+		<-c.done
+		nDone++
+
+		if nDone == c.total {
+			close(c.funnel)
+			close(c.done)
+			return
+		}
+	}
+}

commands/mllint/linterfaces.go

@@ -0,0 +1,17 @@
+package mllint
+
+import "github.com/bvobart/mllint/api"
+
+type WithRunner interface {
+	SetRunner(runner *Runner)
+}
+
+type LinterWithRunner interface {
+	api.Linter
+	WithRunner
+}
+
+type ConfigurableLinterWithRunner interface {
+	api.ConfigurableLinter
+	WithRunner
+}

commands/mllint/progress.go

@@ -0,0 +1,35 @@
+package mllint
+
+import (
+	"io"
+
+	"github.com/fatih/color"
+)
+
+type RunnerProgress interface {
+	Start()
+	RunningTask(task *RunnerTask)
+	CompletedTask(task *RunnerTask)
+	AllTasksDone()
+}
+
+type status string
+
+const (
+	statusRunning status = "⏳ Running -"
+	statusDone    status = "✔️ Done -"
+)
+
+type taskStatus struct {
+	*RunnerTask
+	Status status
+}
+
+func (s taskStatus) PrintStatus(writer io.Writer) {
+	if s.Status == statusRunning {
+		color.New(color.FgYellow).Fprintln(writer, s.Status, s.displayName)
+	}
+	if s.Status == statusDone {
+		color.New(color.FgGreen).Fprintln(writer, s.Status, s.displayName)
+	}
+}

commands/mllint/progress_basic.go

@@ -0,0 +1,33 @@
+package mllint
+
+import (
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/fatih/color"
+)
+
+type BasicRunnerProgress struct {
+	Out io.Writer
+}
+
+func NewBasicRunnerProgress() RunnerProgress {
+	return &BasicRunnerProgress{os.Stdout}
+}
+
+func (p *BasicRunnerProgress) Start() {}
+
+func (p *BasicRunnerProgress) RunningTask(task *RunnerTask) {
+	taskStatus{task, statusRunning}.PrintStatus(p.Out)
+}
+
+func (p *BasicRunnerProgress) CompletedTask(task *RunnerTask) {
+	taskStatus{task, statusDone}.PrintStatus(p.Out)
+}
+
+func (p *BasicRunnerProgress) AllTasksDone() {
+	fmt.Fprintln(p.Out)
+	color.New(color.Bold, color.FgGreen).Fprintln(p.Out, "✔️ All done!")
+	fmt.Fprint(p.Out, "\n---\n\n")
+}

commands/mllint/progress_live.go

@@ -0,0 +1,126 @@
+package mllint
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/fatih/color"
+	"github.com/gosuri/uilive"
+)
+
+// LiveRunnerProgress is used by an `mllint.Runner` to keep track of and pretty-print the progress of the runner in running its tasks.
+//
+// Unless you're changing the implementation of `mllint.Runner`, you probably don't need to interact with this.
+type LiveRunnerProgress struct {
+	w *uilive.Writer
+
+	running chan *RunnerTask
+	done    chan *RunnerTask
+
+	stopped chan struct{}
+
+	// The following properties may only be edited inside p.printWorker()
+
+	// list of all tasks that have been run / scheduled
+	tasks []taskStatus
+	// maps the tasks' IDs to their index in `tasks`, since iterating over a map in Go does not happen in order of insertion and I don't want to do an O(n) search through `tasks` when updating a task's status.
+	taskIndexes map[string]int
+}
+
+func NewLiveRunnerProgress() RunnerProgress {
+	writer := uilive.New()
+	writer.RefreshInterval = time.Hour
+	return &LiveRunnerProgress{
+		w:           writer,
+		running:     make(chan *RunnerTask, queueSize),
+		done:        make(chan *RunnerTask, queueSize),
+		stopped:     make(chan struct{}),
+		tasks:       []taskStatus{},
+		taskIndexes: make(map[string]int),
+	}
+}
+
+// Start starts the printWorker process on a new go-routine.
+func (p *LiveRunnerProgress) Start() {
+	p.w.Start()
+	go p.printWorker()
+}
+
+// RunningTask is the way for the `mllint.Runner` to signal that it has started running a task.
+func (p *LiveRunnerProgress) RunningTask(task *RunnerTask) {
+	p.running <- task
+}
+
+// CompletedTask is the way for the `mllint.Runner` to signal that it has completed running a task.
+func (p *LiveRunnerProgress) CompletedTask(task *RunnerTask) {
+	p.done <- task
+}
+
+// AllTasksDone is the way for the `mllint.Runner` to signal that it has finished running all tasks,
+// and that it won't call p.CompletedTask anymore (if it does, it panics because `p.done` is closed).
+// This method will wait until the printWorker has finished printing and has shutdown.
+func (p *LiveRunnerProgress) AllTasksDone() {
+	close(p.done)
+	<-p.stopped
+}
+
+func (p *LiveRunnerProgress) printWorker() {
+	for {
+		select {
+		case task, open := <-p.running:
+			if !open {
+				break
+			}
+
+			p.onTaskRunning(task)
+		case task, open := <-p.done:
+			if !open {
+				p.w.Stop()
+				close(p.stopped)
+				return
+			}
+
+			p.onTaskDone(task)
+		}
+	}
+}
+
+func (p *LiveRunnerProgress) onTaskRunning(task *RunnerTask) {
+	p.taskIndexes[task.Id] = len(p.tasks)
+	p.tasks = append(p.tasks, taskStatus{task, statusRunning})
+
+	p.printTasks()
+}
+
+func (p *LiveRunnerProgress) onTaskDone(task *RunnerTask) {
+	index, found := p.taskIndexes[task.Id]
+	if !found {
+		return
+	}
+
+	status := p.tasks[index]
+	status.Status = statusDone
+	p.tasks[index] = status
+
+	p.printTasks()
+}
+
+func (p *LiveRunnerProgress) printTasks() {
+	allDone := true
+	for _, task := range p.tasks {
+		if task.Status != statusDone {
+			allDone = false
+		}
+
+		writer := p.w.Newline()
+		task.PrintStatus(writer)
+	}
+
+	if allDone {
+		fmt.Fprintln(p.w.Newline())
+		color.New(color.Bold, color.FgGreen).Fprintln(p.w.Newline(), "✔️ All done!")
+		fmt.Fprint(p.w.Newline(), "\n---\n\n")
+	}
+
+	p.w.Flush()
+}

commands/mllint/progress_test.go

@@ -0,0 +1,52 @@
+package mllint_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/bvobart/mllint/commands/mllint"
+	"github.com/gosuri/uilive"
+)
+
+func TestProgressTest(t *testing.T) {
+	t.Skip("Skipping this uilive.Newline() demo")
+
+	writer := uilive.New()
+	writer2 := writer.Newline()
+	// start listening for updates and render
+	writer.Start()
+
+	for i := 0; i <= 100; i++ {
+		fmt.Fprintf(writer, "Downloading.. (%d/%d) GB\n", i, 100)
+		fmt.Fprintf(writer2, "Downloading.. (%d/%d) MB\n", i, 100)
+		time.Sleep(time.Millisecond * 5)
+	}
+
+	fmt.Fprintln(writer, "Finished: Downloaded 100GB")
+	writer.Stop() // flush and stop rendering
+}
+
+func TestProgressLive(t *testing.T) {
+	progress := mllint.NewLiveRunnerProgress()
+	progress.Start()
+
+	numTasks := 10
+	tasks := []*mllint.RunnerTask{}
+	for i := 0; i < numTasks; i++ {
+		task := &mllint.RunnerTask{Id: fmt.Sprint(i)}
+		mllint.DisplayName(fmt.Sprint("Task ", i))(task)
+		tasks = append(tasks, task)
+	}
+
+	for _, task := range tasks {
+		progress.RunningTask(task)
+		time.Sleep(time.Millisecond * 10)
+	}
+	for _, task := range tasks {
+		progress.CompletedTask(task)
+		time.Sleep(time.Millisecond * 10)
+	}
+
+	progress.AllTasksDone()
+}