Repository: gabrik/mvar-python

Issue:
Potentially dangerous behaviour
Hi there!

An `MVar` is one of the things I miss very much in python, thus I was happy to see there is a library addressing that. Thank you very much for that!
Unfortunately, I've encountered a few problems with the current implementation.

First, and most important: in a many-threads setting, your MVar may end up in a deadlock, because `self.__condition.notify()` wakes up only one thread at a time. If one thread puts something and wakes another "put"-thread rather than a "get"-thread, the "put" thread can either override the previously put value, or cause a deadlock. I'd suggest to use `self.__condition.notify_all` and `self.__condition.wait_for` instead.

Then, it's not really possible to put 'None' in the MVar at this point - it needs a separate 'is_full' condition.

And lastly, more of a feature request: would be nice to have two separate 'get'/'take' functions that actually reflect Haskell behavior.

I've ended up using the following implementation:
```python
class MVar:
    '''
    Haskell-style MVars.

    The context manager protocol behaves like 'get' function,
    but allows to keep the lock while performing some actions.
    The lock behind the MVar is recursive,
    thus you can use the other blocking operations under the scope
    of the 'with mvar' block.
    '''

    __lock: threading.RLock
    __condition: threading.Condition
    __full: bool
    __value: any

    def __init__(self):
        self.__lock = threading.RLock()
        self.__condition = threading.Condition(lock=self.__lock)
        self.__full = False
        self.__value = None

    def is_full(self) -> bool:
        '''Wether the value is there (non-blocking)'''
        return self.__full

    def is_empty(self) -> bool:
        '''Inverse of 'is_full' (non-blocking)'''
        return not self.__full

    def get(self):
        '''
        Wait for the value and get it without emptying the MVar (blocking).
        Does not cause other threads to wake up.
        '''
        with self.__lock:
            self.__condition.wait_for(self.is_full)
            return self.__value

    def take(self):
        '''
        Wait for the value and take it (blocking).
        Empties the MVar and wakes up all waiting threads.
        '''
        with self.__lock:
            self.__condition.wait_for(self.is_full)
            v = self.__value
            self.__value = None
            self.__full = False
            self.__condition.notify_all()
            return v

    def put(self, value):
        '''
        Wait for MVar to become empty and put a value in there (blocking).
        Wakes up all waiting threads.
        '''
        with self.__lock:
            self.__condition.wait_for(self.is_empty)
            self.__value = value
            self.__full = True
            self.__condition.notify_all()

    def __enter__(self):
        self.__lock.__enter__()
        self.__condition.wait_for(self.is_full)
        return self.__value

    def __exit__(self, *args):
        return self.__lock.__exit__(*args)

    @contextlib.contextmanager
    def wait(self, timeout=None):
        '''
        Wait for any changes in the state, or for the timeout event (blocking).

        returns wether it's awaken within the timeout time.
        '''
        with self.__lock:
            yield self.__condition.wait(timeout=timeout)
```

I guess, you have long forgotten about this project. But if you choose to use this implementation and upload it to pip - I would be happy to give away the copy right :)

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 82d11a0..087407b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,6 +1,9 @@
 # Changelog
 All notable changes to this project will be documented in this file.
 
+## [1.0.0] - 2021-02-03
+### Changed
+- **MVar** is redefined to match Haskell's implementation more closely
 
 ## [0.0.1] - 2018-12-21
 ### Added
@@ -13,11 +16,11 @@ All notable changes to this project will be documented in this file.
 ### Removed
 - Nothing
 
-        
-        
+
+
 
 
 #### Remarks
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
-and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
\ No newline at end of file
+and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
diff --git a/README.md b/README.md
index 9e3b0ec..0d149c4 100644
--- a/README.md
+++ b/README.md
@@ -7,10 +7,10 @@
 
 A Pyhton port of Haskell's [Control.Concurrent.MVar](https://hackage.haskell.org/package/base/docs/Control-Concurrent-MVar.html).
 
-This implementation blocks on get is the MVar is empty, and on put if the MVar is not empty.
+This implementation blocks on get if the MVar is empty, and on put if the MVar is not empty.
 
 A MVar is a mutable location which can either be empty, or contain a value.
-The location can be written to and read from safely from multiple concurrent Unix threads.
+The location can be written to and read from safely from multiple concurrent python threads.
 
 
 ### Installation
diff --git a/example/example.py b/example/example.py
index b6bd319..f2588e1 100644
--- a/example/example.py
+++ b/example/example.py
@@ -30,14 +30,14 @@ def worker2(var):
 print('Starting Worker1 and waiting on MVar the result...')
 local_var = MVar()
 Thread(target=worker, args=(local_var,), daemon=True).start()
-res = local_var.get()
+res = local_var.take()
 print("Worker1 returned {}".format(res))
 print("Putting a value on the MVar to block Worker2")
 local_var.put(2)
 print('Starting Worker2...')
 Thread(target=worker2, args=(local_var,), daemon=True).start()
-res2 = local_var.get()
+res2 = local_var.take()
 print("Getting previous value stored in MVar to"
-      "unlock Worker2 - MVar.get() returned {}".format(res2))
-res3 = local_var.get()
+      "unlock Worker2 - MVar.take() returned {}".format(res2))
+res3 = local_var.take()
 print("Worker3 returned {}".format(res3))
diff --git a/mvar/__init__.py b/mvar/__init__.py
index db15154..e8d94d1 100644
--- a/mvar/__init__.py
+++ b/mvar/__init__.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2018 Gabriele Baldoni.
+# Copyright (c) 2018 Gabriele Baldoni, Artem Chirkin.
 #
 # See the NOTICE file(s) distributed with this work for additional
 # information regarding copyright ownership.
@@ -10,6 +10,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 #
-# Contributors: Gabriele Baldoni MVar implementation in Python
+# Contributors: Gabriele Baldoni, Artem Chirkin
 
 from mvar.mvar import MVar
diff --git a/mvar/mvar.py b/mvar/mvar.py
index 8029cb9..fda9193 100644
--- a/mvar/mvar.py
+++ b/mvar/mvar.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2018 Gabriele Baldoni.
+# Copyright (c) 2018 Gabriele Baldoni, Artem Chirkin.
 #
 # See the NOTICE file(s) distributed with this work for additional
 # information regarding copyright ownership.
@@ -10,32 +10,139 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 #
-# Contributors: Gabriele Baldoni MVar implementation in Python
+# Contributors: Gabriele Baldoni, Artem Chirkin
+'''Haskell-style MVars using threading primitives.'''
+import contextlib
+import threading
+import typing
 
-from threading import Condition, Lock
+T = typing.TypeVar('T')
 
 
-class MVar(object):
+class MVar(typing.Generic[T]):
+    '''
+    Haskell-style MVars.
+
+    At its core functionality are methods `take()`, `get()`, and `put()`,
+    which closely reflect their haskell counterparts.
+
+    This implementation also adds a few more methods which enable a more
+    pythonic style of programming.
+    In particular, `wait()` and `locked()` by supporting the context manager protocol
+    allow you to perform atomic operations on the MVar using `with` syntax.
+    Using `with` syntax on MVar itself is equivalent to using `locked()` without arguments.
+    It is safe to nest `with` syntax on the same MVar (it uses a recursive lock).
+
+    For most of the functions, the documentation indicates whether they are blocking or not;
+    this refers to the underlying recursive lock.
+    '''
+
+    __lock: threading.RLock
+    __condition: threading.Condition
+    __full: bool
+    __value: typing.Optional[T]
 
     def __init__(self):
-        self.__lock = Lock()
-        self.__condition = Condition(lock=self.__lock)
+        self.__lock = threading.RLock()
+        self.__condition = threading.Condition(lock=self.__lock)
+        self.__full = False
         self.__value = None
 
-    def get(self):
-        self.__lock.acquire()
-        if self.__value is None:
-            self.__condition.wait()
-        v = self.__value
-        self.__value = None
-        self.__condition.notify()
-        self.__lock.release()
-        return v
-
-    def put(self, value):
-        self.__lock.acquire()
-        if self.__value is not None:
-            self.__condition.wait()
-        self.__value = value
-        self.__condition.notify()
-        self.__lock.release()
+    def __enter__(self) -> typing.Tuple[bool, typing.Optional[T]]:
+        self.__lock.__enter__()
+        return (self.__full, self.__value)
+
+    def __exit__(self, *args):
+        return self.__lock.__exit__(*args)
+
+    def is_full(self) -> bool:
+        '''Whether the value is there (non-blocking)'''
+        return self.__full
+
+    def is_empty(self) -> bool:
+        '''Inverse of 'is_full' (non-blocking)'''
+        return not self.__full
+
+    def get(self, timeout: typing.Optional[float] = None) -> T:
+        '''
+        Wait for the value and get it without emptying the MVar (blocking).
+        Does not cause other waiting threads to wake up.
+        '''
+        with self.__lock:
+            if not self.__condition.wait_for(self.is_full, timeout=timeout):
+                raise TimeoutError('MVar.get: timeout')
+            return self.__value
+
+    def take(self, timeout: typing.Optional[float] = None) -> T:
+        '''
+        Wait for the value and take it (blocking).
+        Empties the MVar and wakes up all waiting threads.
+        '''
+        with self.__lock:
+            if not self.__condition.wait_for(self.is_full, timeout=timeout):
+                raise TimeoutError('MVar.take: timeout')
+            v = self.__value
+            self.__value = None
+            self.__full = False
+            self.__condition.notify_all()
+            return v
+
+    def take_nowait(self) -> T:
+        '''
+        Try to take the value without waiting (blocking).
+        Empties the MVar and wakes up all waiting threads if the MVar was full.
+        Otherwise, throws the ValueError.
+        '''
+        with self.__lock:
+            if self.is_empty():
+                raise ValueError("MVar.take_nowait: empty")
+            v = self.__value
+            self.__value = None
+            self.__full = False
+            self.__condition.notify_all()
+            return v
+
+    def put(self, value: T, timeout: typing.Optional[float] = None) -> None:
+        '''
+        Wait for MVar to become empty and put a value in there (blocking).
+        Wakes up all waiting threads.
+        '''
+        with self.__lock:
+            if not self.__condition.wait_for(self.is_empty, timeout=timeout):
+                raise TimeoutError('MVar.put: timeout')
+            self.__value = value
+            self.__full = True
+            self.__condition.notify_all()
+
+    @contextlib.contextmanager
+    def wait(self, timeout: typing.Optional[float] = None) -> bool:
+        '''
+        Wait for any changes in the state, or for the timeout event (blocking).
+
+        Returns whether it's awaken within the specified time.
+        '''
+        with self.__lock:
+            yield self.__condition.wait(timeout=timeout)
+
+    @contextlib.contextmanager
+    def locked(self, timeout: typing.Optional[float] = None) -> typing.Tuple[bool, typing.Optional[T]]:
+        '''
+        Use the MVar's lock independently of the MVar's state.
+
+        Returns a tuple (is_full, Optional[value]).
+
+        Note, the type of the value `T` may itself be optional,
+        that is why this method returns the tuple with the explicit indication of whether it's full.
+        '''
+        if self.__lock.acquire(True, **({'timeout': timeout} if timeout is not None else {})):
+            try:
+                yield (self.__full, self.__value)
+            finally:
+                self.__lock.release()
+        else:
+            raise TimeoutError('MVar.locked: timeout')
+
+    def notify_all(self) -> None:
+        '''Low-level function: wakeup all threads waiting for this MVar (blocking).'''
+        with self.__lock:
+            self.__condition.notify_all()
diff --git a/setup.py b/setup.py
index 812a8a6..abb7c8e 100644
--- a/setup.py
+++ b/setup.py
@@ -1,4 +1,4 @@
-# Copyright (c) 2018 Gabriele Baldoni.
+# Copyright (c) 2018 Gabriele Baldoni, Artem Chirkin.
 #
 # See the NOTICE file(s) distributed with this work for additional
 # information regarding copyright ownership.
@@ -10,7 +10,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 #
-# Contributors: Gabriele Baldoni MVar implementation in Python
+# Contributors: Gabriele Baldoni, Artem Chirkin
 
 #!/usr/bin/env python3
 
@@ -24,8 +24,8 @@ def read(fname):
 
 setup(
     name='mvar',
-    version='0.0.1',
-    author='Gabriele Baldoni',
+    version='1.0.0',
+    author='Gabriele Baldoni, Artem Chirkin',
     description='Python implementation of Haskell\'s Control.Concurrent.MVar',
     long_description=read('README.md'),
     packages=['mvar'],


Test patch:
diff --git a/mvar/tests/test_mvar.py b/mvar/tests/test_mvar.py
index b37baa5..6cb782b 100644
--- a/mvar/tests/test_mvar.py
+++ b/mvar/tests/test_mvar.py
@@ -32,12 +32,12 @@ class MVarTests(unittest.TestCase):
 
         local_var = MVar()
         Thread(target=worker, args=(local_var,), daemon=True).start()
-        res = local_var.get()
+        res = local_var.take()
 
         local_var.put(2)
         Thread(target=worker2, args=(local_var,), daemon=True).start()
-        res2 = local_var.get()
-        res3 = local_var.get()
+        res2 = local_var.take()
+        res3 = local_var.take()
 
         self.assertEqual(res, 1)
         self.assertEqual(res2, 2)