diff --git a/ipsframework/_internal/__init__.py b/ipsframework/_internal/__init__.py index fab43796..31bfe111 100644 --- a/ipsframework/_internal/__init__.py +++ b/ipsframework/_internal/__init__.py @@ -1 +1 @@ -"""Code which should ONLY be imported by the IPS framework itself.""" \ No newline at end of file +"""Code which should ONLY be imported by the IPS framework itself.""" diff --git a/ipsframework/_internal/bridges/portal_bridge.py b/ipsframework/_internal/bridges/portal_bridge.py index fe90313b..3ebe4df0 100644 --- a/ipsframework/_internal/bridges/portal_bridge.py +++ b/ipsframework/_internal/bridges/portal_bridge.py @@ -25,6 +25,11 @@ _portal_logger = logging.getLogger('ipsframework.bridges.portal_bridge') +EVENT_MESSAGE_TYPE = 'events' +DATA_MESSAGE_TYPE = 'data' +NOTEBOOK_MESSAGE_TYPE = 'notebook' +ENSEMBLE_MESSAGE_TYPE = 'ensemble' + def send_post(conn: Connection, stop: EventType, url: str): fail_count = 0 @@ -34,6 +39,7 @@ def send_post(conn: Connection, stop: EventType, url: str): headers={'Content-Type': 'application/json'}, ) + # TODO - try to figure out ways to send the sim_names back for events, not essential though while True: if conn.poll(0.1): msgs = [] @@ -45,13 +51,13 @@ def send_post(conn: Connection, stop: EventType, url: str): _portal_logger.debug('HTTP event took %s seconds', time.time() - start_walltime) except MaxRetryError as e: fail_count += 1 - conn.send((999, f'Max retry error: {e}')) + conn.send((EVENT_MESSAGE_TYPE, '', 999, f'Max retry error: {e}')) else: - conn.send((resp.status, resp.data.decode())) + conn.send((EVENT_MESSAGE_TYPE, '', resp.status, resp.data.decode())) fail_count = 0 if fail_count >= MAX_RETRIES: - conn.send((-1, 'Too many consecutive failed connections')) + conn.send((EVENT_MESSAGE_TYPE, '', -1, 'Too many consecutive failed connections')) break elif stop.is_set(): break @@ -86,13 +92,13 @@ def send_jupyter_notebook(conn: Connection, stop: EventType, url: str, api_key: ) except MaxRetryError as e: fail_count += 1 - conn.send((999, f'Max retry error: {e}')) + conn.send((NOTEBOOK_MESSAGE_TYPE, next_val['component_id'], 999, f'Max retry error: {e}')) else: - conn.send((resp.status, resp.data.decode())) + conn.send((NOTEBOOK_MESSAGE_TYPE, next_val['component_id'], resp.status, resp.data.decode())) fail_count = 0 if fail_count >= MAX_RETRIES: - conn.send((-1, 'Too many consecutive failed connections')) + conn.send((NOTEBOOK_MESSAGE_TYPE, next_val['component_id'], -1, 'Too many consecutive failed connections')) break elif stop.is_set(): break @@ -137,13 +143,13 @@ def send_jupyter_notebook_data(conn: Connection, stop: EventType, url: str, api_ ) except (MaxRetryError, OSError) as e: fail_count += 1 - conn.send((999, f'Max retry error: {e}')) + conn.send((DATA_MESSAGE_TYPE, next_val['component_id'], 999, f'Max retry error: {e}')) else: - conn.send((resp.status, resp.data.decode())) + conn.send((DATA_MESSAGE_TYPE, next_val['component_id'], resp.status, resp.data.decode())) fail_count = 0 if fail_count >= MAX_RETRIES: - conn.send((-1, 'Too many consecutive failed connections')) + conn.send((DATA_MESSAGE_TYPE, next_val['component_id'], -1, 'Too many consecutive failed connections')) break elif stop.is_set(): break @@ -183,13 +189,13 @@ def send_ensemble_variables(conn: Connection, stop: EventType, url: str, api_key ) except (MaxRetryError, OSError) as e: fail_count += 1 - conn.send((999, f'Max retry error: {e}')) + conn.send((ENSEMBLE_MESSAGE_TYPE, next_val['component_id'], 999, f'Max retry error: {e}')) else: - conn.send((resp.status, resp.data.decode())) + conn.send((ENSEMBLE_MESSAGE_TYPE, next_val['component_id'], resp.status, resp.data.decode())) fail_count = 0 if fail_count >= MAX_RETRIES: - conn.send((-1, 'Too many consecutive failed connections')) + conn.send((ENSEMBLE_MESSAGE_TYPE, next_val['component_id'], -1, 'Too many consecutive failed connections')) break elif stop.is_set(): break @@ -249,6 +255,8 @@ def __init__(self, services, config): self.url_manager_jupyter_data = None self.url_manager_ensemble_uploads = None + ### COMPONENT FUNCTIONS (public) ### + def init(self, timestamp=0.0, **keywords): """ Try to connect to the portal, subscribe to *_IPS_MONITOR* events and @@ -263,6 +271,7 @@ def init(self, timestamp=0.0, **keywords): except AttributeError: pass + # logging configuration if self.services.fwk.log_level == logging.DEBUG: log_level = logging.DEBUG else: @@ -285,11 +294,29 @@ def step(self, timestamp=0.0, **keywords): """ while not self.done: self.services.process_events() + self._check_url_manager_responses() time.sleep(0.5) def finalize(self, timestamp=0.0, **keywords): pass + def terminate(self, status: Literal[0, 1]): + """ + Clean up services and call :py:obj:`sys_exit`. + """ + if self.childProcess: + self.childProcess.terminate() + if self.url_manager_jupyter_data: + self.url_manager_jupyter_data.childProcess.terminate() + if self.url_manager_jupyter_notebook: + self.url_manager_jupyter_notebook.childProcess.terminate() + if self.url_manager_ensemble_uploads: + self.url_manager_ensemble_uploads.childProcess.terminate() + + Component.terminate(self, status) + + ### SUBSCRIPTION CHANNELS (public) ### + def process_event(self, topicName, theEvent): """ Process a single event *theEvent* on topic *topicName*. @@ -301,25 +328,30 @@ def process_event(self, topicName, theEvent): portal_data['sim_name'] = event_body['real_sim_name'] except KeyError: portal_data['sim_name'] = sim_name + portal_data['component_id'] = event_body.get('component_id', '') + + event_type = portal_data['eventtype'] - if portal_data['eventtype'] == 'IPS_START': + if event_type == 'IPS_START': sim_root = event_body['sim_root'] self.init_simulation(sim_name, sim_root, portal_data['portal_runid']) sim_data = self.sim_map[sim_name] - if portal_data['eventtype'] == 'PORTALBRIDGE_UPDATE_TIMESTAMP': + if event_type == 'PORTALBRIDGE_UPDATE_TIMESTAMP': sim_data.phys_time_stamp = portal_data['phystimestamp'] return else: portal_data['phystimestamp'] = sim_data.phys_time_stamp - if portal_data['eventtype'] == 'PORTAL_REGISTER_NOTEBOOK': + ### Portal data events ### + + if event_type == 'PORTAL_REGISTER_NOTEBOOK': with open(portal_data['data_source'], 'rb') as f: portal_data['data'] = f.read() - self.send_jupyter_notebook(sim_data, portal_data) + self._send_jupyter_notebook(sim_data, portal_data) return - if portal_data['eventtype'] == 'PORTAL_ADD_JUPYTER_DATA': + if event_type == 'PORTAL_ADD_JUPYTER_DATA': data_source = portal_data['data_source'] if os.path.isdir(data_source): # assume that we are handling a specialized data format, and do not use compression @@ -330,21 +362,21 @@ def process_event(self, topicName, theEvent): tar.add(data_source, arcname=os.path.basename(data_source)) portal_data['data_source'] = tarpath - self.send_notebook_data(sim_data, portal_data) + self._send_notebook_data(sim_data, portal_data) return - if portal_data['eventtype'] == 'PORTAL_UPLOAD_ENSEMBLE_PARAMS': - self.send_ensemble_variables(sim_data, portal_data) + if event_type == 'PORTAL_UPLOAD_ENSEMBLE_PARAMS': + self._send_ensemble_variables(sim_data, portal_data) return portal_data['portal_runid'] = sim_data.portal_runid - if portal_data['eventtype'] == 'IPS_SET_MONITOR_URL': + if event_type == 'IPS_SET_MONITOR_URL': sim_data.monitor_url = portal_data['vizurl'] elif sim_data.monitor_url: portal_data['vizurl'] = sim_data.monitor_url - if portal_data['eventtype'] == 'IPS_START' and 'parent_portal_runid' not in portal_data: + if event_type == 'IPS_START' and 'parent_portal_runid' not in portal_data: portal_data['parent_portal_runid'] = sim_data.parent_portal_runid portal_data['seqnum'] = sim_data.counter @@ -369,27 +401,29 @@ def process_event(self, topicName, theEvent): except OSError: pass - self.check_send_post_responses(polling_timeout) + self._check_send_post_responses(polling_timeout) - if portal_data['eventtype'] == 'IPS_END': + if event_type == 'IPS_END': del self.sim_map[sim_name] if len(self.sim_map) == 0: + self.done = True if self.childProcess: self.childProcessStop.set() self.childProcess.join() - self.check_send_post_responses(0.0) - self.done = True + self._check_send_post_responses(0.0) self.services.debug('No more simulation to monitor - exiting') time.sleep(1) - def check_send_post_responses(self, polling_timeout: float = 0.0): + ### LOCAL FUNCTIONS (private) ### + + def _check_send_post_responses(self, polling_timeout: float = 0.0): if self.parent_conn is None: self.services.warning('Giving up on polling for portal responses') return while self.parent_conn.poll(timeout=polling_timeout): try: - code, msg = self.parent_conn.recv() + _msg_type, _component_id, code, msg = self.parent_conn.recv() except (EOFError, OSError): break @@ -418,28 +452,29 @@ def check_send_post_responses(self, polling_timeout: float = 0.0): except (TypeError, json.decoder.JSONDecodeError): pass - def http_req_and_response(self, manager: UrlRequestProcessManager, event_data): - try: - manager.parent_conn.send(event_data) - except OSError: - pass - - while manager.parent_conn.poll(): - try: - code, msg = manager.parent_conn.recv() - except (EOFError, OSError): - break - - if code == -1: - # disable portal, stop trying to send more data - self.portal_url = '' - self.services.error('Disabling portal because: %s', msg) - elif code >= 400: - self.services.error('Portal Error: %d %s', code, msg) - else: - self.services.debug('Portal Response: %d %s', code, msg) - - def send_jupyter_notebook(self, sim_data: PortalSimulationData, event_data): + def _check_url_manager_responses(self): + """poll all data API checks""" + for manager in [self.url_manager_jupyter_notebook, self.url_manager_jupyter_data, self.url_manager_ensemble_uploads]: + if manager is None: + continue + while manager.parent_conn.poll(): + try: + msg_type, component_id, code, msg = manager.parent_conn.recv() + except (EOFError, OSError): + break + + if code >= 400: + self.services.error('Portal Error: %d %s', code, msg) + elif code == -1: + # disable portal, stop trying to send more data + self.portal_url = '' + self.services.error('Disabling portal because: %s', msg) + else: + self.services.debug('Portal Response: %d %s', code, msg) + if msg_type == ENSEMBLE_MESSAGE_TYPE: + self.services.publish(f'_IPS_{component_id}', '_IPS_PORTAL_UPLOAD_ENSEMBLE_PARAMS_SUCCESS', 'true') + + def _send_jupyter_notebook(self, sim_data: PortalSimulationData, event_data): if self.portal_url and self.portal_api_key: if not self.url_manager_jupyter_notebook: self.url_manager_jupyter_notebook = UrlRequestProcessManager( @@ -448,9 +483,12 @@ def send_jupyter_notebook(self, sim_data: PortalSimulationData, event_data): self.portal_api_key, self.USER, ) - self.http_req_and_response(self.url_manager_jupyter_notebook, event_data) + try: + self.url_manager_jupyter_notebook.parent_conn.send(event_data) + except OSError: + self.services.error('Failed to send notebook to portal %s', event_data) - def send_notebook_data(self, sim_data: PortalSimulationData, event_data): + def _send_notebook_data(self, sim_data: PortalSimulationData, event_data): if self.portal_url and self.portal_api_key: if not self.url_manager_jupyter_data: self.url_manager_jupyter_data = UrlRequestProcessManager( @@ -459,9 +497,12 @@ def send_notebook_data(self, sim_data: PortalSimulationData, event_data): self.portal_api_key, self.USER, ) - self.http_req_and_response(self.url_manager_jupyter_data, event_data) + try: + self.url_manager_jupyter_data.parent_conn.send(event_data) + except OSError: + self.services.error('Failed to send notebook data to portal %s', event_data) - def send_ensemble_variables(self, sim_data: PortalSimulationData, event_data): + def _send_ensemble_variables(self, sim_data: PortalSimulationData, event_data): if self.portal_url and self.portal_api_key: if not self.url_manager_ensemble_uploads: self.url_manager_ensemble_uploads = UrlRequestProcessManager( @@ -470,7 +511,10 @@ def send_ensemble_variables(self, sim_data: PortalSimulationData, event_data): self.portal_api_key, self.USER, ) - self.http_req_and_response(self.url_manager_ensemble_uploads, event_data) + try: + self.url_manager_ensemble_uploads.parent_conn.send(event_data) + except OSError: + self.services.error('Failed to send ensemble variables to portal %s', event_data) def init_simulation(self, sim_name: str, sim_root: str, portal_runid: str): """ @@ -507,18 +551,3 @@ def init_simulation(self, sim_name: str, sim_root: str, portal_runid: str): self.first_portal_runid = sim_data.portal_runid self.sim_map[sim_name] = sim_data - - def terminate(self, status: Literal[0, 1]): - """ - Clean up services and call :py:obj:`sys_exit`. - """ - if self.childProcess: - self.childProcess.terminate() - if self.url_manager_jupyter_data: - self.url_manager_jupyter_data.childProcess.terminate() - if self.url_manager_jupyter_notebook: - self.url_manager_jupyter_notebook.childProcess.terminate() - if self.url_manager_ensemble_uploads: - self.url_manager_ensemble_uploads.childProcess.terminate() - - Component.terminate(self, status) diff --git a/ipsframework/_internal/definitions.py b/ipsframework/_internal/definitions.py index 4d0dead3..96b79e1b 100644 --- a/ipsframework/_internal/definitions.py +++ b/ipsframework/_internal/definitions.py @@ -33,4 +33,4 @@ 'IPS_UPDATE_STATE', 'IPS_UPDATE_TIME_STAMP', '', -] \ No newline at end of file +] diff --git a/ipsframework/component.py b/ipsframework/component.py index e04958da..45ea5213 100644 --- a/ipsframework/component.py +++ b/ipsframework/component.py @@ -7,8 +7,10 @@ import sys import weakref from copy import copy +from multiprocessing import Queue from typing import TYPE_CHECKING, Any, Dict, Literal +from .componentRegistry import ComponentID from .messages import Message, MethodResultMessage if TYPE_CHECKING: @@ -31,8 +33,8 @@ def __init__(self, services, config: Dict[str, Any]): """ Set up config values and reference to services. """ - self.__component_id = None - self.__invocation_q = None + self.__component_id: ComponentID = None # type: ignore + self.__invocation_q: Queue = None # type: ignore self.__services: ServicesProxy = weakref.proxy(services) self.__config = config self.__start_time = 0.0 @@ -57,7 +59,7 @@ def __copy__(self): setattr(result, k, copy(v)) return result - def __initialize__(self, component_id, invocation_q, start_time=0.0): + def __initialize__(self, component_id: ComponentID, invocation_q: Queue, start_time: float = 0.0): """ Establish connection to *invocation_q*. """ @@ -124,6 +126,9 @@ def __run__(self): self.services._init_event_service() + # the topic prefix must start with '_IPS_' to be a reserved topic + self.services.subscribe(f'_IPS_{self.__component_id.get_serialization()}', self.services._component_id_subscription_callback) + while True: msg = self.__invocation_q.get() self.services.debug('Received Message ') diff --git a/ipsframework/services.py b/ipsframework/services.py index a72f4336..0d805711 100644 --- a/ipsframework/services.py +++ b/ipsframework/services.py @@ -444,6 +444,9 @@ def __init__(self, fwk, fwk_in_q: Queue, svc_response_q: Queue, sim_conf: dict[s self.task_map: dict[int, RunningTask] = {} self.workdir = '' self.full_comp_id = '' + """full_comp_id is a unique identifier for the component, which is used as the directory name on the filesystem.""" + self._ips_serialized_component_id = '' + """_ips_serialized_component_id is used across the IPS message bus as the topic to communicate to this specific component""" self.logger: logging.Logger = None # type: ignore self.start_time = time.time() self.cur_time = self.start_time @@ -480,6 +483,9 @@ def __init__(self, fwk, fwk_in_q: Queue, svc_response_q: Queue, sim_conf: dict[s self._portal_runid_event = threading.Event() """Thread-safe means to detect if self._portal_runid has been set""" + self._ensemble_verification_check = False + """If using the Portal and calling run_ensemble(), the portal can wait for this value to be set to True to verify that the ensemble parameters were set on the Portal.""" + def __initialize__(self, component_ref): """ Initialize the service proxy object, connecting it to its associated @@ -491,6 +497,7 @@ def __initialize__(self, component_ref): self.component_ref = weakref.proxy(component_ref) conf = self.component_ref.config self.full_comp_id = '_'.join([conf['CLASS'], conf['SUB_CLASS'], conf['NAME'], str(self.component_ref.component_id.get_seq_num())]) + self._ips_serialized_component_id = self.component_ref.component_id.get_serialization() # # Set up logging path to the IPS logging daemon # @@ -555,6 +562,15 @@ def _init_event_service(self) -> None: initialize_event_service(self) self.event_service = eventManager(self.component_ref) + def _component_id_subscription_callback(self, topicName: str, theEvent) -> None: + """ + All components subscribe to a topic based on their component_id. This function is the callback function for that subscription. + """ + event_types = list(theEvent.getHeader().keys()) + self.debug('_component_id_subscription_callback(): component_id=%s topic=%s event_header=%s event_body=%s', self._ips_serialized_component_id, topicName, theEvent.getHeader(), theEvent.getBody()) + if '_IPS_PORTAL_UPLOAD_ENSEMBLE_PARAMS_SUCCESS' in event_types: + self._ensemble_verification_check = True + def _get_elapsed_time(self) -> float: """ Return total elapsed time since simulation started in seconds @@ -739,6 +755,7 @@ def _send_monitor_event( event_data = {} event_data['sim_name'] = self.sim_conf['__PORTAL_SIM_NAME'] event_data['real_sim_name'] = self.sim_name + event_data['component_id'] = self._ips_serialized_component_id event_data['portal_data'] = portal_data self.publish('_IPS_MONITOR', 'PORTAL_EVENT', event_data) @@ -1219,7 +1236,7 @@ def wait_task_nonblocking(self, task_id: int) -> Union[int, None]: if task_retval is None: if task.start_time + task.timeout < time.time(): self.kill_task(task_id) - self._send_monitor_event('IPS_TASK_END', 'task_id = %s TIMEOUT elapsed time = %.2f S' % (str(task_id), time.time() - task.start_time)) + self._send_monitor_event('IPS_TASK_END', 'source_func = wait_task_nonblocking, task_id = %s TIMEOUT elapsed time = %.2f S' % (str(task_id), time.time() - task.start_time)) return -1 else: return None @@ -1265,9 +1282,9 @@ def wait_task(self, task_id: int, timeout: int = -1, delay: int = 1) -> int: if task_retval is None: task.process.kill() task_retval = task.process.wait() - event_comment = 'task_id = %s TIMEOUT elapsed time = %.2f S' % (str(task_id), finish_time - task.start_time) + event_comment = 'source_func = wait_task, task_id = %s TIMEOUT elapsed time = %.2f S' % (str(task_id), finish_time - task.start_time) else: - event_comment = 'task_id = %s elapsed time = %.2f S' % (str(task_id), finish_time - task.start_time) + event_comment = 'source_func = wait_task, task_id = %s elapsed time = %.2f S' % (str(task_id), finish_time - task.start_time) self._send_monitor_event( 'IPS_TASK_END', @@ -2046,6 +2063,7 @@ def update_time_stamp(self, new_time_stamp=-1) -> None: event_data = {} event_data['sim_name'] = self.sim_conf['__PORTAL_SIM_NAME'] event_data['real_sim_name'] = self.sim_name + event_data['component_id'] = self._ips_serialized_component_id portal_data = {} portal_data['phystimestamp'] = new_time_stamp @@ -2180,6 +2198,7 @@ def initialize_jupyter_notebook( event_data = {} event_data['sim_name'] = self.sim_conf['__PORTAL_SIM_NAME'] event_data['real_sim_name'] = self.sim_name + event_data['component_id'] = self._ips_serialized_component_id portal_data: dict[str, Any] = {} portal_data['eventtype'] = 'PORTAL_REGISTER_NOTEBOOK' @@ -2218,6 +2237,7 @@ def add_analysis_data_files(self, current_data_file_paths: list[str], timestamp: event_data = {} event_data['sim_name'] = self.sim_conf['__PORTAL_SIM_NAME'] event_data['real_sim_name'] = self.sim_name + event_data['component_id'] = self._ips_serialized_component_id portal_data: dict[str, Any] = {} portal_data['eventtype'] = 'PORTAL_ADD_JUPYTER_DATA' @@ -2790,6 +2810,7 @@ def send_ensemble_instance_to_portal(ensemble_name: str, data_path: Path) -> Non event_data = {} event_data['sim_name'] = self.sim_conf['__PORTAL_SIM_NAME'] event_data['real_sim_name'] = self.sim_name + event_data['component_id'] = self._ips_serialized_component_id portal_data: dict[str, Any] = {} portal_data['eventtype'] = 'PORTAL_UPLOAD_ENSEMBLE_PARAMS' @@ -2800,8 +2821,20 @@ def send_ensemble_instance_to_portal(ensemble_name: str, data_path: Path) -> Non portal_data['username'] = self.get_config_param('USER') portal_data['portal_runid'] = self._portal_runid event_data['portal_data'] = portal_data + self.publish('_IPS_MONITOR', 'PORTAL_UPLOAD_ENSEMBLE_PARAMS', event_data) self._send_monitor_event('IPS_PORTAL_UPLOAD_ENSEMBLE_PARAMS', f'NAME = {name}') + + # wait on a confirmation from the portal before proceeding with launching the ensembles + ensemble_portal_wait_time = 10.0 + while not self._ensemble_verification_check: + time.sleep(1.0) + ensemble_portal_wait_time -= 1.0 + if ensemble_portal_wait_time <= 0.0: + self.warning('Could not confirm ensemble parameters upload, proceeding with task submission anyway') + break + self.process_events() + self._ensemble_verification_check = False # reset in case run_ensemble is called again self.info(f'Preparing to run ensembles in {run_dir}')