diff --git a/taskgraph/Task.py b/taskgraph/Task.py index c13a19a..e0fc4d8 100644 --- a/taskgraph/Task.py +++ b/taskgraph/Task.py @@ -19,6 +19,28 @@ import retrying +import parsl +import os +from parsl.app.app import python_app +from parsl.concurrent import ParslPoolExecutor +from parsl.configs.local_threads import config +from parsl.utils import get_all_checkpoints +from parsl.dataflow.memoization import BasicMemoizer, id_for_memo +from parsl.dataflow.futures import AppFuture +from parsl.executors import HighThroughputExecutor +from parsl.monitoring.monitoring import MonitoringHub +from parsl.addresses import address_by_hostname + + +LOGGER = logging.getLogger(__name__) + + +# @id_for_memo.register(object) +# def id_for_memo_pickle(obj, output_ref=False): +# print('hashing', obj) +# return pickle.dumps(obj) + + try: __version__ = version('taskgraph') except PackageNotFoundError: @@ -47,6 +69,11 @@ _MAX_TIMEOUT = 5.0 # amount of time to wait for threads to terminate +@python_app(cache=True) +def wrap_func(func, *args, inputs=[], **kwargs): + func(*args, **kwargs) + + # 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 @@ -137,105 +164,6 @@ def _logging_queue_monitor(logging_queue): LOGGER.debug('_logging_queue_monitor shutting down') -def _create_taskgraph_table_schema(taskgraph_database_path): - """Create database exists and/or ensures it is compatible and recreate. - - Args: - taskgraph_database_path (str): path to an existing database or desired - location of a new database. - - Returns: - None. - - """ - sql_create_projects_table_script = ( - """ - 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): - 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__,)) - class TaskGraph(object): """Encapsulates the worker and tasks states for parallel processing.""" @@ -290,10 +218,6 @@ def __init__( # 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 @@ -309,9 +233,6 @@ def __init__( # 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 @@ -331,25 +252,33 @@ def __init__( # 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__}') + print('loading config') + config.app_cache = True + config.memoizer = BasicMemoizer( + checkpoint_mode='task_exit', + checkpoint_files=get_all_checkpoints( + self._taskgraph_cache_dir_path)) + config.run_dir = taskgraph_cache_dir_path + config.initialize_logging = False + config.monitoring = MonitoringHub( + hub_address='127.0.0.1', + monitoring_debug=True, + resource_monitoring_interval=10 + ) + print('create pool') + self.pool = ParslPoolExecutor( + config=config, + executors=[ + HighThroughputExecutor( + label="local_htex", + cores_per_worker=1, + max_workers_per_node=4, + address='127.0.0.1' + ) + ] + ) + print('config loaded') + print('pool', self.pool, type(self.pool)) # no need to set up schedulers if n_workers is single threaded self._n_workers = n_workers @@ -369,24 +298,9 @@ 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,)) @@ -403,111 +317,13 @@ def __init__( LOGGER.warning( "NoSuchProcess exception encountered when trying " "to nice %s. This might be a bug in `psutil` so " - "it should be okay to ignore.") + "it should be okay to ignore.")\ def __del__(self): """Ensure all threads have been joined for cleanup.""" + print('__del__') 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") - def add_task( self, func=None, args=None, kwargs=None, task_name=None, target_path_list=None, ignore_path_list=None, @@ -630,9 +446,9 @@ def add_task( new_task = Task( task_name, func, args, kwargs, target_path_list, ignore_path_list, hash_target_files, ignore_directories, - transient_run, self._worker_pool, + transient_run, priority, hash_algorithm, store_result, - self._task_database_path) + dependent_task_list) self._task_name_map[new_task.task_name] = new_task # it may be this task was already created in an earlier call, @@ -714,6 +530,15 @@ def add_task( self._terminate() raise + + except Exception: + # something went wrong, shut down the taskgraph + LOGGER.exception( + "Something went wrong when adding task %s, " + "terminating taskgraph.", task_name) + self._terminate() + raise + def _execution_monitor(self, monitor_wait_event): """Log state of taskgraph every ``self._reporting_interval`` seconds. @@ -770,33 +595,26 @@ def join(self, timeout=None): True if successful join, False if timed out. """ + print('joining taskgraph') 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: + print('computing task', task) + task.join(timeout=timeout) + except 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 @@ -811,6 +629,7 @@ def join(self, timeout=None): def close(self): """Prevent future tasks from being added to the work queue.""" + print('closing task graph') LOGGER.debug("Closing taskgraph.") if self._closed: return @@ -822,10 +641,14 @@ def close(self): def _terminate(self): """Immediately terminate remaining task graph computation.""" + print('terminating task graph') LOGGER.debug( "Invoking terminate. already terminated? %s", self._terminated) if self._terminated: return + print('shut down pool') + print(self.pool) + 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 @@ -833,25 +656,11 @@ def _terminate(self): 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: LOGGER.exception( @@ -864,8 +673,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, + store_result, dependent_task_list): """Make a Task. Args: @@ -947,8 +756,6 @@ def __init__( 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 self._hash_algorithm = hash_algorithm self._store_result = store_result self.exception_object = None @@ -1033,6 +840,10 @@ def __init__( # this will get calculated when ``is_precalculated`` is invoked. self._task_reexecution_hash = None + self.future = wrap_func( + func, *args, **kwargs, + inputs=[task.future for task in dependent_task_list]) + def __eq__(self, other): """Two tasks are equal if their hashes are equal.""" return ( @@ -1067,7 +878,7 @@ def __repr__(self): "self._result": self._result, }) - def _call(self): + def _call(self, timeout=None): """Invoke this method to execute task. Precondition is that the Task dependencies are satisfied. @@ -1080,24 +891,7 @@ def _call(self): 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 + self.future.result(timeout=timeout) # check that the target paths exist and record stats for later if not self._hash_target_files: @@ -1117,161 +911,15 @@ def _call(self): 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) + self._call(timeout=timeout) successful_wait = self.task_done_executing_event.wait(timeout) if self.exception_object: raise self.exception_object @@ -1307,6 +955,85 @@ def get(self, timeout=None): return self._result +@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() + def _get_file_stats( base_value, hash_algorithm, ignore_list, ignore_directories): @@ -1530,82 +1257,3 @@ def _normalize_path(path): 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 cc34616..ebb8600 100644 --- a/taskgraph/__init__.py +++ b/taskgraph/__init__.py @@ -1,8 +1,7 @@ """TaskGraph init module.""" from .Task import TaskGraph -from .Task import Task from .Task import _TASKGRAPH_DATABASE_FILENAME from .Task import __version__ -__all__ = ['__version__', 'TaskGraph', 'Task', '_TASKGRAPH_DATABASE_FILENAME'] +__all__ = ['__version__', 'TaskGraph', '_TASKGRAPH_DATABASE_FILENAME'] diff --git a/tests/test_task.py b/tests/test_task.py index 397946c..175225d 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -14,6 +14,7 @@ import time import unittest +import parsl import retrying import taskgraph @@ -193,18 +194,23 @@ def test_single_task(self): def test_task_hash_source_deleted(self): """TaskGraph: test if old target deleted when hashing duplicate.""" target_a_path = os.path.join(self.workspace_dir, 'a.txt') + print('creating task graph...') task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) + print('adding task...') task_a = task_graph.add_task( func=_create_file, args=(target_a_path, 'test value'), target_path_list=[target_a_path]) + print('joining task...') task_a.join() target_b_path = os.path.join(self.workspace_dir, 'b.txt') _ = task_graph.add_task( func=_create_file, args=(target_b_path, 'test value'), target_path_list=[target_b_path]) + print('close') task_graph.close() + print('join') task_graph.join() del task_graph @@ -429,17 +435,17 @@ 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) + # # 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.""" @@ -517,8 +523,9 @@ def test_broken_task_chain(self): task_name='create list on disk after zero div%d' % task_id) task_graph.close() - with self.assertRaises(ZeroDivisionError): + with self.assertRaises(parsl.dataflow.errors.DependencyError) as err: task_graph.join() + assert isinstance(err.__cause__, ZeroDivisionError) def test_empty_task(self): """TaskGraph: Test an empty task.""" @@ -526,17 +533,17 @@ 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) + # # 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.""" @@ -791,694 +798,694 @@ def test_task_hash_when_ready(self): self.assertEqual(target_string, "wordword") - 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, - 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) - - 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), - target_path_list=[target_path], - ignore_path_list=[target_path], - task_name='first _call_it') - task_graph.close() - 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)) - - 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), - target_path_list=[target_path], - ignore_path_list=[target_path], - task_name='second _call_it') - task_graph.close() - task_graph.join() - - with open(target_path, 'r') as target_file: - result = target_file.read() - - # the second call shouldn't happen - self.assertEqual(result, '1') - - 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), - target_path_list=[target_path], - task_name='first _call_it') - task_graph.close() - task_graph.join() - del task_graph - - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) - task_graph.add_task( - func=_call_it, - args=(_append_val, target_path, 1), - target_path_list=[target_path], - task_name='second _call_it') - task_graph.close() - task_graph.join() - - with open(target_path, 'r') as target_file: - result = target_file.read() - - # the second call shouldn't happen - self.assertEqual(result, '1') - - def test_very_long_string(self): - """TaskGraph: ensure that long strings don't case an OSError.""" - from taskgraph.Task import _get_file_stats - - # this is a list with two super long strings to try to trick some - # os function into thinking it's a path. - base_value = [ - 'c:' + r'\\\\\\\\x\\\\\\\\'*2**10 + 'foo', - 'wfeji3223j8923j9' * 2**10] - self.assertEqual( - list(_get_file_stats(base_value, 'sizetimestamp', [], True)), []) - - def test_duplicate_call_changed_target(self): - """TaskGraph: test that duplicate calls copy target path.""" - 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 - - task_graph.add_task( - func=_create_file_once, - args=(target_path, 'test'), - target_path_list=[target_path], - hash_target_files=False, - task_name='first _create_file_once') - - task_graph.close() - task_graph.join() - del task_graph - - with open(target_path, 'a') as target_file: - target_file.write('updated') - - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) - task_graph.add_task( - func=_create_file_once, - args=(target_path, 'test'), - target_path_list=[target_path], - hash_target_files=False, - task_name='first _create_file_once') - - task_graph.close() - task_graph.join() - del task_graph - - with open(target_path, 'r') as result_file: - result_contents = result_file.read() - self.assertEqual('testupdated', result_contents) - - def test_duplicate_call_modify_timestamp(self): - """TaskGraph: test that duplicate call modified stamp recompute.""" - 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, - args=(target_path, 'test'), - target_path_list=[target_path], - task_name='first _create_file') - task_graph.close() - task_graph.join() - del task_graph - - with open(target_path, 'w') as target_file: - target_file.write('test2') - with open(target_path, 'r') as target_file: - contents = target_file.read() - self.assertEqual(contents, 'test2') - - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) - task_graph.add_task( - func=_create_file, - args=(target_path, 'test'), - target_path_list=[target_path], - task_name='second _create_file') - - task_graph.close() - task_graph.join() - - with open(target_path, 'r') as target_file: - contents = target_file.read() - self.assertEqual(contents, 'test') - - def test_different_target_path_list(self): - """TaskGraph: duplicate calls with different targets should fail.""" - 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, - args=('test', 1, target_path), - target_path_list=[target_path], - task_name='first _create_list_on_disk') - - with self.assertRaises(RuntimeError): - # make the same call but with different target path list - task_graph.add_task( - func=_create_list_on_disk, - args=('test', 1, target_path), - target_path_list=[target_path, 'test.txt'], - task_name='first _create_list_on_disk') - - task_graph.close() - task_graph.join() - - 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) - with self.assertRaises(ZeroDivisionError): - task_graph.join() - - with self.assertRaises(RuntimeError) as cm: - _ = task_graph.add_task(func=_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) - - task_graph.close() - # try closing twice just to mess with coverage - task_graph.close() - - def test_type_list_error(self): - """TaskGraph: Task not passed to dependent task list.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) - 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, - args=('test', 1, target_path), - target_path_list=[target_path], - dependent_task_list=[target_path], - task_name='first _create_list_on_disk') - expected_message = ( - "Objects passed to dependent task list that are not tasks") - actual_message = str(cm.exception) - self.assertTrue(expected_message in actual_message, actual_message) - - def test_target_list_error(self): - """TaskGraph: Path not passed to target list.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) - 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, - args=('test', 1, target_path), - target_path_list=[1], - task_name='_create_list_on_disk') - expected_message = ( - "Values passed to target_path_list are not strings") - actual_message = str(cm.exception) - self.assertTrue(expected_message in actual_message, actual_message) - - def test_target_path_missing_file(self): - """TaskGraph: func runs, but missing target.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) - target_path = os.path.join(self.workspace_dir, 'testfile.txt') - 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, - args=('test', 1, target_path), - target_path_list=[not_target_path], - task_name='_create_list_on_disk') - 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) - # 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( - self.workspace_dir, 'a.txt') - relative_path = os.path.relpath(absolute_target_file_path, - start=self.workspace_dir) - - _ = task_graph.add_task( - func=_create_file, - args=(relative_path, 'test value'), - target_path_list=[relative_path], - task_name='create file') - - task_graph.close() - task_graph.join() - del task_graph - - self.assertTrue('Ran without crashing!') - - def test_kwargs_hashed(self): - """TaskGraph: ensure kwargs are considered in determining id hash.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) - - task_a = task_graph.add_task( - func=_noop_function, - kwargs={ - 'content': ['this value: a']}, - task_name='noop a') - - task_b = task_graph.add_task( - func=_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") - - def test_same_timestamp_and_value(self): - """TaskGraph: ensure identical files but filename are noticed.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) - - file_a_path = os.path.join(self.workspace_dir, 'file_a.txt') - file_b_path = os.path.join(self.workspace_dir, 'file_b.txt') - - with open(file_a_path, 'w') as file_a: - file_a.write('a') - with open(file_b_path, 'w') as file_b: - file_b.write('a') - - os.utime(file_a_path, (0, 0)) - os.utime(file_b_path, (0, 0)) - - task_a = task_graph.add_task( - func=_noop_function, - kwargs={ - 'path': file_a_path}, - task_name='noop a') - - task_b = task_graph.add_task( - func=_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") - - def test_different_hash_different_file(self): - """TaskGraph: ensure identical files but filename are noticed.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) - target_file_path = os.path.join(self.workspace_dir, 'target.txt') - _ = task_graph.add_task( - func=_create_file, - args=(target_file_path, 'content'), - hash_algorithm='exists', - target_path_list=[target_file_path], - task_name='create content, hash with exists') - task_graph.close() - task_graph.join() - del task_graph - - with open(target_file_path, 'r') as target_file: - self.assertEqual(target_file.read(), 'content') - with open(target_file_path, 'w') as target_file: - target_file.write('overwritten') - - task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) - _ = task_graph.add_task( - func=_create_file, - args=(target_file_path, 'content'), - hash_algorithm='exists', - target_path_list=[target_file_path], - task_name='will not overwrite content, hash with exists') - task_graph.close() - task_graph.join() - del task_graph - - 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.add_task( - func=_create_file, - args=(target_file_path, 'content'), - hash_algorithm='md5', - target_path_list=[target_file_path], - task_name='create content again with new hash') - task_graph.close() - task_graph.join() - del task_graph - - 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 - n_iterations = 3 - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0, 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, - transient_run=transient_run, - store_result=True, - args=(expected_value,), - task_name=f'{expected_value} iter {iteration_id}') - value = value_task.get() - self.assertEqual(value, expected_value) - task_graph.close() - task_graph.join() - task_graph = None - - # reset run - del _return_value_once.executed - for iteration_id in range(n_iterations): - LOGGER.debug(iteration_id) - task_graph = taskgraph.TaskGraph(self.workspace_dir, 0, 0) - expected_value = 'transient run' - if iteration_id == 0: - value_task = task_graph.add_task( - func=_return_value_once, - transient_run=True, - store_result=True, - args=(expected_value,), - task_name='first re-run transient') - value = value_task.get() - self.assertEqual(value, expected_value) - task_graph.close() - task_graph.join() - else: - with self.assertRaises(RuntimeError): - value_task = task_graph.add_task( - func=_return_value_once, - transient_run=True, - store_result=True, - args=(expected_value,), - task_name=f'expected error {iteration_id}') - - value = value_task.get() - - with self.assertRaises(RuntimeError): - task_graph.join() - - 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) - dict_arg = {} - x = {None: None} - for _ in range(10000): - dict_arg[_] = x - - def my_op(dict_arg): - pass - task_graph.add_task( - func=my_op, args=(), kwargs={'dict_arg': dict_arg}) - task_graph.join() - self.assertTrue(True, 'no memory error so everything is fine') - - def test_filter_non_files(self): - """TaskGraph: test internal filter non-files function.""" - from taskgraph.Task import _filter_non_files - from taskgraph.Task import _normalize_path - - # Test a passthrough - test_dict = { - 0: {'one': 0, 'two': 1, 'three': 2}, - 1: {'one': 1, 'two': 2, 'three': 3}, - 2: {'one': 2, 'two': 3, 'three': 4}} - self.assertEqual( - test_dict, _filter_non_files(test_dict, [], [], False)) - - # Test combination of files, not existing files, and flags in the - # call - test_file_a_exists = _normalize_path(os.path.join( - self.workspace_dir, 'exists_a.txt')) - pathlib.Path(test_file_a_exists).touch() - test_file_b_exists = _normalize_path(os.path.join( - self.workspace_dir, 'exists_b.txt')) - pathlib.Path(test_file_b_exists).touch() - test_file_not_a_exists = _normalize_path(os.path.join( - self.workspace_dir, 'does_not_exist_a.txt')) - test_file_not_b_exists = _normalize_path(os.path.join( - self.workspace_dir, 'does_not_exist_b.txt')) - - test_dict = { - 0: {'one': 0, 'two': 1, 'three': 2}, - 1: {'one': 1, 'two': 2, 'three': 3}, - 2: {'one': 2, 'two': 3, 'three': 4}, - 4: {'bar': test_file_not_a_exists}, - 5: {'foo': test_file_a_exists}, - 6: test_file_b_exists, - 7: test_file_not_b_exists, - 8: _normalize_path(self.workspace_dir)} - - expected_result_dict = { - 0: {'one': 0, 'two': 1, 'three': 2}, - 1: {'one': 1, 'two': 2, 'three': 3}, - 2: {'one': 2, 'two': 3, 'three': 4}, - 4: {'bar': test_file_not_a_exists}, - 5: {'foo': None}, - 6: test_file_b_exists, - 7: None, - 8: _normalize_path(self.workspace_dir)} - - self.assertEqual( - _filter_non_files( - test_dict, - [test_file_b_exists], - [test_file_not_b_exists], - True), - expected_result_dict) - - # and test same as above but don't keep directories: - expected_result_dict[8] = None - self.assertEqual( - _filter_non_files( - test_dict, - [test_file_b_exists], - [test_file_not_b_exists], - False), - expected_result_dict) - - def test_duplicate_task_hang_on_exit(self): - """TaskGraph: ensure duplicate tasks don't cause taskgraph to hang.""" - task_graph = taskgraph.TaskGraph(self.workspace_dir, 1) - target_path = os.path.join(self.workspace_dir, 'target.txt') - content = 'test' - for _ in range(10): - _ = task_graph.add_task( - func=_create_file, - args=(target_path, content), - target_path_list=[target_path], - task_name='create content') - task_graph.join() - task_graph.close() - - def test_history_rst_format(self): - """TaskGraph: ensure HISTORY.rst is correctly formatted.""" - # ensure there are no errors when checking the history file - history_filepath = os.path.join( - os.path.dirname(__file__), '..', 'HISTORY.rst') - subprocess.check_call(['rstcheck', history_filepath]) - - def test_mtime_mismatch(self): - """TaskGraph: ensure re-run when file mtimes don't match. - - This test addresses the issue described under the github issue - https://github.com/natcap/taskgraph/issues/70. - """ - target_path = os.path.join(self.workspace_dir, 'target.txt') - - # SETUP: When we call 3 similar graphs in rapid succession, the file's - # mtime is not precise enough to detect that the file has actually - # changed. The specific conditions here are: - # * The task "test text" has already been computed once - # * The file written by the first "test text" is replaced by content - # of the same filesize (thus fooling the size part of sizetimestamp) - # * The graphs are executed fast enough that _is_precalculated's mtime - # check via math.isclose() couldn't detect the recalculation. - 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, - args=(target_path, content), - target_path_list=[target_path], - task_name='create content') - task_graph.join() - task_graph.close() - - with open(target_path) as target_file: - self.assertEqual(target_file.read(), content) + # 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, + # 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) + + # 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), + # target_path_list=[target_path], + # ignore_path_list=[target_path], + # task_name='first _call_it') + # task_graph.close() + # 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)) + + # 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), + # target_path_list=[target_path], + # ignore_path_list=[target_path], + # task_name='second _call_it') + # task_graph.close() + # task_graph.join() + + # with open(target_path, 'r') as target_file: + # result = target_file.read() + + # # the second call shouldn't happen + # self.assertEqual(result, '1') + + # 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), + # target_path_list=[target_path], + # task_name='first _call_it') + # task_graph.close() + # task_graph.join() + # del task_graph + + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) + # task_graph.add_task( + # func=_call_it, + # args=(_append_val, target_path, 1), + # target_path_list=[target_path], + # task_name='second _call_it') + # task_graph.close() + # task_graph.join() + + # with open(target_path, 'r') as target_file: + # result = target_file.read() + + # # the second call shouldn't happen + # self.assertEqual(result, '1') + + # def test_very_long_string(self): + # """TaskGraph: ensure that long strings don't case an OSError.""" + # from taskgraph.Task import _get_file_stats + + # # this is a list with two super long strings to try to trick some + # # os function into thinking it's a path. + # base_value = [ + # 'c:' + r'\\\\\\\\x\\\\\\\\'*2**10 + 'foo', + # 'wfeji3223j8923j9' * 2**10] + # self.assertEqual( + # list(_get_file_stats(base_value, 'sizetimestamp', [], True)), []) + + # def test_duplicate_call_changed_target(self): + # """TaskGraph: test that duplicate calls copy target path.""" + # 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 + + # task_graph.add_task( + # func=_create_file_once, + # args=(target_path, 'test'), + # target_path_list=[target_path], + # hash_target_files=False, + # task_name='first _create_file_once') + + # task_graph.close() + # task_graph.join() + # del task_graph + + # with open(target_path, 'a') as target_file: + # target_file.write('updated') + + # task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) + # task_graph.add_task( + # func=_create_file_once, + # args=(target_path, 'test'), + # target_path_list=[target_path], + # hash_target_files=False, + # task_name='first _create_file_once') + + # task_graph.close() + # task_graph.join() + # del task_graph + + # with open(target_path, 'r') as result_file: + # result_contents = result_file.read() + # self.assertEqual('testupdated', result_contents) + + # def test_duplicate_call_modify_timestamp(self): + # """TaskGraph: test that duplicate call modified stamp recompute.""" + # 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, + # args=(target_path, 'test'), + # target_path_list=[target_path], + # task_name='first _create_file') + # task_graph.close() + # task_graph.join() + # del task_graph + + # with open(target_path, 'w') as target_file: + # target_file.write('test2') + # with open(target_path, 'r') as target_file: + # contents = target_file.read() + # self.assertEqual(contents, 'test2') + + # task_graph = taskgraph.TaskGraph(self.workspace_dir, 0) + # task_graph.add_task( + # func=_create_file, + # args=(target_path, 'test'), + # target_path_list=[target_path], + # task_name='second _create_file') + + # task_graph.close() + # task_graph.join() + + # with open(target_path, 'r') as target_file: + # contents = target_file.read() + # self.assertEqual(contents, 'test') + + # def test_different_target_path_list(self): + # """TaskGraph: duplicate calls with different targets should fail.""" + # 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, + # args=('test', 1, target_path), + # target_path_list=[target_path], + # task_name='first _create_list_on_disk') + + # with self.assertRaises(RuntimeError): + # # make the same call but with different target path list + # task_graph.add_task( + # func=_create_list_on_disk, + # args=('test', 1, target_path), + # target_path_list=[target_path, 'test.txt'], + # task_name='first _create_list_on_disk') + + # task_graph.close() + # task_graph.join() + + # 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) + # with self.assertRaises(ZeroDivisionError): + # task_graph.join() + + # with self.assertRaises(RuntimeError) as cm: + # _ = task_graph.add_task(func=_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) + + # task_graph.close() + # # try closing twice just to mess with coverage + # task_graph.close() + + # def test_type_list_error(self): + # """TaskGraph: Task not passed to dependent task list.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) + # 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, + # args=('test', 1, target_path), + # target_path_list=[target_path], + # dependent_task_list=[target_path], + # task_name='first _create_list_on_disk') + # expected_message = ( + # "Objects passed to dependent task list that are not tasks") + # actual_message = str(cm.exception) + # self.assertTrue(expected_message in actual_message, actual_message) + + # def test_target_list_error(self): + # """TaskGraph: Path not passed to target list.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) + # 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, + # args=('test', 1, target_path), + # target_path_list=[1], + # task_name='_create_list_on_disk') + # expected_message = ( + # "Values passed to target_path_list are not strings") + # actual_message = str(cm.exception) + # self.assertTrue(expected_message in actual_message, actual_message) + + # def test_target_path_missing_file(self): + # """TaskGraph: func runs, but missing target.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1) + # target_path = os.path.join(self.workspace_dir, 'testfile.txt') + # 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, + # args=('test', 1, target_path), + # target_path_list=[not_target_path], + # task_name='_create_list_on_disk') + # 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) + # # 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( + # self.workspace_dir, 'a.txt') + # relative_path = os.path.relpath(absolute_target_file_path, + # start=self.workspace_dir) + + # _ = task_graph.add_task( + # func=_create_file, + # args=(relative_path, 'test value'), + # target_path_list=[relative_path], + # task_name='create file') + + # task_graph.close() + # task_graph.join() + # del task_graph + + # self.assertTrue('Ran without crashing!') + + # def test_kwargs_hashed(self): + # """TaskGraph: ensure kwargs are considered in determining id hash.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + + # task_a = task_graph.add_task( + # func=_noop_function, + # kwargs={ + # 'content': ['this value: a']}, + # task_name='noop a') + + # task_b = task_graph.add_task( + # func=_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") + + # def test_same_timestamp_and_value(self): + # """TaskGraph: ensure identical files but filename are noticed.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + + # file_a_path = os.path.join(self.workspace_dir, 'file_a.txt') + # file_b_path = os.path.join(self.workspace_dir, 'file_b.txt') + + # with open(file_a_path, 'w') as file_a: + # file_a.write('a') + # with open(file_b_path, 'w') as file_b: + # file_b.write('a') + + # os.utime(file_a_path, (0, 0)) + # os.utime(file_b_path, (0, 0)) + + # task_a = task_graph.add_task( + # func=_noop_function, + # kwargs={ + # 'path': file_a_path}, + # task_name='noop a') + + # task_b = task_graph.add_task( + # func=_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") + + # def test_different_hash_different_file(self): + # """TaskGraph: ensure identical files but filename are noticed.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + # target_file_path = os.path.join(self.workspace_dir, 'target.txt') + # _ = task_graph.add_task( + # func=_create_file, + # args=(target_file_path, 'content'), + # hash_algorithm='exists', + # target_path_list=[target_file_path], + # task_name='create content, hash with exists') + # task_graph.close() + # task_graph.join() + # del task_graph + + # with open(target_file_path, 'r') as target_file: + # self.assertEqual(target_file.read(), 'content') + # with open(target_file_path, 'w') as target_file: + # target_file.write('overwritten') + + # task_graph = taskgraph.TaskGraph(self.workspace_dir, -1, 0) + # _ = task_graph.add_task( + # func=_create_file, + # args=(target_file_path, 'content'), + # hash_algorithm='exists', + # target_path_list=[target_file_path], + # task_name='will not overwrite content, hash with exists') + # task_graph.close() + # task_graph.join() + # del task_graph + + # 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.add_task( + # func=_create_file, + # args=(target_file_path, 'content'), + # hash_algorithm='md5', + # target_path_list=[target_file_path], + # task_name='create content again with new hash') + # task_graph.close() + # task_graph.join() + # del task_graph + + # 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 + # n_iterations = 3 + # task_graph = taskgraph.TaskGraph(self.workspace_dir, 0, 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, + # transient_run=transient_run, + # store_result=True, + # args=(expected_value,), + # task_name=f'{expected_value} iter {iteration_id}') + # value = value_task.get() + # self.assertEqual(value, expected_value) + # task_graph.close() + # task_graph.join() + # task_graph = None + + # # reset run + # del _return_value_once.executed + # for iteration_id in range(n_iterations): + # LOGGER.debug(iteration_id) + # task_graph = taskgraph.TaskGraph(self.workspace_dir, 0, 0) + # expected_value = 'transient run' + # if iteration_id == 0: + # value_task = task_graph.add_task( + # func=_return_value_once, + # transient_run=True, + # store_result=True, + # args=(expected_value,), + # task_name='first re-run transient') + # value = value_task.get() + # self.assertEqual(value, expected_value) + # task_graph.close() + # task_graph.join() + # else: + # with self.assertRaises(RuntimeError): + # value_task = task_graph.add_task( + # func=_return_value_once, + # transient_run=True, + # store_result=True, + # args=(expected_value,), + # task_name=f'expected error {iteration_id}') + + # value = value_task.get() + + # with self.assertRaises(RuntimeError): + # task_graph.join() + + # 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) + # dict_arg = {} + # x = {None: None} + # for _ in range(10000): + # dict_arg[_] = x + + # def my_op(dict_arg): + # pass + # task_graph.add_task( + # func=my_op, args=(), kwargs={'dict_arg': dict_arg}) + # task_graph.join() + # self.assertTrue(True, 'no memory error so everything is fine') + + # def test_filter_non_files(self): + # """TaskGraph: test internal filter non-files function.""" + # from taskgraph.Task import _filter_non_files + # from taskgraph.Task import _normalize_path + + # # Test a passthrough + # test_dict = { + # 0: {'one': 0, 'two': 1, 'three': 2}, + # 1: {'one': 1, 'two': 2, 'three': 3}, + # 2: {'one': 2, 'two': 3, 'three': 4}} + # self.assertEqual( + # test_dict, _filter_non_files(test_dict, [], [], False)) + + # # Test combination of files, not existing files, and flags in the + # # call + # test_file_a_exists = _normalize_path(os.path.join( + # self.workspace_dir, 'exists_a.txt')) + # pathlib.Path(test_file_a_exists).touch() + # test_file_b_exists = _normalize_path(os.path.join( + # self.workspace_dir, 'exists_b.txt')) + # pathlib.Path(test_file_b_exists).touch() + # test_file_not_a_exists = _normalize_path(os.path.join( + # self.workspace_dir, 'does_not_exist_a.txt')) + # test_file_not_b_exists = _normalize_path(os.path.join( + # self.workspace_dir, 'does_not_exist_b.txt')) + + # test_dict = { + # 0: {'one': 0, 'two': 1, 'three': 2}, + # 1: {'one': 1, 'two': 2, 'three': 3}, + # 2: {'one': 2, 'two': 3, 'three': 4}, + # 4: {'bar': test_file_not_a_exists}, + # 5: {'foo': test_file_a_exists}, + # 6: test_file_b_exists, + # 7: test_file_not_b_exists, + # 8: _normalize_path(self.workspace_dir)} + + # expected_result_dict = { + # 0: {'one': 0, 'two': 1, 'three': 2}, + # 1: {'one': 1, 'two': 2, 'three': 3}, + # 2: {'one': 2, 'two': 3, 'three': 4}, + # 4: {'bar': test_file_not_a_exists}, + # 5: {'foo': None}, + # 6: test_file_b_exists, + # 7: None, + # 8: _normalize_path(self.workspace_dir)} + + # self.assertEqual( + # _filter_non_files( + # test_dict, + # [test_file_b_exists], + # [test_file_not_b_exists], + # True), + # expected_result_dict) + + # # and test same as above but don't keep directories: + # expected_result_dict[8] = None + # self.assertEqual( + # _filter_non_files( + # test_dict, + # [test_file_b_exists], + # [test_file_not_b_exists], + # False), + # expected_result_dict) + + # def test_duplicate_task_hang_on_exit(self): + # """TaskGraph: ensure duplicate tasks don't cause taskgraph to hang.""" + # task_graph = taskgraph.TaskGraph(self.workspace_dir, 1) + # target_path = os.path.join(self.workspace_dir, 'target.txt') + # content = 'test' + # for _ in range(10): + # _ = task_graph.add_task( + # func=_create_file, + # args=(target_path, content), + # target_path_list=[target_path], + # task_name='create content') + # task_graph.join() + # task_graph.close() + + # def test_history_rst_format(self): + # """TaskGraph: ensure HISTORY.rst is correctly formatted.""" + # # ensure there are no errors when checking the history file + # history_filepath = os.path.join( + # os.path.dirname(__file__), '..', 'HISTORY.rst') + # subprocess.check_call(['rstcheck', history_filepath]) + + # def test_mtime_mismatch(self): + # """TaskGraph: ensure re-run when file mtimes don't match. + + # This test addresses the issue described under the github issue + # https://github.com/natcap/taskgraph/issues/70. + # """ + # target_path = os.path.join(self.workspace_dir, 'target.txt') + + # # SETUP: When we call 3 similar graphs in rapid succession, the file's + # # mtime is not precise enough to detect that the file has actually + # # changed. The specific conditions here are: + # # * The task "test text" has already been computed once + # # * The file written by the first "test text" is replaced by content + # # of the same filesize (thus fooling the size part of sizetimestamp) + # # * The graphs are executed fast enough that _is_precalculated's mtime + # # check via math.isclose() couldn't detect the recalculation. + # 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, + # args=(target_path, content), + # target_path_list=[target_path], + # task_name='create content') + # task_graph.join() + # task_graph.close() + + # with open(target_path) as target_file: + # self.assertEqual(target_file.read(), content) def Fail(n_tries, result_path):