2
\$\begingroup\$

I tried to use Go to write a DelayQueue. I referred to the implementation in Java. So the general logic is to have a minimum heap, then try to fetch the top element of the heap every time, if it is found to have expired, then fetch it, otherwise use cond#wait to suspend the current routine.

One of the biggest problems in the process is Cond does not support timeout waiting. For example, if the top element of the heap has 1 minute left to expire, I can just wait 1 minute. So I found an implementation that supports timeout.

As mentioned in this issue, it is not recommended to use Cond in Go, but channel instead. However, I do not know whether using channel is a better choice here.

I am new to Go, so there must be a lot of other problems with my code, and I hope to get your advice.

waitable_cond.go:

package main

import (
    "sync"
    "sync/atomic"
    "time"
    "unsafe"
)

// Code from : https://gist.github.com/zviadm/c234426882bfc8acba88f3503edaaa36#file-cond2-go

type Cond struct {
    L sync.Locker
    n unsafe.Pointer
}

func NewCond(l sync.Locker) *Cond {
    c := &Cond{L: l}
    n := make(chan struct{})
    c.n = unsafe.Pointer(&n)
    return c
}

// Wait Waits for Broadcast calls. Similar to regular sync.Cond, this unlocks the underlying
// locker first, waits on changes and re-locks it before returning.
func (c *Cond) Wait() {
    n := c.NotifyChan()
    c.L.Unlock()
    <-n
    c.L.Lock()
}

// WaitWithTimeout Same as Wait() call, but will only wait up to a given timeout.
func (c *Cond) WaitWithTimeout(t time.Duration) {
    n := c.NotifyChan()
    c.L.Unlock()
    select {
    case <-n:
    case <-time.After(t):
    }
    c.L.Lock()
}

// NotifyChan Returns a channel that can be used to wait for next Broadcast() call.
func (c *Cond) NotifyChan() <-chan struct{} {
    ptr := atomic.LoadPointer(&c.n)
    return *((*chan struct{})(ptr))
}

// Broadcast call notifies everyone that something has changed.
func (c *Cond) Broadcast() {
    n := make(chan struct{})
    ptrOld := atomic.SwapPointer(&c.n, unsafe.Pointer(&n))
    close(*(*chan struct{})(ptrOld))
}

priority_queue.go:

package main

import (
    "container/heap"
)

// Code from :https://pkg.go.dev/container/heap

// An Item is something we manage in a priority queue.
type Item struct {
    value    interface{} // The value of the item; arbitrary.
    priority int64       // The priority of the item in the queue.
    // The index is needed by update and is maintained by the heap.Interface methods.
    index int // The index of the item in the heap.
}

// A PriorityQueue implements heap.Interface and holds Items.
type PriorityQueue []*Item

const initCap = 12

func newPriorityQueue() PriorityQueue {
    return make(PriorityQueue, 0, initCap)
}

func (pq PriorityQueue) Len() int { return len(pq) }

func (pq PriorityQueue) Less(i, j int) bool {
    return pq[i].priority < pq[j].priority
}

func (pq PriorityQueue) Swap(i, j int) {
    pq[i], pq[j] = pq[j], pq[i]
    pq[i].index = i
    pq[j].index = j
}

func (pq *PriorityQueue) Push(x interface{}) {
    n := len(*pq)
    c := cap(*pq)
    if n+1 > c {
        npq := make(PriorityQueue, n, c*2)
        copy(npq, *pq)
        *pq = npq
    }
    item := x.(*Item)
    item.index = n
    *pq = append(*pq, item)
}

func (pq *PriorityQueue) Pop() interface{} {
    old := *pq
    n := len(old)
    item := old[n-1]
    old[n-1] = nil  // avoid memory leak
    item.index = -1 // for safety
    *pq = old[0 : n-1]
    return item
}

func (pq *PriorityQueue) Peek() interface{} {
    if pq.Len() == 0 {
        return nil
    }
    return (*pq)[0]
}

// update modifies the priority and value of an Item in the queue.
func (pq *PriorityQueue) update(item *Item, value string, priority int64) {
    item.value = value
    item.priority = priority
    heap.Fix(pq, item.index)
}

delay_queue.go:

package main

import (
    "container/heap"
    "sync"
    "time"
)

type DelayQueue struct {
    wakeup chan interface{}
    cond   *Cond
    pq     PriorityQueue
}

type DelayItem struct {
    deadline time.Time
    data     interface{}
}

// NewDelayQueue New creates an instance of delayQueue with the specified size.
func NewDelayQueue() *DelayQueue {
    return &DelayQueue{
        wakeup: make(chan interface{}),
        pq:     newPriorityQueue(),
        cond:   NewCond(&sync.Mutex{}),
    }
}

func (dq *DelayQueue) Put(data interface{}, delay time.Duration) {
    dq.cond.L.Lock()
    defer dq.cond.L.Unlock()
    item := &Item{
        value: DelayItem{
            time.Now().Add(delay),
            data,
        },
        priority: time.Now().Add(delay).Unix(),
    }
    heap.Push(&dq.pq, item)
    if dq.pq.Peek() == item {
        dq.cond.Broadcast()
    }
}

// Take This is block until get an item from Queue
func (dq *DelayQueue) Take() interface{} {
    dq.cond.L.Lock()
    defer dq.cond.L.Unlock()
    for {
        first := dq.pq.Peek()
        if first == nil {
            dq.cond.Wait()
        } else {
            deadline := first.(*Item).value.(DelayItem).deadline.Sub(time.Now())
            if first.(*Item).value.(DelayItem).deadline.Before(time.Now()) {
                return heap.Pop(&dq.pq).(*Item).value.(DelayItem).data
            }
            dq.cond.WaitWithTimeout(deadline)
        }
    }
}

\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

I'm going to suggest a different strategy for implementing this, one which relies more on language builtins. I think your queue can be simply a single channel, and to insert an element, you will just start a goroutine that delays for a certain amount of time and then sends the element on the channel.

To demonstrate with a little bit of code:

func Put(queue chan interface{}, data interface{}, delay time.Duration) {
    go func() {
        <-time.After(delay)
        queue <- data
    }()
}

func Take(queue chan interface{}) interface{} {
    // Just read from the queue
    return <-queue
}

I think this strategy will be vastly simpler, and utilize go's strengths of channels and goroutines. Of course, this simpler solution doesn't have exactly the same semantics as the original version, but depending on your use case it may be sufficient.

I also want to recommend reading through the presentation rethinking concurrency which lists many invaluable concurrency patterns for a go programmer.

\$\endgroup\$
3
  • \$\begingroup\$ Thanks for your answer,rose! In this way, if a large number of elements are put in and the timeout period is relatively long, it means that a large number of routines will be blocked. Although I know that routines are not a very expensive resource in Go, it seems that this is not a good choice in the face of large data? \$\endgroup\$
    – zysaaa
    Feb 22, 2022 at 6:53
  • \$\begingroup\$ In fact, I tried to use DelayQueue to implement a map that supports expiration, so it may put a lot of data. \$\endgroup\$
    – zysaaa
    Feb 22, 2022 at 6:58
  • \$\begingroup\$ If your timeouts are relatively long and you have a lot of data, then this implementation might be too expensive. \$\endgroup\$
    – rose
    Feb 22, 2022 at 13:31

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.