Merge pull request #778 from powerapi-ng/refactor/actor-picklable-args

feat!: Modernize actor architecture and rework dependent components

Author: gfieni

src/powerapi/actor/__init__.py

@@ -28,7 +28,7 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 from powerapi.actor.socket_interface import SocketInterface, NotConnectedException
-from powerapi.actor.supervisor import Supervisor, ActorInitError, ActorAlreadySupervisedException
-from powerapi.actor.supervisor import CrashConfigureError, FailConfigureError
+from powerapi.actor.message import Message, StartMessage, OKMessage, ErrorMessage, PoisonPillMessage
 from powerapi.actor.state import State
-from powerapi.actor.actor import Actor, InitializationException
+from powerapi.actor.actor import Actor, ActorProxy, InitializationException
+from powerapi.actor.supervisor import Supervisor, ActorAlreadySupervisedException, ActorInitializationError

src/powerapi/actor/actor.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2018, INRIA
+# Copyright (c) 2018, Inria
 # Copyright (c) 2018, University of Lille
 # All rights reserved.
 #
@@ -27,246 +27,175 @@
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-import ctypes
-import logging
-import multiprocessing
+from hashlib import blake2b
+from pathlib import Path
+from typing import Any
 
 import zmq
 
 from powerapi.exception import PowerAPIException
 
-LOCAL_ADDR = 'tcp://127.0.0.1'
-
 
 class NotConnectedException(PowerAPIException):
     """
-    Exception raised when attempting to send/receinve a message on a socket
-    that is not conected
+    Exception raised when attempting to do an operation on a disconnected socket.
     """
 
 
 class SocketInterface:
     """
-    Interface to handle comunication to/from the actor
-
-    general methods :
-
-    - :meth:`send_control <powerapi.actor.socket_interface.SocketInterface.send_control>`
-
-    client interface methods :
-
-    - :meth:`connect_data <powerapi.actor.socket_interface.SocketInterface.connect_data>`
-    - :meth:`connect_control <powerapi.actor.socket_interface.SocketInterface.connect_control>`
-    - :meth:`send_data <powerapi.actor.socket_interface.SocketInterface.send_data>`
-    - :meth:`close <powerapi.actor.socket_interface.SocketInterface.close>`
-
-    server interface methods :
-
-    - :meth:`setup <powerapi.actor.socket_interface.SocketInterface.setup>`
-    - :meth:`receive <powerapi.actor.socket_interface.SocketInterface.receive>`
-    - :meth:`close <powerapi.actor.socket_interface.SocketInterface.close>`
+    Interface to handle communication between actors.
     """
 
-    def __init__(self, name, timeout):
+    def __init__(self, actor_name: str, timeout: int | None):
         """
-        :param str name: name of the actor using this interface
-        :param int timeout: time in millisecond to wait for a message
+        :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.logger = logging.getLogger(name)
-
-        #: (int): Time in millisecond to wait for a message before execute
-        #:        timeout_handler
+        self.actor_name = actor_name
         self.timeout = timeout
 
-        #: (str): Address of the pull socket
-        self.pull_socket_address = None
-
-        #: (str): Address of the control socket
-        self.control_socket_address = None
-
-        #: (zmq.Poller): ZMQ Poller for read many socket at same time
-        self.poller = zmq.Poller()
-
-        #: (zmq.Socket): ZMQ Pull socket for receiving data message
-        self.pull_socket = None
-
-        #: (zmq.Socket): ZMQ Pair socket for receiving control message
-        self.control_socket = None
+        self.data_socket_filepath = self._generate_socket_path(actor_name, 'data')
+        self.control_socket_filepath = self._generate_socket_path(actor_name, 'control')
 
-        # This socket is used to connect to the pull socket of this actor. It
-        # won't be created on the actor's process but on the process that want
-        # to connect to the pull socket of this actor
-        #: (zmq.Socket): ZMQ Push socket for sending message to this actor
-        self.push_socket = None
+        self._control_socket: zmq.Socket | None = None
+        self._data_socket: zmq.Socket | None = None
+        self._sockets_poller: zmq.Poller | None = None
+        self._is_endpoint: bool = False
 
-        # Shared memory used to communicate the port used to bind sockets
-        self._pull_port = multiprocessing.Value(ctypes.c_int)
-        self._ctrl_port = multiprocessing.Value(ctypes.c_int)
-        self._values_available = multiprocessing.Event()
-
-        self._pull_port.value = -1
-        self._ctrl_port.value = -1
-
-    def setup(self):
+    @staticmethod
+    def _generate_socket_path(actor_name: str, socket_purpose: str, basedir: str = '/tmp') -> Path:
         """
-        Initialize sockets and send the selected port number to the father
-        process with a Pipe
+        Generate a deterministic filesystem path for an IPC socket.
+        :param actor_name: Name of the actor whose endpoint is being accessed
+        :param socket_purpose: Purpose of the socket (control, data, etc.)
+        :param basedir: Base directory used to store the socket
         """
-        # create the pull socket (to communicate with this actor, others
-        # process have to connect a push socket to this socket)
-        self.pull_socket, pull_port = self._create_socket(zmq.PULL, -1)
-
-        # create the control socket (to control this actor, a process have to
-        # connect a pair socket to this socket with the `control` method)
-        self.control_socket, ctrl_port = self._create_socket(zmq.PAIR, 0)
-
-        self.pull_socket_address = LOCAL_ADDR + ':' + str(pull_port)
-        self.control_socket_address = LOCAL_ADDR + ':' + str(ctrl_port)
+        key = f'{actor_name}:{socket_purpose}'
+        digest = blake2b(key.encode('utf-8'), digest_size=16).hexdigest()
+        return Path(basedir) / f'powerapi-ipc-{digest}'
 
-        self._pull_port.value = pull_port
-        self._ctrl_port.value = ctrl_port
-        self._values_available.set()
-
-    def _create_socket(self, socket_type, linger_value):
-        """
-        Create a socket of the given type, bind it to a random port and
-        register it to the poller
-
-        :param int socket_type: type of the socket to open
-        :param int linger_value: -1 mean wait for receive all msg and block
-                                 closing 0 mean hardkill the socket even if msg
-                                 are still here.
-        :return (zmq.Socket, int): the initialized socket and the port where the
-                                   socket is bound
+    def setup(self):
         """
-        socket = zmq.Context.instance().socket(socket_type)
-        socket.setsockopt(zmq.LINGER, linger_value)
-        port_number = socket.bind_to_random_port(LOCAL_ADDR)
-        self.poller.register(socket, zmq.POLLIN)
-        self.logger.debug('Bind socket to %s:%d', LOCAL_ADDR, port_number)
-        return (socket, port_number)
-
-    def receive(self):
+        Initializes the socket interface.
+        This method should only be called by the actor acting as endpoint.
         """
-        Block until a message was received (or until timeout) an return the
-        received messages
+        self._is_endpoint = True
 
-        :return: the list of received messages or None if timeout
-        :rtype: a list of Object or None
-        """
-        events = dict(self.poller.poll(self.timeout))
+        self._control_socket = zmq.Context.instance().socket(zmq.DEALER)
+        self._control_socket.setsockopt(zmq.LINGER, 0)
+        self._control_socket.bind(f'ipc://{self.control_socket_filepath}')
 
-        if len(events) == 0:
-            return None
+        self._data_socket = zmq.Context.instance().socket(zmq.PULL)
+        self._data_socket.bind(f'ipc://{self.data_socket_filepath}')
 
-        return self._recv_serialized(next(iter(events)))
+        self._sockets_poller = zmq.Poller()
+        self._sockets_poller.register(self._control_socket, zmq.POLLIN)
+        self._sockets_poller.register(self._data_socket, zmq.POLLIN)
 
-    def receive_control(self, timeout):
+    def close(self) -> None:
         """
-        Block until a message was received on the control canal (client side)
-        (or until timeout) an return the received messages
-
-        :return: the received message or an None if timeout
-        :rtype: a list of Object
-
+        Closes the socket interface.
         """
-        if self.control_socket is None:
-            raise NotConnectedException
-
-        event = self.control_socket.poll(timeout)
-
-        if event == 0:
-            return None
+        self._sockets_poller = None
 
-        return self._recv_serialized(self.control_socket)
-
-    def close(self):
-        """
-        Close all socket handle by this interface
-        """
-        if self.push_socket is not None:
-            self.push_socket.close()
+        if self._control_socket is not None:
+            self._control_socket.close()
+            self._control_socket = None
 
-        if self.pull_socket is not None:
-            self.pull_socket.close()
+        if self._data_socket is not None:
+            self._data_socket.close()
+            self._data_socket = None
 
-        if self.control_socket is not None:
-            self.control_socket.close()
+        if self._is_endpoint:
+            # Calling `close()` on the bound sockets doesn't remove the file when using the `ipc` transport protocol.
+            # Manually unlinking files after calling `close()` is the most reliable way to fix it.
+            self.control_socket_filepath.unlink(missing_ok=True)
+            self.data_socket_filepath.unlink(missing_ok=True)
 
     @staticmethod
-    def _send_serialized(socket: zmq.Socket, msg):
+    def _send_serialized(socket: zmq.Socket, msg: Any) -> None:
         """
-        Send a message to the given socket.
+        Sends a serialized message to the given socket.
         :param socket: Socket to use
-        :param msg: Message to send
+        :param msg: Message to serialize and send
         """
         socket.send_pyobj(msg)
 
     @staticmethod
-    def _recv_serialized(socket: zmq.Socket):
+    def _recv_serialized(socket: zmq.Socket) -> Any:
         """
-        Receive and returns a message from the given socket.
+        Receive, deserialize and returns a message from the given socket.
         :param socket: Socket to use
         :return: Message received
         """
         return socket.recv_pyobj()
 
-    def connect_data(self):
+    def connect_control(self) -> None:
         """
-        Connect to the pull socket of this actor
-
-        Open a push socket on the process that want to communicate with this
-        actor
+        Connect to the control socket of the actor.
+        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.connect(f'ipc://{self.control_socket_filepath}')
+        self._control_socket.poll(zmq.POLLIN | zmq.POLLOUT)  # Very important, prevents synchronization problems.
 
-        this method shouldn't be called if socket interface was not initialized
-        with the setup method
+    def receive_control(self, timeout: int | None = None) -> Any:
         """
+        Receive a message from the control socket of the actor.
+        :param timeout: Timeout of the operation in milliseconds, if None block indefinitely
+        :return: Message received
+        """
+        if self._control_socket is None:
+            raise NotConnectedException()
 
-        if self.pull_socket_address is None:
-            self._values_available.wait()
-            self.pull_socket_address = LOCAL_ADDR + ':' + str(self._pull_port.value)
-            self.control_socket_address = LOCAL_ADDR + ':' + str(self._ctrl_port.value)
+        if self._control_socket.poll(timeout):
+            return self._recv_serialized(self._control_socket)
 
-        self.push_socket = zmq.Context.instance().socket(zmq.PUSH)
-        self.push_socket.setsockopt(zmq.LINGER, -1)
-        self.push_socket.connect(self.pull_socket_address)
-        self.logger.debug('Connected data socket to %s', self.pull_socket_address)
+        return None
 
-    def connect_control(self):
+    def send_control(self, msg: Any) -> None:
         """
-        Connect to the control socket of this actor
-
-        Open a pair socket on the process that want to control this actor
-        this method shouldn't be called if socket interface was not initialized
-        with the setup method
+        Send a message to the control socket of the actor.
+        :param msg: Message to send
         """
-        if self.pull_socket_address is None:
-            self._values_available.wait()
-            self.pull_socket_address = LOCAL_ADDR + ':' + str(self._pull_port.value)
-            self.control_socket_address = LOCAL_ADDR + ':' + str(self._ctrl_port.value)
+        if self._control_socket is None:
+            raise NotConnectedException()
 
-        self.control_socket = zmq.Context.instance().socket(zmq.PAIR)
-        self.control_socket.setsockopt(zmq.LINGER, 0)
-        self.control_socket.connect(self.control_socket_address)
-        self.logger.debug('Connected control socket to %s', self.control_socket_address)
+        self._send_serialized(self._control_socket, msg)
 
-    def send_control(self, msg):
+    def connect_data(self) -> None:
         """
-        Send a message on the control canal
-
-        :param Object msg: message to send
+        Connect to the data socket of the actor.
+        This method should only be called by actors that wants to communicate with another actor's endpoint.
         """
+        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.
 
-        if self.control_socket is None:
+    def send_data(self, msg: Any) -> None:
+        """
+        Send a message to the data socket of the actor.
+        :param msg: Message to send
+        """
+        if self._data_socket is None:
             raise NotConnectedException()
-        self._send_serialized(self.control_socket, msg)
 
-    def send_data(self, msg):
-        """
-        Send a message on data canal
+        self._send_serialized(self._data_socket, msg)
 
-        :param Object msg: message to send
+    def receive(self, timeout: int | None = None) -> Any:
+        """
+        Receive a message from either the control or the data sockets.
+        This method should only be called by an actor acting as endpoint.
+        :param timeout: Timeout of the operation in milliseconds, if None block indefinitely
+        :return: The received message or None if the timeout is reached
         """
-        if self.push_socket is None:
+        if self._sockets_poller is None:
             raise NotConnectedException()
-        self._send_serialized(self.push_socket, msg)
+
+        for socket, _ in self._sockets_poller.poll(timeout):
+            return self._recv_serialized(socket)
+
+        return None

src/powerapi/actor/socket_interface.py

@@ -66,7 +66,7 @@ def get_corresponding_handler(self, msg: Message) -> Handler:
         """
         try:
             return self.handlers[msg.__class__.__name__]
-        except ValueError as e:
+        except KeyError as e:
             raise UnknownMessageTypeException() from e
 
     def add_handler(self, message_type: type[Message], handler: Handler, include_subclasses: bool = True):