go shutdown go

Author: YvesDup

Lib/multiprocessing/queues.py

@@ -47,12 +47,11 @@ def __init__(self, maxsize=0, *, ctx):
             self._wlock = ctx.Lock()
         self._sem = ctx.BoundedSemaphore(maxsize)
 
-
         self._lock_shutdown = ctx.Lock()
         # Cannot use a ctx.Value because 'ctypes' library is
         # not always available on all Linux platforms.
-        # Using Semaphores instead of heap.BufferWrapper
-        # as an array of int is more explicit.
+        # Use of Semaphores instead of an array from `heap.BufferWrapper'
+        # is here more efficient and explicit.
         self._sem_flag_shutdown = ctx.Semaphore(0)
         self._sem_flag_shutdown_immediate = ctx.Semaphore(0)
         self._sem_pending_getters = ctx.Semaphore(0)
@@ -64,14 +63,6 @@ def __init__(self, maxsize=0, *, ctx):
         if sys.platform != 'win32':
             register_after_fork(self, Queue._after_fork)
 
-    def _is_shutdown(self):
-        return not self._sem_flag_shutdown.locked()
-
-    def _set_shutdown(self, immediate=False):
-        self._sem_flag_shutdown.release()
-        if immediate:
-            self._sem_flag_shutdown_immediate.release()
-
     def __getstate__(self):
         context.assert_spawning(self)
         return (self._ignore_epipe, self._maxsize, self._reader, self._writer,
@@ -107,6 +98,14 @@ def _reset(self, after_fork=False):
         self._recv_bytes = self._reader.recv_bytes
         self._poll = self._reader.poll
 
+    def _is_shutdown(self):
+        return not self._sem_flag_shutdown.locked()
+
+    def _set_shutdown(self, immediate=False):
+        self._sem_flag_shutdown.release()
+        if immediate:
+            self._sem_flag_shutdown_immediate.release()
+
     @contextmanager
     def _handle_pending_processes(self, sem):
         # Count pending getter or putter processes in a dedicated
@@ -189,7 +188,7 @@ def get(self, block=True, timeout=None):
         if self._is_shutdown() \
             and isinstance(item, _ShutdownSentinel):
             # A pending getter process is just unblocked,
-            # we try to unblock a next one if exists.
+            # Unblock a next one if exists.
             self._release_pending_getters()
             raise ShutDown
 
@@ -220,7 +219,7 @@ def _put_sentinel(self):
         # When put a sentinel into an empty queue,
         # dont forget to call to _sem.acquire in order to
         # maintain a correct count of acquire/release
-        #calls for BoudedSempaphore.
+        # calls for BoudedSempaphore.
         self._sem.acquire()
 
         with self._notempty:
@@ -248,13 +247,13 @@ def shutdown(self, immediate=False):
             self._set_shutdown(immediate)
 
             # Shut down is immediatly and there is no pending getter,
-            # we purge the queue (pipe). If data is into the buffer and
-            # not into pipe, the 'put' thread should erase remaining data.
+            # we purge the queue (pipe). If there are datas into the buffer
+            # the 'feeder' thread should erase all of them.
             if immediate and not is_pending_getters:
                 self._clear()
 
             # Starting release one pending getter process.
-            # Put a first shutdown sentinel into the pipe.
+            # Put a first shutdown sentinel data into the pipe.
             if self.empty() and is_pending_getters:
                 self._put_sentinel()
 
@@ -386,12 +385,12 @@ def _feed(buffer, notempty, send_bytes, writelock, reader_close,
                             writer_close()
                             return
 
-                        # When queue shuts down immediatly, don`t insert
+                        # When queue shuts down immediatly, do not insert
                         # regular data in pipe, only shutdown sentinel.
                         if is_shutdown_immediate() \
                             and not isinstance(obj, _ShutdownSentinel):
                             debug("Queue shuts down immediatly, " \
-                                  "don't feed regular data in pipe")
+                                  "don't feed regular data to pipe")
                             continue
 
                         # serialize the data before acquiring the lock
@@ -504,7 +503,8 @@ def _clear(self):
         super()._clear()
 
         # Data could be in the buffer, not in the pipe.
-        # Call acquire until Semaphore counter is zero.
+        # Call acquire method of '_unfinished_tasks' Semaphore
+        # until counter is zero.
         with self._cond:
             while not self._unfinished_tasks.locked():
                 self._unfinished_tasks.acquire(block=False)

Lib/test/_test_multiprocessing.py

@@ -1687,7 +1687,9 @@ def test_queue_shutdown_count_pending_put(self):
 
         self._start_processes_pending(n, target=self._pending_put,
                                       args=(q, b, results))
+        # Wait for all procesess start.
         b.wait()
+
         # to be sure that queue is full, and all 'n-size' others processes
         # are pending.
         self._wait()
@@ -1719,7 +1721,9 @@ def test_queue_shutdown_immediate_pending_put(self):
         b = self.Barrier(n+1)
         self._start_processes_pending(n, target=self._pending_put,
                                       args=(q, b, results))
+        # Wait for all procesess start.
         b.wait()
+
         # We need to call _wait to be sure that queue is full,
         # and all others processes are pending.
         self._wait()
@@ -1757,8 +1761,12 @@ def test_queue_shutdown_count_pending_get(self):
         b = self.Barrier(n+1)
         self._start_processes_pending(n, target=self._pending_get,
                                       args=(q, b, results))
+        # Wait for all procesess start.
         b.wait()
-        self._wait() # wait for all pending get processes to be blocked.
+
+        # wait for all pending get processes to be blocked.
+        self._wait()
+
         self.assertTrue(q.empty())
         self.assertEqual(q._sem_pending_getters.get_value(), n)
         self.assertEqual(q._sem_pending_putters.get_value(), 0)
@@ -1784,8 +1792,11 @@ def test_queue_shutdown_immediate_pending_get(self):
         b = self.Barrier(n+1)
         self._start_processes_pending(n, target=self._pending_get,
                                       args=(q, b, results))
+        # Wait for all procesess start.
         b.wait()
-        self._wait() # wait for all pending get processes to be blocked.
+
+        # wait for all pending get processes to be blocked.
+        self._wait()
 
         q.shutdown(immediate=True)
         self.assertTrue(q.empty())
@@ -1814,8 +1825,11 @@ def test_queue_shutdown_pending_get(self):
         b = self.Barrier(n+1)
         self._start_processes_pending(n, target=self._pending_get,
                                       args=(q, b, results))
+        # Wait for all procesess start.
         b.wait()
-        self._wait() # wait for all pending get processes to be blocked.
+
+        # wait for all pending get processes to be blocked.
+        self._wait()
 
         q.shutdown(immediate=False)
         self.assertTrue(q._is_shutdown())
@@ -1828,7 +1842,7 @@ def test_queue_shutdown_pending_get(self):
         close_queue(q)
 
     @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
-    def queue_shutdown_immediate_purge_buffer(self):
+    def queue_shutdown_immediate_check_purge_buffer(self):
         if self.TYPE != 'processes':
             self.skipTest(f'test not appropriate for {self.TYPE}')
 
@@ -1850,7 +1864,7 @@ def queue_shutdown_immediate_purge_buffer(self):
             q.join()
 
             log_record = stream.getvalue()
-            self.assertIn("don't feed regular data in pipe", log_record)
+            self.assertIn("don't feed regular data to pipe", log_record)
 
             close_queue(q)
         finally:
@@ -1882,7 +1896,9 @@ def _join_joinablequeue(self, immediate):
         return_process = 1000
         p = self.Process(target=self._shutdown,
                          args=(q, immediate, results,
-                              return_process+immediate))
+                              return_process+immediate
+                              )
+                        )
         p.start()
         q.join()
         p.join()