/// Pseudocodes
template <typename Thread>
class ThreadPoolImpl
{
public:
      using Job = std::function<void()>;
      ThreadPoolImpl(size_t max_threads_);
    void schedule(Job job);
      void wait();
      ~ThreadPoolImpl();
private:
      mutable std::mutex mutex;
      std::condition_variable job_finished;
      std::condition_variable new_job_or_shutdown;
  
      boost::heap::priority_queue<JobWithPriority> jobs;
      std::list<Thread> threads;
      
     void worker(typename std::list<Thread>::iterator thread_it);
};

/// Producer
template <typename Thread>
void ThreadPoolImpl<Thread>::schedule(Job job)
{
    {
        std::unique_lock lock(mutex);
      
          /// 等待空闲线程
          auto pred = [this] { return !queue_size || scheduled_jobs < queue_size || shutdown; };
          job_finished.wait(lock, pred);
      
          /// 不是在线程池构造是一口气创建完所有线程，而是 lazy 地创建线程
          if (threads.size() < std::min(max_threads, scheduled_jobs + 1))
        {
             try
            {
                threads.emplace_front();
            }
            catch (...)
            {
                /// 线程容器无法分配新的内存
            }
          
              /// 给创建出来的线程分配消费者
              try
            {
                threads.front() = Thread([this, it = threads.begin()] { worker(it); });
            }
            catch (...)
            {
                  /// 创建新的线程失败
                threads.pop_front();
            }
        }
          
          /// 将任务放入任务队列
          jobs.emplace(job);
    }
      /// 唤醒一个空闲的 worker 线程
    new_job_or_shutdown.notify_one();
}

/// Consumer
template <typename Thread>
void ThreadPoolImpl<Thread>::worker(typename std::list<Thread>::iterator thread_it)
{
      bool job_is_done = false;
      /// 线程消费者实现中常见的死循环
     while (true)
    {
          {
              std::unique_lock lock(mutex);
            /// 结束当前任务
            if (job_is_done)
            {
              job_is_done = false;
              --scheduled_jobs;
              job_finished.notify_all();
              if (shutdown)
                new_job_or_shutdown.notify_all(); /// `shutdown` was set, wake up other threads so they can finish themselves.
            }

            /// 等待新任务到来
            new_job_or_shutdown.wait(lock, [&] { return !jobs.empty() || shutdown || threads.size() > std::min(max_threads, scheduled_jobs + max_free_threads); });

            /// 如果没有设置 max_free_threads，则将线程释放掉
            if (jobs.empty() || threads.size() > std::min(max_threads, scheduled_jobs + max_free_threads))
            {
                // We enter here if:
                //  - either this thread is not needed anymore due to max_free_threads excess;
                //  - or shutdown happened AND all jobs are already handled.
                if (threads_remove_themselves)
                {
                    thread_it->detach();
                    threads.erase(thread_it);
                }
                return;
            }

            job_data = std::move(const_cast<JobWithPriority &>(jobs.top()));
            jobs.pop();
        }
      
          /// 执行任务
          /// 每一个任务独占一个线程，具体的 CPU 调度交给操作系统完成
          try
        {
              /// 运行任务
              job_data()->job();
              /// 任务执行完毕，销毁任务
              job_data.reset();
        }
          catch (...)
        {
              /// 任务执行异常
              job_data.reset();
        }
      
          job_is_done = true;
    }
}

/// 为了保证线程池析构时所有资源都得到了释放，需要实现 wait 方法等待所有线程执行完毕
template <typename Thread>
void ThreadPoolImpl<Thread>::wait()
{
    std::unique_lock lock(mutex);
    /// Signal here just in case.
    /// If threads are waiting on condition variables, but there are some jobs in the queue
    /// then it will prevent us from deadlock.
    new_job_or_shutdown.notify_all();
    job_finished.wait(lock, [this] { return scheduled_jobs == 0; });
}

template <bool propagate_opentelemetry_context = true>
class ThreadFromGlobalPoolImpl : boost::noncopyable
{
public:
    ThreadFromGlobalPoolImpl() = default;

    template <typename Function, typename... Args>
    explicit ThreadFromGlobalPoolImpl(Function && func, Args &&... args)
        : state(std::make_shared<State>())
    {
        /// NOTE:
        /// - If this will throw an exception, the destructor won't be called
        /// - this pointer cannot be passed in the lambda, since after detach() it will not be valid
        GlobalThreadPool::instance().scheduleOrThrow([
            my_state = state,
            my_func = std::forward<Function>(func),
            my_args = std::make_tuple(std::forward<Args>(args)...)]() mutable /// mutable is needed to destroy capture
        {
            SCOPE_EXIT(
                my_state->thread_id = std::thread::id();
                my_state->event.set(); /// 线程执行完毕，触发 event，方便 join 等方法判断线程是否结束。
            );

            my_state->thread_id = std::this_thread::get_id();

            /// This moves are needed to destroy function and arguments before exit.
            /// It will guarantee that after ThreadFromGlobalPool::join all captured params are destroyed.
            auto function = std::move(my_func);
            auto arguments = std::move(my_args);

            /// Thread status holds raw pointer on query context, thus it always must be destroyed
            /// before sending signal that permits to join this thread.
            DB::ThreadStatus thread_status;
            std::apply(function, arguments);
        },
        {}, // default priority
        0, // default wait_microseconds
        propagate_opentelemetry_context
        );
    }
  
      void join()
    {
        if (!initialized())
            abort();

        state->event.wait(); /// 等待上面的函数执行结束
        state.reset();
    }
protected:
    struct State
    {
        /// Should be atomic() because of possible concurrent access between
        /// assignment and joinable() check.
        std::atomic<std::thread::id> thread_id;

        /// The state used in this object and inside the thread job.
        Poco::Event event;
    };
      std::shared_ptr<State> state;
};

/**
 * We use **constinit** here to tell the compiler the current_thread variable is initialized.
 * If we didn't help the compiler, then it would most likely add a check before every use of the variable to initialize it if needed.
 * Instead it will trust that we are doing the right thing (and we do initialize it to nullptr) and emit more optimal code.
 * This is noticeable in functions like CurrentMemoryTracker::free and CurrentMemoryTracker::allocImpl
 * See also:
 * - https://en.cppreference.com/w/cpp/language/constinit
 * - https://github.com/ClickHouse/ClickHouse/pull/40078
 */
extern thread_local constinit ThreadStatus * current_thread;

class BackgroundSchedulePool
{
public:
    friend class BackgroundSchedulePoolTaskInfo;
    using TaskInfo = BackgroundSchedulePoolTaskInfo;
    using TaskInfoPtr = std::shared_ptr<TaskInfo>;
    using TaskFunc = std::function<void()>;
    using TaskHolder = BackgroundSchedulePoolTaskHolder;
    using DelayedTasks = std::multimap<Poco::Timestamp, TaskInfoPtr>;
    TaskHolder createTask(const std::string & log_name, const TaskFunc & function);
      /// ... others
private:
      using Threads = std::vector<ThreadFromGlobalPoolNoTracingContextPropagation>;
    void threadFunction();
    void delayExecutionThreadFunction();
    void scheduleTask(TaskInfoPtr task_info);
    void scheduleDelayedTask(const TaskInfoPtr & task_info, size_t ms, std::lock_guard<std::mutex> & task_schedule_mutex_lock);

    std::deque<TaskInfoPtr> tasks;
    Threads threads;
    std::unique_ptr<ThreadFromGlobalPoolNoTracingContextPropagation> delayed_thread;
    DelayedTasks delayed_tasks;

      /// ... others
};

void BackgroundSchedulePool::threadFunction()
{
    setThreadName(thread_name.c_str());

    while (!shutdown)
    {
        TaskInfoPtr task;

        {
            std::unique_lock<std::mutex> tasks_lock(tasks_mutex);

            tasks_cond_var.wait(tasks_lock, [&]()
            {
                return shutdown || !tasks.empty();
            });

            if (!tasks.empty())
            {
                task = tasks.front();
                tasks.pop_front();
            }
        }

        if (task)
            task->execute();
    }
}

void BackgroundSchedulePool::scheduleTask(TaskInfoPtr task_info)
{
    {
        std::lock_guard tasks_lock(tasks_mutex);
        tasks.push_back(std::move(task_info));
    }
    tasks_cond_var.notify_one();
}

void BackgroundSchedulePoolTaskInfo::scheduleImpl(std::lock_guard<std::mutex> & schedule_mutex_lock)
{
    scheduled = true;

    if (delayed)
        pool.cancelDelayedTask(shared_from_this(), schedule_mutex_lock);

    /// 如果任务正在被执行，但是被 schedule 了，它会在执行结束之后再一次被放入任务队列，
      /// 这个逻辑在 execute 函数中。
    if (!executing)
        pool.scheduleTask(shared_from_this());
}

bool BackgroundSchedulePoolTaskInfo::schedule()
{
    std::lock_guard lock(schedule_mutex);

    if (deactivated || scheduled)
        return false;

    scheduleImpl(lock);
    return true;
}

void BackgroundSchedulePoolTaskInfo::execute()
{
    std::lock_guard lock_exec(exec_mutex);

    {
        std::lock_guard lock_schedule(schedule_mutex);

        if (deactivated)
            return;

          /// 执行前设置为 false，以保证执行期间能够被 schedule。
        scheduled = false;
        executing = true;
    }

    try
    {
        function();
    }
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);
    }

    {
        std::lock_guard lock_schedule(schedule_mutex);

        executing = false;

          /// 在任务执行期间，任务被 schedule，将其重新放回任务队列。
          /// 不立刻重新执行的原因：保证公平，防止其他任务被饿死。
        if (scheduled)
            pool.scheduleTask(shared_from_this());
    }
}

class IExecutableTask
{
public:
    using TaskResultCallback = std::function<void(bool)>;
    virtual bool executeStep() = 0;
    virtual bool printExecutionException() const { return true; }
    virtual void onCompleted() = 0;
    virtual StorageID getStorageID() const = 0;
    virtual String getQueryId() const = 0;
    virtual Priority getPriority() const = 0;
    virtual ~IExecutableTask() = default;
};

bool BackgroundJobsAssignee::scheduleMergeMutateTask(ExecutableTaskPtr merge_task)
{
    bool res = getContext()->getMergeMutateExecutor()->trySchedule(merge_task);
    res ? trigger() : postpone();
    return res;
}

#define THREAD_NAME_SIZE 16
static thread_local char thread_name[THREAD_NAME_SIZE]{};
void setThreadName(const char * name, bool truncate)
{
    size_t name_len = strlen(name);
    if (!truncate && name_len > THREAD_NAME_SIZE - 1)
        throw DB::Exception(DB::ErrorCodes::PTHREAD_ERROR, "Thread name cannot be longer than 15 bytes");

    size_t name_capped_len = std::min<size_t>(1 + name_len, THREAD_NAME_SIZE - 1);
    char name_capped[THREAD_NAME_SIZE];
    memcpy(name_capped, name, name_capped_len);
    name_capped[name_capped_len] = '\0';

#if defined(OS_FREEBSD)
    pthread_set_name_np(pthread_self(), name_capped);
    if ((false))
#elif defined(OS_DARWIN)
    if (0 != pthread_setname_np(name_capped))
#elif defined(OS_SUNOS)
    if (0 != pthread_setname_np(pthread_self(), name_capped))
#else
    if (0 != prctl(PR_SET_NAME, name_capped, 0, 0, 0))
#endif
        if (errno != ENOSYS && errno != EPERM)    /// It's ok if the syscall is unsupported or not allowed in some environments.
            throw DB::ErrnoException(DB::ErrorCodes::PTHREAD_ERROR, "Cannot set thread name with prctl(PR_SET_NAME, ...)");

    memcpy(thread_name, name_capped, name_capped_len);
}