التزامن والخيوط في Rust: معالجة متوازية آمنة للبيانات الصناعية

لماذا التزامن مهم في المصانع

في المصنع الحقيقي، عشرات الآلات تعمل في وقت واحد. كل آلة تُنتج بيانات: درجات حرارة، ضغط، اهتزاز، سرعة دوران. لا يمكن قراءة مستشعر واحد ثم الانتظار حتى ينتهي قبل قراءة التالي - نحتاج قراءتها جميعاً بالتوازي.

خط الإنتاج الواقعي:
┌──────────┐  ┌──────────┐  ┌──────────┐
│ آلة القص │  │ فرن التجفيف│  │ آلة التغليف│
│ خيط #1   │  │ خيط #2    │  │ خيط #3    │
└────┬─────┘  └────┬─────┘  └─────┬────┘
     │             │              │
     └─────────────┼──────────────┘
                   ▼
          ┌────────────────┐
          │ المجمّع المركزي │
          │  (الخيط الرئيسي) │
          └────────────────┘

Rust يضمن سلامة التزامن عند التجميع - لا سباق بيانات في وقت التشغيل.

إنشاء الخيوط مع std::thread

كل خيط (thread) يعمل بشكل مستقل، مثل عامل مسؤول عن آلة واحدة:

use std::thread;
use std::time::Duration;

fn main() {
    // إنشاء خيط لمراقبة آلة القص
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("[آلة القص] قراءة #{i}: {}°", 70.0 + i as f64 * 0.3);
            thread::sleep(Duration::from_millis(100));
        }
        "اكتملت قراءات القص" // القيمة المُرجعة
    });

    // الخيط الرئيسي يستمر بالعمل بالتوازي
    for i in 1..=3 {
        println!("[الرئيسي] فحص دوري #{i}");
        thread::sleep(Duration::from_millis(150));
    }

    // انتظار انتهاء الخيط واستلام النتيجة
    let result = handle.join().expect("الخيط فشل");
    println!("النتيجة: {result}");
}

لنقل بيانات إلى الخيط، نستخدم move لنقل الملكية:

let machine_id = String::from("CUT-01");
let threshold = 85.0;

let handle = thread::spawn(move || {
    // machine_id و threshold انتقلت ملكيتهما إلى هذا الخيط
    println!("مراقبة الآلة: {machine_id} (حد: {threshold}°)");
});

// لا يمكن استخدام machine_id هنا بعد الآن
handle.join().unwrap();

تمرير الرسائل عبر القنوات

القنوات (channels) تسمح للخيوط بإرسال بيانات بأمان، مثل حزام ناقل بين الأقسام:

use std::sync::mpsc; // multiple producer, single consumer
use std::thread;
use std::time::Duration;

struct SensorReading {
    machine: String,
    value: f64,
}

fn main() {
    // إنشاء قناة: المرسل والمستقبل
    let (tx, rx) = mpsc::channel();

    // خيط آلة القص
    let tx1 = tx.clone();
    thread::spawn(move || {
        let readings = vec![72.5, 73.1, 74.0];
        for val in readings {
            tx1.send(SensorReading {
                machine: "CUT-01".into(),
                value: val,
            }).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });

    // خيط فرن التجفيف
    let tx2 = tx.clone();
    thread::spawn(move || {
        let readings = vec![180.0, 182.5, 179.8];
        for val in readings {
            tx2.send(SensorReading {
                machine: "OVEN-03".into(),
                value: val,
            }).unwrap();
            thread::sleep(Duration::from_millis(120));
        }
    });

    // إسقاط المرسل الأصلي حتى تنتهي الحلقة
    drop(tx);

    // المجمّع المركزي يستقبل من جميع الخيوط
    for reading in rx {
        println!("[{}] القيمة: {:.1}°", reading.machine, reading.value);
    }
    println!("انتهت جميع القراءات");
}

الحالة المشتركة: Mutex و Arc

أحياناً نحتاج أن تكتب عدة خيوط في مكان واحد، مثل سجل مشترك. Mutex يضمن أن خيطاً واحداً فقط يكتب في كل لحظة، وArc يسمح بالملكية المشتركة.

use std::sync::{Arc, Mutex};
use std::thread;

struct Reading {
    machine: String,
    value: f64,
}

fn main() {
    // سجل مشترك محمي بقفل، ملفوف في عدّاد مراجع ذري
    let log = Arc::new(Mutex::new(Vec::<Reading>::new()));

    let mut handles = vec![];

    let machines = vec!["CUT-01", "OVEN-03", "PACK-07"];

    for machine_name in machines {
        let log_clone = Arc::clone(&log); // نسخة من المؤشر الذكي
        let name = machine_name.to_string();

        let handle = thread::spawn(move || {
            for i in 0..3 {
                let reading = Reading {
                    machine: name.clone(),
                    value: 70.0 + i as f64 * 1.5,
                };

                // اكتساب القفل، الإضافة، ثم تحرير تلقائي
                let mut shared_log = log_clone.lock().unwrap();
                shared_log.push(reading);
                // القفل يُحرّر هنا تلقائياً عند خروج المتغير من النطاق
            }
        });
        handles.push(handle);
    }

    // انتظار جميع الخيوط
    for h in handles {
        h.join().unwrap();
    }

    // قراءة السجل النهائي
    let final_log = log.lock().unwrap();
    println!("إجمالي القراءات المسجّلة: {}", final_log.len());
    for r in final_log.iter() {
        println!("  {} => {:.1}°", r.machine, r.value);
    }
}

Send و Sync: ضمانات التزامن في Rust

المترجم يفرض قواعد التزامن من خلال سمتين (traits) تلقائيتين:

// Send: يمكن نقل ملكية القيمة إلى خيط آخر
// معظم الأنواع تحقق Send تلقائياً

// Sync: يمكن مشاركة المرجع بين عدة خيوط
// T تحقق Sync إذا كانت &T تحقق Send

// أمثلة عملية:
// Vec<f64>          -> Send + Sync  (آمن للنقل والمشاركة)
// Arc<Mutex<Vec>>   -> Send + Sync  (آمن بفضل القفل)
// Rc<T>             -> !Send + !Sync (خيط واحد فقط!)

use std::rc::Rc;
use std::sync::Arc;

fn main() {
    // هذا يعمل: Arc مصمم للخيوط المتعددة
    let data = Arc::new(vec![1, 2, 3]);
    let data_clone = Arc::clone(&data);
    std::thread::spawn(move || {
        println!("من خيط آخر: {:?}", data_clone);
    }).join().unwrap();

    // هذا لن يُجمّع: Rc ليس آمناً للخيوط
    // let bad = Rc::new(vec![1, 2, 3]);
    // std::thread::spawn(move || {
    //     println!("{:?}", bad); // خطأ: Rc لا يحقق Send
    // });
}

المترجم يرفض الكود غير الآمن قبل التشغيل - هذه ميزة فريدة في Rust.

مثال عملي: مراقبة متوازية للآلات

use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::time::Duration;
use std::collections::HashMap;

/// قراءة مستشعر من آلة معيّنة
fn simulate_sensor(machine: &str, base_temp: f64) -> f64 {
    // محاكاة تذبذب طبيعي
    base_temp + (rand_simple() * 5.0 - 2.5)
}

fn rand_simple() -> f64 {
    // محاكاة مبسّطة لرقم عشوائي
    0.5
}

fn main() {
    let (tx, rx) = mpsc::channel();
    let alarm_count = Arc::new(Mutex::new(0u32));

    // تعريف الآلات ودرجات حرارتها الأساسية
    let machines = vec![
        ("CUT-01", 72.0, 85.0),   // (اسم، حرارة أساسية، حد الإنذار)
        ("OVEN-03", 180.0, 200.0),
        ("PACK-07", 25.0, 35.0),
    ];

    let mut handles = vec![];

    for (name, base, threshold) in machines {
        let tx = tx.clone();
        let alarms = Arc::clone(&alarm_count);

        let handle = thread::spawn(move || {
            for cycle in 1..=5 {
                let temp = simulate_sensor(name, base);
                let is_alarm = temp > threshold;

                if is_alarm {
                    let mut count = alarms.lock().unwrap();
                    *count += 1;
                }

                tx.send(format!(
                    "[دورة {cycle}] {name}: {temp:.1}° {}",
                    if is_alarm { "⚠ إنذار!" } else { "✓" }
                )).unwrap();

                thread::sleep(Duration::from_millis(80));
            }
        });
        handles.push(handle);
    }

    drop(tx); // إغلاق المرسل الأصلي

    // المجمّع المركزي: طباعة جميع الرسائل
    println!("=== بدء المراقبة المتوازية ===");
    for msg in rx {
        println!("{msg}");
    }

    // انتظار انتهاء الجميع
    for h in handles {
        h.join().unwrap();
    }

    let total_alarms = alarm_count.lock().unwrap();
    println!("=== انتهت المراقبة ===");
    println!("إجمالي الإنذارات: {total_alarms}");
}

الخلاصة

• التزامن ضروري في الأنظمة الصناعية حيث عشرات الآلات تعمل معاً
• thread::spawn يُنشئ خيطاً مستقلاً، وjoin() ينتظر انتهاءه
• القنوات (mpsc::channel) تنقل البيانات بأمان بين الخيوط كحزام ناقل
• Mutex<T> يحمي البيانات المشتركة بقفل، وArc<T> يسمح بمشاركة الملكية
• Send و Sync يضمنان سلامة التزامن في وقت التجميع، لا وقت التشغيل
• استخدم القنوات عندما تتدفق البيانات في اتجاه واحد، والقفل عند الحاجة لحالة مشتركة
• في الدرس القادم سنتعلم معالجة الأخطاء بأسلوب Rust باستخدام Result و Option