message: Use an inner struct for work queue data instead of atomics
diff: use std::{
  collections::VecDeque,
  - sync::{
  - atomic::{AtomicUsize, Ordering},
  - Arc, Mutex,
  - },
  + sync::{Arc, Mutex},
  };
  use futures::{stream, Stream};
  @@ -81,21 +78,36 @@ use tokio::sync::Notify;
  /// ```
  #[derive(Debug)]
  pub struct WorkQueue<T> {
  - /// Jobs currently in the queue.
  - jobs: Mutex<VecDeque<T>>,
  - /// Number of jobs that have been popped from the queue using `next_job` but not finished.
  - in_progress: Arc<AtomicUsize>,
  + /// Inner workings of the queue.
  + inner: Mutex<QueueInner<T>>,
  /// Notifier for waiting tasks.
  notify: Notify,
  }
  +/// Queue inner state.
  +#[derive(Debug)]
  +struct QueueInner<T> {
  + /// Jobs currently in the queue.
  + jobs: VecDeque<T>,
  + /// Number of jobs that have been popped from the queue using `next_job` but not finished.
  + in_progress: usize,
  +}
  +
  // Manual default implementation, since the derivation would require a `T: Default` trait bound.
  impl<T> Default for WorkQueue<T> {
  + fn default() -> Self {
  + Self {
  + inner: Default::default(),
  + notify: Default::default(),
  + }
  + }
  +}
  +
  +impl<T> Default for QueueInner<T> {
  fn default() -> Self {
  Self {
  jobs: Default::default(),
  in_progress: Default::default(),
  - notify: Default::default(),
  }
  }
  }
  @@ -114,11 +126,11 @@ impl<T> WorkQueue<T> {
  loop {
  let waiting;
  {
  - let mut jobs = self.jobs.lock().expect("lock poisoned");
  - match jobs.pop_front() {
  + let mut inner = self.inner.lock().expect("lock poisoned");
  + match inner.jobs.pop_front() {
  Some(job) => {
  // We got a job, increase the `in_progress` count and return.
  - self.in_progress.fetch_add(1, Ordering::SeqCst);
  + inner.in_progress += 1;
  return Some(JobHandle {
  job,
  queue: self.clone(),
  @@ -126,7 +138,7 @@ impl<T> WorkQueue<T> {
  }
  None => {
  // No job found. Check if we are completely done.
  - if self.in_progress.load(Ordering::SeqCst) == 0 {
  + if inner.in_progress == 0 {
  // No more jobs, no jobs in progress. We are done!
  return None;
  }
  @@ -151,9 +163,9 @@ impl<T> WorkQueue<T> {
  ///
  /// If there are any worker waiting on `next_job`, one of them will receive the job.
  pub fn push_job(&self, job: T) {
  - let mut guard = self.jobs.lock().expect("lock poisoned");
  + let mut inner = self.inner.lock().expect("lock poisoned");
  - guard.push_back(job);
  + inner.jobs.push_back(job);
  self.notify.notify_waiters();
  }
  @@ -171,13 +183,9 @@ impl<T> WorkQueue<T> {
  /// This is an internal function to be used by `JobHandle`, which locks the internal queue and
  /// decreases the in-progress count by one.
  fn complete_job(&self) {
  - // We need to lock the queue to prevent someone adding a job while we are notifying workers
  - // about the completion of what might appear to be the last job. This also prevents workers
  - // starving in the case of the last job being completed while they are checking for more
  - // work.
  - let _guard = self.jobs.lock().expect("lock poisoned");
  + let mut inner = self.inner.lock().expect("lock poisoned");
  - self.in_progress.fetch_sub(1, Ordering::SeqCst);
  + inner.in_progress -= 1;
  self.notify.notify_waiters();
  }
  }
  