tokio_delayed_queue 0.2.0

////////////////////////////////////////////////////////////////////////////////

// Еще можно сделать просто свою футуру-обертку, которая будет
// в себя включать очередь + футуру задержки работы
//
// Есть уже кое-какая альтернатива:
// - https://crates.io/crates/futures-delay-queue
// - https://github.com/bells307/throttled-stream
// - https://crates.io/crates/stream_throttle
// - https://crates.io/crates/delay-queue
// - https://crates.io/crates/deadqueue

////////////////////////////////////////////////////////////////////////////////

use crate::{future::DelayedPopFuture, item::DelayItem};
use async_condvar_fair::{BatonExt, Condvar};
use parking_lot::Mutex;
use std::{
    collections::VecDeque,
    num::NonZeroUsize,
    sync::{
        atomic::{AtomicU64, Ordering},
        Arc,
    },
    time::{Duration, Instant},
};

////////////////////////////////////////////////////////////////////////////////

/// Структура данных, которую шарим между потоков
struct Inner<T> {
    /// Максимальный размер очереди
    max_size: Option<NonZeroUsize>,

    /// Очередь с синхронной блокировкой
    queue: Mutex<VecDeque<DelayItem<T>>>,

    /// Асинхронный condvar для оповещения об изменениях
    size_condvar: Condvar,

    /// Асинхронный condvar для оповещения об изменениях
    reserve_condvar: Condvar,

    /// Счетчик футур ожидания
    counter: AtomicU64,
}

////////////////////////////////////////////////////////////////////////////////

/// Delayed queue with atomic pop with cancellation support.
///
/// ```rust
/// # use tokio_delayed_queue::DelayedQueue;
/// # use std::{time::Duration, num::NonZeroUsize};
/// # tokio_test::block_on(async {
///
/// // New queue
/// let queue = DelayedQueue::new_bound(NonZeroUsize::new(16).unwrap());
///
/// // Push
/// queue.push(1, Duration::from_secs(1)).await;
///
/// // Pop
/// let v = queue.pop().await;
/// assert_eq!(v, 1);
///
/// # });
/// ```
//
// Используем `derive_where`,
// чтобы не накладывать дополнительные условия на тип `T`.
// #[derive_where(Clone)]
pub struct DelayedQueue<T> {
    inner: Arc<Inner<T>>,
}

impl<T> DelayedQueue<T> {
    /// Creates new queue with max size limitation.
    // Создание очереди c ограничением на размер.
    pub fn new_bound(size: NonZeroUsize) -> DelayedQueue<T> {
        DelayedQueue {
            inner: Arc::new(Inner {
                max_size: Some(size),
                queue: Mutex::new(VecDeque::new()),
                size_condvar: Condvar::new(),
                reserve_condvar: Condvar::new(),
                counter: AtomicU64::new(1),
            }),
        }
    }

    /// Creates new queue without max size limitation and capacity.
    // Создание очереди без какого-то ограничения в размерностях.
    pub fn new_unbound() -> DelayedQueue<T> {
        DelayedQueue {
            inner: Arc::new(Inner {
                max_size: None,
                queue: Mutex::new(VecDeque::new()),
                size_condvar: Condvar::new(),
                reserve_condvar: Condvar::new(),
                counter: AtomicU64::new(1),
            }),
        }
    }

    /// Push new item.
    // Добавляем новый итем с задержкой
    #[allow(clippy::await_holding_lock)]
    pub async fn push(&self, item: T, delay: Duration) {
        // Для удобства
        let this = self.inner.as_ref();

        // Когда будем пробуждаться
        let pop_time = Instant::now() + delay;

        // Пробуем получить блокировку над очередью
        let mut lock = loop {
            // Берем блокировку короткую над очередью
            let lock = this.queue.lock();

            // Надо ли нам как-то ограничивать размерность очереди?
            if let Some(max_size) = this.max_size {
                // Проверяем размер очереди, если превышен
                if lock.len() >= max_size.get() {
                    // Тогда подождем возможности запихнуть новый элемент
                    this.size_condvar.wait_no_relock(lock).await.dispose();
                } else {
                    // Все норм - отдаем дальше блокировку
                    break lock;
                }
            } else {
                // Все норм - отдаем дальше блокировку
                break lock;
            }
        };

        // Учитываем в том числе время нахождения в очереди,
        // поэтому создаем фиксированнную точку пробуждения
        // let pop_time = tokio::time::Instant::now() + delay;

        // Новый итем, сразу с футурой ожидания
        let queue_item = DelayItem {
            pop_time,
            item,
            reserved: Arc::new(AtomicU64::new(0)),
        };

        // Выполним увеличение размера буффера до нужного размера если надо
        if let Some(max_size) = this.max_size {
            // Следующее добавление должно будет привести к аллокации больше максимального размера?
            let will_allocate =
                ((lock.len() + 1) > lock.capacity()) && ((lock.len() + 1) >= (max_size.get() / 2));

            if will_allocate {
                // Может быть надо наоборот упаковать?
                let diff = (max_size.get() as i64) - (lock.capacity() as i64);
                if diff > 0 {
                    // Резервируем точно до максимального размера память
                    lock.reserve_exact(diff as usize);
                } else {
                    // Просто упакуем до максимального размера
                    lock.shrink_to(max_size.get());
                }
            }
        }

        // Добавляем итем
        lock.push_back(queue_item);

        // Снимаем блокировку
        drop(lock);

        // Теперь уведомляем, что итем стал доступен новый
        this.size_condvar.notify_one();
    }

    /// Atomically pop delayed item. It supports pop cancelation by returned future drop.
    // Получение нового итема с нужной задержкой
    pub fn pop(&self) -> DelayedPopFuture<T> {
        DelayedPopFuture {
            queue: &self.inner.queue,
            size_condvar: &self.inner.size_condvar,
            reserve_condvar: &self.inner.reserve_condvar,
            condvar_future: None,
            sleep_future: None,
            reserve_waker: None,
            future_id: self.inner.counter.fetch_add(1, Ordering::Release),
        }
    }
}

impl<T> Clone for DelayedQueue<T> {
    fn clone(&self) -> Self {
        DelayedQueue {
            inner: self.inner.clone(),
        }
    }
}