Repository: tqdm/tqdm (Python)

Issue:
PyTorch TQDM conflict
``DataLoader`` when interacting with ``DistributedDataParallel`` and ``tqdm==4.26.0`` causes semaphores to leak.

PyTorch issue: https://github.com/pytorch/pytorch/issues/9985#issuecomment-421502889

Any ideas on resolving this? 

```python
from torch import multiprocessing

# DEPENDANCY: This is required for ``DistributedDataParallel``
# https://pytorch.org/docs/stable/nn.html?highlight=distributeddataparallel#torch.nn.parallel.DistributedDataParallel
try:
    multiprocessing.set_start_method('spawn')
except RuntimeError:
    pass

# DEPENDANCY: This is required for ``from tqdm import tqdm``
# https://github.com/tqdm/tqdm/blob/96d8a3c3642474144f53f74331ef2172d1c39496/tqdm/_tqdm.py#L74
mp_lock = multiprocessing.RLock()

import torch
from torch.utils.data import DataLoader

if __name__ == '__main__':
    data_iterator = torch.utils.data.DataLoader([torch.tensor(i) for i in range(10)], num_workers=4)
    for batch in data_iterator:
        pass
```
```
/Users/michaelp/.pyenv/versions/3.6.5/lib/python3.6/multiprocessing/semaphore_tracker.py:143: UserWarning: semaphore_tracker: There appear to be 1 leaked semaphores to clean up at shutdown
  len(cache))
/Users/michaelp/.pyenv/versions/3.6.5/lib/python3.6/multiprocessing/semaphore_tracker.py:143: UserWarning: semaphore_tracker: There appear to be 1 leaked semaphores to clean up at shutdown
  len(cache))
/Users/michaelp/.pyenv/versions/3.6.5/lib/python3.6/multiprocessing/semaphore_tracker.py:143: UserWarning: semaphore_tracker: There appear to be 1 leaked semaphores to clean up at shutdown
  len(cache))
/Users/michaelp/.pyenv/versions/3.6.5/lib/python3.6/multiprocessing/semaphore_tracker.py:143: UserWarning: semaphore_tracker: There appear to be 1 leaked semaphores to clean up at shutdown
  len(cache))
```


Hints:
@PetrochukM 
I tried the following code with the lastest pytorch(0.4.1) and tqdm(v4.26) in python2.7 and python3.6
```python
from torch import multiprocessing

# DEPENDANCY: This is required for ``DistributedDataParallel``
# https://pytorch.org/docs/stable/nn.html?highlight=distributeddataparallel#torch.nn.parallel.DistributedDataParallel
# try:
#     multiprocessing.set_start_method('spawn')
# except RuntimeError:
#     pass

# DEPENDANCY: This is required for ``from tqdm import tqdm``
# https://github.com/tqdm/tqdm/blob/96d8a3c3642474144f53f74331ef2172d1c39496/tqdm/_tqdm.py#L74
mp_lock = multiprocessing.RLock()

import torch
from torch.utils.data import DataLoader

data_iterator = torch.utils.data.DataLoader([torch.tensor(i) for i in range(10)], num_workers=4)
for batch in data_iterator:
    pass
```

I cannot reproduce your error.
It works perfectly.

**But perhaps because I use the No-CUDA pytorch installation, as I only have AMD GPU**
Thanks for trying to reproduce. Looks the code example you used to reproduce the problem misses critical requirements to reproduce the problem. 

Looks like you commented out:
```
# try:
#     multiprocessing.set_start_method('spawn')
# except RuntimeError:
#     pass
```
And removed:
```
if __name__ == '__main__':
```

This example is reproducible on a CPU, no GPU is required. The latest python version is required, I am using: ``Python 3.6.5``. 
@PetrochukM 
Yes, now I reproduce the error.
Then I find comment the line of mp_lock can fix it.
Why you need to set RLock apparently?? How about let tqdm decide it?
@chengs 

It'd be nice if you were a bit more thoughtful before relying on this thread. For example:
- Please don't change the code and then claim you can't reproduce it.
- Please don't ask me questions that are answered in the original post.

-------------------------

Originally, I commented that:
```
# DEPENDANCY: This is required for ``from tqdm import tqdm``
# https://github.com/tqdm/tqdm/blob/96d8a3c3642474144f53f74331ef2172d1c39496/tqdm/_tqdm.py#L74
```

Sorry if that's not clear enough. If you run ``from tqdm import tqdm`` it runs this line of code ``mp_lock = multiprocessing.RLock()`` ([source](https://github.com/tqdm/tqdm/blob/96d8a3c3642474144f53f74331ef2172d1c39496/tqdm/_tqdm.py#L74))

To make it more clear.... this code below causes the bug as well.
```python
from torch import multiprocessing

# DEPENDANCY: This is required for ``DistributedDataParallel``
# https://pytorch.org/docs/stable/nn.html?highlight=distributeddataparallel#torch.nn.parallel.DistributedDataParallel
try:
    multiprocessing.set_start_method('spawn')
except RuntimeError:
    pass

from tqdm import tqdm

import torch
from torch.utils.data import DataLoader

if __name__ == '__main__':
    data_iterator = torch.utils.data.DataLoader([torch.tensor(i) for i in range(10)], num_workers=4)
    for batch in data_iterator:
        pass
```

@PetrochukM 

Check this 
https://docs.python.org/3/library/multiprocessing.html#multiprocessing.set_start_method

It says:
> multiprocessing.set_start_method(method)
Set the method which should be used to start child processes. method can be 'fork', 'spawn' or 'forkserver'.
> Note that this should be called at most once, and **it should be protected inside the if __name__ == '__main__' clause of the main module.**

Indeed, ``DistributedDataParallel`` asks the call of this method, well I think it should be called inside ``__main__``.
Related to https://github.com/pytorch/pytorch/issues/11727

@casperdcl Maybe it's better to add a known issue into README about the conflict between ``RLock`` and the usage of ``set_start_method('spawn')``. 
Due to such rare compatibility problems, we should have a simple version (#610) ASAP, to give users a fallback solution.
> Indeed, `DistributedDataParallel` asks the call of this method, well I think it should be called inside `__main__`.
@chengs If you are going to put the line `multiprocessing.set_start_method('spawn')` inside `if __name__=='main'`you'll invoke the Runtime Error Context has already been set from
https://github.com/python/cpython/blob/b3b8cb419e496629873fa7dda82a01863f58617a/Lib/multiprocessing/context.py#L242
tqdm may be setting the context. If you then remove the line `from tqdm import tqdm`, it works. So I guess you should not manually import tqdm; rather let the multiprocessing library use tqdm in whatever way it wants.
I'd also like to point out that compatibility problems are not that rare. There are a bunch of libraries that break on OS X unless you use 'spawn' or 'forkserver' as your start method. In addition to pytorch:
- https://github.com/numpy/numpy/issues/4776#issuecomment-45114910
- https://github.com/opencv/opencv/issues/5150#issuecomment-161371095

I'm not sure if this is what was suggested above, but a workaround for now is to import tqdm inside of `if __name__ == '__main__'` or inside of the function where you use it (after you have set the multiprocessing context). It's not very clean to have imports in the middle of code, but it does avoid the error.

I think another possible solution is to initialize the write lock in a lazy way. This means that the main process would have to create a progress bar or call some init function before forking and creating subprocesses, so it would be a little worse for usability, but it would solve the issue.

Edit: A python 3.7-only way to make sure lazy initialization works is with https://python.readthedocs.io/en/latest/library/os.html#os.register_at_fork
@casperdcl @mjstevens777 @PetrochukM Eh, you're right. tqdm use RLock as global var. this way is not very good. On linux/mac it only works well when multiprocessing start as "fork". There should be a better way to use this.

Patch:
diff --git a/tqdm/_tqdm.py b/tqdm/_tqdm.py
--- a/tqdm/_tqdm.py
+++ b/tqdm/_tqdm.py
@@ -69,30 +69,23 @@ class TqdmMonitorWarning(TqdmWarning, RuntimeWarning):
     pass
 
 
-# Create global parallelism locks to avoid racing issues with parallel bars
-# works only if fork available (Linux, MacOSX, but not on Windows)
-try:
-    mp_lock = mp.RLock()  # multiprocessing lock
-except ImportError:  # pragma: no cover
-    mp_lock = None
-except OSError:  # pragma: no cover
-    mp_lock = None
-try:
-    th_lock = th.RLock()  # thread lock
-except OSError:  # pragma: no cover
-    th_lock = None
-
-
 class TqdmDefaultWriteLock(object):
     """
     Provide a default write lock for thread and multiprocessing safety.
     Works only on platforms supporting `fork` (so Windows is excluded).
+    You must initialize an instance of tqdm or TqdmDefaultWriteLock before
+    forking in order for the write lock to work.
     On Windows, you need to supply the lock from the parent to the children as
     an argument to joblib or the parallelism lib you use.
     """
+
     def __init__(self):
-        global mp_lock, th_lock
-        self.locks = [lk for lk in [mp_lock, th_lock] if lk is not None]
+        # Create global parallelism locks to avoid racing issues with parallel bars
+        # works only if fork available (Linux, MacOSX, but not on Windows)
+        self.create_mp_lock()
+        self.create_th_lock()
+        cls = type(self)
+        self.locks = [lock for lock in [cls.mp_lock, cls.th_lock] if lock is not None]
 
     def acquire(self):
         for lock in self.locks:
@@ -108,6 +101,31 @@ def __enter__(self):
     def __exit__(self, *exc):
         self.release()
 
+    @classmethod
+    def create_mp_lock(cls):
+        if not hasattr(cls, 'mp_lock'):
+            try:
+                cls.mp_lock = mp.RLock()  # multiprocessing lock
+            except ImportError:  # pragma: no cover
+                cls.mp_lock = None
+            except OSError:  # pragma: no cover
+                cls.mp_lock = None
+
+    @classmethod
+    def create_th_lock(cls):
+        if not hasattr(cls, 'th_lock'):
+            try:
+                cls.th_lock = th.RLock()  # thread lock
+            except OSError:  # pragma: no cover
+                cls.th_lock = None
+
+
+# Create a thread lock before instantiation so that no setup needs to be done
+# before running in a multithreaded environment.
+# Do not create the multiprocessing lock because it sets the multiprocessing
+# context and does not allow the user to use 'spawn' or 'forkserver' methods.
+TqdmDefaultWriteLock.create_th_lock()
+
 
 class tqdm(Comparable):
     """
@@ -118,7 +136,7 @@ class tqdm(Comparable):
 
     monitor_interval = 10  # set to 0 to disable the thread
     monitor = None
-    _lock = TqdmDefaultWriteLock()
+    # _instances and _lock defined in __new__
 
     @staticmethod
     def format_sizeof(num, suffix='', divisor=1000):
@@ -444,9 +462,9 @@ def __new__(cls, *args, **kwargs):
         # Create a new instance
         instance = object.__new__(cls)
         # Add to the list of instances
-        if "_instances" not in cls.__dict__:
+        if not hasattr(cls, '_instances'):
             cls._instances = WeakSet()
-        if "_lock" not in cls.__dict__:
+        if not hasattr(cls, '_lock'):
             cls._lock = TqdmDefaultWriteLock()
         with cls._lock:
             cls._instances.add(instance)


Test patch:
diff --git a/tqdm/tests/tests_tqdm.py b/tqdm/tests/tests_tqdm.py
--- a/tqdm/tests/tests_tqdm.py
+++ b/tqdm/tests/tests_tqdm.py
@@ -710,6 +710,44 @@ def test_smoothed_dynamic_min_iters_with_min_interval():
     assert '14%' in out and '14%' in out2
 
 
+@with_setup(pretest, posttest)
+def test_rlock_creation():
+    """Test that importing tqdm does not create multiprocessing objects."""
+    import multiprocessing as mp
+    if sys.version_info < (3, 3):
+        # unittest.mock is a 3.3+ feature
+        raise SkipTest
+
+    # Use 'spawn' instead of 'fork' so that the process does not inherit any
+    # globals that have been constructed by running other tests
+    ctx = mp.get_context('spawn')
+    with ctx.Pool(1) as pool:
+        # The pool will propagate the error if the target method fails
+        pool.apply(_rlock_creation_target)
+
+
+def _rlock_creation_target():
+    """Check that the RLock has not been constructed."""
+    from unittest.mock import patch
+    import multiprocessing as mp
+
+    # Patch the RLock class/method but use the original implementation
+    with patch('multiprocessing.RLock', wraps=mp.RLock) as rlock_mock:
+        # Importing the module should not create a lock
+        from tqdm import tqdm
+        assert rlock_mock.call_count == 0
+        # Creating a progress bar should initialize the lock
+        with closing(StringIO()) as our_file:
+            with tqdm(file=our_file) as t:
+                pass
+        assert rlock_mock.call_count == 1
+        # Creating a progress bar again should reuse the lock
+        with closing(StringIO()) as our_file:
+            with tqdm(file=our_file) as t:
+                pass
+        assert rlock_mock.call_count == 1
+
+
 @with_setup(pretest, posttest)
 def test_disable():
     """Test disable"""