o bit @sddlZddlZddlZddlZddlmZmZmZmZm Z ddl Z ddl m m Z ddlmZddlmZddlmZeeZdZddZded efd d Zded efd d Zde eeefefdeddd e eeefeffddZde eeefefded e eeefeffddZde eeefefddd e eeefeffddZ ded eefddZ!ded eefddZ"ded eefddZ#dS) N)AnyCallableDictOptionalUnion) load_class) RuntimeEnvzray_runtime_env_func::cCsttjtjdS)N60)intosenvironget ray_constants2RAY_WORKER_PROCESS_SETUP_HOOK_LOAD_TIMEOUT_ENV_VARrrW/home/ubuntu/.local/lib/python3.10/site-packages/ray/_private/runtime_env/setup_hook.pyget_import_export_timeouts rkeyreturncCstt|SN)RUNTIME_ENV_FUNC_IDENTIFIERbase64 b64encodedecoderrrr_decode_function_keysrcCs$|tsJt|ttdSr) startswithrr b64decodelenrrrr_encode_function_keysr runtime_env setup_funcworkerz ray.Workerc Cst|tsJz |jj|td}Wnty%}ztjd|d}~ww| di}t j |vs:Jdt j dt ||t j <||d<|j |d<|S)Ntimeoutz$Failed to export the setup function.env_vars The env var, ?, is not permitted because it is reserved for the internal use.worker_process_setup_hook) isinstancerfunction_actor_managerexport_setup_funcr Exceptionray exceptionsRuntimeEnvSetupErrorr r!WORKER_PROCESS_SETUP_HOOK_ENV_VARr__name__)r r!r"rer%rrrexport_setup_func_callable$s.      r3setup_func_modulecCsLt|tsJ|di}tj|vsJdtjd||tj<||d<|S)Nr%r&r')r)strr rr0)r r4r%rrrexport_setup_func_moduleCs    r6cCsT|d}|dur |St|trt|||St|tr t||Stdt|d)aUploads the worker_process_setup_hook to GCS with a key. runtime_env["worker_process_setup_hook"] is converted to a decoded key that can load the worker setup hook function from GCS. i.e., you can use internalKV.Get(runtime_env["worker_process_setup_hook]) to access the worker setup hook from GCS. Args: runtime_env: The runtime_env. The value will be modified when returned. worker: ray.worker instance. decoder: GCS requires the function key to be bytes. However, we cannot json serialize (which is required to serialize runtime env) the bytes. So the key should be decoded to a string. The given decoder is used to decode the function key. r(Nz2worker_process_setup_hook must be a function, got .)r r)rr3r5r6 TypeErrortype)r r"r!rrr*upload_worker_process_setup_hook_if_neededRs     r:worker_process_setup_hook_keycCs&|dusJ|tst|St|S)zLoad the setup hook from a given key and execute. Args: worker_process_setup_hook_key: The key to import the setup hook from GCS. Returns: An error message if it fails. None if it succeeds. N)rr"load_and_execute_setup_hook_module load_and_execute_setup_hook_func)r;rrrload_and_execute_setup_hookvs r>c CsFz t|}|WdSty"d|d|dt}|YSw)Nz)Failed to execute the setup hook method, z; from ``ray.init(runtime_env={'worker_process_setup_hook': z})``. Please make sure the given module exists and is available from ray workers. For more details, see the error trace below. )rr, traceback format_exc)r;r! error_messagerrrr<s  r<cCstjjj}|js J|j}z |jt|td}Wnt y/dtdt }|YSwzt |j}Wnt yJdt }|YSwz|WdSt ygd|jdt }|YSw)Nr#z#Failed to import setup hook within z seconds. z-Failed to deserialize the setup hook method. z7Failed to execute the setup hook method. Function name: )r-_privater" global_worker connectedr*fetch_registered_methodrrr,r?r@pickleloadsfunction function_name)r;r" func_managerworker_setup_func_inforAr!rrrr=sH      r=)$rloggingr r?typingrrrrrr-ray._private.ray_constantsrCrray.cloudpickle cloudpicklerGray._common.utilsrray.runtime_envr getLoggerr1loggerrrbytesr5rrr3r6r:r>r<r=rrrrsj       $