-
Notifications
You must be signed in to change notification settings - Fork 44
/
workerqueue.go
117 lines (98 loc) · 2.61 KB
/
workerqueue.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
package main
import (
"fmt"
"strings"
"sync"
"sync/atomic"
)
type workerQueue struct {
// manifest job channel
jobQueue chan manifestJob
// channel for sending messages from jobs to the printer
msgQueue chan string
// channel for sending errors from jobs to the collector
errQueue chan error
// global error list
errors []error
// total job count defined on workerQueue creation
// sets the length of the job queue so that pushing to the queue doesn't block
njobs uint32
// total workers defined on workerQueue creation
nworkers uint32
// active worker count
activeWorkers int32
// wait group for all workers
workerWG sync.WaitGroup
// wait group for internal routines (printer and error collector)
utilWG sync.WaitGroup
}
func newWorkerQueue(nworkers uint32, njobs uint32) *workerQueue {
wq := workerQueue{
jobQueue: make(chan manifestJob, njobs),
msgQueue: make(chan string, nworkers),
errQueue: make(chan error, nworkers),
errors: make([]error, 0, nworkers),
nworkers: nworkers,
activeWorkers: 0,
njobs: njobs,
}
return &wq
}
func (wq *workerQueue) start() {
wq.startMessagePrinter()
wq.startErrorCollector()
for idx := uint32(0); idx < wq.nworkers; idx++ {
wq.startWorker(idx)
}
}
// close all queues and wait for waitgroups
func (wq *workerQueue) wait() []error {
// close job channel and wait for workers to finish
close(wq.jobQueue)
wq.workerWG.Wait()
// close message channels and wait for them to finish their work so we don't miss any messages or errors
close(wq.msgQueue)
close(wq.errQueue)
wq.utilWG.Wait()
return wq.errors
}
func (wq *workerQueue) startWorker(idx uint32) {
wq.workerWG.Add(1)
go func() {
atomic.AddInt32(&(wq.activeWorkers), 1)
defer atomic.AddInt32(&(wq.activeWorkers), -1)
defer wq.workerWG.Done()
for job := range wq.jobQueue {
err := job(wq.msgQueue)
if err != nil {
wq.errQueue <- err
}
}
}()
}
func (wq *workerQueue) startMessagePrinter() {
wq.utilWG.Add(1)
go func() {
defer wq.utilWG.Done()
var msglen int
for msg := range wq.msgQueue {
// clear previous line (avoids leftover trailing characters from progress)
fmt.Printf(strings.Repeat(" ", msglen) + "\r")
fmt.Println(msg)
msglen, _ = fmt.Printf(" == Jobs == Queue: %4d Active: %4d Total: %4d\r", len(wq.jobQueue), wq.activeWorkers, wq.njobs)
}
fmt.Println()
}()
}
func (wq *workerQueue) startErrorCollector() {
wq.utilWG.Add(1)
go func() {
defer wq.utilWG.Done()
for err := range wq.errQueue {
wq.errors = append(wq.errors, err)
}
}()
}
func (wq *workerQueue) submitJob(j manifestJob) {
wq.jobQueue <- j
}