refactor(actor)!: Remove the unused global socket timeout

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:
         """

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:
         """

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

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
 

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
 

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
 

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] = []
 

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

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

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