From 1bcc1307be1d198f4b5c7b08744922e44b291ec0 Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Tue, 24 Feb 2026 02:41:19 -0500 Subject: [PATCH 1/9] perf: lazy page rendering in convert_pdf_to_image to reduce peak memory --- unstructured_inference/inference/layout.py | 33 ++++++++++++---------- 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/unstructured_inference/inference/layout.py b/unstructured_inference/inference/layout.py index 5b956d2a..74f37c6a 100644 --- a/unstructured_inference/inference/layout.py +++ b/unstructured_inference/inference/layout.py @@ -422,8 +422,13 @@ def convert_pdf_to_image( raise ValueError("output_folder must be specified if path_only is true") if filename is None and file is None: raise ValueError("Either filename or file must be provided") + if output_folder: + assert Path(output_folder).exists() + assert Path(output_folder).is_dir() + with _pdfium_lock, pdfium.PdfDocument(filename or file, password=password) as pdf: images: dict[int, Image.Image] = {} + filenames: list[str] = [] if dpi is None: dpi = inference_config.PDF_RENDER_DPI scale = dpi / 72.0 @@ -441,24 +446,22 @@ def convert_pdf_to_image( optimize_mode="print", ) try: - images[i] = bitmap.to_pil() + pil_image = bitmap.to_pil() finally: bitmap.close() + if output_folder: + fn: str = os.path.join(str(output_folder), f"page_{i}.png") + pil_image.save(fn, format="PNG", compress_level=1, optimize=False) + filenames.append(fn) + if not path_only: + images[i] = pil_image + else: + images[i] = pil_image finally: page.close() - if not output_folder: + if output_folder and path_only: + return filenames + if output_folder: return list(images.values()) - else: - # Save images to output_folder - filenames: list[str] = [] - assert Path(output_folder).exists() - assert Path(output_folder).is_dir() - for i, image in images.items(): - fn: str = os.path.join(str(output_folder), f"page_{i}.png") - image.save(fn, format="PNG", compress_level=1, optimize=False) - filenames.append(fn) - if path_only: - return filenames - images_values: list[Image.Image] = list(images.values()) - return images_values + return list(images.values()) From a09174a312bcac485060493275c5c4022995dad3 Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Wed, 25 Feb 2026 02:27:15 -0500 Subject: [PATCH 2/9] test: add coverage for convert_pdf_to_image branches --- .../inference/test_layout.py | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/test_unstructured_inference/inference/test_layout.py b/test_unstructured_inference/inference/test_layout.py index dabfa2c5..2844229b 100644 --- a/test_unstructured_inference/inference/test_layout.py +++ b/test_unstructured_inference/inference/test_layout.py @@ -591,6 +591,25 @@ def test_exposed_pdf_image_dpi(pdf_image_dpi, expected, monkeypatch): assert mock_from_image.call_args[0][0].height == expected +def test_convert_pdf_to_image_no_output_folder(): + result = layout.convert_pdf_to_image(filename="sample-docs/loremipsum.pdf", dpi=72) + assert len(result) == 1 + assert isinstance(result[0], Image.Image) + + +def test_convert_pdf_to_image_output_folder_returns_images(tmp_path): + result = layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=tmp_path, + path_only=False, + ) + assert len(result) == 1 + assert isinstance(result[0], Image.Image) + saved = list(tmp_path.glob("*.png")) + assert len(saved) == 1 + + @pytest.mark.parametrize( ("filename", "img_num", "should_complete"), [ From cf00e7d6c17354ee334e4d16b1bce3066b7e30bc Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Mon, 16 Mar 2026 16:36:27 -0600 Subject: [PATCH 3/9] chore: bump version to 1.5.3 and update changelog --- CHANGELOG.md | 5 +++++ unstructured_inference/__version__.py | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a3fb7ab6..f4555b4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,8 @@ +## 1.5.3 + +### Enhancement +- Lazy page rendering in `convert_pdf_to_image` to reduce peak memory from O(N pages) to O(1 page) + ## 1.5.2 ### Fix diff --git a/unstructured_inference/__version__.py b/unstructured_inference/__version__.py index 928be505..e375e78d 100644 --- a/unstructured_inference/__version__.py +++ b/unstructured_inference/__version__.py @@ -1 +1 @@ -__version__ = "1.5.2" # pragma: no cover +__version__ = "1.5.3" # pragma: no cover From 9bf79efc3a0fed33856b49b8442c7651d492b575 Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Thu, 26 Mar 2026 23:21:14 -0500 Subject: [PATCH 4/9] changelog: move lazy page rendering entry to 1.5.4 --- CHANGELOG.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 156164ba..15530f91 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,10 +1,12 @@ -## 1.5.3 - -- Store routing in LayoutElement +## 1.5.4 ### Enhancement - Lazy page rendering in `convert_pdf_to_image` to reduce peak memory from O(N pages) to O(1 page) +## 1.5.3 + +- Store routing in LayoutElement + ## 1.5.2 ### Fix From cd9954dfc12cd887689df70aa32af592621b0d20 Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Thu, 26 Mar 2026 23:21:55 -0500 Subject: [PATCH 5/9] bump version to 1.5.4 --- unstructured_inference/__version__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/unstructured_inference/__version__.py b/unstructured_inference/__version__.py index e375e78d..e4b9acb5 100644 --- a/unstructured_inference/__version__.py +++ b/unstructured_inference/__version__.py @@ -1 +1 @@ -__version__ = "1.5.3" # pragma: no cover +__version__ = "1.5.4" # pragma: no cover From 1451b72345b3e32213c5a0f33d1b32df90591360 Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Fri, 27 Mar 2026 02:00:11 -0500 Subject: [PATCH 6/9] fix: use per-page locking so disk I/O is not under _pdfium_lock Move from document-level locking to per-page locking: acquire _pdfium_lock only for pdfium operations (page load, render, bitmap conversion, close), release it before saving images to disk. This preserves the O(1) memory benefit for path_only=True (render, save, discard per page) while fixing the concurrency regression where PNG compression and filesystem writes held the global pdfium lock. Add tests for path_only=True hot path and a lock-scope regression test. --- .../inference/test_layout.py | 38 +++++++++ unstructured_inference/inference/layout.py | 77 +++++++++++-------- 2 files changed, 82 insertions(+), 33 deletions(-) diff --git a/test_unstructured_inference/inference/test_layout.py b/test_unstructured_inference/inference/test_layout.py index 2844229b..c700cb32 100644 --- a/test_unstructured_inference/inference/test_layout.py +++ b/test_unstructured_inference/inference/test_layout.py @@ -610,6 +610,44 @@ def test_convert_pdf_to_image_output_folder_returns_images(tmp_path): assert len(saved) == 1 +def test_convert_pdf_to_image_path_only(tmp_path): + result = layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=tmp_path, + path_only=True, + ) + assert len(result) == 1 + assert all(isinstance(p, str) for p in result) + for p in result: + assert os.path.exists(p) + assert p.endswith(".png") + saved = sorted(tmp_path.glob("*.png")) + assert [str(s) for s in saved] == sorted(result) + + +def test_convert_pdf_to_image_save_not_under_pdfium_lock(tmp_path): + """Verify that PIL save (disk I/O) is NOT performed while holding _pdfium_lock.""" + original_save = Image.Image.save + lock_held_during_save = [] + + def spy_save(self, *args, **kwargs): + lock_held_during_save.append(layout._pdfium_lock.locked()) + return original_save(self, *args, **kwargs) + + with patch.object(Image.Image, "save", spy_save): + layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=tmp_path, + path_only=True, + ) + assert lock_held_during_save, "save was never called" + assert not any(lock_held_during_save), ( + "pil_image.save() was called while _pdfium_lock was held" + ) + + @pytest.mark.parametrize( ("filename", "img_num", "should_complete"), [ diff --git a/unstructured_inference/inference/layout.py b/unstructured_inference/inference/layout.py index 6c470db7..c8090282 100644 --- a/unstructured_inference/inference/layout.py +++ b/unstructured_inference/inference/layout.py @@ -430,42 +430,53 @@ def convert_pdf_to_image( assert Path(output_folder).exists() assert Path(output_folder).is_dir() - with _pdfium_lock, pdfium.PdfDocument(filename or file, password=password) as pdf: + if dpi is None: + dpi = inference_config.PDF_RENDER_DPI + scale = dpi / 72.0 + + with _pdfium_lock: + pdf = pdfium.PdfDocument(filename or file, password=password) + n_pages = len(pdf) + + try: images: dict[int, Image.Image] = {} filenames: list[str] = [] - if dpi is None: - dpi = inference_config.PDF_RENDER_DPI - scale = dpi / 72.0 - for i, page in enumerate(pdf, start=1): - try: - if first_page is not None and i < first_page: - continue - if last_page is not None and i > last_page: - break - bitmap = page.render( - scale=scale, - no_smoothtext=False, - no_smoothimage=False, - no_smoothpath=False, - optimize_mode="print", - ) + for i in range(n_pages): + page_num = i + 1 + if first_page is not None and page_num < first_page: + continue + if last_page is not None and page_num > last_page: + break + + with _pdfium_lock: + page = pdf[i] try: - pil_image = bitmap.to_pil() + bitmap = page.render( + scale=scale, + no_smoothtext=False, + no_smoothimage=False, + no_smoothpath=False, + optimize_mode="print", + ) + try: + pil_image = bitmap.to_pil() + finally: + bitmap.close() finally: - bitmap.close() - if output_folder: - fn: str = os.path.join(str(output_folder), f"page_{i}.png") - pil_image.save(fn, format="PNG", compress_level=1, optimize=False) - filenames.append(fn) - if not path_only: - images[i] = pil_image - else: - images[i] = pil_image - finally: - page.close() - - if output_folder and path_only: + page.close() + + if output_folder: + fn: str = os.path.join(str(output_folder), f"page_{page_num}.png") + pil_image.save(fn, format="PNG", compress_level=1, optimize=False) + filenames.append(fn) + if not path_only: + images[page_num] = pil_image + else: + images[page_num] = pil_image + finally: + with _pdfium_lock: + pdf.close() + + if path_only: return filenames - if output_folder: - return list(images.values()) return list(images.values()) From 63277761f33a49e22e68a5939665fbff1979199a Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Fri, 27 Mar 2026 02:01:07 -0500 Subject: [PATCH 7/9] style: ruff format test_layout.py --- test_unstructured_inference/inference/test_layout.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/test_unstructured_inference/inference/test_layout.py b/test_unstructured_inference/inference/test_layout.py index c700cb32..bd602ef9 100644 --- a/test_unstructured_inference/inference/test_layout.py +++ b/test_unstructured_inference/inference/test_layout.py @@ -643,9 +643,7 @@ def spy_save(self, *args, **kwargs): path_only=True, ) assert lock_held_during_save, "save was never called" - assert not any(lock_held_during_save), ( - "pil_image.save() was called while _pdfium_lock was held" - ) + assert not any(lock_held_during_save), "pil_image.save() was called while _pdfium_lock was held" @pytest.mark.parametrize( From a7358a8e2ea58e46c2da71ba27cf4ad5fe86d9cc Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Fri, 27 Mar 2026 02:07:07 -0500 Subject: [PATCH 8/9] test: add concurrency regression test for save outside lock Two threads call convert_pdf_to_image with a deliberately slow save(). Verifies that disk writes can overlap rather than being serialized behind _pdfium_lock. --- .../inference/test_layout.py | 63 +++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/test_unstructured_inference/inference/test_layout.py b/test_unstructured_inference/inference/test_layout.py index bd602ef9..1e1f51dd 100644 --- a/test_unstructured_inference/inference/test_layout.py +++ b/test_unstructured_inference/inference/test_layout.py @@ -646,6 +646,69 @@ def spy_save(self, *args, **kwargs): assert not any(lock_held_during_save), "pil_image.save() was called while _pdfium_lock was held" +def test_convert_pdf_to_image_concurrent_saves_not_serialized(tmp_path): + """Two concurrent callers must be able to overlap their disk writes. + + Inject a slow save() and verify both threads can save concurrently rather + than being serialized behind _pdfium_lock. + """ + import threading + import time + + save_delay = 0.3 + original_save = Image.Image.save + concurrent_saves = threading.Event() + save_entered = threading.Event() + + def slow_save(self, *args, **kwargs): + save_entered.set() + # Wait briefly for the other thread's save to also start. If saves are + # serialized under the lock this will time out. + concurrent_saves.wait(timeout=save_delay) + if not concurrent_saves.is_set(): + # We're the first thread in; the other may not have started yet. + # Signal so the other thread can proceed, then do a short sleep + # to give it a window to overlap. + concurrent_saves.set() + time.sleep(0.05) + return original_save(self, *args, **kwargs) + + errors: list[str] = [] + + def run(folder): + try: + layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=folder, + path_only=True, + ) + except Exception as exc: + errors.append(str(exc)) + + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + + with patch.object(Image.Image, "save", slow_save): + t1 = threading.Thread(target=run, args=(dir_a,)) + t2 = threading.Thread(target=run, args=(dir_b,)) + t1.start() + # Wait for t1 to reach its save before starting t2 + save_entered.wait(timeout=5) + t2.start() + t1.join(timeout=10) + t2.join(timeout=10) + + assert not errors, f"threads raised: {errors}" + assert concurrent_saves.is_set(), ( + "saves were serialized — concurrent_saves event was never set by overlap" + ) + assert list(dir_a.glob("*.png")), "thread A produced no output" + assert list(dir_b.glob("*.png")), "thread B produced no output" + + @pytest.mark.parametrize( ("filename", "img_num", "should_complete"), [ From 9d2bd70f35910c29ad2275a2dee735167c40b0ff Mon Sep 17 00:00:00 2001 From: Kevin Turcios Date: Fri, 27 Mar 2026 02:16:08 -0500 Subject: [PATCH 9/9] test: fix concurrency test and add render-interleave, multi-page, and error-isolation tests - Replace broken Event-based concurrency test with a Barrier that correctly detects whether two threads are inside save() simultaneously. - Add test that thread B can render while thread A is blocked in save(). - Add test that two threads produce all 10 pages from a multi-page PDF. - Add test that a failure in one thread does not block or poison the other. --- .../inference/test_layout.py | 193 +++++++++++++++--- 1 file changed, 170 insertions(+), 23 deletions(-) diff --git a/test_unstructured_inference/inference/test_layout.py b/test_unstructured_inference/inference/test_layout.py index 1e1f51dd..b2859a5d 100644 --- a/test_unstructured_inference/inference/test_layout.py +++ b/test_unstructured_inference/inference/test_layout.py @@ -649,28 +649,23 @@ def spy_save(self, *args, **kwargs): def test_convert_pdf_to_image_concurrent_saves_not_serialized(tmp_path): """Two concurrent callers must be able to overlap their disk writes. - Inject a slow save() and verify both threads can save concurrently rather - than being serialized behind _pdfium_lock. + Uses a threading.Barrier to verify both threads are inside save() + simultaneously. If saves are serialized under _pdfium_lock, the second + thread can never reach save() while the first is there, so the barrier + times out and the test fails. """ import threading - import time - save_delay = 0.3 original_save = Image.Image.save - concurrent_saves = threading.Event() - save_entered = threading.Event() - - def slow_save(self, *args, **kwargs): - save_entered.set() - # Wait briefly for the other thread's save to also start. If saves are - # serialized under the lock this will time out. - concurrent_saves.wait(timeout=save_delay) - if not concurrent_saves.is_set(): - # We're the first thread in; the other may not have started yet. - # Signal so the other thread can proceed, then do a short sleep - # to give it a window to overlap. - concurrent_saves.set() - time.sleep(0.05) + barrier = threading.Barrier(2, timeout=5) + overlap_detected = threading.Event() + + def barrier_save(self, *args, **kwargs): + try: + barrier.wait() + overlap_detected.set() + except threading.BrokenBarrierError: + pass return original_save(self, *args, **kwargs) errors: list[str] = [] @@ -691,24 +686,176 @@ def run(folder): dir_a.mkdir() dir_b.mkdir() - with patch.object(Image.Image, "save", slow_save): + with patch.object(Image.Image, "save", barrier_save): t1 = threading.Thread(target=run, args=(dir_a,)) t2 = threading.Thread(target=run, args=(dir_b,)) t1.start() - # Wait for t1 to reach its save before starting t2 - save_entered.wait(timeout=5) t2.start() t1.join(timeout=10) t2.join(timeout=10) assert not errors, f"threads raised: {errors}" - assert concurrent_saves.is_set(), ( - "saves were serialized — concurrent_saves event was never set by overlap" + assert overlap_detected.is_set(), ( + "saves were serialized under _pdfium_lock — threads could not overlap" ) assert list(dir_a.glob("*.png")), "thread A produced no output" assert list(dir_b.glob("*.png")), "thread B produced no output" +def test_render_can_proceed_while_other_thread_saves(tmp_path): + """Thread B can acquire _pdfium_lock and render while thread A is in save(). + + Blocks thread A inside save() (outside the lock), then starts thread B. + If B completes entirely while A is still blocked, the lock was not held + during save — rendering and saving can overlap across callers. + """ + import threading + + original_save = Image.Image.save + a_in_save = threading.Event() + b_done = threading.Event() + + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + + def gated_save(self, *args, **kwargs): + fp = str(args[0]) if args else "" + if str(dir_a) in fp: + a_in_save.set() + b_done.wait(timeout=5) + return original_save(self, *args, **kwargs) + + errors: list[str] = [] + + def run(folder, done_event=None): + try: + layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=folder, + path_only=True, + ) + except Exception as exc: + errors.append(str(exc)) + finally: + if done_event: + done_event.set() + + with patch.object(Image.Image, "save", gated_save): + t_a = threading.Thread(target=run, args=(dir_a,)) + t_b = threading.Thread(target=run, args=(dir_b, b_done)) + t_a.start() + a_in_save.wait(timeout=5) + # A is now blocked in save (outside lock). B should render + save freely. + t_b.start() + t_b.join(timeout=10) + t_a.join(timeout=10) + + assert not errors, f"threads raised: {errors}" + assert b_done.is_set(), "Thread B could not complete while A was saving" + assert list(dir_a.glob("*.png")), "thread A produced no output" + assert list(dir_b.glob("*.png")), "thread B produced no output" + + +def test_multi_page_concurrent_output_complete(tmp_path): + """Two threads processing a multi-page PDF both produce correct, complete output.""" + import threading + + errors: list[str] = [] + + def run(folder): + try: + layout.convert_pdf_to_image( + filename="sample-docs/loremipsum_multipage.pdf", + dpi=72, + output_folder=folder, + path_only=True, + ) + except Exception as exc: + errors.append(str(exc)) + + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + + t1 = threading.Thread(target=run, args=(dir_a,)) + t2 = threading.Thread(target=run, args=(dir_b,)) + t1.start() + t2.start() + t1.join(timeout=60) + t2.join(timeout=60) + + assert not errors, f"threads raised: {errors}" + a_files = sorted(dir_a.glob("*.png")) + b_files = sorted(dir_b.glob("*.png")) + assert len(a_files) == 10, f"thread A produced {len(a_files)} files, expected 10" + assert len(b_files) == 10, f"thread B produced {len(b_files)} files, expected 10" + for i in range(1, 11): + assert (dir_a / f"page_{i}.png").exists(), f"thread A missing page_{i}.png" + assert (dir_b / f"page_{i}.png").exists(), f"thread B missing page_{i}.png" + + +def test_error_in_one_thread_does_not_block_other(tmp_path): + """If one thread fails mid-processing, the other still completes.""" + import threading + + original_save = Image.Image.save + + dir_a = tmp_path / "a" + dir_b = tmp_path / "b" + dir_a.mkdir() + dir_b.mkdir() + + def failing_save(self, *args, **kwargs): + fp = str(args[0]) if args else "" + if str(dir_a) in fp: + raise OSError("simulated disk failure") + return original_save(self, *args, **kwargs) + + a_error: list[Exception] = [] + b_result: list[str] = [] + b_error: list[Exception] = [] + + def run_a(): + try: + layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=dir_a, + path_only=True, + ) + except Exception as exc: + a_error.append(exc) + + def run_b(): + try: + result = layout.convert_pdf_to_image( + filename="sample-docs/loremipsum.pdf", + dpi=72, + output_folder=dir_b, + path_only=True, + ) + b_result.extend(result) + except Exception as exc: + b_error.append(exc) + + with patch.object(Image.Image, "save", failing_save): + t_a = threading.Thread(target=run_a) + t_b = threading.Thread(target=run_b) + t_a.start() + t_b.start() + t_a.join(timeout=10) + t_b.join(timeout=10) + + assert a_error, "Thread A should have failed" + assert not b_error, f"Thread B should have succeeded: {b_error}" + assert b_result, "Thread B produced no result" + assert list(dir_b.glob("*.png")), "Thread B produced no output files" + + @pytest.mark.parametrize( ("filename", "img_num", "should_complete"), [