%`isddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z m Z ddlmZmZddlZddlmcmZddlmZddlmZddlmZddlmZmZmZddlm Z dd l!m"Z"m#Z#m$Z$dd l%m&Z&m'Z'm(Z(dd l)m*Z*e d gd Z+e dgdZ, ej-e.Z/de0de'dee0fdZ1GddZ2dS)N) defaultdict namedtuple)CallableOptional) cloudpickle) pickle_dumps) ray_constants)is_class_methodis_function_or_methodis_static_method)KV_NAMESPACE_FUNCTION_TABLE)check_oversized_function ensure_strformat_error_message)&WORKER_PROCESS_SETUP_HOOK_KEY_NAME_GCSJobIDPythonFunctionDescriptor)RemoteFunctionFunctionExecutionInfofunction function_name max_callsImportedFunctionInfojob_id function_idrrmodulerkey_typerkeyc|;d||gSd|||gS)N:)joinhexencode)rrr s q/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/_private/function_manager.pymake_function_table_keyr'3s_ {yy(FJJLL$7$7$9$9:;;;yy(FJJLL$7$7$9$93?@@@ceZdZdZdZdZdZdZdZ dde d e e d e fd Z d Z dd ed e e d e efdZdZdZdZddefdZdZdZdZdZdZdefdZdS)FunctionActorManagerarA class used to export/load remote functions and actors. Attributes: _worker: The associated worker that this manager related. _functions_to_export: The remote functions to export when the worker gets connected. _actors_to_export: The actors to export when the worker gets connected. _function_execution_info: The function_id and execution_info. _num_task_executions: The function execution times. imported_actor_classes: The set of actor classes keys (format: ActorClass:function_id) that are already in GCS. c4||_g|_g|_td|_td|_t |_i|_tj |_ i|_ d|_ tj|_dS)NciSNr.r(r&z/FunctionActorManager.__init__..QsBr(ciSr-r.r.r(r&r/z/FunctionActorManager.__init__..Rsr(r)_worker_functions_to_export_actors_to_exportr_function_execution_info_num_task_executionssetimported_actor_classes_loaded_actor_classes threadingRLocklockexecution_infos _num_exportedLock _export_lock)selfworkers r&__init__zFunctionActorManager.__init__Js $&!!#)4JJ(?(?%$/ $;$;!'*ee#%'"O%% !%N,,r(c>|j}|j|xxdz cc<dS)Nrr5r@function_descriptorrs r&increase_task_counterz*FunctionActorManager.increase_task_countergs/)5  !+...!3.....r(c*|j}|j|Sr-rErFs r&get_task_counterz%FunctionActorManager.get_task_counterks)5 (55r(cddl}|}tj||d|jdz|z}t j|dS)aDThe identifier is used to detect excessive duplicate exports. The identifier is used to determine when the same function or class is exported many times. This can yield false positives. Args: function_or_class: The function or class to compute an identifier for. Returns: The identifier. Note that different functions or classes can give rise to same identifier. However, the same function should hopefully always give rise to the same identifier. TODO(rkn): verify if this is actually the case. Note that if the identifier is incorrect in any way, then we may give warnings unnecessarily or fail to give warnings, but the application's behavior won't change. rN)filedepth:zutf-8) ioStringIOdis__name__getvaluehashlibsha1r%digest)r@function_or_classrP string_filecollision_identifiers r&compute_collision_identifierz1FunctionActorManager.compute_collision_identifieros~  kkmm  ! 1====09C?+BVBVBXBXX|077@@AAHHJJJr(ctj|}d|dD}|} |D]}t||}|S#t$rYdSwxYw)z9Try to load a function or class in the module from local.cg|]}||Sr.r.).0parts r& zJFunctionActorManager.load_function_or_class_from_local..sLLL$tLLLLr(.N) importlib import_modulesplitgetattr Exception)r@ module_namefunction_or_class_namerpartsobjectr_s r&!load_function_or_class_from_localz6FunctionActorManager.load_function_or_class_from_locals(55LL"8">">s"C"CLLL  / / ..M   44 sA AAN setup_functimeoutreturnc ~t|d|j}tj|t j}ttj |j j |}t||jd|j  |j j|t!j|j j ||ddt j|n7#t&$r*}t(d|jd|d}~wwxYw|S) z2Export the setup hook function and return the key.z/Cannot serialize the worker_process_setup_hook r)rrrTrmz Failed to export the setup hook raN)rrSrU shake_128rWr ID_SIZEr'rr%r1current_job_idbinaryr gcs_clientinternal_kv_putpickledumpsr rflogger exception)r@rlrmpickled_functionfunction_to_run_idr es r&export_setup_funcz&FunctionActorManager.export_setup_funcso(  VAT V V   %./?@@GG  !  & 3 9 ; ; L ' . . 0 0     ! j1:t|     L # 3 3 "&,"="D"D"F"F'9$49 4          Lj6ILLL   G    s%A D D:%D55D:c |jjr.|j}|j|j}}||| dS|j}|j}t||j d|jtd|jj |jj }|jj|t rdSt#j|jj |jj |j |j||||jd}|jj||dt dS)zPickle a remote function and export it to redis. Args: remote_function: the RemoteFunction object. Nzremote functionRemoteFunction)rrrrrrZrT)r1load_code_from_local_function_descriptorrgrrk _function_pickled_functionr_function_namer'rsrrtruinternal_kv_existsr rwrx __module__r[ _max_callsrv) r@remote_functionrGrgrrr{r vals r&exportzFunctionActorManager.exportsx < , "1"F #/#1'K66{MRR",*<   *  L    &  L '  0 < C C E E   < " 5 5c;V W W  Fl,5<<>>.COVVXX!0!?"-,(,(I(I((S(S,7      // d7     r(r c|jj|t|dSt jgd}t fd|DS)Nrprc3BK|]}|VdSr-getr^fieldvalss r& z?FunctionActorManager.fetch_registered_method..s--R-R%dhhuoo-R-R-R-R-R-Rr()r1ruinternal_kv_getr rwloadsr_make)r@r rmfieldsrs @r&fetch_registered_methodz,FunctionActorManager.fetch_registered_methods|&66 ,g7   <4<%%DF(---R-R-R-R6-R-R-RRR Rr(c  ||}|sdS|\}}}}}}tj|} tj|} t |}|j5d|j| < tj|} || _ t| |||j | <n#t$r{ttj fd} t| |||j | <t d|d| d| d d YnwxYwd d d n #1swxYwYd S) zImport a remote function.FrrcBtdtjd)NzThe remote function failed to import on the worker. This may be because needed library dependencies are not installed in the worker environment or cannot be found from sys.path z: ) RuntimeErrorsyspath)argskwargs traceback_strs r&fzBFunctionActorManager.fetch_and_register_remote_function..f)s:&:8 ::+8 ::r(z(Failed to unpickle the remote function 'z' with function ID z . Job ID:z .Traceback: z. NT)rray FunctionIDrintr;r5rwrrrr4rfr traceback format_excrydebugr$)r@r remote_function_info job_id_strfunction_id_strrserialized_functionrrrrrrrs @r&"fetch_and_register_remote_functionz7FunctionActorManager.fetch_and_register_remote_function s#;;C@@# 5 !      n_55 :&& NN Y, , 56D %k 2) !<(;<<J'-#=R%]i>>>-k::K   !5Y5I5K5K L L >Smy>>>-k: 5%55#.??#4#455%55$1 555/  , , , , , , , , , , , , , , , Zts7" D8.B$"D8$BD)&D8(D))D88D<?D<c|j}||jvr |j|S|jjr8|s$||dur |j|St jd5|||dddn #1swxYwY |j}|j|}n,#t$r}d|d|d|}t|d}~wwxYw|S)aGet the FunctionExecutionInfo of a remote function. Args: job_id: ID of the job that the function belongs to. function_descriptor: The FunctionDescriptor of the function to get. Returns: A FunctionExecutionInfo object. Twait_for_functionNz,Error occurs in get_execution_info: job_id: z, function_descriptor: z . Message: ) rr4r1ris_actor_method_load_function_from_local profilingprofile_wait_for_functionKeyError)r@rrGrinfor}messages r&get_execution_infoz'FunctionActorManager.get_execution_infoMs*5  $7 7 70= = < , F&6688 F112EFF$NN8EE 2 3 3 A A  # #$7 @ @ @ A A A A A A A A A A A A A A A $-9K0=DD $ $ $ $66...3  7## #  $ s*6BB B%B:: C#CC#c|rJ|j}|j|j}}|||}|Et |t r|j}n|}t||d|j |<d|j |<dSdS)NrrTF) rrrgrrk isinstancerrrr4r5)r@rGrrgrrjrs r&rz.FunctionActorManager._load_function_from_localvs&6688888)5  +  -# 77 ]SS  &.11 "!+!9N!+:::D )+ 6 67D %k 245r( rc tj}d} |j5|jjrj|j|jvr ddddStd||j}| |dur ddddSnH|jjrJ|jj|jj vsJ ddddS dddn #1swxYwYtj|z |krd|d|jj d|jj dtjd }|sQt |t$jj|jt,j|| d}tjd ) a{Wait until the function to be executed is present on this worker. This method will simply loop until the import thread has imported the relevant function. If we spend too long in this loop, that may indicate a problem somewhere and we will push an error message to the user. If this worker is an actor, then this will wait until the actor has been defined. Args: function_descriptor : The FunctionDescriptor of the function that we want to execute. job_id: The ID of the job to push the error message to if this times out. FTNrzJThis worker was asked to execute a function that has not been registered (z, node=z , worker_id=z, pid=z). You may have to restart Ray.)rgMbP?)timer;r1actor_idis_nilrr4r'rtractorsnode_ip_address worker_idr$osgetpidryerrorr_privateutilspush_error_to_driverr WAIT_FOR_FUNCTION_PUSH_ERRORsleep)r@rGrrm start_time warning_sentr warning_messages r&rz'FunctionActorManager._wait_for_functionsY[[  $   <(//11*6$:WWW         6-"/;BBDD  BB3GG4OO!        P $|4;;=====<0DL4GGGGG!        P               $y{{Z''11H5HHH L8HH"&!7!;!;!=!=HH9;; HHH $LL111L&;; %B'% < $ Ju   I$ s.C<AC<&2=$K66{JOO<.5577     7 ,%   . : A A C C   ".  916 9 9" "  =GMMcRRSUV8D+mmoo$($E$Ee$L$L"&*T2D-E-E"F"F    ! W % \ *  L     // .//7R     r(c.|j}|j|d}||jjr.||}||||}n|||}||j|<|j}|j}tj |t}|D]`\}} |dkr|} nt|||} | j} | || } t| |d|j| <d|j| <ad|j|<|S)zLoad the actor class. Args: job_id: job ID of the actor. actor_creation_function_descriptor: Function descriptor of the actor constructor. Returns: The actor class. N) predicaterBrr)rr8rr1r_load_actor_class_from_local_load_actor_class_from_gcsrgrinspect getmembersr r_make_actor_method_executorrr4r5) r@rrr actor_classrgactor_class_name actor_methodsactor_method_name actor_methodmethod_descriptor method_idexecutors r&load_actor_classz%FunctionActorManager.load_actor_classs9D 044[$GG  |0 "??6 &"&"A"A B##K #==> 7BD &{ 3=HKAL #.'<M4A 9 9/!<% 22(J%%(@#%68H))%.9 ;;%|.TemporaryActorc KdS)zDummy method for this fake actor class to work for async actors. Without this method, this temporary actor class fails to initialize if the original actor class was async.Nr.)r@s r&__dummy_methodzTFunctionActorManager._create_fake_actor_class..TemporaryActor.__dummy_method[s r(N)rSr __qualname___TemporaryActor__dummy_methodr.r(r&TemporaryActorrZs#     r(rc.tdd)NzThe actor with name z failed to import on the worker. This may be because needed library dependencies are not installed in the worker environment: )r)rrrrs r&temporary_actor_methodzMFunctionActorManager._create_fake_actor_class..temporary_actor_methodas6:'7::+8:: r()setattr)r@rrrrrmethods ` ` r&_create_fake_actor_classz-FunctionActorManager._create_fake_actor_classWsy              ) D DF NF,B C C C Cr(c td||j}|jj|t  gd} i ntj  fd|D\}}}}} t|}t|} tj |}tjt| } d} |j 5tj|} dddn #1swxYwYnb#t$rUtd|t#t%j} ||| | } YnwxYw| | _| S)zLoad actor class from GCS.r)rrrrrNc3BK|]}|VdSr-rrs r&rzBFunctionActorManager._load_actor_class_from_gcs..}sFO O %DHHUOOO O O O O O r(zFailed to load actor class %s.)r'rrtr1rurr rwrrrrrr;rfryrrrrrr)r@rrr rrrr pickled_classrrgrrrs @r&rz/FunctionActorManager._load_actor_class_from_gcsns%   . : A A C C  |&66s2 D >DD DD AE)(E) method_namec2fd}|_|_|S)aMake an executor that wraps a user-defined actor method. The wrapped method updates the worker's internal state and performs any necessary checkpointing operations. Args: method_name: The name of the actor method. method: The actor method to wrap. This should be a method defined on the actor class and should therefore take an instance of the actor as the first argument. Returns: A function that executes the given actor method on the worker's stored instance of the actor. The function also updates the worker's internal state to record the executed method. ctptt|}|r|i|S|g|Ri|Sr-)r r type) _FunctionActorManager__ray_actorrris_boundrrs r&actor_method_executorzOFunctionActorManager._make_actor_method_executor..actor_method_executorsn&v..2B[!!;33H <vt.v...vk;D;;;F;;;r()namer)r@rrrs `` r&rz0FunctionActorManager._make_actor_method_executors< < < < < < <&1"'-$$$r(r-)r)rSrr__doc__rBrHrJr[rkrrrbytesr~rstrrrrrrrrrrrrrr.r(r&r*r*:s  ---:444666KKK2   >B..".-5c]. ....`/ / / d26SSS!)#S & 'SSSS(BBBH'''R644c4444l7 7 7 xAAAF"....`!%s!%!%!%!%!%!%r(r*)3rRrUrbrrloggingrrr9rr collectionsrrtypingrrrray._private.profilingrrrrwray._common.serializationr ray._privater ray._private.inspect_utilr r r ray._private.ray_constantsr ray._private.utilsrrr ray._rayletrrrray.remote_functionrrr getLoggerrSryr r'r*r.r(r&rs    ////////%%%%%%%% *********%%%%%%222222&&&&&& CBBBBB  /....." GGG"zQQQ P  8 $ $AeAUA%AAAAE %E %E %E %E %E %E %E %E %E %r(