@@ -60,8 +60,6 @@ func New[T comparable](name string, o ...Opt[T]) PriorityQueue[T] {
60
60
items : map [T ]* item [T ]{},
61
61
queue : btree .NewG (32 , less [T ]),
62
62
metrics : newQueueMetrics [T ](opts .MetricProvider , name , clock.RealClock {}),
63
- // Use a buffered channel to try and not block callers
64
- adds : make (chan * itemsWithOpts [T ], 1000 ),
65
63
// itemOrWaiterAdded indicates that an item or
66
64
// waiter was added. It must be buffered, because
67
65
// if we currently process items we can't tell
@@ -90,7 +88,6 @@ type priorityqueue[T comparable] struct {
90
88
items map [T ]* item [T ]
91
89
queue bTree [* item [T ]]
92
90
metrics queueMetrics [T ]
93
- adds chan * itemsWithOpts [T ]
94
91
95
92
// addedCounter is a counter of elements added, we need it
96
93
// because unixNano is not guaranteed to be unique.
@@ -120,67 +117,54 @@ type priorityqueue[T comparable] struct {
120
117
}
121
118
122
119
func (w * priorityqueue [T ]) AddWithOpts (o AddOpts , items ... T ) {
123
- w .adds <- & itemsWithOpts [T ]{opts : o , items : items }
124
- }
125
-
126
- func (w * priorityqueue [T ]) drainAddsLocked (added * itemsWithOpts [T ]) {
127
- for {
128
- for _ , key := range added .items {
129
- if added .opts .RateLimited {
130
- after := w .rateLimiter .When (key )
131
- if added .opts .After == 0 || after < added .opts .After {
132
- added .opts .After = after
133
- }
134
- }
135
-
136
- var readyAt * time.Time
137
- if added .opts .After > 0 {
138
- readyAt = ptr .To (w .now ().Add (added .opts .After ))
139
- w .metrics .retry ()
140
- }
141
- if _ , ok := w .items [key ]; ! ok {
142
- item := & item [T ]{
143
- key : key ,
144
- addedAtUnixNano : w .now ().UnixNano (),
145
- addedCounter : w .addedCounter ,
146
- priority : added .opts .Priority ,
147
- readyAt : readyAt ,
148
- }
149
- w .items [key ] = item
150
- w .queue .ReplaceOrInsert (item )
151
- w .metrics .add (key )
152
- w .addedCounter ++
153
- continue
154
- }
120
+ w .lock .Lock ()
121
+ defer w .lock .Unlock ()
155
122
156
- // The b-tree de-duplicates based on ordering and any change here
157
- // will affect the order - Just delete and re-add.
158
- item , _ := w .queue . Delete ( w . items [ key ] )
159
- if added . opts . Priority > item . priority {
160
- item . priority = added . opts . Priority
123
+ for _ , key := range items {
124
+ if o . RateLimited {
125
+ after := w .rateLimiter . When ( key )
126
+ if o . After == 0 || after < o . After {
127
+ o . After = after
161
128
}
129
+ }
162
130
163
- if item .readyAt != nil && (readyAt == nil || readyAt .Before (* item .readyAt )) {
164
- item .readyAt = readyAt
131
+ var readyAt * time.Time
132
+ if o .After > 0 {
133
+ readyAt = ptr .To (w .now ().Add (o .After ))
134
+ w .metrics .retry ()
135
+ }
136
+ if _ , ok := w .items [key ]; ! ok {
137
+ item := & item [T ]{
138
+ key : key ,
139
+ addedAtUnixNano : w .now ().UnixNano (),
140
+ addedCounter : w .addedCounter ,
141
+ priority : o .Priority ,
142
+ readyAt : readyAt ,
165
143
}
166
-
144
+ w . items [ key ] = item
167
145
w .queue .ReplaceOrInsert (item )
146
+ w .metrics .add (key )
147
+ w .addedCounter ++
148
+ continue
168
149
}
169
150
170
- // Drain the remainder of the channel. Has to be at the end,
171
- // because the first item is read in spin() and passed on
172
- // to us.
173
- select {
174
- case added = <- w .adds :
175
- default :
176
- return
151
+ // The b-tree de-duplicates based on ordering and any change here
152
+ // will affect the order - Just delete and re-add.
153
+ item , _ := w .queue .Delete (w .items [key ])
154
+ if o .Priority > item .priority {
155
+ item .priority = o .Priority
156
+ }
157
+
158
+ if item .readyAt != nil && (readyAt == nil || readyAt .Before (* item .readyAt )) {
159
+ item .readyAt = readyAt
177
160
}
161
+
162
+ w .queue .ReplaceOrInsert (item )
178
163
}
179
- }
180
164
181
- type itemsWithOpts [ T comparable ] struct {
182
- opts AddOpts
183
- items [] T
165
+ if len ( items ) > 0 {
166
+ w . notifyItemOrWaiterAdded ()
167
+ }
184
168
}
185
169
186
170
func (w * priorityqueue [T ]) notifyItemOrWaiterAdded () {
@@ -195,13 +179,11 @@ func (w *priorityqueue[T]) spin() {
195
179
var nextReady <- chan time.Time
196
180
nextReady = blockForever
197
181
198
- var addedItem * itemsWithOpts [T ]
199
182
for {
200
183
select {
201
184
case <- w .done :
202
185
return
203
186
case <- w .itemOrWaiterAdded :
204
- case addedItem = <- w .adds :
205
187
case <- nextReady :
206
188
}
207
189
@@ -211,11 +193,6 @@ func (w *priorityqueue[T]) spin() {
211
193
w .lock .Lock ()
212
194
defer w .lock .Unlock ()
213
195
214
- if addedItem != nil {
215
- w .drainAddsLocked (addedItem )
216
- }
217
- addedItem = nil
218
-
219
196
w .lockedLock .Lock ()
220
197
defer w .lockedLock .Unlock ()
221
198
0 commit comments