6dZddlmZddlmZddlmZddlZddlmZddlm Z ddl m Z d Z d Z d Zd ZGd dZGdde j"ZGddZy)zImplements logic for tracking task dependencies in task_graph_executor. See go/parallel-processing-in-gcloud-storage for more information. )absolute_import)division)unicode_literalsN)List)errors)logz Task Graph:z - Task ID: {} z^ - Task: {} - Dependency Count: {} - Dependent Task IDs: {} - Is Submitted: {} ceZdZdZdZdZy) TaskWrapperaEmbeds a Task instance in a dependency graph. Attributes: id (Hashable): A unique identifier for this task wrapper. task (googlecloudsdk.command_lib.storage.tasks.task.Task): An instance of a task class. dependency_count (int): The number of unexecuted dependencies this task has, i.e. this node's in-degree in a graph where an edge from A to B indicates that A must be executed before B. dependent_task_ids (Optional[Iterable[Hashable]]): The id of the tasks that require this task to be completed for their own completion. This value should be None if no tasks depend on this one. is_submitted (bool): True if this task has been submitted for execution. cJ||_||_d|_||_d|_y)NrF)idtaskdependency_countdependent_task_ids is_submitted)selftask_idrrs :lib/googlecloudsdk/command_lib/storage/tasks/task_graph.py__init__zTaskWrapper.__init__:s(DGDID0DDctj|jtj|jj j |jrt|jnd|j|jzS)z3Returns a string representation of the TaskWrapper.r) TASK_WRAPPER_IDformatr TASK_DETAILSr __class____name__rlenr)rs r__str__zTaskWrapper.__str__Aso tww' II   ( (&& '' (,-  # #       rN)r __module__ __qualname____doc__rrrrr r *s  rr ceZdZdZy)InvalidDependencyErrorzRaised on attempts to create an invalid dependency. Invalid dependencies are self-dependencies and those that involve nodes that do not exist. N)rrr r!r"rrr$r$Osrr$cBeZdZdZdZd dZdZdZdZde e fd Z y) TaskGrapha7Tracks dependencies between Task instances. See googlecloudsdk.command_lib.storage.tasks.task.Task for the definition of the Task class. The public methods in this class are thread safe. Attributes: is_empty (threading.Event): is_empty.is_set() is True when the graph has no tasks in it. ctj|_|jjtj|_i|_tj||_y)aSInitializes a TaskGraph instance. Args: top_level_task_limit (int): A top-level task is a task that no other tasks depend on for completion (i.e. dependent_task_ids is None). Adding top-level tasks with TaskGraph.add will block until there are fewer than this number of top-level tasks in the graph. N) threadingEventis_emptysetRLock_lock_task_wrappers_in_graph Semaphore_top_level_task_semaphore)rtop_level_task_limits rrzTaskGraph.__init__dsROO%DMMM"DJ$&D &/%8%89M%ND"rNct|du}|r|jj|j5|j |j}n t |}||j vr|jNt jjdj|jj|jnBt jjdj|jj|r|jj dddyt|||}|xsgD]%} |j |xjdz c_'||j |j<|j"j%ddd|S#t$rt wxYw#1swYSxYw)a Adds a task to the graph. Args: task (googlecloudsdk.command_lib.storage.tasks.task.Task): The task to be added. dependent_task_ids (Optional[List[Hashable]]): TaskWrapper.id attributes for tasks already in the graph that require the task being added to complete before being executed. This argument should be None for top-level tasks, which no other tasks depend on. Returns: A TaskWrapper instance for the task passed into this function, or None if task.parallel_processing_key was the same as another task's parallel_processing_key. Raises: InvalidDependencyError if any id in dependent_task_ids is not in the graph, or if a the add operation would have created a self-dependency. NzcSkipping {} for {}. This can occur if a cp command results in multiple writes to the same resource.zoSkipping {}. This is probably because due to a bug that caused it to be submitted for execution more than once.)r0acquirer-parallel_processing_keyr r.rstatusPrintrrrreleaser rKeyErrorr$r*clear)rrris_top_level_task identifier task_wrapperrs raddz TaskGraph.add~s(+d2 $$,,.  % % 111 X t33 3  ' ' 3 **  66append) rexecuted_task_wrapper task_outputrrBparent_tasks_for_next_layer task_iteratorr=rrs rupdate_from_executed_taskz#TaskGraph.update_from_executed_tasks;:   !"".#66B,??G#'#?#?#H % % 7 7 > >"" $@  K$I$I}}23 &;$;!$K$I$IJ-0K 0K LOO0K  ')#!D$;MN,  % ' . .| <" K) +, (C ( ) s*BD(D(>D#D(0)D(#D((D1cdtd|jjdt|jg}|jrIt }|j |j|jjt|n|jddj|S)z1Returns a string representation of the TaskGraph.z - Empty: z - Task Wrappers: zNo tasks in the graph to print. ) TASK_GRAPH_HEADERr*is_setrr.r+rF_print_task_wrapper_recursivevaluesINITIAL_INDENT_LEVELrIjoin)routput printed_taskss rrzTaskGraph.__str__/s  T]]))+,- S!=!=>?@F  ##em mm  , ,**113"  mm56 99V rreturnc#8K|D]}|j|vs|j|jt||jsG|jDcgc]}|j|}}|j ||dz|Ed{ycc}w7 w)aRecursively yields task wrappers and their dependencies. Example: Suppose we have task wrappers representing tasks with dependencies: task_wrapper1 = TaskWrapper(id='task1', dependent_task_ids=['task2', 'task3']), task_wrapper2 = TaskWrapper(id='task2', dependent_task_ids=['task4']) task_wrapper3 = TaskWrapper(id='task3', dependent_task_ids=[]) task_wrapper4 = TaskWrapper(id='task4', dependent_task_ids=[]) task_wrappers = [task_wrapper1, task_wrapper2, task_wrapper3, task_wrapper4] Calling _print_task_wrapper_recursive(task_wrappers, 0, set()) would produce: ['task1', ' task2', ' task4', ' task3'] This shows the tasks and their dependencies formatted with appropriate indentation levels. Args: task_wrappers (list): List of task wrappers to print. indent_level (int): Current level of indentation for formatting. printed_tasks (set): Set of task IDs that have already been printed. Yields: List of formatted strings representing the task wrappers and their dependencies. r N)r r>strrr.rS)r task_wrappers indent_levelrXr=rdependent_task_wrapperss rrSz'TaskGraph._print_task_wrapper_recursiveCsN&  -,//*,  * **<<% <'**73< "% 77%|a'7H H H& %  Hs(B4B BB0B B B)N) rrr r!rr>r@rNrrr[rSr"rrr&r&Ws7 O48t5n>)@(1H Cy1Hrr&)r! __future__rrrr(typingr"googlecloudsdk.command_lib.storagergooglecloudsdk.corerrUrQrrr Errorr$r&r"rrrdsg ''5#!& ""JV\\]H]Hr