Project Link: https://github.com/Joker666/goworkerpool In modern programming languages concurrency has become an unequivocal need. Almost all programming languages today have some method of concurrency. Some have really powerful constructs, that can offload works to different threads in the OS, e.g., Java, some mimic this behavior in the same thread, e.g., Ruby. Golang has a very powerful concurrency model called CSP (communicating sequential processes), which breaks a problem into smaller sequential processes and then schedules several instances of these processes called Goroutines. The communication between these processes happens by passing immutable messages via Channels. Here we will explore, how we can take advantage of concurrency in golang and how we can limit its usage with worker pools. In the second part, we will explore how to build a robust solution for concurrency around this concept. A Simple Case Let's imagine we have an external API call which takes about 100ms to get done. If we have 1000 of these calls, and we call these synchronously, it would take about to complete 100s model SimpleData {
	ID } basic ( ) {
	start := time.Now() i, _ := allData {
		Process(allData[i])
	}
	elapsed := time.Since(start)
	fmt.Printf( , elapsed)
} {
	fmt.Printf( , data.ID)
	time.Sleep( * time.Millisecond)
	fmt.Printf( , data.ID)
} main ( ) { allData []model.SimpleData i := ; i < ; i++ {
		data := model.SimpleData{ ID: i }
		allData = (allData, data)
	}
	fmt.Printf( ) basic.Work(allData)
} //// model/data.go package type struct int //// basic/basic.go package import "fmt" "github.com/Joker666/goworkerpool/model" "time" func Work (allData []model.SimpleData) for range "Took ===============> %s\n" func Process (data model.SimpleData) "Start processing %d\n" 100 "Finish processing %d\n" //// main.go package import "fmt" "github.com/Joker666/goworkerpool/basic" "github.com/Joker666/goworkerpool/model" "github.com/Joker666/goworkerpool/worker" func main () // Prepare the data var for 0 1000 append "Start processing all work \n" // Process Start processing all work 
Took ===============> 1m40.226679665s Here, we have a simple model that holds a data struct with just an integer value. We process an array of this data synchronously. This is obviously not an optimum solution since these tasks can be processed concurrently. Let's turn this into an asynchronous process with goroutines and channels. Asynchronous {
	start := time.Now() wg sync.WaitGroup

	dataCh := ( model.SimpleData, )

	wg.Add( ) { wg.Done() data := dataCh {
			wg.Add( ) { wg.Done()
				basic.Process(data)
			}(data)
		}
	}() i, _ := allData {
		dataCh <- allData[i]
	} (dataCh)
	wg.Wait()
	elapsed := time.Since(start)
	fmt.Printf( , elapsed)
} worker.NotPooledWork(allData) //// worker/notPooled.go func NotPooledWork (allData []model.SimpleData) var make chan 100 1 go func () defer for range 1 go func (data model.SimpleData) defer for range close "Took ===============> %s\n" //// main.go // Process Start processing all work 
Took ===============> 101.191534ms Here, we create a buffered channel of 100 and add all the data passed to to the channel. Since it is a buffered channel, it would not be able to enter any more data than 100 until data is extracted from it. Which we do it inside a goroutine. We range over the channel, extract data from it, add a goroutine, and process. There's no limit to how many goroutines we are creating here. This can process as many tasks as passed. Theoretically, it is possible to handle as much data as possible given the resources needed. If we run this, . Crazy right. Not quite. NoPooledWork we complete 1000 tasks around 100ms The Problem Unless we have all the resources on earth, we have limited allocations we can do for a given period of time. The minimal size of a goroutine object is 2K, but it can reach up to 1GB. The above solution runs all the tasks concurrently and let's say for a million of these tasks, it can quickly run out of memory and CPU of the machine. Either we have to upgrade our machine or find a different solution. Computer scientists have thought about this problem long before and came up with a brilliant solution called Thread Pool or Worker Pool. The idea is to have a limited pool of workers to process the tasks. Once a worker is done processing one task, it goes for handling the next task. So tasks keep waiting to be processed. This reduces the burst on CPU and memory and rather distributes the tasks over time. The Solution: Worker Pool Let's fix the earlier problem with worker pool implementation {
	start := time.Now() wg sync.WaitGroup
	workerPoolSize := dataCh := ( model.SimpleData, workerPoolSize) i := ; i < workerPoolSize; i++ {
		wg.Add( ) { wg.Done() data := dataCh {
				basic.Process(data)
			}
		}()
	} i, _ := allData {
		dataCh <- allData[i]
	} (dataCh)
	wg.Wait()
	elapsed := time.Since(start)
	fmt.Printf( , elapsed)
} worker.PooledWork(allData) //// worker/pooled.go func PooledWork (allData []model.SimpleData) var 100 make chan for 0 1 go func () defer for range for range close "Took ===============> %s\n" //// main.go // Process Start processing all work 
Took ===============> 1.002972449s Here, we have a limited number of worker size, 100. We spin up exactly 100 goroutines to process the tasks. We can think of channels as queues and each worker goroutine is a consumer. Multiple goroutines can listen to the same channel but each item on a channel would be processed just once. Channels work like queue in Golang This is a nice solution, if we run it, we see it takes to finish all the tasks. Not quite 100ms, but we don't need it. We get a much better solution that distributes the load over time. 1s Handling Error We are not quite done yet. This looks like a complete solution except it isn't. We are not handling errors here. So let's create a scenario where we have to handle errors and let's see how we can go about it. {
	start := time.Now() wg sync.WaitGroup
	workerPoolSize := dataCh := ( model.SimpleData, workerPoolSize)
	errors := ( error, ) i := ; i < workerPoolSize; i++ {
		wg.Add( ) { wg.Done() data := dataCh {
				process(data, errors)
			}
		}()
	} i, _ := allData {
		dataCh <- allData[i]
	} (dataCh)

	wg.Add( ) { wg.Done() { { err := <-errors:
				fmt.Println( , err.Error()) <-time.After(time.Second * ):
				fmt.Println( ) }
		}
	}() (errors)
	wg.Wait()
	elapsed := time.Since(start)
	fmt.Printf( , elapsed)
} {
	fmt.Printf( , data.ID)
	time.Sleep( * time.Millisecond) data.ID % == {
		errors <- fmt.Errorf( , data.ID)
	} {
		fmt.Printf( , data.ID)
	}
} worker.PooledWorkError(allData) //// worker/pooledError.go func PooledWorkError (allData []model.SimpleData) var 100 make chan make chan 1000 for 0 1 go func () defer for range for range close 1 go func () defer for select case "finished with error:" case 1 "Timeout: errors finished" return defer close "Took ===============> %s\n" func process (data model.SimpleData, errors <- error) chan "Start processing %d\n" 100 if 29 0 "error on job %v" else "Finish processing %d\n" //// main.go // Process We have modified our method to handle some randomized errors and add it to the passed channel. So, to handle errors in a concurrent model, we need channel to hold the error data. After all the tasks are done processing, we check the channel for errors. The error object holds ID of the task so that if needed, we can process those again. process errors errors errors This is a better solution than the one which was not handling errors at all. But we are not done with it yet, in the second part, we will go over how we can make a dedicated worker pool package that is robust and can handle concurrent tasks with a limiting worker pool number. Exercise We don't always want to keep processing our tasks if multiple tasks keep failing. Indicates that we have a deeper issue to resolve first. For exercise, you can try to improve the last solution to stop processing tasks, when there are more than 2 tasks that have resulted in an error. Send pull request in the original when you find a solution and I will send a review for that. repo Conclusion Golang's concurrency model is powerful enough to just build a worker pool solution without much overhead, that's why it's not included in the standard library. But we can spin up our own solution that fits our needs. I will write the next article very soon. Til then, stay tuned.

Different

Keep

Read My Stories

Too Long; Didn't Read

In your car, at home, or at work — Bosch technology shapes many areas of life.

Concurrency in Golang And WorkerPool [Part 1]

Too Long; Didn't Read

Companies Mentioned

MD Ahad Hasan

About Author

TOPICS

THIS ARTICLE WAS FEATURED IN...

Concurrency in Golang And WorkerPool [Part 1]

Too Long; Didn't Read

Companies Mentioned

MD Ahad Hasan

About Author

TOPICS

THIS ARTICLE WAS FEATURED IN...

RELATED STORIES