Feat/marker pdf chunking

benchmarks/marker/marker_page_chunking.md

@@ -0,0 +1,74 @@
+# Marker Page Chunking Benchmark
+
+## Objective
+
+Measure the impact of `MARKER_CHUNK_SIZE` on PDF parsing speed and GPU memory usage. Page chunking splits large PDFs into fixed-size page ranges and dispatches them across all available Marker workers in parallel, rather than sending the entire file to a single worker.
+
+## Setup
+
+### Worker configuration
+
+| Variable | Value |
+|----------|-------|
+| `MARKER_MAX_PROCESSES` | 5 |
+| `MARKER_MAX_TASKS_PER_CHILD` | 100 |
+
+### Disabled features
+
+The following features were disabled to isolate the measurement to pure PDF parsing time, avoiding bias from downstream processing:
+
+```bash
+CONTEXTUAL_RETRIEVAL=false
+IMAGE_CAPTIONING=false
+VDB_ENABLE_INSERTION=false
+```
+
+### Dataset
+
+| Metric | Min | Max | Mean | Std |
+|--------|-----|-----|------|-----|
+| Pages per PDF | 11 | 40 | 21.0 | 8.3 |
+| Size per PDF (MB) | 0.03 | 25.81 | 1.89 | 3.84 |
+
+## Results
+
+Tested with `MARKER_CHUNK_SIZE` values of 10, 20, and 30
+
+| Chunk size | Parsing duration | Max GPU spike (GB) | Spike duration |
+|------------|-----------------|---------------------|----------------|
+| 30 | 16m 27s | 2.2 - 4.0 | 5s to ~2 min (file-dependent) |
+| 20 | 16m 44s | 2.2 - 3 | 5s to ~1 min |
+| 10 | 17m 29s | 1.9 - 2.5 | 5s - 30s |
+
+## Analysis
+
+### Speed
+
+These results are at **equal number of workers** (`MARKER_MAX_PROCESSES=5`). All chunk sizes perform similarly (~16-17 min), and with smaller chunks the workload per worker actually *increases* since each worker handles more tasks (more chunks to process).
+
+The main advantage of chunking is therefore not raw speed at fixed worker count, but the ability to **scale the number of workers without risking OOM**. By keeping per-worker memory spikes low and spike duration low aswell, chunking allows safely increasing `MARKER_MAX_PROCESSES`, which is where the real speed gains come from.
+
+The GPU memory constraint can be estimated as:
+
+```text
+available_gpu_mem >= max_spike * num_workers + marker_model_gpu_size + other_gpu_processes
+```
+
+With a chunk size of 10 (max spike ~2.2 GB) you can fit more workers in the same GPU budget than with unchunked processing (where spikes can reach 4+ GB per worker).
+
+### GPU memory
+
+Smaller chunk sizes produce **lower and shorter memory spikes**:
+
+- **Chunk size 10**: Safest option. Peak stays around 1.9-2.5 GB with spikes lasting at most 30 seconds. Memory drops quickly since 10 pages are processed fast.
+- **Chunk size 20-30**: Spikes can reach 3-4 GB and persist for up to 2 minutes, increasing the risk of OOM when multiple workers hit peak usage simultaneously.
+
+### Spike behavior
+
+- Spikes occur primarily during Marker's **"Recognizing text"** phase.
+- For chunk sizes 20 and 30, spike duration can extend to ~2 minutes with peaks between 2.3 and 4 GB. This raises OOM risk when several processes spike concurrently.
+- Files with complex or non-searchable text are the worst case: Marker spends significantly more time in the recognition phase (layout recognition, text recognition, OCR error detection, bbox detection), keeping memory elevated for longer.
+
+## Recommendation
+
+A chunk size of **10** offers the best trade-off: parsing speed is comparable to larger chunks, while GPU memory stays controlled with short-lived spikes. This reduces OOM risk in production, especially under concurrent load.

charts/openrag-stack/values.yaml

@@ -308,7 +308,6 @@ env:
     XDG_CACHE_HOME: "/app/model_weights"
     MARKER_MAX_TASKS_PER_CHILD: "10"
     MARKER_MAX_PROCESSES: "15"
-    MARKER_MIN_PROCESSES: "3"
     MARKER_POOL_SIZE: "3"
     MARKER_NUM_GPUS: "0.6"
     TRANSCRIBER_BASE_URL: "http://{{ .Release.Name }}-whisper-engine-service/v1"

conf/config.yaml

@@ -47,7 +47,8 @@ embedder:
   max_model_len: 8192
 
 # --- Vector Database (Milvus) ---
-# Env: VDB_HOST, VDB_PORT, VDB_CONNECTOR_NAME, VDB_COLLECTION_NAME, VDB_HYBRID_SEARCH
+# Env: VDB_HOST, VDB_PORT, VDB_CONNECTOR_NAME, VDB_COLLECTION_NAME, VDB_HYBRID_SEARCH,
+#      VDB_ENABLE_INSERTION
 vectordb:
   host: milvus
   port: 19530
@@ -177,14 +178,16 @@ loader:
     md: MarkdownLoader
 
   # Env: MARKER_MAX_TASKS_PER_CHILD, MARKER_POOL_SIZE, MARKER_MAX_PROCESSES,
-  #      MARKER_MIN_PROCESSES, MARKER_NUM_GPUS, MARKER_TIMEOUT, MARKER_PDFTEXT_WORKERS
-  marker_max_tasks_per_child: 10
+  #      MARKER_NUM_GPUS, MARKER_TIMEOUT, MARKER_PDFTEXT_WORKERS, MARKER_CHUNK_SIZE
+  marker_max_tasks_per_child: 20
   marker_pool_size: 1
   marker_max_processes: 2
-  marker_min_processes: 1
   marker_num_gpus: 0.01
   marker_timeout: 3600
   marker_pdftext_workers: 2
+  # Split large PDFs into chunks of this many pages for parallel processing across workers.
+  # 0 = no chunking (process entire PDF in one worker).
+  marker_chunk_size: 10
 
   # Env: DOCLING_NUM_GPUS, DOCLING_POOL_SIZE, DOCLING_MAX_TASKS_PER_WORKER
   docling_num_gpus: 0.01

docs/content/docs/documentation/deploy_ray_cluster.md

@@ -23,7 +23,6 @@ RAY_MAX_TASKS_PER_WORKER=5
 # PDF specific resources when using marker
 MARKER_MAX_TASKS_PER_CHILD=10
 MARKER_MAX_PROCESSES=5 # Number of subprocesses <-> Number of concurrent pdfs per worker
-MARKER_MIN_PROCESSES=3 # Minimum number of subprocesses available before triggering a process pool reset.
 MARKER_POOL_SIZE=1 # Number of workers (typically 1 worker per cluster node)
 MARKER_NUM_GPUS=0.6
 

docs/content/docs/documentation/env_vars.md

@@ -36,10 +36,18 @@ The `MarkerLoader` is the default PDF parsing engine. It can be configured using
 |----------|------|---------|-------------|
 | `MARKER_POOL_SIZE` | int | 1 | Number of workers (typically 1 worker per cluster node) |
 | `MARKER_MAX_PROCESSES` | int | 2 | Number of subprocesses <-> Number of concurrent PDFs per worker (to increase depending on your available GPU resources)|
-| `MARKER_MAX_TASKS_PER_CHILD` | int | 10 | Number of tasks a child (PDF worker) has to process before it gets restarted to clean up memory leaks |
-| `MARKER_MIN_PROCESSES` | int | 1 | Minimum number of subprocesses available before triggering a process pool reset |
+| `MARKER_MAX_TASKS_PER_CHILD` | int | 20 | Number of tasks a child (PDF worker) has to process before it gets restarted to clean up memory leaks |
 | `MARKER_TIMEOUT` | int | 3600 | Timeout in seconds for marker processes |
 | `MARKER_PDFTEXT_WORKERS` | int | 2 | Number of PDF text extractor workers inside marker. |
+| `MARKER_CHUNK_SIZE` | int | 10 | Split large PDFs into chunks of this many pages for parallel processing across workers. Use <= 0 to deactivate chunking. |
+
+:::note[Page chunking with `MARKER_CHUNK_SIZE`]
+Enabling page chunking allows processing large PDFs **significantly faster** by dispatching page ranges to all available workers in parallel rather than sending the entire file to a single worker. The main benefit is the ability to safely scale `MARKER_MAX_PROCESSES` without risking OOM.
+
+It also **reduces per-worker GPU memory spikes** on large files. With a reasonable chunk size (around 10 pages), spikes are shorter and lower, making it safer to run more concurrent workers. See `benchmarks/marker/marker_page_chunking.md` for measured results.
+
+**NB:** Consider increasing `MARKER_MAX_TASKS_PER_CHILD` when using page chunking, as worker utilization increases significantly and you may observe frequent subprocess restarts with the default value.
+:::
 
 
 
@@ -165,6 +173,7 @@ The vector database stores embeddings and is configured using the following envi
 | `VDB_CONNECTOR_NAME` | str | milvus | Connector/driver to use for the vector DB. Currently only `milvus` is implemented |
 | `VDB_COLLECTION_NAME` | str | vdb_test | Name of the collection storing embeddings |
 |`VDB_HYBRID_SEARCH`| `bool` | true |To activate hybrid search (semantic similarity + Keyword search)|
+| `VDB_ENABLE_INSERTION` | bool | true | Enable or disable vector database insertion. When disabled, documents are processed but not inserted into Milvus. Useful for testing. |
 
 These variables can be overridden when using an external vector database service.
 

openrag/components/indexer/loaders/pdf_loaders/marker.py

@@ -4,6 +4,7 @@
 import time
 from pathlib import Path
 
+import pypdfium2
 import ray
 import torch
 from config import load_config
@@ -103,27 +104,36 @@ def _worker_init(model_dict):
     def _process_pdf(file_path, config):
         global worker_model_dict
 
+        page_range = config.get("page_range")
+        if page_range is not None:
+            label = f"[p{page_range[0]}-{page_range[-1]}]"
+        else:
+            label = "(all pages)"
+
         try:
-            logger.debug("Processing PDF", path=file_path)
+            logger.debug("Processing PDF", path=file_path, label=label)
             converter = PdfConverter(
                 artifact_dict=worker_model_dict,
                 config=config,
             )
             render = converter(file_path)
             return render
         except Exception as e:
-            logger.exception("Error processing PDF", path=file_path, error=str(e))
+            logger.exception("Error processing PDF", path=file_path, label=label, error=str(e))
             raise
         finally:
             gc.collect()
             if torch.cuda.is_available():
                 torch.cuda.empty_cache()
                 torch.cuda.ipc_collect()
 
-    async def process_pdf(self, file_path: str):
+    async def process_pdf(self, file_path: str, page_range: list[int] | None = None):
         from concurrent.futures import TimeoutError as FuturesTimeoutError
 
         converter_config = self.converter_config.copy()
+        if page_range is not None:
+            converter_config["page_range"] = page_range
+
         loop = asyncio.get_event_loop()
         timeout = self.config.loader.marker_timeout
 
@@ -142,12 +152,11 @@ def run_with_timeout():
         result = await loop.run_in_executor(None, run_with_timeout)
         return result.markdown, result.images
 
-    def get_current_pool_size(self):
-        # ProcessPoolExecutor manages worker lifecycle automatically
-        # Return count of alive worker processes
-        if self.executor is None:
-            return 0
-        return len([p for p in self.executor._processes.values() if p.is_alive()])
+    def is_pool_broken(self):
+        # ProcessPoolExecutor auto-replaces dead/finished workers on next
+        # submit(), so counting live processes is unreliable and unnecessary.
+        # Only a None or shut-down executor requires reinitialization.
+        return self.executor is None or bool(getattr(self.executor, "_broken", False))
 
     def __del__(self):
         """Clean up ProcessPoolExecutor on actor destruction"""
@@ -166,7 +175,6 @@ def __init__(self):
 
         self.logger = get_logger()
         self.config = load_config()
-        self.min_processes = self.config.loader.marker_min_processes
         self.max_processes = self.config.loader.marker_max_processes
         self.pool_size = self.config.loader.marker_pool_size
         self.actors = [MarkerWorker.remote() for _ in range(self.pool_size)]
@@ -181,34 +189,94 @@ def __init__(self):
             f"{self.pool_size * self.max_processes} PDF concurrency"
         )
 
+    @staticmethod
+    def _get_page_count(file_path: str) -> int:
+        with pypdfium2.PdfDocument(file_path) as pdf:
+            return len(pdf)
+
+    @staticmethod
+    def _create_chunks(page_count: int, chunk_size: int) -> list[tuple[list[int], str]]:
+        if page_count <= chunk_size:
+            return [(list(range(page_count)), f"({page_count}p)")]
+        chunks = []
+        for start in range(0, page_count, chunk_size):
+            end = min(start + chunk_size, page_count)
+            page_range = list(range(start, end))
+            label = f"[p{start}-{end - 1}]"
+            chunks.append((page_range, label))
+        return chunks
+
     async def ensure_worker_pool_healthy(self, worker):
-        current_alive = await worker.get_current_pool_size.remote()
-        if current_alive < self.min_processes:
-            self.logger.warning(
-                f"Only {current_alive}/{self.min_processes} worker processes alive. Reinitializing pool..."
+        from components.ray_utils import call_ray_actor_with_timeout
+
+        timeout = self.config.loader.marker_timeout
+        broken = await call_ray_actor_with_timeout(
+            worker.is_pool_broken.remote(),
+            timeout=timeout,
+            task_description="MarkerWorker pool health check",
+        )
+        if broken:
+            self.logger.warning("Worker ProcessPoolExecutor is broken. Reinitializing pool...")
+            await call_ray_actor_with_timeout(
+                worker.setup_mp.remote(),
+                timeout=timeout,
+                task_description="MarkerWorker pool reset",
             )
-            await worker.setup_mp.remote()
 
-    async def process_pdf(self, file_path: str):
+    async def _process_chunk(self, file_path: str, page_range: list[int] | None, label: str):
+        """Acquire a worker slot, process a PDF chunk, and release the slot."""
         from components.ray_utils import call_ray_actor_with_timeout
 
-        # Wait until any slot is free
         worker = await self._queue.get()
-        if worker:
-            self.logger.info("MarkerWorker allocated")
-            # Ensure the worker pool is healthy
-            await self.ensure_worker_pool_healthy(worker)
         try:
+            self.logger.info(f"MarkerWorker allocated for {label}")
+            await self.ensure_worker_pool_healthy(worker)
             timeout = self.config.loader.marker_timeout
-            future = worker.process_pdf.remote(file_path)
+            future = worker.process_pdf.remote(file_path, page_range=page_range)
             return await call_ray_actor_with_timeout(
                 future,
                 timeout=timeout,
-                task_description=f"MarkerPool PDF processing ({file_path})",
+                task_description=f"MarkerPool PDF {label} ({file_path})",
             )
         finally:
             await self._queue.put(worker)
-            self.logger.debug("MarkerWorker returned to pool")
+            self.logger.debug(f"MarkerWorker returned to pool for {label}")
+
+    async def process_pdf(self, file_path: str):
+        chunk_size = self.config.loader.marker_chunk_size
+
+        if chunk_size <= 0:
+            return await self._process_chunk(file_path, page_range=None, label="(all pages)")
+
+        page_count = self._get_page_count(file_path)
+        chunks = self._create_chunks(page_count, chunk_size)
+
+        if len(chunks) == 1:
+            page_range, label = chunks[0]
+            return await self._process_chunk(file_path, page_range=None, label=label)
+
+        self.logger.info(
+            f"Splitting {page_count}-page PDF into {len(chunks)} chunks of ~{chunk_size} pages for parallel processing"
+        )
+
+        tasks = [asyncio.create_task(self._process_chunk(file_path, page_range, label)) for page_range, label in chunks]
+        try:
+            results = await asyncio.gather(*tasks)
+        except Exception:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            raise
+
+        # Reassemble: concatenate markdown in order, merge image dicts
+        all_markdown = []
+        all_images = {}
+        for markdown, images in results:
+            all_markdown.append(markdown)
+            all_images.update(images)
+
+        combined_markdown = "\n\n".join(all_markdown)
+        return combined_markdown, all_images
 
 
 class MarkerLoader(BaseLoader):

openrag/config/loader.py

@@ -46,6 +46,7 @@
     ("VDB_CONNECTOR_NAME", "vectordb.connector_name", str),
     ("VDB_COLLECTION_NAME", "vectordb.collection_name", str),
     ("VDB_HYBRID_SEARCH", "vectordb.hybrid_search", bool),
+    ("VDB_ENABLE_INSERTION", "vectordb.enable", bool),
     # RDB (Postgres)
     ("POSTGRES_HOST", "rdb.host", str),
     ("POSTGRES_PORT", "rdb.port", int),
@@ -87,10 +88,10 @@
     ("MARKER_MAX_TASKS_PER_CHILD", "loader.marker_max_tasks_per_child", int),
     ("MARKER_POOL_SIZE", "loader.marker_pool_size", int),
     ("MARKER_MAX_PROCESSES", "loader.marker_max_processes", int),
-    ("MARKER_MIN_PROCESSES", "loader.marker_min_processes", int),
     ("MARKER_NUM_GPUS", "loader.marker_num_gpus", float),
     ("MARKER_TIMEOUT", "loader.marker_timeout", int),
     ("MARKER_PDFTEXT_WORKERS", "loader.marker_pdftext_workers", int),
+    ("MARKER_CHUNK_SIZE", "loader.marker_chunk_size", int),
     ("DOCLING_NUM_GPUS", "loader.docling_num_gpus", float),
     ("DOCLING_POOL_SIZE", "loader.docling_pool_size", int),
     ("DOCLING_MAX_TASKS_PER_WORKER", "loader.docling_max_tasks_per_worker", int),

openrag/config/models.py

@@ -331,13 +331,13 @@ class LoaderConfig(ConfigMixin):
     mimetypes: MimetypesConfig = Field(default_factory=MimetypesConfig)
     local_whisper: LocalWhisperConfig = Field(default_factory=LocalWhisperConfig)
     file_loaders: FileLoadersConfig = Field(default_factory=FileLoadersConfig)
-    marker_max_tasks_per_child: int = 10
+    marker_max_tasks_per_child: int = 20
     marker_pool_size: int = 1
     marker_max_processes: int = 2
-    marker_min_processes: int = 1
     marker_num_gpus: float = 0.01
     marker_timeout: int = 3600
     marker_pdftext_workers: int = 2
+    marker_chunk_size: int = 10
     docling_num_gpus: float = Field(default=0.01, ge=0)
     docling_pool_size: int = Field(default=1, ge=1)
     docling_max_tasks_per_worker: int = Field(default=2, ge=1)

pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "langchain-openai>=0.3.7",
     "loguru>=0.7.3",
     "marker-pdf>=0.2.17",
+    "pypdfium2>=4.30.0",
     "pydub>=0.25.1",
     "pymupdf4llm>=0.0.17",
     "spire-doc>=13.1.0",