Protect set of ready tasks by lock to avoid concurrent updates. (#1489)

When there is no locking then there is a possibility that multiple threads
manipulate with the same object at the same time.
The issue is manifested as:
```
RuntimeError: Set changed size during iteration
```

See: https://github.com/celery/celery/issues/7162

1 files changed, 17 insertions, 0 deletions

diff --git a/t/unit/asynchronous/test_hub.py b/t/unit/asynchronous/test_hub.py
index eae25357..0cf14194 100644
--- a/t/unit/asynchronous/test_hub.py
+++ b/t/unit/asynchronous/test_hub.py
@@ -187,6 +187,12 @@ class test_Hub:
         assert promise() in self.hub._ready
         assert ret is promise()
 
+    def test_call_soon_uses_lock(self):
+        callback = Mock(name='callback')
+        with patch.object(self.hub, '_ready_lock', autospec=True) as lock:
+            self.hub.call_soon(callback)
+            assert lock.__enter__.called_once()
+
     def test_call_soon__promise_argument(self):
         callback = promise(Mock(name='callback'), (1, 2, 3))
         ret = self.hub.call_soon(callback)
@@ -533,3 +539,14 @@ class test_Hub:
         callbacks[0].assert_called_once_with()
         callbacks[1].assert_called_once_with()
         deferred.assert_not_called()
+
+    def test__pop_ready_pops_ready_items(self):
+        self.hub._ready.add(None)
+        ret = self.hub._pop_ready()
+        assert ret == {None}
+        assert self.hub._ready == set()
+
+    def test__pop_ready_uses_lock(self):
+        with patch.object(self.hub, '_ready_lock', autospec=True) as lock:
+            self.hub._pop_ready()
+            assert lock.__enter__.called_once()