Pop the next item using deficit round-robin across all virtual queues. Returns `None` if all virtual queues are empty. Each database is served for up to `priority_weight(class)` consecutive items before the scheduler rotates to the next database. Deficit credits are added once per turn (when a DB's deficit reaches zero and it re-enters the front of the rotation) and carried across calls so datab
(&mut self)
| 135 | /// the front of the rotation) and carried across calls so databases with |
| 136 | /// lower arrival rates still accumulate credits fairly. |
| 137 | pub fn pop_next(&mut self) -> Option<T> { |
| 138 | if self.total == 0 { |
| 139 | return None; |
| 140 | } |
| 141 | |
| 142 | // Walk the round-robin ring. We may need to skip empty queues, so we |
| 143 | // bound the scan to at most `n` DB rotations to avoid an infinite loop |
| 144 | // when all but one queue is empty. |
| 145 | let n = self.db_order.len(); |
| 146 | for _ in 0..n { |
| 147 | let db_id = match self.db_order.front().copied() { |
| 148 | Some(id) => id, |
| 149 | None => break, |
| 150 | }; |
| 151 | |
| 152 | let vq = match self.queues.get_mut(&db_id) { |
| 153 | Some(vq) => vq, |
| 154 | None => { |
| 155 | self.db_order.pop_front(); |
| 156 | continue; |
| 157 | } |
| 158 | }; |
| 159 | |
| 160 | // If this DB has no deficit remaining from its previous turn, grant |
| 161 | // a new quantum now (beginning of a new turn for this DB). |
| 162 | if vq.deficit == 0 { |
| 163 | let cls = self.priorities.get(&db_id).copied().unwrap_or_default(); |
| 164 | vq.deficit = priority_weight(cls); |
| 165 | } |
| 166 | |
| 167 | if let Some(item) = vq.items.pop_front() { |
| 168 | vq.deficit -= 1; |
| 169 | self.total -= 1; |
| 170 | |
| 171 | // If this DB's deficit is now exhausted, rotate it to the back |
| 172 | // so the next DB gets its turn. Otherwise leave it at the front |
| 173 | // so we keep draining it next call. |
| 174 | if vq.deficit == 0 { |
| 175 | self.db_order.pop_front(); |
| 176 | self.db_order.push_back(db_id); |
| 177 | } |
| 178 | |
| 179 | self.pops_since_reap += 1; |
| 180 | if self.pops_since_reap >= self.reap_after_pops { |
| 181 | self.reap_empty_queues(); |
| 182 | self.pops_since_reap = 0; |
| 183 | } |
| 184 | return Some(item); |
| 185 | } else { |
| 186 | // Queue drained; reset deficit so credits don't accumulate |
| 187 | // unboundedly for an inactive DB, then rotate to next. |
| 188 | vq.deficit = 0; |
| 189 | self.db_order.pop_front(); |
| 190 | self.db_order.push_back(db_id); |
| 191 | } |
| 192 | } |
| 193 | None |
| 194 | } |