o 1 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) PublicAPIZ worker_poolZworker_concurrencyZtask_ignore_resultZtask_acks_lateZtask_reject_on_worker_lostalpha)Z 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)selfrargskwargsZconflicting_keys __class__d/home/app/PaddleOCR-VL-test/.venv_paddleocr/lib/python3.10/site-packages/ray/serve/task_processor.pyr6s    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)backendZbrokerthreadsFTdirect)ZexchangeZ exchange_typeZ routing_key*queue)Z task_queuesZ task_routes)broker_transport_options)rr queue_namerZ backend_urlZ broker_urlrCELERY_WORKER_POOLCELERY_WORKER_CONCURRENCYCELERY_TASK_IGNORE_RESULTCELERY_TASK_ACKS_LATE!CELERY_TASK_REJECT_ON_WORKER_LOSTrupdateconffailed_task_queue_nameunprocessable_task_queue_namer1rconnect_handle_task_failurer_handle_unknown_task)r$r+Zapp_configurationZ queue_configr)r)r* initializeNsX   z%CeleryTaskProcessorAdapter.initializecCsptfd|jjidddd}|jjjr||jjj|r+|jjdd|i||dS|jjdi||dS)N max_retriesT<F)Z autoretry_forZ retry_kwargsZ retry_backoffZretry_backoff_maxZ retry_jitternamer)) Exceptionrr@rZtask_custom_configr8rtask)r$funcrBZ task_optionsr)r)r*register_task_handles  z/CeleryTaskProcessorAdapter.register_task_handlereturncKs:|jj|f|||jjd|}t|j|jt|jdS)N)r%r&r0)idstatusZ created_atresult) r send_taskrr2rrHrItimerJ)r$ task_namer%r&optionsZ task_responser)r)r*enqueue_task_syncsz,CeleryTaskProcessorAdapter.enqueue_task_synccCs |j|}t|j|j|jdS)N)rHrJrI)rZ AsyncResultrrHrJrI)r$task_idZ task_detailsr)r)r*get_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._Zworkerz --hostname=z-Q)targetr%z,Celery worker thread started with hostname: )ris_aliveloggerinfor Z replica_tagrmainrrr2 threadingThreadZ worker_mainstart)r$r& unique_idZ worker_argsr)r)r*start_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.) rrTrUrVrcontrol broadcastrjoinwarning)r$r^r)r)r* stop_consumers     z(CeleryTaskProcessorAdapter.stop_consumercCs$td|jjtddS)NzShutting down Celery worker...z"Celery worker shutdown complete...)rUrVrrar_r$r)r)r*r_s  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)rraZrevoke)r$rPr)r)r*cancel_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 )rrainspectstatsrfr)r)r*get_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 )rraZpingrfr)r)r*health_check_syncs z,CeleryTaskProcessorAdapter.health_check_syncsenderrPr%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) rUrVstr exceptionrr:_move_task_to_queuerBerror)r$rlrPr%r&rmkwZdlq_argsr)r)r*r=s&z/CeleryTaskProcessorAdapter._handle_task_failurerBrHmessageexcc 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)rUrVrnrr;rp)r$rlrBrHrsrtr&r)r)r*r>-sz/CeleryTaskProcessorAdapter._handle_unknown_taskr2rMc 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: )rBr0r%zFailed to move task: z , error: N)rUrVrrKrCrq)r$r2rMr%er)r)r*rpUs  z.CeleryTaskProcessorAdapter._move_task_to_queue)N)NN)r])NNNNN)%__name__ __module__ __qualname____doc__r__annotations__rrrrXrYrrnr rintrr?rFrrOrQr\floatrer_rgrrrjrrkr=r>listrp __classcell__r)r)r'r*r(sl   =   / (r)"loggingrXrLtypingrrrrZceleryrZcelery.signalsrrZ ray.server Zray.serve._private.constantsr r Zray.serve.schemar r rrZray.util.annotationsr getLoggerrUr3r4r5r6r7r!rr)r)r)r*s0