Go concurrency pattern: Semaphore
Implementing concurrent processes in the Go programming language is very easy. Just run your function with go prefix, and boooom, your function will not blocking your main Goroutine.
But, what if your awesome concurrent function have responsibility with the machine’s I/O? For example, your function will do 100 parallel HTTP POST request to REST API from other services. Yeah, your network I/O will be busy handling 100 requests concurrently and this can slow down your I/O performance.
Here comes the rescue Semaphore, based on Wikipedia,
“Semaphore is a variable or abstract data type used to control access to a common resource by multiple processes in a concurrent system such as a multitasking operating system.”.
Based on our case, instead of doing 100 HTTP requests concurrently, we can scale down the concurrent process to 20 and we repeat the process 5 times. Yeah, it will take more time to process than doing 100 requests concurrently, but this will giving a breath for your I/O network since your network only needs to handle 20 requests concurrently.
That is the power of Semaphore, we can split 100 concurrent requests into 5 x 20 concurrent requests. Unfortunately, Golang doesn’t have built-in Semaphore implementation, but it can be emulated easily using buffered channel. Because when buffered channel is full, the channel will lock the Goroutine and make it wait until a buffer becomes available.
That is enough for a basic introduction, let’s get our hands dirty by implementing Semaphore in Go. First, we need to define our Semaphore interface:
type Semaphore interface { Acquire() Release()}
Our implementation has two methods Acquire() and Release(). The Acquire method is used to lock resources and will be called before calling our heavy / long-running process. And Release method should be called after the long-running process has been processed.
And then here comes our implementation:
type semaphore struct { semC chan struct{}}func New(maxConcurrency int) Semaphore { return &semaphore{ semC: make(chan struct{}, maxConcurrency), }}func (s *semaphore) Acquire() { s.semC <- struct{}{}}func (s *semaphore) Release() { <-s.semC}
That’s it, we leverage the blocking behavior on Go buffered channel when channel is full by setting maxConcurrency parameter as the channel size. When we calling Acquire(), the channel will be filled with an empty struct, this channel will be blocking if it reaches its maximum value. And when we calling Release(), we take out the empty struct from the channel, and the channel will be available for the next value and the channel will be unblocking.
let’s see our Semaphore implementation in action.
func main() { sem := semaphore.New(3) doneC := make(chan bool, 1) totProcess := 10 for i := 1; i <= totProcess; i++ { sem.Acquire() go func(v int) { defer sem.Release() longRunningProcess(v) if v == totProcess { doneC <- true } }(i) } <-doneC}func longRunningProcess(taskID int) { fmt.Println(
time.Now().Format(“15:04:05”),
“Running task with ID”,
taskID) time.Sleep(2 * time.Second)}
We instantiate a new Semaphore by the size 3, that means our maximum concurrent process will be limited to 3. And then we emulate some heavy tasks on longRunningProcess function (we use 2 seconds sleep time to block the process). Finally, we try to run longRunningProcess concurrently 10 Goroutines. Let’s see the output:
As you can see from the example above, our function can run 3 processes concurrently and since we using Semaphore, our example needs 6 seconds to complete the process, in contrast without Semaphore, the process only needs 2 seconds to be completed. By using Semaphore we can controls access to a shared resource eg: Database, Network, Disk, etc.
Reference:
