10076152260

Committed 24 Jul 2024 11:57AM UTC coverage: 58.456% (-0.03%) from 58.483%

Build # 10076152260

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #8923 from yyforyongyu/fix-test-interface

chainntnfs+lntest: fix `TestInterfaces`

Run Details

24 of 24 new or added lines in 2 files covered. (100.0%)

90 existing lines in 18 files now uncovered.

124010 of 212144 relevant lines covered (58.46%)

27644.9 hits per line

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

96.51

/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 {	46✔
63		46✔
64	q := &GCQueue{	46✔
65	takeBuffer: make(chan interface{}),	46✔
66	returnBuffer: make(chan interface{}, returnQueueSize),	46✔
67	expiryInterval: expiryInterval,	46✔
68	freeList: list.New(),	46✔
69	recycleTicker: ticker.New(gcInterval),	46✔
70	newItem: newItem,	46✔
71	quit: make(chan struct{}),	46✔
72	}	46✔
73		46✔
74	go q.queueManager()	46✔
75		46✔
76	return q	46✔
77	}	46✔
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{} {	68✔
82	select {	68✔
83	case item := <-q.takeBuffer:	68✔
84	return item	68✔
UNCOV 85	case <-time.After(time.Millisecond):	×
UNCOV 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{}) {	52✔
94	select {	52✔
95	case q.returnBuffer <- item:	52✔
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() {	46✔
115	for {	218✔
116	// If the pool is empty, initialize a buffer pool to serve a	172✔
117	// client that takes a buffer immediately. If this happens, this	172✔
118	// is either:	172✔
119	// 1) the first iteration of the loop,	172✔
120	// 2) after all entries were garbage collected, or	172✔
121	// 3) the freelist was emptied after the last entry was taken.	172✔
122	//	172✔
123	// In all of these cases, it is safe to pause the recycle ticker	172✔
124	// since it will be resumed as soon an entry is returned to the	172✔
125	// freelist.	172✔
126	if q.freeList.Len() == 0 {	286✔
127	q.freeList.PushBack(gcQueueEntry{	114✔
128	item: q.newItem(),	114✔
129	time: time.Now(),	114✔
130	})	114✔
131		114✔
132	q.recycleTicker.Pause()	114✔
133	}	114✔
134
135	next := q.freeList.Front()	172✔
136		172✔
137	select {	172✔
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:	68✔
142	q.freeList.Remove(next)	68✔
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:	52✔
148	// Add the returned buffer to the freelist, recording	52✔
149	// the current time so we can determine when the entry	52✔
150	// expires.	52✔
151	q.freeList.PushBack(gcQueueEntry{	52✔
152	item: item,	52✔
153	time: time.Now(),	52✔
154	})	52✔
155		52✔
156	// Adding the buffer implies that we now have a non-zero	52✔
157	// number of elements in the free list. Resume the	52✔
158	// recycle ticker to cleanup any entries that go unused.	52✔
159	q.recycleTicker.Resume()	52✔
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():	6✔
166	// Since the insert time of all entries will be	6✔
167	// monotonically increasing, iterate over elements and	6✔
168	// remove all entries that have expired.	6✔
169	var next *list.Element	6✔
170	for e := q.freeList.Front(); e != nil; e = next {	35✔
171	// Cache the next element, since it will become	29✔
172	// unreachable from the current element if it is	29✔
173	// removed.	29✔
174	next = e.Next()	29✔
175	entry := e.Value.(gcQueueEntry)	29✔
176		29✔
177	// Use now - insertTime <= expiryInterval to	29✔
178	// determine if this entry has not expired.	29✔
179	if time.Since(entry.time) <= q.expiryInterval {	31✔
180	// If this entry hasn't expired, then	2✔
181	// all entries that follow will still be	2✔
182	// valid.	2✔
183	break	2✔
184	}
185
186	// Otherwise, remove the expired entry from the
187	// linked-list.
188	q.freeList.Remove(e)	27✔
189	entry.item = nil	27✔
190	e.Value = nil	27✔
191	}
192	}
193	}
194	}

lightningnetwork / lnd / 10076152260

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