Feat: Support production deployment mode by using gunicorn gevent.

Author: hzy@singhand.com

api/db/db_models.py

@@ -417,6 +417,7 @@ class Meta:
 
 
 @DB.connection_context()
+@DB.lock("init_database_tables", 60)
 def init_database_tables(alter_fields=[]):
     members = inspect.getmembers(sys.modules[__name__], inspect.isclass)
     table_objs = []
@@ -428,7 +429,7 @@ def init_database_tables(alter_fields=[]):
             if not obj.table_exists():
                 logging.debug(f"start create table {obj.__name__}")
                 try:
-                    obj.create_table()
+                    obj.create_table(safe=True)
                     logging.debug(f"create table success: {obj.__name__}")
                 except Exception as e:
                     logging.exception(e)

api/wsgi.py

@@ -14,14 +14,26 @@
 #  limitations under the License.
 #
 
+# Gevent monkey patching - must be done before importing other modules
+import os
+if os.environ.get('GUNICORN_WORKER_CLASS') == 'gevent':
+    from gevent import monkey
+    monkey.patch_all()
+
+    # Import gevent for greenlet spawning
+    import gevent
+    from gevent import spawn
+    USE_GEVENT = True
+else:
+    USE_GEVENT = False
+
 from api.utils.log_utils import initRootLogger
 from plugin import GlobalPluginManager
 
 # Initialize logging first
 initRootLogger("ragflow_server")
 
 import logging
-import os
 import signal
 import threading
 import uuid
@@ -40,9 +52,15 @@
 from rag.settings import print_rag_settings
 from rag.utils.redis_conn import RedisDistributedLock
 
-# Global stop event and executor for background tasks
-stop_event = threading.Event()
-background_executor = None
+# Global variables for background tasks
+if USE_GEVENT:
+    stop_event = None
+    background_executor = None
+    background_greenlet = None
+else:
+    stop_event = threading.Event()
+    background_executor = None
+    background_greenlet = None
 
 RAGFLOW_DEBUGPY_LISTEN = int(os.environ.get('RAGFLOW_DEBUGPY_LISTEN', "0"))
 
@@ -52,24 +70,50 @@ def update_progress():
     lock_value = str(uuid.uuid4())
     redis_lock = RedisDistributedLock("update_progress", lock_value=lock_value, timeout=60)
     logging.info(f"update_progress lock_value: {lock_value}")
-    while not stop_event.is_set():
-        try:
-            if redis_lock.acquire():
-                DocumentService.update_progress()
+
+    if USE_GEVENT:
+        # Use gevent sleep and loop for greenlet compatibility
+        while True:
+            try:
+                if redis_lock.acquire():
+                    DocumentService.update_progress()
+                    redis_lock.release()
+                gevent.sleep(6)  # Use gevent.sleep instead of stop_event.wait
+            except Exception:
+                logging.exception("update_progress exception")
+                redis_lock.release()
+                break
+    else:
+        # Traditional threading approach
+        while not stop_event.is_set():
+            try:
+                if redis_lock.acquire():
+                    DocumentService.update_progress()
+                    redis_lock.release()
+                stop_event.wait(6)
+            except Exception:
+                logging.exception("update_progress exception")
+            finally:
                 redis_lock.release()
-            stop_event.wait(6)
-        except Exception:
-            logging.exception("update_progress exception")
-        finally:
-            redis_lock.release()
 
 
 def signal_handler(sig, frame):
     """Handle shutdown signals gracefully"""
-    logging.info("Received interrupt signal, shutting down...")
-    stop_event.set()
-    if background_executor:
-        background_executor.shutdown(wait=False)
+    logging.info("Received shutdown signal, stopping background tasks...")
+
+    if USE_GEVENT:
+        # Kill the background greenlet
+        global background_greenlet
+        if background_greenlet and not background_greenlet.dead:
+            background_greenlet.kill()
+    else:
+        # Traditional threading approach
+        stop_event.set()
+        if hasattr(background_executor, 'shutdown'):
+            background_executor.shutdown(wait=False)
+
+    logging.info("Background tasks stopped")
+    exit(0)
 
 
 def initialize_ragflow():
@@ -113,8 +157,16 @@ def initialize_ragflow():
     signal.signal(signal.SIGTERM, signal_handler)
 
     # Start background progress update task
-    background_executor = ThreadPoolExecutor(max_workers=1)
-    background_executor.submit(update_progress)
+    if USE_GEVENT:
+        # Use gevent spawn for greenlet-based execution
+        global background_greenlet
+        background_greenlet = spawn(update_progress)
+        logging.info("Started document progress update task in gevent mode")
+    else:
+        # Use thread pool for traditional threading
+        background_executor = ThreadPoolExecutor(max_workers=1)
+        background_executor.submit(update_progress)
+        logging.info("Started document progress update task in threading mode")
 
     logging.info("RAGFlow WSGI application initialized successfully in production mode")
 

conf/gunicorn.conf.py

@@ -1,22 +1,27 @@
 # Gunicorn configuration file for RAGFlow production deployment
 import multiprocessing
 import os
+from api import settings
+from rag.utils.infinity_conn import InfinityConnection
+from graphrag import search as kg_search
+from rag.nlp import search
 
 # Server socket
 bind = f"{os.environ.get('RAGFLOW_HOST_IP', '0.0.0.0')}:{os.environ.get('RAGFLOW_HOST_PORT', '9380')}"
 backlog = 2048
 
 # Worker processes
 workers = int(os.environ.get('GUNICORN_WORKERS', min(multiprocessing.cpu_count() * 2 + 1, 8)))
-worker_class = 'sync'
+worker_class = 'gevent'
+
+# Gevent-specific settings
 worker_connections = 1000
-timeout = 120
-keepalive = 2
-max_requests = 1000
-max_requests_jitter = 100
+timeout = 300
+keepalive = 10
+max_requests = 2000
+max_requests_jitter = 200
 
-# Restart workers after this many requests, to help prevent memory leaks
-preload_app = True
+preload_app = False
 
 # Logging
 accesslog = '-'
@@ -64,6 +69,10 @@ def pre_fork(server, worker):
 def post_fork(server, worker):
     """Called just after a worker has been forked."""
     server.log.info("RAGFlow worker spawned (pid: %s)", worker.pid)
+    if os.environ.get("DOC_ENGINE") == "infinity":
+        settings.docStoreConn = InfinityConnection()
+        settings.retrievaler = search.Dealer(settings.docStoreConn)
+        settings.kg_retrievaler = kg_search.KGSearch(settings.docStoreConn)
 
 def worker_abort(worker):
     """Called when a worker received the SIGABRT signal."""

docker/.env

@@ -181,3 +181,6 @@ REGISTER_ENABLED=1
 # COMPOSE_PROFILES=infinity,sandbox
 #    - For OpenSearch:
 # COMPOSE_PROFILES=opensearch,sandbox
+
+# Gunicorn setting
+GUNICORN_WORKERS=4

docker/entrypoint.sh

@@ -190,9 +190,12 @@ if [[ "${ENABLE_WEBSERVER}" -eq 1 ]]; then
     RAGFLOW_HOST=${RAGFLOW_HOST_IP:-0.0.0.0}
     RAGFLOW_PORT=${RAGFLOW_HOST_PORT:-9380}
     GUNICORN_WORKERS=${GUNICORN_WORKERS:-4}
-    GUNICORN_TIMEOUT=${GUNICORN_WORKERS:120}
+    GUNICORN_TIMEOUT=${GUNICORN_TIMEOUT:-120}
 
-    echo "Gunicorn config: Workers=${GUNICORN_WORKERS}, Host=${RAGFLOW_HOST}, Port=${RAGFLOW_PORT}"
+    # Set environment variable for gevent worker class
+    export GUNICORN_WORKER_CLASS=gevent
+
+    echo "Gunicorn config: Workers=${GUNICORN_WORKERS}, Host=${RAGFLOW_HOST}, Port=${RAGFLOW_PORT}, Worker Class=gevent"
 
     # Check if gunicorn config file exists and use it, otherwise use command line options
     if [[ -f "/ragflow/conf/gunicorn.conf.py" ]]; then
@@ -202,7 +205,7 @@ if [[ "${ENABLE_WEBSERVER}" -eq 1 ]]; then
         echo "Using Gunicorn command line configuration..."
         # Start gunicorn with our WSGI application
         exec gunicorn --workers ${GUNICORN_WORKERS} \
-                       --worker-class sync \
+                       --worker-class gevent \
                        --worker-connections 1000 \
                        --max-requests 1000 \
                        --max-requests-jitter 100 \

pyproject.toml

@@ -127,6 +127,7 @@ dependencies = [
     "opensearch-py==2.7.1",
     "pluginlib==0.9.4",
     "gunicorn>=21.2.0,<22.0.0",
+    "gevent>=23.9.0,<24.0.0",
 ]
 
 [project.optional-dependencies]