Update celery forwarder to use greenlets instead of processes

Author: dmchoiboi

model-engine/model_engine_server/inference/forwarding/celery_forwarder.py

@@ -3,7 +3,9 @@
 from datetime import datetime, timedelta
 from typing import Any, Dict, Optional, TypedDict, Union
 
+from aiohttp import ClientConnectionError
 from celery import Celery, Task, states
+from gevent import monkey
 from model_engine_server.common.constants import DEFAULT_CELERY_TASK_NAME, LIRA_CELERY_TASK_NAME
 from model_engine_server.common.dtos.model_endpoints import BrokerType
 from model_engine_server.common.dtos.tasks import EndpointPredictV1Request
@@ -23,7 +25,8 @@
 from model_engine_server.inference.infra.gateways.datadog_inference_monitoring_metrics_gateway import (
     DatadogInferenceMonitoringMetricsGateway,
 )
-from requests import ConnectionError
+
+monkey.patch_all()
 
 logger = make_logger(logger_name())
 
@@ -132,7 +135,7 @@ def after_return(
         base=ErrorHandlingTask,
         name=LIRA_CELERY_TASK_NAME,
         track_started=True,
-        autoretry_for=(ConnectionError,),
+        autoretry_for=(ClientConnectionError,),
     )
     def exec_func(payload, arrival_timestamp, *ignored_args, **ignored_kwargs):
         if len(ignored_args) > 0:
@@ -177,12 +180,7 @@ def start_celery_service(
         concurrency=concurrency,
         loglevel="INFO",
         optimization="fair",
-        # Don't use pool="solo" so we can send multiple concurrent requests over
-        # Historically, pool="solo" argument fixes the known issues of celery and some of the libraries.
-        # Particularly asyncio and torchvision transformers. This isn't relevant since celery-forwarder
-        # is quite lightweight
-        # TODO: we should probably use eventlet or gevent for the pool, since
-        # the forwarder is nearly the most extreme example of IO bound.
+        pool="gevent",
     )
     worker.start()