Multithreading — C++ المتقدم

الخيوط والتزامن (Multithreading)

المعالجات الحديثة فيها عدة أنوية. الخيط (thread) مسار تنفيذ مستقل داخل البرنامج؛ عدة خيوط تعمل معاً تستغل الأنوية. لكن الخيوط تتشارك الذاكرة، وهذا مصدر أصعب الأخطاء في البرمجة.

إنشاء خيط

#include <iostream>
#include <thread>
using namespace std;

void work(int id) { cout << "خيط " << id << '\n'; }

int main() {
    thread t1(work, 1);
    thread t2([] { cout << "خيط lambda\n"; });
    t1.join();            // انتظر انتهاءه
    t2.join();
}

- يجب استدعاء join() (انتظار) أو detach() (فصل) قبل تدمير كائن thread، وإلا يُستدعى std::terminate.

- ترتيب الطباعة بين الخيوط غير محدد: قد يتغير كل تشغيل.

jthread (C++20) — RAII للخيوط

std::jthread يستدعي join() تلقائياً في مدمّره، فلا تنسى:

#include <thread>
#include <vector>

void runAll() {
    std::vector<std::jthread> workers;
    for (int i = 0; i < 4; i++)
        workers.emplace_back([i] { /* عمل */ });
}   // كل الخيوط تُنتظر هنا تلقائياً

سباق البيانات (Data Race)

خيطان يعدّلان نفس المتغير بلا تزامن = سلوك غير معرّف. counter++ ليست عملية واحدة بل ثلاث (قراءة، زيادة، كتابة)، فتضيع زيادات:

الخيط A: يقرأ 5         الخيط B: يقرأ 5
الخيط A: يكتب 6         الخيط B: يكتب 6     ← زيادتان، والنتيجة 6!

المثال الكامل يعرض هذا للتعليم فقط؛ في كود حقيقي هو خطأ وليس "نتيجة تقريبية".

الحل 1: mutex

#include <mutex>

std::mutex m;
int counter = 0;

void safeAdd(int times) {
    for (int i = 0; i < times; i++) {
        std::lock_guard<std::mutex> lock(m);   // يقفل هنا (RAII)
        ++counter;
    }                                          // يفتح تلقائياً عند نهاية كل دورة
}

لا تستدعِ m.lock() و m.unlock() يدوياً: استثناء في المنتصف يترك القفل مغلقاً للأبد. استخدم lock_guard أو scoped_lock (يقفل عدة mutex معاً دون جمود).

الحل 2: atomic

للعدادات والأعلام البسيطة، std::atomic أسرع من mutex:

#include <atomic>

std::atomic<int> hits{0};
void hit() { hits++; }        // عملية واحدة غير قابلة للتجزئة

الجمود (Deadlock)

خيط A يقفل m1 وينتظر m2، وخيط B يقفل m2 وينتظر m1 ← كلاهما ينتظر للأبد. الوقاية: اقفل دائماً بنفس الترتيب، أو استخدم scoped_lock lock(m1, m2);.

async و future: نتائج من خيوط

عندما تريد قيمة من حساب متوازٍ، async أبسط من thread:

#include <future>
#include <iostream>
using namespace std;

long long sumRange(long long a, long long b) {
    long long s = 0;
    for (long long i = a; i <= b; i++) s += i;
    return s;
}

int main() {
    auto first = async(launch::async, sumRange, 1, 500000);
    auto second = async(launch::async, sumRange, 500001, 1000000);
    cout << first.get() + second.get() << '\n';   // 500000500000
}

get() تنتظر النتيجة، وإن رمى الخيط استثناء فإن get() تعيد رميه في الخيط المستدعي. لاحظ long long: المجموع أكبر من حد int.

نصائح

- أقل مشاركة = أقل مشاكل: أعطِ كل خيط بياناته الخاصة واجمع النتائج في النهاية.

- لا تُنشئ خيطاً لكل مهمة صغيرة؛ إنشاء الخيط مكلف. عدد الأنوية: thread::hardware_concurrency().

- أدوات اكتشاف السباقات: -fsanitize=thread مع g++/clang.

تمرين

قسّم vector من 10 ملايين عدد على 4 أجزاء، واحسب مجموع كل جزء بـ async، ثم اجمع النتائج. قارن الزمن مع الحساب في خيط واحد باستخدام chrono::steady_clock.

#include <iostream>
#include <thread>
#include <mutex>
#include <vector>
#include <future>
#include <chrono>
#include <atomic>
using namespace std;

mutex mtx;
atomic<int> atomicCounter(0);

// ===== Thread أساسي =====
void sayHello(int id) {
    lock_guard<mutex> lock(mtx);
    cout << "مرحباً من Thread " << id << endl;
}

void printNumbers(int start, int end) {
    for (int i = start; i <= end; i++) {
        lock_guard<mutex> lock(mtx);
        cout << this_thread::get_id() << ": " << i << endl;
        this_thread::sleep_for(chrono::milliseconds(100));
    }
}

// ===== مع Lambda =====
void lambdaExample() {
    int x = 10;
    
    thread t([&x]() {
        x *= 2;
        cout << "Thread: x = " << x << endl;
    });
    
    t.join();
    cout << "Main: x = " << x << endl;
}

// ===== Race Condition =====
int unsafeCounter = 0;
int safeCounter = 0;

void unsafeIncrement(int iterations) {
    for (int i = 0; i < iterations; i++) {
        unsafeCounter++;  // Race Condition! (سلوك غير معرّف — للعرض فقط)
    }
}

void safeIncrement(int iterations) {
    for (int i = 0; i < iterations; i++) {
        lock_guard<mutex> lock(mtx);
        safeCounter++;
    }
}

void atomicIncrement(int iterations) {
    for (int i = 0; i < iterations; i++) {
        atomicCounter++;  // Thread-safe!
    }
}

// ===== async و future =====
long long computeSum(int start, int end) {
    long long sum = 0;  // int يفيض: مجموع 1..500000 أكبر من 2^31
    for (int i = start; i <= end; i++) {
        sum += i;
    }
    return sum;
}

long long fibonacci(int n) {
    if (n <= 1) return n;
    return fibonacci(n - 1) + fibonacci(n - 2);
}

int main() {
    cout << "=== Threads الأساسية ===" << endl;
    
    vector<thread> threads;
    for (int i = 0; i < 5; i++) {
        threads.push_back(thread(sayHello, i));
    }
    
    for (auto& t : threads) {
        t.join();
    }
    
    // ===== Lambda =====
    cout << "\n=== Lambda Thread ===" << endl;
    lambdaExample();
    
    // ===== Race Condition Demo =====
    cout << "\n=== Race Condition ===" << endl;
    
    vector<thread> workers;
    
    // Unsafe
    for (int i = 0; i < 4; i++) {
        workers.push_back(thread(unsafeIncrement, 10000));
    }
    for (auto& w : workers) w.join();
    cout << "Unsafe counter: " << unsafeCounter << " (expected: 40000)" << endl;
    
    workers.clear();
    
    // Safe with mutex
    for (int i = 0; i < 4; i++) {
        workers.push_back(thread(safeIncrement, 10000));
    }
    for (auto& w : workers) w.join();
    cout << "Safe counter: " << safeCounter << endl;
    
    workers.clear();
    
    // Atomic
    for (int i = 0; i < 4; i++) {
        workers.push_back(thread(atomicIncrement, 10000));
    }
    for (auto& w : workers) w.join();
    cout << "Atomic counter: " << atomicCounter << endl;
    
    // ===== async =====
    cout << "\n=== async ===" << endl;
    
    auto start = chrono::high_resolution_clock::now();
    
    // Parallel computation
    auto f1 = async(launch::async, computeSum, 1, 500000);
    auto f2 = async(launch::async, computeSum, 500001, 1000000);
    
    long long result = f1.get() + f2.get();
    
    auto end = chrono::high_resolution_clock::now();
    auto duration = chrono::duration_cast<chrono::milliseconds>(end - start);
    
    cout << "المجموع: " << result << endl;
    cout << "الوقت: " << duration.count() << " ms" << endl;
    
    // ===== Parallel Fibonacci =====
    cout << "\n=== Parallel Fibonacci ===" << endl;
    
    start = chrono::high_resolution_clock::now();
    
    auto fib30 = async(launch::async, fibonacci, 30);
    auto fib35 = async(launch::async, fibonacci, 35);
    
    cout << "fib(30) = " << fib30.get() << endl;
    cout << "fib(35) = " << fib35.get() << endl;
    
    end = chrono::high_resolution_clock::now();
    duration = chrono::duration_cast<chrono::milliseconds>(end - start);
    cout << "الوقت: " << duration.count() << " ms" << endl;
    
    return 0;
}