13481548301

Committed 23 Feb 2025 09:06AM UTC coverage: 4.031% (-54.8%) from 58.825%

Build # 13481548301

Build Type

Pull #9521

github

Committed by

web-flow

Commit Message

Merge 1ffbe99fe into 5fe900d18

Pull Request Pull Request #9521: unit: remove GOACC, use Go 1.20 native coverage functionality

Run Details

2852 of 70750 relevant lines covered (4.03%)

0.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/queue/gc_queue.go

package queue

import (
        "container/list"
        "time"

        "github.com/lightningnetwork/lnd/ticker"
)

// GCQueue is garbage collecting queue, which dynamically grows and contracts
// based on load. If the queue has items which have been returned, the queue
// will check every gcInterval amount of time to see if any elements are
// eligible to be released back to the runtime. Elements that have been in the
// queue for a duration of least expiryInterval will be released upon the next
// iteration of the garbage collection, thus the maximum amount of time an
// element remain in the queue is expiryInterval+gcInterval. The gc ticker will
// be disabled after all items in the queue have been taken or released to
// ensure that the GCQueue becomes quiescent, and imposes minimal overhead in
// the steady state.
type GCQueue struct {
        // takeBuffer coordinates the delivery of items taken from the queue
        // such that they are delivered to requesters.
        takeBuffer chan interface{}

        // returnBuffer coordinates the return of items back into the queue,
        // where they will be kept until retaken or released.
        returnBuffer chan interface{}

        // newItem is a constructor, used to generate new elements if none are
        // otherwise available for reuse.
        newItem func() interface{}

        // expiryInterval is the minimum amount of time an element will remain
        // in the queue before being released.
        expiryInterval time.Duration

        // recycleTicker is a resumable ticker used to trigger a sweep to
        // release elements that have been in the queue longer than
        // expiryInterval.
        recycleTicker ticker.Ticker

        // freeList maintains a list of gcQueueEntries, sorted in order of
        // increasing time of arrival.
        freeList *list.List

        quit chan struct{}
}

// NewGCQueue creates a new garbage collecting queue, which dynamically grows
// and contracts based on load. If the queue has items which have been returned,
// the queue will check every gcInterval amount of time to see if any elements
// are eligible to be released back to the runtime. Elements that have been in
// the queue for a duration of least expiryInterval will be released upon the
// next iteration of the garbage collection, thus the maximum amount of time an
// element remain in the queue is expiryInterval+gcInterval. The gc ticker will
// be disabled after all items in the queue have been taken or released to
// ensure that the GCQueue becomes quiescent, and imposes minimal overhead in
// the steady state. The returnQueueSize parameter is used to size the maximal
// number of items that can be returned without being dropped during large
// bursts in attempts to return items to the GCQUeue.
func NewGCQueue(newItem func() interface{}, returnQueueSize int,
        gcInterval, expiryInterval time.Duration) *GCQueue {

        q := &GCQueue{
                takeBuffer:     make(chan interface{}),
                returnBuffer:   make(chan interface{}, returnQueueSize),
                expiryInterval: expiryInterval,
                freeList:       list.New(),
                recycleTicker:  ticker.New(gcInterval),
                newItem:        newItem,
                quit:           make(chan struct{}),
        }

        go q.queueManager()

        return q
}

// Take returns either a recycled element from the queue, or creates a new item
// if none are available.
func (q *GCQueue) Take() interface{} {
        select {
        case item := <-q.takeBuffer:
                return item
        case <-time.After(time.Millisecond):
                return q.newItem()
        }
}

// Return adds the returned item to freelist if the queue's returnBuffer has
// available capacity. Under load, items may be dropped to ensure this method
// does not block.
func (q *GCQueue) Return(item interface{}) {
        select {
        case q.returnBuffer <- item:
        default:
        }
}

// gcQueueEntry is a tuple containing an interface{} and the time at which the
// item was added to the queue. The recorded time is used to determine when the
// entry becomes stale, and can be released if it has not already been taken.
type gcQueueEntry struct {
        item interface{}
        time time.Time
}

// queueManager maintains the free list of elements by popping the head of the
// queue when items are needed, and appending them to the end of the queue when
// items are returned. The queueManager will periodically attempt to release any
// items that have been in the queue longer than the expiry interval.
//
// NOTE: This method SHOULD be run as a goroutine.
func (q *GCQueue) queueManager() {
        for {
                // If the pool is empty, initialize a buffer pool to serve a
                // client that takes a buffer immediately. If this happens, this
                // is either:
                //   1) the first iteration of the loop,
                //   2) after all entries were garbage collected, or
                //   3) the freelist was emptied after the last entry was taken.
                //
                // In all of these cases, it is safe to pause the recycle ticker
                // since it will be resumed as soon an entry is returned to the
                // freelist.
                if q.freeList.Len() == 0 {
                        q.freeList.PushBack(gcQueueEntry{
                                item: q.newItem(),
                                time: time.Now(),
                        })

                        q.recycleTicker.Pause()
                }

                next := q.freeList.Front()

                select {

                // If a client requests a new write buffer, deliver the buffer
                // at the head of the freelist to them.
                case q.takeBuffer <- next.Value.(gcQueueEntry).item:
                        q.freeList.Remove(next)

                // If a client is returning a write buffer, add it to the free
                // list and resume the recycle ticker so that it can be cleared
                // if the entries are not quickly reused.
                case item := <-q.returnBuffer:
                        // Add the returned buffer to the freelist, recording
                        // the current time so we can determine when the entry
                        // expires.
                        q.freeList.PushBack(gcQueueEntry{
                                item: item,
                                time: time.Now(),
                        })

                        // Adding the buffer implies that we now have a non-zero
                        // number of elements in the free list. Resume the
                        // recycle ticker to cleanup any entries that go unused.
                        q.recycleTicker.Resume()

                // If the recycle ticker fires, we will aggressively release any
                // write buffers in the freelist for which the expiryInterval
                // has elapsed since their insertion. If after doing so, no
                // elements remain, we will pause the recycle ticker.
                case <-q.recycleTicker.Ticks():
                        // Since the insert time of all entries will be
                        // monotonically increasing, iterate over elements and
                        // remove all entries that have expired.
                        var next *list.Element
                        for e := q.freeList.Front(); e != nil; e = next {
                                // Cache the next element, since it will become
                                // unreachable from the current element if it is
                                // removed.
                                next = e.Next()
                                entry := e.Value.(gcQueueEntry)

                                // Use now - insertTime <= expiryInterval to
                                // determine if this entry has not expired.
                                if time.Since(entry.time) <= q.expiryInterval {
                                        // If this entry hasn't expired, then
                                        // all entries that follow will still be
                                        // valid.
                                        break
                                }

                                // Otherwise, remove the expired entry from the
                                // linked-list.
                                q.freeList.Remove(e)
                                entry.item = nil
                                e.Value = nil
                        }
                }
        }
}

1	package queue
2
3	import (
4	"container/list"
5	"time"
6
7	"github.com/lightningnetwork/lnd/ticker"
8	)
9
10	// GCQueue is garbage collecting queue, which dynamically grows and contracts
11	// based on load. If the queue has items which have been returned, the queue
12	// will check every gcInterval amount of time to see if any elements are
13	// eligible to be released back to the runtime. Elements that have been in the
14	// queue for a duration of least expiryInterval will be released upon the next
15	// iteration of the garbage collection, thus the maximum amount of time an
16	// element remain in the queue is expiryInterval+gcInterval. The gc ticker will
17	// be disabled after all items in the queue have been taken or released to
18	// ensure that the GCQueue becomes quiescent, and imposes minimal overhead in
19	// the steady state.
20	type GCQueue struct {
21	// takeBuffer coordinates the delivery of items taken from the queue
22	// such that they are delivered to requesters.
23	takeBuffer chan interface{}
24
25	// returnBuffer coordinates the return of items back into the queue,
26	// where they will be kept until retaken or released.
27	returnBuffer chan interface{}
28
29	// newItem is a constructor, used to generate new elements if none are
30	// otherwise available for reuse.
31	newItem func() interface{}
32
33	// expiryInterval is the minimum amount of time an element will remain
34	// in the queue before being released.
35	expiryInterval time.Duration
36
37	// recycleTicker is a resumable ticker used to trigger a sweep to
38	// release elements that have been in the queue longer than
39	// expiryInterval.
40	recycleTicker ticker.Ticker
41
42	// freeList maintains a list of gcQueueEntries, sorted in order of
43	// increasing time of arrival.
44	freeList *list.List
45
46	quit chan struct{}
47	}
48
49	// NewGCQueue creates a new garbage collecting queue, which dynamically grows
50	// and contracts based on load. If the queue has items which have been returned,
51	// the queue will check every gcInterval amount of time to see if any elements
52	// are eligible to be released back to the runtime. Elements that have been in
53	// the queue for a duration of least expiryInterval will be released upon the
54	// next iteration of the garbage collection, thus the maximum amount of time an
55	// element remain in the queue is expiryInterval+gcInterval. The gc ticker will
56	// be disabled after all items in the queue have been taken or released to
57	// ensure that the GCQueue becomes quiescent, and imposes minimal overhead in
58	// the steady state. The returnQueueSize parameter is used to size the maximal
59	// number of items that can be returned without being dropped during large
60	// bursts in attempts to return items to the GCQUeue.
61	func NewGCQueue(newItem func() interface{}, returnQueueSize int,
62	gcInterval, expiryInterval time.Duration) *GCQueue {	×
63		×
64	q := &GCQueue{	×
65	takeBuffer: make(chan interface{}),	×
66	returnBuffer: make(chan interface{}, returnQueueSize),	×
67	expiryInterval: expiryInterval,	×
68	freeList: list.New(),	×
69	recycleTicker: ticker.New(gcInterval),	×
70	newItem: newItem,	×
71	quit: make(chan struct{}),	×
72	}	×
73		×
74	go q.queueManager()	×
75		×
76	return q	×
77	}	×
78
79	// Take returns either a recycled element from the queue, or creates a new item
80	// if none are available.
81	func (q *GCQueue) Take() interface{} {	×
82	select {	×
83	case item := <-q.takeBuffer:	×
84	return item	×
85	case <-time.After(time.Millisecond):	×
86	return q.newItem()	×
87	}
88	}
89
90	// Return adds the returned item to freelist if the queue's returnBuffer has
91	// available capacity. Under load, items may be dropped to ensure this method
92	// does not block.
93	func (q *GCQueue) Return(item interface{}) {	×
94	select {	×
95	case q.returnBuffer <- item:	×
96	default:	×
97	}
98	}
99
100	// gcQueueEntry is a tuple containing an interface{} and the time at which the
101	// item was added to the queue. The recorded time is used to determine when the
102	// entry becomes stale, and can be released if it has not already been taken.
103	type gcQueueEntry struct {
104	item interface{}
105	time time.Time
106	}
107
108	// queueManager maintains the free list of elements by popping the head of the
109	// queue when items are needed, and appending them to the end of the queue when
110	// items are returned. The queueManager will periodically attempt to release any
111	// items that have been in the queue longer than the expiry interval.
112	//
113	// NOTE: This method SHOULD be run as a goroutine.
114	func (q *GCQueue) queueManager() {	×
115	for {	×
116	// If the pool is empty, initialize a buffer pool to serve a	×
117	// client that takes a buffer immediately. If this happens, this	×
118	// is either:	×
119	// 1) the first iteration of the loop,	×
120	// 2) after all entries were garbage collected, or	×
121	// 3) the freelist was emptied after the last entry was taken.	×
122	//	×
123	// In all of these cases, it is safe to pause the recycle ticker	×
124	// since it will be resumed as soon an entry is returned to the	×
125	// freelist.	×
126	if q.freeList.Len() == 0 {	×
127	q.freeList.PushBack(gcQueueEntry{	×
128	item: q.newItem(),	×
129	time: time.Now(),	×
130	})	×
131		×
132	q.recycleTicker.Pause()	×
133	}	×
134
135	next := q.freeList.Front()	×
136		×
137	select {	×
138
139	// If a client requests a new write buffer, deliver the buffer
140	// at the head of the freelist to them.
141	case q.takeBuffer <- next.Value.(gcQueueEntry).item:	×
142	q.freeList.Remove(next)	×
143
144	// If a client is returning a write buffer, add it to the free
145	// list and resume the recycle ticker so that it can be cleared
146	// if the entries are not quickly reused.
147	case item := <-q.returnBuffer:	×
148	// Add the returned buffer to the freelist, recording	×
149	// the current time so we can determine when the entry	×
150	// expires.	×
151	q.freeList.PushBack(gcQueueEntry{	×
152	item: item,	×
153	time: time.Now(),	×
154	})	×
155		×
156	// Adding the buffer implies that we now have a non-zero	×
157	// number of elements in the free list. Resume the	×
158	// recycle ticker to cleanup any entries that go unused.	×
159	q.recycleTicker.Resume()	×
160
161	// If the recycle ticker fires, we will aggressively release any
162	// write buffers in the freelist for which the expiryInterval
163	// has elapsed since their insertion. If after doing so, no
164	// elements remain, we will pause the recycle ticker.
165	case <-q.recycleTicker.Ticks():	×
166	// Since the insert time of all entries will be	×
167	// monotonically increasing, iterate over elements and	×
168	// remove all entries that have expired.	×
169	var next *list.Element	×
170	for e := q.freeList.Front(); e != nil; e = next {	×
171	// Cache the next element, since it will become	×
172	// unreachable from the current element if it is	×
173	// removed.	×
174	next = e.Next()	×
175	entry := e.Value.(gcQueueEntry)	×
176		×
177	// Use now - insertTime <= expiryInterval to	×
178	// determine if this entry has not expired.	×
179	if time.Since(entry.time) <= q.expiryInterval {	×
180	// If this entry hasn't expired, then	×
181	// all entries that follow will still be	×
182	// valid.	×
183	break	×
184	}
185
186	// Otherwise, remove the expired entry from the
187	// linked-list.
188	q.freeList.Remove(e)	×
189	entry.item = nil	×
190	e.Value = nil	×
191	}
192	}
193	}
194	}

lightningnetwork / lnd / 13481548301

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous