fixed size worker pool to serve new job requests #72

dtheodor · 2018-05-04T11:41:16Z

Based on https://gobyexample.com/worker-pools, with error and result handling improvements

X amount of goroutines listen to a queue for new job requests
HTTP new job requests place the work into the queue
the result is communicated back to the HTTP request goroutine through
a per-request unique result channel

worker pool
configurable settings
tests
on exit wait for jobs to finish

agis · 2018-05-04T11:51:27Z

cmd/mistryd/server.go

+type WorkResult struct {
+	BuildInfo *types.BuildInfo
+	Err       error
+}


Since BuildInfo already contains an .Err, can't we just use BuildInfo instead?

BuildInfo.Err contains build errors. The outer error is about failing to retrieve or construct the BuildInfo.

This struct captures the result of Work() (BuildInfo, error) which needs to be forwarded through a channel, so I'm capturing the two output parameters into a struct to pass it as one unit through channels

I see, fair enough.

agis · 2018-05-04T11:53:41Z

cmd/mistryd/server.go

+type WorkItem struct {
+	Job        *Job
+	JobRequest types.JobRequest
+	ResultChan chan WorkResult


ResultChan -> Result

agis · 2018-05-04T11:53:58Z

cmd/mistryd/server.go

+// WorkItem contains a job and a channel to place the job result
+type WorkItem struct {
+	Job        *Job
+	JobRequest types.JobRequest


*types.JobRequest

agis · 2018-05-04T11:55:24Z

cmd/mistryd/server.go

+	for i := 0; i < s.maxWorkers; i++ {
+		go s.worker(i)
+	}
+	s.Log.Printf("Setup %d workers", s.maxWorkers)


[nit] "Set up" or "Spawned"

agis · 2018-05-04T11:58:02Z

cmd/mistryd/server.go

@@ -87,6 +92,9 @@ func NewServer(cfg *Config, logger *log.Logger) (*Server, error) {
 	s.pq = NewProjectQueue()
 	s.br = broker.NewBroker(s.Log)
 	s.tq = new(sync.Map)
+	s.workQueueSize = 100
+	s.maxWorkers = 1


This should just be a cfg.GlobalBuildConcurrency or something.

agis · 2018-05-04T12:45:05Z

cmd/mistryd/server.go

+	s.Log.Printf("%s work items channel closed, returning...", logPrefix)
+}
+
+func (s *Server) writeJobResultResponse(j *Job, r WorkResult, w http.ResponseWriter) {


Are you sure this plays correct? I'd expect it to receive a pointer to the ResponseWriter, not a copy.

Same goes for WorkResult.

ResponseWriter is an interface. The WorkResult doesn't need to be mutated so we don't need to pass a pointer

j isn't mutated either.

agis · 2018-05-04T12:45:37Z

cmd/mistryd/server.go

+	s.Log.Printf("%s work items channel closed, returning...", logPrefix)
+}
+
+func (s *Server) writeJobResultResponse(j *Job, r WorkResult, w http.ResponseWriter) {


Perhaps writeWorkResult()?

agis · 2018-05-04T12:46:48Z

cmd/mistryd/server.go

 	go s.br.ListenForClients()
 	return s.srv.ListenAndServe()
 }

+func (s *Server) setupWorkerPool() {


This is too small and I think it should be inlined to ListenAndServe. We're not going to use it anywhere else for sure.

agis · 2018-05-04T12:47:56Z

cmd/mistryd/server.go

+	ResultChan chan WorkResult
+}
+
+func (s *Server) worker(id int) {


Needs a more accurate name (work?) and documentation. Also, do we need to have an id?

agis · 2018-05-04T12:48:32Z

cmd/mistryd/server.go

+}
+
+func (s *Server) worker(id int) {
+	logPrefix := fmt.Sprintf("Worker %d:", id)


We tend to prefix things like this:

[worker] foo

agis · 2018-05-07T08:41:26Z

cmd/mistryd/config.go

@@ -21,6 +22,9 @@ type Config struct {
 	ProjectsPath string            `json:"projects_path"`
 	BuildPath    string            `json:"build_path"`
 	Mounts       map[string]string `json:"mounts"`
+
+	PoolMaxWorkers   int `json:"pool_max_workers"`


Let's not expose implementation details to the user; we can use "Concurrency" or "BuildConcurrency" or "GlobalBuildConcurrency".

agis · 2018-05-07T08:44:34Z

cmd/mistryd/config.go

@@ -21,6 +22,9 @@ type Config struct {
 	ProjectsPath string            `json:"projects_path"`
 	BuildPath    string            `json:"build_path"`
 	Mounts       map[string]string `json:"mounts"`
+
+	PoolMaxWorkers   int `json:"pool_max_workers"`
+	PoolMaxQueueSize int `json:"pool_max_queue_size"`


I don't this configuration is need to be exposed to the user at all. We can use a reasonable fixed number for it; I believe 100 is a sane choice.

agis · 2018-05-07T08:45:27Z

cmd/mistryd/config.sample.json

-  }
+  },
+  "pool_max_workers": 5,
+  "pool_max_queue_size": 100


Please also update config.test.json accordingly.

agis · 2018-05-07T08:48:21Z

cmd/mistryd/server.go

+	jq         *JobQueue
+	pq         *ProjectQueue
+	cfg        *Config
+	WorkerPool *WorkerPool


Is there a reason this is exposed?

agis · 2018-05-07T08:51:17Z

cmd/mistryd/server.go

+		return
+	}
+
+	// if async, we're done, otherwise wait for the result in the result channel
 	if _, isAsync := r.URL.Query()["async"]; isAsync {


[out of scope] I think this should be renamed to "async". Also, we should be consistent in our style: the if ... should be on a new line.

agis · 2018-05-07T09:52:52Z

cmd/mistryd/worker_pool.go

+	return p
+}
+
+// Stop stops the pool workers and closes the queue


"closes the queue" is implementation detail. How about something like:

Stop signals the workers to close. Stop does not block until they are closed.

agis · 2018-05-07T10:10:23Z

cmd/mistryd/worker_pool.go

+	close(p.queue)
+}
+
+// SendWork sends work to the pool, and receives a work result that can be waited on.


SendWork schedules work on p and returns a FutureWorkResult. The actual result can be obtained by FutureWorkResult.Wait(). [...]

agis · 2018-05-07T10:11:42Z

cmd/mistryd/worker_pool.go

+
+// SendWork sends work to the pool, and receives a work result that can be waited on.
+// An error is returned if the backlog is full and cannot accept any new work items
+func (p *WorkerPool) SendWork(j *Job, jr types.JobRequest) (FutureWorkResult, error) {


Can we somehow make this agnostic to Job and types.JobRequest? I think we can. Couldn't this accept a WorkItem? This would seem more intuitive and better since we get rid of the tight coupling with the server.

agis · 2018-05-07T10:12:52Z

cmd/mistryd/worker_pool.go

+	case p.queue <- wi:
+		return result, nil
+	default:
+		return result, fmt.Errorf("queue is full")


No need to use Errorf since you don't format anything. Use errors.New. I think go vet should fail here anyway.

agis · 2018-05-07T10:13:38Z

cmd/mistryd/worker_pool.go

+
+// poolWorker listens to the workQueue, runs Work() on any incoming work items, and
+// sends the result through the result queue
+func poolWorker(s *Server, id int, workQueue <-chan workItem) {


Maybe just spawn or start is a better name?

Also, same goes here regarding to Server. I think this is unnecessary coupling that we could and should avoid.

"workQueue" could be renamed to the simpler "queue". The kind of thing that's queued in there is apparent from the type.

apostergiou · 2018-05-08T06:52:37Z

cmd/mistryd/testdata/projects/sleep/docker-entrypoint.sh

+#!/bin/bash
+set -e
+
+sleep 10


[nit] consider putting a newline at the end of the file

apostergiou · 2018-05-08T06:55:51Z

cmd/mistryd/worker_pool.go

+	p.wg.Wait()
+}
+
+// SendWork chedules work on p and returns a FutureWorkResult. The actual result can be


replace chedules with schedules 🐦

apostergiou · 2018-05-08T07:05:38Z

cmd/mistryd/worker_pool_test.go

+	cfg.Backlog = backlog
+
+	s, err := NewServer(cfg, nil)
+	failIfError(err, t)


👍

As a side note, I opened an issue #74 to package and refactor the test helpers.

apostergiou · 2018-05-08T07:06:06Z

LGTM 👍

* X amount of goroutines listen to a queue for new job requests * HTTP new job requests place the work into the queue * the result is communicated back to the HTTP request goroutine through a per-request unique result channel

dtheodor requested review from ctrochalakis, agis and apostergiou May 4, 2018 11:41

agis suggested changes May 4, 2018

View reviewed changes

dtheodor force-pushed the throttling branch 4 times, most recently from 38b925e to d8bb7f5 Compare May 4, 2018 14:29

agis suggested changes May 7, 2018

View reviewed changes

apostergiou approved these changes May 8, 2018

View reviewed changes

fixed size worker pool to serve new job requests

dc7ee83

* X amount of goroutines listen to a queue for new job requests * HTTP new job requests place the work into the queue * the result is communicated back to the HTTP request goroutine through a per-request unique result channel

dtheodor force-pushed the throttling branch from 4c1cfe9 to dc7ee83 Compare May 8, 2018 07:19

dtheodor merged commit 73c44ec into master May 8, 2018

dtheodor deleted the throttling branch May 8, 2018 07:25

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fixed size worker pool to serve new job requests #72

fixed size worker pool to serve new job requests #72

dtheodor commented May 4, 2018 •

edited

Loading

agis May 4, 2018

dtheodor May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

dtheodor May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 4, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

agis May 7, 2018

apostergiou May 8, 2018

apostergiou May 8, 2018

apostergiou May 8, 2018

apostergiou commented May 8, 2018

fixed size worker pool to serve new job requests #72

fixed size worker pool to serve new job requests #72

Conversation

dtheodor commented May 4, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apostergiou commented May 8, 2018

dtheodor commented May 4, 2018 •

edited

Loading