CODEWITHSUNDEEP
gothreadpoolchannelworker
nitimalh
nitimalh

Reputation: 961

Worker thread pool

In the example provided at http://marcio.io/2015/07/handling-1-million-requests-per-minute-with-golang/ which has been cited in a lot of places.

func (d *Dispatcher) dispatch() {
for {
    select {
    case job := <-JobQueue:
        // a job request has been received
        go func(job Job) {
            // try to obtain a worker job channel that is available.
            // this will block until a worker is idle
            jobChannel := <-d.WorkerPool

            // dispatch the job to the worker job channel
            jobChannel <- job
         }(job)
    }
}
}

Wouldn't the worker pool (chan chan job) get depleted after MaxWorker number of jobs have been serviced by the dispatch? Since <-d.WorkerPool is pulling from the channel and job channels are not being replenished after the first type dispatcher.Run() is invoked the first time? Or am I missing/misreading something ? How is the WorkerPool getting replenished with available job channels ?

go func(job Job) {
            // try to obtain a worker job channel that is available.
            // this will block until a worker is idle
            jobChannel := <-d.WorkerPool

            // dispatch the job to the worker job channel
            jobChannel <- job
        }(job)

Upvotes: 1

Views: 1676

Answers (1)

simon_xia
simon_xia

Reputation: 2544

if you read the code of worker carefully, you will notice

w.WorkerPool <- w.JobChannel

each time a loop begin, the channel of worker itself has been put back

I copy the whole function below:

func (w Worker) Start() {
    go func() {
        for {
            // register the current worker into the worker queue.
            w.WorkerPool <- w.JobChannel

            select {
            case job := <-w.JobChannel:
                // we have received a work request.
                if err := job.Payload.UploadToS3(); err != nil {
                    log.Errorf("Error uploading to S3: %s", err.Error())
                }

            case <-w.quit:
                // we have received a signal to stop
                return
            }
        }
    }()
}

Upvotes: 3

Related Questions