marker_api/gunicorn.conf.py

import os
import multiprocessing
import threading

port = int(os.environ.get("PORT", "8001"))
workers = int(os.environ.get("GUNICORN_WORKERS", "3"))
threads = int(os.environ.get("GUNICORN_THREADS", "4"))
timeout = int(os.environ.get("GUNICORN_TIMEOUT", "600"))

bind = f"0.0.0.0:{port}"
workers = min(workers, 8)

worker_class = "gthread"
graceful_timeout = 60

accesslog = "-"
errorlog = "-"
loglevel = os.environ.get("GUNICORN_LOGLEVEL", "info")

# Do NOT preload — each worker starts its own background thread and loads models
preload_app = False

max_requests = 1000
max_requests_jitter = 100

tempdir = "/app/uploads"


def post_fork(server, worker):
    """Start background worker thread after fork (filesystem-based store works across workers)."""
    t = threading.Thread(target=_bg_worker, daemon=True)
    t.start()


def _bg_worker():
    import app as marker_api
    from request_store import _job_queue, _job_cond, get_request, complete_request, fail_request, get_result

    while True:
        import time
        import traceback
        with _job_cond:
            while not _job_queue:
                _job_cond.wait()
            job = _job_queue.pop(0)
        request_id, endpoint, raw_bytes, filename, opts = job
        try:
            if endpoint == "convert":
                result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts)
            elif endpoint == "ocr":
                from deepseek_ocr import ocr_pdf
                result = ocr_pdf(raw_bytes, max_pages=opts.get("max_pages"), page_range=opts.get("page_range"))
            elif endpoint == "extract":
                conv_result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts)
                if conv_result["success"]:
                    import json
                    schema_json = opts.get("page_schema") or "{}"
                    try:
                        schema = json.loads(schema_json) if isinstance(schema_json, str) else schema_json
                    except json.JSONDecodeError:
                        schema = {}
                    extraction = marker_api._apply_extraction_schema(conv_result["output"], schema)
                    result = {**conv_result, "extraction": extraction}
                else:
                    result = conv_result
            elif endpoint == "segment":
                conv_result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts)
                if conv_result["success"]:
                    import json
                    schema_json = opts.get("segmentation_schema") or "{}"
                    try:
                        schema = json.loads(schema_json) if isinstance(schema_json, str) else schema_json
                    except json.JSONDecodeError:
                        schema = {}
                    segments = marker_api._apply_segmentation_schema(conv_result["output"], schema)
                    result = {**conv_result, "segments": segments}
                else:
                    result = conv_result
            elif endpoint == "table_rec":
                conv_result = marker_api.convert_with_ocr_backend(raw_bytes, filename, **opts, output_format="json")
                tables = marker_api._extract_tables(conv_result)
                result = {**conv_result, "tables": tables}
            else:
                result = {"success": False, "error": f"Unknown endpoint: {endpoint}"}
            if result["success"]:
                complete_request(request_id, result)
            else:
                fail_request(request_id, result.get("error", "Unknown error"))
        except Exception as exc:
            traceback.print_exc()
            fail_request(request_id, str(exc))