C# lock: תנאי מרוץ, Interlocked ו-deadlock

כשכמה תהליכונים (threads) קוראים וכותבים את אותם נתונים באותו זמן, עדכונים עלולים ללכת לאיבוד או להיראות חצי גמורים. ההוראה lock הופכת בלוק קוד לבלעדי: כל עוד תהליכון אחד נמצא בתוכו, כל תהליכון אחר שמגיע ל-lock על אותו אובייקט מחכה לתורו.

הבעיה: תנאי מרוץ

ארבע משימות מוסיפות כל אחת 100,000 למונה משותף. התוצאה אמורה להיות 400,000:

using System;
using System.Threading.Tasks;

class Program
{
    static int count = 0;

    static void Main()
    {
        Task[] workers = new Task[4];
        for (int w = 0; w < workers.Length; w++)
        {
            workers[w] = Task.Run(() =>
            {
                for (int i = 0; i < 100000; i++)
                {
                    count++;   // not thread safe
                }
            });
        }
        Task.WaitAll(workers);
        Console.WriteLine("Expected 400000, got " + count);
    }
}

פלט לדוגמה:

Expected 400000, got 245609

הסכום בדרך כלל יוצא נמוך מדי, ובכל הרצה בהפרש אחר (במחשב עם ליבה אחת הוא עשוי לפעמים לצאת נכון, וזה מה שהופך את הבאגים האלה לקשים לתפיסה). count++ נראה כמו פעולה אחת, אבל הוא שלוש: קריאת count, הוספת אחד וכתיבה חזרה. שני תהליכונים יכולים שניהם לקרוא 500, שניהם לחשב 501 ושניהם לכתוב 501, והגדלה אחת נעלמת.

הפתרון: lock

עטפו את הקריאה, השינוי והכתיבה ב-lock על אובייקט משותף:

using System;
using System.Threading.Tasks;

class Program
{
    static readonly object sync = new object();
    static int count = 0;

    static void Main()
    {
        Task[] workers = new Task[4];
        for (int w = 0; w < workers.Length; w++)
        {
            workers[w] = Task.Run(() =>
            {
                for (int i = 0; i < 100000; i++)
                {
                    lock (sync)
                    {
                        count++;
                    }
                }
            });
        }
        Task.WaitAll(workers);
        Console.WriteLine("Expected 400000, got " + count);
    }
}

פלט:

Expected 400000, got 400000

עכשיו רק תהליכון אחד בכל רגע יכול להיות בתוך הבלוק, כך שכל הגדלה מסתיימת לפני שהבאה קוראת את הערך. הנעילה גם מבטיחה נראוּת: תהליכון שנכנס לנעילה רואה כל כתיבה שהמחזיק הקודם ביצע לפני שיצא.

ההגנה עובדת רק אם כל גישה ל-count עוברת דרך אותה נעילה. count++ אחד בלי נעילה במקום אחר בתוכנית מחזיר את תנאי המרוץ.

למה lock מתקמפל

lock הוא קיצור למחלקה Monitor, עם try/finally כדי שהנעילה תשתחרר גם כשהבלוק זורק חריגה:

lock (sync)
{
    count++;
}

// is compiled roughly as:
bool taken = false;
try
{
    Monitor.Enter(sync, ref taken);
    count++;
}
finally
{
    if (taken) Monitor.Exit(sync);
}

Monitor מציעה גם את TryEnter(obj, timeout), שמוותרת אחרי זמן קצוב במקום לחכות לנצח, ואת Wait/Pulse לאיתות בין תהליכונים. תהליכון שכבר מחזיק בנעילה יכול להיכנס אליה שוב (נעילות הן reentrant), כך שמתודה נעולה יכולה לקרוא למתודה נעולה אחרת על אותו אובייקט.

בחירת אובייקט הנעילה

האובייקט ב-lock (...) הוא רק אסימון שהתהליכונים מסכימים עליו. הכללים:

• השתמשו בשדה פרטי לקריאה בלבד מטיפוס object. private readonly object sync = new object();. פרטי אומר ששום קוד חיצוני לא יכול לקחת את אותה נעילה; readonly אומר שאי אפשר להחליף את האסימון בזמן שתהליכון מחזיק בו.
• לעולם לא lock (this). כל מי שמחזיק הפניה לאובייקט שלכם יכול לנעול עליו גם, ואז הקוד שלו חוסם את שלכם או נכנס איתו ל-deadlock.
• לעולם אל תנעלו על מחרוזת או על typeof(...). ליטרלים של מחרוזות עוברים interning (כל "orders" בתהליך הוא אותו אובייקט), ואובייקטים מסוג Type משותפים לכל האפליקציה, כך שקוד שאין לו קשר עלול להתחרות על אותה נעילה.
• לעולם אל תנעלו על טיפוס ערך. lock (count) על int לא מתקמפל (שגיאה CS0185), ו-boxing ידני יוצר אובייקט חדש בכל פעם, כך שהוא לא היה נועל כלום.
• השתמשו בנעילה אחת לכל קבוצת נתונים שחייבת להישאר עקבית: שדה נעילה static לנתונים סטטיים, ושדה מופע לנתונים של כל מופע.

הגנה על פעולות מרובות שלבים

lock נחוץ במיוחד כשבדיקה ופעולה חייבות לקרות יחד. חשבון בטוח לתהליכונים מדגים גם את תבנית ה"בדוק ואז פעל" וגם תמונת מצב עקבית של שני שדות:

using System;
using System.Threading.Tasks;

class Account
{
    private readonly object sync = new object();
    private decimal balance;
    private int withdrawals;

    public Account(decimal opening) { balance = opening; }

    public bool TryWithdraw(decimal amount)
    {
        lock (sync)
        {
            if (balance < amount) return false; // check...
            balance -= amount;                  // ...and act, with no gap between
            withdrawals++;
            return true;
        }
    }

    public string Summary()
    {
        lock (sync)
        {
            return balance + " left after " + withdrawals + " withdrawals";
        }
    }
}

class Program
{
    static void Main()
    {
        var acct = new Account(1000m);

        // 50 concurrent attempts to withdraw 30 each: only 33 can succeed.
        Parallel.For(0, 50, i => acct.TryWithdraw(30m));

        Console.WriteLine(acct.Summary());
    }
}

פלט:

10 left after 33 withdrawals

בלי הנעילה, שני תהליכונים יכלו שניהם לראות יתרה של 40, שניהם לעבור את הבדיקה ושניהם למשוך 30, כך שהחשבון היה נשאר במינוס 20. גם המתודה Summary נועלת, כך שהיא אף פעם לא מדווחת על balance מאחרי משיכה יחד עם ספירת withdrawals מלפניה.

גם Dictionary<TKey, TValue> ו-List<T> אינם בטוחים לתהליכונים: כתיבות מקבילות יכולות לשבש את המערכים הפנימיים שלהם, ולא רק לאבד עדכונים. הגנו עליהם בנעילה או השתמשו ב-ConcurrentDictionary<TKey, TValue> ובשאר הטיפוסים ב-System.Collections.Concurrent.

Interlocked לערכים בודדים

כשהמצב המשותף הוא מספר שלם אחד או הפניה אחת והעדכון הוא צעד אחד, המחלקה Interlocked מבצעת אותו באופן אטומי בלי נעילה:

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static int visits = 0;
    static long bytesServed = 0;
    static int peak = 0;

    static void Main()
    {
        Parallel.For(0, 10000, i =>
        {
            int now = Interlocked.Increment(ref visits);   // returns the new value
            Interlocked.Add(ref bytesServed, 512);

            // Record a new maximum with compare-and-swap.
            int seen;
            do
            {
                seen = peak;
                if (now <= seen) break;
            } while (Interlocked.CompareExchange(ref peak, now, seen) != seen);
        });

        Console.WriteLine("Visits: " + visits);
        Console.WriteLine("Bytes: " + bytesServed);
        Console.WriteLine("Peak: " + peak);
    }
}

פלט:

Visits: 10000
Bytes: 5120000
Peak: 10000

Increment, Decrement, Add ו-Exchange ממופות לפעולות אטומיות של המעבד (פקודה אחת ב-x64). CompareExchange(ref location, newValue, expected) כותבת רק אם המיקום עדיין מכיל את expected, ומחזירה את מה שמצאה, וכך אפשר לבנות כל עדכון כלולאת ניסיון חוזר. כל דבר שנוגע בשני משתנים בבת אחת עדיין צריך lock.

Deadlock

Deadlock (קיפאון) קורה כששני תהליכונים מחזיקים כל אחד בנעילה שהשני צריך:

// Thread 1                          // Thread 2
lock (accountA)                      lock (accountB)
{                                    {
    lock (accountB) { /* ... */ }        lock (accountA) { /* ... */ }
}                                    }

אם תהליכון 1 לוקח את accountA בזמן שתהליכון 2 לוקח את accountB, כל אחד מהם מחכה אחר כך לנצח לשני. אין חריגה ואין זמן קצוב; התוכנית פשוט נתקעת. העברה בין שני חשבונות שנועלת קודם את "המקור" ואחר כך את "היעד" יוצרת בדיוק את זה כששתי העברות בכיוונים הפוכים רצות בו זמנית.

ההגנות המקובלות:

• נעלו בסדר גלובלי קבוע. בהעברה, נעלו קודם את החשבון עם המזהה הקטן יותר, לא משנה לאיזה כיוון הכסף זז.
• החזיקו נעילות לזמן קצר ובצעו עבודה איטית (קלט/פלט, לוגים, קריאות רשת) מחוץ להן.
• אל תקראו לקוד לא מוכר בזמן שאתם מחזיקים בנעילה: אירועים, callbacks ומתודות וירטואליות עשויים לקחת נעילות משלהם.
• השתמשו ב-Monitor.TryEnter עם זמן קצוב במקום שבו היתקעות גרועה יותר מכישלון.

אין await בתוך lock: השתמשו ב-SemaphoreSlim

await אסור בתוך בלוק lock (שגיאת קומפילציה CS1996). אחרי await המתודה עשויה להמשיך בתהליכון אחר, ונעילת Monitor חייבת להשתחרר על ידי התהליכון שלקח אותה. בקוד async, SemaphoreSlim עם מונה 1 מתפקד כנעילה שמתאימה ל-async:

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static readonly SemaphoreSlim gate = new SemaphoreSlim(1, 1);
    static readonly List<string> log = new List<string>();

    static async Task SaveAsync(string item)
    {
        await gate.WaitAsync();          // wait without blocking a thread
        try
        {
            await Task.Delay(20);        // awaiting while holding the gate is fine
            log.Add(item);
        }
        finally
        {
            gate.Release();
        }
    }

    static void Main()
    {
        var tasks = new List<Task>();
        for (int i = 1; i <= 5; i++) tasks.Add(SaveAsync("order " + i));
        Task.WhenAll(tasks).GetAwaiter().GetResult();
        Console.WriteLine(log.Count + " saved, one at a time");
    }
}

פלט:

5 saved, one at a time

תמיד צמדו את WaitAsync ל-Release בתוך finally, אחרת חריגה תשאיר את השער סגור לתמיד. SemaphoreSlim(3, 3) מכניס שלושה קוראים בבת אחת, וכך מגבילים את מספר הבקשות המקבילות ל-API עם מגבלת קצב.

System.Threading.Lock (.NET 9)

.NET 9 עם C# 13 מוסיפה טיפוס ייעודי, System.Threading.Lock. כשהאובייקט בהוראת lock הוא Lock, הקומפיילר משתמש ב-API המהיר יותר שלו, EnterScope, במקום ב-Monitor:

private readonly Lock sync = new Lock();

public void Add(decimal amount)
{
    lock (sync)          // uses Lock.EnterScope(), not Monitor
    {
        balance += amount;
    }
}

הכללים לבחירת האובייקט נשארים אותם כללים. בגרסאות קודמות, private readonly object היא הבחירה הנכונה.

טעויות נפוצות

• נעילה של חלק מהגישות ולא של כולן. כל קריאה וכתיבה של הנתונים המשותפים חייבת לקחת את אותה נעילה.
• lock (this), lock (typeof(X)), lock ("name"). קוד חיצוני יכול לקחת את אותה נעילה.
• ביצוע קלט/פלט איטי בתוך נעילה. כל שאר התהליכונים מחכים; שמרו על הבלוק קצר.
• לקיחת שתי נעילות בסדרים שונים במקומות שונים. ה-deadlock הקלאסי.
• שימוש ב-lock סביב await. זה לא מתקמפל; השתמשו ב-SemaphoreSlim.
• נעילה חדשה בכל קריאה. lock (new object()) לא מגן על כלום; האובייקט חייב להיות משותף.