Concurrency are part of the Golang core. They are similar to light weight threads. We run a routine using the go keyword.
go matchRecordsWithDb(record)
Channels
Channels are a way to synchronize and communicate with go routines.
ch := make(chan string)
ch <- "test" //Send data to unbuffered channel
v := <-ch //receive data from channel and assign to v
The receive here will block till data is available on the channel.
Most programs use multiple go routines and buffered channels.
doneCh := make(chan bool, 4)
Here we will be able to run a routine 4 times without blocking.
select
select is like a switch with cases that allows you to wait on multiple communication operations. It is a way to receive channel data will block until one of its case is ready. Select with a default clause is a way to implement non-blocking sends, receives.
select {
case ac <- a:
// sent a on ac
case b:= <-bc:
// received b from bc
}
WorkerPools
I encountered the classic scenario where I had to make thousands of database calls to match records in a payment file. Finding viable matches in the database per line in the file, was slow and proving to much of a hassle. I wanted to add concurrency to my calls to achieve this faster. However, I was restricted by the database connection. I could only send a set number of queries at a time or it would error out.
I started with the naive approach. I create a buffered channel of 100 that went out and call the matching routine. The requirement was to match with a key in a table and return results. Some improvement. It did about 100 queries. Wait for those to finish and start the next batch of 100.
const workers = 100
jobsCh := make(chan int, workers)
for rowIdx := 0; rowIdx < len(fileRecords); rowIdx += workers {
for j = 0; j < workers; j++ {
if (rowIndex + j) >=len(fileRecords) {
break;
}
go matchRecordsWithDb(jobsCh, &fileRecords[rowIdx+j])
} // wait for the 100 workers to return
for i := 0; i < j; i++ {
fmt.Printf("%d", <-jobsCh)
}
}
There was a major snag in this approach. We had a condition if for some reason the line in the file didn’t have the main key, we had to query on another field. This field was not indexed and took a while to query. It is a very old legacy system so I can’t change the indexing at this point.
In my entire file I had one such record. The iteration of the 100 workers that had among it the routine to do this one query waited almost a minute and a half on that one query, while the 99 others finished. That is when I started looking at design patterns with channels and came across worker pools.
Worker pools is an approach to concurrency in which a fixed number of m workers have to do n number of tasks in a work queue. Rather than wait on all the workers (channels) at once, as the workers get idle they can be assigned jobs.
The three main players are :
Collector: Gathers all the jobs
AvailableWorkers Pool: Buffered channel of channels that is used to process the requests
Dispatcher: Pulls work requests off the Collector and sends them to available channels
All the jobs are add to a collector pool. The dispatcher picks jobs off the collector. If there are availableWorkers it gives them the job else it tries to createOne. If all m workers are busy doing jobs the dispatcher will wait on completion to assign the job.
After reading and experimenting with workerPools, I have written a workerPool package that can be used directly or as a guideline to implement your own.
