Replace use of queue.qsize() in monitoring database manager

Prior to this PR, each use of .qsize() was of the form

q.qsize() != 0

which is equivalent to (not q.empty())

Queue.empty() is better supported across platforms than .qsize()

This PR makes that substitution, meaning the broken (#3856) MacSafeQueue
subclass will not be needed in this situation any more and the database
manager will be able to work with a regular multiprocessing Queue even
on Mac.

This PR does not actually change the Queue type - it only reduces the
functionality required by whichever queue is supplied.

This is part of a series of PRs to remove that platform-specific
functionality entirely - see PR #3932.

Author: benclifford

parsl/monitoring/db_manager.py

@@ -346,9 +346,9 @@ def start(self,
         exception_happened = False
 
         while (not self._kill_event.is_set() or
-               self.pending_priority_queue.qsize() != 0 or self.pending_resource_queue.qsize() != 0 or
-               self.pending_node_queue.qsize() != 0 or self.pending_block_queue.qsize() != 0 or
-               resource_queue.qsize() != 0):
+               not self.pending_priority_queue.empty() or not self.pending_resource_queue.empty() or
+               not self.pending_node_queue.empty() or not self.pending_block_queue.empty() or
+               not resource_queue.empty()):
 
             """
             WORKFLOW_INFO and TASK_INFO messages (i.e. priority messages)
@@ -357,9 +357,9 @@ def start(self,
             try:
                 logger.debug("""Checking STOP conditions: {}, {}, {}, {}, {}, {}""".format(
                                   self._kill_event.is_set(),
-                                  self.pending_priority_queue.qsize() != 0, self.pending_resource_queue.qsize() != 0,
-                                  self.pending_node_queue.qsize() != 0, self.pending_block_queue.qsize() != 0,
-                                  resource_queue.qsize() != 0))
+                                  not self.pending_priority_queue.empty(), not self.pending_resource_queue.empty(),
+                                  not self.pending_node_queue.empty(), not self.pending_block_queue.empty(),
+                                  not resource_queue.empty()))
 
                 # This is the list of resource messages which can be reprocessed as if they
                 # had just arrived because the corresponding first task message has been
@@ -558,9 +558,9 @@ def start(self,
     def _migrate_logs_to_internal(self, logs_queue: mpq.Queue, kill_event: threading.Event) -> None:
         logger.info("Starting _migrate_logs_to_internal")
 
-        while not kill_event.is_set() or logs_queue.qsize() != 0:
+        while not kill_event.is_set() or not logs_queue.empty():
             logger.debug("Checking STOP conditions: kill event: %s, queue has entries: %s",
-                         kill_event.is_set(), logs_queue.qsize() != 0)
+                         kill_event.is_set(), not logs_queue.empty())
 
             try:
                 x = logs_queue.get(timeout=0.1)