diff --git a/pyproject.toml b/pyproject.toml index 973dd58c..a3158b3e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ classifiers = [ dynamic = ["version"] dependencies = [ "retrying>=1.3.0", + "parsl @ git+https://github.com/emlys/parsl.git@windows-support" ] [project.optional-dependencies] diff --git a/taskgraph/Task.py b/taskgraph/Task.py index c13a19a3..f1a25d51 100644 --- a/taskgraph/Task.py +++ b/taskgraph/Task.py @@ -1,23 +1,45 @@ """Task graph framework.""" -import collections +import concurrent.futures +from concurrent.futures import Future +from functools import lru_cache import hashlib import inspect import logging import logging.handlers -import multiprocessing -import multiprocessing.pool import os import pathlib import pickle -import pprint -import queue -import sqlite3 +import sys import threading import time +import types +from typing import Callable +from typing import Any, Dict, List, Literal, Optional, Sequence from importlib.metadata import PackageNotFoundError from importlib.metadata import version -import retrying +import parsl +from parsl.concurrent import ParslPoolExecutor +from parsl.configs.local_threads import config +from parsl.dataflow.taskrecord import TaskRecord +from parsl.errors import ConfigurationError +from parsl.utils import Timer +from parsl.utils import get_all_checkpoints +from parsl.dataflow.memoization import BasicMemoizer, id_for_memo +from parsl.dataflow.memoization import CheckpointCommand +from parsl.dataflow.errors import BadCheckpoint +from parsl.executors import HighThroughputExecutor +from parsl.executors.high_throughput import interchange +from parsl.executors import ThreadPoolExecutor +from parsl.log_utils import DEFAULT_FORMAT +from parsl.logconfigs.base import LogConfig +from .parsl_app import create_parsl_app + +LOGGER = logging.getLogger(__name__) + +# LOGGING_MANAGER = multiprocessing.Manager() +# LOGGING_QUEUE = LOGGING_MANAGER.Queue(-1) + try: __version__ = version('taskgraph') @@ -25,9 +47,7 @@ # package is not installed; no metadata available pass - _VALID_PATH_TYPES = (str, pathlib.PurePath) -_TASKGRAPH_DATABASE_FILENAME = 'taskgraph_data.db' try: import psutil @@ -43,198 +63,655 @@ except ImportError: HAS_PSUTIL = False -LOGGER = logging.getLogger(__name__) _MAX_TIMEOUT = 5.0 # amount of time to wait for threads to terminate -# We want our processing pool to be nondeamonic so that workers could use -# multiprocessing if desired (deamonic processes cannot start new processes) -# the following bit of code to do this was taken from -# https://stackoverflow.com/a/8963618/42897 -class NoDaemonProcess(multiprocessing.Process): - """Make 'daemon' attribute always return False.""" +# Because you'll want to define the logging configurations for listener and workers, the +# listener and worker process functions take a configurer parameter which is a callable +# for configuring logging for that process. These functions are also passed the queue, +# which they use for communication. +# +# In practice, you can configure the listener however you want, but note that in this +# simple example, the listener does not apply level or filter logic to received records. +# In practice, you would probably want to do this logic in the worker processes, to avoid +# sending events which would be filtered out between processes. +# +# The size of the rotated files is made small so you can see the results easily. +def listener_configurer(): + root = logging.getLogger() + h = logging.handlers.RotatingFileHandler('mptest.log', 'a', 300, 10) + f = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s') + h.setFormatter(f) + root.addHandler(h) + + +# This is the listener process top-level loop: wait for logging events +# (LogRecords)on the queue and handle them, quit when you get a None for a +# LogRecord. +def listener_process(queue, configurer): + configurer() + while True: + try: + record = queue.get() + if record is None: # We send this as a sentinel to tell the listener to quit. + break + logger = logging.getLogger(record.name) + logger.handle(record) # No level or filter logic applied - just do it! + except Exception: + import sys, traceback + print('Whoops! Problem:', file=sys.stderr) + traceback.print_exc(file=sys.stderr) - @property - def daemon(self): - """Return False indicating not a daemon process.""" - return False - @daemon.setter - def daemon(self, value): - """Do not allow daemon value to be overriden.""" - pass +def _null_func(): + """Use when func=None on add_task.""" + return None -class NoDaemonContext(type(multiprocessing.get_context('spawn'))): - """From https://stackoverflow.com/a/8963618/42897. +class FileLogging(LogConfig): + """Copy of parsl.logconfigs.file.FileLogging, but doesn't filter out non-parsl logs.""" - "As the current implementation of multiprocessing [3.7+] has been - extensively refactored to be based on contexts, we need to provide a - NoDaemonContext class that has our NoDaemonProcess as attribute. - [NonDaemonicPool] will then use that context instead of the default - one." "spawn" is chosen as default since that is the default and only - context option for Windows and is the default option for Mac OS as - well since 3.8. + def __init__(self, *, + format_string: str = DEFAULT_FORMAT, + level: int = logging.DEBUG): + super().__init__() + self.format_string = format_string + self.level = level - """ + def initialize_logging(self, *, log_dir: pathlib.Path, log_name: str) -> Callable[[], None]: - Process = NoDaemonProcess + log_dir.mkdir(exist_ok=True, parents=True) + filename = log_dir.joinpath(log_name + ".log") + handler = logging.FileHandler(filename) + handler.setLevel(self.level) + formatter = logging.Formatter(self.format_string, datefmt='%Y-%m-%d %H:%M:%S') + handler.setFormatter(formatter) -class NonDaemonicPool(multiprocessing.pool.Pool): - """NonDaemonic Process Pool.""" + logger = logging.getLogger() - def __init__(self, *args, **kwargs): - """Invoke super to set the context of Pool class explicitly.""" - kwargs['context'] = NoDaemonContext() - super(NonDaemonicPool, self).__init__(*args, **kwargs) + if logger.level == logging.NOTSET: + logger.setLevel(self.level) + else: + logger.setLevel(min(logger.level, self.level)) + logger.addHandler(handler) + logger.info(f'logging to {filename}') -def _null_func(): - """Use when func=None on add_task.""" - return None + # Concurrent.futures errors are also of interest, as exceptions + # which propagate out of the top of a callback are logged this way + # and then discarded. (see #240) + futures_logger = logging.getLogger("concurrent.futures") + if futures_logger.level == logging.NOTSET: + futures_logger.setLevel(self.level) + else: + futures_logger.setLevel(min(futures_logger.level, self.level)) + futures_logger.addHandler(handler) + def unregister_callback() -> None: + logger.info('Unregistering this logger') + handler.close() + logger.removeHandler(handler) + futures_logger.removeHandler(handler) -def _initialize_logging_to_queue(logging_queue): - """Add a synchronized queue to a new process. + return unregister_callback - This is intended to be called as an initialization function to - ``multiprocessing.Pool`` to establish logging from a Pool worker to the - main python process via a multiprocessing Queue. - Args: - logging_queue (multiprocessing.Queue): The queue to use for passing - log records back to the main process. +FUNC_SOURCE_MAP = {} +def get_task_hash_info(func, args, kwargs, target_path_list): + """Return hash used to compare equality of task executions. - Returns: - None + Data used in the hash: + + - the __name__ of func, or '' if it has no __name__ attribute + - the source code of func as returned by inspect.getsource, or + '' if inspecting source isn't possible + - 'cleaned' args and kwargs: """ - root_logger = logging.getLogger() + try: + # memoize func source code because it's likely we'll import + # the same func many times and reflection is slow + if func not in FUNC_SOURCE_MAP: + FUNC_SOURCE_MAP[func] = inspect.getsource(func) + source_code = FUNC_SOURCE_MAP[func] + except (IOError, TypeError): + # many reasons for this, for example, frozen Python code won't + # have source code, so just leave blank + source_code = '' + + args_clean = [] + for index, arg in enumerate(args): + try: + scrubbed_value = _scrub_task_args(arg, target_path_list) + _ = pickle.dumps(scrubbed_value) + args_clean.append(scrubbed_value) + except TypeError: + LOGGER.warning( + "could not pickle argument at index %d (%s). " + "Skipping argument which means it will not be considered " + "when calculating whether inputs have been changed " + "on a successive run.", index, arg) + + kwargs_clean = {} + # iterate through sorted order so we get the same hash result with the + # same set of kwargs irrespective of the item dict order. + for key, arg in sorted(kwargs.items()): + try: + scrubbed_value = _scrub_task_args(arg, target_path_list) + _ = pickle.dumps(scrubbed_value) + kwargs_clean[key] = scrubbed_value + except TypeError: + LOGGER.warning( + "could not pickle kw argument %s (%s) scrubbed to %s. " + "Skipping argument which means it will not be considered " + "when calculating whether inputs have been changed " + "on a successive run.", key, arg, scrubbed_value) + + return { + 'func_name': getattr(func, '__name__', ''), + 'args_clean': args_clean, + 'kwargs_clean': kwargs_clean, + 'source_code_hash': hashlib.sha1( + source_code.encode('utf-8')).hexdigest(), + } + + +def get_task_hash(func, args, kwargs, target_path_list): + hash_info = get_task_hash_info(func, args, kwargs, target_path_list) + argument_hash_string = ':'.join([ + repr(hash_info[key]) + for key in sorted(hash_info.keys())]) + return hashlib.sha1( + argument_hash_string.encode('utf-8')).hexdigest() + + +# overwrite parsl's default function serializer to include the source code +# +# the LRU cache decorator must be applied closer to the id_for_memo_function call +# that the .register() call, so that the cache-decorated version is registered. +@id_for_memo.register(types.FunctionType) +@lru_cache() +def id_for_memo_function(f: types.FunctionType, output_ref: bool = False) -> bytes: + """Checkpoint a function based on its name, module, and source code.""" + try: + # memoize func source code because it's likely we'll import + # the same func many times and reflection is slow + if f not in FUNC_SOURCE_MAP: + FUNC_SOURCE_MAP[f] = inspect.getsource(f) + source_code = FUNC_SOURCE_MAP[f] + except (IOError, TypeError): + # many reasons for this, for example, frozen Python code won't + # have source code, so just leave blank + source_code = '' + return pickle.dumps(["types.FunctionType", f.__name__, f.__module__, source_code]) - # By the time this function is called, `root_logger` has a copy of all of - # the logging handlers registered to it within the parent process, which - # leads to duplicate logging in some cases. By removing all of the - # handlers here, we ensure that log messages can only be passed back to the - # parent process by the `logging_queue`, where they will be handled. - for handler in root_logger.handlers[:]: - root_logger.removeHandler(handler) - root_logger.setLevel(logging.NOTSET) - handler = logging.handlers.QueueHandler(logging_queue) - root_logger.addHandler(handler) +class TaskGraphMemoizer(BasicMemoizer): + """Memoizer is responsible for ensuring that identical work is not repeated. + When a task is repeated, i.e., the same function is called with the same exact arguments, the + result from a previous execution is reused. `wiki `_ -def _logging_queue_monitor(logging_queue): - """Monitor ``logging_queue`` for message and pass to ``logger``.""" - LOGGER.debug('Starting logging worker') - while True: - record = logging_queue.get() - if record is None: - break - logger = logging.getLogger(record.name) - logger.handle(record) - LOGGER.debug('_logging_queue_monitor shutting down') + The memoizer implementation here does not collapse duplicate calls + at call time, but works **only** when the result of a previous + call is available at the time the duplicate call is made. + For instance:: -def _create_taskgraph_table_schema(taskgraph_database_path): - """Create database exists and/or ensures it is compatible and recreate. + No advantage from Memoization helps + memoization here: here: - Args: - taskgraph_database_path (str): path to an existing database or desired - location of a new database. + TaskA TaskB + | TaskA | + | | TaskA done (TaskB) + | | | (TaskB) + done | | + done | + done - Returns: - None. + The memoizer creates a lookup table by hashing the function name + and its inputs, and storing the results of the function. + + When a task is ready for launch, i.e., all of its arguments + have resolved, we add its hash to the task datastructure. """ - sql_create_projects_table_script = ( + + hash_algorithm: str = 'sizetimestamp' + + def __init__(self, *, + checkpoint_period: Optional[str] = None, + checkpoint_mode: Literal['task_exit', 'periodic', 'dfk_exit', 'manual'] | None = None, + memoize: bool = True): + """Initialize the memoizer. + + KWargs: + + - checkpoint_files : sequence of str, optional + List of paths to checkpoint files to load. By default, all checkpoints from the + run directory will be restored. This is usually the right behaviour, but this + parameter allows that behaviour to be overridden. + See :func:`parsl.utils.get_all_checkpoints` and :func:`parsl.utils.get_last_checkpoint` + for helpers. + - checkpoint_period : str, optional + Time interval (in "HH:MM:SS") at which to checkpoint completed tasks. Only has an effect if + ``checkpoint_mode='periodic'``. + - checkpoint_mode : str, optional + Checkpoint mode to use, can be ``'dfk_exit'``, ``'task_exit'``, ``'periodic'`` or ``'manual'``. + If set to `None`, checkpointing will be disabled. Default is None. + - memoize : str, enable memoization or not. + """ - CREATE TABLE taskgraph_data ( - task_reexecution_hash TEXT NOT NULL, - target_path_stats BLOB NOT NULL, - result BLOB NOT NULL, - PRIMARY KEY (task_reexecution_hash) - ); - CREATE TABLE global_variables ( - key TEXT NOT NULL, - value BLOB, - PRIMARY KEY (key) - ); - """) - - table_valid = True - expected_table_column_name_map = { - 'taskgraph_data': [ - 'task_reexecution_hash', 'target_path_stats', 'result'], - 'global_variables': ['key', 'value']} - if os.path.exists(taskgraph_database_path): + self.checkpointed_tasks = 0 + + # this lock must be held when: + # * writing to any checkpoint files + # * interacting with self.checkpointable_tasks + self._checkpoint_lock = threading.Lock() + + self.checkpoint_mode = checkpoint_mode + self.checkpoint_period = checkpoint_period + + self.checkpointable_tasks: List[CheckpointCommand] = [] + + self._checkpoint_timer: Timer | None = None + self.memoize = memoize + + def is_precalculated( + self, task, target_hash_algorithm, func, args, kwargs, + target_path_list, ignore_path_list, ignore_directories): + """Return true if _call need not be invoked. + + If the task has been precalculated it will fetch the return result from + the previous run. + + Returns: + True if the Task's target paths exist in the same state as the + last recorded run at the time this function is called. It is + possible this value could change without running the Task if + input parameter file stats change. False otherwise. + + """ + # This gets a list of the files and their file stats that can be found + # in args and kwargs but ignores anything specifically targeted or + # an expected result. This will allow a task to change its hash in + # case a different version of a file was passed in. + # these are the stats of the files that exist that aren't ignored try: - # check that the tables exist and the column names are as expected - for expected_table_name in expected_table_column_name_map: - table_result = _execute_sqlite( - ''' - SELECT name - FROM sqlite_master - WHERE type='table' AND name=? - ''', taskgraph_database_path, - argument_list=[expected_table_name], - mode='read_only', execute='execute', fetch='all') - if not table_result: - raise ValueError(f'missing table {expected_table_name}') - - # this query returns a list of results of the form - # [(0, 'task_reexecution_hash', 'TEXT', 1, None, 1), ... ] - # we'll just check that the header names are the same, no - # need to be super aggressive, also need to construct the - # PRAGMA string directly since it doesn't take arguments - table_info_result = _execute_sqlite( - f'PRAGMA table_info({expected_table_name})', - taskgraph_database_path, mode='read_only', - execute='execute', fetch='all') - - expected_column_names = expected_table_column_name_map[ - expected_table_name] - header_count = 0 - for header_line in table_info_result: - column_name = header_line[1] - if column_name not in expected_column_names: - raise ValueError( - f'expected {column_name} in table ' - f'{expected_table_name} but not found') - header_count += 1 - if header_count < len(expected_column_names): - raise ValueError( - f'found only {header_count} of an expected ' - f'{len(expected_column_names)} columns in table ' - f'{expected_table_name}') - if not table_info_result: - raise ValueError(f'missing table {expected_table_name}') - except Exception: - # catch all "Exception"s because anything that goes wrong while - # checking the database should be considered a bad database and we - # should make a new one. - LOGGER.exception( - f'{taskgraph_database_path} exists, but is incompatible ' - 'somehow. Deleting and making a new one.') - os.remove(taskgraph_database_path) - table_valid = False - else: - # table does not exist - table_valid = False - - if not table_valid: - # create the base table - _execute_sqlite( - sql_create_projects_table_script, taskgraph_database_path, - mode='modify', execute='script') - # set the database version - _execute_sqlite( - ''' - INSERT OR REPLACE INTO global_variables - VALUES ("version", ?) - ''', taskgraph_database_path, mode='modify', - argument_list=(__version__,)) + # memoize func source code because it's likely we'll import + # the same func many times and reflection is slow + if func not in FUNC_SOURCE_MAP: + FUNC_SOURCE_MAP[func] = inspect.getsource(func) + source_code = FUNC_SOURCE_MAP[func] + except (IOError, TypeError): + # many reasons for this, for example, frozen Python code won't + # have source code, so just leave blank + source_code = '' + + args_clean = [] + for index, arg in enumerate(args): + try: + scrubbed_value = _scrub_task_args(arg, target_path_list) + _ = pickle.dumps(scrubbed_value) + args_clean.append(scrubbed_value) + except TypeError: + LOGGER.warning( + "could not pickle argument at index %d (%s). " + "Skipping argument which means it will not be considered " + "when calculating whether inputs have been changed " + "on a successive run.", index, arg) + + kwargs_clean = {} + # iterate through sorted order so we get the same hash result with the + # same set of kwargs irrespective of the item dict order. + for key, arg in sorted(kwargs.items()): + try: + scrubbed_value = _scrub_task_args(arg, target_path_list) + _ = pickle.dumps(scrubbed_value) + kwargs_clean[key] = scrubbed_value + except TypeError: + LOGGER.warning( + "could not pickle kw argument %s (%s) scrubbed to %s. " + "Skipping argument which means it will not be considered " + "when calculating whether inputs have been changed " + "on a successive run.", key, arg, scrubbed_value) + + for target_path in target_path_list: + if not os.path.exists(_normalize_path(target_path)): + LOGGER.debug(f'Could not create a hash because {target_path} does not exist.') + return False + + target_file_stat_list = list(_get_file_stats( + base_value=target_path_list, + hash_algorithm=target_hash_algorithm, + ignore_list=[], + ignore_directories=ignore_directories)) + + non_target_file_stat_list = list(_get_file_stats( + base_value=[args, kwargs], + hash_algorithm=target_hash_algorithm, + ignore_list=target_path_list + ignore_path_list, + ignore_directories=ignore_directories)) + + other_arguments = _filter_non_files( + [args_clean, kwargs_clean], + target_path_list, + ignore_path_list, + ignore_directories) + + try: + mismatched_target_file_list = [] + # for path, hash_string in result_target_path_stats: + # if path not in target_path_list: + # mismatched_target_file_list.append( + # 'Recorded path not in target path list %s' % path) + # if not os.path.exists(path): + # mismatched_target_file_list.append( + # 'Path not found: %s' % path) + # continue + # elif target_hash_algorithm == 'exists': + # # this is the case where hash_algorithm == 'exists' but + # # we already know the file exists so we do nothing + # continue + # if target_hash_algorithm == 'sizetimestamp': + # size, modified_time, actual_path = [ + # x for x in hash_string.split('::')] + # if actual_path != path: + # mismatched_target_file_list.append( + # "Path names don't match\n" + # "cached: (%s)\nactual (%s)" % (path, actual_path)) + + # # Using nanosecond resolution for mtime (instead of the + # # usual float result of os.path.getmtime()) allows us to + # # precisely compare modification time because we're + # # comparing ints: st_mtime_ns always returns an int. + # # + # # Timestamp resolution: the python docs note that "many + # # filesystems do not provide nanosecond precision". + # # This is true (e.g. FAT, FAT32 timestamps are only + # # accurate to within 2 seconds), but the data read from the + # # filesystem will be consistent. This lets us know + # # whether the timestamp changed. This also means that, on + # # FAT filesystems, if a file is changed within 2s of its + # # creation time, we might not be able to detect it. This + # # is a weakness of FAT, not taskgraph. + # target_modified_time = os.stat(path).st_mtime_ns + # if not int(modified_time) == target_modified_time: + # mismatched_target_file_list.append( + # "Modified times don't match " + # "cached: (%f) actual: (%f)" % ( + # float(modified_time), target_modified_time)) + # continue + # target_size = os.path.getsize(path) + # if float(size) != target_size: + # mismatched_target_file_list.append( + # "File sizes don't match " + # "cached: (%s) actual: (%s)" % ( + # size, target_size)) + # else: + # target_hash = _hash_file(path, target_hash_algorithm) + # if hash_string != target_hash: + # mismatched_target_file_list.append( + # "File hashes are different. cached: (%s) " + # "actual: (%s)" % (hash_string, target_hash)) + if mismatched_target_file_list: + LOGGER.info( + "Task hash exists, but there are these mismatches: %s", + '\n'.join(mismatched_target_file_list)) + return False + + # # if kwargs contains an outputs parameter, that parameter is removed + # # and normalised differently - with output_ref set to True. + # # kwargs listed in ignore_for_cache will also be removed + filtered_kw = kwargs.copy() + # ignore_list = task['ignore_for_cache'] + + # LOGGER.debug("Ignoring these kwargs for checkpointing: %s", ignore_list) + # for k in ignore_list: + # LOGGER.debug("Ignoring kwarg %s", k) + # del filtered_kw[k] + + # add the file stat list to the already existing reexecution info + # dictionary that contains stats that should not change whether + # files have been created/updated/or not. + reexecution_info = { + 'func_name': getattr(func, '__name__', ''), + 'args_clean': args_clean, + 'kwargs_clean': kwargs_clean, + 'source_code_hash': hashlib.sha1( + source_code.encode('utf-8')).hexdigest(), + 'target_file_stat_list': target_file_stat_list, + 'non_target_file_stat_list': non_target_file_stat_list, + 'other_arguments': other_arguments, + 'non_outputs': get_task_hash_info( + task['func'], + task['args'], + filtered_kw, + task['kwargs']['_taskgraph_task_info']['target_path_list']), + 'filtered_kw': filtered_kw, + 'func': task['func'], + 'args': task['args'] + } + + binary_reexecution_info = id_for_memo(reexecution_info) + + if 'outputs' in task['kwargs']: + del filtered_kw['outputs'] + reexecution_info['outputs'] = task['kwargs']['outputs'] + binary_reexecution_info += id_for_memo( + task['kwargs']['outputs'], output_ref=True) + + hashsum = hashlib.md5(binary_reexecution_info).hexdigest() + LOGGER.debug("Task {} has memoization hash {}".format(task['id'], hashsum)) + task['hashsum'] = hashsum + + checkpoint_dir = task['kwargs']['_taskgraph_task_info']['checkpoint_dir'] + if checkpoint_dir in self.memo_lookup_table: + if hashsum in self.memo_lookup_table[checkpoint_dir]: + return True + else: + return False + else: + LOGGER.info('task checkpoint dir has not been loaded') + return False + except EOFError: + LOGGER.exception("EOFError") + return False + + def check_memo(self, task: TaskRecord) -> Optional[Future[Any]]: + """Create a hash of the task and its inputs and check the lookup table for this hash. + + If present, the results are returned. + + Args: + - task(task) : task from the dfk.tasks table + + Returns: + - Result of the function if present in table, wrapped in a Future + + This call will also set task['hashsum'] to the unique hashsum for the func+inputs. + """ + task_id = task['id'] + + if not self.memoize or not task['memoize']: + task['hashsum'] = None + LOGGER.debug("Task {} will not be memoized".format(task_id)) + return None + + result = None + if self.is_precalculated( + task=task, + target_hash_algorithm=task['kwargs']['_taskgraph_task_info']['target_hash_algorithm'], + func=task['kwargs']['_taskgraph_task_info']['func'], + args=task['kwargs']['_taskgraph_task_info']['args'], + kwargs=task['kwargs']['_taskgraph_task_info']['kwargs'], + target_path_list=task['kwargs']['_taskgraph_task_info']['target_path_list'], + ignore_path_list=task['kwargs']['_taskgraph_task_info']['ignore_path_list'], + ignore_directories=task['kwargs']['_taskgraph_task_info']['ignore_directories']): + checkpoint_dir = task['kwargs']['_taskgraph_task_info']['checkpoint_dir'] + result = self.memo_lookup_table[checkpoint_dir][task['hashsum']] + LOGGER.info("Task %s using result from cache", task_id) + else: + LOGGER.info("Task %s had no result in cache", task_id) + + assert isinstance(result, Future) or result is None + return result + + def add_checkpoint_dir(self, checkpoint_dir): + self.memo_lookup_table[checkpoint_dir] = self.load_checkpoints( + get_all_checkpoints(checkpoint_dir)) + + def start(self, *, run_dir: str, config_run_dir: str) -> None: + + self.run_dir = run_dir + self.config_run_dir = config_run_dir + + if self.memoize: + LOGGER.info("App caching enabled") + self.memo_lookup_table = {} + else: + LOGGER.info("App caching disabled for all apps") + + if self.checkpoint_mode == "periodic": + if self.checkpoint_period is None: + raise ConfigurationError("Checkpoint period must be specified with periodic checkpoint mode") + else: + try: + h, m, s = map(int, self.checkpoint_period.split(':')) + except Exception: + raise ConfigurationError("invalid checkpoint_period provided: {0} expected HH:MM:SS".format(self.checkpoint_period)) + checkpoint_period = (h * 3600) + (m * 60) + s + self._checkpoint_timer = Timer(self.checkpoint_queue, interval=checkpoint_period, name="Checkpoint") + + def _load_checkpoints(self, checkpointDirs: Sequence[str]) -> Dict[str, Future[Any]]: + """Load a checkpoint file into a lookup table. + + The data being loaded from the pickle file mostly contains input + attributes of the task: func, args, kwargs, env... + To simplify the check of whether the exact task has been completed + in the checkpoint, we hash these input params and use it as the key + for the memoized lookup table. + + Args: + - checkpointDirs (list) : List of filepaths to checkpoints + Eg. ['runinfo/001', 'runinfo/002'] + + Returns: + - memoized_lookup_table (dict) + """ + memo_lookup_table = {} + + for checkpoint_dir in checkpointDirs: + LOGGER.info("Loading checkpoints from {}".format(checkpoint_dir)) + checkpoint_file = os.path.join(checkpoint_dir, 'tasks.pkl') + try: + with open(checkpoint_file, 'rb') as f: + while True: + try: + data = pickle.load(f) + # Copy and hash only the input attributes + memo_fu: Future = Future() + assert data['exception'] is None + memo_fu.set_result(data['result']) + memo_lookup_table[data['hash']] = memo_fu + + except EOFError: + # Done with the checkpoint file + break + except FileNotFoundError: + reason = "Checkpoint file was not found: {}".format( + checkpoint_file) + LOGGER.error(reason) + raise BadCheckpoint(reason) + except Exception as ex: + reason = "Failed to load checkpoint: {}".format( + checkpoint_file) + LOGGER.error(reason) + raise BadCheckpoint(reason) + + LOGGER.info("Completed loading checkpoint: {0} with {1} tasks".format(checkpoint_file, + len(memo_lookup_table.keys()))) + return memo_lookup_table + + def _update_memo(self, task: TaskRecord) -> None: + """Updates the memoization lookup table with the result from a task. + This doesn't move any values around but associates the memoization + hashsum with the completed (by success or failure) AppFuture. + + Args: + - task (TaskRecord) : A task record from dfk.tasks + """ + task_id = task['id'] + self.is_precalculated( + task=task, + target_hash_algorithm=task['kwargs']['_taskgraph_task_info']['target_hash_algorithm'], + func=task['kwargs']['_taskgraph_task_info']['func'], + args=task['kwargs']['_taskgraph_task_info']['args'], + kwargs=task['kwargs']['_taskgraph_task_info']['kwargs'], + target_path_list=task['kwargs']['_taskgraph_task_info']['target_path_list'], + ignore_path_list=task['kwargs']['_taskgraph_task_info']['ignore_path_list'], + ignore_directories=task['kwargs']['_taskgraph_task_info']['ignore_directories']) + + if not self.memoize or not task['memoize'] or 'hashsum' not in task: + return + + if not isinstance(task['hashsum'], str): + LOGGER.error("Attempting to update app cache entry but hashsum is not a string key") + return + + checkpoint_dir = task['kwargs']['_taskgraph_task_info']['checkpoint_dir'] + if task['hashsum'] in self.memo_lookup_table[checkpoint_dir]: + LOGGER.info(f"Replacing app cache entry {task['hashsum']} with result from task {task_id}") + else: + LOGGER.debug(f"Storing app cache entry {task['hashsum']} with result from task {task_id}") + self.memo_lookup_table[checkpoint_dir][task['hashsum']] = task['app_fu'] + + def _checkpoint_these_tasks(self, checkpoint_queue: List[CheckpointCommand]) -> None: + """Play a sequence of CheckpointCommands into a checkpoint file. + + The checkpoint lock must be held when invoking this method. + """ + + # This checks that the lock is held, at least, but does not check that + # it is held by the current thread - threading.Lock does not have a + # concept of locking thread for threading.Lock. + assert self._checkpoint_lock.locked(), "checkpoint system should be locked" + + count = 0 + + for cc in checkpoint_queue: + if cc.exception is None: + checkpoint_dir = pathlib.Path(cc.task_record['kwargs'][ + '_taskgraph_task_info']['checkpoint_dir'] + ) / 'all_runs' / 'checkpoint' + if not os.path.exists(checkpoint_dir): + os.makedirs(checkpoint_dir, exist_ok=True) + checkpoint_filepath = checkpoint_dir / 'tasks.pkl' + with open(checkpoint_filepath, 'ab') as f: + hashsum = cc.task_record['hashsum'] + if not hashsum: + continue + t = {'hash': hashsum, 'exception': None, 'result': cc.result} + + # We are using pickle here since pickle dumps to a file in 'ab' + # mode behave like a incremental log. + pickle.dump(t, f) + count += 1 + LOGGER.debug("Task %s checkpointed", cc.task_record['id']) + + self.checkpointed_tasks += count + + if count == 0: + if self.checkpointed_tasks == 0: + LOGGER.warning("No tasks checkpointed so far in this run. Please ensure caching is enabled") + else: + LOGGER.debug("No tasks checkpointed in this pass.") + else: + LOGGER.info("Done checkpointing {} tasks".format(count)) + + +_DFK = None class TaskGraph(object): @@ -279,86 +756,73 @@ def __init__( # to the taskgraph during `add_task` self._task_hash_map = dict() - # use this to keep track of all the tasks added to the graph by their - # task names. Used to map a unique task name to the task object it - # represents - self._task_name_map = dict() - # used to remember if task_graph has been closed self._closed = False # keep track if the task graph has been forcibly terminated self._terminated = False - # if n_workers > 0 this will be a multiprocessing pool used to execute - # the __call__ functions in Tasks - self._worker_pool = None - - # If n_workers > 0 this will be a threading.Thread used to propagate - # log records from another process into the current process. - self._logging_monitor_thread = None - - # If n_workers > 0, this will be a multiprocessing.Queue used to pass - # log records from the process pool to the parent process. - self._logging_queue = None - # keeps track of the tasks currently being processed for logging. self._active_task_list = [] - # keeps track of how many tasks have all their dependencies satisfied - # and are waiting for a worker - self._task_waiting_count = 0 - - # this might hold the threads to execute tasks if n_workers >= 0 - self._task_executor_thread_list = [] - - # executor threads wait on this event that gets set when new tasks are - # added to the queue. If the queue is empty an executor will clear - # the event to halt other executors - self._executor_ready_event = threading.Event() - - # tasks that have all their dependencies satisfied go in this queue - # and can be executed immediately - self._task_ready_priority_queue = queue.PriorityQueue() - - # maps a list of task names that need to be executed before the key - # task can - self._task_dependent_map = collections.defaultdict(set) - - # maps a list of task names that are dependent to a task - self._dependent_task_map = collections.defaultdict(set) - # tasks that complete are added to this set self._completed_task_names = set() - self._task_database_path = os.path.join( - self._taskgraph_cache_dir_path, _TASKGRAPH_DATABASE_FILENAME) - - # create new table if needed - _create_taskgraph_table_schema(self._task_database_path) - - # check the version of the database and warn if a problem - local_version = _execute_sqlite( - ''' - SELECT value - FROM global_variables - WHERE key=? - ''', self._task_database_path, mode='read_only', - fetch='one', argument_list=['version'])[0] - if local_version != __version__: - LOGGER.warning( - f'the database located at {self._task_database_path} was ' - f'created with TaskGraph version {local_version} but the ' - f'current version is {__version__}') + # set_file_logger(f'{self._taskgraph_cache_dir_path}/parsl.log') + + # self.listener = logging.handlers.QueueListener( + # LOGGING_QUEUE, logging.FileHandler( + # f'{self._taskgraph_cache_dir_path}/taskgraph.log')) + # self.listener.start() + global _DFK + if _DFK: + try: + _DFK.cleanup() + except Exception: + pass + config.run_dir = self._taskgraph_cache_dir_path + config.app_cache = True + config.memoizer = TaskGraphMemoizer(checkpoint_mode='task_exit') + config.initialize_logging = FileLogging() + _DFK = parsl.load(config) + + _DFK.config.memoizer.add_checkpoint_dir(self._taskgraph_cache_dir_path) + if n_workers > 0: + self.executor_label = "htex" + executor = HighThroughputExecutor( + label=self.executor_label, + cores_per_worker=1, + max_workers_per_node=n_workers, + working_dir=self._taskgraph_cache_dir_path, + address='127.0.0.1', + encrypted=False, + interchange_launch_cmd=[ + sys.executable, + interchange.__file__]) + elif n_workers == 0: + self.executor_label = "threaded" + executor = ThreadPoolExecutor( + label=self.executor_label, + max_threads=2) + else: + self.executor_label = "threaded" + executor = ThreadPoolExecutor( + label=self.executor_label, + max_threads=1) + + _DFK.add_executors([executor]) + self.pool = ParslPoolExecutor(dfk=_DFK) - # no need to set up schedulers if n_workers is single threaded + # # no need to set up schedulers if n_workers is single threaded self._n_workers = n_workers - if n_workers < 0: - return + # if n_workers < 0: + # return # start concurrent reporting of taskgraph if reporting interval is set self._reporting_interval = reporting_interval if reporting_interval is not None: + if reporting_interval <= 0: + raise ValueError('reporting_interval must be greater than 0') self._execution_monitor_wait_event = threading.Event() self._execution_monitor_thread = threading.Thread( target=self._execution_monitor, @@ -369,30 +833,8 @@ def __init__( self._execution_monitor_thread.daemon = True self._execution_monitor_thread.start() - # launch executor threads - self._executor_thread_count = max(0, n_workers) - for thread_id in range(max(1, n_workers)): - task_executor_thread = threading.Thread( - target=self._task_executor, - name='task_executor_%s' % thread_id) - # make daemons in case there's a catastrophic error the main - # thread won't hang - task_executor_thread.daemon = True - task_executor_thread.start() - self._task_executor_thread_list.append(task_executor_thread) - # set up multiprocessing if n_workers > 0 if n_workers > 0: - self._logging_queue = multiprocessing.Queue() - self._worker_pool = NonDaemonicPool( - n_workers, initializer=_initialize_logging_to_queue, - initargs=(self._logging_queue,)) - self._logging_monitor_thread = threading.Thread( - target=_logging_queue_monitor, - args=(self._logging_queue,)) - - self._logging_monitor_thread.daemon = True - self._logging_monitor_thread.start() if HAS_PSUTIL: parent = psutil.Process() parent.nice(PROCESS_LOW_PRIORITY) @@ -408,113 +850,15 @@ def __init__( def __del__(self): """Ensure all threads have been joined for cleanup.""" self._terminate() - - def _task_executor(self): - """Worker that executes Tasks that have satisfied dependencies.""" - while True: - # this event blocks until the task graph has signaled it wants - # the executors to read the state of the queue or a stop event or - # a timeout exceeded just to protect against a worst case deadlock - self._executor_ready_event.wait(_MAX_TIMEOUT) - # this lock synchronizes changes between the queue and - # executor_ready_event - if self._terminated: - LOGGER.debug( - "taskgraph is terminated, ending %s", - threading.current_thread()) - break - task = None - try: - task = self._task_ready_priority_queue.get_nowait() - self._task_waiting_count -= 1 - task_name_time_tuple = (task.task_name, time.time()) - self._active_task_list.append(task_name_time_tuple) - except queue.Empty: - # no tasks are waiting could be because the taskgraph is - # closed or because the queue is just empty. - if (self._closed and len(self._completed_task_names) == - self._added_task_count): - # the graph is closed and there are as many completed tasks - # as there are added tasks, so none left. The executor can - # terminate. - self._executor_thread_count -= 1 - if self._executor_thread_count == 0 and self._worker_pool: - # only the last executor should terminate the worker - # pool, because otherwise who knows if it's still - # executing anything - try: - self._worker_pool.close() - self._worker_pool.terminate() - self._worker_pool = None - self._terminate() - except Exception: - # there's the possibility for a race condition here - # where another thread already closed the worker - # pool, so just guard against it - LOGGER.warning('worker pool was already closed') - LOGGER.debug( - "no tasks are pending and taskgraph closed, normally " - "terminating executor %s." % threading.current_thread()) - break - else: - # there's still the possibility for work to be added or - # still work in the pipeline - self._executor_ready_event.clear() - if task is None: - continue - try: - task._call() - task.task_done_executing_event.set() - except Exception as e: - # An error occurred on a call, terminate the taskgraph - task.exception_object = e - LOGGER.exception( - 'A taskgraph _task_executor failed on Task ' - '%s. Terminating taskgraph.', task.task_name) - self._terminate() - break - - LOGGER.debug( - "task %s is complete, checking to see if any dependent " - "tasks can be executed now", task.task_name) - self._completed_task_names.add(task.task_name) - self._active_task_list.remove(task_name_time_tuple) - for waiting_task_name in ( - self._task_dependent_map[task.task_name]): - # remove `task` from the set of tasks that - # `waiting_task` was waiting on. - self._dependent_task_map[waiting_task_name].remove( - task.task_name) - # if there aren't any left, we can push `waiting_task` - # to the work queue - if not self._dependent_task_map[waiting_task_name]: - # if we removed the last task we can put it to the - # work queue - LOGGER.debug( - "Task %s is ready for processing, sending to " - "task_ready_priority_queue", - waiting_task_name) - del self._dependent_task_map[waiting_task_name] - self._task_ready_priority_queue.put( - self._task_name_map[waiting_task_name]) - self._task_waiting_count += 1 - # indicate to executors there is work to do - self._executor_ready_event.set() - del self._task_dependent_map[task.task_name] - # this extra set ensures that recently emptied map won't get - # ignored by the executor if no work is left to do and the graph is - # closed - self._executor_ready_event.set() - LOGGER.debug("task %s done processing", task.task_name) - LOGGER.debug("task executor shutting down") + # self.listener.stop() def add_task( - self, func=None, args=None, kwargs=None, task_name=None, + self, func=_null_func, args=None, kwargs=None, + task_name='UNNAMED TASK', target_path_list=None, ignore_path_list=None, hash_target_files=True, dependent_task_list=None, ignore_directories=True, priority=0, - hash_algorithm='sizetimestamp', transient_run=False, - store_result=False): + hash_algorithm='sizetimestamp', transient_run=False): """Add a task to the task graph. Args: @@ -576,9 +920,6 @@ def add_task( some sort of initialization that's needed every time a TaskGraph is instantiated. Perhaps to acquire dynamic resources or authenticate permissions. - store_result (bool): If True, the result of ``func`` will be stored - in the TaskGraph database and retrievable with a call to - ``.get()`` on a ``Task`` object. Returns: Task which was just added to the graph or an existing Task that @@ -607,16 +948,12 @@ def add_task( args = [] if kwargs is None: kwargs = {} - if task_name is None: - task_name = 'UNNAMED TASK' if dependent_task_list is None: dependent_task_list = [] if target_path_list is None: target_path_list = [] if ignore_path_list is None: ignore_path_list = [] - if func is None: - func = _null_func # this is a pretty common error to accidentally not pass a # Task to the dependent task list. @@ -626,84 +963,54 @@ def add_task( "Objects passed to dependent task list that are not " "tasks: %s", dependent_task_list) - task_name = '%s (%d)' % (task_name, len(self._task_hash_map)) - new_task = Task( - task_name, func, args, kwargs, target_path_list, - ignore_path_list, hash_target_files, ignore_directories, - transient_run, self._worker_pool, - priority, hash_algorithm, store_result, - self._task_database_path) + # it is a common error to accidentally pass a non string as to the + # target path list, this terminates early if so + if any([not (isinstance(path, _VALID_PATH_TYPES)) + for path in target_path_list]): + raise ValueError( + "Values passed to target_path_list are not strings: %s", + target_path_list) + target_path_list = [_normalize_path(path) for path in target_path_list] + hash = get_task_hash(func, args, kwargs, target_path_list) - self._task_name_map[new_task.task_name] = new_task # it may be this task was already created in an earlier call, # use that object in its place - if new_task in self._task_hash_map: - duplicate_task = self._task_hash_map[new_task] - new_task_target_set = set(new_task._target_path_list) + if hash in self._task_hash_map: + duplicate_task = self._task_hash_map[hash] + new_task_target_set = set(sorted([ + _normalize_path(path) for path in target_path_list])) duplicate_task_target_set = set( duplicate_task._target_path_list) if new_task_target_set == duplicate_task_target_set: LOGGER.warning( - "A duplicate task was submitted: %s original: %s", - new_task, self._task_hash_map[new_task]) + "A duplicate task of %s was submitted", + self._task_hash_map[hash]) self._added_task_count -= 1 return duplicate_task disjoint_target_set = ( new_task_target_set.symmetric_difference( duplicate_task_target_set)) - if len(disjoint_target_set) == ( + if len(disjoint_target_set) != ( len(new_task_target_set) + len(duplicate_task_target_set)): - if duplicate_task not in dependent_task_list: - LOGGER.info( - "A task was created that had an identical " - "args signature sans target paths, but a " - "different target_path_list of the same " - "length. To avoid recomputation, dynamically " - "adding previous Task (%s) as a dependent " - "task to this one (%s).", - duplicate_task.task_name, task_name) - dependent_task_list = ( - dependent_task_list + [duplicate_task]) - else: raise RuntimeError( "A task was created that has the same arguments " "as another task, but only partially different " "expected target paths. This runs the risk of " "unpredictably overwriting output so treating as " - "a runtime error: submitted task: %s, existing " - "task: %s" % (new_task, duplicate_task)) - self._task_hash_map[new_task] = new_task - if self._n_workers < 0: - # call directly if single threaded - new_task._call() - else: - # determine if task is ready or is dependent on other - # tasks - LOGGER.debug( - "multithreaded: %s sending to new task queue.", - task_name) - outstanding_dep_task_name_list = [ - dep_task.task_name for dep_task in dependent_task_list - if dep_task.task_name - not in self._completed_task_names] - if not outstanding_dep_task_name_list: - LOGGER.debug( - "sending task %s right away", new_task.task_name) - self._task_ready_priority_queue.put(new_task) - self._task_waiting_count += 1 - self._executor_ready_event.set() - else: - # there are unresolved tasks that the waiting - # process scheduler has not been notified of. - # Record dependencies. - for dep_task_name in outstanding_dep_task_name_list: - # record tasks that are dependent on dep_task_name - self._task_dependent_map[dep_task_name].add( - new_task.task_name) - # record tasks that new_task depends on - self._dependent_task_map[new_task.task_name].add( - dep_task_name) + "a runtime error. Existing task: %s" % (duplicate_task)) + + task_name = '%s (%d)' % (task_name, len(self._task_hash_map)) + new_task = Task( + task_name, func, args, kwargs, target_path_list, + ignore_path_list, hash_target_files, ignore_directories, + transient_run, + priority, hash_algorithm, + dependent_task_list, + checkpoint_dir=self._taskgraph_cache_dir_path, + executor_label=self.executor_label) + + self._task_hash_map[hash] = new_task return new_task except Exception: @@ -731,7 +1038,6 @@ def _execution_monitor(self, monitor_wait_event): if self._terminated: break active_task_count = len(self._active_task_list) - queue_length = self._task_ready_priority_queue.qsize() active_task_message = '\n'.join( ['\t%s: executing for %.2fs' % ( task_name, time.time() - task_time) @@ -746,10 +1052,9 @@ def _execution_monitor(self, monitor_wait_event): LOGGER.info( "\n\ttaskgraph execution status: tasks added: %d \n" "\ttasks complete: %d (%.1f%%) \n" - "\ttasks waiting for a free worker: %d (qsize: %d)\n" "\ttasks executing (%d): graph is %s\n%s", self._added_task_count, completed_tasks, percent_complete, - self._task_waiting_count, queue_length, active_task_count, + active_task_count, 'closed' if self._closed else 'open', active_task_message) @@ -771,32 +1076,21 @@ def join(self, timeout=None): """ LOGGER.debug("joining taskgraph") - if self._n_workers < 0: - # Join() is meaningless since tasks execute synchronously. - LOGGER.debug( - 'n_workers: %s; join is vacuously true' % self._n_workers) - return True - try: LOGGER.debug("attempting to join threads") - timedout = False for task in self._task_hash_map.values(): - LOGGER.debug("attempting to join task %s", task.task_name) - # task.join() will raise any exception that resulted from the - # task's execution. - timedout = not task.join(timeout) - LOGGER.debug("task %s was joined", task.task_name) - # if the last task timed out then we want to timeout for all - # of the task graph - if timedout: + try: + task.join(timeout=timeout) + except concurrent.futures._base.TimeoutError: LOGGER.info( "task %s timed out in graph join", task.task_name) return False + if self._closed: # Close down the taskgraph; ok if already terminated - self._executor_ready_event.set() self._terminate() return True + except Exception: # If there's an exception on a join it means that a task failed # to execute correctly. Print a helpful message then terminate the @@ -817,7 +1111,6 @@ def close(self): self._closed = True # this wakes up all the executors and any that wouldn't otherwise # have work to do will see there are no tasks left and terminate - self._executor_ready_event.set() LOGGER.debug("taskgraph closed") def _terminate(self): @@ -826,31 +1119,11 @@ def _terminate(self): "Invoking terminate. already terminated? %s", self._terminated) if self._terminated: return + self.pool.shutdown() try: # it's possible the global state is not well defined, so just in # case we'll wrap it all up in a try/except self._terminated = True - if self._executor_ready_event is not None: - # alert executors to check that _terminated is True - self._executor_ready_event.set() - LOGGER.debug("shutting down workers") - if self._worker_pool is not None: - self._worker_pool.close() - self._worker_pool.terminate() - self._worker_pool = None - - # This will terminate the logging worker - if self._logging_queue is not None: - self._logging_queue.put(None) - - # This will cause all 'join'ed Tasks to join. - if self._n_workers >= 0: - self._executor_ready_event.set() - if self._reporting_interval is not None: - self._execution_monitor_wait_event.set() - for task in self._task_hash_map.values(): - # shortcut to get the tasks to mark as joined - task.task_done_executing_event.set() LOGGER.debug('taskgraph terminated') except Exception: @@ -864,8 +1137,8 @@ class Task(object): def __init__( self, task_name, func, args, kwargs, target_path_list, ignore_path_list, hash_target_files, ignore_directories, - transient_run, worker_pool, priority, hash_algorithm, - store_result, task_database_path): + transient_run, priority, hash_algorithm, + dependent_task_list, checkpoint_dir, executor_label): """Make a Task. Args: @@ -895,8 +1168,6 @@ def __init__( to the same object it will not be re-run in any scenario. Otherwise if False, subsequent tasks with an identical execution hash will be skipped. - worker_pool (multiprocessing.Pool): if not None, is a - multiprocessing pool that can be used for ``_call`` execution. priority (numeric): the priority of a task is considered when there is more than one task whose dependencies have been met and are ready for scheduling. Tasks are inserted into the @@ -910,20 +1181,6 @@ def __init__( of any files found in the arguments. If 'exists' will be considered the same file only if a file with the same filename exists on disk. - store_result (bool): If true, the result of ``func`` will be - stored in the TaskGraph database and retrievable with a call - to ``.get()`` on the Task object. - task_database_path (str): path to an SQLITE database that has - table named "taskgraph_data" with the three fields: - task_hash TEXT NOT NULL, - target_path_stats BLOB NOT NULL - result BLOB NOT NULL - If a call is successful its hash is inserted/updated in the - table, the target_path_stats stores the base/target stats - for the target files created by the call and listed in - ``target_path_list``, and the result of ``func`` is stored in - ``result``. - """ # it is a common error to accidentally pass a non string as to the # target path list, this terminates early if so @@ -939,343 +1196,80 @@ def __init__( self._target_path_list = sorted([ _normalize_path(path) for path in target_path_list]) self.task_name = task_name - self._func = func - self._args = args - self._kwargs = kwargs - self._ignore_path_list = [ - _normalize_path(path) for path in ignore_path_list] - self._hash_target_files = hash_target_files - self._ignore_directories = ignore_directories - self._transient_run = transient_run - self._worker_pool = worker_pool - self._task_database_path = task_database_path + ignore_path_list = [_normalize_path(path) for path in ignore_path_list] self._hash_algorithm = hash_algorithm - self._store_result = store_result - self.exception_object = None - - # invert the priority since sorting goes smallest to largest and we - # want more positive priority values to be executed first. - self._priority = -priority - - # Used to ensure only one attempt at executing and also a mechanism - # to see when Task is complete. This can be set if a Task finishes - # a _call and there are no more attempts at reexecution. - self.task_done_executing_event = threading.Event() + if not hash_target_files: + self._target_hash_algorithm = 'exists' + else: + self._target_hash_algorithm = self._hash_algorithm + + self._taskgraph_task_info = { + 'target_hash_algorithm': self._target_hash_algorithm, + 'func': func, + 'args': args, + 'kwargs': kwargs, + 'target_path_list': self._target_path_list, + 'ignore_path_list': ignore_path_list, + 'ignore_directories': ignore_directories, + 'checkpoint_dir': checkpoint_dir + } # These are used to store and later access the result of the call. self._result = None - # Calculate a hash based only on argument inputs. - try: - if not hasattr(Task, 'func_source_map'): - Task.func_source_map = {} - # memoize func source code because it's likely we'll import - # the same func many times and reflection is slow - if self._func not in Task.func_source_map: - Task.func_source_map[self._func] = ( - inspect.getsource(self._func)) - source_code = Task.func_source_map[self._func] - except (IOError, TypeError): - # many reasons for this, for example, frozen Python code won't - # have source code, so just leave blank - source_code = '' + parsl_resource_specification = {} + if executor_label == 'htex': + parsl_resource_specification['priority'] = priority - if not hasattr(self._func, '__name__'): - LOGGER.warning( - "function does not have a __name__ which means it will not " - "be considered when calculating a successive input has " - "been changed with another function without __name__.") - self._func.__name__ = '' + wrap_func = create_parsl_app( + cache=not transient_run, + executor_label=executor_label, + ignore_for_cache=ignore_path_list, + parsl_resource_specification=parsl_resource_specification) - args_clean = [] - for index, arg in enumerate(self._args): - try: - scrubbed_value = _scrub_task_args(arg, self._target_path_list) - _ = pickle.dumps(scrubbed_value) - args_clean.append(scrubbed_value) - except TypeError: - LOGGER.warning( - "could not pickle argument at index %d (%s). " - "Skipping argument which means it will not be considered " - "when calculating whether inputs have been changed " - "on a successive run.", index, arg) - - kwargs_clean = {} - # iterate through sorted order so we get the same hash result with the - # same set of kwargs irrespective of the item dict order. - for key, arg in sorted(self._kwargs.items()): - try: - scrubbed_value = _scrub_task_args(arg, self._target_path_list) - _ = pickle.dumps(scrubbed_value) - kwargs_clean[key] = scrubbed_value - except TypeError: - LOGGER.warning( - "could not pickle kw argument %s (%s) scrubbed to %s. " - "Skipping argument which means it will not be considered " - "when calculating whether inputs have been changed " - "on a successive run.", key, arg, scrubbed_value) - - self._reexecution_info = { - 'func_name': self._func.__name__, - 'args_clean': args_clean, - 'kwargs_clean': kwargs_clean, - 'source_code_hash': hashlib.sha1( - source_code.encode('utf-8')).hexdigest(), - } - - argument_hash_string = ':'.join([ - repr(self._reexecution_info[key]) - for key in sorted(self._reexecution_info.keys())]) - - self._task_id_hash = hashlib.sha1( - argument_hash_string.encode('utf-8')).hexdigest() - - # this will get calculated when ``is_precalculated`` is invoked. - self._task_reexecution_hash = None - - def __eq__(self, other): - """Two tasks are equal if their hashes are equal.""" - return ( - isinstance(self, other.__class__) and - (self._task_id_hash == other._task_id_hash)) - - def __hash__(self): - """Return the base-16 integer hash of this hash string.""" - return int(self._task_id_hash, 16) - - def __ne__(self, other): - """Inverse of __eq__.""" - return not self.__eq__(other) - - def __lt__(self, other): - """Less than based on priority.""" - return self._priority < other._priority + self.future = wrap_func( + func, *args, **kwargs, + _taskgraph_task_info=self._taskgraph_task_info, + inputs=[task.future for task in dependent_task_list]) def __repr__(self): """Create a string representation of a Task.""" - return "Task object %s:\n\n" % (id(self)) + pprint.pformat( - { - "task_name": self.task_name, - "priority": self._priority, - "ignore_path_list": self._ignore_path_list, - "ignore_directories": self._ignore_directories, - "target_path_list": self._target_path_list, - "task_id_hash": self._task_id_hash, - "task_reexecution_hash": self._task_reexecution_hash, - "exception_object": self.exception_object, - "self._reexecution_info": self._reexecution_info, - "self._result": self._result, - }) - - def _call(self): + return f"Task object {id(self)} [{self.task_name}]" + + def _call(self, timeout=None): """Invoke this method to execute task. Precondition is that the Task dependencies are satisfied. - Sets the ``self.task_done_executing_event`` flag if execution is - successful. - Raises: RuntimeError if any target paths are not generated after the function call is complete. """ - LOGGER.debug("_call check if precalculated %s", self.task_name) - if not self._transient_run and self.is_precalculated(): - self.task_done_executing_event.set() - return - LOGGER.debug("not precalculated %s", self.task_name) - - if self._worker_pool is not None: - result = self._worker_pool.apply_async( - func=self._func, args=self._args, kwds=self._kwargs) - # the following blocks and raises an exception if result - # raised an exception - LOGGER.debug("apply_async for task %s", self.task_name) - payload = result.get() - else: - LOGGER.debug("direct _func for task %s", self.task_name) - payload = self._func(*self._args, **self._kwargs) - if self._store_result: - self._result = payload - - # check that the target paths exist and record stats for later - if not self._hash_target_files: - target_hash_algorithm = 'exists' - else: - target_hash_algorithm = self._hash_algorithm - result_target_path_stats = list( - _get_file_stats( - self._target_path_list, target_hash_algorithm, [], False)) + self._result = self.future.result(timeout=timeout) + + # check that the target paths exist result_target_path_set = set( - [x[0] for x in result_target_path_stats]) - target_path_set = set(self._target_path_list) - if target_path_set != result_target_path_set: + [x[0] for x in _get_file_stats( + base_value=self._target_path_list, + hash_algorithm=self._target_hash_algorithm, + ignore_list=[], + ignore_directories=False)]) + + if set(self._target_path_list) != result_target_path_set: raise RuntimeError( "In Task: %s\nMissing expected target path results.\n" "Expected: %s\nObserved: %s\n" % ( self.task_name, self._target_path_list, result_target_path_set)) - # this step will only record the run if there is an expected - # target file. Otherwise we infer the result of this call is - # transient between taskgraph executions and we should expect to - # run it again. - if not self._transient_run: - _execute_sqlite( - "INSERT OR REPLACE INTO taskgraph_data VALUES (?, ?, ?)", - self._task_database_path, mode='modify', - argument_list=( - self._task_reexecution_hash, - pickle.dumps(result_target_path_stats), - pickle.dumps(self._result))) - self.task_done_executing_event.set() LOGGER.debug("successful run on task %s", self.task_name) - def is_precalculated(self): - """Return true if _call need not be invoked. - - If the task has been precalculated it will fetch the return result from - the previous run. - - Returns: - True if the Task's target paths exist in the same state as the - last recorded run at the time this function is called. It is - possible this value could change without running the Task if - input parameter file stats change. False otherwise. - - """ - # This gets a list of the files and their file stats that can be found - # in args and kwargs but ignores anything specifically targeted or - # an expected result. This will allow a task to change its hash in - # case a different version of a file was passed in. - # these are the stats of the files that exist that aren't ignored - if not self._hash_target_files: - target_hash_algorithm = 'exists' - else: - target_hash_algorithm = self._hash_algorithm - file_stat_list = list(_get_file_stats( - [self._args, self._kwargs], - target_hash_algorithm, - self._target_path_list+self._ignore_path_list, - self._ignore_directories)) - - other_arguments = _filter_non_files( - [self._reexecution_info['args_clean'], - self._reexecution_info['kwargs_clean']], - self._target_path_list, - self._ignore_path_list, - self._ignore_directories) - - LOGGER.debug("file_stat_list: %s", file_stat_list) - LOGGER.debug("other_arguments: %s", other_arguments) - - # add the file stat list to the already existing reexecution info - # dictionary that contains stats that should not change whether - # files have been created/updated/or not. - self._reexecution_info['file_stat_list'] = file_stat_list - self._reexecution_info['other_arguments'] = other_arguments - - reexecution_string = '%s:%s:%s:%s:%s' % ( - self._reexecution_info['func_name'], - self._reexecution_info['source_code_hash'], - self._reexecution_info['other_arguments'], - self._store_result, - # the x[1] is to only take the digest part of the 'file_stat' - str([x[1] for x in file_stat_list])) - - self._task_reexecution_hash = hashlib.sha1( - reexecution_string.encode('utf-8')).hexdigest() - try: - database_result = _execute_sqlite( - """SELECT target_path_stats, result from taskgraph_data - WHERE (task_reexecution_hash == ?)""", - self._task_database_path, mode='read_only', - argument_list=(self._task_reexecution_hash,), fetch='one') - if database_result is None: - LOGGER.debug( - "not precalculated, Task hash does not " - "exist (%s)", self.task_name) - LOGGER.debug("is_precalculated full task info: %s", self) - return False - result_target_path_stats = pickle.loads(database_result[0]) - mismatched_target_file_list = [] - for path, hash_string in result_target_path_stats: - if path not in self._target_path_list: - mismatched_target_file_list.append( - 'Recorded path not in target path list %s' % path) - if not os.path.exists(path): - mismatched_target_file_list.append( - 'Path not found: %s' % path) - continue - elif target_hash_algorithm == 'exists': - # this is the case where hash_algorithm == 'exists' but - # we already know the file exists so we do nothing - continue - if target_hash_algorithm == 'sizetimestamp': - size, modified_time, actual_path = [ - x for x in hash_string.split('::')] - if actual_path != path: - mismatched_target_file_list.append( - "Path names don't match\n" - "cached: (%s)\nactual (%s)" % (path, actual_path)) - - # Using nanosecond resolution for mtime (instead of the - # usual float result of os.path.getmtime()) allows us to - # precisely compare modification time because we're - # comparing ints: st_mtime_ns always returns an int. - # - # Timestamp resolution: the python docs note that "many - # filesystems do not provide nanosecond precision". - # This is true (e.g. FAT, FAT32 timestamps are only - # accurate to within 2 seconds), but the data read from the - # filesystem will be consistent. This lets us know - # whether the timestamp changed. This also means that, on - # FAT filesystems, if a file is changed within 2s of its - # creation time, we might not be able to detect it. This - # is a weakness of FAT, not taskgraph. - target_modified_time = os.stat(path).st_mtime_ns - if not int(modified_time) == target_modified_time: - mismatched_target_file_list.append( - "Modified times don't match " - "cached: (%f) actual: (%f)" % ( - float(modified_time), target_modified_time)) - continue - target_size = os.path.getsize(path) - if float(size) != target_size: - mismatched_target_file_list.append( - "File sizes don't match " - "cached: (%s) actual: (%s)" % ( - size, target_size)) - else: - target_hash = _hash_file(path, target_hash_algorithm) - if hash_string != target_hash: - mismatched_target_file_list.append( - "File hashes are different. cached: (%s) " - "actual: (%s)" % (hash_string, target_hash)) - if mismatched_target_file_list: - LOGGER.info( - "not precalculated (%s), Task hash exists, " - "but there are these mismatches: %s", - self.task_name, '\n'.join(mismatched_target_file_list)) - return False - if self._store_result: - self._result = pickle.loads(database_result[1]) - LOGGER.debug("precalculated (%s)" % self) - return True - except EOFError: - LOGGER.exception("not precalculated %s, EOFError", self.task_name) - return False - def join(self, timeout=None): """Block until task is complete, raise exception if runtime failed.""" - LOGGER.debug( - "joining %s done executing: %s", self.task_name, - self.task_done_executing_event) - successful_wait = self.task_done_executing_event.wait(timeout) - if self.exception_object: - raise self.exception_object - return successful_wait + LOGGER.debug(f'joining task {self.task_name}') + self._call(timeout=timeout) + return True def get(self, timeout=None): """Return the result of the ``func`` once it is ready. @@ -1293,17 +1287,9 @@ def get(self, timeout=None): Raises: RuntimeError when ``timeout`` exceeded. - ValueError if ``store_result`` was set to ``False`` when the task - was created. """ - if not self._store_result: - raise ValueError( - 'must set `store_result` to True in `add_task` to invoke this ' - 'function') - timeout = not self.join(timeout) - if timeout: - raise RuntimeError('call to get timed out') + self.join(timeout) return self._result @@ -1436,6 +1422,20 @@ def _scrub_task_args(base_value, target_path_list): This function can be called before the Task dependencies are satisfied since it doesn't inspect any file stats on disk. + Attempts to standardize task arguments according to these rules: + - if value is callable, return a string containing the __name__ and + source code as returned by inspect.getsource + - if value is a dict, return a copy of the dict where each value is + replaced with the result of calling _scrub_task_args on the original + value + - if value is a list, set, or tuple, return a copy of where each value is + replaced with the result of calling _scrub_task_args on the original + value + - if value is a string or pathlib.PurePath, attempt to convert into + a normalized, normcase, absolute filepath string. if this normalized + string exists in `target_path_list`, return 'in_target_path_list' + - otherwise, return the value unchanged + Args: base_value: any python value target_path_list (list): a list of strings that if found in @@ -1448,15 +1448,13 @@ def _scrub_task_args(base_value, target_path_list): """ if callable(base_value): try: - if not hasattr(Task, 'func_source_map'): - Task.func_source_map = {} # memoize func source code because it's likely we'll import # the same func many times and reflection is slow - if base_value not in Task.func_source_map: - Task.func_source_map[base_value] = ( + if base_value not in FUNC_SOURCE_MAP: + FUNC_SOURCE_MAP[base_value] = ( inspect.getsource(base_value)).replace( ' ', '').replace('\t', '') - source_code = Task.func_source_map[base_value] + source_code = FUNC_SOURCE_MAP[base_value] except (IOError, TypeError): # many reasons for this, for example, frozen Python code won't # have source code, so just leave blank @@ -1475,10 +1473,11 @@ def _scrub_task_args(base_value, target_path_list): return type(base_value)(result_list) elif isinstance(base_value, _VALID_PATH_TYPES): normalized_path = _normalize_path(base_value) - if normalized_path in target_path_list: - return 'in_target_path_list' - else: - return normalized_path + return normalized_path + # if normalized_path in target_path_list: + # return 'in_target_path_list' + # else: + # return normalized_path else: return base_value @@ -1529,83 +1528,3 @@ def _normalize_path(path): "failed to abspath %s so returning normalized path instead") abs_path = norm_path return os.path.normcase(abs_path) - - -@retrying.retry( - wait_exponential_multiplier=500, wait_exponential_max=3200, - stop_max_attempt_number=100) -def _execute_sqlite( - sqlite_command, database_path, argument_list=None, - mode='read_only', execute='execute', fetch=None): - """Execute SQLite command and attempt retries on a failure. - - Args: - sqlite_command (str): a well formatted SQLite command. - database_path (str): path to the SQLite database to operate on. - argument_list (list): ``execute == 'execute'`` then this list is passed - to the internal sqlite3 ``execute`` call. - mode (str): must be either 'read_only' or 'modify'. - execute (str): must be either 'execute' or 'script'. - fetch (str): if not ``None`` can be either 'all' or 'one'. - If not None the result of a fetch will be returned by this - function. - - Returns: - result of fetch if ``fetch`` is not None. - - """ - cursor = None - connection = None - try: - if mode == 'read_only': - ro_uri = r'%s?mode=ro' % pathlib.Path( - os.path.abspath(database_path)).as_uri() - LOGGER.debug( - '%s exists: %s', ro_uri, os.path.exists(os.path.abspath( - database_path))) - connection = sqlite3.connect(ro_uri, uri=True) - elif mode == 'modify': - connection = sqlite3.connect(database_path) - else: - raise ValueError('Unknown mode: %s' % mode) - - if execute == 'execute': - if argument_list is None: - cursor = connection.execute(sqlite_command) - else: - cursor = connection.execute(sqlite_command, argument_list) - elif execute == 'script': - cursor = connection.executescript(sqlite_command) - else: - raise ValueError('Unknown execute mode: %s' % execute) - - result = None - payload = None - if fetch == 'all': - payload = (cursor.fetchall()) - elif fetch == 'one': - payload = (cursor.fetchone()) - elif fetch is not None: - raise ValueError('Unknown fetch mode: %s' % fetch) - if payload is not None: - result = list(payload) - cursor.close() - connection.commit() - connection.close() - cursor = None - connection = None - return result - except sqlite3.OperationalError: - LOGGER.warning( - 'TaskGraph database is locked because another process is using ' - 'it, waiting for a bit of time to try again') - raise - except Exception: - LOGGER.exception('Exception on _execute_sqlite: %s', sqlite_command) - raise - finally: - if cursor is not None: - cursor.close() - if connection is not None: - connection.commit() - connection.close() diff --git a/taskgraph/__init__.py b/taskgraph/__init__.py index cc346160..c31c2aea 100644 --- a/taskgraph/__init__.py +++ b/taskgraph/__init__.py @@ -1,8 +1,5 @@ -"""TaskGraph init module.""" - -from .Task import TaskGraph from .Task import Task -from .Task import _TASKGRAPH_DATABASE_FILENAME +from .Task import TaskGraph from .Task import __version__ -__all__ = ['__version__', 'TaskGraph', 'Task', '_TASKGRAPH_DATABASE_FILENAME'] +__all__ = ['__version__', 'Task', 'TaskGraph'] diff --git a/taskgraph/_test_fixtures.py b/taskgraph/_test_fixtures.py new file mode 100644 index 00000000..3a524e27 --- /dev/null +++ b/taskgraph/_test_fixtures.py @@ -0,0 +1,124 @@ +import logging +import pickle +import shutil +import time + + +def _return_value_once(value): + """Return the value passed to it only once.""" + if hasattr(_return_value_once, 'executed'): + raise RuntimeError("this function was called twice") + _return_value_once.executed = True + return value + + +def _noop_function(**kwargs): + """Do nothing except allow kwargs to be passed.""" + pass + + +def _long_running_function(delay): + """Wait for ``delay`` seconds.""" + time.sleep(delay) + + +def _create_two_files_on_disk(value, target_a_path, target_b_path): + """Create two files and write ``value`` and append if possible.""" + with open(target_a_path, 'a') as a_file: + a_file.write(value) + + with open(target_b_path, 'a') as b_file: + b_file.write(value) + + +def _merge_and_append_files(base_a_path, base_b_path, target_path): + """Merge two files and append if possible to new file.""" + with open(target_path, 'a') as target_file: + for base_path in [base_a_path, base_b_path]: + with open(base_path, 'r') as base_file: + target_file.write(base_file.read()) + + +def _create_list_on_disk(value, length, target_path=None): + """Create a numpy array on disk filled with value of ``size``.""" + target_list = [value] * length + pickle.dump(target_list, open(target_path, 'wb')) + + +def _call_it(target, *args): + """Invoke ``target`` with ``args``.""" + target(*args) + + +def _append_val(path, *val): + """Append a ``val`` to file at ``path``.""" + with open(path, 'a') as target_file: + for v in val: + target_file.write(str(v)) + + +_append_val_2 = _append_val + + +def _sum_lists_from_disk(list_a_path, list_b_path, target_path): + """Read two lists, add them and save result.""" + list_a = pickle.load(open(list_a_path, 'rb')) + list_b = pickle.load(open(list_b_path, 'rb')) + target_list = [] + for a, b in zip(list_a, list_b): + target_list.append(a+b) + pickle.dump(target_list, open(target_path, 'wb')) + + +def _div_by_zero(): + """Divide by zero to raise an exception.""" + return 1/0 + + +def _create_file(target_path, content): + """Create a file with contents.""" + with open(target_path, 'w') as target_file: + target_file.write(content) + + +def _create_file_once(target_path, content): + """Create a file on the first call, raise an exception on the second.""" + if hasattr(_create_file_once, 'executed'): + raise RuntimeError("this function was called twice") + _create_file_once.executed = True + with open(target_path, 'w') as target_file: + target_file.write(content) + + +def _copy_file_once(base_path, target_path): + """Copy base to target on the first call, raise exception on second.""" + if hasattr(_copy_file_once, 'executed'): + raise RuntimeError("this function was called twice") + _copy_file_once.executed = True + shutil.copyfile(base_path, target_path) + + +def _copy_two_files_once(base_path, target_a_path, target_b_path): + """Copy base to target a/b on first call, raise exception on second.""" + if hasattr(_copy_two_files_once, 'executed'): + raise RuntimeError("this function was called twice") + _copy_two_files_once.executed = True + shutil.copyfile(base_path, target_a_path) + shutil.copyfile(base_path, target_b_path) + + +def _log_from_another_process(logger_name, log_message): + """Write a log message to a given logger. + + Args: + logger_name (string): The string logger name to which ``log_message`` + will be logged. + log_message (string): The string log message to be logged (at INFO + level) to the logger at ``logger_name``. + + Returns: + ``None`` + + """ + logger = logging.getLogger(logger_name) + logger.info(log_message) diff --git a/tests/test_task.py b/tests/test_task.py index 397946ce..8e75ee77 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -8,138 +8,23 @@ import pickle import re import shutil -import sqlite3 import subprocess +import sys import tempfile -import time import unittest +import parsl import retrying import taskgraph +from taskgraph import _test_fixtures + LOGGER = logging.getLogger(__name__) N_TEARDOWN_RETRIES = 5 MAX_TRY_WAIT_MS = 500 -def _return_value_once(value): - """Return the value passed to it only once.""" - if hasattr(_return_value_once, 'executed'): - raise RuntimeError("this function was called twice") - _return_value_once.executed = True - return value - - -def _noop_function(**kwargs): - """Do nothing except allow kwargs to be passed.""" - pass - - -def _long_running_function(delay): - """Wait for ``delay`` seconds.""" - time.sleep(delay) - - -def _create_two_files_on_disk(value, target_a_path, target_b_path): - """Create two files and write ``value`` and append if possible.""" - with open(target_a_path, 'a') as a_file: - a_file.write(value) - - with open(target_b_path, 'a') as b_file: - b_file.write(value) - - -def _merge_and_append_files(base_a_path, base_b_path, target_path): - """Merge two files and append if possible to new file.""" - with open(target_path, 'a') as target_file: - for base_path in [base_a_path, base_b_path]: - with open(base_path, 'r') as base_file: - target_file.write(base_file.read()) - - -def _create_list_on_disk(value, length, target_path=None): - """Create a numpy array on disk filled with value of ``size``.""" - target_list = [value] * length - pickle.dump(target_list, open(target_path, 'wb')) - - -def _call_it(target, *args): - """Invoke ``target`` with ``args``.""" - target(*args) - - -def _append_val(path, *val): - """Append a ``val`` to file at ``path``.""" - with open(path, 'a') as target_file: - for v in val: - target_file.write(str(v)) - - -def _sum_lists_from_disk(list_a_path, list_b_path, target_path): - """Read two lists, add them and save result.""" - list_a = pickle.load(open(list_a_path, 'rb')) - list_b = pickle.load(open(list_b_path, 'rb')) - target_list = [] - for a, b in zip(list_a, list_b): - target_list.append(a+b) - pickle.dump(target_list, open(target_path, 'wb')) - - -def _div_by_zero(): - """Divide by zero to raise an exception.""" - return 1/0 - - -def _create_file(target_path, content): - """Create a file with contents.""" - with open(target_path, 'w') as target_file: - target_file.write(content) - - -def _create_file_once(target_path, content): - """Create a file on the first call, raise an exception on the second.""" - if hasattr(_create_file_once, 'executed'): - raise RuntimeError("this function was called twice") - _create_file_once.executed = True - with open(target_path, 'w') as target_file: - target_file.write(content) - - -def _copy_file_once(base_path, target_path): - """Copy base to target on the first call, raise exception on second.""" - if hasattr(_copy_file_once, 'executed'): - raise RuntimeError("this function was called twice") - _copy_file_once.executed = True - shutil.copyfile(base_path, target_path) - - -def _copy_two_files_once(base_path, target_a_path, target_b_path): - """Copy base to target a/b on first call, raise exception on second.""" - if hasattr(_copy_two_files_once, 'executed'): - raise RuntimeError("this function was called twice") - _copy_two_files_once.executed = True - shutil.copyfile(base_path, target_a_path) - shutil.copyfile(base_path, target_b_path) - - -def _log_from_another_process(logger_name, log_message): - """Write a log message to a given logger. - - Args: - logger_name (string): The string logger name to which ``log_message`` - will be logged. - log_message (string): The string log message to be logged (at INFO - level) to the logger at ``logger_name``. - - Returns: - ``None`` - - """ - logger = logging.getLogger(logger_name) - logger.info(log_message) - - class TaskGraphTests(unittest.TestCase): """Tests for the taskgraph.""" @@ -154,11 +39,14 @@ def setUp(self): wait_exponential_multiplier=250, wait_exponential_max=MAX_TRY_WAIT_MS) def tearDown(self): """Remove temporary directory.""" + try: + parsl.dfk().cleanup() + except parsl.errors.NoDataFlowKernelError: + pass try: shutil.rmtree(self.workspace_dir) except Exception: - LOGGER.exception('error when tearing down.') - raise + pass def test_version_loaded(self): """TaskGraph: verify we can load the version.""" @@ -179,7 +67,7 @@ def test_single_task(self): value = 5 list_len = 1000 _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -195,13 +83,13 @@ def test_task_hash_source_deleted(self): target_a_path = os.path.join(self.workspace_dir, 'a.txt') task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) task_a = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_a_path, 'test value'), target_path_list=[target_a_path]) task_a.join() target_b_path = os.path.join(self.workspace_dir, 'b.txt') _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_b_path, 'test value'), target_path_list=[target_b_path]) task_graph.close() @@ -214,7 +102,7 @@ def test_task_hash_source_deleted(self): target_c_path = os.path.join(self.workspace_dir, 'c.txt') task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_c_path, 'test value'), target_path_list=[target_c_path]) task_graph.close() @@ -233,13 +121,13 @@ def test_task_rel_vs_absolute(self): target_b_path = os.path.abspath(target_a_path) _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_a_path, 'test value'), target_path_list=[target_a_path], task_name='task a') _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_b_path, 'test value'), target_path_list=[target_b_path], task_name='task b') @@ -262,7 +150,7 @@ def test_timeout_task(self): """TaskGraph: Test timeout functionality.""" task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) _ = task_graph.add_task( - func=_long_running_function, + func=_test_fixtures._long_running_function, args=(5,)) task_graph.close() timedout = not task_graph.join(0.5) @@ -276,7 +164,7 @@ def test_precomputed_task(self): value = 5 list_len = 1000 _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -291,7 +179,7 @@ def test_precomputed_task(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -316,21 +204,21 @@ def test_task_chain(self): value_b = 10 list_len = 10 task_a = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value_a, list_len), kwargs={ 'target_path': target_a_path, }, target_path_list=[target_a_path]) task_b = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value_b, list_len), kwargs={ 'target_path': target_b_path, }, target_path_list=[target_b_path]) sum_task = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, target_b_path), kwargs={ 'target_path': result_path, @@ -343,7 +231,7 @@ def test_task_chain(self): self.assertEqual(result, [value_a+value_b]*list_len) sum_2_task = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, result_path, result_2_path), target_path_list=[result_2_path], dependent_task_list=[task_a, sum_task]) @@ -353,7 +241,7 @@ def test_task_chain(self): self.assertEqual(result2, expected_result) sum_3_task = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, result_path, result_2_path), target_path_list=[result_2_path], dependent_task_list=[task_a, sum_task]) @@ -375,7 +263,7 @@ def test_task_chain_single_thread(self): value_b = 10 list_len = 10 task_a = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value_a, list_len), kwargs={ 'target_path': target_a_path, @@ -383,7 +271,7 @@ def test_task_chain_single_thread(self): target_path_list=[target_a_path], task_name='task a') task_b = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value_b, list_len), kwargs={ 'target_path': target_b_path, @@ -391,7 +279,7 @@ def test_task_chain_single_thread(self): target_path_list=[target_b_path], task_name='task b') sum_task = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, target_b_path), kwargs={ 'target_path': result_path, @@ -405,7 +293,7 @@ def test_task_chain_single_thread(self): self.assertEqual(result, [value_a+value_b]*list_len) sum_2_task = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, result_path, result_2_path), target_path_list=[result_2_path], dependent_task_list=[task_a, sum_task], @@ -416,7 +304,7 @@ def test_task_chain_single_thread(self): self.assertEqual(result2, expected_result) sum_3_task = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, result_path, result_2_path), target_path_list=[result_2_path], dependent_task_list=[task_a, sum_task], @@ -429,18 +317,6 @@ def test_task_chain_single_thread(self): task_graph = None self.assertEqual(result3, expected_result) - # we should have 4 completed values in the database, 5 total but one - # was a duplicate - database_path = os.path.join( - self.workspace_dir, taskgraph._TASKGRAPH_DATABASE_FILENAME) - conn = sqlite3.connect(database_path) - with conn: - cursor = conn.cursor() - cursor.execute("SELECT * FROM taskgraph_data") - result = cursor.fetchall() - conn.close() - self.assertEqual(len(result), 4) - def test_task_broken_chain(self): """TaskGraph: Test a multiprocess chain with exception raised.""" task_graph = taskgraph.TaskGraph(self.workspace_dir, 4) @@ -450,17 +326,17 @@ def test_task_broken_chain(self): value_a = 5 list_len = 10 task_a = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value_a, list_len), kwargs={ 'target_path': target_a_path, }, target_path_list=[target_a_path]) task_b = task_graph.add_task( - func=_div_by_zero, + func=_test_fixtures._div_by_zero, dependent_task_list=[task_a]) _ = task_graph.add_task( - func=_sum_lists_from_disk, + func=_test_fixtures._sum_lists_from_disk, args=(target_a_path, target_b_path), kwargs={ 'target_path': result_path, @@ -477,7 +353,7 @@ def test_broken_task(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 1) broken_task = task_graph.add_task( - func=_div_by_zero, task_name='test_broken_task') + func=_test_fixtures._div_by_zero, task_name='test_broken_task') with self.assertRaises(ZeroDivisionError): _ = broken_task.join() @@ -489,33 +365,29 @@ def test_broken_task(self): def test_broken_task_chain(self): """TaskGraph: test dependent tasks fail on ancestor fail.""" task_graph = taskgraph.TaskGraph(self.workspace_dir, 4) - - target_path = os.path.join(self.workspace_dir, '1000.dat') value = 5 list_len = 1000 - for task_id in range(1): - target_path = os.path.join( - self.workspace_dir, '1000_%d.dat' % task_id) - normal_task = task_graph.add_task( - func=_create_list_on_disk, - args=(value, list_len), - kwargs={'target_path': target_path}, - target_path_list=[target_path], - task_name='create list on disk %d' % task_id) - zero_div_task = task_graph.add_task( - func=_div_by_zero, - dependent_task_list=[normal_task], - task_name='test_broken_task_chain_%d' % task_id) - target_path = os.path.join( - self.workspace_dir, 'after_zerodiv_1000_%d.dat' % task_id) - _ = task_graph.add_task( - func=_create_list_on_disk, - args=(value, list_len), - kwargs={'target_path': target_path}, - dependent_task_list=[zero_div_task], - target_path_list=[target_path], - task_name='create list on disk after zero div%d' % task_id) - + target_path = os.path.join( + self.workspace_dir, '1000.dat') + normal_task = task_graph.add_task( + func=_test_fixtures._create_list_on_disk, + args=(value, list_len), + kwargs={'target_path': target_path}, + target_path_list=[target_path], + task_name='create list on disk') + zero_div_task = task_graph.add_task( + func=_test_fixtures._div_by_zero, + dependent_task_list=[normal_task], + task_name='test_broken_task_chain') + target_path = os.path.join( + self.workspace_dir, 'after_zerodiv_1000.dat') + _ = task_graph.add_task( + func=_test_fixtures._create_list_on_disk, + args=(value, list_len), + kwargs={'target_path': target_path}, + dependent_task_list=[zero_div_task], + target_path_list=[target_path], + task_name='create list on disk after zero div') task_graph.close() with self.assertRaises(ZeroDivisionError): task_graph.join() @@ -526,17 +398,6 @@ def test_empty_task(self): _ = task_graph.add_task() task_graph.close() task_graph.join() - # we shouldn't have anything in the database - database_path = os.path.join( - self.workspace_dir, taskgraph._TASKGRAPH_DATABASE_FILENAME) - - conn = sqlite3.connect(database_path) - with conn: - cursor = conn.cursor() - cursor.executescript("SELECT * FROM taskgraph_data") - result = cursor.fetchall() - conn.close() - self.assertEqual(len(result), 0) def test_closed_graph(self): """TaskGraph: Test adding to an closed task graph fails.""" @@ -547,7 +408,7 @@ def test_closed_graph(self): list_len = 1000 with self.assertRaises(ValueError): _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={'target_path': target_path}, target_path_list=[target_path]) @@ -560,7 +421,7 @@ def test_single_task_multiprocessing(self): value = 5 list_len = 1000 _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -598,13 +459,13 @@ def test_get_file_stats(self): self.assertEqual(result, []) def test_transient_runs(self): - """TaskGraph: ensure that transent tasks reexecute.""" + """TaskGraph: ensure that transient tasks reexecute.""" task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) target_path = os.path.join(self.workspace_dir, '1000.dat') value = 5 list_len = 1000 _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -617,7 +478,7 @@ def test_transient_runs(self): task_graph2 = taskgraph.TaskGraph(self.workspace_dir, -1) _ = task_graph2.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), transient_run=True, kwargs={ @@ -638,7 +499,7 @@ def test_repeat_targeted_runs(self): value = 5 list_len = 1000 _ = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -650,7 +511,7 @@ def test_repeat_targeted_runs(self): task_graph2 = taskgraph.TaskGraph(self.workspace_dir, -1) task = task_graph2.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={ 'target_path': target_path, @@ -667,17 +528,17 @@ def test_task_equality(self): value = 5 list_len = 1000 task_a = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={'target_path': target_path}, target_path_list=[target_path]) task_a_same = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value, list_len), kwargs={'target_path': target_path}, target_path_list=[target_path]) task_b = task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=(value+1, list_len), kwargs={'target_path': target_path}, target_path_list=[target_path]) @@ -691,7 +552,7 @@ def test_async_logging(self): task_graph = taskgraph.TaskGraph( self.workspace_dir, 0, reporting_interval=0.5) _ = task_graph.add_task( - func=_long_running_function, + func=_test_fixtures._long_running_function, args=(1.0,)) task_graph.close() task_graph.join() @@ -705,18 +566,18 @@ def test_scrub(self): target_path = os.path.join(self.workspace_dir, 'a.txt') first_task = task_graph.add_task( - func=_append_val, + func=_test_fixtures._append_val, args=(target_path, 1, [1], {'x': 1}), task_name='first append') second_task = task_graph.add_task( - func=_append_val, + func=_test_fixtures._append_val, args=(target_path, 1, [1], {'x': 2}), dependent_task_list=[first_task], task_name='second append') _ = task_graph.add_task( - func=_append_val, + func=_test_fixtures._append_val, args=(target_path, 1, [2], {'x': 1}), dependent_task_list=[second_task], task_name='third append') @@ -735,12 +596,12 @@ def test_target_path_order(self): target_b_path = os.path.join(self.workspace_dir, 'b.txt') task_graph.add_task( - func=_create_two_files_on_disk, + func=_test_fixtures._create_two_files_on_disk, args=("word", target_a_path, target_b_path), target_path_list=[target_a_path, target_b_path]) task_graph.add_task( - func=_create_two_files_on_disk, + func=_test_fixtures._create_two_files_on_disk, args=("word", target_a_path, target_b_path), target_path_list=[target_b_path, target_a_path]) @@ -763,13 +624,13 @@ def test_task_hash_when_ready(self): target_b_path = os.path.join(self.workspace_dir, 'b.txt') create_files_task = task_graph.add_task( - func=_create_two_files_on_disk, + func=_test_fixtures._create_two_files_on_disk, args=("word", target_a_path, target_b_path), target_path_list=[target_a_path, target_b_path]) target_merged_path = os.path.join(self.workspace_dir, 'merged.txt') task_graph.add_task( - func=_merge_and_append_files, + func=_test_fixtures._merge_and_append_files, args=(target_a_path, target_b_path, target_merged_path), target_path_list=[target_merged_path], dependent_task_list=[create_files_task]) @@ -778,7 +639,7 @@ def test_task_hash_when_ready(self): # this second task shouldn't execute because it's a copy of the first task_graph.add_task( - func=_merge_and_append_files, + func=_test_fixtures._merge_and_append_files, args=(target_a_path, target_b_path, target_merged_path), target_path_list=[target_merged_path], dependent_task_list=[create_files_task]) @@ -795,50 +656,31 @@ def test_multiprocessed_logging(self): """TaskGraph: ensure tasks can log from multiple processes.""" logger_name = 'test.task.queuelogger' log_message = 'This is coming from another process' - logger = logging.getLogger(logger_name) - logger.setLevel(logging.DEBUG) - file_log_path = os.path.join( - self.workspace_dir, 'test_multiprocessed_logging.log') - file_handler = logging.FileHandler(file_log_path) - file_handler.setFormatter( - logging.Formatter(fmt=':%(processName)s:%(message)s:')) - logger.addHandler(file_handler) task_graph = taskgraph.TaskGraph(self.workspace_dir, 1) log_task = task_graph.add_task( - func=_log_from_another_process, + func=_test_fixtures._log_from_another_process, args=(logger_name, log_message)) log_task.join() - file_handler.flush() task_graph.close() task_graph.join() - file_handler.close() - - @retrying.retry(wait_exponential_multiplier=100, - wait_exponential_max=1000, - stop_max_attempt_number=5) - def get_name_and_message(): - with open(file_log_path, 'r') as log_file: - message = log_file.read().rstrip() - print(message) - process_name, logged_message = re.match( - ':([^:]*):([^:]*):', message).groups() - return process_name, logged_message - process_name, logged_message = get_name_and_message() - self.assertEqual(logged_message, log_message) - self.assertNotEqual( - process_name, multiprocessing.current_process().name) + file_log_path = list(pathlib.Path(self.workspace_dir).glob( + '000/htex/block-0/*/worker_0.log'))[0] + with open(file_log_path, 'r') as log_file: + message = log_file.read().rstrip() + print(message) + self.assertIn( + '_log_from_another_process INFO: This is coming from another process', + message) def test_repeated_function(self): """TaskGraph: ensure no reruns if argument is a function.""" - global _append_val - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) target_path = os.path.join(self.workspace_dir, 'testfile.txt') task_graph.add_task( - func=_call_it, - args=(_append_val, target_path, 1), + func=_test_fixtures._call_it, + args=(_test_fixtures._append_val, target_path, 1), target_path_list=[target_path], ignore_path_list=[target_path], task_name='first _call_it') @@ -846,18 +688,14 @@ def test_repeated_function(self): task_graph.join() del task_graph - # this causes the address to change - def _append_val(path, *val): - """Append a ``val`` to file at ``path``.""" - with open(path, 'a') as target_file: - for v in val: - target_file.write(str(v)) + # change the address of the function + _test_fixtures._append_val = _test_fixtures._append_val_2 task_graph = taskgraph.TaskGraph(self.workspace_dir, 1) target_path = os.path.join(self.workspace_dir, 'testfile.txt') task_graph.add_task( - func=_call_it, - args=(_append_val, target_path, 1), + func=_test_fixtures._call_it, + args=(_test_fixtures._append_val, target_path, 1), target_path_list=[target_path], ignore_path_list=[target_path], task_name='second _call_it') @@ -872,16 +710,13 @@ def _append_val(path, *val): def test_unix_path_repeated_function(self): """TaskGraph: ensure no reruns if path is unix style.""" - global _append_val - _append_val = _append_val # flake8 complains if not defined - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) target_dir = self.workspace_dir + '/foo/bar/rad/' os.makedirs(target_dir) target_path = target_dir + '/testfile.txt' task_graph.add_task( - func=_call_it, - args=(_append_val, target_path, 1), + func=_test_fixtures._call_it, + args=(_test_fixtures._append_val, target_path, 1), target_path_list=[target_path], task_name='first _call_it') task_graph.close() @@ -890,8 +725,8 @@ def test_unix_path_repeated_function(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) task_graph.add_task( - func=_call_it, - args=(_append_val, target_path, 1), + func=_test_fixtures._call_it, + args=(_test_fixtures._append_val, target_path, 1), target_path_list=[target_path], task_name='second _call_it') task_graph.close() @@ -920,11 +755,11 @@ def test_duplicate_call_changed_target(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) target_path = os.path.join(self.workspace_dir, 'testfile.txt') - if hasattr(_create_file_once, 'executed'): - del _create_file_once.executed + if hasattr(_test_fixtures._create_file_once, 'executed'): + del _test_fixtures._create_file_once.executed task_graph.add_task( - func=_create_file_once, + func=_test_fixtures._create_file_once, args=(target_path, 'test'), target_path_list=[target_path], hash_target_files=False, @@ -939,7 +774,7 @@ def test_duplicate_call_changed_target(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) task_graph.add_task( - func=_create_file_once, + func=_test_fixtures._create_file_once, args=(target_path, 'test'), target_path_list=[target_path], hash_target_files=False, @@ -958,7 +793,7 @@ def test_duplicate_call_modify_timestamp(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) target_path = os.path.join(self.workspace_dir, 'testfile.txt') task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_path, 'test'), target_path_list=[target_path], task_name='first _create_file') @@ -974,7 +809,7 @@ def test_duplicate_call_modify_timestamp(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_path, 'test'), target_path_list=[target_path], task_name='second _create_file') @@ -991,7 +826,7 @@ def test_different_target_path_list(self): task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) target_path = os.path.join(self.workspace_dir, 'testfile.txt') task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=('test', 1, target_path), target_path_list=[target_path], task_name='first _create_list_on_disk') @@ -999,7 +834,7 @@ def test_different_target_path_list(self): with self.assertRaises(RuntimeError): # make the same call but with different target path list task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=('test', 1, target_path), target_path_list=[target_path, 'test.txt'], task_name='first _create_list_on_disk') @@ -1010,12 +845,12 @@ def test_different_target_path_list(self): def test_terminated_taskgraph(self): """TaskGraph: terminated task graph raises exception correctly.""" task_graph = taskgraph.TaskGraph(self.workspace_dir, 4) - _ = task_graph.add_task(func=_div_by_zero) + _ = task_graph.add_task(func=_test_fixtures._div_by_zero) with self.assertRaises(ZeroDivisionError): task_graph.join() with self.assertRaises(RuntimeError) as cm: - _ = task_graph.add_task(func=_div_by_zero) + _ = task_graph.add_task(func=_test_fixtures._div_by_zero) expected_message = "add_task when Taskgraph is terminated" actual_message = str(cm.exception) self.assertTrue(expected_message in actual_message, actual_message) @@ -1030,7 +865,7 @@ def test_type_list_error(self): target_path = os.path.join(self.workspace_dir, 'testfile.txt') with self.assertRaises(ValueError) as cm: task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=('test', 1, target_path), target_path_list=[target_path], dependent_task_list=[target_path], @@ -1046,7 +881,7 @@ def test_target_list_error(self): target_path = os.path.join(self.workspace_dir, 'testfile.txt') with self.assertRaises(ValueError) as cm: task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=('test', 1, target_path), target_path_list=[1], task_name='_create_list_on_disk') @@ -1062,21 +897,18 @@ def test_target_path_missing_file(self): not_target_path = os.path.join(self.workspace_dir, 'not_target.txt') with self.assertRaises(RuntimeError) as cm: task_graph.add_task( - func=_create_list_on_disk, + func=_test_fixtures._create_list_on_disk, args=('test', 1, target_path), target_path_list=[not_target_path], task_name='_create_list_on_disk') + task_graph.join() expected_message = "Missing expected target path results" actual_message = str(cm.exception) self.assertTrue(expected_message in actual_message, actual_message) def test_expected_path_list(self): """TaskGraph: test expected path list matches actual path list.""" - def _create_file(target_path, content): - with open(target_path, 'w') as target_file: - target_file.write(content) - - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) # note it is important this is a relative path that does not # contain the drive letter on Windows. absolute_target_file_path = os.path.join( @@ -1085,7 +917,7 @@ def _create_file(target_path, content): start=self.workspace_dir) _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(relative_path, 'test value'), target_path_list=[relative_path], task_name='create file') @@ -1098,31 +930,30 @@ def _create_file(target_path, content): def test_kwargs_hashed(self): """TaskGraph: ensure kwargs are considered in determining id hash.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) - task_a = task_graph.add_task( - func=_noop_function, + task_graph.add_task( + func=_test_fixtures._noop_function, kwargs={ 'content': ['this value: a']}, task_name='noop a') - task_b = task_graph.add_task( - func=_noop_function, + task_graph.add_task( + func=_test_fixtures._noop_function, kwargs={ 'content': ['this value b']}, task_name='noop b') task_graph.close() task_graph.join() - del task_graph - self.assertNotEqual( - task_a._task_id_hash, task_b._task_id_hash, - "task ids should be different since the kwargs are different") + # taskgraph should have registered two different tasks + # because the kwarg values are different + self.assertEqual(len(task_graph._task_hash_map), 2) def test_same_timestamp_and_value(self): """TaskGraph: ensure identical files but filename are noticed.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) file_a_path = os.path.join(self.workspace_dir, 'file_a.txt') file_b_path = os.path.join(self.workspace_dir, 'file_b.txt') @@ -1135,32 +966,30 @@ def test_same_timestamp_and_value(self): os.utime(file_a_path, (0, 0)) os.utime(file_b_path, (0, 0)) - task_a = task_graph.add_task( - func=_noop_function, + task_graph.add_task( + func=_test_fixtures._noop_function, kwargs={ 'path': file_a_path}, task_name='noop a') - task_b = task_graph.add_task( - func=_noop_function, + task_graph.add_task( + func=_test_fixtures._noop_function, kwargs={ 'path': file_b_path}, task_name='noop b') task_graph.close() task_graph.join() - del task_graph - self.assertNotEqual( - task_a._task_id_hash, task_b._task_id_hash, - "task ids should be different since the filenames are different") + # taskgraph should have registered two different tasks + self.assertEqual(len(task_graph._task_hash_map), 2) def test_different_hash_different_file(self): """TaskGraph: ensure identical files but filename are noticed.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) target_file_path = os.path.join(self.workspace_dir, 'target.txt') _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_file_path, 'content'), hash_algorithm='exists', target_path_list=[target_file_path], @@ -1174,9 +1003,9 @@ def test_different_hash_different_file(self): with open(target_file_path, 'w') as target_file: target_file.write('overwritten') - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_file_path, 'content'), hash_algorithm='exists', target_path_list=[target_file_path], @@ -1188,9 +1017,9 @@ def test_different_hash_different_file(self): with open(target_file_path, 'r') as target_file: self.assertEqual(target_file.read(), 'overwritten') - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_file_path, 'content'), hash_algorithm='md5', target_path_list=[target_file_path], @@ -1202,34 +1031,19 @@ def test_different_hash_different_file(self): with open(target_file_path, 'r') as target_file: self.assertEqual(target_file.read(), 'content') - def test_return_value_no_record(self): - """TaskGraph: test ``get`` raises exception if not set to record.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) - value_task = task_graph.add_task( - func=_noop_function, - store_result=False) - - # get wil raise a ValueError because store_result is not True - with self.assertRaises(ValueError) as cm: - _ = value_task.get() - expected_message = 'must set `store_result` to True in `add_task`' - actual_message = str(cm.exception) - self.assertTrue(expected_message in actual_message, actual_message) - def test_return_value(self): """TaskGraph: test that ``.get`` behavior works as expected.""" - if hasattr(_return_value_once, 'executed'): - del _return_value_once.executed + if hasattr(_test_fixtures._return_value_once, 'executed'): + del _test_fixtures._return_value_once.executed n_iterations = 3 - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) for iteration_id in range(n_iterations): transient_run = iteration_id == n_iterations-1 LOGGER.debug(iteration_id) expected_value = 'a good value' value_task = task_graph.add_task( - func=_return_value_once, + func=_test_fixtures._return_value_once, transient_run=transient_run, - store_result=True, args=(expected_value,), task_name=f'{expected_value} iter {iteration_id}') value = value_task.get() @@ -1239,16 +1053,15 @@ def test_return_value(self): task_graph = None # reset run - del _return_value_once.executed + del _test_fixtures._return_value_once.executed for iteration_id in range(n_iterations): LOGGER.debug(iteration_id) - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0, 0) + task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) expected_value = 'transient run' if iteration_id == 0: value_task = task_graph.add_task( - func=_return_value_once, + func=_test_fixtures._return_value_once, transient_run=True, - store_result=True, args=(expected_value,), task_name='first re-run transient') value = value_task.get() @@ -1258,9 +1071,8 @@ def test_return_value(self): else: with self.assertRaises(RuntimeError): value_task = task_graph.add_task( - func=_return_value_once, + func=_test_fixtures._return_value_once, transient_run=True, - store_result=True, args=(expected_value,), task_name=f'expected error {iteration_id}') @@ -1271,86 +1083,6 @@ def test_return_value(self): task_graph = None - def test_malformed_taskgraph_database(self): - """TaskGraph: Test an empty task.""" - db_schema_test_list = [ - ''' - CREATE TABLE taskgraph_data ( - bad_name_1 TEXT NOT NULL, - bad_name_2 BLOB NOT NULL, - bad_name_3 BLOB NOT NULL); - ''', - ''' - CREATE TABLE taskgraph_data ( - task_reexecution_hash TEXT NOT NULL, - target_path_stats BLOB NOT NULL); - ''', - ''' - CREATE TABLE bad_table_name ( - task_reexecution_hash TEXT NOT NULL, - target_path_stats BLOB NOT NULL, - result BLOB NOT NULL, - PRIMARY KEY (task_reexecution_hash)); - ''' - ] - - for db_schema in db_schema_test_list: - database_path = os.path.join( - self.workspace_dir, taskgraph._TASKGRAPH_DATABASE_FILENAME) - if os.path.exists(database_path): - os.remove(database_path) - connection = sqlite3.connect(database_path) - cursor = connection.cursor() - cursor.executescript(db_schema) - cursor.close() - connection.commit() - connection.close() - - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) - _ = task_graph.add_task() - task_graph.close() - task_graph.join() - del task_graph - - expected_column_name_list = [ - 'task_reexecution_hash', 'target_path_stats', 'result'] - connection = sqlite3.connect(database_path) - cursor = connection.cursor() - cursor.execute('PRAGMA table_info(taskgraph_data)') - result = list(cursor.fetchall()) - cursor.close() - connection.commit() - connection.close() - for header_line in result: - column_name = header_line[1] - if column_name not in expected_column_name_list: - raise ValueError( - f'unexpected column name {column_name} in ' - 'taskgraph_data ') - self.assertEqual(len(result), len(expected_column_name_list)) - - def test_terminate_log(self): - """TaskGraph: test that the logger thread terminates on .join.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, 1, 5.0) - _ = task_graph.add_task() - task_graph.join() - - # logger should not terminate until after join, give it enough time - # to have a chance to close, but not so long the test hangs - task_graph._logging_monitor_thread.join(0.1) - self.assertTrue(task_graph._logging_monitor_thread.is_alive()) - task_graph._execution_monitor_thread.join(0.1) - self.assertTrue(task_graph._execution_monitor_thread.is_alive()) - - task_graph.close() - task_graph.join() - - # 5 seconds should be way too much time to expect the thread to join - task_graph._logging_monitor_thread.join(5) - self.assertFalse(task_graph._logging_monitor_thread.is_alive()) - task_graph._execution_monitor_thread.join(5) - self.assertFalse(task_graph._execution_monitor_thread.is_alive()) - def test_dictionary_arguments(self): """TaskGraph: test that large dictionary arguments behave well.""" task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) @@ -1437,7 +1169,7 @@ def test_duplicate_task_hang_on_exit(self): content = 'test' for _ in range(10): _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_path, content), target_path_list=[target_path], task_name='create content') @@ -1470,7 +1202,7 @@ def test_mtime_mismatch(self): for content in ('test text', 'TEST TEXT', 'test text'): task_graph = taskgraph.TaskGraph(self.workspace_dir, n_workers=-1) _ = task_graph.add_task( - func=_create_file, + func=_test_fixtures._create_file, args=(target_path, content), target_path_list=[target_path], task_name='create content') diff --git a/tox.ini b/tox.ini index 0f242c4e..5885b580 100644 --- a/tox.ini +++ b/tox.ini @@ -14,7 +14,7 @@ python = [testenv] commands = - pytest --log-level=DEBUG \ + pytest -s --log-level=DEBUG \ --cov=taskgraph \ --cov-report=term \ --cov-report=xml \ @@ -22,6 +22,8 @@ commands = --junitxml={toxinidir}/testresults.xml {toxinidir}/tests changedir= {envtmpdir} +setenv = + USERNAME = test # If tox-conda is installed (https://github.com/tox-dev/tox-conda), # use conda-forge python builds for the environments.