This issue tracker has been migrated to GitHub, and is currently read-only.
For more information, see the GitHub FAQs in the Python's Developer Guide.

Title: Something like Rust's std::sync::Mutex – combining a mutex primitive and a piece of data it's protecting
Type: enhancement Stage:
Components: Library (Lib) Versions: Python 3.10
Status: open Resolution:
Dependencies: Superseder:
Assigned To: Nosy List: benjamin.peterson, jstasiak, pitrou
Priority: normal Keywords:

Created on 2020-12-07 15:56 by jstasiak, last changed 2022-04-11 14:59 by admin.

Messages (1)
msg382650 - (view) Author: Jakub Stasiak (jstasiak) * Date: 2020-12-07 15:56
I've been wondering if it's worth it to have something like Rust's std::sync::Mutex[1] which is used like this:

    let data = Mutex::new(0);
        let mut unlocked = data.lock().unwrap();
        *unlocked += 1;
    // unlocked is no longer available here, we need to use data.lock() again

Our (Python) [R]Lock is typically used like this:

    data_to_protect = whatever()
    lock = Lock()

    # ...
    with lock:

The inconvenience of this is obvious to me and it's more error prone if one forgets the lock when accessing data_to_protect. I wrote a quick prototype to get something like Mutex in Python:

    import threading
    from contextlib import contextmanager
    from typing import Any, cast, Dict, Generator, Generic, Optional, TypeVar

    T = TypeVar('T')

    class LockedData(Generic[T]):
        def __init__(self, data: T, lock: Any = None) -> None:
            self._data = data
            if lock is None:
                lock = threading.Lock()
            self._lock = lock

        def unlocked(self, timeout: float = -1.0) -> Generator[T, None, None]:
            acquired = None
            unlocked = None
                acquired = self._lock.acquire(timeout=timeout)
                if acquired is False:
                    raise LockTimeout()
                unlocked = UnlockResult(self._data)
                yield unlocked
                if acquired is True:
                    if unlocked is not None:
                        unlocked._unlocked = False
                        self._data = unlocked._data
                        unlocked._data = None

    class UnlockResult(Generic[T]):
        _data: Optional[T]

        def __init__(self, data: T) -> None:
            self._data = data
            self._unlocked = True

        def data(self) -> T:
            assert self._unlocked
            return cast(T, self._data)

        def data(self, data: T) -> None:
            assert self._unlocked
            self._data = data

    class LockTimeout(Exception):

    if __name__ == '__main__':
        locked_dict: LockedData[Dict[str, bool]] = LockedData({})

        # Mutating the dictionary
        with locked_dict.unlocked() as result:
  ['hello'] = True

        with locked_dict.unlocked() as result:

        # Replacing the dictionary
        with locked_dict.unlocked() as result:
   = {'a': True, 'b': False}

        with locked_dict.unlocked() as result:

        # Trying to access data after context closes

Now this is obviously quite far from what Rust offers, as there's nothing to prevent a person from doing something like this:

        with locked_dict.unlocked() as result:
            data =

        print('Oh no, look: %r' % (data,))
but it seems to me it's still an improvement.

Date User Action Args
2022-04-11 14:59:39adminsetgithub: 86756
2020-12-07 21:57:16vstinnersetnosy: - vstinner
2020-12-07 15:56:57jstasiakcreate