Câu lệnh lock trong C#: race condition, Interlocked và deadlock

Khi nhiều luồng (thread) cùng đọc và ghi một dữ liệu tại cùng thời điểm, các cập nhật có thể bị mất hoặc bị thấy khi mới làm được một nửa. Câu lệnh lock khiến một khối code trở nên loại trừ lẫn nhau: khi một luồng đang ở bên trong, mọi luồng khác gặp lock trên cùng đối tượng phải chờ đến lượt.

Vấn đề: race condition

Bốn task, mỗi task cộng 100.000 vào một bộ đếm dùng chung. Kết quả đúng phải là 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);
    }
}

Ví dụ đầu ra:

Expected 400000, got 245609

Tổng thường bị thiếu, và thiếu một lượng khác nhau mỗi lần chạy (trên máy một nhân đôi khi nó lại đúng, chính điều đó khiến loại bug này khó bắt). count++ trông như một thao tác nhưng thực ra là ba: đọc count, cộng một, ghi lại. Hai luồng có thể cùng đọc được 500, cùng tính ra 501 và cùng ghi 501, thế là mất một lần tăng.

Cách sửa: lock

Bọc thao tác đọc, sửa, ghi trong một lock trên một đối tượng dùng chung:

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);
    }
}

Đầu ra:

Expected 400000, got 400000

Giờ chỉ một luồng được ở trong khối tại mỗi thời điểm, nên mỗi lần tăng hoàn tất trước khi lần sau đọc giá trị. Khóa còn đảm bảo tính hiển thị: luồng vào khóa sẽ thấy mọi lần ghi mà luồng giữ khóa trước đã thực hiện trước khi rời đi.

Sự bảo vệ chỉ có tác dụng khi mọi truy cập vào count đều đi qua cùng một khóa. Chỉ cần một count++ không khóa ở đâu đó trong chương trình là race condition quay lại.

lock được biên dịch thành gì

lock là cách viết tắt cho lớp Monitor, kèm try/finally để khóa được nhả kể cả khi khối lệnh ném ngoại lệ:

lock (sync)
{
    count++;
}

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

Monitor còn có TryEnter(obj, timeout), bỏ cuộc sau một khoảng thời gian thay vì chờ mãi, và Wait/Pulse để báo hiệu giữa các luồng. Một luồng đang giữ khóa có thể vào lại chính khóa đó (khóa có tính reentrant), nên một phương thức có khóa có thể gọi một phương thức có khóa khác trên cùng đối tượng.

Chọn đối tượng khóa

Đối tượng trong lock (...) chỉ là một tín hiệu mà các luồng cùng thống nhất. Các quy tắc:

• Dùng một field private, chỉ đọc, kiểu object. private readonly object sync = new object();. Private nghĩa là không code bên ngoài nào lấy được cùng khóa; readonly nghĩa là tín hiệu không thể bị thay thế trong khi một luồng đang giữ nó.
• Đừng bao giờ lock (this). Bất kỳ ai có tham chiếu đến đối tượng của bạn cũng khóa được trên nó, và code của họ sẽ chặn code của bạn hoặc gây deadlock.
• Đừng bao giờ khóa trên chuỗi hoặc typeof(...). Chuỗi literal được intern (mọi "orders" trong tiến trình là cùng một đối tượng), và đối tượng Type được dùng chung trong toàn ứng dụng, nên những đoạn code không liên quan có thể tranh nhau cùng một khóa.
• Đừng bao giờ khóa trên kiểu giá trị. lock (count) trên một int không biên dịch được (lỗi CS0185), còn tự boxing thì mỗi lần lại tạo một đối tượng mới, nên chẳng khóa được gì.
• Dùng một khóa cho mỗi nhóm dữ liệu cần giữ nhất quán, một field khóa static cho dữ liệu tĩnh, một field instance cho dữ liệu riêng của từng instance.

Bảo vệ thao tác nhiều bước

lock cần thiết nhất ở nơi một phép kiểm tra và một hành động phải diễn ra cùng nhau. Một tài khoản thread safe cho thấy cả mẫu kiểm tra rồi hành động lẫn ảnh chụp nhất quán của hai field:

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());
    }
}

Đầu ra:

10 left after 33 withdrawals

Không có khóa, hai luồng có thể cùng thấy số dư 40, cùng vượt qua phép kiểm tra và cùng rút 30, để lại tài khoản âm 20. Phương thức Summary cũng khóa, nên nó không bao giờ báo balance sau một lần rút cùng với số withdrawals trước lần rút đó.

Dictionary<TKey, TValue> và List<T> cũng không thread safe: các lần ghi đồng thời có thể làm hỏng mảng bên trong của chúng, chứ không chỉ làm mất cập nhật. Hãy bảo vệ chúng bằng khóa hoặc dùng ConcurrentDictionary<TKey, TValue> và các kiểu khác trong System.Collections.Concurrent.

Interlocked cho giá trị đơn lẻ

Khi trạng thái dùng chung là một số nguyên hoặc một tham chiếu và việc cập nhật chỉ có một bước, lớp Interlocked thực hiện nó một cách nguyên tử mà không cần khóa:

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);
    }
}

Đầu ra:

Visits: 10000
Bytes: 5120000
Peak: 10000

Increment, Decrement, Add và Exchange tương ứng với các thao tác nguyên tử của bộ xử lý (một lệnh duy nhất trên x64). CompareExchange(ref location, newValue, expected) chỉ ghi nếu vị trí đó vẫn đang giữ expected, và trả về giá trị nó tìm thấy, nhờ vậy bạn có thể xây mọi phép cập nhật thành một vòng lặp thử lại. Mọi thứ chạm đến hai biến cùng lúc vẫn cần lock.

Deadlock

Deadlock xảy ra khi hai luồng, mỗi luồng giữ một khóa mà luồng kia cần:

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

Nếu luồng 1 lấy accountA trong khi luồng 2 lấy accountB, thì sau đó mỗi luồng chờ luồng kia mãi mãi. Không có ngoại lệ, không có timeout; chương trình treo. Một giao dịch chuyển tiền giữa hai tài khoản khóa "from" rồi đến "to" sẽ gây ra đúng tình huống này khi hai giao dịch ngược chiều chạy cùng lúc.

Các cách phòng vệ chuẩn:

• Khóa theo một thứ tự toàn cục cố định. Với giao dịch chuyển tiền, khóa tài khoản có ID nhỏ hơn trước, bất kể tiền đi theo chiều nào.
• Giữ khóa thật ngắn và làm các việc chậm (I/O, ghi log, gọi mạng) bên ngoài khóa.
• Không gọi code không rõ nguồn gốc khi đang giữ khóa: sự kiện, callback và phương thức virtual có thể tự lấy khóa riêng.
• Dùng Monitor.TryEnter với timeout ở nơi chương trình bị treo còn tệ hơn một lần thất bại.

Không có await trong lock: dùng SemaphoreSlim

await không được phép trong khối lock (lỗi biên dịch CS1996). Sau một await, phương thức có thể tiếp tục trên một luồng khác, mà khóa Monitor phải được nhả bởi chính luồng đã lấy nó. Với code bất đồng bộ, SemaphoreSlim với số lượng 1 hoạt động như một khóa tương thích với 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");
    }
}

Đầu ra:

5 saved, one at a time

Luôn ghép WaitAsync với Release trong một finally, nếu không một ngoại lệ sẽ khiến cổng đóng vĩnh viễn. Một SemaphoreSlim(3, 3) cho ba bên gọi vào cùng lúc, đó là cách giới hạn số request đồng thời tới một API có giới hạn tốc độ.

System.Threading.Lock (.NET 9)

.NET 9 cùng C# 13 bổ sung kiểu System.Threading.Lock chuyên dụng. Khi đối tượng trong câu lệnh lock là một Lock, trình biên dịch dùng API EnterScope nhanh hơn của nó thay cho Monitor:

private readonly Lock sync = new Lock();

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

Quy tắc chọn đối tượng vẫn như cũ. Trên các phiên bản trước, private readonly object là lựa chọn đúng.

Lỗi thường gặp

• Khóa một số lần truy cập nhưng không phải tất cả. Mọi lần đọc và ghi dữ liệu dùng chung phải lấy cùng một khóa.
• lock (this), lock (typeof(X)), lock ("name"). Code bên ngoài có thể lấy cùng khóa đó.
• Làm I/O chậm bên trong khóa. Mọi luồng khác đều phải chờ; hãy giữ khối lệnh thật ngắn.
• Lấy hai khóa theo thứ tự khác nhau ở những chỗ khác nhau. Deadlock kinh điển.
• Dùng lock bao quanh await. Không biên dịch được; hãy dùng SemaphoreSlim.
• Mỗi lần gọi một khóa mới. lock (new object()) không bảo vệ gì cả; đối tượng phải được dùng chung.