Job Queue in Golang

1. what is the different unbuffered and buffered channel?
2. how to implement a job queue in golang?
3. how to stop the worker in a container?
4. Shutdown with Sigterm Handling
5. Canceling Workers without Context
6. Graceful shutdown with worker
7. How to auto-scaling build agent?
8. How to cancel the current Job?

Transcript

1. Job Queue in Golang Bo-Yi Wu 2019.10.19
2. About me • Software Engineer in Mediatek • Member of Drone CI/CD Platform • Member of Gitea Platform • Member of Gin Golang Framework • Maintain Some GitHub Actions Plugins. • Teacher of Udemy Platform: Golang + Drone
4. Queue (Open Source) 3BCCJU.2
   /42
5. Why not use Open Source?
6. Single Service • Build in SQLite (LDAP, Job List) • Build in Cache (LRU) • Build for single binary • No third-party dependency
7. Before talking about Job Queue
8. Buffered vs Unbuffered http://bit.ly/understand-channel
9. Goroutine func main() { go func() { fmt.Println("GO GO GO") }() time.Sleep(1 * time.Second) }
10. Unbuffered make(chan bool)
11. Unbuffered Channel func main() { c := make(chan bool) go func() { fmt.Println("GO GO GO") c <- true }() <-c }
12. Unbuffered Channel func main() { c := make(chan bool) go func() { fmt.Println("GO GO GO") c <- true }() <-c }
13. Unbuffered Channel func main() { c := make(chan bool) go func() { fmt.Println("GO GO GO") <-c }() c <- true }
14. func main() { c := make(chan bool) go func() { fmt.Println("GO GO GO") c <- true c <- true }() <-c time.Sleep(1 * time.Second) } Unbuffered Channel
15. buffered make(chan bool, 1)
16. Buffered channel func main() { c := make(chan bool, 1) go func() { fmt.Println("GO GO GO") <-c }() c <- true }
17. Buffered channel func main() { c := make(chan bool, 1) go func() { fmt.Println("GO GO GO") <-c }() c <- true }
18. Buffered channel func main() { c := make(chan bool, 1) go func() { fmt.Println("GO GO GO") c <- true }() <-c }
19. How to implement Job Queue in Go
20. Sometimes you don’t need A job queue go process("job01")
21. func worker(jobChan <-chan Job) { for job := range jobChan { process(job) } } // make a channel with a capacity of 1024. jobChan := make(chan Job, 1024) // start the worker go worker(jobChan) // enqueue a job jobChan <- job
22. func worker(jobChan <-chan Job) { for job := range jobChan { process(job) } } // make a channel with a capacity of 1024. jobChan := make(chan Job, 1024) // start the worker go worker(jobChan) // enqueue a job jobChan <- job
23. Block if there already are 1024 jobs jobChan := make(chan Job, 1024)
24. Enqueue without blocking
25. func Enqueue(job Job, jobChan chan<- Job) bool { select { case jobChan <- job: return true default: return false } } if !Enqueue(job, job100) { Error( http.StatusServiceUnavailable, "max capacity reached", ) return }
26. Stopping the worker?
27. func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 }() for n := range ch { fmt.Println(n) } }
28. func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 close(ch) }() for n := range ch { fmt.Println(n) } }
29. func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 }() go func() { for n := range ch { fmt.Println(n) } }() time.Sleep(1 * time.Second) }
31. Setup Consumer
32. type Consumer struct { inputChan chan int jobsChan chan int } const PoolSize = 2 func main() { // create the consumer consumer := Consumer{ inputChan: make(chan int, 1), jobsChan: make(chan int, PoolSize), } }
34. func (c *Consumer) queue(input int) { fmt.Println("send input value:", input) c.jobsChan <- input } func (c *Consumer) worker(num int) { for job := range c.jobsChan { fmt.Println("worker:", num, " job value:", job) } } for i := 0; i < PoolSize; i++ { go consumer.worker(i) }
35. consumer.queue(1) consumer.queue(2) consumer.queue(3) consumer.queue(4) blocking by poll size = 2
36. Output (Poll Size = 2) send input value: 1 send input value: 2 send input value: 3 worker: 0 job value: 1 send input value: 4 worker: 0 job value: 2 worker: 0 job value: 3 worker: 0 job value: 4
37. rewrite queue func func (c *Consumer) queue(input int) bool { fmt.Println("send input value:", input) select { case c.jobsChan <- input: return true default: return false } }
38. Output (Poll Size = 2) send input value: 1 send input value: 2 send input value: 3 send input value: 4 worker: 0 job value: 1 worker: 0 job value: 2
39. Shutdown with Sigterm Handling
40. func WithContextFunc(ctx context.Context, f func()) context.Context { ctx, cancel := context.WithCancel(ctx) go func() { c := make(chan os.Signal) signal.Notify(c, syscall.SIGINT, syscall.SIGTERM) defer signal.Stop(c) select { case <-ctx.Done(): case <-c: f() cancel() } }() return ctx }
42. func WithContextFunc(ctx context.Context, f func()) context.Context { ctx, cancel := context.WithCancel(ctx) go func() { c := make(chan os.Signal) signal.Notify(c, syscall.SIGINT, syscall.SIGTERM) defer signal.Stop(c) select { case <-ctx.Done(): case <-c: f() cancel() } }() return ctx }
43. func (c Consumer) startConsumer(ctx context.Context) { for { select { case job := <-c.inputChan: if ctx.Err() != nil { close(c.jobsChan) return } c.jobsChan <- job case <-ctx.Done(): close(c.jobsChan) return } } }
44. Cancel by ctx.Done() event func (c *Consumer) worker(num int) { for job := range c.jobsChan { fmt.Println("worker:", num, " job value:", job) } }
45. Canceling Workers without Context
46. cancelChan := make(chan struct{}) go worker(jobChan, cancelChan) func worker(jobChan <-chan Job, cancelChan <-chan struct{}) { for { select { case <-cancelChan: return case job := <-jobChan: process(job) } } } // to cancel the worker, close the cancel channel close(cancelChan) Create a cancel channel
47. cancelChan := make(chan struct{}) go worker(jobChan, cancelChan) func worker(jobChan <-chan Job, cancelChan <-chan struct{}) { for { select { case <-cancelChan: return case job := <-jobChan: process(job) } } } // to cancel the worker, close the cancel channel close(cancelChan) Create a cancel channel close(cancelChan)
48. Graceful shutdown with worker sync.WaitGroup
50. wg := &sync.WaitGroup{} wg.Add(numberOfWorkers) // Start [PoolSize] workers for i := 0; i < PoolSize; i++ { go consumer.worker(i) }
51. WaitGroup WaitGroup WaitGroup WaitGroup
52. func (c Consumer) worker(wg *sync.WaitGroup) { defer wg.Done() for job := range c.jobsChan { // handle the job event } }
53. Add WaitGroup after Cancel Function
54. func WithContextFunc(ctx context.Context, f func()) context.Context { ctx, cancel := context.WithCancel(ctx) go func() { c := make(chan os.Signal) signal.Notify(c, syscall.SIGINT, syscall.SIGTERM) defer signal.Stop(c) select { case <-ctx.Done(): case <-c: cancel() f() } }() return ctx } Add WaitGroup after Cancel Function
55. wg := &sync.WaitGroup{} wg.Add(numberOfWorkers) ctx := signal.WithContextFunc( context.Background(), func() { wg.Wait() close(finishChan) }, ) go consumer.startConsumer(ctx)
56. End of Program select { case <-finished: case err := <-errChannel: if err != nil { return err } }
57. How to auto-scaling build agent?
59. Communicate between server and agent
61. Jobs Schema
62. r := e.Group("/rpc") r.Use(rpc.Check()) { r.POST("/v1/healthz", web.RPCHeartbeat) r.POST("/v1/request", web.RPCRquest) r.POST("/v1/accept", web.RPCAccept) r.POST("/v1/details", web.RPCDetails) r.POST("/v1/updateStatus", web.RPCUpdateStatus) r.POST("/v1/upload", web.RPCUploadBytes) r.POST("/v1/reset", web.RPCResetStatus) } Check RPC Secret
63. /rpc/v1/accept Update jobs set version = (oldVersion + 1) where machine = "fooBar" and version = oldVersion
64. Create multiple worker
65. if r.Capacity != 0 { var g errgroup.Group for i := 0; i < r.Capacity; i++ { g.Go(func() error { return r.start(ctx, 0) }) time.Sleep(1 * time.Second) } return g.Wait() }
66. Break for and select loop func (r *Runner) start(ctx context.Context, id int64) error { LOOP: for { select { case <-ctx.Done(): return ctx.Err() default: r.poll(ctx, id) if r.Capacity == 0 { break LOOP } } time.Sleep(1 * time.Second) } return nil }
67. How to cancel the current Job?
71. Context with Cancel or Timeout ctx, cancel := context.WithCancel(context.Background()) defer cancel() timeout, cancel := context.WithTimeout(ctx, 60*time.Minute) defer cancel() Job03 context
72. Context with Cancel or Timeout ctx, cancel := context.WithCancel(context.Background()) defer cancel() timeout, cancel := context.WithTimeout(ctx, 60*time.Minute) defer cancel() Job03 context Job05 context
73. Watch the Cancel event go func() { done, _ := r.Manager.Watch(ctx, id) if done { cancel() } }()
74. Handle cancel event on Server subscribers: make(map[chan struct{}]int64), cancelled: make(map[int64]time.Time),
75. User cancel running job c.Lock() c.cancelled[id] = time.Now().Add(time.Minute * 5) for subscriber, build := range c.subscribers { if id == build { close(subscriber) } } c.Unlock()
76. Agent subscribe the cancel event for { select { case <-ctx.Done(): return false, ctx.Err() case <-time.After(time.Minute): c.Lock() _, ok := c.cancelled[id] c.Unlock() if ok { return true, nil } case <-subscriber: return true, nil } }
77. case <-time.After(time.Minute): c.Lock() _, ok := c.cancelled[id] c.Unlock() if ok { return true, nil }
78. case <-time.After(time.Minute): c.Lock() _, ok := c.cancelled[id] c.Unlock() if ok { return true, nil } 1 Cancel
79. case <-time.After(time.Minute): c.Lock() _, ok := c.cancelled[id] c.Unlock() if ok { return true, nil } 1 2 Reconnect Server Cancel
80. https://www.udemy.com/course/golang-fight/?couponCode=GOLANG201911
81. https://www.udemy.com/course/devops-oneday/?couponCode=DEVOPS201911
82. Any Question?