Hiện tượng Race Condition (Tranh chấp dữ liệu) xảy ra khi nhiều luồng (threads) cùng truy cập và thay đổi một vùng nhớ dùng chung (shared state) mà không có cơ chế đồng bộ hóa, dẫn đến kết quả cuối cùng phụ thuộc vào trật tự ngẫu nhiên của bộ lập lịch hệ điều hành (OS Scheduler).

Dưới đây là bài thực hành chuẩn mực giúp bạn mổ xẻ hiện tượng này ở cấp độ Bytecode và làm chủ cơ chế khóa với Lock và RLock.

1. Bản chất cơ học: Tại sao counter += 1 lại không an toàn?

Nhiều người lầm tưởng counter += 1 là một thao tác đơn nguyên (atomic operation). Nhưng ở cấp độ máy ảo Python (CPython Bytecode), phép tính này gồm 3 chỉ lệnh độc lập:

1. LOAD_GLOBAL     (Đọc giá trị hiện tại của counter từ RAM vào Stack)
2. BINARY_OP (+)   (Cộng thêm 1 vào giá trị trong Stack)
3. STORE_GLOBAL    (Ghi giá trị mới từ Stack ngược trở lại biến counter trong RAM)

Kịch bản xảy ra Race Condition giữa Thread A và Thread B:

• Giả sử counter = 100.

• Thread A đọc counter $\rightarrow$ được 100. Đang chuẩn bị cộng thì hết lượt CPU (Context Switch).

• Thread B được cấp CPU $\rightarrow$ đọc counter (vẫn là 100), cộng thành 101, và ghi đè counter = 101.

• Thread A được đánh thức dậy $\rightarrow$ tiếp tục thực hiện phép cộng trên giá trị cũ đã nạp vào Stack ($100 + 1 = 101$) $\rightarrow$ ghi đè counter = 101.

• Hậu quả: 2 lượt tăng biến đếm diễn ra nhưng counter chỉ tăng thêm 1 đơn vị. Lượt cập nhật của Thread B bị mất hoàn toàn (Lost Update).

2. Thực nghiệm 1: Chứng minh Race Condition bằng mã nguồn

Chạy kịch bản tạo 10 threads, mỗi thread tăng biến đếm lên $100.000$ lần.

• Kỳ vọng lý thuyết: $10 \times 100.000 = 1.000.000$.

• Thực tế chạy: Kết quả luôn nhỏ hơn $1.000.000$ và thay đổi sau mỗi lần chạy.

import threading
import time

# Vùng nhớ dùng chung (Shared State)
counter = 0
NUM_THREADS = 10
ITERATIONS = 100_000

def worker():
    global counter
    for _ in range(ITERATIONS):
        # Đoạn mã không được bảo vệ (Critical Section)
        counter += 1

print(f"Bắt đầu chạy với {NUM_THREADS} threads, mỗi thread lặp {ITERATIONS:,} lần...")
threads = []
start_time = time.time()

for _ in range(NUM_THREADS):
    t = threading.Thread(target=worker)
    threads.append(t)
    t.start()

# Chờ tất cả các thread chạy xong
for t in threads:
    t.join()

elapsed = time.time() - start_time
expected = NUM_THREADS * ITERATIONS

print("-" * 50)
print(f"Giá trị kỳ vọng : {expected:,}")
print(f"Giá trị thực tế : {counter:,}")
print(f"Số lượt bị mất  : {expected - counter:,} (Sai lệch: {((expected - counter)/expected)*100:.2f}%)")
print(f"Thời gian chạy  : {elapsed:.4f} giây")

Kết quả mẫu trên máy tính:

Bắt đầu chạy với 10 threads, mỗi thread lặp 100,000 lần...
--------------------------------------------------
Giá trị kỳ vọng : 1,000,000
Giá trị thực tế : 648,312
Số lượt bị mất  : 351,688 (Sai lệch: 35.17%)
Thời gian chạy  : 0.0812 giây

3. Thực nghiệm 2: Khắc phục triệt để bằng threading.Lock

threading.Lock (Mutual Exclusion Lock - Mutex) tạo ra một rào chắn: tại một thời điểm, chỉ duy nhất một thread được sở hữu khóa để bước vào Vùng xung yếu (Critical Section). Các thread khác muốn vào buộc phải chuyển sang trạng thái chờ (Blocked/Waiting).

Cách dùng chuẩn là kết hợp với Context Manager (with lock:) để tự động giải phóng khóa ngay cả khi code phát sinh ngoại lệ.

import threading
import time

counter_safe = 0
counter_lock = threading.Lock()

NUM_THREADS = 10
ITERATIONS = 100_000

def safe_worker():
    global counter_safe
    for _ in range(ITERATIONS):
        # Yêu cầu sở hữu khóa trước khi đọc/ghi
        with counter_lock:
            counter_safe += 1
        # Ra khỏi khối with: Khóa tự động được nhả (release)

threads = []
start_time = time.time()

for _ in range(NUM_THREADS):
    t = threading.Thread(target=safe_worker)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

elapsed = time.time() - start_time
expected = NUM_THREADS * ITERATIONS

print("-" * 50)
print(f"Giá trị kỳ vọng : {expected:,}")
print(f"Giá trị an toàn : {counter_safe:,}")
print(f"Thời gian chạy  : {elapsed:.4f} giây")

Kết quả:

--------------------------------------------------
Giá trị kỳ vọng : 1,000,000
Giá trị an toàn : 1,000,000  <-- CHÍNH XÁC TUYỆT ĐỐI!
Thời gian chạy  : 0.3541 giây

Lưu ý về hiệu năng: Thời gian chạy có Lock (0.35s) chậm hơn không Lock (0.08s) khoảng 4 lần do chi phí xin cấp khóa, nhả khóa và chuyển trạng thái luồng (Lock contention).

4. Khi nào cần RLock (Reentrant Lock)?

Một Lock tiêu chuẩn không có khả năng tái nhập: Nếu cùng một thread đã giữ khóa mà lại cố tình xin khóa thêm một lần nữa, nó sẽ tự chặn chính nó và gây ra hiện tượng Self-Deadlock (Treo ứng dụng vĩnh viễn).

threading.RLock giải quyết bài toán này: Nó ghi nhớ thread nào đang giữ khóa và số lần khóa đã được gọi (recursion level). Cùng một thread có thể acquire() nhiều lần liên tiếp mà không bị khóa cứng.

Kịch bản lỗi kinh điển với Lock:

import threading

lock = threading.Lock()

def step_b():
    with lock:
        print("Đang chạy Step B...")

def step_a():
    with lock:
        print("Đang chạy Step A...")
        step_b() # CHẾT ĐỨNG TẠI ĐÂY: step_b cố xin lock mà step_a chưa nhả!

# Lệnh này sẽ làm chương trình bị treo vô tận (Deadlock)
# step_a()

Xử lý an toàn với RLock:

import threading

# Đổi sang RLock
rlock = threading.RLock()

def step_b():
    with rlock:
        print("Đang chạy Step B...")

def step_a():
    with rlock:
        print("Đang chạy Step A...")
        step_b() # Hoàn toàn trơn tru vì cùng một thread sở hữu!

step_a()
print("Hoàn thành không bị Deadlock!")

Bảng tóm tắt quy tắc cốt lõi