o $iz2@sddlZddlZddlZddlmZmZmZmZddlm Z ddl m Z m Z ddl mZddlmZmZddlmZmZmZmZddlmZeeZd Zd Zd Zd Zd ZeeeeegZ eddGdddeZ!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 s\eZdZUdZeed<eed<dZee j ed<dZ ee ed<e Zeed<deffd d Ze fd efd d Zd4ddZ d5defddZdefddZddZd6defddZddZddZdee effdd Zdeefd!d"Z     d7d#ed$e d%ed&ed'ef d(d)Z     d7d#ed*e d+e d,ed-ef d.d/Z!d0e d1e d%e"fd2d3Z#Z$S)8CeleryTaskProcessorAdapterz 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_concurrencyconfigcshtj|i|t|jtstd|jjr/t|jjtt @}|r/t dt |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)selfrargskwargsconflicting_keys __class__U/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/serve/task_processor.pyr 6s    z#CeleryTaskProcessorAdapter.__init__consumer_concurrencyc Cs6t|jj|jjj|jjjd|_tdt|t dt dt di}|jjj r)| |jjj |jj ||jj|jjd|jjdi}|jjrP|jjd|jjd||jj<|jjrb|jjd|jjd||jj<|jjj |dd|jjiid |jjjdur|jjj |jjjd |jjrt|j|jjrt|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_namer>rconnect_handle_task_failurer_handle_unknown_task)r*r2app_configuration queue_configr0r0r1 initializeNsX   z%CeleryTaskProcessorAdapter.initializecCsptfd|jjidddd}|jjjr||jjj|r+|jjdd|i||dS|jjdi||dS)N max_retriesT<F) autoretry_for retry_kwargs retry_backoffretry_backoff_max retry_jitternamer0) ExceptionrrQr"task_custom_configrGrtask)r*funcrX task_optionsr0r0r1register_task_handles  z/CeleryTaskProcessorAdapter.register_task_handlereturncKs:|jj|f|||jjd|}t|j|jt|jdS)N)r+r,r;)idstatus created_atresult) r send_taskrr?rr`ratimerc)r* task_namer+r,options task_responser0r0r1enqueue_task_syncsz,CeleryTaskProcessorAdapter.enqueue_task_synccCs |j|}t|j|j|jdS)N)r`rcra)r AsyncResultrr`rcra)r*task_id task_detailsr0r0r1get_task_status_syncs z/CeleryTaskProcessorAdapter.get_task_status_synccKs|jdur|jrtddStj}|jjd||_dd|jd|j j g}t j |jj |fd|_|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_tagrmainrrr? threadingThread worker_mainstart)r*r, unique_id worker_argsr0r0r1start_consumers$    z)CeleryTaskProcessorAdapter.start_consumer$@timeoutcCs|jdus |jstddStd|jjjdd|jgd|jj|d|jr:t 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)r~z&Worker thread did not terminate after z seconds.z!Celery worker thread has stopped.) rrqrrrsrcontrol broadcastrjoinwarning)r*r~r0r0r1 stop_consumers     z(CeleryTaskProcessorAdapter.stop_consumercCs$td|jjtddS)NzShutting down Celery worker...z"Celery worker shutdown complete...)rrrsrrrr*r0r0r1rs  z#CeleryTaskProcessorAdapter.shutdowncCs|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*rkr0r0r1cancel_task_syncsz+CeleryTaskProcessorAdapter.cancel_task_synccCs|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 )rrinspectstatsrr0r0r1get_metrics_syncsz+CeleryTaskProcessorAdapter.get_metrics_synccCs |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 )rrpingrr0r0r1health_check_syncs z,CeleryTaskProcessorAdapter.health_check_syncsenderrkr+r,einfoc Kstd|dt||t|jt|t|t|g}|jjr>||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) rrrsstr exceptionrrI_move_task_to_queuerXerror)r*rrkr+r,rkwdlq_argsr0r0r1rLs&z/CeleryTaskProcessorAdapter._handle_task_failurerXr`messageexcc KsXtd|d|dt||jjr*||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)rrrsrrrJr)r*rrXr`rrr,r0r0r1rM-sz/CeleryTaskProcessorAdapter._handle_unknown_taskr?rfc Csnztd|d|d||jj|||dWdSty6}ztd|d|d||d}~ww)z4Helper function to move a task to a specified queue.z Moving task: z to queue: z, args: )rXr;r+zFailed to move task: z , error: N)rrrsrrdrYr)r*r?rfr+er0r0r1rUs  z.CeleryTaskProcessorAdapter._move_task_to_queue)N)NN)r})NNNNN)%__name__ __module__ __qualname____doc__r__annotations__rrrrvrwrrr rintr rPr^rrirmr|floatrrrrrrrrrLrMlistr __classcell__r0r0r.r1r(sl   =   / (r)"loggingrvretypingrrrrceleryrcelery.signalsrr ray.server ray.serve._private.constantsr r ray.serve.schemar r rrray.util.annotationsr getLoggerrrrBrCrDrErFr'rr0r0r0r1s0