Bài 50: Distributed Semaphore — Giới Hạn Concurrency

• Hiểu distributed semaphore là gì và khi nào cần dùng thay vì lock đơn giản.
• Giải thích bẫy rò rỉ của INCR/DECR semaphore khi client crash giữa chừng.
• Implement semaphore đúng bằng Sorted Set với timeout per holder.
• Viết Lua script atomic acquire và Python class DistributedSemaphore hoàn chỉnh.
• Dùng context manager để đảm bảo release trong mọi trường hợp (kể cả exception).
• Biết fair semaphore là gì và khi nào cần xét đến.
• Nhận diện vấn đề process pause và cách giảm thiểu.
• Phân biệt distributed semaphore với local connection pool.
• Biết các metric cần monitor và anti-pattern cần tránh.

Các bài trước xây distributed lock — đúng 1 client giữ lock tại một thời điểm. Nhưng nhiều tài nguyên thực tế không bị ràng buộc nghiêm ngặt như vậy. Ví dụ:

• Legacy database có connection pool size = 10. Không phải "1 client dùng", mà là "tối đa 10 client dùng đồng thời". Client thứ 11 trở đi phải chờ.
• Một 3rd-party API giới hạn 5 concurrent request (concurrent — không phải rate). Gọi lần thứ 6 song song bị trả về 429 hoặc lỗi.
• GPU inference server chỉ chịu được 20 concurrent job trước khi OOM.
• LLM provider đặt giới hạn concurrent request per API key.

Lock giải quyết N=1 (mutual exclusion). Semaphore tổng quát hóa: N slot, tối đa N holder đồng thời, client thứ N+1 phải chờ cho đến khi có slot trống.

Khái niệm semaphore (E.W. Dijkstra, 1965) ban đầu là biến đếm nguyên với hai thao tác P (wait/acquire, giảm counter) và V (signal/release, tăng counter). Distributed semaphore áp dụng ý tưởng đó xuyên qua nhiều process và nhiều máy, dùng Redis làm coordination point.

Ba primitive hay bị nhầm lẫn vì đều "giới hạn" thứ gì đó:

Điểm khác nhau cốt lõi: semaphore giới hạn concurrent — bao nhiêu request đang chạy tại cùng một thời điểm. Rate limit giới hạn throughput — bao nhiêu request được thực hiện trong một khoảng thời gian.

Một request kéo dài 10 giây chiếm 1 semaphore slot trong suốt 10 giây đó. Cùng request đó chỉ tốn 1 unit trong sliding window rate limiter (bài 34), bất kể nó mất bao lâu. Ngược lại: 100 request mỗi cái chỉ mất 1ms có thể đồng loạt pass semaphore N=100 mà không vi phạm gì, nhưng sẽ bị rate limiter 50 req/s chặn lại.

Trên thực tế đôi khi cần cả hai: rate limit để kiểm soát throughput tổng, semaphore để kiểm soát concurrent trong từng instant. Bài này chỉ tập trung vào semaphore.

Cách tiếp cận đầu tiên ai cũng nghĩ đến: dùng counter đơn giản.

# SAI — INCR/DECR không an toàn cho semaphore

N = 10  # tối đa 10 concurrent

# Acquire
count = redis.incr("sem:db-pool")
if count > N:
    redis.decr("sem:db-pool")  # trả lại
    raise NoSlotAvailable

# ... dùng resource ...

# Release
redis.decr("sem:db-pool")

Đoạn code này có vấn đề nghiêm trọng: nếu client crash, bị kill, hoặc exception ném ra giữa phần "dùng resource" trước khi đến decr, counter sẽ không bao giờ được giảm. Theo thời gian, counter tích lũy giá trị cao hơn thực tế — semaphore mất slot một cách vĩnh viễn. Trong hệ thống chạy liên tục, chỉ cần vài sự cố nhỏ là toàn bộ pool bị "rò rỉ" hết.

Một vấn đề khác: check-then-act (incr + kiểm tra count > N + decr nếu vượt) không atomic. Giữa incr và kiểm tra có thể có race condition, mặc dù với INCR pattern này ít nguy hiểm hơn nhưng vẫn không đáng tin cậy.

Vấn đề cốt lõi: counter không biết ai đang giữ slot, nên không có cơ chế timeout per holder. INCR/DECR semaphore không phù hợp cho distributed systems.

Thay vì counter vô danh, dùng Sorted Set lưu từng holder cụ thể:

• Mỗi holder = 1 member trong ZSet, với score = timestamp acquire (milliseconds).
• Holder id là UUID duy nhất — giống token trong lock.
• Acquire: trước tiên xóa các holder đã hết hạn (score < now - timeout), sau đó đếm holder còn lại, nếu số lượng < N thì thêm mình vào.
• Release: ZREM member của mình.
• Holder crash không release: entry tồn tại trong ZSet nhưng score cũ → bị ZREMRANGEBYSCORE dọn ở acquire kế tiếp → slot tự phục hồi.

# ZSet "sem:llm-api" khi có 3 holder đang active
ZADD sem:llm-api 1716883200000 "uuid-aaa"   # acquire lúc T=0
ZADD sem:llm-api 1716883201500 "uuid-bbb"   # acquire lúc T=1.5s
ZADD sem:llm-api 1716883203000 "uuid-ccc"   # acquire lúc T=3s

ZCARD sem:llm-api  # => 3 (slot đang dùng)

# Giả sử timeout = 30s, now = T+35s
# uuid-aaa hết hạn (35s > 30s), bị ZREMRANGEBYSCORE dọn
# => chỉ còn 2 holder, slot trống

Điểm khác so với bài 34 (sliding window log): bài 34 dùng ZSet để đếm request trong cửa sổ thời gian (rate limiting). Bài này dùng ZSet để theo dõi holder đang active (concurrency limiting). Cấu trúc tương tự nhưng mục đích và logic acquire khác nhau.

Toàn bộ acquire (dọn hết hạn + đếm + thêm holder) phải atomic. Nếu không, giữa ZCARD và ZADD có thể có N client khác chen vào — tất cả thấy count < N và tất cả được thêm vào, vượt quá giới hạn. Lua script chạy single-threaded và không bị interrupt:

-- sem_acquire.lua
-- KEYS[1] = semaphore key (vd "sem:llm-api")
-- ARGV[1] = now (ms epoch)
-- ARGV[2] = timeout (ms) — holder tồn tại tối đa bao lâu
-- ARGV[3] = limit (N) — số slot tối đa
-- ARGV[4] = holder_id (UUID)

local now     = tonumber(ARGV[1])
local timeout = tonumber(ARGV[2])
local limit   = tonumber(ARGV[3])

-- Bước 1: Dọn holder hết hạn (score < now - timeout)
redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, now - timeout)

-- Bước 2: Đếm holder còn active
local count = redis.call('ZCARD', KEYS[1])

-- Bước 3: Nếu còn slot, thêm holder này
if count < limit then
    redis.call('ZADD', KEYS[1], now, ARGV[4])
    -- Đặt TTL cho key, phòng khi mọi holder đều chết
    -- và không ai acquire lại (key bị bỏ lơ vĩnh viễn)
    redis.call('EXPIRE', KEYS[1], math.ceil(timeout / 1000) + 1)
    return 1  -- acquired
end

return 0  -- no slot available

Script release đơn giản hơn:

-- sem_release.lua
-- KEYS[1] = semaphore key
-- ARGV[1] = holder_id

return redis.call('ZREM', KEYS[1], ARGV[1])

Tại sao ZREMRANGEBYSCORE dùng now - timeout thay vì chỉ dùng TTL của key?

• TTL của key (qua EXPIRE) bảo vệ trường hợp key bị bỏ lơ.
• ZREMRANGEBYSCORE bảo vệ trường hợp holder riêng lẻ hết hạn trong khi key vẫn còn holder khác active. Nếu chỉ dùng TTL của key, holder "thây ma" sẽ ngồi trong ZSet cho đến khi key bị xóa.

import time
import uuid
import redis as redis_lib


ACQUIRE_LUA = """
local now     = tonumber(ARGV[1])
local timeout = tonumber(ARGV[2])
local limit   = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', KEYS[1], 0, now - timeout)

local count = redis.call('ZCARD', KEYS[1])

if count < limit then
    redis.call('ZADD', KEYS[1], now, ARGV[4])
    redis.call('EXPIRE', KEYS[1], math.ceil(timeout / 1000) + 1)
    return 1
end

return 0
"""

RELEASE_LUA = "return redis.call('ZREM', KEYS[1], ARGV[1])"


class DistributedSemaphore:
    """
    Distributed semaphore — tối đa `limit` holder đồng thời.
    Holder chết không release: tự dọn sau `timeout` giây.
    """

    def __init__(
        self,
        redis: redis_lib.Redis,
        name: str,
        limit: int,
        timeout: int = 30,
    ):
        """
        Args:
            redis:   redis client (redis-py >= 4.x)
            name:    tên semaphore, vd "llm-api"
            limit:   số slot tối đa (N)
            timeout: giây một holder được giữ slot tối đa
                     Phải > thời gian dùng resource hợp lệ dài nhất
        """
        self.redis = redis
        self.key = f"sem:{name}"
        self.limit = limit
        self.timeout_ms = timeout * 1000
        self._acquire = redis.register_script(ACQUIRE_LUA)
        self._release = redis.register_script(RELEASE_LUA)

    def acquire(self) -> str | None:
        """
        Cố gắng lấy 1 slot. Non-blocking.
        Returns: holder_id (UUID) nếu thành công, None nếu không còn slot.
        """
        holder = str(uuid.uuid4())
        now = int(time.time() * 1000)
        ok = self._acquire(
            keys=[self.key],
            args=[now, self.timeout_ms, self.limit, holder],
        )
        return holder if ok else None

    def release(self, holder: str) -> None:
        """Giải phóng slot. Safe to call nhiều lần (idempotent)."""
        self._release(keys=[self.key], args=[holder])

    def current_count(self) -> int:
        """Số slot đang bị giữ (bao gồm có thể cả holder hết hạn chưa dọn)."""
        return self.redis.zcard(self.key)


# Khởi tạo
r = redis_lib.Redis(host="localhost", port=6379, decode_responses=True)
llm_sem = DistributedSemaphore(r, name="llm-api", limit=20, timeout=60)

acquire() phía trên là non-blocking (thử một lần, thất bại thì trả về None). Trong practice, thường cần blocking retry: thử lại cho đến khi có slot hoặc hết thời gian chờ.

import time
from contextlib import contextmanager


class NoSlotAvailable(Exception):
    pass


@contextmanager
def semaphore(sem: DistributedSemaphore, blocking_timeout: float = 10.0, retry_interval: float = 0.1):
    """
    Context manager: acquire slot khi vào, release khi thoát (kể cả exception).

    Args:
        sem:              DistributedSemaphore instance
        blocking_timeout: giây tối đa chờ có slot (0 = non-blocking)
        retry_interval:   giây giữa các lần retry
    """
    holder = None
    deadline = time.monotonic() + blocking_timeout

    while True:
        holder = sem.acquire()
        if holder:
            break
        if time.monotonic() >= deadline:
            raise NoSlotAvailable(f"No slot in semaphore '{sem.key}' after {blocking_timeout}s")
        time.sleep(retry_interval)

    try:
        yield holder
    finally:
        sem.release(holder)


# Sử dụng
def process_request(prompt: str) -> str:
    with semaphore(llm_sem, blocking_timeout=30):
        # Đảm bảo tối đa 20 concurrent call tới LLM
        return call_llm_provider(prompt)  # type: ignore

Nếu call_llm_provider ném exception, finally vẫn chạy và release slot. Không cần try/except thủ công trong caller.

Với asyncio, thay time.sleep bằng await asyncio.sleep và dùng asynccontextmanager. Cấu trúc logic giữ nguyên.

Timeout per holder là cơ chế tự phục hồi slot quan trọng nhất. Nếu holder crash, bị kill, network partition — entry không bị xóa (không có release) nhưng score của nó không được cập nhật. Khi acquire kế tiếp gọi ZREMRANGEBYSCORE, entry quá hạn bị dọn và slot trở về trạng thái trống.

Chọn timeout cần cẩn thận:

Nguyên tắc: timeout phải lớn hơn thời gian dùng resource hợp lệ trong trường hợp chậm nhất. Nếu LLM call tệ nhất mất 45 giây, đặt timeout = 60+ giây.

Nếu cần giữ slot lâu hơn timeout cố định (ví dụ GPU job kéo dài không đoán được), implement renewal: một background thread/task gọi ZADD định kỳ để cập nhật score lên timestamp hiện tại — tương tự watchdog trong bài 44. Script renewal:

def renew(sem: DistributedSemaphore, holder: str) -> bool:
    """
    Cập nhật score của holder lên now để gia hạn.
    Trả về True nếu holder vẫn tồn tại trong ZSet (chưa bị dọn).
    """
    now = int(time.time() * 1000)
    # ZADD XX: chỉ update nếu member đã tồn tại (không tạo mới)
    result = sem.redis.zadd(sem.key, {holder: now}, xx=True)
    # zadd XX trả về số member được update; 0 nếu không tìm thấy
    return result is not None  # redis-py trả về số bản ghi update

Semaphore đã implement không đảm bảo fairness (thứ tự FIFO). Client nào retry nhanh hơn và trúng vào đúng lúc slot trống thì thắng, bất kể ai chờ lâu hơn. Điều này thường chấp nhận được vì:

• Phần lớn workload không yêu cầu strict ordering.
• Unfairness tự san bằng theo thống kê khi có nhiều client retry.

Khi cần strict FIFO, có thể dùng pattern queue + semaphore:

1. Mỗi client ghi vào một "waiting queue" ZSet (score = thời điểm xếp hàng).
2. Khi slot trống, chỉ client có score nhỏ nhất (chờ lâu nhất) được phép acquire.
3. Cần atomic check "tôi có phải đầu hàng không + acquire semaphore" — phức tạp hơn và thêm độ trễ.

Trên thực tế, fair semaphore hiếm khi cần thiết trong các hệ thống AI/backend thông thường. Chấp nhận unfair và ưu tiên đơn giản hóa là lựa chọn hợp lý.

Vấn đề process pause cũng xảy ra với semaphore, giống như với lock (bài 45). Kịch bản:

1. Client A acquire semaphore, đang dùng resource.
2. Client A bị pause (GC stop-the-world, container throttling, debug breakpoint) trong một khoảng thời gian dài hơn timeout.
3. Entry của A bị ZREMRANGEBYSCORE dọn → slot được cấp cho Client B.
4. Client A wake up, tiếp tục dùng resource, không biết slot đã bị thu hồi → có N+1 client đang dùng.

Không có giải pháp hoàn hảo cho distributed systems. Các biện pháp giảm thiểu:

• Timeout đủ lớn: chọn timeout vượt trội so với GC pause thực tế trong môi trường production. Với JVM, 99.9th percentile GC pause thường < 1 giây; với Python CPython hiếm khi pause quá vài trăm milliseconds.
• Renewal (watchdog): A liên tục gia hạn nên entry không bị dọn khi còn active.
• Idempotent operation: operation được bảo vệ bởi semaphore phải idempotent và có thể chịu đựng vài concurrent request vượt N mà không gây hư hỏng dữ liệu — concurrency limit lúc đó là "soft limit" chứ không phải correctness requirement.
• Fencing token: với resource ngoài (vd database), gắn fencing token để reject request đến trễ từ holder "thây ma" — tương tự bài 45.

Connection pool của thư viện (SQLAlchemy pool, asyncpg pool, aiohttp connector) là local per instance. Nếu bạn có 10 instance mỗi cái có pool size = 10, tổng connection tới database lên đến 100 — bất kể bạn muốn giới hạn ở mức nào.

Kiến trúc local pool:
  Instance 1: pool_size=10 ─┐
  Instance 2: pool_size=10 ─┼──> Legacy DB (max_connections=30)
  Instance 3: pool_size=10 ─┘
  Tổng: có thể mở 30 connections → đầy DB pool

Kiến trúc distributed semaphore:
  Instance 1 ─┐
  Instance 2 ─┼──> Redis "sem:db" (limit=20) ──> Legacy DB
  Instance 3 ─┘
  Tổng concurrent query tới DB: luôn ≤ 20

Distributed semaphore phù hợp khi:

• Bạn cần global concurrency limit (không phải per-instance limit).
• Resource chia sẻ có hard limit nhỏ (legacy DB với max_connections thấp, external API).
• Số instance biến động động (auto-scaling) — không thể cố định pool size per instance.

Local pool vẫn hữu ích kết hợp: dùng semaphore để giới hạn global, dùng local pool để tái sử dụng connection. Pool size mỗi instance = semaphore_limit / expected_instance_count (ước lượng).

Semaphore không tự báo cáo trạng thái. Cần instrument chủ động:

import prometheus_client as prom

# Metrics
sem_current = prom.Gauge("semaphore_current_holders", "Slot đang bị giữ", ["name"])
sem_acquire_total = prom.Counter("semaphore_acquire_total", "Tổng lần acquire", ["name", "result"])
sem_wait_seconds = prom.Histogram("semaphore_wait_seconds", "Thời gian chờ có slot", ["name"])

@contextmanager
def semaphore_monitored(sem: DistributedSemaphore, blocking_timeout: float = 10.0):
    start = time.monotonic()
    holder = None
    deadline = start + blocking_timeout

    while True:
        holder = sem.acquire()
        if holder:
            sem_acquire_total.labels(name=sem.key, result="ok").inc()
            break
        if time.monotonic() >= deadline:
            sem_acquire_total.labels(name=sem.key, result="timeout").inc()
            raise NoSlotAvailable

        time.sleep(0.05)

    wait_time = time.monotonic() - start
    sem_wait_seconds.labels(name=sem.key).observe(wait_time)
    sem_current.labels(name=sem.key).set(sem.current_count())

    try:
        yield holder
    finally:
        sem.release(holder)
        sem_current.labels(name=sem.key).set(sem.current_count())

Những gì cần alert:

• Semaphore luôn full (ZCARD == limit trong thời gian dài): limit quá thấp hoặc holder bị leak.
• Acquire timeout rate cao: resource đang bị overwhelm, cần tăng limit hoặc scale resource.
• ZCARD tăng không ngừng mà không giảm: holder leak (có thể do renewal loop bị lỗi, hoặc client không release).
• Hold time p99 vượt timeout * 0.8: nguy cơ cao holder bị dọn nhầm khi đang active.

INCR/DECR semaphore

Counter không có timeout per holder. Client chết → counter rò rỉ → pool cạn dần vĩnh viễn. Không dùng trong distributed systems.

Không có timeout per holder

Dùng ZSet nhưng không ZREMRANGEBYSCORE holder hết hạn. Holder chết chiếm slot mãi mãi. Toàn bộ cơ chế tự phục hồi biến mất.

Non-atomic acquire

Gọi ZCARD rồi ZADD riêng biệt (không qua Lua). Race condition: N client cùng thấy count < limit, cùng ZADD, vượt quá N slot.

Holder id không unique

Dùng hostname hoặc process id cố định làm holder id. ZADD sẽ overwrite entry của holder cùng tên, gây mất tracking. Mỗi acquire cần UUID riêng.

timeout < thời gian dùng hợp lệ

Timeout quá ngắn → holder đang active bị dọn nhầm → có thể vượt N concurrent. Phải đo p99 hold time thực tế trước khi đặt timeout.

Quên release

Không dùng try/finally hoặc context manager. Exception ném ra → slot không được trả → leak cho đến timeout. Luôn dùng context manager.

Semaphore limit cố định không đổi

Resource capacity thay đổi (ví dụ DB nâng cấp max_connections) nhưng semaphore limit không được cập nhật. Nên đọc limit từ config/env, không hardcode.

• Dùng Sorted Set với timeout: không phải INCR/DECR. Mỗi holder có identity và có thể bị dọn độc lập.
• Lua script atomic acquire: dọn hết hạn + đếm + thêm holder trong một thao tác không thể interrupt.
• Holder id UUID: unique per acquire, không tái sử dụng.
• timeout > p99 hold time + buffer: đo thực tế, thêm buffer an toàn. Với LLM call 10s p99, dùng timeout 60s.
• Renewal nếu cần giữ lâu: background task cập nhật score định kỳ. Dừng renewal trước khi release.
• try/finally release: luôn wrap trong context manager hoặc try/finally thủ công.
• Monitor holder count + acquire fail rate: alert khi luôn full hoặc khi holder count tăng bất thường.
• Đặt EXPIRE cho ZSet key: phòng trường hợp tất cả holder chết và không ai acquire lại — key vẫn bị xóa sau timeout thay vì tồn tại vĩnh viễn.

Module 4 (Distributed Coordination) đã đi qua đủ bộ coordination primitives cần thiết cho production:

Quiz

1. Tại sao INCR/DECR không phù hợp cho distributed semaphore? Vấn đề gì xảy ra khi client crash?
2. Vì sao acquire phải atomic (qua Lua script)? Điều gì xảy ra nếu ZCARD và ZADD không atomic?
3. Holder id cần là UUID mỗi lần acquire, không phải hostname cố định. Tại sao?
4. Giải thích sự khác nhau giữa semaphore N=20 và rate limit 20 req/s với một request kéo dài 10 giây.
5. Khi nào nên dùng distributed semaphore thay vì local connection pool?

Đáp án gợi ý

1. Counter không có timeout per holder. Client crash không DECR → counter tăng nhưng không giảm → semaphore mất slot vĩnh viễn. Sorted Set với timeout per holder tự dọn entry hết hạn ở acquire kế tiếp.
2. Giữa ZCARD (đếm) và ZADD (thêm holder) không atomic: N client cùng thấy count < limit, cùng ZADD, tổng holder vượt quá limit. Lua script chạy single-threaded, ZREMRANGEBYSCORE + ZCARD + ZADD là một thao tác không bị interrupt.
3. ZADD ghi đè entry nếu member đã tồn tại. Nếu hai acquire dùng cùng holder id (vd hostname), acquire thứ hai sẽ update score của acquire đầu (mà không biết đó là holder đang active của instance khác), phá vỡ tracking. UUID unique per acquire tránh hoàn toàn vấn đề này.
4. Semaphore N=20: tại mỗi thời điểm tối đa 20 request đang xử lý. 1 request 10 giây chiếm 1 slot trong 10 giây đó. Rate limit 20 req/s: tối đa 20 request mới được bắt đầu mỗi giây, bất kể chúng mất bao lâu. 20 request 10 giây mỗi cái: pass rate limit (1 request mỗi 0.5s) nhưng sau vài giây sẽ có 20 concurrent, slot đầy.
5. Khi cần global concurrency limit bất kể số instance: tổng concurrent tới resource luôn ≤ N, không phụ thuộc bao nhiêu instance đang chạy. Local pool chỉ giới hạn per instance, tổng tăng tuyến tính theo số instance.

Bài tiếp theo

Bài 51 đi vào ứng dụng thực tế của lock và leader election: Singleton Worker & Cron Deduplication — đảm bảo chỉ một worker chạy một cron job tại một thời điểm trong hệ thống multi-instance.

Tham khảo

• Redis Docs — ZADD command
• Redis Docs — ZREMRANGEBYSCORE command
• Redis Docs — ZCARD command
• Redis Docs — Lua scripting (EVAL)
• Redis in Action — Counting Semaphores (Josiah Carlson)
• redis-py — Redis client cho Python