Example to explain golang simulation implementation of semaphore with timeout

This article mainly introduces to you the relevant information about golang simulation implementation of semaphore with timeout. The article introduces it in detail through the example code. It has certain reference learning value for everyone's study or work. Friends who need it Let’s learn with the editor below.

Preface

I am writing a project recently and need to use semaphores to wait for some resources to complete, but the maximum wait is N milliseconds. Before looking at the main text of this article, let's first look at the implementation method in C language.

In C language, there is the following API to implement semaphore waiting with timeout:

SYNOPSIS
  #include <pthread.h>
 
  int
  pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime);

Then check golang After reading the document, I found that semaphore with timeout is not implemented in golang. The official document is here.

Principle

My business scenario is this: I have a cache dictionary, when multiple users request a non-existent key At this time, only one request will penetrate to the backend, and all users will have to queue up and wait for this request to be completed, or return with a timeout.

How to achieve it? In fact, if you think about the principle of cond for a moment, you can simulate a cond with timeout.

In golang, to implement "suspend waiting" and "timeout return" at the same time, you generally need to use select case syntax. One case waits for blocked resources, and one case waits for a timer. This is very certain. .

Originally blocked resources should be notified of completion through the mechanism of condition variables. Since it is decided to use select case here, it is natural to think of using channel to replace this completion notification.

The next problem is that many requesters come to obtain this resource concurrently, but the resource is not ready yet, so everyone has to queue and hang up, waiting for the resource to be completed, and notify everyone when the resource is completed.

So, it is natural to create a queue for this resource. Each requester creates a chan, puts the chan in the queue, and then selects the case to wait for the notification of the chan. On the other end, after the resource is completed, it traverses the queue and notifies each chan.

The last problem is that only the first requester can penetrate the request to the backend, and subsequent requesters should not penetrate repeated requests. This can be determined by judging whether there is this key in the cache as the first time. condition, and flag bit init to determine whether the requester should queue.

My scenario

The above is the idea, and the following is the implementation of my business scenario.

func (cache *Cache) Get(key string, keyType int) *string {
 if keyType == KEY_TYPE_DOMAIN {
 key = "#" + key
 } else {
 key = "=" + key
 }
 
 cache.mutex.Lock()
 item, existed := cache.dict[key]
 if !existed {
 item = &cacheItem{}
 item.key = &key
 item.waitQueue = list.New()
 cache.dict[key] = item
 }
 cache.mutex.Unlock()
 
 conf := config.GetConfig()
 
 lastGet := getCurMs()
 
 item.mutex.Lock()
 item.lastGet = lastGet
 if item.init { // 已存在并且初始化
 defer item.mutex.Unlock()
 return item.value
 }
 
 // 未初始化，排队等待结果
 wait := waitItem{}
 wait.wait_chan = make(chan *string, 1)
 item.waitQueue.PushBack(&wait)
 item.mutex.Unlock()
 
 // 新增key, 启动goroutine获取初始值
 if !existed {
 go cache.initCacheItem(item, keyType)
 }
 
 timer := time.NewTimer(time.Duration(conf.Cache_waitTime) * time.Millisecond)
 
 var retval *string = nil
 
 // 等待初始化完成
 select {
 case retval = <- wait.wait_chan:
 case <- timer.C:
 }
 return retval
}

Briefly describe the whole process:

• First lock the dictionary. If the key does not exist, explain I am the first requester, and I will create the value corresponding to this key, but init=false means that it is being initialized. Finally, release the dictionary lock.

• Next, lock the key and judge that it has been initialized, then return the value directly. Otherwise, create a chan and put it into the waitQueue waiting queue. Finally, release the key lock.

• Next, if it is the first requester, it will penetrate the request to the backend (initiate a network call in an independent coroutine).

• Now, create a timer for timeout.

• Finally, regardless of whether it is the first requester of the key or a concurrent requester during initialization, they are all completed by waiting for the result of the select case timeout.

In the initCacheItem function, the data has been obtained successfully

 // 一旦标记为init, 后续请求将不再操作waitQueue
 item.mutex.Lock()
 item.value = newValue
 item.init = true
 item.expire = expire
 item.mutex.Unlock()
 
 // 唤醒所有排队者
 waitQueue := item.waitQueue
 for elem := waitQueue.Front(); elem != nil; elem = waitQueue.Front() {
 wait := elem.Value.(*waitItem)
 wait.wait_chan <- newValue
 waitQueue.Remove(elem)
 }

• First, lock the key and mark it init=true, assign value, and release the lock. Subsequent requests can be returned immediately without queuing.

• After that, because init=true has been marked, there are no requests to modify waitQueue at this moment, so there is no need to lock, traverse the queue directly, and notify each chan in it.

Finally

This achieves the condition variable effect with timeout. In fact, my scene is a broadcast Cond example, you can refer to the ideas to achieve the effect you want, learn and use it.

The above is the detailed content of Example to explain golang simulation implementation of semaphore with timeout. For more information, please follow other related articles on the PHP Chinese website!