Convert monitoring type annotations to PEP-526 from comments (#3573)

This is in preparation for future type work in the monitoring
codebase (for example, see PR #3572).

This PR does not claim that the types it is moving around are
correct (and PR #3572 contains some instances where the types
are incorrect). It is a purely syntactic PR.

After this PR, $ git grep '# type:' parsl/monitoring/
returns two remaining comment style annotations, which are
'type: ignore' exclusions not specific types.

parsl/monitoring/db_manager.py

@@ -283,7 +283,7 @@ def __init__(self,
                  ):
 
         self.workflow_end = False
-        self.workflow_start_message = None  # type: Optional[MonitoringMessage]
+        self.workflow_start_message: Optional[MonitoringMessage] = None
         self.logdir = logdir
         os.makedirs(self.logdir, exist_ok=True)
 
@@ -299,10 +299,10 @@ def __init__(self,
         self.batching_interval = batching_interval
         self.batching_threshold = batching_threshold
 
-        self.pending_priority_queue = queue.Queue()  # type: queue.Queue[TaggedMonitoringMessage]
-        self.pending_node_queue = queue.Queue()  # type: queue.Queue[MonitoringMessage]
-        self.pending_block_queue = queue.Queue()  # type: queue.Queue[MonitoringMessage]
-        self.pending_resource_queue = queue.Queue()  # type: queue.Queue[MonitoringMessage]
+        self.pending_priority_queue: queue.Queue[TaggedMonitoringMessage] = queue.Queue()
+        self.pending_node_queue: queue.Queue[MonitoringMessage] = queue.Queue()
+        self.pending_block_queue: queue.Queue[MonitoringMessage] = queue.Queue()
+        self.pending_resource_queue: queue.Queue[MonitoringMessage] = queue.Queue()
 
     def start(self,
               priority_queue: "queue.Queue[TaggedMonitoringMessage]",
@@ -351,18 +351,18 @@ def start(self,
         If that happens, the message will be added to deferred_resource_messages and processed later.
 
         """
-        inserted_tasks = set()  # type: Set[object]
+        inserted_tasks: Set[object] = set()
 
         """
         like inserted_tasks but for task,try tuples
         """
-        inserted_tries = set()  # type: Set[Any]
+        inserted_tries: Set[Any] = set()
 
         # for any task ID, we can defer exactly one message, which is the
         # assumed-to-be-unique first message (with first message flag set).
         # The code prior to this patch will discard previous message in
         # the case of multiple messages to defer.
-        deferred_resource_messages = {}  # type: MonitoringMessage
+        deferred_resource_messages: MonitoringMessage = {}
 
         exception_happened = False
 
@@ -505,7 +505,7 @@ def start(self,
                         "Got {} messages from block queue".format(len(block_info_messages)))
                     # block_info_messages is possibly a nested list of dict (at different polling times)
                     # Each dict refers to the info of a job/block at one polling time
-                    block_messages_to_insert = []  # type: List[Any]
+                    block_messages_to_insert: List[Any] = []
                     for block_msg in block_info_messages:
                         block_messages_to_insert.extend(block_msg)
                     self._insert(table=BLOCK, messages=block_messages_to_insert)
@@ -686,7 +686,7 @@ def _insert(self, table: str, messages: List[MonitoringMessage]) -> None:
                 logger.exception("Rollback failed")
 
     def _get_messages_in_batch(self, msg_queue: "queue.Queue[X]") -> List[X]:
-        messages = []  # type: List[X]
+        messages: List[X] = []
         start = time.time()
         while True:
             if time.time() - start >= self.batching_interval or len(messages) >= self.batching_threshold:

parsl/monitoring/remote.py

@@ -199,10 +199,10 @@ def monitor(pid: int,
 
     pm = psutil.Process(pid)
 
-    children_user_time = {}  # type: Dict[int, float]
-    children_system_time = {}  # type: Dict[int, float]
-    children_num_ctx_switches_voluntary = {}  # type: Dict[int, float]
-    children_num_ctx_switches_involuntary = {}  # type: Dict[int, float]
+    children_user_time: Dict[int, float] = {}
+    children_system_time: Dict[int, float] = {}
+    children_num_ctx_switches_voluntary: Dict[int, float] = {}
+    children_num_ctx_switches_involuntary: Dict[int, float] = {}
 
     def accumulate_and_prepare() -> Dict[str, Any]:
         d = {"psutil_process_" + str(k): v for k, v in pm.as_dict().items() if k in simple}