diff --git a/src/powerapi/actor/actor.py b/src/powerapi/actor/actor.py index de302b4b..9ea588e4 100644 --- a/src/powerapi/actor/actor.py +++ b/src/powerapi/actor/actor.py @@ -56,18 +56,17 @@ class Actor(multiprocessing.Process): """ state: State - def __init__(self, name: str, level_logger: int = logging.WARNING, timeout: int | None = None): + def __init__(self, name: str, level_logger: int = logging.WARNING): """ Initialize :param name: Name of the actor :param level_logger: Logging level of the actor - :param timeout: Timeout of IPC operations in milliseconds, None for waiting indefinitely """ super().__init__(name=name) self.logging_level = level_logger - self.socket_interface = SocketInterface(name, timeout) + self.socket_interface = SocketInterface(name) def run(self) -> None: """ @@ -202,7 +201,7 @@ def __init__(self, actor_name: str, actor_type: type[Actor]): self.actor_name = actor_name self.actor_type = actor_type - self._ipc_interface = SocketInterface(actor_name, None) + self._ipc_interface = SocketInterface(actor_name) def connect_control(self) -> None: """ diff --git a/src/powerapi/actor/socket_interface.py b/src/powerapi/actor/socket_interface.py index e093a1e9..4b16e21d 100644 --- a/src/powerapi/actor/socket_interface.py +++ b/src/powerapi/actor/socket_interface.py @@ -40,13 +40,11 @@ class SocketInterface: Interface to handle communication between actors. """ - def __init__(self, actor_name: str, timeout: int | None): + def __init__(self, actor_name: str): """ :param str actor_name: Name of the actor whose endpoint is being accessed - :param int timeout: Maximum time, in milliseconds, to wait for an operation """ self.actor_name = actor_name - self.timeout = timeout self.data_socket_filepath = self._generate_socket_path(actor_name, 'data') self.control_socket_filepath = self._generate_socket_path(actor_name, 'control') @@ -112,9 +110,8 @@ def connect_control(self) -> None: This method should only be called by actors that wants to communicate with another actor's endpoint. """ self._control_socket = zmq.Context.instance().socket(zmq.DEALER) - self._control_socket.setsockopt(zmq.LINGER, 0) + self._control_socket.setsockopt(zmq.LINGER, -1) self._control_socket.connect(f'ipc://{self.control_socket_filepath}') - self._control_socket.poll(zmq.POLLIN | zmq.POLLOUT) # Very important, prevents synchronization problems. def receive_control(self, timeout: int | None = None) -> Any: """ @@ -148,7 +145,6 @@ def connect_data(self) -> None: self._data_socket = zmq.Context.instance().socket(zmq.PUSH) self._data_socket.setsockopt(zmq.LINGER, -1) self._data_socket.connect(f'ipc://{self.data_socket_filepath}') - self._data_socket.poll(zmq.POLLOUT) # Very important, prevents synchronization problems. def send_data(self, msg: Any) -> None: """ diff --git a/src/powerapi/dispatcher/dispatcher_actor.py b/src/powerapi/dispatcher/dispatcher_actor.py index 168b0546..9b73f4cd 100644 --- a/src/powerapi/dispatcher/dispatcher_actor.py +++ b/src/powerapi/dispatcher/dispatcher_actor.py @@ -116,16 +116,15 @@ class DispatcherActor(Actor): state: DispatcherState def __init__(self, name: str, formula_factory: FormulaFactory, pushers: dict[type[Report], list[ActorProxy]], - route_table: RouteTable, level_logger: int = logging.WARNING, timeout=None): + route_table: RouteTable, level_logger: int = logging.WARNING): """ Initialize a new dispatcher actor. :param name: Actor name :param formula_factory: Factory function for Formula actors :param route_table: Routing table to use for dispatching the reports between formulas :param level_logger: Logging level of the actor - :param timeout: Maximum time to wait for a message (in milliseconds) """ - super().__init__(name, level_logger, timeout) + super().__init__(name, level_logger) self.formula_factory = formula_factory self.pushers = pushers diff --git a/src/powerapi/formula/formula_actor.py b/src/powerapi/formula/formula_actor.py index 3ee69514..d17ac377 100644 --- a/src/powerapi/formula/formula_actor.py +++ b/src/powerapi/formula/formula_actor.py @@ -82,7 +82,7 @@ def __init__(self, name: str, pushers: dict[type[Report], list[ActorProxy]], lev :param pushers: Mapping of report types to pusher actors :param level_logger: Level of the logger """ - super().__init__(name, level_logger, None) + super().__init__(name, level_logger) self.pushers = pushers diff --git a/src/powerapi/processor/pre/k8s/actor.py b/src/powerapi/processor/pre/k8s/actor.py index 734e080b..f0f54dab 100644 --- a/src/powerapi/processor/pre/k8s/actor.py +++ b/src/powerapi/processor/pre/k8s/actor.py @@ -92,7 +92,7 @@ def __init__(self, name: str, monitor_config: KubernetesMonitorConfig, level_log :param monitor_config: Configuration of the monitoring agent :param level_logger: logging level of the actor """ - super().__init__(name, level_logger, 5000) + super().__init__(name, level_logger) self.monitor_config = monitor_config diff --git a/src/powerapi/processor/pre/openstack/actor.py b/src/powerapi/processor/pre/openstack/actor.py index 9b33e868..bd83815f 100644 --- a/src/powerapi/processor/pre/openstack/actor.py +++ b/src/powerapi/processor/pre/openstack/actor.py @@ -94,7 +94,7 @@ def __init__(self, name: str, monitor_config: OpenStackMonitorConfig, level_logg :param monitor_config: Monitoring agent configuration :param level_logger: Logging level of the actor """ - super().__init__(name, level_logger, 5000) + super().__init__(name, level_logger) self.monitor_config = monitor_config diff --git a/src/powerapi/processor/processor_actor.py b/src/powerapi/processor/processor_actor.py index 5a9ce070..9dd4785b 100644 --- a/src/powerapi/processor/processor_actor.py +++ b/src/powerapi/processor/processor_actor.py @@ -44,14 +44,13 @@ class ProcessorActor(Actor): Used to process reports before sending them to target actor(s). """ - def __init__(self, name: str, level_logger: int = logging.WARNING, timeout: int = 5000): + def __init__(self, name: str, level_logger: int = logging.WARNING): """ Initializes a new processor actor. :param name: Name of the processor actor :param level_logger: Logging level of the actor - :param timeout: Timeout in milliseconds """ - super().__init__(name, level_logger, timeout) + super().__init__(name, level_logger) self.target_actors: list[ActorProxy] = [] diff --git a/src/powerapi/puller/puller_actor.py b/src/powerapi/puller/puller_actor.py index ddfb58f0..b6df5eb5 100644 --- a/src/powerapi/puller/puller_actor.py +++ b/src/powerapi/puller/puller_actor.py @@ -94,7 +94,7 @@ def __init__(self, name: str, database_factory: ReadableDatabaseFactory, report_ :param report_filter: Filter to use when dispatching reports :param level_logger: Define the level of the logger for the actor """ - super().__init__(name, level_logger, 1000) + super().__init__(name, level_logger) self.database_factory = database_factory self.report_filter = report_filter diff --git a/src/powerapi/pusher/pusher_actor.py b/src/powerapi/pusher/pusher_actor.py index bb205cb7..f40da3c2 100644 --- a/src/powerapi/pusher/pusher_actor.py +++ b/src/powerapi/pusher/pusher_actor.py @@ -102,7 +102,7 @@ def __init__(self, name: str, database_factory: WritableDatabaseFactory, flush_i :param max_buffer_size: Maximum number of reports that can be buffered before a forced flush to the database :param logger_level: Define the level of the logger for the actor """ - super().__init__(name, logger_level, 1000) + super().__init__(name, logger_level) self.database_factory = database_factory self.flush_interval = flush_interval diff --git a/tests/unit/actor/test_socket_interface.py b/tests/unit/actor/test_socket_interface.py index b8b9ec9e..ef260a20 100644 --- a/tests/unit/actor/test_socket_interface.py +++ b/tests/unit/actor/test_socket_interface.py @@ -57,7 +57,7 @@ def socket_interface(): Returns an uninitialized socket interface. """ actor_name = f'pytest-{secrets.token_hex()}' # Prevent collisions on IPC files between runs. - return SocketInterface(actor_name, 100) + return SocketInterface(actor_name) @pytest.fixture