Source code for structum_lab.plugins.dynaconf.features.locks

# src/structum_lab.plugins.dynaconf/locks.py
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: 2025 PythonWoods

"""Synchronization primitives for concurrent configuration access.

Provides a ReadWriteLock to allow multiple concurrent readers but exclusive writers,
optimizing performance for read-heavy configuration workloads.
"""

import threading
from collections.abc import Generator
from contextlib import contextmanager



[docs]
class ReadWriteLock:
    """A lock allowing multiple simultaneous readers but exclusive access for writers.

    This implementation gives priority to writers to prevent starvation.
    """


[docs]
    def __init__(self) -> None:
        """Initialize the read-write lock."""
        self._readers = 0
        self._writers = 0
        self._read_ready = threading.Condition(threading.RLock())
        self._write_ready = threading.Condition(threading.RLock())



[docs]
    @contextmanager
    def read_lock(self) -> Generator[None, None, None]:
        """Acquire a read lock. Blocks if there are active writers."""
        with self._read_ready:
            while self._writers > 0:
                self._read_ready.wait()
            self._readers += 1

        try:
            yield
        finally:
            with self._read_ready:
                self._readers -= 1
                if self._readers == 0:
                    with self._write_ready:
                        self._write_ready.notify_all()



[docs]
    @contextmanager
    def write_lock(self) -> Generator[None, None, None]:
        """Acquire a write lock. Blocks until all readers and writers release."""
        with self._write_ready:
            while self._writers > 0 or self._readers > 0:
                self._write_ready.wait()
            self._writers += 1

        try:
            yield
        finally:
            with self._write_ready:
                self._writers -= 1
                self._write_ready.notify_all()
                with self._read_ready:
                    self._read_ready.notify_all()