Concurrent-kolleksiyalarına giriş

1. Problemin fonu

Tək iplikli tətbiqdə List<T>, Dictionary<T> kimi kolleksiyalar gözlənilən şəkildə işləyir. Amma eyni kolleksiyaya bir neçə iplik eyni vaxtda müraciət etməyə başladıqda tanış problem yaranır: məlumat yarışları (race conditions).

Əgər bir neçə iplik eyni kolleksiyadan oxumağa və/və ya yazmağa cəhd edirsə və lazımi sinxronizasiya yoxdursa, aşağıdakıları ala bilərsiniz:

• Yanlış/uyğunsuz məlumat: bir element bir iplik tərəfindən silinə bilər, başqa iplik isə onu yeniləməyə çalışırmış kimi davrana bilər.
• Məlumat itkisi: bir iplik elementi əlavə etdi, digər iplik isə əvvəlki yazını bilmədən onu üzərindən yazdı.
• Exceptions: kolleksiya qeyri-valid vəziyyətə düşə və InvalidOperationException (məsələn, "Collection was modified; enumeration operation may not execute.") və ya hətta NullReferenceException verə bilər.

Nümunə 1: List<T>-də məlumat yarışı (sadə inkrement)

İki iplik eyni list elementini eyni vaxtda inkrement edir.

using System.Collections.Generic;
using System.Threading.Tasks; // Task.Run üçün

class RaceConditionExample
{
    static List<int> numbers = new List<int> { 0 }; // Bir elementli siyahı

    static void Main(string[] args)
    {
        Console.WriteLine("Başlanğıc dəyər: " + numbers[0]); // 0

        // İki iplik işə salırıq, hər biri numbers[0]-ı inkrement edir
        Task task1 = Task.Run(() => IncrementNumbers(500_000));
        Task task2 = Task.Run(() => IncrementNumbers(500_000));

        Task.WaitAll(task1, task2); // Hər iki ipliyin bitməsini gözləyirik

        Console.WriteLine("Son dəyər: " + numbers[0]); // 1_000_000 gözləyirik, amma...
        // Nəticə demək olar ki, həmişə 1_000_000-dan az olacaq!
    }

    static void IncrementNumbers(int count)
    {
        for (int i = 0; i < count; i++)
        {
            // Bu əməliyyat "numbers[0]++" əslində 3 addımdan ibarətdir:
            // 1. numbers[0]-ı oxu
            // 2. dəyəri 1 artır
            // 3. yeni dəyəri numbers[0]-a yaz
            numbers[0]++;
        }
    }
}

Niyə bu yarışdır? Əgər iplik A numbers[0]-ı (dəyəri 0) oxuyub, sonra iplik B də numbers[0]-ı (yenə 0) A yazmadan öncə oxuyarsa, hər iki iplik 0-ı 1-ə çevirib 1 yazacaq. Bir inkrement itəcək. numbers[0]++ atomik deyil.

Nümunə 2: InvalidOperationException Dictionary-də dəyişiklik zamanı

Bir iplik sözlüyü iterasiya edir, digəri isə onu dəyişdirir.

using System.Collections.Generic;
using System.Threading; // Thread.Sleep üçün

class DictionaryRaceExample
{
    static Dictionary<int, string> users = new Dictionary<int, string>();

    static void Main(string[] args)
    {
        // Sözlüyün ilkinləşdirilməsi
        for (int i = 0; i < 5; i++) users.Add(i, $"User {i}");

        // Oxuyan iplik
        Thread readerThread = new Thread(() =>
        {
            try
            {
                foreach (var user in users) // Sözlüyü iterasiya edirik
                {
                    Console.WriteLine($"Oxucu: {user.Key} - {user.Value}");
                    Thread.Sleep(10); // İşin simulyasiyası
                }
            }
            catch (InvalidOperationException ex)
            {
                Console.WriteLine($"Oxucu: XƏTA! {ex.Message}");
            }
        });

        // Yazıcı iplik
        Thread writerThread = new Thread(() =>
        {
            Thread.Sleep(5); // Oxucuya bir az başlamaq üçün vaxt veririk
            for (int i = 5; i < 10; i++)
            {
                users.Add(i, $"New User {i}"); // Elementlər əlavə olunur
                Console.WriteLine($"Yazıcı: User {i} əlavə olundu");
                Thread.Sleep(15);
            }
        });

        readerThread.Start();
        writerThread.Start();

        readerThread.Join(); // İpliklərin bitməsini gözləyirik
        writerThread.Join();
        Console.WriteLine("Nümunə tamamlandı.");
    }
}

Niyə xəta baş verir? Dictionary<TKey, TValue> (və List<T>) sinxronizasiya olmadan müxtəlif ipliklər tərəfindən eyni vaxtda oxuma və yazma üçün nəzərdə tutulmayıb. Yazıcı iplik daxili strukturu dəyişdirəndə, oxuyan iplik artıq dəyişən məlumatlar üzərində foreach etməyə davam edir və bu da InvalidOperationException-a gətirib çıxarır.

2. Niyə sadə bloklamalar (lock) hər zaman optimal deyil?

"Hamısını lock ilə əhatə et" ideyası sadə görünür, amma öz dezavantajları var:

// Pis nümunə: çox geniş lock istifadə edilir
// (Yalnız demonstrasiya üçün, belə etməyin!)
static object _lock = new object();
static List<int> _sharedList = new List<int>();

void AddItem(int item)
{
    lock (_lock)
    {
        _sharedList.Add(item);
    }
}

int GetItemCount()
{
    lock (_lock)
    {
        return _sharedList.Count;
    }
}

• Performans (bottleneck): lock bütün kolleksiyaya girişini bloklayır. 100 iplik olduqda 99-u birinin bitməsini gözləyəcək, hətta əməliyyatlar birbaşa konflikt etmirsə belə.
• Çətinlik: kolleksiyanın istifadə olunduğu hər yerdə lock xatırlanmalıdır. Bir yerdə unutmaq — və yarış yenidən ortaya çıxır.
• Deadlock riski: müxtəlif obyektlər üzərində bir neçə lock asanlıqla deadlock-a gətirə bilər.
• Iteratorlar: foreach başqa iplik kolleksiyanı modifikasiya edərsə sizi qurtarmır.

Buna görə .NET-də xüsusi olaraq thread-safe kolleksiyalar təqdim edildi.

Atomik əməliyyatlar

Thread-safe kolleksiya — istifadəçinin xarici lock qoymadan bir neçə iplikdən eyni anda istifadə edildikdə düzgün işləməyi təmin edir. Əsas prinsiplərdən biri — atomik əməliyyatlar: əməliyyat ya tam baş verir, ya da ümumiyyətlə baş vermir — digər ipliklər "yarımçıq" vəziyyət görmür.

• Əlavə etmə, silmə, oxuma — sanki ardıcıl olaraq tək-tək icra olunur.
• İçəridə aşağı səviyyəli üsullar istifadə olunur: interlocked əməliyyatlar (Interlocked), Compare-And-Swap (CAS), incə granularlıqlarla lock-lar — ümumi kolleksiyanı bloklamaq əvəzinə.

3. System.Collections.Concurrent icmalı

System.Collections.Concurrent namespace-i çoxiplikli işləmək üçün sıfırdan yazılmış kolleksiyalar dəsti verir. Onların fəlsəfəsi — maksimal paralellik və minimum bloklama.

• Performans: nüvələrin sayının artması ilə yaxşı skalyasiya olunur.
• Sadəlik: hər əməliyyat ətrafında əl ilə lock qoymaq lazım deyil.
• Az səhvlər: əl ilə sinxronizasiya etməklə bağlı sinif problemləri azalır.
• Rəqabət üçün optimizasiya: eyni anda əlavə/silmə əməliyyatlarında effektiv işləyir.

4. Əsas siniflər

ConcurrentQueue<T> (thread-safe növbə)

Prinsip: FIFO — "ilk gələn, ilk çıxan". Ssenarilər: producer–consumer, logging, task queue-lar.

using System.Collections.Concurrent;

ConcurrentQueue<string> messageQueue = new ConcurrentQueue<string>();

void Producer() => messageQueue.Enqueue("Mesaj 1");

void Consumer()
{
    if (messageQueue.TryDequeue(out string message))
    {
        Console.WriteLine($"Emal edildi: {message}");
    }
    else
    {
        Console.WriteLine("Növbə boşdur.");
    }
}

ConcurrentStack<T> (thread-safe stek)

Prinsip: LIFO — "son gələn, ilk çıxan". Ssenarilər: əməliyyat tarixçəsi, DFS gəzintisi, obyekt hovuzları.

using System.Collections.Concurrent;

ConcurrentStack<int> historyStack = new ConcurrentStack<int>();

void PushAction(int value) => historyStack.Push(value);

void PopAction()
{
    if (historyStack.TryPop(out int action))
    {
        Console.WriteLine($"Əməliyyat geri alındı: {action}");
    }
    else
    {
        Console.WriteLine("Stek boşdur.");
    }
}

ConcurrentBag<T> (thread-safe "çanta")

Sırasız kolleksiya, sıra təmin olunmur. "Hər iplik tez-tez öz qoyduğunu götürür" ssenarisinə optimizasiya edilib. Obyekt hovuzları üçün əladır.

using System.Collections.Concurrent;

ConcurrentBag<System.Guid> objectPool = new ConcurrentBag<System.Guid>();

void AddObject() => objectPool.Add(System.Guid.NewGuid());

void TakeObject()
{
    if (objectPool.TryTake(out System.Guid obj))
    {
        Console.WriteLine($"Obyekt götürüldü: {obj}");
    }
    else
    {
        Console.WriteLine("Hovuz boşdur.");
    }
}

ConcurrentDictionary<TKey, TValue> (thread-safe sözlük)

Açar üzrə atomik əlavə, yeniləmə və dəyər əldə etmə əməliyyatlarını dəstəkləyir. Keşlər, sessiyalar, sayğaclar üçün uyğundur.

using System.Collections.Concurrent;

ConcurrentDictionary<string, int> userScores = new ConcurrentDictionary<string, int>();

void UpdateScore(string user, int score)
{
    // Atomik olaraq əlavə edəcək, yoxdursa, və ya varsa yeniləyəcək
    userScores.AddOrUpdate(user, score, (key, existingVal) => existingVal + score);
    Console.WriteLine($"Bal {user}: {userScores[user]}");
}

void GetScore(string user)
{
    if (userScores.TryGetValue(user, out int score))
    {
        Console.WriteLine($"{user}-in cari balı: {score}");
    }
    else
    {
        Console.WriteLine($"{user} tapılmadı.");
    }
}

5. Bu kolleksiyaları adi kolleksiyalar əvəzinə nə zaman istifadə etmək lazımdır?

• Tətbiq çoxipliklidirsə: tək iplikli tətbiqdə adi kolleksiyalar daha sürətlidir (overhead yoxdur).
• Bir kolleksiya bir neçə iplik tərəfindən paylaşılıdırsa: bu, System.Collections.Concurrent-dan istifadə üçün əsas göstəricidir.
• Yüksək performans və skalyasiya lazımdırsa: kolleksiyalar minimum gözləmə üçün dizayn edilib.
• Kodu sadələşdirmək istəyirsinizsə: hər əməliyyat ətrafında əl ilə lock yazmağa ehtiyac qalmaz.
• Atomik əməliyyatlara ehtiyac varsa: əlavə/silmə/əldə etmə kolleksiyanı qeyri-konsistent vəziyyətə salmayacaq.

Concurrent-kolleksiyaları istifadə etməyin:

• Tətbiq tamamilə tək ipliklidirsə.
• Bir neçə əlaqəli əməliyyatın "transactional" olması tələb olunursa (bəlkə xarici sinxronizasiya və ya başqa mexanizmlər lazım olacaq).
• Sıra tələb olunan yer varsa və o zəmanət olunmursa (məsələn, ConcurrentBag<T> üçün sıra zəmanət olunmur).