234
func (s *InterleavedWeightedRoundRobinScheduler[T, K]) getOrCreateTaskChannel(
235
channelKey K,
237
>
s.RLock()
238
>
channel, ok := s.weightedChannels[channelKey]
239
>
if ok {
240
channel.IncrementRefCount()
241
s.RUnlock()
242
return channel, channel.DecrementRefCount
243
}
245
>
246
>
s.Lock()
247
>
defer s.Unlock()
248
>
249
>
channel, ok = s.weightedChannels[channelKey]
250
>
if ok {
251
channel.IncrementRefCount()
252
return channel, channel.DecrementRefCount
253
}
254
256
>
channel = NewWeightedChannel[T](weight, WeightedChannelDefaultSize, s.ts.Now())
257
>
s.weightedChannels[channelKey] = channel
258
>
259
>
s.flattenWeightedChannelsLocked()
260
>
channel.IncrementRefCount()
261
>
return channel, channel.DecrementRefCount
262
}
263
264
>
func (s *InterleavedWeightedRoundRobinScheduler[T, K]) flattenWeightedChannelsLocked() {
interleaved_weighted_round_robin.go
265
>
weightedChannels := make(WeightedChannels[T], 0, len(s.weightedChannels))
266
>
for _, weightedChan := range s.weightedChannels {
267
>
weightedChannels = append(weightedChannels, weightedChan)
268
>
}
269
>
sort.Sort(weightedChannels)
270
>
271
>
iwrrChannels := make(WeightedChannels[T], 0, len(weightedChannels))
272
>
if len(weightedChannels) == 0 {
273
s.iwrrChannels.Store(iwrrChannels)
274
return
275
}
276
277
>
maxWeight := weightedChannels[len(weightedChannels)-1].Weight()
interleaved_weighted_round_robin.go
278
>
for round := maxWeight - 1; round > -1; round-- {
279
>
for index := len(weightedChannels) - 1; index > -1 && weightedChannels[index].Weight() > round; index-- {
280
>
iwrrChannels = append(iwrrChannels, weightedChannels[index])
281
>
}
282
}
284
}
285