diff --git a/ipsframework/_internal/bridges/portal_bridge.py b/ipsframework/_internal/bridges/portal_bridge.py index 8a75a013..a683324e 100644 --- a/ipsframework/_internal/bridges/portal_bridge.py +++ b/ipsframework/_internal/bridges/portal_bridge.py @@ -21,7 +21,7 @@ from ipsframework import Component -MAX_RETRIES = 10 +MAX_RETRIES = 3 _portal_logger = logging.getLogger('ipsframework.bridges.portal_bridge') @@ -36,7 +36,9 @@ def send_post(conn: Connection, stop: EventType, url: str): fail_count = 0 http = PoolManager( - retries=Urllib3Retry(total=MAX_RETRIES, backoff_factor=1, respect_retry_after_header=True), + retries=Urllib3Retry( + total=MAX_RETRIES, backoff_factor=0.25, respect_retry_after_header=True + ), headers={'Content-Type': 'application/json'}, ) @@ -68,7 +70,9 @@ def send_jupyter_notebook(conn: Connection, stop: EventType, url: str, api_key: fail_count = 0 http = PoolManager( - retries=Urllib3Retry(total=MAX_RETRIES, backoff_factor=1, respect_retry_after_header=True) + retries=Urllib3Retry( + total=MAX_RETRIES, backoff_factor=0.25, respect_retry_after_header=True + ) ) while True: @@ -129,7 +133,9 @@ def send_jupyter_notebook_data( fail_count = 0 http = PoolManager( - retries=Urllib3Retry(total=MAX_RETRIES, backoff_factor=1, respect_retry_after_header=True) + retries=Urllib3Retry( + total=MAX_RETRIES, backoff_factor=0.25, respect_retry_after_header=True + ) ) while True: @@ -195,7 +201,9 @@ def send_ensemble_variables( fail_count = 0 http = PoolManager( - retries=Urllib3Retry(total=MAX_RETRIES, backoff_factor=1, respect_retry_after_header=True) + retries=Urllib3Retry( + total=MAX_RETRIES, backoff_factor=0.25, respect_retry_after_header=True + ) ) while True: diff --git a/ipsframework/component_registry.py b/ipsframework/component_registry.py index 32306831..d311bb09 100644 --- a/ipsframework/component_registry.py +++ b/ipsframework/component_registry.py @@ -147,7 +147,7 @@ def add_entry( print( 'Error creating component registry entry for ', key, ' : ', str(e), file=sys.stderr ) - raise e + raise def remove_entry(self, component_id): key = component_id.get_serialization() diff --git a/ipsframework/debug.py b/ipsframework/debug.py index ec7c8612..a68af828 100644 --- a/ipsframework/debug.py +++ b/ipsframework/debug.py @@ -6,42 +6,22 @@ variable 'IPSES_DEBUG' is defined. """ +import logging import os +_logger = logging.getLogger(__name__) -class Debug: # pragma: no cover - def __init__(self): - self.file = None - if 'IPSES_DEBUG' in os.environ: - self.file = open('debug.out', 'w') +if 'IPSES_DEBUG' in os.environ: + _logger.setLevel(logging.DEBUG) + _logger.addHandler(logging.FileHandler('debug.out', mode='w')) +else: + _logger.setLevel(logging.WARNING) - def output(self, s: str, id1=0, id2=0): - if self.file: - tmp = '' - if id1 != 0: - """ one subscriber/listener """ - if id2 == 0: - tmp += ', id = ' + str(id1) - else: - tmp += ', listenerid = ' + str(id1) + ', subscriberid = ' + str(id2) - self.file.write(s + tmp + '\n') - - def msg(self, s1: str, ret=99, s2=''): - if self.file: - if s2 == '': - if ret != 99: - self.file.write(s1 + ' ' + str(ret) + '\n') - else: - self.file.write(s1 + '\n') - elif ret != 99: - self.file.write(s1 + ' ' + str(ret) + ' ' + s2 + '\n') - else: - self.file.write(s1 + ' ' + s2 + '\n') - - def __del__(self): - if self.file: - self.file.close() - - -debug = Debug() +def output(s: str, id1=0, id2=0): + if id1 != 0: + if id2 == 0: + s += ', id = ' + str(id1) + else: + s += ', listenerid = ' + str(id1) + ', subscriberid = ' + str(id2) + _logger.debug(s) diff --git a/ipsframework/event_service.py b/ipsframework/event_service.py index dd3c6170..bf68431c 100644 --- a/ipsframework/event_service.py +++ b/ipsframework/event_service.py @@ -9,7 +9,7 @@ """ from .cca_es_spec import Event, EventServiceError, Topic -from .debug import debug +from .debug import output as debug_output from .topic_manager import TopicManager @@ -74,27 +74,27 @@ def process_service_request(self, msg): method = getattr(self, msg.target_method) return method(*msg.args) - """""" """PublisherEventService methods start here""" """""" + """PublisherEventService methods start here""" def get_topic(self, topic_name): """Add an entry to the topicDirectory for a new topic.""" if topic_name not in self.topicDirectory: - debug.output('get_topic %s' % topic_name) + debug_output('get_topic %s' % topic_name) self.topicDirectory[topic_name] = TopicManager() return Topic(topic_name) def exists_topic(self, topic_name): return topic_name in self.topicDirectory - """""" """PublisherEventService methods end here""" """""" + """PublisherEventService methods end here""" - """""" """SubscriberEventService methods start here""" """""" + """SubscriberEventService methods start here""" def register_subscriber(self): self.numSubscribers += 1 subscriberid = self.numSubscribers self.subscriberDirectory[subscriberid] = {} - debug.output('Subscriber registered', subscriberid) + debug_output('Subscriber registered', subscriberid) return subscriberid """ @@ -108,7 +108,7 @@ def register_subscriber(self): def unregister_subscriber(self, subscriberid): listener_list = [] if subscriberid in self.subscriberDirectory: - debug.output('\n\n------Subscriber is unregistering', subscriberid) + debug_output('\n\n------Subscriber is unregistering', subscriberid) """ Step through all the listeners for the subscriber in turn, @@ -120,7 +120,7 @@ def unregister_subscriber(self, subscriberid): listenerid = self.subscriberDirectory[subscriberid][subscription_name][ listener_key ] - debug.output( + debug_output( 'Unregistering listener on listener_key %s, subscription %s' % (listener_key, subscription_name), listenerid, @@ -131,7 +131,7 @@ def unregister_subscriber(self, subscriberid): ) for topic_name in topic_list: self.topicDirectory[topic_name].unregister_listener(listenerid) - debug.output( + debug_output( 'Listener on listener_key %s, subscription %s unregistered' % (listener_key, subscription_name), listenerid, @@ -140,7 +140,7 @@ def unregister_subscriber(self, subscriberid): listener_list.append(listenerid) """ Remove the subscriber entry in subscriberDirectory. """ del self.subscriberDirectory[subscriberid] - debug.output('Subscriber unregistered', subscriberid) + debug_output('Subscriber unregistered', subscriberid) else: raise EventServiceError('Subscriber not recognized.') return listener_list @@ -157,7 +157,7 @@ def get_subscription(self, subscriberid, subscription_name): if subscription_name not in self.subscriberDirectory[subscriberid]: self.subscriberDirectory[subscriberid][subscription_name] = {} - debug.output('Subscriber subscribed to %s' % subscription_name, subscriberid) + debug_output('Subscriber subscribed to %s' % subscription_name, subscriberid) """ A Subscription object cannot be safely returned without screwing @@ -205,9 +205,9 @@ def process_events(self, subscriberid): raise EventServiceError('Subscriber not recognized.') return event_list - """""" """SubscriberEventService methods end here""" """""" + """SubscriberEventService methods end here""" - """""" """Topic methods start here""" """""" + """Topic methods start here""" """ send_event adds an event to the topic's TopicManager object. @@ -218,24 +218,24 @@ def send_event(self, topic_name, event_name, event_body): event_header = {} event_header[event_name] = event_name the_event = Event(event_header, event_body) - debug.output('Event %s sent to topic %s' % (the_event, topic_name)) + debug_output('Event %s sent to topic %s' % (the_event, topic_name)) self.topicDirectory[topic_name].send_event(the_event) else: raise EventServiceError('Topic not recognized.') - """""" """Topic methods end here""" """""" + """Topic methods end here""" - """""" """EventListener methods start here""" """""" + """EventListener methods start here""" def create_listener(self): self.numListeners += 1 listenerid = self.numListeners - debug.output('Listener created', listenerid) + debug_output('Listener created', listenerid) return listenerid - """""" """EventListener methods end here""" """""" + """EventListener methods end here""" - """""" """Subscription methods start here""" """""" + """Subscription methods start here""" """ register_event_listener adds a listener to its subscriber's subscriberDirectory @@ -255,7 +255,7 @@ def register_event_listener(self, subscriberid, subscription_name, listener_key, listener_key not in self.subscriberDirectory[subscriberid][subscription_name] ): - debug.output( + debug_output( 'Registering listener on listener_key %s, subscription %s' % (listener_key, subscription_name), listenerid, @@ -296,7 +296,7 @@ def unregister_event_listener(self, subscriberid, subscription_name, listener_ke listenerid = self.subscriberDirectory[subscriberid][subscription_name][ listener_key ] - debug.output( + debug_output( 'Unregistering listener on listener_key %s, subscription %s' % (listener_key, subscription_name), listenerid, @@ -308,7 +308,7 @@ def unregister_event_listener(self, subscriberid, subscription_name, listener_ke for topic_name in topic_list: self.topicDirectory[topic_name].unregister_listener(listenerid) del self.subscriberDirectory[subscriberid][subscription_name][listener_key] - debug.output( + debug_output( 'Listener on listener_key %s, subscription %s unregistered' % (listener_key, subscription_name), listenerid, @@ -335,7 +335,7 @@ def remove_subscription(self, subscriberid, subscription_name): listener_list = [] if subscriberid in self.subscriberDirectory: if subscription_name in self.subscriberDirectory[subscriberid]: - debug.output( + debug_output( "\n\n------Subscriber's subscription to %s is being removed" % subscription_name, subscriberid, @@ -344,7 +344,7 @@ def remove_subscription(self, subscriberid, subscription_name): listenerid = self.subscriberDirectory[subscriberid][subscription_name][ listener_key ] - debug.output( + debug_output( 'Unregistering listener on listener_key %s, subscription %s' % (listener_key, subscription_name), listenerid, @@ -355,7 +355,7 @@ def remove_subscription(self, subscriberid, subscription_name): ) for topic_name in topic_list: self.topicDirectory[topic_name].unregister_listener(listenerid) - debug.output( + debug_output( 'Listener on listener_key %s, subscription %s unregistered' % (listener_key, subscription_name), listenerid, @@ -363,7 +363,7 @@ def remove_subscription(self, subscriberid, subscription_name): ) listener_list.append(listenerid) del self.subscriberDirectory[subscriberid][subscription_name] - debug.output( + debug_output( "Subscriber's subscription to %s removed" % subscription_name, subscriberid ) """ @@ -374,9 +374,9 @@ def remove_subscription(self, subscriberid, subscription_name): """ return listener_list - """""" """Subscription methods end here""" """""" + """Subscription methods end here""" - """""" """Methods internal to the event service start here""" """""" + """Methods internal to the event service start here""" """ A listener_key may specify a bunch of topics using wildcarding. @@ -389,4 +389,4 @@ def _map_listener_key_to_topic_list(self, subscription_name, listener_key): topic_list.append(listener_key) return topic_list - """""" """Methods internal to the event service end here""" """""" + """Methods internal to the event service end here""" diff --git a/ipsframework/ips.py b/ipsframework/ips.py index 0aed6638..f8e3df13 100755 --- a/ipsframework/ips.py +++ b/ipsframework/ips.py @@ -84,8 +84,8 @@ from ipsframework.resource_manager import ResourceManager from ipsframework.task_manager import TaskManager -if sys.version_info[0] != 3 or sys.version_info[1] < 9: - print('IPS is only compatible with Python 3.9 or higher', file=sys.stderr) +if sys.version_info < (3, 10): + print('IPS is only compatible with Python 3.10 or higher', file=sys.stderr) sys.exit(1) diff --git a/ipsframework/ips_es_spec.py b/ipsframework/ips_es_spec.py index f04293af..bb0d5692 100644 --- a/ipsframework/ips_es_spec.py +++ b/ipsframework/ips_es_spec.py @@ -17,17 +17,14 @@ class EventManager: def __init__(self, obj_ref): self.obj_ref = obj_ref self.objcache = {} - self.publisher = 'self.publisher' - self.subscriber = 'self.subscriber' + self._publisher_svc = None + self._subscriber_svc = None def publish(self, topic_name, event_name, event_body): - if self.publisher in self.objcache: - pub = self.objcache[self.publisher] - else: - pub = PublisherEventService() - self.objcache[self.publisher] = pub + if self._publisher_svc is None: + self._publisher_svc = PublisherEventService() - topic = pub.get_topic(topic_name) + topic = self._publisher_svc.get_topic(topic_name) topic.send_event(event_name, event_body) def subscribe(self, topic_name, callback): @@ -41,11 +38,8 @@ def subscribe(self, topic_name, callback): # throw an exception? return - if self.subscriber in self.objcache: - sub = self.objcache[self.subscriber] - else: - sub = SubscriberEventService() - self.objcache[self.subscriber] = sub + if self._subscriber_svc is None: + self._subscriber_svc = SubscriberEventService() if topic_name in self.objcache: # TODO: do we notify the client to do an unsubscribe before @@ -53,7 +47,7 @@ def subscribe(self, topic_name, callback): # currently throws an exception in this scenario... scp = self.objcache[topic_name] else: - scp = sub.get_subscription(topic_name) + scp = self._subscriber_svc.get_subscription(topic_name) self.objcache[topic_name] = scp evl = MyEventListener(callback_method) @@ -68,8 +62,8 @@ def unsubscribe(self, topic_name): # throw an exception? def process_events(self): - if self.subscriber in self.objcache: - self.objcache[self.subscriber].process_events() + if self._subscriber_svc is not None: + self._subscriber_svc.process_events() # else: # TODO: do we notify the client to do a subscribe before processing? # throw an exception? diff --git a/ipsframework/ipsutil.py b/ipsframework/ipsutil.py index 90330f16..cb7c0236 100644 --- a/ipsframework/ipsutil.py +++ b/ipsframework/ipsutil.py @@ -11,32 +11,13 @@ import time from collections.abc import Iterable -try: - import Pyro4 -except ImportError: - pass - def which(program, alt_paths: list[str] | None = None): - def is_exe(fpath): - return os.path.exists(fpath) and os.access(fpath, os.X_OK) - - fpath, _ = os.path.split(program) - if fpath: - if is_exe(program): - return program - else: - for path in os.environ['PATH'].split(os.pathsep): - exe_file = os.path.join(path, program) - if is_exe(exe_file): - return exe_file - - # Trust locations in platform file over those in environment path - if alt_paths: - for path in alt_paths: - exe_file = os.path.join(path, program) - if is_exe(exe_file): - return exe_file + path = os.environ.get('PATH', '') + if alt_paths: + # trust locations in platform file over those in environment path + path = os.pathsep.join([*alt_paths, path]) + return shutil.which(program, path=path) def copy_files( @@ -54,12 +35,6 @@ def copy_files( Wild-cards in file name specification are allowed. """ - use_data_server = os.getenv('USE_DATA_SERVER', 'DATA_SERVER_NOT_USED') - if use_data_server != 'DATA_SERVER_NOT_USED': - data_server = Pyro4.Proxy('PYRONAME:DataServer') - data_server.copy_files(src_dir, src_file_list, target_dir, prefix, keep_old) - return - try: file_list = src_file_list.split() except AttributeError: # srcFileList is not a string diff --git a/ipsframework/node_structure.py b/ipsframework/node_structure.py index 8a9107ea..5a56ed51 100644 --- a/ipsframework/node_structure.py +++ b/ipsframework/node_structure.py @@ -5,6 +5,8 @@ Node structures for RM are implemented here for convenience. """ +import sys + # local version @@ -74,11 +76,11 @@ def print_sockets(self, fname=''): else: for sock in self.sockets: - print(' socket:', sock.name) - print(' availablilty:', sock.avail_cores) - print(' task ids:', sock.task_ids) - print(' owners:', sock.owners) - print(' cores:', sock.total_cores) + print(' socket:', sock.name, file=sys.stderr) + print(' availablilty:', sock.avail_cores, file=sys.stderr) + print(' task ids:', sock.task_ids, file=sys.stderr) + print(' owners:', sock.owners, file=sys.stderr) + print(' cores:', sock.total_cores, file=sys.stderr) sock.print_cores() def allocate(self, whole_nodes, whole_sockets, tid, o, procs): @@ -209,12 +211,12 @@ def print_cores(self, fname=''): print(' - owner:', c.owner, file=fname) else: for c in self.cores: - print(' core:', c.name, end=' ') + print(' core:', c.name, end=' ', file=sys.stderr) if c.is_available: - print(' - available') + print(' - available', file=sys.stderr) else: - print(' - task_id:', c.task_id, end=' ') - print(' - owner:', c.owner) + print(' - task_id:', c.task_id, end=' ', file=sys.stderr) + print(' - owner:', c.owner, file=sys.stderr) def allocate(self, whole, tid, o, num_procs): """ @@ -266,7 +268,7 @@ def release(self, tid): self.available.append(c.name) count += 1 if count != k: - print('<<>>') + print('<<>>', file=sys.stderr) # set avail_cores self.avail_cores += k return count @@ -297,7 +299,7 @@ def allocate(self, tid: int, o): self.owner = o return self.name else: - print('trying to allocate core that is not available') + print('trying to allocate core that is not available', file=sys.stderr) raise RuntimeError('trying to allocate core that is not available') def release(self) -> None: @@ -305,7 +307,7 @@ def release(self) -> None: Mark core as available. """ if self.is_available: - print('warning: trying to release core when not in use') + print('warning: trying to release core when not in use', file=sys.stderr) else: self.is_available = True self.task_id = -1 diff --git a/ipsframework/resource_helper.py b/ipsframework/resource_helper.py index 75843fb3..d5311c23 100644 --- a/ipsframework/resource_helper.py +++ b/ipsframework/resource_helper.py @@ -8,6 +8,7 @@ import os import platform import subprocess +import sys from math import ceil import psutil @@ -149,8 +150,8 @@ def get_checkjob_info(): if line.strip() != '': data_lines.append(line.strip()) except Exception as e: - print(e) - raise e + print(e, file=sys.stderr) + raise # return nodes, procs # parse output to get allocated nodes data """ @@ -170,11 +171,11 @@ def get_checkjob_info(): for i in pairs: ndata.append(i.split(':')) # parse allocated nodes data [nid:nprocs]... - for m, _ in ndata: - nodes.append(m) - except Exception as e: - print('problem parsing - small format') - raise e + nodes = [m for m, _ in ndata] + p = ndata[-1][1] + except Exception: + print('problem parsing - small format', file=sys.stderr) + raise elif data_lines[0].find('*') > -1: # large node number format try: @@ -195,15 +196,21 @@ def get_checkjob_info(): # this is a single node id nodes.append(r) ndata = [(n, p) for n in nodes] - except Exception as e: - print('problem parsing - large format') - raise e + except Exception: + print('problem parsing - large format', file=sys.stderr) + raise else: # TODO: make this into a real exception type raise Exception('could not parse resource data') if abs(len(nodes) * int(p) - tot_procs) > 1: - print('len(nodes) = %d p = %d tot_procs = %d' % (len(nodes), int(p), tot_procs)) - print('something wrong with parsing - node count*cores does not match task count') + print( + 'len(nodes) = %d p = %d tot_procs = %d' % (len(nodes), int(p), tot_procs), + file=sys.stderr, + ) + print( + 'something wrong with parsing - node count*cores does not match task count', + file=sys.stderr, + ) raise Exception('something wrong with parsing - node count*cores does not match task count') return nodes, int(p), mixed_nodes, ndata @@ -243,7 +250,7 @@ def get_slurm_info(): cmd = 'scontrol show hostname %s' % nodelist sys_nodes = subprocess.check_output(cmd.split(), encoding='UTF-8').strip().split('\n') nodes.extend([(k, ppn) for k in sys_nodes]) - print('IPS SLURM_NODES = ', nodes) + print('IPS SLURM_NODES = ', nodes, file=sys.stderr) except Exception: raise @@ -331,8 +338,8 @@ def get_resource_list(services, host, partial_nodes=False): node_detect_str = services.get_platform_parameter('NODE_DETECTION', silent=True) if node_detect_str == 'checkjob': num_nodes, ppn, mixed_nodes, list_of_nodes = get_checkjob_info() - print('=======================================================') - print(num_nodes, ppn, mixed_nodes, list_of_nodes) + print('=======================================================', file=sys.stderr) + print(num_nodes, ppn, mixed_nodes, list_of_nodes, file=sys.stderr) accurate_nodes = False elif node_detect_str == 'qstat': num_nodes, ppn, mixed_nodes, list_of_nodes = get_qstat_jobinfo() @@ -358,7 +365,8 @@ def get_resource_list(services, host, partial_nodes=False): else: print( "WARNING: no node detection strategy specified in platform config file ('NODE_DETECTION'). " - 'Valid options are: checkjob, qstat, pbs_env, slurm_env, manual. Trying all detection schemes.' + 'Valid options are: checkjob, qstat, pbs_env, slurm_env, manual. Trying all detection schemes.', + file=sys.stderr, ) try: num_nodes, ppn, mixed_nodes, list_of_nodes = get_checkjob_info() @@ -386,7 +394,7 @@ def get_resource_list(services, host, partial_nodes=False): num_nodes, ppn, mixed_nodes, list_of_nodes = manual_detection(services) accurate_nodes = False except Exception: - print('*** NO DETECTION MECHANISM WORKS ***') + print('*** NO DETECTION MECHANISM WORKS ***', file=sys.stderr) raise # detect topology cpn = int(services.get_platform_parameter('CORES_PER_NODE')) diff --git a/ipsframework/resource_manager.py b/ipsframework/resource_manager.py index dddddeb4..e7f72c4e 100644 --- a/ipsframework/resource_manager.py +++ b/ipsframework/resource_manager.py @@ -3,6 +3,7 @@ # ------------------------------------------------------------------------------- # local version import os +import sys import time from collections import namedtuple from math import ceil @@ -151,7 +152,7 @@ def initialize(self, data_mngr, task_mngr, config_mngr, cmd_nodes=0, cmd_ppn=0): self.accurate_nodes = False self.fwk.warning('RM: User set accurate_nodes to False') except Exception: - print("can't get resource info") + print("can't get resource info", file=sys.stderr) raise # ------------------------------- @@ -280,11 +281,11 @@ def print_rm_state(self) -> None: """ Print the node tree to ``stdout``. """ - print('*** RM.nodeTable ***') + print('*** RM.nodeTable ***', file=sys.stderr) for n, i in self.nodes.items(): - print(n) + print(n, file=sys.stderr) i.print_sockets() - print('=====================') + print('=====================', file=sys.stderr) def add_nodes(self, list_of_nodes: list[tuple[str, int]]) -> int: """ @@ -482,20 +483,20 @@ def get_allocation( self.avail_cores -= cores_allocated self.active_tasks.update({task_id: (comp_id, nproc, cores_allocated)}) except Exception: - print('Available Nodes:') + print('Available Nodes:', file=sys.stderr) for nm in self.avail_nodes: n = self.nodes[nm] - print(n.name, n.avail_cores) + print(n.name, n.avail_cores, file=sys.stderr) n.print_sockets() - print('\nAllocated Nodes:') + print('\nAllocated Nodes:', file=sys.stderr) for nm in self.alloc_nodes: n = self.nodes[nm] - print(n.name, n.avail_cores) + print(n.name, n.avail_cores, file=sys.stderr) n.print_sockets() - print('\n ***** Neither List!') + print('\n ***** Neither List!', file=sys.stderr) for nm, n in self.nodes.items(): if nm not in self.avail_nodes and nm not in self.alloc_nodes: - print(nm, n.avail_cores) + print(nm, n.avail_cores, file=sys.stderr) n.print_sockets() raise diff --git a/ipsframework/services.py b/ipsframework/services.py index 822afe4c..6df99004 100644 --- a/ipsframework/services.py +++ b/ipsframework/services.py @@ -43,7 +43,7 @@ from ipsframework.task_manager import TaskInit pretty.install() -console = Console() +console = Console(stderr=True) rich.traceback.install(show_locals=True) if TYPE_CHECKING: @@ -96,9 +96,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg log.info( f'Launching task {task_name} with id {task_key!s} and worker {worker.name!s} in {working_dir}' ) - print( - f'Launching task {task_name} with id {task_key!s} and worker {worker.name!s} in {working_dir}' - ) start_time = time.time() working_dir_path = Path(working_dir) @@ -115,8 +112,7 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg try: log_filename = kwargs['logfile'] except KeyError: - log.info('No logfile specified, using stdout for task output') - print('No logfile specified, using stdout for task output') + log.info('No logfile specified, using stderr for task output') else: log_path = Path(log_filename) if not log_path.is_absolute(): @@ -124,7 +120,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg subprocess_stdout = open(log_path, 'a') close_stdout = True # Welp, gotta close it now log.info(f'Task output log file: {log_path}') - print(f'Task output log file: {log_path}') # Repeat the same for stderr subprocess_stderr = subprocess.STDOUT @@ -133,7 +128,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg err_filename = kwargs['errfile'] except KeyError: log.info('No errfile specified, using STDOUT for task errors') - print('No errfile specified, using STDOUT for task errors') else: err_path = Path(err_filename) if not err_path.is_absolute(): @@ -142,18 +136,15 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg strict=False ): log.info(f'Task error log file matches output log file: {log_path}') - print(f'Task error log file matches output log file: {log_path}') else: try: subprocess_stderr = open(err_path, 'a') except OSError: log.info(f'Could not open errfile {err_path}, using STDOUT for task errors') - print(f'Could not open errfile {err_path}, using STDOUT for task errors') subprocess_stderr = subprocess.STDOUT else: close_stderr = True log.info(f'Task error log file: {err_path}') - print(f'Task error log file: {err_path}') task_env = kwargs.get('task_env', {}) new_env = os.environ.copy() @@ -168,43 +159,27 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg dvm_uri_file = Path(worker.dvm_uri_file) if not dvm_uri_file.exists(): log.error(f'DVM URI file {dvm_uri_file} does not exist') - print(f'DVM URI file {dvm_uri_file} does not exist') - # print(f'DVM URI file {dvm_uri_file} does not exist', flush=True) else: log.debug(f'Using DVM URI file: {dvm_uri_file}') - print(f'Using DVM URI file: {dvm_uri_file}') - # print(f'Using DVM URI file: {dvm_uri_file}', flush=True) # PMIX_SERVER_URI41 is used by prun to figure out how to talk to the DVM # It can be defined in `task_env` or in `os.environ`, so we look in - # both locations to just echo its presence. The flushes are necessary - # in some HPC environments to ensure the output appears in the logs. + # both locations to just echo its presence. if task_env is not None and task_env != {}: if 'PMIX_SERVER_URI41' in task_env: log.debug( f'DVM environment variable PMIX_SERVER_URI41 set in task_env to {task_env["PMIX_SERVER_URI41"]}' ) - print( - f'DVM environment variable PMIX_SERVER_URI41 set in task_env to {task_env["PMIX_SERVER_URI41"]}' - ) - # print(f'DVM environment variable PMIX_SERVER_URI41 set in task_' - # f'env to {task_env["PMIX_SERVER_URI41"]}', flush=True) if 'PMIX_SERVER_URI41' in os.environ: log.debug( f'DVM environment variable PMIX_SERVER_URI41 set in os.environ to {os.environ["PMIX_SERVER_URI41"]}' ) - print( - f'DVM environment variable PMIX_SERVER_URI41 set in os.environ to {os.environ["PMIX_SERVER_URI41"]}' - ) - # print(f'DVM environment variable PMIX_SERVER_URI41 set in os.environ ' - # f'to {os.environ["PMIX_SERVER_URI41"]}', flush=True) timeout = float(kwargs.get('timeout', 1.0e9)) cmd = f'{executable} {" ".join(map(str, args))}' log.debug(f'Launching task {task_name} with command: {cmd}') - print(f'Launching task {task_name} with command: {cmd}') worker.log_event( 'ips', @@ -241,7 +216,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg }, ) log.error(f'Failed to launch task {task_name} with command {cmd}: {e}') - print(f'Failed to launch task {task_name} with command {cmd}: {e}') raise try: @@ -273,7 +247,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg ) process.kill() log.error(f'Task {task_name} with command {cmd} timed out after {timeout}s') - print(f'Task {task_name} with command {cmd} timed out after {timeout}s') ret_val = -1 except Exception as e: worker.log_event( @@ -286,12 +259,11 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg }, ) log.error(f'Task {task_name} with command {cmd} failed with {e!s}') - print(f'Task {task_name} with command {cmd} failed with {e!s}') finally: if 'logfile' not in kwargs: print(process.stdout.read() if process and process.stdout else '') if 'errfile' not in kwargs: - print(process.stderr.read() if process and process.stderr else '') + print(process.stderr.read() if process and process.stderr else '', file=sys.stderr) if close_stdout: subprocess_stdout.close() @@ -345,7 +317,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg }, ) log.error(f'Task {task_name} with callable {executable!s} failed with {e!s}') - print(f'Task {task_name} with callable {executable!s} failed with {e!s}') finally: os.chdir(str(original_dir)) else: @@ -354,7 +325,6 @@ def launch(executable: Any, task_name: str, working_dir: str | os.PathLike, *arg ) log.info(f'Task {task_name} finished with return value: {ret_val}') - print(f'Task {task_name} finished with return value: {ret_val}') return task_name, ret_val @@ -1831,7 +1801,7 @@ def stage_input_files(self, input_file_list: str | Iterable[str]) -> None: ok=False, ) self.exception('Error in stage_input_files') - raise e + raise for _, old_conf, _, _ in self.sub_flows.values(): ports = old_conf['PORTS']['NAMES'].split() comps = [old_conf['PORTS'][p]['IMPLEMENTATION'] for p in ports] @@ -1849,7 +1819,7 @@ def stage_input_files(self, input_file_list: str | Iterable[str]) -> None: ok=False, ) self.exception('Error in stage_input_files') - raise e + raise elapsed_time = time.time() - start_time self._send_monitor_event( event_type='IPS_STAGE_INPUTS', @@ -3300,23 +3270,21 @@ def setup(self, worker: Worker): command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT ) except Exception as e: - print(f'Exception during setting up DVM: {e}') + print(f'Exception during setting up DVM: {e}', file=sys.stderr) console.print(Traceback.from_exception(type(e), e, e.__traceback__)) # If there was an exception, dump any stdout/stderr we have if hasattr(self.worker.dvm_proc, 'stdout'): - print(self.worker.dvm_proc.stdout, file=sys.stdout, flush=True) + print(self.worker.dvm_proc.stdout, file=sys.stderr, flush=True) print(self.worker.dvm_proc.stderr, file=sys.stderr, flush=True) # TODO What if there was a subprocess exception? ready = self.worker.dvm_proc.stdout.readline() self.logger.info(f'Ready Message : {ready}') - print(f'Ready Message : {ready}', flush=True) with open(self.worker.dvm_uri_file, 'r') as f: self.worker.dvm_uri = f.readline() - print(f'Read DVM URI: {self.worker.dvm_uri}', flush=True) self.logger.debug(f'Read DVM URI: {self.worker.dvm_uri}') os.environ['PMIX_SERVER_URI41'] = self.worker.dvm_uri @@ -3381,6 +3349,7 @@ def __init__(self, name: str, services: ServicesProxy): self.dask_sched_popen = None self.dask_workers_tid = None self.futures = None + self.dask_results = None self.dask_scheduler_file = None self.dask_client = None self.worker_event_logfile = None @@ -3493,6 +3462,9 @@ def _process_dask_event(self, event): # this argument. del message['worker'] + # return_value isn't part of _send_monitor_event()'s API either. + message.pop('return_value', None) + self.services._send_monitor_event(**message) def submit_dask_tasks( @@ -3693,9 +3665,6 @@ def _make_worker_args( self.services.debug( f'Using {dask_ppw} processes per Dask worker via dask_ppw argument' ) - print( - f'Using {dask_ppw} processes per Dask worker via dask_ppw argument', flush=True - ) nthreads = cores_per_node // dask_ppw else: nthreads = cores_per_node @@ -3707,12 +3676,9 @@ def _make_worker_args( nthreads = 1 if nthreads is None or nthreads == 0 else nthreads self.services.debug(f'Number of threads: {nthreads}') - print(f'(submit_dask_tasks: Number of threads: {nthreads})', flush=True) if dask_ppw is not None: self.services.debug(f'Using {dask_ppw} processes per Dask worker via dask_ppw argument') - # FIXME Redundant print since debug() appears to be ignored. - print(f'Using {dask_ppw} processes per Dask worker via dask_ppw argument', flush=True) else: dask_ppw = int(services.get_config_param('PROCS_PER_NODE')) self.services.debug( @@ -3830,6 +3796,8 @@ def _make_worker_args( self.active_tasks = self.queued_tasks self.queued_tasks = {} + nsubmitted = len(self.futures) + if block: self.services.debug('submit_dask_tasks: blocking tasks to await results') # Await all the futures to finish, thereby blocking until they @@ -3838,8 +3806,10 @@ def _make_worker_args( self.services.debug(f'submit_dask_tasks: have {len(result)} results, block released') # TODO check actual result values for problems - # Set this to empty list so that get_dask_finished_tasks_status + # Stash the results so get_dask_finished_tasks_status() can + # report them later, and set futures to empty list so it # doesn't try to gather() needlessly again. + self.dask_results = result self.futures = [] # Since we're done with Dask, let's shut it down @@ -3847,7 +3817,7 @@ def _make_worker_args( else: self.services.debug('submit_dask_tasks: not blocking tasks') - return len(self.futures) + return nsubmitted def submit_tasks( self, @@ -4057,6 +4027,14 @@ def get_dask_finished_tasks_status(self): """ result = None + if self.dask_results is not None: + # submit_dask_tasks was called with block = True, so the + # results were already gathered (and Dask already shut down) + # before this was called. + result = self.dask_results + self.dask_results = None + return dict(result) + if self.dask_client is None: # FIXME How does this happen and is it ok when it does? self.services.warning('No dask client in call to finished tasks status') @@ -4112,7 +4090,7 @@ def get_finished_tasks_status(self): :return: dict mapping task name to exit status :rtype: dict """ - if self.dask_pool: + if self.dask_pool or self.dask_results is not None: return self.get_dask_finished_tasks_status() if len(self.active_tasks) + len(self.finished_tasks) == 0: diff --git a/ipsframework/task_manager.py b/ipsframework/task_manager.py index a843cd15..bf972d22 100644 --- a/ipsframework/task_manager.py +++ b/ipsframework/task_manager.py @@ -129,11 +129,11 @@ def print_curr_task_table(self): """ ctt = self.curr_task_table for c, i in ctt.items(): - print(c) + print(c, file=sys.stderr) for k, v in i.items(): - print(' ', k, '=', v) - print('------') - print('=====================') + print(' ', k, '=', v, file=sys.stderr) + print('------', file=sys.stderr) + print('=====================', file=sys.stderr) # TM call def init_call(self, init_call_msg, manage_return=True): @@ -435,7 +435,7 @@ def build_launch_cmd( nproc_flag = '-np' ppn_flag = '-npernode' host_select = '-H' - if smp_node or mpi_binary == 'prun': + if mpi_binary == 'prun': # --display MAP-DEVEL is added to show the DVM state when # invoking this prun. We do this so that we can verify the # resources managed by DVM for this task as displayed in @@ -450,6 +450,8 @@ def build_launch_cmd( str(nproc), ] ) + elif smp_node: + cmd = ' '.join([mpicmd, nproc_flag, str(nproc)]) else: cmd = ' '.join([mpicmd, nproc_flag, str(nproc), ppn_flag, str(ppn)]) cmd = f'{cmd} -x PYTHONPATH' # Propagate PYTHONPATH to compute nodes diff --git a/ipsframework/topic_manager.py b/ipsframework/topic_manager.py index 4daf9be9..7db820cb 100644 --- a/ipsframework/topic_manager.py +++ b/ipsframework/topic_manager.py @@ -18,7 +18,7 @@ """ from .cca_es_spec import Event, EventServiceError -from .debug import debug +from .debug import output as debug_output class TopicManager: @@ -45,7 +45,7 @@ def __init__(self, limit_pending_events=10): """ self.limit_pending_events = limit_pending_events - debug.output('TopicManager.__init__') + debug_output('TopicManager.__init__') self.print_events_and_listeners() """ @@ -66,7 +66,7 @@ def send_event(self, the_event): self.eventList.append(the_event) event_list_len = len(self.eventList) self.maxPendingEvents = max(event_list_len, self.maxPendingEvents) - debug.output('TopicManager.send_event') + debug_output('TopicManager.send_event') self.print_events_and_listeners() def register_listener(self, listenerid): @@ -77,7 +77,7 @@ def register_listener(self, listenerid): """ if listenerid not in self.listenerDirectory: self.listenerDirectory[listenerid] = len(self.eventList) - debug.output('TopicManager.register_listener') + debug_output('TopicManager.register_listener') self.print_events_and_listeners() else: raise EventServiceError('Event listener registered earlier.') @@ -111,7 +111,7 @@ def cleanup_events(self, listenerid): def unregister_listener(self, listenerid): self.cleanup_events(listenerid) del self.listenerDirectory[listenerid] - debug.output('TopicManager.unregister_listener') + debug_output('TopicManager.unregister_listener') self.print_events_and_listeners() """ @@ -123,7 +123,7 @@ def get_event_list_for_listener(self, listenerid): for the_event in self.eventList[self.listenerDirectory[listenerid] :]: event_list_for_listener.append(Event(the_event.header, the_event.body)) self.cleanup_events(listenerid) - debug.output('TopicManager.get_event_list_for_listener') + debug_output('TopicManager.get_event_list_for_listener') self.print_events_and_listeners() return event_list_for_listener @@ -137,12 +137,12 @@ def print_events_and_listeners(self): for i, e in enumerate(self.eventList): string += '\n' + str(i) + '---' + str(e) string += '\n\n' + 'List of listeners:' - debug.output(string) + debug_output(string) sorted_keys = sorted(self.listenerDirectory.keys()) for listenerid in sorted_keys: string = 'event = ' + str(self.listenerDirectory[listenerid]) - debug.output(string, listenerid) - debug.output(':::::::::') + debug_output(string, listenerid) + debug_output(':::::::::') """ Gives a profile of events posted to this topic, currently just diff --git a/pyproject.toml b/pyproject.toml index 4578508a..311ead33 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,7 @@ dependencies = [ dev = [ "codespell>=2.4.1", "flask", + "mpi4py", # mpi4py is required for some tests, but is not used in primary ipsframework code "mypy>=1.15.0", "pre-commit>=4.2.0", "pytest>=9.0.3", @@ -128,7 +129,7 @@ extend-select = [ 'N', # pep8-naming #'D', # pydocstyle #'UP', # pyupgrade - #'YTT', # flake8-2020 + 'YTT', # flake8-2020 #'ANN', # flake8-annotations #'ASYNC', # flake8-async #'S', # flake8-bandit @@ -141,7 +142,7 @@ extend-select = [ #'T10', # flake8-debugger #'EM', # flake8-error-message #'FA', # flake8-future-annotations - #'ISC', # flake8-implicit-string-concat + 'ISC', # flake8-implicit-string-concat (ISC001 itself stays off, see ignore list) #'ICN', # flake8-import-conventions #'G', # flake8-logging-format #'INP', # flake8-no-pep420 @@ -161,15 +162,15 @@ extend-select = [ #'PGH', # pygrep-hooks 'PL', # pylint #'TRY', # tryceratops - 'FLY', # flynt - 'RUF', # RUFF additional rules + 'TRY201', # tryceratops: use bare `raise` to re-raise, don't name the exception + 'FLY', # flynt + 'RUF', # RUFF additional rules ] # If you're seeking to disable a rule, first consider whether the rule is overbearing, or if it should only be turned off for your usecase. ignore = [ ### TODO move these to extend-select when ready 'D', # pydocstyle 'UP', # pyupgrade - 'YTT', # flake8-2020 'ANN', # flake8-annotations 'ASYNC', # flake8-async 'S', # flake8-bandit @@ -179,7 +180,6 @@ ignore = [ 'T10', # flake8-debugger 'EM', # flake8-error-message 'FA', # flake8-future-annotations - 'ISC', # flake8-implicit-string-concat 'ICN', # flake8-import-conventions 'G', # flake8-logging-format 'INP', # flake8-no-pep420 diff --git a/tests/components/drivers/basic_concurrent1.py b/tests/components/drivers/basic_concurrent_1.py similarity index 97% rename from tests/components/drivers/basic_concurrent1.py rename to tests/components/drivers/basic_concurrent_1.py index fdf347c6..128f2285 100644 --- a/tests/components/drivers/basic_concurrent1.py +++ b/tests/components/drivers/basic_concurrent_1.py @@ -10,7 +10,7 @@ """ from ipsframework import Component -from ipsframework.ipsExceptions import IncompleteCallError +from ipsframework.ips_exceptions import IncompleteCallError class BasicConcurrent1(Component): diff --git a/tests/components/drivers/driver_dataManager.py b/tests/components/drivers/driver_data_manager.py similarity index 100% rename from tests/components/drivers/driver_dataManager.py rename to tests/components/drivers/driver_data_manager.py diff --git a/tests/dakota/test_dakota.py b/tests/dakota/test_dakota.py index 7e289aef..6132d080 100644 --- a/tests/dakota/test_dakota.py +++ b/tests/dakota/test_dakota.py @@ -58,7 +58,7 @@ def test_dakota(tmpdir): # Check PARENT CHILD relationship # Get parent PORTAL_RUNID json_files = glob.glob( - str(tmpdir.join('DAKOTA_Gaussian_TEST_1').join('simulation_log').join('*.json')) + str(tmpdir.join('DAKOTA_Gaussian_TEST_1').join('simulation_log').join('*.jsonl')) ) assert len(json_files) == 1 @@ -81,7 +81,7 @@ def test_dakota(tmpdir): tmpdir.join('DAKOTA_Gaussian_TEST_1') .join('simulation_*_0000') .join('simulation_log') - .join('*.json') + .join('*.jsonl') ) ) assert len(json_files) == 1 diff --git a/tests/hello-world-nested/test_hello-world-nested.py b/tests/hello-world-nested/test_hello-world-nested.py index 84c9ff0e..2787ff64 100644 --- a/tests/hello-world-nested/test_hello-world-nested.py +++ b/tests/hello-world-nested/test_hello-world-nested.py @@ -115,7 +115,7 @@ def test_hello_world_nested(tmpdir, capfd): # check the simulation log json json_files = glob.glob( - str(tmpdir.join('hello_example_SUPER').join('simulation_log').join('*.json')) + str(tmpdir.join('hello_example_SUPER').join('simulation_log').join('*.jsonl')) ) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: diff --git a/tests/helloworld/test_helloworld.py b/tests/helloworld/test_helloworld.py index 9566a7e5..29bf08c2 100644 --- a/tests/helloworld/test_helloworld.py +++ b/tests/helloworld/test_helloworld.py @@ -51,7 +51,7 @@ def test_helloworld(tmpdir, capfd): assert framework.log_file_name.endswith('test.log') fwk_components = framework.config_manager.get_framework_components() - assert len(fwk_components) == 1 + assert len(fwk_components) == 2 assert 'Hello_world_1_FWK@RunspaceInitComponent@3' in fwk_components component_map = framework.config_manager.get_component_map() @@ -81,8 +81,9 @@ def test_helloworld(tmpdir, capfd): assert captured_out[5] == 'Hello from HelloWorker' assert captured_out[6] == 'HelloDriver: finished worker call' - # check that portal didn't write anything since USE_PORTAL=False - assert not os.path.exists(tmpdir.join('simulation_log')) + # LocalLoggingBridge always writes simulation_log; portal-only www dir + # should not exist since USE_PORTAL=False + assert os.path.exists(tmpdir.join('simulation_log')) assert not os.path.exists(tmpdir.join('www')) @@ -111,7 +112,7 @@ def test_helloworld_launch_task(tmpdir, capfd): assert framework.log_file_name.endswith('test.log') fwk_components = framework.config_manager.get_framework_components() - assert len(fwk_components) == 1 + assert len(fwk_components) == 2 assert 'Hello_world_1_FWK@RunspaceInitComponent@3' in fwk_components component_map = framework.config_manager.get_component_map() @@ -179,7 +180,7 @@ def test_helloworld_task_pool(tmpdir, capfd): assert framework.log_file_name.endswith('test.log') - assert len(framework.config_manager.get_framework_components()) == 1 + assert len(framework.config_manager.get_framework_components()) == 2 component_map = framework.config_manager.get_component_map() @@ -273,7 +274,7 @@ def test_helloworld_task_pool_dask(tmpdir, capfd): assert 'ret_val = 9' in captured_out for duration in ('0.2', '0.4', '0.6'): - for task in ['my_fun', 'myMethod']: + for task in ['my_fun', 'my_method']: assert f'{task}({duration})' in captured_out exit_status = json.loads(captured_out[-3].replace("'", '"')) diff --git a/tests/multirun/test_basic_serial.py b/tests/multirun/test_basic_serial.py index 3c8614d7..2d71f374 100644 --- a/tests/multirun/test_basic_serial.py +++ b/tests/multirun/test_basic_serial.py @@ -50,9 +50,9 @@ def test_basic_serial_1(tmpdir, capfd): assert captured_out[0] == "Created " assert captured_out[1] == "Created " assert captured_out[2] == "Created " - assert captured_out[3] == 'small_worker : init() called' - assert captured_out[5] == 'medium_worker : init() called' - assert captured_out[7] == 'large_worker : init() called' + assert captured_out[3] == 'SmallWorker : init() called' + assert captured_out[5] == 'MediumWorker : init() called' + assert captured_out[7] == 'LargeWorker : init() called' assert captured_out[9] == 'Current time = 3.50' assert captured_out[10] == 'Current time = 3.60' assert captured_out[11] == 'Current time = 3.70' @@ -61,7 +61,7 @@ def test_basic_serial_1(tmpdir, capfd): driver_files = [ os.path.basename(f) for f in glob.glob( - str(tmpdir.join('test_basic_serial_1_0/work/drivers_testing_basic_serial_1_*/*')) + str(tmpdir.join('test_basic_serial_1_0/work/drivers_testing_BasicSerial1_*/*')) ) ] for infile in ['file1', 'ofile1', 'ofile2', 'sfile1', 'sfile2']: @@ -189,7 +189,7 @@ def test_basic_serial_multi(tmpdir, capfd): driver_files = [ os.path.basename(f) for f in glob.glob( - str(tmpdir.join(f'test_basic_serial_{no}_0/work/drivers_testing_basic_serial*_*/*')) + str(tmpdir.join(f'test_basic_serial_{no}_0/work/drivers_testing_BasicSerial*_*/*')) ) ] for infile in ['file1', 'ofile1', 'ofile2', 'sfile1', 'sfile2']: @@ -306,7 +306,7 @@ def test_basic_serial_multi(tmpdir, capfd): # check that the parent_portal_runid is correctly set serial1_json_files = glob.glob( - str(tmpdir.join('test_basic_serial_1_0').join('simulation_log').join('*.json')) + str(tmpdir.join('test_basic_serial_1_0').join('simulation_log').join('*.jsonl')) ) assert len(serial1_json_files) == 1 with open(serial1_json_files[0], 'r') as json_file: @@ -317,7 +317,7 @@ def test_basic_serial_multi(tmpdir, capfd): serial1_portal_runid = serial1_ips_start['portal_runid'] serial2_json_files = glob.glob( - str(tmpdir.join('test_basic_serial_2_0').join('simulation_log').join('*.json')) + str(tmpdir.join('test_basic_serial_2_0').join('simulation_log').join('*.jsonl')) ) assert len(serial2_json_files) == 1 with open(serial2_json_files[0], 'r') as json_file: @@ -363,19 +363,17 @@ def test_basic_concurrent_1(tmpdir, capfd): assert captured_out[5] == 'MediumWorker : init() called' assert captured_out[7] == 'LargeWorker : init() called' assert captured_out[9] == 'Current time = 3.50' - assert captured_out[10] == 'nonblocking wait_call() invoked before call 10 finished' + assert captured_out[10] == 'nonblocking wait_call() invoked before call 12 finished' assert captured_out[11] == 'Current time = 3.60' - assert captured_out[12] == 'nonblocking wait_call() invoked before call 13 finished' + assert captured_out[12] == 'nonblocking wait_call() invoked before call 15 finished' assert captured_out[13] == 'Current time = 3.70' - assert captured_out[14] == 'nonblocking wait_call() invoked before call 16 finished' + assert captured_out[14] == 'nonblocking wait_call() invoked before call 18 finished' # check files copied and created driver_files = [ os.path.basename(f) for f in glob.glob( - str( - tmpdir.join('test_basic_concurrent_1_0/work/drivers_testing_basic_concurrent_1_*/*') - ) + str(tmpdir.join('test_basic_concurrent_1_0/work/drivers_testing_BasicConcurrent1_*/*')) ) ] for infile in ['file1', 'ofile1', 'ofile2', 'sfile1', 'sfile2']: diff --git a/tests/new/test_bad_components.py b/tests/new/test_bad_components.py index e6dec797..431de3fa 100644 --- a/tests/new/test_bad_components.py +++ b/tests/new/test_bad_components.py @@ -102,7 +102,7 @@ def test_exception(tmpdir): assert not worker_call_end_event['ok'] assert ( worker_call_end_event['comment'] - == 'Error: "Runtime error" Target = test@exception_worker@2:step(0)' + == 'Error: "Runtime error" Target = test@ExceptionWorker@2:step(0)' ) sim_end_event = events[10] @@ -153,7 +153,7 @@ def test_bad_task(tmpdir): assert not worker_call_end_event['ok'] assert ( worker_call_end_event['comment'] - == 'Error: "task binary of wrong type, expected str but found int" Target = test@bad_task_worker@2:step(0)' + == 'Error: "task binary of wrong type, expected str but found int" Target = test@BadTaskWorker@2:step(0)' ) sim_end_event = events[10] @@ -254,13 +254,13 @@ def test_assign_protected_attribute(tmpdir): assert ( "AttributeError: can't set attribute\n" in lines or "AttributeError: can't set attribute 'args'\n" in lines - or "AttributeError: property 'args' of 'assign_protected_attribute' object has no setter\n" + or "AttributeError: property 'args' of 'AssignProtectedAttribute' object has no setter\n" in lines ) assert ( "Exception: can't set attribute\n" in lines or "Exception: can't set attribute 'args'\n" in lines - or "Exception: property 'args' of 'assign_protected_attribute' object has no setter\n" + or "Exception: property 'args' of 'AssignProtectedAttribute' object has no setter\n" in lines ) @@ -284,9 +284,9 @@ def test_assign_protected_attribute(tmpdir): assert not worker_call_end_event['ok'] # python 3.10 and 3.11 have different error messages assert worker_call_end_event['comment'] in ( - 'Error: "can\'t set attribute" Target = test@assign_protected_attribute@2:step(0)', - "Error: \"can't set attribute 'args'\" Target = test@assign_protected_attribute@2:step(0)", - "Error: \"property 'args' of 'assign_protected_attribute' object has no setter\" Target = test@assign_protected_attribute@2:step(0)", + 'Error: "can\'t set attribute" Target = test@AssignProtectedAttribute@2:step(0)', + "Error: \"can't set attribute 'args'\" Target = test@AssignProtectedAttribute@2:step(0)", + "Error: \"property 'args' of 'AssignProtectedAttribute' object has no setter\" Target = test@AssignProtectedAttribute@2:step(0)", ) sim_end_event = events[10] @@ -298,7 +298,7 @@ def test_assign_protected_attribute(tmpdir): def read_event_log(tmpdir): sim_event_log_json = next( - f for f in os.listdir(tmpdir.join('simulation_log')) if f.endswith('.json') + f for f in os.listdir(tmpdir.join('simulation_log')) if f.endswith('.jsonl') ) with open(str(tmpdir.join('simulation_log').join(sim_event_log_json)), 'r') as f: lines = f.readlines() diff --git a/tests/new/test_component_logging.py b/tests/new/test_component_logging.py index 3ac092be..88c7be3a 100644 --- a/tests/new/test_component_logging.py +++ b/tests/new/test_component_logging.py @@ -82,7 +82,7 @@ def test_component_logging(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - component_id = 'LOGGING__loggingTester_1' + component_id = 'LOGGING__LoggingTester_1' # for log_level=WARNING only WARNING, ERROR and CRITICAL logs should be included # DEBUG and INFO should be excluded diff --git a/tests/new/test_cori_srun.py b/tests/new/test_cori_srun.py index 87ba2e65..d9e748f7 100644 --- a/tests/new/test_cori_srun.py +++ b/tests/new/test_cori_srun.py @@ -67,7 +67,7 @@ def test_srun_openmp_on_cori(tmpdir): framework.run() # check simulation_log - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: comments = [json.loads(line)['comment'].split(', ', maxsplit=4)[3:] for line in json_file] @@ -277,7 +277,7 @@ def test_srun_openmp_on_cori_pool(tmpdir): framework.run() # check simulation_log - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: comments = [json.loads(line)['comment'].split(', ', maxsplit=5)[3:] for line in json_file] diff --git a/tests/new/test_dask.py b/tests/new/test_dask.py index 38239f66..c6b5979a 100644 --- a/tests/new/test_dask.py +++ b/tests/new/test_dask.py @@ -108,7 +108,7 @@ def test_dask(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format('cmd = /bin/sleep') in lines assert log.format('ret_val = 4') in lines @@ -117,7 +117,7 @@ def test_dask(tmpdir): assert log.format(f'task_{i} 0') in lines # check simulation_log, make sure it includes events from dask tasks - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: lines = json_file.readlines() @@ -126,16 +126,19 @@ def test_dask(tmpdir): eventtypes = [e.get('eventtype') for e in lines] assert eventtypes.count('IPS_LAUNCH_DASK_TASK') == 4 - assert eventtypes.count('IPS_TASK_END') == 5 + assert eventtypes.count('IPS_DASK_TASK_END') == 4 launch_dask_comments = [ e.get('comment') for e in lines if e.get('eventtype') == 'IPS_LAUNCH_DASK_TASK' ] for task in range(4): - assert f'task_name = task_{task}, Target = /bin/sleep 1' in launch_dask_comments + assert ( + f'task_name = task_{task}, Task key = task_{task}, Target = /bin/sleep 1' + in launch_dask_comments + ) task_end_comments = [ - e.get('comment')[:-4] for e in lines if e.get('eventtype') == 'IPS_TASK_END' + e.get('comment')[:-4] for e in lines if e.get('eventtype') == 'IPS_DASK_TASK_END' ] for task in range(4): assert f'task_name = task_{task}, elapsed time = 1' in task_end_comments @@ -169,12 +172,12 @@ def test_dask_shifter_fail(tmpdir): lines = [line[24:] for line in lines] assert ( - 'DASK__dask_worker_2 ERROR Requested to run dask within shifter but shifter not available\n' + 'DASK__DaskWorker_2 ERROR Requested to run dask within shifter but shifter not available\n' in lines ) # check simulation_log, make sure it includes events from dask tasks - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: lines = json_file.readlines() @@ -222,7 +225,7 @@ def test_dask_fake_shifter(tmpdir, monkeypatch): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format('cmd = /bin/sleep') in lines assert log.format('ret_val = 4') in lines @@ -231,7 +234,7 @@ def test_dask_fake_shifter(tmpdir, monkeypatch): assert log.format(f'task_{i} 0') in lines # check simulation_log, make sure it includes events from dask tasks - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: lines = json_file.readlines() @@ -240,28 +243,31 @@ def test_dask_fake_shifter(tmpdir, monkeypatch): eventtypes = [e.get('eventtype') for e in lines] assert eventtypes.count('IPS_LAUNCH_DASK_TASK') == 4 - assert eventtypes.count('IPS_TASK_END') == 5 + assert eventtypes.count('IPS_DASK_TASK_END') == 4 launch_dask_comments = [ e.get('comment') for e in lines if e.get('eventtype') == 'IPS_LAUNCH_DASK_TASK' ] for task in range(4): - assert f'task_name = task_{task}, Target = /bin/sleep 1' in launch_dask_comments + assert ( + f'task_name = task_{task}, Task key = task_{task}, Target = /bin/sleep 1' + in launch_dask_comments + ) task_end_comments = [ - e.get('comment')[:-4] for e in lines if e.get('eventtype') == 'IPS_TASK_END' + e.get('comment')[:-4] for e in lines if e.get('eventtype') == 'IPS_DASK_TASK_END' ] for task in range(4): assert f'task_name = task_{task}, elapsed time = 1' in task_end_comments # check shifter.log file - with open(str(tmpdir.join('/work/DASK__dask_worker_2').join('shifter.log')), 'r') as f: + with open(str(tmpdir.join('/work/DASK__DaskWorker_2').join('shifter.log')), 'r') as f: lines = sorted(f.readlines()) - assert lines[0].startswith('Running dask scheduler --no-dashboard --scheduler-file') + assert ' scheduler --no-dashboard --no-jupyter --no-show --idle-timeout' in lines[0] assert lines[0].endswith('--port 0 in shifter\n') - assert lines[1].startswith('Running dask worker --scheduler-file') - assert lines[1].endswith('--nworkers 1 --nthreads 2 --no-dashboard in shifter\n') + assert ' worker --no-dashboard --no-nanny --scheduler-file' in lines[1] + assert lines[1].endswith('--nworkers 1 --nthreads 2 in shifter\n') def test_dask_timeout(tmpdir): @@ -286,7 +292,7 @@ def test_dask_timeout(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format('cmd = /bin/sleep') in lines assert log.format('ret_val = 4') in lines @@ -295,7 +301,7 @@ def test_dask_timeout(tmpdir): assert log.format(f'task_{i} -1') in lines # check simulation_log, make sure it includes events from dask tasks - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: lines = json_file.readlines() @@ -304,15 +310,20 @@ def test_dask_timeout(tmpdir): eventtypes = [e.get('eventtype') for e in lines] assert eventtypes.count('IPS_LAUNCH_DASK_TASK') == 4 - assert eventtypes.count('IPS_TASK_END') == 5 + assert eventtypes.count('IPS_DASK_TASK_END') == 4 launch_dask_comments = [ e.get('comment') for e in lines if e.get('eventtype') == 'IPS_LAUNCH_DASK_TASK' ] for task in range(4): - assert f'task_name = task_{task}, Target = /bin/sleep 100' in launch_dask_comments + assert ( + f'task_name = task_{task}, Task key = task_{task}, Target = /bin/sleep 100' + in launch_dask_comments + ) - task_end_comments = [e.get('comment') for e in lines if e.get('eventtype') == 'IPS_TASK_END'] + task_end_comments = [ + e.get('comment') for e in lines if e.get('eventtype') == 'IPS_DASK_TASK_END' + ] for task in range(4): assert f'task_name = task_{task}, timed-out after 1.0s' in task_end_comments @@ -341,7 +352,7 @@ def test_dask_nproc(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format('cmd = /bin/sleep') in lines assert log.format('ret_val = 4') in lines @@ -351,7 +362,7 @@ def test_dask_nproc(tmpdir): # check for warning message that dask isn't being used assert ( - 'DASK__dask_worker_2 WARNING Requested use_dask but cannot because multiple processors requested\n' + 'DASK__DaskWorker_2 WARNING Requested use_dask but cannot because multiple processors requested\n' in lines ) @@ -384,7 +395,7 @@ def test_dask_logfile(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format(f'cmd = {exe}') in lines assert log.format('ret_val = 4') in lines @@ -393,7 +404,7 @@ def test_dask_logfile(tmpdir): assert log.format(f'task_{i} 0') in lines # check that the process output log files are created - work_dir = tmpdir.join('work').join('DASK__dask_worker_2') + work_dir = tmpdir.join('work').join('DASK__DaskWorker_2') for i in range(4): log_file = work_dir.join(f'task_{i}.log') assert log_file.exists() @@ -430,7 +441,7 @@ def test_dask_logfile_errfile(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format(f'cmd = {exe}') in lines assert log.format('ret_val = 4') in lines @@ -439,7 +450,7 @@ def test_dask_logfile_errfile(tmpdir): assert log.format(f'task_{i} 0') in lines # check that the process output log files are created - work_dir = tmpdir.join('work').join('DASK__dask_worker_2') + work_dir = tmpdir.join('work').join('DASK__DaskWorker_2') for i in range(4): log_file = work_dir.join(f'task_{i}.log') assert log_file.exists() @@ -489,7 +500,7 @@ def test_dask_shifter_on_cori(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format(f'cmd = {exe}') in lines assert log.format('ret_val = 4') in lines @@ -498,7 +509,7 @@ def test_dask_shifter_on_cori(tmpdir): assert log.format(f'task_{i} 0') in lines # check that the process output log files are created - work_dir = tmpdir.join('work').join('DASK__dask_worker_2') + work_dir = tmpdir.join('work').join('DASK__DaskWorker_2') for i in range(4): log_file = work_dir.join(f'task_{i}.log') assert log_file.exists() @@ -531,7 +542,7 @@ def test_dask_with_1_gpu(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format('ret_val = 4') in lines # task successful and return 0 @@ -539,15 +550,15 @@ def test_dask_with_1_gpu(tmpdir): assert log.format(f'task_{i} 0') in lines # check simulation_log - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: comments = [json.loads(line)['comment'].split(', ', maxsplit=5)[2:] for line in json_file] assert comments[10][0] == 'nproc = 1 ' assert comments[10][1].startswith('Target = ') - assert 'dask-worker --scheduler-file' in comments[10][1] - assert comments[10][1].endswith('s 1 --nthreads 2 --no-dashboard') + assert 'dask worker --no-dashboard --no-nanny --scheduler-file' in comments[10][1] + assert comments[10][1].endswith('--nworkers 1 --nthreads 2') def test_dask_with_2_gpus(tmpdir): @@ -572,7 +583,7 @@ def test_dask_with_2_gpus(tmpdir): # remove timestamp lines = [line[24:] for line in lines] - log = 'DASK__dask_worker_2 INFO {}\n' + log = 'DASK__DaskWorker_2 INFO {}\n' assert log.format('ret_val = 4') in lines # task successful and return 0 @@ -580,12 +591,12 @@ def test_dask_with_2_gpus(tmpdir): assert log.format(f'task_{i} 0') in lines # check simulation_log - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: comments = [json.loads(line)['comment'].split(', ', maxsplit=5)[2:] for line in json_file] assert comments[10][0] == 'nproc = 2 ' assert comments[10][1].startswith('Target = ') - assert 'dask-worker --scheduler-file' in comments[10][1] - assert comments[10][1].endswith('s 1 --nthreads 1 --no-dashboard') + assert 'dask worker --no-dashboard --no-nanny --scheduler-file' in comments[10][1] + assert comments[10][1].endswith('--nworkers 1 --nthreads 1') diff --git a/tests/new/test_ips_framework.py b/tests/new/test_ips_framework.py index c5552ca8..56e3404f 100644 --- a/tests/new/test_ips_framework.py +++ b/tests/new/test_ips_framework.py @@ -10,7 +10,7 @@ def write_basic_config_and_platform_files(tmpdir): test_component = tmpdir.join('test_component.py') driver = """from ipsframework.component import Component -class test_driver(Component): +class TestDriver(Component): def __init__(self, services, config): super().__init__(services, config) """ @@ -84,10 +84,10 @@ def test_framework_simple(tmpdir, capfd): assert 'test' in component_map test = component_map['test'] assert len(test) == 1 - assert test[0].get_class_name() == 'test_driver' - assert test[0].get_instance_name().startswith('test@test_driver') + assert test[0].get_class_name() == 'TestDriver' + assert test[0].get_instance_name().startswith('test@TestDriver') assert test[0].get_seq_num() == 1 - assert test[0].get_serialization().startswith('test@test_driver') + assert test[0].get_serialization().startswith('test@TestDriver') assert test[0].get_sim_name() == 'test' # check all registered service handlers @@ -97,12 +97,12 @@ def test_framework_simple(tmpdir, capfd): 'create_simulation', 'exists_topic', 'finish_task', - 'get_subscription', - 'get_topic', 'get_allocation', 'get_config_parameter', 'get_port', + 'get_subscription', 'get_time_loop', + 'get_topic', 'init_call', 'init_task', 'init_task_pool', @@ -125,7 +125,7 @@ def test_framework_simple(tmpdir, capfd): framework.run() # check simulation_log - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: json_lines = json_file.readlines() @@ -258,7 +258,7 @@ def test_framework_log_output_debug(tmpdir): with open(str(tmpdir.join('framework_log_debug_test.log')), 'r') as f: lines = f.readlines() - assert len(lines) == 32 + assert len(lines) == 30 assert 'Traceback (most recent call last):\n' in lines assert " raise ValueError('wrong value')\n" in lines diff --git a/tests/new/test_perlmutter_srun.py b/tests/new/test_perlmutter_srun.py index 52195bc6..45cddca0 100644 --- a/tests/new/test_perlmutter_srun.py +++ b/tests/new/test_perlmutter_srun.py @@ -75,7 +75,7 @@ def test_srun_gpu_on_perlmutter(tmpdir): framework.run() # check simulation_log - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: comments = [json.loads(line)['comment'].split(', ', maxsplit=4)[3:] for line in json_file] diff --git a/tests/new/test_portal.py b/tests/new/test_portal.py index a02f8b4a..c6cf20b8 100644 --- a/tests/new/test_portal.py +++ b/tests/new/test_portal.py @@ -118,7 +118,7 @@ def api(): with open(str(tmpdir.join('ips.log')), 'r') as f: lines = f.readlines() - ur_ls = [line[57:] for line in lines if 'FWK_COMP_PortalBridge_4 INFO' in line] + ur_ls = [line[57:] for line in lines if 'FWK_COMP_PortalBridge_5 INFO' in line] assert len(ur_ls) > 0 assert ur_ls[0] == 'Run Portal URL = http://localhost:18080/42\n' @@ -128,7 +128,7 @@ def api(): for (code, data) in [ line[74:].strip().split(maxsplit=1) for line in lines - if 'FWK_COMP_PortalBridge_4 DEBUG Portal Response: ' in line + if 'FWK_COMP_PortalBridge_5 DEBUG Portal Response: ' in line ] ] @@ -197,11 +197,11 @@ def test_portal_no_server(tmpdir): lines = f.readlines() # remove timestamp and common start - lines = [line[57:] for line in lines if 'FWK_COMP_PortalBridge_4 ERROR' in line] + lines = [line[57:] for line in lines if 'FWK_COMP_PortalBridge_5 ERROR' in line] assert len(lines) == 4 # should fail 3 time then disable the portal for n in range(3): - assert lines[n].startswith('Portal Error: 999 HTTPConnectionPool') + assert lines[n].startswith('Portal Error: 999 Max retry error: HTTPConnectionPool') assert lines[-1] == 'Disabling portal because: Too many consecutive failed connections\n' diff --git a/tests/new/test_run_ensemble.py b/tests/new/test_run_ensemble.py index e3b40f24..a09bca96 100644 --- a/tests/new/test_run_ensemble.py +++ b/tests/new/test_run_ensemble.py @@ -1,6 +1,5 @@ import logging import os -from importlib import import_module from ipsframework import ServicesProxy, TaskPool from ipsframework import services as services_module @@ -186,12 +185,10 @@ def record_launch(executable, task_name, working_dir, *args, **kwargs): def test_launch_writes_stderr_to_logfile_when_errfile_is_omitted(tmpdir, monkeypatch): script = write_stdout_stderr_script(tmpdir) - dask_distributed = import_module('dask.distributed') - def get_worker(): return DummyDaskWorker() - monkeypatch.setattr(dask_distributed, 'get_worker', get_worker) + monkeypatch.setattr(services_module, 'get_worker', get_worker) assert services_module.launch( str(script), @@ -209,12 +206,10 @@ def get_worker(): def test_launch_writes_stderr_to_logfile_when_errfile_matches_logfile(tmpdir, monkeypatch): script = write_stdout_stderr_script(tmpdir) - dask_distributed = import_module('dask.distributed') - def get_worker(): return DummyDaskWorker() - monkeypatch.setattr(dask_distributed, 'get_worker', get_worker) + monkeypatch.setattr(services_module, 'get_worker', get_worker) assert services_module.launch( str(script), diff --git a/tests/new/test_service_checkpoint_component.py b/tests/new/test_service_checkpoint_component.py index a78827b4..1eaf1794 100644 --- a/tests/new/test_service_checkpoint_component.py +++ b/tests/new/test_service_checkpoint_component.py @@ -52,7 +52,7 @@ def test_checkpoint_components_force(): services_proxy = ServicesProxy(None, None, None, {}, None) services_proxy._dispatch_checkpoint = MagicMock(name='dispatch_checkpoint') - services_proxy.checkpoint_components([], 0, Force=True) + services_proxy.checkpoint_components([], 0, force=True) services_proxy._dispatch_checkpoint.assert_called_once_with(0, [], False) diff --git a/tests/new/test_timeloop_checkpoint.py b/tests/new/test_timeloop_checkpoint.py index ab2faa81..89cb0b21 100644 --- a/tests/new/test_timeloop_checkpoint.py +++ b/tests/new/test_timeloop_checkpoint.py @@ -35,7 +35,7 @@ def write_basic_config_and_platform_files(tmpdir, restart=False): LOG_FILE = {tmpdir!s}/{sim_log} LOG_LEVEL = INFO SIM_ROOT = {tmpdir!s} -simulation_mode = {simulation_mode} +SIMULATION_MODE = {simulation_mode} CURRENT_STATE = ${{SIM_NAME}}_ps.dat STATE_FILES = $CURRENT_STATE STATE_WORK_DIR = $SIM_ROOT/work/state @@ -83,9 +83,9 @@ def write_basic_config_and_platform_files(tmpdir, restart=False): MODULE = components.workers.timeloop_comp [TIME_LOOP] MODE = REGULAR - start = {start} - finish = {finish} - nstep = {nstep} + START = {start} + FINISH = {finish} + NSTEP = {nstep} [CHECKPOINT] MODE = ALL NUM_CHECKPOINT = 2 @@ -120,12 +120,12 @@ def test_timeloop_checkpoint_restart(tmpdir): lines = [line[24:] for line in lines] for time in ['100.0', '112.5', '125.0', '137.5', '150.0']: - assert f'TIMELOOP_COMP__timeloop_comp_2 INFO step({time})\n' in lines - assert f'TIMELOOP_COMP2__timeloop_comp_3 INFO step({time})\n' in lines + assert f'TIMELOOP_COMP__TimeloopComp_2 INFO step({time})\n' in lines + assert f'TIMELOOP_COMP2__TimeloopComp_3 INFO step({time})\n' in lines for comp in [ - 'TIMELOOP__timeloop_driver_1', - 'TIMELOOP_COMP__timeloop_comp_2', - 'TIMELOOP_COMP2__timeloop_comp_3', + 'TIMELOOP__TimeloopDriver_1', + 'TIMELOOP_COMP__TimeloopComp_2', + 'TIMELOOP_COMP2__TimeloopComp_3', ]: assert f'{comp} INFO checkpoint({time})\n' in lines @@ -140,20 +140,20 @@ def test_timeloop_checkpoint_restart(tmpdir): # restart files restart_dir = tmpdir.join('restart') assert len(restart_dir.listdir()) == 2 - assert restart_dir.join('137.500').join('TIMELOOP_COMP__timeloop_comp').exists() - assert restart_dir.join('150.000').join('TIMELOOP_COMP__timeloop_comp').exists() - assert restart_dir.join('137.500').join('TIMELOOP_COMP2__timeloop_comp').exists() - assert restart_dir.join('150.000').join('TIMELOOP_COMP2__timeloop_comp').exists() + assert restart_dir.join('137.500').join('TIMELOOP_COMP__TimeloopComp').exists() + assert restart_dir.join('150.000').join('TIMELOOP_COMP__TimeloopComp').exists() + assert restart_dir.join('137.500').join('TIMELOOP_COMP2__TimeloopComp').exists() + assert restart_dir.join('150.000').join('TIMELOOP_COMP2__TimeloopComp').exists() # 137.500 - restart_files = restart_dir.join('137.500').join('TIMELOOP_COMP__timeloop_comp') + restart_files = restart_dir.join('137.500').join('TIMELOOP_COMP__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w1_1.dat').exists() assert len(restart_files.join('w1_1.dat').readlines()) == 5 assert restart_files.join('test_ps.dat').exists() assert len(restart_files.join('test_ps.dat').readlines()) == 14 - restart_files = restart_dir.join('137.500').join('TIMELOOP_COMP2__timeloop_comp') + restart_files = restart_dir.join('137.500').join('TIMELOOP_COMP2__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w2_1.dat').exists() assert len(restart_files.join('w2_1.dat').readlines()) == 5 @@ -161,14 +161,14 @@ def test_timeloop_checkpoint_restart(tmpdir): assert len(restart_files.join('test_ps.dat').readlines()) == 15 # 150.000 - restart_files = restart_dir.join('150.000').join('TIMELOOP_COMP__timeloop_comp') + restart_files = restart_dir.join('150.000').join('TIMELOOP_COMP__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w1_1.dat').exists() assert len(restart_files.join('w1_1.dat').readlines()) == 6 assert restart_files.join('test_ps.dat').exists() assert len(restart_files.join('test_ps.dat').readlines()) == 17 - restart_files = restart_dir.join('150.000').join('TIMELOOP_COMP2__timeloop_comp') + restart_files = restart_dir.join('150.000').join('TIMELOOP_COMP2__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w2_1.dat').exists() assert len(restart_files.join('w2_1.dat').readlines()) == 6 @@ -181,10 +181,10 @@ def test_timeloop_checkpoint_restart(tmpdir): assert len(results_dir.listdir()) == 8 for time in ['100.0', '112.5', '125.0', '137.5', '150.0']: - assert results_dir.join('TIMELOOP_COMP__timeloop_comp_2').join(f'w1_1_{time}.dat').exists() - assert results_dir.join('TIMELOOP_COMP__timeloop_comp_2').join(f'w1_2_{time}.dat').exists() - assert results_dir.join('TIMELOOP_COMP2__timeloop_comp_3').join(f'w2_1_{time}.dat').exists() - assert results_dir.join('TIMELOOP_COMP2__timeloop_comp_3').join(f'w2_2_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP__TimeloopComp_2').join(f'w1_1_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP__TimeloopComp_2').join(f'w1_2_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP2__TimeloopComp_3').join(f'w2_1_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP2__TimeloopComp_3').join(f'w2_2_{time}.dat').exists() # Now do simulation_mode=RESTART @@ -210,12 +210,12 @@ def test_timeloop_checkpoint_restart(tmpdir): lines = [line[24:] for line in lines] for time in ['162.5', '175.0', '187.5', '200.0']: - assert f'TIMELOOP_COMP__timeloop_comp_8 INFO step({time})\n' in lines - assert f'TIMELOOP_COMP2__timeloop_comp_9 INFO step({time})\n' in lines + assert f'TIMELOOP_COMP__TimeloopComp_8 INFO step({time})\n' in lines + assert f'TIMELOOP_COMP2__TimeloopComp_9 INFO step({time})\n' in lines for comp in [ - 'TIMELOOP__timeloop_driver_7', - 'TIMELOOP_COMP__timeloop_comp_8', - 'TIMELOOP_COMP2__timeloop_comp_9', + 'TIMELOOP__TimeloopDriver_7', + 'TIMELOOP_COMP__TimeloopComp_8', + 'TIMELOOP_COMP2__TimeloopComp_9', ]: assert f'{comp} INFO checkpoint({time})\n' in lines @@ -230,20 +230,20 @@ def test_timeloop_checkpoint_restart(tmpdir): # restart files restart_dir = tmpdir.join('restart') assert len(restart_dir.listdir()) == 2 - assert restart_dir.join('187.500').join('TIMELOOP_COMP__timeloop_comp').exists() - assert restart_dir.join('200.000').join('TIMELOOP_COMP__timeloop_comp').exists() - assert restart_dir.join('187.500').join('TIMELOOP_COMP2__timeloop_comp').exists() - assert restart_dir.join('200.000').join('TIMELOOP_COMP2__timeloop_comp').exists() + assert restart_dir.join('187.500').join('TIMELOOP_COMP__TimeloopComp').exists() + assert restart_dir.join('200.000').join('TIMELOOP_COMP__TimeloopComp').exists() + assert restart_dir.join('187.500').join('TIMELOOP_COMP2__TimeloopComp').exists() + assert restart_dir.join('200.000').join('TIMELOOP_COMP2__TimeloopComp').exists() # 137.500 - restart_files = restart_dir.join('187.500').join('TIMELOOP_COMP__timeloop_comp') + restart_files = restart_dir.join('187.500').join('TIMELOOP_COMP__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w1_1.dat').exists() assert len(restart_files.join('w1_1.dat').readlines()) == 10 assert restart_files.join('test_ps.dat').exists() assert len(restart_files.join('test_ps.dat').readlines()) == 29 - restart_files = restart_dir.join('187.500').join('TIMELOOP_COMP2__timeloop_comp') + restart_files = restart_dir.join('187.500').join('TIMELOOP_COMP2__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w2_1.dat').exists() assert len(restart_files.join('w2_1.dat').readlines()) == 10 @@ -251,14 +251,14 @@ def test_timeloop_checkpoint_restart(tmpdir): assert len(restart_files.join('test_ps.dat').readlines()) == 30 # 200.000 - restart_files = restart_dir.join('200.000').join('TIMELOOP_COMP__timeloop_comp') + restart_files = restart_dir.join('200.000').join('TIMELOOP_COMP__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w1_1.dat').exists() assert len(restart_files.join('w1_1.dat').readlines()) == 11 assert restart_files.join('test_ps.dat').exists() assert len(restart_files.join('test_ps.dat').readlines()) == 32 - restart_files = restart_dir.join('200.000').join('TIMELOOP_COMP2__timeloop_comp') + restart_files = restart_dir.join('200.000').join('TIMELOOP_COMP2__TimeloopComp') assert len(restart_files.listdir()) == 2 assert restart_files.join('w2_1.dat').exists() assert len(restart_files.join('w2_1.dat').readlines()) == 11 @@ -267,25 +267,22 @@ def test_timeloop_checkpoint_restart(tmpdir): # work files, w[1,2]_1.dat should include previous data where w[1,2]_2.dat shouldn't work_files = tmpdir.join('work') - assert work_files.join('TIMELOOP_COMP__timeloop_comp_8').join('w1_1.dat').exists() - assert work_files.join('TIMELOOP_COMP__timeloop_comp_8').join('w1_2.dat').exists() - assert work_files.join('TIMELOOP_COMP__timeloop_comp_8').join('test_ps.dat').exists() - assert len(work_files.join('TIMELOOP_COMP__timeloop_comp_8').join('w1_1.dat').readlines()) == 11 - assert len(work_files.join('TIMELOOP_COMP__timeloop_comp_8').join('w1_2.dat').readlines()) == 5 + assert work_files.join('TIMELOOP_COMP__TimeloopComp_8').join('w1_1.dat').exists() + assert work_files.join('TIMELOOP_COMP__TimeloopComp_8').join('w1_2.dat').exists() + assert work_files.join('TIMELOOP_COMP__TimeloopComp_8').join('test_ps.dat').exists() + assert len(work_files.join('TIMELOOP_COMP__TimeloopComp_8').join('w1_1.dat').readlines()) == 11 + assert len(work_files.join('TIMELOOP_COMP__TimeloopComp_8').join('w1_2.dat').readlines()) == 5 assert ( - len(work_files.join('TIMELOOP_COMP__timeloop_comp_8').join('test_ps.dat').readlines()) == 32 + len(work_files.join('TIMELOOP_COMP__TimeloopComp_8').join('test_ps.dat').readlines()) == 32 ) - assert work_files.join('TIMELOOP_COMP2__timeloop_comp_9').join('w2_1.dat').exists() - assert work_files.join('TIMELOOP_COMP2__timeloop_comp_9').join('w2_2.dat').exists() - assert work_files.join('TIMELOOP_COMP2__timeloop_comp_9').join('test_ps.dat').exists() + assert work_files.join('TIMELOOP_COMP2__TimeloopComp_9').join('w2_1.dat').exists() + assert work_files.join('TIMELOOP_COMP2__TimeloopComp_9').join('w2_2.dat').exists() + assert work_files.join('TIMELOOP_COMP2__TimeloopComp_9').join('test_ps.dat').exists() + assert len(work_files.join('TIMELOOP_COMP2__TimeloopComp_9').join('w2_1.dat').readlines()) == 11 + assert len(work_files.join('TIMELOOP_COMP2__TimeloopComp_9').join('w2_2.dat').readlines()) == 5 assert ( - len(work_files.join('TIMELOOP_COMP2__timeloop_comp_9').join('w2_1.dat').readlines()) == 11 - ) - assert len(work_files.join('TIMELOOP_COMP2__timeloop_comp_9').join('w2_2.dat').readlines()) == 5 - assert ( - len(work_files.join('TIMELOOP_COMP2__timeloop_comp_9').join('test_ps.dat').readlines()) - == 33 + len(work_files.join('TIMELOOP_COMP2__TimeloopComp_9').join('test_ps.dat').readlines()) == 33 ) # check output from services.stage_output_files @@ -294,33 +291,33 @@ def test_timeloop_checkpoint_restart(tmpdir): assert len(results_dir.listdir()) == 14 for time in ['162.5', '175.0', '187.5', '200.0']: - assert results_dir.join('TIMELOOP_COMP__timeloop_comp_8').join(f'w1_1_{time}.dat').exists() - assert results_dir.join('TIMELOOP_COMP__timeloop_comp_8').join(f'w1_2_{time}.dat').exists() - assert results_dir.join('TIMELOOP_COMP2__timeloop_comp_9').join(f'w2_1_{time}.dat').exists() - assert results_dir.join('TIMELOOP_COMP2__timeloop_comp_9').join(f'w2_2_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP__TimeloopComp_8').join(f'w1_1_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP__TimeloopComp_8').join(f'w1_2_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP2__TimeloopComp_9').join(f'w2_1_{time}.dat').exists() + assert results_dir.join('TIMELOOP_COMP2__TimeloopComp_9').join(f'w2_2_{time}.dat').exists() def test_time_loop(): - sim_conf = {'TIME_LOOP': {'MODE': 'REGULAR', 'start': '0', 'finish': '10', 'nstep': '10'}} + sim_conf = {'TIME_LOOP': {'MODE': 'REGULAR', 'START': '0', 'FINISH': '10', 'NSTEP': '10'}} services_proxy = ServicesProxy(None, None, None, sim_conf, None) tl = services_proxy.get_time_loop() assert tl == list(range(11)) sim_conf = { - 'TIME_LOOP': {'MODE': 'REGULAR', 'start': '0 + 20 / 2', 'finish': '13 - 1', 'nstep': '2'} + 'TIME_LOOP': {'MODE': 'REGULAR', 'START': '0 + 20 / 2', 'FINISH': '13 - 1', 'NSTEP': '2'} } services_proxy = ServicesProxy(None, None, None, sim_conf, None) tl = services_proxy.get_time_loop() assert tl == [10, 11, 12] sim_conf = { - 'TIME_LOOP': {'MODE': 'REGULAR', 'start': '10 * 2', 'finish': '10 ** 2', 'nstep': '2'} + 'TIME_LOOP': {'MODE': 'REGULAR', 'START': '10 * 2', 'FINISH': '10 ** 2', 'NSTEP': '2'} } services_proxy = ServicesProxy(None, None, None, sim_conf, None) tl = services_proxy.get_time_loop() assert tl == [20, 60, 100] - sim_conf = {'TIME_LOOP': {'MODE': 'REGULAR', 'start': '1e2', 'finish': '5e1', 'nstep': '2'}} + sim_conf = {'TIME_LOOP': {'MODE': 'REGULAR', 'START': '1e2', 'FINISH': '5e1', 'NSTEP': '2'}} services_proxy = ServicesProxy(None, None, None, sim_conf, None) tl = services_proxy.get_time_loop() assert tl == [100, 75, 50] @@ -330,9 +327,9 @@ def test_time_loop(): tl = services_proxy.get_time_loop() assert tl == [7, 13, -42, 1000] - sim_conf = {'TIME_LOOP': {'MODE': 'REGULAR', 'start': '1p2', 'finish': '10', 'nstep': '2'}} + sim_conf = {'TIME_LOOP': {'MODE': 'REGULAR', 'START': '1p2', 'FINISH': '10', 'NSTEP': '2'}} services_proxy = ServicesProxy(None, None, None, sim_conf, None) services_proxy.error = MagicMock(name='error') with pytest.raises(ValueError) as excinfo: services_proxy.get_time_loop() - assert str(excinfo.value) == 'Invalid TIME_LOOP value of start = 1p2' + assert str(excinfo.value) == 'Invalid TIME_LOOP value of START = 1p2' diff --git a/tests/new/test_trace.py b/tests/new/test_trace.py index a707c5dd..919e3c86 100644 --- a/tests/new/test_trace.py +++ b/tests/new/test_trace.py @@ -39,7 +39,7 @@ def write_basic_config_and_platform_files( [DRIVER] CLASS = DRIVER SUB_CLASS = - NAME = driver + NAME = Driver BIN_PATH = NPROC = 1 INPUT_FILES = @@ -49,7 +49,7 @@ def write_basic_config_and_platform_files( [WORKER] CLASS = WORKER SUB_CLASS = - NAME = simple_sleep + NAME = SimpleSleep NPROC = 1 BIN_PATH = INPUT_FILES = @@ -80,7 +80,7 @@ def test_trace_info(tmpdir): framework.run() # check simulation_log, make sure it includes events from dask tasks - json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.json'))) + json_files = glob.glob(str(tmpdir.join('simulation_log').join('*.jsonl'))) assert len(json_files) == 1 with open(json_files[0], 'r') as json_file: lines = json_file.readlines() @@ -95,13 +95,13 @@ def test_trace_info(tmpdir): call_ids = [5, 1, 8, 2, 9, 7, 10, None] service_names = [ - 'trace@driver@1', + 'trace@Driver@1', '/bin/sleep', - 'trace@simple_sleep@2', + 'trace@SimpleSleep@2', '/bin/sleep', - 'trace@simple_sleep@2', - 'trace@driver@1', - 'trace@driver@1', + 'trace@SimpleSleep@2', + 'trace@Driver@1', + 'trace@Driver@1', 'trace@FRAMEWORK@Framework@0', ] names = ['init(0)', '1', 'step(0)', '1', 'step(0)', 'step(0)', 'finalize(0)', None] diff --git a/tests/utils/test_ensemble_csv.py b/tests/utils/test_ensemble_csv.py index bf9dea8c..4d6b127a 100644 --- a/tests/utils/test_ensemble_csv.py +++ b/tests/utils/test_ensemble_csv.py @@ -19,7 +19,7 @@ def test_instances_to_csv(): } instances = group_ensemble_variables_into_instances(variables, 'this_is_my_name') expected_result = b'''\ -ensemble_name,a_comp:A,a_comp:B,a_comp:C,another_comp:D,another_comp:B,another_comp:F\r +sim_name,a_comp:A,a_comp:B,a_comp:C,another_comp:D,another_comp:B,another_comp:F\r this_is_my_name0,3,2.34,"""the quick, brown fox""",7,0.775,xyzzy\r this_is_my_name1,2,5.82,baz,5,0.08,plud\r this_is_my_name2,4,0.1,quux,9,29.2,thud\r