Skip to content

Commit

Permalink
Correct and check types on monitoring router and database processes (#…
Browse files Browse the repository at this point in the history
…3572)

Prior to this PR, the startup code for the monitoring router and database processes had type annotations on queues; but these types were not checked, and were incorrect - they were labelled process-local Queue instead of multiprocessing queues. This did not cause much trouble execution- and mypy-wise, as the interfaces of those two classes are similar enough, but it is confusing to read in a part of the codebase that is already confusing (that confusion is probably what lead to the incorrect annotations in the first place...)

They were not checked because the informal policy of "internal stuff is checked with mypy, external interfaces are checked with typeguard" works badly here:

The startup methods are launched using multiprocessing.Process, and function invocations are not type-checked by mypy across a multiprocessing Process constructor.

Changed Behaviour
This PR introduces typeguard decorators onto the router and database start methods so that this internal checking happens at runtime. This consequently reveals that the type annotations of these methods are incorrect, and so this PR makes those consequential changes.

Further, generic types (Queue[MessageType]) are not supported on multiprocessing.Queues before Python 3.12 - so those generic indices are removed from the type annotations. That is unfortunate and weakens in-process static verification - but they could be re-introduced after Parsl drops Python 3.11 support (around 2027 in the present informal support policy)
  • Loading branch information
benclifford authored Aug 14, 2024
1 parent 2067b40 commit ffb3644
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 21 deletions.
22 changes: 13 additions & 9 deletions parsl/monitoring/db_manager.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,14 @@
import datetime
import logging
import multiprocessing.queues as mpq
import os
import queue
import threading
import time
from typing import Any, Dict, List, Optional, Set, Tuple, TypeVar, cast

import typeguard

from parsl.dataflow.states import States
from parsl.errors import OptionalModuleMissing
from parsl.log_utils import set_file_logger
Expand Down Expand Up @@ -305,10 +308,10 @@ def __init__(self,
self.pending_resource_queue: queue.Queue[MonitoringMessage] = queue.Queue()

def start(self,
priority_queue: "queue.Queue[TaggedMonitoringMessage]",
node_queue: "queue.Queue[MonitoringMessage]",
block_queue: "queue.Queue[MonitoringMessage]",
resource_queue: "queue.Queue[MonitoringMessage]") -> None:
priority_queue: mpq.Queue,
node_queue: mpq.Queue,
block_queue: mpq.Queue,
resource_queue: mpq.Queue) -> None:

self._kill_event = threading.Event()
self._priority_queue_pull_thread = threading.Thread(target=self._migrate_logs_to_internal,
Expand Down Expand Up @@ -719,11 +722,12 @@ def close(self) -> None:


@wrap_with_logs(target="database_manager")
def dbm_starter(exception_q: "queue.Queue[Tuple[str, str]]",
priority_msgs: "queue.Queue[TaggedMonitoringMessage]",
node_msgs: "queue.Queue[MonitoringMessage]",
block_msgs: "queue.Queue[MonitoringMessage]",
resource_msgs: "queue.Queue[MonitoringMessage]",
@typeguard.typechecked
def dbm_starter(exception_q: mpq.Queue,
priority_msgs: mpq.Queue,
node_msgs: mpq.Queue,
block_msgs: mpq.Queue,
resource_msgs: mpq.Queue,
db_url: str,
logdir: str,
logging_level: int) -> None:
Expand Down
26 changes: 14 additions & 12 deletions parsl/monitoring/router.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
from __future__ import annotations

import logging
import multiprocessing.queues as mpq
import os
import pickle
import queue
import socket
import threading
import time
from multiprocessing.synchronize import Event
from typing import Optional, Tuple, Union
from typing import Optional, Tuple

import typeguard
import zmq

from parsl.log_utils import set_file_logger
Expand All @@ -33,10 +34,10 @@ def __init__(self,
logdir: str = ".",
logging_level: int = logging.INFO,
atexit_timeout: int = 3, # in seconds
priority_msgs: "queue.Queue[AddressedMonitoringMessage]",
node_msgs: "queue.Queue[AddressedMonitoringMessage]",
block_msgs: "queue.Queue[AddressedMonitoringMessage]",
resource_msgs: "queue.Queue[AddressedMonitoringMessage]",
priority_msgs: mpq.Queue,
node_msgs: mpq.Queue,
block_msgs: mpq.Queue,
resource_msgs: mpq.Queue,
exit_event: Event,
):
""" Initializes a monitoring configuration class.
Expand Down Expand Up @@ -202,12 +203,13 @@ def start_zmq_listener(self) -> None:


@wrap_with_logs
def router_starter(comm_q: "queue.Queue[Union[Tuple[int, int], str]]",
exception_q: "queue.Queue[Tuple[str, str]]",
priority_msgs: "queue.Queue[AddressedMonitoringMessage]",
node_msgs: "queue.Queue[AddressedMonitoringMessage]",
block_msgs: "queue.Queue[AddressedMonitoringMessage]",
resource_msgs: "queue.Queue[AddressedMonitoringMessage]",
@typeguard.typechecked
def router_starter(comm_q: mpq.Queue,
exception_q: mpq.Queue,
priority_msgs: mpq.Queue,
node_msgs: mpq.Queue,
block_msgs: mpq.Queue,
resource_msgs: mpq.Queue,
exit_event: Event,

hub_address: str,
Expand Down

0 comments on commit ffb3644

Please sign in to comment.