# Copyright (C) 2015. BMW Car IT GmbH. All rights reserved. """Handlers are classes that assist dlt_broker in receiving and filtering DLT messages """ from abc import ABC, abstractmethod from collections import defaultdict import ctypes import logging from multiprocessing import Lock, Process, Value from queue import Empty import socket import time from threading import Thread, Event from dlt.dlt import ( DLTClient, DLT_DAEMON_TCP_PORT, DLT_UDP_MULTICAST_BUFFER_SIZE, DLT_UDP_MULTICAST_FD_BUFFER_SIZE, cDLT_FILE_NOT_OPEN_ERROR, load, py_dlt_client_main_loop, py_dlt_file_main_loop, ) DLT_CLIENT_TIMEOUT = 5 logger = logging.getLogger(__name__) # pylint: disable=invalid-name class DLTTimeValue(object): """Create a share memory to pass the timestamp between processes The type of dlt time is float (4 bytes). There are several ways to send the value between DLTMessageHandler (it's a process) and DLTBroker. Since DLTMessageHandler has to send the value many times every second, choosing a lightweight solution is must. khiz678 studied and evaluated the following solutions for the problem. - multiprocessing.Queue (Queue in the following discussion) - multiprocessing.Pipe (Pipe in the following discussion) - multiprocessing.Value (Value in the following discussion) Value is our final solution. Queue's implementation is based on Pipe (in cpython). If the solution is based on Queue or Pipe, it needs another thread in DLTBroker process to receive and write the value to a local variable. The solution based on Value does not have such problem, only assigns the value to the shared memory directly. khiz678 also did a simple benchmark for the Value solution. It could receive more than 100000 timestamps per second. It's twice faster than Pipe's implementation. """ def __init__(self, default_value=0.0): self._timestamp_mem = Value(ctypes.c_double, default_value) @property def timestamp(self): """Get the seconds from 1970/1/1 0:00:00 :rtype: float """ with self._timestamp_mem.get_lock(): return self._timestamp_mem.value @timestamp.setter def timestamp(self, new_timestamp): with self._timestamp_mem.get_lock(): self._timestamp_mem.value = new_timestamp class DLTFilterAckMessageHandler(Thread): """Receive filter-set ack message and pass it to the corresponding ack queue""" def __init__(self, filter_ack_queue): # (multiprocessing.Queue[Tuple[ContextQueueID, bool]]) -> None super(DLTFilterAckMessageHandler, self).__init__() self.filter_ack_queue = filter_ack_queue self.stop_flag = Event() self.context_map_lock = Lock() self.context_map = {} # Dict[ContextQueueID, Queue.Queue[bool]] def stop(self): """Stops thread execution""" self.stop_flag.set() self.filter_ack_queue.put((None, None)) if self.is_alive(): self.join() def register(self, filter_ack_queue): # (Queue.Queue[bool]) -> None """Register an ack queue""" with self.context_map_lock: self.context_map[id(filter_ack_queue)] = filter_ack_queue def unregister(self, filter_ack_queue): # (Queue.Queue[bool]) -> None """Unregister an ack queue""" with self.context_map_lock: key = id(filter_ack_queue) if key in self.context_map: del self.context_map[key] def run(self): """Run the thread to recieve the message and dispatch it""" while not self.stop_flag.is_set(): queue_ack_id, enable = self.filter_ack_queue.get() if not queue_ack_id: continue with self.context_map_lock: if queue_ack_id in self.context_map: self.context_map[queue_ack_id].put(enable) else: logger.warning("Could not send an ack to the queue: %s", queue_ack_id) class DLTContextHandler(Thread): """Communication layer between the DLTContext instances and DLTMessageHandler Process. This class handles the transfer of messages between the process receiving traces from the DLT Daemon and the DLTContext queues. """ def __init__(self, filter_queue, message_queue): super(DLTContextHandler, self).__init__() self.stop_flag = Event() self.context_map = {} self.lock = Lock() self.filter_queue = filter_queue self.message_queue = message_queue def _make_send_filter_msg(self, queue, filters, is_register, context_filter_ack_queue=None): """Send a filter message to the filter message queue""" queue_id = id(queue) if context_filter_ack_queue: return queue_id, id(context_filter_ack_queue), filters, is_register return queue_id, filters, is_register def register(self, queue, filters=None, context_filter_ack_queue=None): """Register a queue to collect messages matching specific filters :param Queue queue: The new queue to add :param tuple filters: An tuple with (apid, ctid) used to filter messages that go into this queue. """ if filters is None: filters = [(None, None)] queue_id = id(queue) # - unique identifier for this queue with self.lock: self.context_map[queue_id] = (queue, filters) # - inform the DLTMessageHandler process about this new # (queue, filter) pair self.filter_queue.put( self._make_send_filter_msg(queue, filters, True, context_filter_ack_queue=context_filter_ack_queue) ) def unregister(self, queue, context_filter_ack_queue=None): """Remove a queue from set of queues being handled :param Queue queue: The queue to remove """ queue_id = id(queue) _, filters = self.context_map.get(queue_id, (None, None)) if filters: with self.lock: try: del self.context_map[queue_id] except KeyError: pass # - inform the DLTMessageHandler process about removal of this # (queue, filter) pair self.filter_queue.put( self._make_send_filter_msg(queue, filters, False, context_filter_ack_queue=context_filter_ack_queue) ) def run(self): """The thread's main loop""" while not self.stop_flag.is_set(): queue_id, message = None, None try: if self.message_queue.full(): logger.error("message_queue is full ! put() on this queue will block") queue_id, message = self.message_queue.get_nowait() except Empty: pass if message: queue, _ = self.context_map.get(queue_id, (None, None)) if queue: queue.put(message) else: time.sleep(0.01) def stop(self): """Stops thread execution""" self.stop_flag.set() self.filter_queue.close() if self.is_alive(): self.join() class DLTMessageDispatcherBase(ABC, Process): """Base class for different dlt message dispatchers The derived class could dispatch dlt messages from dlt-daemon, or from at-runtime written file. """ def __init__(self, filter_queue, message_queue, mp_stop_event, dlt_time_value=None, filter_ack_queue=None): """ Common members needed for common dispatching behavirours :param Queue filter_queue: contexts for filtering received dlt message :param Queue message_queue: received dlt messages after filtering against context :param multiprocessing.Event mp_stop_event: stop signal for this process :param bool enable_dlt_time: Record the latest dlt message timestamp if enabled. :param bool filter_ack_queue: acks for accepting contexts """ super().__init__() self.filter_queue = filter_queue self.filter_ack_queue = filter_ack_queue self.message_queue = message_queue self.mp_stop_flag = mp_stop_event # - dict mapping filters to queue ids self.context_map = defaultdict(list) self._dlt_time_value = dlt_time_value def _process_filter_queue(self): """Check if filters have been added or need to be removed""" while not self.filter_queue.empty(): queue_ack_id = None msg = self.filter_queue.get_nowait() logger.debug("Process filter queue message: %s", msg) if isinstance(msg, tuple) and len(msg) == 4: queue_id, queue_ack_id, filters, add = msg else: queue_id, filters, add = msg if add: for apid_ctid in filters: self.context_map[apid_ctid].append(queue_id) else: try: for apid_ctid in filters: self.context_map[apid_ctid].remove(queue_id) if not self.context_map[apid_ctid]: del self.context_map[apid_ctid] except (KeyError, ValueError): # - queue_id already removed or not inserted pass if self.filter_ack_queue and queue_ack_id: logger.debug("Send filter ack message: queue_ack_id: %s, add: %s", queue_ack_id, add) self.filter_ack_queue.put((queue_ack_id, add)) @abstractmethod def is_valid_message(self, message): """Validate if the received message is a valid message according to AUTOSAR doc""" return True def handle(self, message): """Function to be called for every message received :param DLTMessage message: received new DLTMessage instance :returns: True if the loop should continue, False to stop the loop and exit :rtype: bool """ self._process_filter_queue() if self.is_valid_message(message): # Dispatch the message msg_ctx = ((message.apid, message.ctid), (None, None), (message.apid, None), (None, message.ctid)) qids = ( queue_id for filters, queue_ids in self.context_map.items() for queue_id in queue_ids if filters in msg_ctx ) for queue_id in qids: if self.message_queue.full(): logger.error("message_queue is full ! put() on this queue will block") self.message_queue.put((queue_id, message)) # Send the message's timestamp if self._dlt_time_value: self._dlt_time_value.timestamp = message.storage_timestamp return not self.mp_stop_flag.is_set() @abstractmethod def run(self) -> None: pass def break_blocking_main_loop(self): """All message dispatchers need a main loop to fetch dlt messages from source. If it could constantly dispatch messages, then the main loop will not get into blocking state. Only when no more message could not be dispatched, the main loop would get into blocking state. Not all message dispatchers need to implement this method """ pass class DLTFileSpinner(DLTMessageDispatcherBase): """Process receiving the DLT messages and handing them to DLTContextHandler This process instance is responsible for collecting messages from the at-runtime written dlt log, tagging them with the correct queue id and placing them on the messages queue. """ def __init__( self, filter_queue, message_queue, mp_stop_event, file_name, dlt_time_value=None, filter_ack_queue=None ): super().__init__(filter_queue, message_queue, mp_stop_event, dlt_time_value, filter_ack_queue) self.file_name = file_name self.dlt_reader = load(filename=self.file_name, live_run=True) def is_valid_message(self, message): """According to AUTOSAR doc, message with empty apid and empty ctid is still valid""" return message is not None def run(self): """DLTFileSpinner worker method""" logger.info("Start to process dlt file %s", self.file_name) # Even though dlt connector for ioc should only be instantiated after successful SerialConsole with fibex, # the corner case of not-existing dlt file will still be handled here with max 5 retires retries_for_non_existing_file = 5 while not self.mp_stop_flag.is_set(): try: logger.debug("py_dlt_file_main_loop") res = py_dlt_file_main_loop(self.dlt_reader, callback=self.handle) if res is False and not self.mp_stop_flag.is_set(): # main loop returned False logger.error("Too many bad messages read from %s", self.file_name) self.mp_stop_flag.set() break except KeyboardInterrupt: logger.debug("main loop manually interrupted") break except IOError as err: if str(err) == cDLT_FILE_NOT_OPEN_ERROR: # Not every time of non-existing file, cDLTFile will report error # Sometimes, it simply works through without issue. # So, no unittest could be done for this error handling if retries_for_non_existing_file == 0: logger.error("After retries, dlt file %s still does not exist", self.file_name) raise err logger.warning( "DLT file %s does not exist, will try %d times again", self.file_name, retries_for_non_existing_file, ) retries_for_non_existing_file = retries_for_non_existing_file - 1 time.sleep(1) else: raise err except Exception: # pylint: disable=broad-except logger.exception("Exception during the DLT message receive") logger.debug("DLTFileSpinner starts to quit...") if not self.dlt_reader.stop_reading_proc.is_set(): self.dlt_reader.stop_reading_proc.set() self.message_queue.close() logger.info("DLTFileSpinner worker execution complete") def break_blocking_main_loop(self): """A big user for DLTFileSpinner is IOC dlt, which does not have so many dlt messages as HU, so it is quite easy for the main loop to get into blocking state, at the moment that no more dlt messages could be dispatched. """ logger.debug("Stop iterating to file %s", self.file_name) self.dlt_reader.stop_reading_proc.set() class DLTMessageHandler(DLTMessageDispatcherBase): """Process receiving the DLT messages and handing them to DLTContextHandler This process instance is responsible for collecting messages from the DLT daemon, tagging them with the correct queue id and placing them on the messages queue. """ def __init__( self, filter_queue, message_queue, mp_stop_event, client_cfg, dlt_time_value=None, filter_ack_queue=None ): super().__init__(filter_queue, message_queue, mp_stop_event, dlt_time_value, filter_ack_queue) self._ip_address = client_cfg["ip_address"] self._port = client_cfg.get("port", DLT_DAEMON_TCP_PORT) self._filename = client_cfg.get("filename") self.verbose = client_cfg.get("verbose", 0) self.timeout = client_cfg.get("timeout", DLT_CLIENT_TIMEOUT) self._client = None self.tracefile = None self.last_connected = time.time() self.last_message = time.time() - 120.0 self._udp_fd_buffer_size_bytes = client_cfg.get("udp_fd_buffer_size_bytes", DLT_UDP_MULTICAST_FD_BUFFER_SIZE) self._udp_buffer_size_bytes = client_cfg.get("udp_buffer_size_bytes", DLT_UDP_MULTICAST_BUFFER_SIZE) def is_valid_message(self, message): return message and (message.apid != "" or message.ctid != "") def _client_connect(self): """Create a new DLTClient :param int timeout: Time in seconds to wait for connection. :returns: True if connected, False otherwise :rtype: bool """ if self.verbose: logger.debug( "Creating DLTClient (ip_address='%s', Port='%s', logfile='%s')", self._ip_address, self._port, self._filename, ) self._client = DLTClient( servIP=self._ip_address, port=self._port, verbose=self.verbose, udp_fd_buffer_size_bytes=self._udp_fd_buffer_size_bytes, udp_buffer_size_bytes=self._udp_buffer_size_bytes, ) connected = self._client.connect(self.timeout) if connected: logger.info("DLTClient connected to %s", self._client.servIP) return connected def run(self): """DLTMessageHandler worker method""" if self._filename is not None: logger.info("Opening the DLT trace file '%s'", self._filename) self.tracefile = open(self._filename, mode="ab", buffering=False) while not self.mp_stop_flag.is_set(): exception_occured = False if not self._client_connect(): # keep trying to reconnect, until we either successfully # connect or the stop_flag is set if time.time() - self.last_message > 60: # Once per minute log that we still have no DLT Connection logger.info( "DLT connection to %s missing since %s seconds", self._ip_address, time.time() - self.last_connected, ) self.last_message = time.time() continue try: if self.last_connected: logger.info( "DLT connection to %s re-established after %s seconds", self._ip_address, time.time() - self.last_connected, ) self.last_connected = time.time() res = py_dlt_client_main_loop(self._client, verbose=0, callback=self.handle, dumpfile=self.tracefile) if res is False and not self.mp_stop_flag.is_set(): # main loop returned False logger.warning("DLT connection to %s lost. Restarting DLT client", self._ip_address) self.last_connected = time.time() self.last_message = time.time() exception_occured = True except KeyboardInterrupt: exception_occured = True logger.debug("main loop manually interrupted") break except socket.timeout as exc: exception_occured = True logger.error("socket timeout error") logger.debug(exc) except Exception: # pylint: disable=broad-except exception_occured = True logger.exception("Exception during the DLT message receive") finally: if exception_occured: logger.debug("Closing open socket connections.") self._client.disconnect() self.message_queue.close() self._client.disconnect() logger.info("DLTMessageHandler worker execution complete")