設計任務排程器 — Priority Queue、Time Wheel 與 Cron 排程實戰 | 資料結構與演算法

// ============================================================
// Task Scheduler — TypeScript 完整實作
// 包含：Priority Queue + Timer Wheel + DAG 依賴 + Cron 解析器
// ============================================================

// ─── 型別定義 ───────────────────────────────────────────────

type TaskStatus = "pending" | "ready" | "running" | "success" | "failed" | "cancelled";
type TaskType = "immediate" | "delayed" | "recurring";
type TaskPriority = 1 | 2 | 3 | 4 | 5; // 1=最高, 5=最低

interface Task {
  id: string;
  name: string;
  type: TaskType;
  priority: TaskPriority;
  executeAt: number;          // Unix timestamp（ms）
  cronExpression?: string;    // 週期任務的 Cron 表達式
  dependencies: string[];     // 依賴的任務 ID 列表
  payload: Record<string, unknown>;
  status: TaskStatus;
  retryCount: number;
  maxRetries: number;
  createdAt: number;
  execute: () => Promise<void>;
}

// ─── Min-Heap Priority Queue ────────────────────────────────

class TaskPriorityQueue {
  private heap: Task[] = [];

  /**
   * 比較函式：先按 executeAt 排序，再按 priority（數值小 = 優先級高）
   */
  private compare(a: Task, b: Task): boolean {
    if (a.executeAt !== b.executeAt) return a.executeAt < b.executeAt;
    return a.priority < b.priority;
  }

  push(task: Task): void {
    this.heap.push(task);
    this.bubbleUp(this.heap.length - 1);
  }

  pop(): Task | undefined {
    if (this.heap.length === 0) return undefined;
    const top = this.heap[0];
    const last = this.heap.pop()!;
    if (this.heap.length > 0) {
      this.heap[0] = last;
      this.sinkDown(0);
    }
    return top;
  }

  peek(): Task | undefined {
    return this.heap[0];
  }

  get size(): number {
    return this.heap.length;
  }

  removeById(taskId: string): boolean {
    const idx = this.heap.findIndex(t => t.id === taskId);
    if (idx === -1) return false;
    // 標記刪除（lazy deletion）— 取出時再跳過
    this.heap[idx].status = "cancelled";
    return true;
  }

  private bubbleUp(i: number): void {
    while (i > 0) {
      const parent = Math.floor((i - 1) / 2);
      if (this.compare(this.heap[i], this.heap[parent])) {
        [this.heap[i], this.heap[parent]] = [this.heap[parent], this.heap[i]];
        i = parent;
      } else break;
    }
  }

  private sinkDown(i: number): void {
    const n = this.heap.length;
    while (true) {
      let smallest = i;
      const left = 2 * i + 1;
      const right = 2 * i + 2;
      if (left < n && this.compare(this.heap[left], this.heap[smallest])) smallest = left;
      if (right < n && this.compare(this.heap[right], this.heap[smallest])) smallest = right;
      if (smallest !== i) {
        [this.heap[i], this.heap[smallest]] = [this.heap[smallest], this.heap[i]];
        i = smallest;
      } else break;
    }
  }
}

// ─── Cron 解析器 ─────────────────────────────────────────────

class CronParser {
  /**
   * 解析 Cron 表達式，計算下次執行時間
   * 支援標準 5 欄 Cron：分 時 日 月 週
   * 支援特殊值：* , - /
   */
  static nextExecutionTime(expression: string, fromTime: Date = new Date()): Date {
    const fields = expression.trim().split(/\s+/);
    if (fields.length !== 5) {
      throw new Error(`無效的 Cron 表達式：${expression}（需要 5 個欄位）`);
    }

    const [minExpr, hourExpr, domExpr, monthExpr, dowExpr] = fields;

    const parsedMin = this.parseField(minExpr, 0, 59);
    const parsedHour = this.parseField(hourExpr, 0, 23);
    const parsedDom = this.parseField(domExpr, 1, 31);
    const parsedMonth = this.parseField(monthExpr, 1, 12);
    const parsedDow = this.parseField(dowExpr, 0, 6); // 0=週日, 6=週六

    // 從 fromTime + 1 分鐘開始搜尋
    const candidate = new Date(fromTime);
    candidate.setSeconds(0, 0);
    candidate.setMinutes(candidate.getMinutes() + 1);

    // 最多搜尋 4 年（防止無限迴圈）
    const maxDate = new Date(fromTime.getTime() + 4 * 365 * 24 * 60 * 60 * 1000);

    while (candidate < maxDate) {
      const month = candidate.getMonth() + 1; // 1-12
      const dom = candidate.getDate();
      const dow = candidate.getDay(); // 0-6
      const hour = candidate.getHours();
      const min = candidate.getMinutes();

      if (!parsedMonth.has(month)) {
        candidate.setMonth(candidate.getMonth() + 1, 1);
        candidate.setHours(0, 0, 0, 0);
        continue;
      }
      if (!parsedDom.has(dom) || !parsedDow.has(dow)) {
        candidate.setDate(candidate.getDate() + 1);
        candidate.setHours(0, 0, 0, 0);
        continue;
      }
      if (!parsedHour.has(hour)) {
        candidate.setHours(candidate.getHours() + 1, 0, 0, 0);
        continue;
      }
      if (!parsedMin.has(min)) {
        candidate.setMinutes(candidate.getMinutes() + 1, 0, 0);
        continue;
      }

      return new Date(candidate);
    }

    throw new Error(`無法在 4 年內找到下次執行時間：${expression}`);
  }

  private static parseField(expr: string, min: number, max: number): Set<number> {
    const result = new Set<number>();

    // 星期別名
    const dowAlias: Record<string, number> = {
      SUN: 0, MON: 1, TUE: 2, WED: 3, THU: 4, FRI: 5, SAT: 6,
    };

    const resolveAlias = (s: string): number => {
      const upper = s.toUpperCase();
      return dowAlias[upper] ?? parseInt(s, 10);
    };

    for (const part of expr.split(",")) {
      if (part === "*") {
        for (let i = min; i <= max; i++) result.add(i);
      } else if (part.includes("/")) {
        const [range, step] = part.split("/");
        const stepNum = parseInt(step, 10);
        const start = range === "*" ? min : resolveAlias(range.split("-")[0]);
        const end = range === "*"
          ? max
          : range.includes("-") ? resolveAlias(range.split("-")[1]) : max;
        for (let i = start; i <= end; i += stepNum) result.add(i);
      } else if (part.includes("-")) {
        const [start, end] = part.split("-").map(resolveAlias);
        for (let i = start; i <= end; i++) result.add(i);
      } else {
        result.add(resolveAlias(part));
      }
    }

    return result;
  }
}

// ─── Timer Wheel 實作 ──────────────────────────────────────

class TimerWheel {
  private slots: Map<number, Task[]>;
  private currentSlot: number = 0;
  private readonly slotCount: number;
  private readonly tickMs: number;
  private intervalHandle?: ReturnType<typeof setInterval>;

  constructor(slotCount: number = 3600, tickMs: number = 10) {
    this.slotCount = slotCount;
    this.tickMs = tickMs;
    this.slots = new Map();
  }

  /** 插入延遲任務 — 時間複雜度：O(1) */
  schedule(task: Task): void {
    const delayMs = Math.max(0, task.executeAt - Date.now());
    const ticks = Math.floor(delayMs / this.tickMs);
    const rounds = Math.floor(ticks / this.slotCount);
    const slot = (this.currentSlot + ticks) % this.slotCount;

    // 記錄剩餘圈數
    (task as any).__timerRounds = rounds;

    if (!this.slots.has(slot)) this.slots.set(slot, []);
    this.slots.get(slot)!.push(task);
  }

  /** 啟動 Timer Wheel */
  start(onTaskReady: (task: Task) => void): void {
    this.intervalHandle = setInterval(() => {
      const slotTasks = this.slots.get(this.currentSlot) ?? [];
      const remaining: Task[] = [];

      for (const task of slotTasks) {
        if (task.status === "cancelled") continue;
        const rounds = (task as any).__timerRounds;
        if (rounds === 0) {
          onTaskReady(task);
        } else {
          (task as any).__timerRounds = rounds - 1;
          remaining.push(task);
        }
      }

      if (remaining.length > 0) {
        this.slots.set(this.currentSlot, remaining);
      } else {
        this.slots.delete(this.currentSlot);
      }

      this.currentSlot = (this.currentSlot + 1) % this.slotCount;
    }, this.tickMs);
  }

  stop(): void {
    if (this.intervalHandle) clearInterval(this.intervalHandle);
  }
}

// ─── DAG Engine — Kahn's 拓撲排序 ───────────────────────────

interface WorkflowResult {
  order: string[];
  parallelGroups: string[][];
  hasCycle: boolean;
}

class DAGEngine {
  /**
   * Kahn's Algorithm — BFS 拓撲排序
   * 同時計算可並行執行的任務群組
   * 時間複雜度：O(V + E)
   */
  static topoSort(tasks: Map<string, Task>): WorkflowResult {
    const inDegree = new Map<string, number>();
    const adjList = new Map<string, string[]>();

    for (const [id] of tasks) {
      inDegree.set(id, 0);
      adjList.set(id, []);
    }

    for (const [id, task] of tasks) {
      for (const dep of task.dependencies) {
        if (!tasks.has(dep)) {
          throw new Error(`任務 ${id} 依賴不存在的任務 ${dep}`);
        }
        inDegree.set(id, (inDegree.get(id) ?? 0) + 1);
        adjList.get(dep)!.push(id);
      }
    }

    const order: string[] = [];
    const parallelGroups: string[][] = [];
    let queue = [...inDegree.entries()]
      .filter(([, deg]) => deg === 0)
      .map(([id]) => id);

    while (queue.length > 0) {
      parallelGroups.push([...queue]);
      const nextQueue: string[] = [];

      for (const id of queue) {
        order.push(id);
        for (const neighbor of adjList.get(id) ?? []) {
          const newDeg = (inDegree.get(neighbor) ?? 1) - 1;
          inDegree.set(neighbor, newDeg);
          if (newDeg === 0) nextQueue.push(neighbor);
        }
      }

      queue = nextQueue;
    }

    const hasCycle = order.length !== tasks.size;
    return { order, parallelGroups, hasCycle };
  }
}

// ─── TaskScheduler（主排程器）──────────────────────────────────

class TaskScheduler {
  private priorityQueue: TaskPriorityQueue;
  private timerWheel: TimerWheel;
  private allTasks: Map<string, Task> = new Map();
  private completedTasks: Set<string> = new Set();
  private runningTasks: Set<string> = new Set();
  private readonly maxConcurrency: number;
  private isRunning: boolean = false;
  private schedulerInterval?: ReturnType<typeof setInterval>;

  constructor(maxConcurrency: number = 10) {
    this.maxConcurrency = maxConcurrency;
    this.priorityQueue = new TaskPriorityQueue();
    this.timerWheel = new TimerWheel(360000, 1); // 100 小時，1ms 精度
  }

  /** 提交即時/延遲任務 */
  submit(task: Omit<Task, "status" | "retryCount" | "createdAt">): string {
    const fullTask: Task = {
      ...task,
      status: "pending",
      retryCount: 0,
      createdAt: Date.now(),
    };
    this.allTasks.set(task.id, fullTask);

    if (task.dependencies.length === 0) {
      this.enqueueTask(fullTask);
    }

    console.log(`[Scheduler] 任務已提交：${task.id} (${task.type}, priority=${task.priority})`);
    return task.id;
  }

  /** 提交 DAG 工作流 */
  submitWorkflow(tasks: Omit<Task, "status" | "retryCount" | "createdAt">[]): WorkflowResult {
    const taskMap = new Map<string, Task>();
    for (const t of tasks) {
      const fullTask: Task = {
        ...t,
        status: "pending",
        retryCount: 0,
        createdAt: Date.now(),
      };
      this.allTasks.set(t.id, fullTask);
      taskMap.set(t.id, fullTask);
    }

    const result = DAGEngine.topoSort(taskMap);
    if (result.hasCycle) {
      throw new Error("工作流包含循環依賴，無法執行");
    }

    console.log("[Scheduler] 工作流拓撲排序結果：");
    result.parallelGroups.forEach((group, i) => {
      console.log(`  第 ${i + 1} 波（可並行）：${group.join(", ")}`);
    });

    // 入度為 0 的任務加入排程
    for (const id of (result.parallelGroups[0] ?? [])) {
      this.enqueueTask(taskMap.get(id)!);
    }

    return result;
  }

  /** 提交週期任務（Cron） */
  submitCron(
    taskTemplate: Omit<Task, "id" | "status" | "retryCount" | "createdAt" | "executeAt" | "type">,
    cronExpression: string
  ): void {
    const scheduleNext = () => {
      const nextTime = CronParser.nextExecutionTime(cronExpression);
      const taskId = `${taskTemplate.name}_${nextTime.getTime()}`;

      const cronTask: Task = {
        ...taskTemplate,
        id: taskId,
        type: "recurring",
        status: "pending",
        retryCount: 0,
        createdAt: Date.now(),
        executeAt: nextTime.getTime(),
        cronExpression,
        execute: async () => {
          await taskTemplate.execute();
          scheduleNext(); // 執行完畢後安排下次
        },
      };

      this.allTasks.set(taskId, cronTask);
      this.timerWheel.schedule(cronTask);
      console.log(`[Scheduler] Cron 任務排程：${taskTemplate.name}，下次：${nextTime.toISOString()}`);
    };

    scheduleNext();
  }

  /** 取消任務 */
  cancel(taskId: string): boolean {
    const task = this.allTasks.get(taskId);
    if (!task || task.status === "running") return false;
    task.status = "cancelled";
    this.priorityQueue.removeById(taskId);
    console.log(`[Scheduler] 任務已取消：${taskId}`);
    return true;
  }

  /** 查詢任務狀態 */
  getStatus(taskId: string): TaskStatus | null {
    return this.allTasks.get(taskId)?.status ?? null;
  }

  /** 啟動排程器主迴圈 */
  start(): void {
    if (this.isRunning) return;
    this.isRunning = true;

    this.timerWheel.start((task) => {
      if (task.status === "pending") this.enqueueTask(task);
    });

    this.schedulerInterval = setInterval(() => this.processQueue(), 10);
    console.log("[Scheduler] 排程器啟動");
  }

  stop(): void {
    this.isRunning = false;
    this.timerWheel.stop();
    if (this.schedulerInterval) clearInterval(this.schedulerInterval);
    console.log("[Scheduler] 排程器停止");
  }

  private enqueueTask(task: Task): void {
    if (task.type === "delayed" && task.executeAt > Date.now() + 10) {
      this.timerWheel.schedule(task);
    } else {
      task.status = "ready";
      this.priorityQueue.push(task);
    }
  }

  private processQueue(): void {
    while (
      this.runningTasks.size < this.maxConcurrency &&
      this.priorityQueue.size > 0
    ) {
      const task = this.priorityQueue.peek();
      if (!task) break;
      if (task.status === "cancelled") { this.priorityQueue.pop(); continue; }
      if (task.executeAt > Date.now()) break;

      this.priorityQueue.pop();
      this.executeTask(task);
    }
  }

  private async executeTask(task: Task): Promise<void> {
    task.status = "running";
    this.runningTasks.add(task.id);

    try {
      await task.execute();
      task.status = "success";
      this.completedTasks.add(task.id);
      console.log(`[Scheduler] 任務完成：${task.id}`);
      this.onTaskComplete(task.id);
    } catch (error) {
      console.error(`[Scheduler] 任務失敗：${task.id}`, error);
      if (task.retryCount < task.maxRetries) {
        task.retryCount++;
        task.status = "pending";
        // 指數退避：2^retryCount 秒
        task.executeAt = Date.now() + 2 ** task.retryCount * 1000;
        this.enqueueTask(task);
        console.log(`[Scheduler] 重試（${task.retryCount}/${task.maxRetries}）：${task.id}`);
      } else {
        task.status = "failed";
        console.log(`[Scheduler] 任務永久失敗（進入 Dead Letter Queue）：${task.id}`);
        this.onTaskComplete(task.id);
      }
    } finally {
      this.runningTasks.delete(task.id);
    }
  }

  /** 任務完成後，檢查依賴此任務的任務是否可就緒 */
  private onTaskComplete(completedId: string): void {
    for (const [, task] of this.allTasks) {
      if (task.status !== "pending") continue;
      if (!task.dependencies.includes(completedId)) continue;

      const allDepsComplete = task.dependencies.every(
        depId => this.completedTasks.has(depId)
      );

      if (allDepsComplete) {
        console.log(`[Scheduler] 依賴解除，任務就緒：${task.id}`);
        task.executeAt = Date.now();
        this.enqueueTask(task);
      }
    }
  }
}

// ─── 使用示範 ──────────────────────────────────────────────

async function demo(): Promise<void> {
  console.log("=== 任務排程器 Demo ===\n");

  const scheduler = new TaskScheduler(5);
  scheduler.start();

  // 建立任務的輔助函式
  const makeTask = (
    id: string, name: string, priority: TaskPriority,
    deps: string[] = [], delay: number = 0
  ): Omit<Task, "status" | "retryCount" | "createdAt"> => ({
    id, name,
    type: delay > 0 ? "delayed" : "immediate",
    priority,
    executeAt: Date.now() + delay,
    dependencies: deps,
    payload: {},
    maxRetries: 2,
    execute: async () => {
      await new Promise(r => setTimeout(r, 100));
      console.log(`  → 執行任務：${name} (ID: ${id})`);
    },
  });

  // 1. 即時優先級任務
  scheduler.submit(makeTask("t1", "高優先級緊急任務", 1));
  scheduler.submit(makeTask("t2", "普通任務A", 3));
  scheduler.submit(makeTask("t3", "低優先級任務", 5));

  // 2. DAG 工作流（ML 訓練管線）
  console.log("\n--- 提交 ML 訓練工作流 ---");
  scheduler.submitWorkflow([
    makeTask("fetch", "抓取資料", 3),
    makeTask("augment", "擴增資料", 3),
    makeTask("clean", "清洗資料", 2, ["fetch"]),
    makeTask("transform", "格式轉換", 2, ["augment"]),
    makeTask("feature", "特徵工程", 2, ["clean", "transform"]),
    makeTask("train", "訓練模型", 1, ["feature"]),
  ]);

  // 3. Cron 排程（示範計算下次時間）
  console.log("\n--- Cron 排程 ---");
  const nextTime = CronParser.nextExecutionTime("0 9 * * MON-FRI");
  console.log(`  每週一至週五 9:00，下次：${nextTime.toLocaleString("zh-TW")}`);

  const nextEveryFive = CronParser.nextExecutionTime("*/5 * * * *");
  console.log(`  每 5 分鐘，下次：${nextEveryFive.toLocaleString("zh-TW")}`);

  // 4. 延遲任務
  scheduler.submit(makeTask("delayed1", "3 秒後執行", 2, [], 3000));

  await new Promise(r => setTimeout(r, 2000));
  scheduler.stop();
  console.log("\n=== Demo 結束 ===");
}

demo().catch(console.error);
// 輸出：
//   [Scheduler] 排程器啟動
//   [Scheduler] 任務已提交：t1 (immediate, priority=1)
//   [Scheduler] 任務已提交：t2 (immediate, priority=3)
//   [Scheduler] 任務已提交：t3 (immediate, priority=5)
//     → 執行任務：高優先級緊急任務 (ID: t1)
//   [Scheduler] 任務完成：t1
//     → 執行任務：普通任務A (ID: t2)
//   [Scheduler] 任務完成：t2
//     → 執行任務：低優先級任務 (ID: t3)
//   [Scheduler] 任務完成：t3
//   --- 提交 ML 訓練工作流 ---
//   [Scheduler] 工作流拓撲排序結果：
//     第 1 波（可並行）：fetch, augment
//     第 2 波（可並行）：clean, transform
//     第 3 波（可並行）：feature
//     第 4 波（可並行）：train
//   --- Cron 排程 ---
//     每週一至週五 9:00，下次：2026/7/27 09:00:00
//   [Scheduler] 排程器停止
//   === Demo 結束 ===

// ============================================================
// Timer Wheel — C++ 高效實作
// 多層時間輪，支援百萬級延遲任務，O(1) 插入與觸發
// ============================================================

#include <vector>
#include <list>
#include <functional>
#include <string>
#include <memory>
#include <iostream>
#include <chrono>
#include <thread>
#include <mutex>
#include <atomic>
#include <unordered_map>
#include <queue>

// ─── 任務定義 ──────────────────────────────────────────────

struct TimerTask {
    std::string id;
    std::function<void()> callback;
    int64_t executeAtMs;       // 絕對執行時間（ms）
    int32_t rounds;            // 剩餘圈數
    bool cancelled = false;
};

using TaskPtr = std::shared_ptr<TimerTask>;

// ─── 單層 Timer Wheel ──────────────────────────────────────

class SingleTimerWheel {
public:
    explicit SingleTimerWheel(int slotCount = 512, int tickMs = 10)
        : slotCount_(slotCount), tickMs_(tickMs),
          slots_(slotCount), currentSlot_(0) {}

    /** 插入任務 — O(1) */
    void insert(TaskPtr task) {
        std::lock_guard<std::mutex> lock(mutex_);
        int64_t delayMs = std::max(0LL, task->executeAtMs - nowMs());
        int64_t ticks = delayMs / tickMs_;
        task->rounds = static_cast<int32_t>(ticks / slotCount_);
        int slot = (currentSlot_ + static_cast<int>(ticks % slotCount_)) % slotCount_;
        slots_[slot].push_back(task);
        taskIndex_[task->id] = task;
    }

    /** 取消任務 — O(1) 標記刪除 */
    bool cancel(const std::string& id) {
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = taskIndex_.find(id);
        if (it == taskIndex_.end()) return false;
        it->second->cancelled = true;
        taskIndex_.erase(it);
        return true;
    }

    /** 推進一個 tick，回傳到期任務 */
    std::vector<TaskPtr> tick() {
        std::lock_guard<std::mutex> lock(mutex_);
        std::vector<TaskPtr> ready;
        auto& slot = slots_[currentSlot_];

        for (auto it = slot.begin(); it != slot.end(); ) {
            auto& task = *it;
            if (task->cancelled) {
                it = slot.erase(it);
                continue;
            }
            if (task->rounds == 0) {
                ready.push_back(task);
                taskIndex_.erase(task->id);
                it = slot.erase(it);
            } else {
                task->rounds--;
                ++it;
            }
        }

        currentSlot_ = (currentSlot_ + 1) % slotCount_;
        return ready;
    }

private:
    static int64_t nowMs() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now().time_since_epoch()
        ).count();
    }

    int slotCount_;
    int64_t tickMs_;
    std::vector<std::list<TaskPtr>> slots_;
    int currentSlot_;
    std::mutex mutex_;
    std::unordered_map<std::string, TaskPtr> taskIndex_;
};

// ─── 多層 Timer Wheel（Hierarchical）────────────────────────

class HierarchicalTimerWheel {
public:
    HierarchicalTimerWheel()
        : wheel1_(256, 1),      // 256 槽，1ms/tick
          wheel2_(256, 256),    // 256 槽，256ms/tick
          wheel3_(256, 65536),  // 256 槽，65536ms/tick
          running_(false) {}

    void insert(TaskPtr task) {
        int64_t delayMs = task->executeAtMs - nowMs();
        if (delayMs < 256) {
            wheel1_.insert(task);
        } else if (delayMs < 65536) {
            wheel2_.insert(task);
        } else {
            wheel3_.insert(task);
        }
    }

    bool cancel(const std::string& id) {
        return wheel1_.cancel(id) || wheel2_.cancel(id) || wheel3_.cancel(id);
    }

    /** 啟動後台 Tick 執行緒 */
    void start(std::function<void(TaskPtr)> onTaskReady) {
        running_ = true;
        tickThread_ = std::thread([this, onTaskReady]() {
            while (running_) {
                // 推進第一層
                auto tasks = wheel1_.tick();
                for (auto& task : tasks) {
                    if (!task->cancelled) onTaskReady(task);
                }

                // 每 256 次推進第二層（降級到第一層）
                if (tick1Count_++ % 256 == 0) {
                    auto tasks2 = wheel2_.tick();
                    for (auto& task : tasks2) wheel1_.insert(task);
                }

                // 每 65536 次推進第三層
                if (tick2Count_++ % 256 == 0) {
                    auto tasks3 = wheel3_.tick();
                    for (auto& task : tasks3) wheel2_.insert(task);
                }

                std::this_thread::sleep_for(std::chrono::milliseconds(1));
            }
        });
    }

    void stop() {
        running_ = false;
        if (tickThread_.joinable()) tickThread_.join();
    }

private:
    static int64_t nowMs() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now().time_since_epoch()
        ).count();
    }

    SingleTimerWheel wheel1_, wheel2_, wheel3_;
    std::atomic<bool> running_;
    std::thread tickThread_;
    uint64_t tick1Count_ = 0;
    uint64_t tick2Count_ = 0;
};

// ─── Priority Queue 排程器（對比實作）─────────────────────────

class PriorityQueueScheduler {
public:
    struct TaskComparator {
        bool operator()(const TaskPtr& a, const TaskPtr& b) const {
            return a->executeAtMs > b->executeAtMs; // Min-Heap
        }
    };

    void insert(TaskPtr task) {
        std::lock_guard<std::mutex> lock(mutex_);
        pq_.push(task);
        cv_.notify_one();
    }

    void start(std::function<void(TaskPtr)> onTaskReady) {
        running_ = true;
        thread_ = std::thread([this, onTaskReady]() {
            while (running_) {
                std::unique_lock<std::mutex> lock(mutex_);
                if (pq_.empty()) {
                    cv_.wait_for(lock, std::chrono::milliseconds(100));
                    continue;
                }
                auto now = nowMs();
                if (pq_.top()->executeAtMs <= now) {
                    auto task = pq_.top();
                    pq_.pop();
                    lock.unlock();
                    if (!task->cancelled) onTaskReady(task);
                } else {
                    auto sleepMs = pq_.top()->executeAtMs - now;
                    cv_.wait_for(lock, std::chrono::milliseconds(sleepMs));
                }
            }
        });
    }

    void stop() {
        running_ = false;
        cv_.notify_all();
        if (thread_.joinable()) thread_.join();
    }

private:
    static int64_t nowMs() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now().time_since_epoch()
        ).count();
    }

    std::priority_queue<TaskPtr, std::vector<TaskPtr>, TaskComparator> pq_;
    std::mutex mutex_;
    std::condition_variable cv_;
    std::atomic<bool> running_{false};
    std::thread thread_;
};

// ─── main 功能示範 ─────────────────────────────────────────

int main() {
    std::cout << "=== Timer Wheel 排程器 Demo ===\n\n";

    HierarchicalTimerWheel scheduler;
    std::atomic<int> executedCount{0};
    std::mutex printMutex;

    scheduler.start([&](TaskPtr task) {
        std::lock_guard<std::mutex> lock(printMutex);
        std::cout << "  [執行] " << task->id << "\n";
        task->callback();
        executedCount++;
    });

    auto nowMs = []() {
        return std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now().time_since_epoch()
        ).count();
    };

    // 插入不同延遲的任務
    std::vector<std::pair<std::string, int>> taskDefs = {
        {"task-immediate", 0},
        {"task-50ms", 50},
        {"task-200ms", 200},
        {"task-500ms", 500},
        {"task-to-cancel", 300},
    };

    for (const auto& [id, delay] : taskDefs) {
        auto task = std::make_shared<TimerTask>();
        task->id = id;
        task->executeAtMs = nowMs() + delay;
        task->rounds = 0;
        task->callback = [id]() {
            std::cout << "    → 回呼：" << id << "\n";
        };
        scheduler.insert(task);
        std::cout << "已排程：" << id << "（delay=" << delay << "ms）\n";
    }

    // 取消一個任務
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    bool cancelled = scheduler.cancel("task-to-cancel");
    std::cout << "\n取消 task-to-cancel：" << (cancelled ? "成功" : "失敗") << "\n";

    std::this_thread::sleep_for(std::chrono::milliseconds(700));
    scheduler.stop();

    std::cout << "\n共執行任務：" << executedCount.load() << " 個\n";
    // 輸出：共執行任務：4 個（task-to-cancel 已被取消）
    std::cout << "=== Demo 結束 ===\n";

    return 0;
}