Method scheduleCacheTask

src/Storages/DiskCache/IDiskCache.cpp:209–245

Schedule cache task, when threadpool's current running task exceed certain ratio, start random drop disk cache task

// Schedule cache task, when threadpool's current running task exceed certain ratio, start random
// drop disk cache task
bool IDiskCache::scheduleCacheTask(const std::function<void()> & task)
{
    if (shutdown_called)
        return false;

    auto & thread_pool = IDiskCache::getThreadPool();
    size_t active_task_size = thread_pool.active();
    size_t max_queue_size = thread_pool.getMaxQueueSize();
    // (Running + Pending tasks) / (Max Running + Max Pending tasks)
    size_t current_ratio = max_queue_size == 0 ? 0 : ((active_task_size * 100) / max_queue_size);

    if (current_ratio <= settings.random_drop_threshold || settings.random_drop_threshold >= 100)
    {
        return thread_pool.trySchedule(task);
    }
    else
    {
        // Drop disk cache task base on queue's full ratio
        // (current task queue full ratio/ (100 - random_drop_threshold)) * 100
        // The drop possibility when current_ratio == random_drop_threshold is 0%
        // The drop possibility when current_ratio == 100 is 100%
        size_t drop_possibility = (100 * (current_ratio - settings.random_drop_threshold)) / (100 - settings.random_drop_threshold);
        std::random_device rd;
        std::mt19937 random_generator(rd());
        std::uniform_int_distribution<size_t> dist(1, 100);
        if (dist(random_generator) <= drop_possibility)
        {
            LOG_DEBUG(log, "Drop disk cache since queue is almost full, Queue length: {}, Max: {}, curren_ratio: {} ", active_task_size, max_queue_size, current_ratio);
            ProfileEvents::increment(ProfileEvents::DiskCacheTaskDropCount, 1, Metrics::MetricType::Meter);
            return false;
        }
        else
        {
            return thread_pool.trySchedule(task);
        }
    }
}

}