&`iz2ddlZddlZddlZddlmZmZmZmZddlm Z ddl m Z m Z ddl mZddlmZmZddlmZmZmZmZddlmZejeZd Zd Zd Zd Zd ZeeeeegZ edGddeZ!dS)N)AnyDictListOptional)Celery) task_failure task_unknown)get_replica_context)DEFAULT_CONSUMER_CONCURRENCYSERVE_LOGGER_NAME)CeleryAdapterConfigTaskProcessorAdapterTaskProcessorConfig TaskResult) PublicAPI worker_poolworker_concurrencytask_ignore_resulttask_acks_latetask_reject_on_worker_lostalpha) stabilityc veZdZUdZeed<eed<dZee j ed<dZ ee ed<e Zeed<deffd Ze fd efd Zd&d Z d'd efdZd efdZdZd(defdZdZdZd ee effdZd eefdZ d)dede dededef dZ d)dede de d ed!ef d"Z!d#e d$e de"fd%Z#xZ$S)*CeleryTaskProcessorAdapterz Celery-based task processor adapter. This adapter does NOT support any async operations. All operations must be performed synchronously. _app_configN_worker_thread_worker_hostname_worker_concurrencyconfigcvtj|i|t|jtst d|jjrbt|jjttz}|r tdt|d||_ dS)NzMTaskProcessorConfig.adapter_config must be an instance of CeleryAdapterConfigzJThe following configuration keys cannot be changed via app_custom_config: zA. These are managed internally by the CeleryTaskProcessorAdapter.) super__init__ isinstanceadapter_configr TypeErrorapp_custom_configsetkeysCELERY_DEFAULT_APP_CONFIG ValueErrorsortedr)selfr argskwargsconflicting_keys __class__s l/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/serve/task_processor.pyr#z#CeleryTaskProcessorAdapter.__init__6s$)&)))&/1DEE _   2 "%7<<>>  -.. /    WaghxayayWWW  consumer_concurrencyc t|jj|jjj|jjj|_tdt|tdtdtdi}|jjj r$| |jjj |jj ||jj|jjd|jjdi}|jjr'|jjd|jjd||jj<|jjr'|jjd|jjd||jj<|jj |dd|jjii |jjj/|jj |jjj |jjrt%j|j|jjrt+j|jdSdS) N)backendbrokerthreadsFTdirect)exchange exchange_type routing_key*queue) task_queues task_routes)broker_transport_options)rr queue_namer% backend_url broker_urlrCELERY_WORKER_POOLCELERY_WORKER_CONCURRENCYCELERY_TASK_IGNORE_RESULTCELERY_TASK_ACKS_LATE!CELERY_TASK_REJECT_ON_WORKER_LOSTr'updateconffailed_task_queue_nameunprocessable_task_queue_namerArconnect_handle_task_failurer _handle_unknown_task)r-r4app_configuration queue_configs r2 initializez%CeleryTaskProcessorAdapter.initializeNs L #L/;<.9      %'; %u !4 -t   < & 8 T  $ $T\%@%R S S S /000 L # L3!)#|6&&  < .  L?!)#|BAAL< = < 5  LF!)#|IHHLC D $gt|67     < & ? K IN ! !)-)D)] "    < . <  !: ; ; ; < 5 <  !: ; ; ; ; ; < )idstatus created_atresult) r send_taskrrBrrfrgtimeri)r- task_namer.r/options task_responses r2enqueue_task_syncz,CeleryTaskProcessorAdapter.enqueue_task_syncss, +  ,)       'y{{ '     r3cx|j|}t|j|j|jS)N)rfrirg)r AsyncResultrrfrirg)r-task_id task_detailss r2get_task_status_syncz/CeleryTaskProcessorAdapter.get_task_status_syncs?y,,W55 &&    r3c |j5|jrtddSt j}|jjd||_dd|jd|j j g}tj |jj |f|_|jtd|jdS) z Starts the Celery worker thread.Nz(Celery worker thread is already running._workerz --hostname=z-Q)targetr.z,Celery worker thread started with hostname: )ris_aliveloggerinfor replica_tagrmainrrrB threadingThread worker_mainstart)r-r/ unique_id worker_argss r2start_consumerz)CeleryTaskProcessorAdapter.start_consumers   *t/B/K/K/M/M * KKB C C C F'))5 #'9> ? ?I ? ?  1$/ 1 1  L #  (.9(    !!### R4;P R R     r3$@timeoutc|j|jstddStd|jjdd|jg|j||jrt d|d ntd d|_dS) zESignals the Celery worker to shut down and waits for it to terminate.Nz$Celery worker thread is not running.z+Sending shutdown signal to Celery worker...shutdownzcelery@) destination)rz&Worker thread did not terminate after z seconds.z!Celery worker thread has stopped.) rryrzr{rcontrol broadcastrjoinwarning)r-rs r2 stop_consumerz(CeleryTaskProcessorAdapter.stop_consumers   &d.A.J.J.L.L & KK> ? ? ? F ABBB ## %Ft/D%F%F$G $       111   ' ' ) ) = NNVGVVV W W W W KK; < < <"r3ctd|jjtddS)NzShutting down Celery worker...z"Celery worker shutdown complete...)rzr{rrrr-s r2rz#CeleryTaskProcessorAdapter.shutdownsE 4555 ""$$$ 899999r3cD|jj|dS)z Cancels a task synchronously. Only supported for Redis and RabbitMQ brokers by Celery. More details can be found here: https://docs.celeryq.dev/en/stable/userguide/workers.html#revoke-revoking-tasks N)rrrevoke)r-rrs r2cancel_task_syncz+CeleryTaskProcessorAdapter.cancel_task_syncs#   )))))r3cb|jjS)z Returns the metrics of the Celery worker synchronously. More details can be found here: https://docs.celeryq.dev/en/stable/reference/celery.app.control.html#celery.app.control.Inspect.stats )rrinspectstatsrs r2get_metrics_syncz+CeleryTaskProcessorAdapter.get_metrics_syncs' y ((**00222r3c>|jjS)a Checks the health of the Celery worker synchronously. Returns a list of dictionaries, each containing the worker name and a dictionary with the health status. Example: [{'celery@192.168.1.100': {'ok': 'pong'}}] More details can be found here: https://docs.celeryq.dev/en/stable/reference/celery.app.control.html#celery.app.control.Control.ping )rrpingrs r2health_check_syncz,CeleryTaskProcessorAdapter.health_check_syncsy %%'''r3senderrrr.r/einfoc td|dt||t|jt|t|t|g}|jjrV||jj|j|td|d|d|jjddSdS)arHandle task failures and route them to appropriate dead letter queues. This method is called when a task fails after all retry attempts have been exhausted. It logs the failure and moves the task to failed_task_queue Args: sender: The task object that failed task_id: Unique identifier of the failed task args: Positional arguments passed to the task kwargs: Keyword arguments passed to the task einfo: Exception info object containing exception details and traceback **kw: Additional keyword arguments passed by Celery z#Task failure detected for task_id: z , einfo: zTask z& failed after max retries. Exception: z. Moved it to the z queue.N) rzr{str exceptionrrL_move_task_to_queuer\error)r-rrrr.r/rkwdlq_argss r2rOz/CeleryTaskProcessorAdapter._handle_task_failures ,  P' P PCJJ P P     II KK JJ   < .   $ $ 3     LLMMMuMM`d`laDMMM       r3r\rfmessageexcc td|d|dt||jjrO||jj|||t|t|t|gdSdS)aHandle unknown or unregistered tasks received by Celery. This method is called when Celery receives a task that it doesn't recognize (i.e., a task that hasn't been registered with the Celery app). These tasks are moved to the unprocessable task queue if configured. Args: sender: The Celery app or worker that detected the unknown task name: Name of the unknown task id: Task ID of the unknown task message: The raw message received for the unknown task exc: The exception raised when trying to process the unknown task **kwargs: Additional context information from Celery z'Unknown task detected by Celery. Name: z, ID: z, Exc: N)rzr{rrrMr)r-rr\rfrrr/s r2rPz/CeleryTaskProcessorAdapter._handle_unknown_task-s.  Wd W W" W WSQTXX W W    < 5   $ $ :LLHHKK        r3rBrlc  td|d|d||j|||dS#t$r*}td|d|d||d}~wwxYw)z4Helper function to move a task to a specified queue.z Moving task: z to queue: z, args: )r\r>r.zFailed to move task: z , error: N)rzr{rrjr^r)r-rBrlr.es r2rz.CeleryTaskProcessorAdapter._move_task_to_queueUs  KKP PPjPP$PP    I              LLV VVjVVSTVV   G  sAA A8%A33A8)N)NN)r)NNNNN)%__name__ __module__ __qualname____doc__r__annotations__rrrr~rrrr rintr#rSrcrrortrfloatrrrrrrrrrOrPlistr __classcell__)r1s@r2rr(s LLL    15NHY-.555&*hsm***;;;;206R;<;rs ,,,,,,,,,,,,55555555)))))) +*****  , - -#00($@!%  W{{{{{!5{{{{{r3