async_future.cpp

// async_future.cpp
// 編譯指令：g++ -std=c++17 -Wall -pthread async_future.cpp -o async_future
//
// 注意：多執行緒程式必須加上 -pthread 旗標
//
// 本程式示範 std::async、std::future、std::promise、std::shared_future

#include <iostream>
#include <future>
#include <thread>
#include <vector>
#include <string>
#include <numeric>
#include <chrono>
#include <cmath>
#include <stdexcept>
#include <mutex>
#include <sstream>

std::mutex cout_mutex;

template<typename... Args>
void safe_print(const Args&... args) {
    std::lock_guard<std::mutex> lock(cout_mutex);
    (std::cout << ... << args) << "\n";
}

// ============================================================
// 第一部分：std::async 基本用法
// ============================================================

// 模擬耗時計算
long long compute_sum(int start, int end) {
    long long sum = 0;
    for (int i = start; i <= end; ++i) {
        sum += i;
    }
    return sum;
}

// 模擬下載檔案
std::string download_file(const std::string& url) {
    safe_print("  下載中: ", url);
    std::this_thread::sleep_for(std::chrono::milliseconds(200));
    return "內容來自 " + url;
}

void demo_async_basics() {
    std::cout << "========================================\n";
    std::cout << "  std::async 基本用法\n";
    std::cout << "========================================\n\n";

    // 基本 async 呼叫
    std::cout << "1. 基本非同步呼叫：\n";
    auto future1 = std::async(std::launch::async, compute_sum, 1, 1000000);
    std::cout << "  計算已在背景啟動...\n";
    std::cout << "  主執行緒可以做其他事情...\n";
    long long result = future1.get();  // 阻塞直到結果就緒
    std::cout << "  1 到 1000000 的總和 = " << result << "\n";

    // 使用 lambda
    std::cout << "\n2. 使用 lambda：\n";
    auto future2 = std::async(std::launch::async, []() {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        return 42;
    });
    std::cout << "  等待結果: " << future2.get() << "\n";

    // 多個非同步任務
    std::cout << "\n3. 多個非同步任務（模擬平行下載）：\n";
    auto start = std::chrono::steady_clock::now();

    auto f1 = std::async(std::launch::async, download_file, "file1.txt");
    auto f2 = std::async(std::launch::async, download_file, "file2.txt");
    auto f3 = std::async(std::launch::async, download_file, "file3.txt");

    std::cout << "  " << f1.get() << "\n";
    std::cout << "  " << f2.get() << "\n";
    std::cout << "  " << f3.get() << "\n";

    auto end = std::chrono::steady_clock::now();
    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << "  三個下載共耗時 " << ms << " ms（平行執行，非 3x200ms）\n\n";
}

// ============================================================
// 第二部分：啟動策略
// ============================================================

void demo_launch_policies() {
    std::cout << "========================================\n";
    std::cout << "  啟動策略（launch policies）\n";
    std::cout << "========================================\n\n";

    // std::launch::async — 保證在新執行緒執行
    std::cout << "1. std::launch::async（在新執行緒執行）：\n";
    auto f1 = std::async(std::launch::async, []() {
        safe_print("  [async] 執行緒 ID: ", std::this_thread::get_id());
        return 100;
    });
    std::cout << "  主執行緒 ID: " << std::this_thread::get_id() << "\n";
    std::cout << "  結果: " << f1.get() << "\n";

    // std::launch::deferred — 延遲到 get() 時在呼叫者執行緒執行
    std::cout << "\n2. std::launch::deferred（延遲執行）：\n";
    auto f2 = std::async(std::launch::deferred, []() {
        safe_print("  [deferred] 執行緒 ID: ", std::this_thread::get_id());
        return 200;
    });
    std::cout << "  呼叫 get() 前，函式尚未執行\n";
    std::cout << "  主執行緒 ID: " << std::this_thread::get_id() << "\n";
    std::cout << "  結果: " << f2.get() << " (此時才執行)\n";

    // 預設策略
    std::cout << "\n3. 預設策略（async | deferred）：\n";
    auto f3 = std::async([]() {
        return 300;
    });
    std::cout << "  由系統決定在新執行緒或延遲執行\n";
    std::cout << "  結果: " << f3.get() << "\n\n";
}

// ============================================================
// 第三部分：future 的操作
// ============================================================

void demo_future_operations() {
    std::cout << "========================================\n";
    std::cout << "  std::future 操作\n";
    std::cout << "========================================\n\n";

    // valid() — 檢查 future 是否有效
    std::cout << "1. valid() 檢查：\n";
    std::future<int> f1;
    std::cout << "  預設建構的 future valid: " << std::boolalpha << f1.valid() << "\n";

    f1 = std::async(std::launch::async, []() { return 42; });
    std::cout << "  設定後 valid: " << f1.valid() << "\n";
    f1.get();
    std::cout << "  get() 後 valid: " << f1.valid() << "\n";

    // wait() — 等待但不取值
    std::cout << "\n2. wait() 等待：\n";
    auto f2 = std::async(std::launch::async, []() {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        return 99;
    });
    std::cout << "  等待完成...\n";
    f2.wait();
    std::cout << "  已完成，取值: " << f2.get() << "\n";

    // wait_for() — 限時等待
    std::cout << "\n3. wait_for() 限時等待：\n";
    auto f3 = std::async(std::launch::async, []() {
        std::this_thread::sleep_for(std::chrono::milliseconds(300));
        return 77;
    });

    auto status = f3.wait_for(std::chrono::milliseconds(50));
    if (status == std::future_status::timeout) {
        std::cout << "  50ms 後：超時（尚未完成）\n";
    } else if (status == std::future_status::ready) {
        std::cout << "  50ms 後：已就緒\n";
    }

    // 繼續等待直到完成
    status = f3.wait_for(std::chrono::milliseconds(500));
    if (status == std::future_status::ready) {
        std::cout << "  再等 500ms 後：已就緒，值 = " << f3.get() << "\n";
    }

    // future_status 的所有值
    std::cout << "\n4. future_status 列舉：\n";
    std::cout << "  ready    - 結果已就緒\n";
    std::cout << "  timeout  - 等待逾時\n";
    std::cout << "  deferred - 使用 deferred 策略且尚未開始\n";

    std::cout << "\n";
}

// ============================================================
// 第四部分：std::promise 與 std::future 配對
// ============================================================

void demo_promise_future() {
    std::cout << "========================================\n";
    std::cout << "  std::promise 與 std::future\n";
    std::cout << "========================================\n\n";

    // 基本使用：在執行緒間傳遞值
    std::cout << "1. 基本值傳遞：\n";
    {
        std::promise<int> promise;
        std::future<int> future = promise.get_future();

        std::thread t([&promise]() {
            safe_print("  [工作者] 計算中...");
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            promise.set_value(42);
            safe_print("  [工作者] 已設定結果");
        });

        std::cout << "  [主程式] 等待結果...\n";
        int result = future.get();
        std::cout << "  [主程式] 收到: " << result << "\n";
        t.join();
    }

    // 傳遞字串
    std::cout << "\n2. 傳遞複雜型別：\n";
    {
        std::promise<std::string> promise;
        auto future = promise.get_future();

        std::thread t([&promise]() {
            std::string result = "計算結果: ";
            for (int i = 0; i < 5; ++i) {
                result += std::to_string(i * i) + " ";
            }
            promise.set_value(std::move(result));
        });

        std::cout << "  收到: \"" << future.get() << "\"\n";
        t.join();
    }

    // 傳遞例外
    std::cout << "\n3. 傳遞例外：\n";
    {
        std::promise<double> promise;
        auto future = promise.get_future();

        std::thread t([&promise]() {
            try {
                // 模擬錯誤
                throw std::runtime_error("除以零錯誤");
            } catch (...) {
                promise.set_exception(std::current_exception());
            }
        });

        try {
            double result = future.get();
            std::cout << "  結果: " << result << "\n";
        } catch (const std::exception& e) {
            std::cout << "  捕獲例外: " << e.what() << "\n";
        }
        t.join();
    }

    // 多階段流水線
    std::cout << "\n4. 多階段流水線：\n";
    {
        std::promise<int> p1, p2, p3;
        auto f1 = p1.get_future();
        auto f2 = p2.get_future();
        auto f3 = p3.get_future();

        // 階段 1：產生資料
        std::thread stage1([&p1]() {
            safe_print("  [階段1] 產生資料...");
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
            p1.set_value(10);
        });

        // 階段 2：處理資料
        std::thread stage2([&f1, &p2]() {
            int input = f1.get();
            safe_print("  [階段2] 收到 ", input, "，處理中...");
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
            p2.set_value(input * 3);
        });

        // 階段 3：最終處理
        std::thread stage3([&f2, &p3]() {
            int input = f2.get();
            safe_print("  [階段3] 收到 ", input, "，最終處理...");
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
            p3.set_value(input + 7);
        });

        int final_result = f3.get();
        std::cout << "  最終結果: " << final_result << " (10 * 3 + 7 = 37)\n";

        stage1.join();
        stage2.join();
        stage3.join();
    }

    std::cout << "\n";
}

// ============================================================
// 第五部分：std::shared_future
// ============================================================

void demo_shared_future() {
    std::cout << "========================================\n";
    std::cout << "  std::shared_future（多個消費者）\n";
    std::cout << "========================================\n\n";

    // shared_future 可以被多次 get()，多個執行緒共用
    std::promise<int> promise;
    std::shared_future<int> shared = promise.get_future().share();

    std::vector<std::thread> consumers;
    for (int i = 0; i < 4; ++i) {
        consumers.emplace_back([shared, i]() {
            safe_print("  [消費者 ", i, "] 等待結果...");
            int value = shared.get();  // 所有消費者都能取得相同的值
            safe_print("  [消費者 ", i, "] 收到: ", value);
        });
    }

    // 生產者
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    safe_print("  [生產者] 設定值 = 999");
    promise.set_value(999);

    for (auto& t : consumers) t.join();

    std::cout << "\nshared_future 允許多個執行緒共享同一個結果\n";
    std::cout << "（普通 future 的 get() 只能呼叫一次）\n\n";
}

// ============================================================
// 第六部分：例外傳遞
// ============================================================

double risky_computation(double x) {
    if (x < 0) {
        throw std::invalid_argument("輸入值不能為負數: " + std::to_string(x));
    }
    if (x == 0) {
        throw std::runtime_error("除以零");
    }
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    return std::sqrt(x) / x;
}

void demo_exception_propagation() {
    std::cout << "========================================\n";
    std::cout << "  例外傳遞（通過 future）\n";
    std::cout << "========================================\n\n";

    std::vector<double> inputs = {4.0, -1.0, 0.0, 16.0};

    for (double x : inputs) {
        auto future = std::async(std::launch::async, risky_computation, x);

        try {
            double result = future.get();
            std::cout << "  risky_computation(" << x << ") = " << result << "\n";
        } catch (const std::invalid_argument& e) {
            std::cout << "  risky_computation(" << x << ") 錯誤: " << e.what() << "\n";
        } catch (const std::runtime_error& e) {
            std::cout << "  risky_computation(" << x << ") 錯誤: " << e.what() << "\n";
        }
    }

    std::cout << "\n例外會從工作執行緒傳遞到呼叫 get() 的執行緒\n\n";
}

// ============================================================
// 第七部分：平行計算範例
// ============================================================

// 分段平行加總
long long parallel_sum(const std::vector<int>& data, int num_tasks) {
    int chunk_size = static_cast<int>(data.size()) / num_tasks;
    std::vector<std::future<long long>> futures;

    for (int i = 0; i < num_tasks; ++i) {
        int start = i * chunk_size;
        int end = (i == num_tasks - 1) ? static_cast<int>(data.size()) : start + chunk_size;

        futures.push_back(std::async(std::launch::async,
            [&data, start, end]() -> long long {
                return std::accumulate(data.begin() + start,
                                       data.begin() + end, 0LL);
            }));
    }

    long long total = 0;
    for (auto& f : futures) {
        total += f.get();
    }
    return total;
}

// 平行質數計數
int count_primes_in_range(int start, int end) {
    int count = 0;
    for (int n = start; n < end; ++n) {
        if (n < 2) continue;
        bool is_prime = true;
        for (int i = 2; i * i <= n; ++i) {
            if (n % i == 0) { is_prime = false; break; }
        }
        if (is_prime) ++count;
    }
    return count;
}

void demo_parallel_computation() {
    std::cout << "========================================\n";
    std::cout << "  平行計算範例\n";
    std::cout << "========================================\n\n";

    // 平行加總
    const int n = 10000000;
    std::vector<int> data(n);
    std::iota(data.begin(), data.end(), 1);

    std::cout << "1. 平行加總 (1 到 " << n << ")：\n";

    auto t1 = std::chrono::steady_clock::now();
    long long serial_sum = std::accumulate(data.begin(), data.end(), 0LL);
    auto t2 = std::chrono::steady_clock::now();
    auto serial_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();

    t1 = std::chrono::steady_clock::now();
    long long par_sum = parallel_sum(data, 4);
    t2 = std::chrono::steady_clock::now();
    auto par_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();

    std::cout << "  序列加總: " << serial_sum << " (" << serial_ms << " ms)\n";
    std::cout << "  平行加總: " << par_sum << " (" << par_ms << " ms, 4 任務)\n";
    std::cout << "  結果" << (serial_sum == par_sum ? "一致 ✓" : "不一致 ✗") << "\n";

    // 平行質數計數
    std::cout << "\n2. 平行質數計數 (1 到 100000)：\n";
    const int range = 100000;
    const int tasks = 4;
    int chunk = range / tasks;

    std::vector<std::future<int>> prime_futures;
    t1 = std::chrono::steady_clock::now();
    for (int i = 0; i < tasks; ++i) {
        int start = i * chunk;
        int end = (i == tasks - 1) ? range : start + chunk;
        prime_futures.push_back(
            std::async(std::launch::async, count_primes_in_range, start, end));
    }

    int total_primes = 0;
    for (auto& f : prime_futures) {
        total_primes += f.get();
    }
    t2 = std::chrono::steady_clock::now();
    par_ms = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();

    std::cout << "  1 到 " << range << " 之間有 " << total_primes
              << " 個質數 (" << par_ms << " ms)\n";

    std::cout << "\n";
}

// ============================================================
// 第八部分：模擬非同步檔案處理
// ============================================================

struct FileData {
    std::string filename;
    std::string content;
    std::size_t size;
};

FileData async_read_file(const std::string& filename) {
    safe_print("  讀取 ", filename, "...");
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    std::string content = "File content of " + filename + " (模擬資料)";
    return {filename, content, content.size()};
}

std::string async_process_file(const FileData& data) {
    safe_print("  處理 ", data.filename, "...");
    std::this_thread::sleep_for(std::chrono::milliseconds(50));
    std::ostringstream oss;
    oss << "已處理: " << data.filename << " (" << data.size << " bytes)";
    return oss.str();
}

void demo_async_file_processing() {
    std::cout << "========================================\n";
    std::cout << "  非同步檔案處理模擬\n";
    std::cout << "========================================\n\n";

    std::vector<std::string> filenames = {
        "config.json", "data.csv", "log.txt", "report.html"
    };

    auto start = std::chrono::steady_clock::now();

    // 階段 1：平行讀取所有檔案
    std::vector<std::future<FileData>> read_futures;
    for (const auto& name : filenames) {
        read_futures.push_back(
            std::async(std::launch::async, async_read_file, name));
    }

    // 階段 2：讀取完成後平行處理
    std::vector<std::future<std::string>> process_futures;
    for (auto& rf : read_futures) {
        FileData data = rf.get();  // 等待讀取完成
        process_futures.push_back(
            std::async(std::launch::async, async_process_file, data));
    }

    // 收集結果
    std::cout << "\n  處理結果：\n";
    for (auto& pf : process_futures) {
        std::cout << "    " << pf.get() << "\n";
    }

    auto end = std::chrono::steady_clock::now();
    auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
    std::cout << "\n  總耗時: " << ms << " ms\n";
    std::cout << "  (若序列執行需 " << filenames.size() * 150 << " ms)\n\n";
}

// ============================================================
// 主程式
// ============================================================

int main() {
    std::cout << "╔══════════════════════════════════════════╗\n";
    std::cout << "║  C++17 async、future 與 promise          ║\n";
    std::cout << "╚══════════════════════════════════════════╝\n\n";

    demo_async_basics();
    demo_launch_policies();
    demo_future_operations();
    demo_promise_future();
    demo_shared_future();
    demo_exception_propagation();
    demo_parallel_computation();
    demo_async_file_processing();

    std::cout << "=== 程式結束 ===\n";
    return 0;
}