&`i| ddlZddlZddlZddlmZddlmZddlmZddl m Z m Z m Z m Z mZddlZejeZGddeZGdd ZeGd d Z d+dedededefdZde ee eeffdeddfdZde de efde ee eeffdeddfdZde de efde ee eeffde e efdZde edfde de e efde ee eefffdZd,d!Zded"ed#efd$Zd%e de efd&e e de efde ee eefffd'Z de ee eeffde de effd(Z!d%e de efde de effd)Z"d%e de efde de effd*Z#dS)-N) defaultdict)Enum)total_ordering)DictListOptionalSetTuplec$eZdZdZdZdZdZdZdS)_DAGNodeOperationTypez There are three types of operations that a DAG node can perform: 1. READ: Read from an input channel. 2. COMPUTE: Execute the method corresponding to the node. 3. WRITE: Write to an output channel. READCOMPUTEWRITEc|tjkrdS|tjkrdS|tjkrdSJd|)z A string representation of the operation type to be used in visualization. The result string is a single character because conciseness is preferred. RCWFzUnknown operation type: )r r rrselfs n/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/dag/dag_node_operation.pyviz_strz_DAGNodeOperationType.viz_strsV (- - -3 *2 2 23 *0 0 03777777uN)__name__ __module__ __qualname____doc__r rrrrrr r s> DG E 8 8 8 8 8rr cJeZdZ d dededeefdZdZdZ dZ d Z dS) _DAGNodeOperationN exec_task_idxoperation_type method_namec0||_||_||_dS)a# Args: exec_task_idx: The index of the task that this operation belongs to in the actor's ExecutableTask list. The index is not the same as bind_index because there may be more tasks bound to an actor than tasks that appear in the current compiled DAG. operation_type: The type of operation to perform. method_name: The name of the method that this operation originates from. This is only for visualization and debugging purposes. Nr typer")rr r!r"s r__init__z_DAGNodeOperation.__init__*s! +" &rc8d|jd|jd|jdS)Nz!_DAGNodeOperation(exec_task_idx: z, type: z, method_name: )r$rs r__repr__z_DAGNodeOperation.__repr__>sB 0"0 0 0Y 0 0!, 0 0 0 rcZd|jd|jd|jS)R A string representation of the node to be used in visualization. [z]  )r r"r%rrs rrz_DAGNodeOperation.viz_strFs7R4%QQ)9QQDIin_edges out_edgesset sync_idxspending_sync_idxs)rr:r;r<r>s rr&z_DAGOperationGraphNode.__init__WsY(#  ($8!TV TVBE JMrc Rd|jd|jd|jjd|jd S)Nz"_DAGOperationGraphNode(operation: z , task_idx: z , actor_id: z, requires_accelerator: r()r:r;r< _ray_actor_idr>rs rr)z_DAGOperationGraphNode.__repr__s` B. B B B B*8 B B&*%>  B B B rr4c|j|jkr|jS|jj|jf|jj|jfkS)z This function defines the order of the nodes in the priority queue used in `_select_next_nodes`. The priority queue is a min-heap, so the node with higher priority is considered "less than" the other node. )is_accelerator_opr:r r;r3s r__lt__z_DAGOperationGraphNode.__lt__sN  !U%< < <) ) N0$-@-D rc|j|jko3|jj|jjko|jj|jjkS)z Two operations are equal only when they have the same `exec_task_idx` and `type` and belong to the same actor. )r<r:r r%r3s rr5z_DAGOperationGraphNode.__eq__sE  !3 3 <,0MM <#u';; rc8t|j|jfS)zQ An operation is uniquely identified by its `task_idx` and type. )r0r:r;rs rr1z_DAGOperationGraphNode.__hash__sT^T]3444rreturnc*t|jSr/)lenr@rs r in_degreez _DAGOperationGraphNode.in_degrees4=!!!rcl|jdko)t|jt|jkS)z If a node is not an accelerator operation, it is ready when it has a zero in-degree. If it is an accelerator operation, it is ready when all the nodes in the operation have zero in-degrees. r)rOrNrDrCrs ris_readyz_DAGOperationGraphNode.is_readys5~" & ' '3t~+>+> > rc6|jjtjkSr/)r:r%r r rs ris_readz_DAGOperationGraphNode.is_reads~"&;&@@@rcD|jjtjko|jS)z^ A node is an accelerator read if it is a read node and requires accelerator. )r:r%r r r>rs ris_accelerator_readz*_DAGOperationGraphNode.is_accelerator_reads% N #8#= = *) rcD|jjtjko|jS)zd A node is an accelerator compute if it is a compute node and requires accelerator. )r:r%r rr>rs ris_accelerator_computez-_DAGOperationGraphNode.is_accelerator_computes% N #8#@ @ *) rcD|jjtjko|jS)z` A node is an accelerator write if it is a write node and requires accelerator. )r:r%r rr>rs ris_accelerator_writez+_DAGOperationGraphNode.is_accelerator_writes% N #8#> > *) rc,|jp |jp|jSr/)rUrWrYrs rrHz(_DAGOperationGraphNode.is_accelerator_ops%  $ )* )( rc4|jS)r+)r:rrs rrz_DAGOperationGraphNode.viz_strs~%%'''rc>|jjSr/)r<rFhexrs r _actor_idz _DAGOperationGraphNode._actor_ids .22444rN)r4r9)rrrrr6boolr&r)rIr5r1propertyrOrQrSrUrWrYrHrr^rrrr9r9Us(O$(O(O. (O # (O(O(O(OT   (     555 "3"""X"  $    X  AAAAXA T   X     X  d   X  4   X ((( 55X555rr9F from_nodeto_nodelabelcontrol_dependencycz||f|j|j|jjf<||f|j|j|jjf<dS)a8 Add an edge from `from_node` to `to_node`. Args: from_node: The node from which the edge originates. to_node: The node to which the edge points. label: The label of the edge. This will be used to annotate the edge in the visualization of the execution schedule. N)rAr;r:r%r@)rbrcrdres r _add_edgergsT GI)7+<+ABC HGi()*=*BCDDDrgraphnoderLc|j|jjf}|jD]-\}}|||}|j|.dS)z? Update the node as pending for its synchronous nodes. N)r;r:r%rCrDadd)rhriidxr;op_type sync_nodes r_update_pending_sync_idxsro s] =$.- .C!^--'(OG, #'',,,,--ractor_to_candidateszray._raylet.ActorIDc|jdks Jd|jrw|jD]o\}}|||}|j|j|jjf|jrt|jdksJt||pt|j dkrt|||j rt|j dkr'tj||jj|dS|j D]:\}}|||}tj||jj|9dSdS)z Push the node with a zero in-degree to the candidates if its operation is ready. If it has synchronous nodes, its accelerator operation is not ready until all the nodes are pending, then all the nodes will be pushed to the candidates. rz!Expected to have a zero in-degreeN)rOrYrAr@popr;r:r%rUrNrorCrQheapqheappushr<r^)rprhrir;rm read_noderns r_push_candidate_node_if_readyrvs >Q    C      8!% 8 8 Hgh0I   " "DM4>3F#G H H H0 QS9K5L5LPQ5Q5Q5QQ %eY 7 7 7 7 4>a!%... }  t~  ! # # N#D$5$?@      &*^  !'!(OG4 ' (>(HI    rc>d}|D],}t|dkr| |d|kr|d}-|dS|g}t|jdkr6|jD].\}}|||}||kr||/|D]=}||jj}||tj|>|D]A}||jj}||vr)||tj|B|S)a This function selects the next nodes for the topological sort to generate execution schedule. If there are multiple candidate _DAGOperationGraphNodes, select the node with the top priority. The priority is defined in `_DAGOperationGraphNode.__lt__`. For the implementation details, we maintain a priority queue for each actor, where the head of the priority queue is the node with the smallest `exec_task_idx`. When a node has a zero in-degree, it is added to the corresponding actor's priority queue. For a node other than an accelerator collective node, it is ready to be executed if it has a zero in-degree. For an accelerator collective node, it is ready to be executed when all the nodes in its collective operation have zero in-degrees. If a node is an accelerator collective node, it updates the `ready_collective_nodes` of all the nodes in its collective operation. Unless all the nodes in its collective group have zero in-degrees, this node is removed from the candidate list. Eventually, exactly one accelerator collective node from its collective operation is selected from the candidate list. If the selected node is an accelerator write node, select all the downstream accelerator read nodes. If the selected node is an accelerator collective node, select all the accelerator compute nodes in its collective operation. Args: actor_to_candidates: A dictionary mapping an actor id to a list of candidate nodes. The list is maintained as a priority queue, so the head of the queue, i.e., `candidates[0]`, is the node with the smallest `bind_index`. graph: A dictionary mapping the index of a task to a dictionary of its _DAGOperationGraphNodes for different operations. Returns: A list of _DAGOperationGraphNodes to be placed into the corresponding execution schedules. Nr) valuesrNrCappendr<r^removersheapify)rprhtop_priority_node candidates next_nodesr;rmris r_select_next_nodesrEskN)0022.. z??a     $ 1 8I(I(I *1  t#$J  &''1,,!2!< ( ( Hg?7+D(((!!$'''""():)DE $ j!!!!&&():)DE :     d # # # M* % % % r idx_to_taskz&ray.dag.compiled_dag_node.CompiledTaskactor_to_operation_nodesr=cv|sJi}|D]\}}d}|D]}|dj}|d|d|d} } }t|| t| | |t|| dd| }||vsJtj|tj| tj| i||<ddlm} m } m } ddl m }|D]\}}t|j| st|j| s1t|j| r |jjrS|jD]S}||j}t|| r!||tj} t|| r|jr||j}|D]|}||vrv||tj}t| || jrd nd | jr=|tjf|tjfh}| |fD]}|j|}||tj}t| || jrd nd | jr=|tjf|tjfh}| |fD]}|j|Ut)t*}|D]W\}}t|j| r=|jjs1||jj|tjfX|D]}|D]\}}||||_|S) ai Generate a DAG node operation graph by adding edges based on the following rules: #1 Add edges from READ to COMPUTE, and from COMPUTE to WRITE, which belong to the same task. #2 Add an edge from COMPUTE with bind_index i to COMPUTE with bind_index i+1 if they belong to the same actor. #3 Add an edge from WRITE of the writer task to READ of the reader task. This is the step one of building an execution schedule for each actor. Args: idx_to_task: A dictionary that maps the `task_idx` to the `CompiledTask`. `CompiledTask` contains information about a DAGNode and its downstream nodes. actor_to_operation_nodes: A dictionary that maps an actor handle to a list of lists of _DAGOperationGraphNode. For the same actor, the index of the outer list corresponds to the index of the ExecutableTask in the list of `executable_tasks` in `actor_to_executable_tasks`. In the inner list, the order of operations is READ, COMPUTE, and WRITE. Returns: A graph where each node is a _DAGOperationGraphNode. The key is `task_idx`, the index to retrieve its task from `idx_to_task`, and the value is a dictionary that maps the _DAGNodeOperationType (READ, COMPUTE, or WRITE) to the corresponding _DAGOperationGraphNode NrraT)ClassMethodNodeCollectiveOutputNodeMultiOutputNode)_CollectiveOperation acceleratorshm)itemsr;rgr r rrray.dagrrrray.dag.collective_noder isinstancedag_nodeis_class_method_outputdownstream_task_idxsr>rCupdaterrB collective_oprkrx)rrrh_operation_nodes_listprev_compute_nodeoperation_nodesr;ru compute_node write_noderrrrtaskdownstream_task_idxdownstream_dag_node consumer_idxs consumer_idxidxsricollective_op_to_idxsrms r_build_dag_node_operation_graphrsBF ;LNE#;#A#A#C#C  3  O&q)2H"""&0|I i . . . lJ / / /!,+\2tDDD , 5((((%*I%-|%+ZE(OO# 0ONNNNNNNNN<<<<<<&++--6060$ t}o 6 6 $-)=>>   t}o 6 6  4  #'#<& 0& 0 "-.A"B"K -?? x)>)DEJ.@@ '> !,,? @ U $1< B B089   &,,..66!% 6 6 Hg15E(OG $ . . 6 Lractorc\|jj}|j}d|d|S)z Returns the label of an actor in the visualization of the execution schedule. Args: actor: The actor to be represented. zActor class name: z Actor ID: )'_ray_actor_creation_function_descriptor class_namerFr])rractor_ids r_actor_viz_labelrs9>IJ"&&((H B B B B BBrrloptimized_indexc\|d|d|z}|jd|}||fS)a7 Returns the visualization id and label of a node. The visualization id is unique across all nodes. Args: node: The node to be represented. idx: The index of the node in the execution schedule. optimized_index: The index of the node in the optimized execution schedule. r-,r)rr^)rirlrnode_viz_label node_viz_ids r_node_viz_id_and_labelr+sG\\^^&A#&A&A&A&AAN^66n66K  &&ractor_to_execution_scheduleactor_to_overlapped_schedulec d ddl}n#t$rtdwxYw|d}i}||}|D]\}}||}dt |D} |j} |d| 5} | | t| t |D]U\} } | | }t| | |\}}|| krd nd }| ||| ||| <V dddn #1swxYwY|D]\}}t |D]v\} } || }| j D]O\}}|\}}|\}}|||}||}|d krdnd }|rdnd}||||||Pw|d5}|dddd| ddd}| d|d|ddddddn #1swxYwYtd |d!d"d#$dS)%a  Visualize the execution schedule for each actor. The visualization will be saved as a PNG file named `compiled_graph_schedule.png`. Details of the visualization: # noqa Node description format: [] , Node description fields: operation: is R(READ), C(COMPUTE), or W(WRITE) orig_index: the index in the original execution schedule overlap_index: the index in the overlap-communication optimized execution schedule If this is different from orig_index, the node is highlighted in red color Node grouping: The nodes belonging to the same actor are grouped in the same rectangle The actor class name and the actor id are shown in the rectangle Edges: black color (without label): data dependency black color (annotated with "shm"): shared memory channel blue color (annotated with "accelerator): accelerator channel dashed edge: control dependency between compute operations Args: actor_to_execution_schedule: A dictionary that maps an actor handle to the execution schedule which is a list of operation nodes. actor_to_overlapped_schedule: A dictionary that maps an actor handle to the optimized execution schedule which is a list of operation nodes. graph: A graph where each node is a _DAGOperationGraphNode. The key is `task_idx`, the index to retrieve its task from `idx_to_task`, and the value is a dictionary that maps the _DAGNodeOperationType (READ, COMPUTE, or WRITE) to the corresponding _DAGOperationGraphNode. It is generated by `_build_dag_node_operation_graph`. rNzrPlease install graphviz to visualize the execution schedule. You can install it by running `pip install graphviz`.DAG)commentci|]\}}|| Srr).0iris r z1_visualize_execution_schedule..{s+# # # 4D!# # # rcluster_)name)rankrdredblack)colorrbluedashedsolid)rdrstylecluster_legendLegendt20 lightgrey)rdlabellocfontsizebgcolor example_nodez[0] bwd C 10,10 a<
Node description format:
[<task_index>] <method_name> <operation> <orig_index>, <overlap_index>
Node description fields:
operation: is R(READ), C(COMPUTE), or W(WRITE)
orig_index: the index in the original execution schedule
overlap_index: the index in the overlap-communication optimized execution schedule
If this is different from orig_index, the node is highlighted in red color
Node grouping:
The nodes belonging to the same actor are grouped in the same rectangle
The actor class name and the actor id are shown in the rectangle
Edges:
black color (without label): data dependency
black color (annotated with "shm"): shared memory channel
blue color (annotated with "accelerator): accelerator channel
dashed edge: control dependency between compute operations
>example_explanation plaintext)shapeinvis)rzLWriting compiled graph schedule visualization to compiled_graph_schedule.pngcompiled_graph_schedulepngF)formatview)graphviz ImportErrorDigraphr enumeraterFr]subgraphattrrgetrrirAedgeloggerinforender)rrrhrdotnode_to_viz_idrexecution_nodesoverlapped_schedulenode_to_optimized_indexrrrrirrrrout_edgeviz_infordre out_task_idx out_op_typeout_nodeout_node_viz_idrlegend explanations r_visualize_execution_scheduler<sZ      D      5  ) )C8:N#+(C$"="C"C"E"E33:5A# # #,-@#A#A# # # &**,, \\4(44\ 5 5 3 MMx/?/F/FM G G G$_55 3 34"9"="=d"C"C.D!_//+ ^"1A!5!57 k> GGG'2t$$ 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3 3#>"C"C"E"E   11  GAt(.K&*n&:&:&<&<  "(,4)),4) k .{;"0":"'="8"8g$6CGURW   + , ,J (S4 UUU  N$7888  .  );k JJJ N$9 III;JJJJJJJJJJJJJJJ> KK )JJ(UJCCCCCs.!,B EE E A"I33I7:I7ctt}tt}|D];\}}|D]!\}}|jdkrt |||"|D]i}|j D]_\}} ||| } | |vr| j |j |jjf| jdkrt ||| `jt!|t!|dzks Jd|S)a  Generate an execution schedule for each actor. The schedule is a list of operation nodes to be executed. The function uses a topological sort algorithm to generate the schedule. Args: graph: A graph where each node is a _DAGOperationGraphNode. The key is `task_idx`, the index to retrieve its task from `idx_to_task`, and the value is a dictionary that maps the _DAGNodeOperationType (READ, COMPUTE, or WRITE) to the corresponding _DAGOperationGraphNode. It is generated by `_build_dag_node_operation_graph`. Returns: actor_to_execution_schedule: A dictionary that maps an actor handle to the execution schedule which is a list of operation nodes to be executed. rTNz Expected all nodes to be visited)rlistrrOrvrBrrkr<ryrAr@rrr;r:r%rN) rhrrpr node_dictri visited_nodesnodesout_node_task_idx out_node_typers r%_generate_actor_to_execution_schedulers2 D  D PP 9 (( P PGAt~""-.A5$OOO  PEEMX##6>> =  H HD},,,,   d # # # '(9 : A A$ G G G G X XD48N X X0!= !23MB},,!%%t}dn6I&JKKK%**12EuhWWW X#X6 }  Ua / / /1S / / / &&rcLtj|}|D]}tt |D]}||jjtjkr||j rt|dz ddD]}||jjtj kr&||}|||}|||dz|dz<|||<nP||jjtjks ||jjtj kr||j rn܌|S)a From an existing execution schedule, generate a new schedule by overlapping computation and communication. Currently, the algorithm generates a new schedule for each actor as follows: For each accelerator read operation (i.e., recv), scan backwards to find the nearest compute node to swap with so that the accelerator read operation can be overlapped with computation. Collective operations are not yet supported. Args: actor_to_execution_schedule: A dictionary that maps an actor handle to the existing execution schedule for the actor. The schedule is a list is a list of operations to be executed. Returns: A dictionary that maps an actor handle to the overlapped execution schedule for the actor. r) copydeepcopyrxrangerNr:r%r r r>rr)rrrrjaccelerator_read_opprev_opss r'_generate_overlapped_execution_scheduler sd8 122! <BBDDs.//00  A#A&059N9SSS'*?T q1ub"--A+A.8=0899/B!.D+#6qs#;=E+AEAEM:1D+A.+A.8=0566.q1;@0677-a0E7 7 8 ('rc>d|DS)zx Extract _DAGNodeOperation from _DAGOperationGraphNode in the schedule and discard unnecessary information. c.i|]\}}|d|DS)cg|] }|j Sr)r:)rris r z:_extract_execution_schedule...Ss1114111rr)rrrs rrz/_extract_execution_schedule..Rs=    E5 115111   r)r)rs r_extract_execution_schedulerIs/  7==??   r)raF)rr=)$rrslogging collectionsrenumr functoolsrtypingrrrr r ray getLoggerrrr rr9r7r_rgr6rorvrrrrrrrrrrrr s ######$$$$$$33333333333333  8 $ $88888D8886)U)U)U)U)U)U)U)UX[5[5[5[5[5[5[5[5B$ % #  2 - T/1GGHH I -  -  - - - -(3T:P5QQR( T/1GGHH I( !( ((((VG3T:P5QQRG T/1GGHH IGd)*+GGGGTMcCCDM"d+A&B!CCM  #t)+AAB BC MMMM` C C C C' ''*'=@''''"AD!%&#? ?@I'I'I'I'X:(!%&#? ?@ :(:(:(:(z !%&