From 34e564cd7de9847c63093b8d3601968341b22148 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Thu, 17 Dec 2020 01:57:05 -0800 Subject: [PATCH 01/27] Use queue.Queue instead of multiprocessing.Queue in threaded mode --- src/ocrmypdf/_concurrent.py | 39 +++++++++++++++++++++++++------------ 1 file changed, 27 insertions(+), 12 deletions(-) diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index 177ffe0a..f38150dd 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -9,20 +9,23 @@ import logging import logging.handlers import multiprocessing import os +import queue import signal import sys import threading from contextlib import suppress from multiprocessing import Pool as ProcessPool from multiprocessing.dummy import Pool as ThreadPool -from typing import Callable, Iterable, Optional +from typing import Callable, Iterable, Optional, Union from tqdm import tqdm from ocrmypdf.exceptions import InputFileError +Queue = Union[multiprocessing.Queue, queue.Queue] -def log_listener(queue): + +def log_listener(q: Queue): """Listen to the worker processes and forward the messages to logging For simplicity this is a thread rather than a process. Only one process @@ -34,7 +37,7 @@ def log_listener(queue): while True: try: - record = queue.get() + record = q.get() if record is None: break logger = logging.getLogger(record.name) @@ -50,7 +53,7 @@ def process_sigbus(*args): raise InputFileError("A worker process lost access to an input file") -def process_init(queue, user_init, loglevel): +def process_init(q: Queue, user_init: Callable[[], None], loglevel): """Initialize a process pool worker""" # Ignore SIGINT (our parent process will kill us gracefully) @@ -61,23 +64,24 @@ def process_init(queue, user_init, loglevel): signal.signal(signal.SIGBUS, process_sigbus) # Reconfigure the root logger for this process to send all messages to a queue - h = logging.handlers.QueueHandler(queue) + h = logging.handlers.QueueHandler(q) root = logging.getLogger() root.setLevel(loglevel) root.handlers = [] root.addHandler(h) - if user_init: - user_init() + user_init() + return -def thread_init(_queue, user_init, _loglevel): +def thread_init(_queue: Queue, user_init: Callable[[], None], _loglevel): # As a thread, block SIGBUS so the main thread deals with it... with suppress(AttributeError): # Windows and Cygwin do not have pthread_sigmask or SIGBUS signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGBUS}) - if user_init: - user_init() + + user_init() + return def exec_progress_pool( @@ -90,15 +94,26 @@ def exec_progress_pool( task_arguments: Optional[Iterable] = None, task_finished: Optional[Callable] = None, ): - log_queue: multiprocessing.Queue = multiprocessing.Queue(-1) - listener = threading.Thread(target=log_listener, args=(log_queue,)) if use_threads: + log_queue = queue.Queue(-1) pool_class = ThreadPool initializer = thread_init else: + log_queue = multiprocessing.Queue(-1) pool_class = ProcessPool initializer = process_init + + if not task_initializer: + + def _noop(): + return + + task_initializer = _noop + + # Regardless of whether we use_threads for worker processes, the log_listener + # must be a thread + listener = threading.Thread(target=log_listener, args=(log_queue,)) listener.start() with tqdm(**tqdm_kwargs) as pbar: From 26b4d9bb4b4bf508ed8d22419eff8a8e64512c67 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Tue, 22 Dec 2020 00:44:59 -0800 Subject: [PATCH 02/27] Refactor concurrency so that it is pluggable However, this may not be the best idea because it involves global state that could be overridden by a parallel call to ocrmypdf.ocr. --- src/ocrmypdf/_concurrent.py | 152 +++--------------- src/ocrmypdf/_plugin_manager.py | 10 +- src/ocrmypdf/_sync.py | 6 +- src/ocrmypdf/builtin_plugins/concurrency.py | 164 ++++++++++++++++++++ src/ocrmypdf/pdfinfo/info.py | 2 +- src/ocrmypdf/pluginspec.py | 43 ++++- 6 files changed, 242 insertions(+), 135 deletions(-) create mode 100644 src/ocrmypdf/builtin_plugins/concurrency.py diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index f38150dd..8dc3a6c3 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -4,84 +4,20 @@ # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. - -import logging -import logging.handlers -import multiprocessing -import os -import queue -import signal -import sys -import threading -from contextlib import suppress -from multiprocessing import Pool as ProcessPool -from multiprocessing.dummy import Pool as ThreadPool -from typing import Callable, Iterable, Optional, Union - -from tqdm import tqdm - -from ocrmypdf.exceptions import InputFileError - -Queue = Union[multiprocessing.Queue, queue.Queue] +from typing import Callable, Iterable, Optional -def log_listener(q: Queue): - """Listen to the worker processes and forward the messages to logging - - For simplicity this is a thread rather than a process. Only one process - should actually write to sys.stderr or whatever we're using, so if this is - made into a process the main application needs to be directed to it. - - See https://docs.python.org/3/howto/logging-cookbook.html#logging-to-a-single-file-from-multiple-processes - """ - - while True: - try: - record = q.get() - if record is None: - break - logger = logging.getLogger(record.name) - logger.handle(record) - except Exception: # pylint: disable=broad-except - import traceback # pylint: disable=import-outside-toplevel - - print("Logging problem", file=sys.stderr) - traceback.print_exc(file=sys.stderr) - - -def process_sigbus(*args): - raise InputFileError("A worker process lost access to an input file") - - -def process_init(q: Queue, user_init: Callable[[], None], loglevel): - """Initialize a process pool worker""" - - # Ignore SIGINT (our parent process will kill us gracefully) - signal.signal(signal.SIGINT, signal.SIG_IGN) - - # Install SIGBUS handler (so our parent process can abort somewhat gracefully) - with suppress(AttributeError): # Windows and Cygwin do not have SIGBUS - signal.signal(signal.SIGBUS, process_sigbus) - - # Reconfigure the root logger for this process to send all messages to a queue - h = logging.handlers.QueueHandler(q) - root = logging.getLogger() - root.setLevel(loglevel) - root.handlers = [] - root.addHandler(h) - - user_init() +def _task_noop(*_args, **_kwargs): return -def thread_init(_queue: Queue, user_init: Callable[[], None], _loglevel): - # As a thread, block SIGBUS so the main thread deals with it... - with suppress(AttributeError): - # Windows and Cygwin do not have pthread_sigmask or SIGBUS - signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGBUS}) +def _model(**_kwargs): + raise RuntimeError("Parallel executor not set up") - user_init() - return + +def set_execution_model(model): + global _model + _model = model def exec_progress_pool( @@ -89,64 +25,22 @@ def exec_progress_pool( use_threads: bool, max_workers: int, tqdm_kwargs: dict, - task_initializer: Optional[Callable] = None, - task: Optional[Callable] = None, + worker_initializer: Optional[Callable] = None, + task: Callable, task_arguments: Optional[Iterable] = None, task_finished: Optional[Callable] = None, ): + if not worker_initializer: + worker_initializer = _task_noop + if not task_finished: + task_finished = _task_noop - if use_threads: - log_queue = queue.Queue(-1) - pool_class = ThreadPool - initializer = thread_init - else: - log_queue = multiprocessing.Queue(-1) - pool_class = ProcessPool - initializer = process_init - - if not task_initializer: - - def _noop(): - return - - task_initializer = _noop - - # Regardless of whether we use_threads for worker processes, the log_listener - # must be a thread - listener = threading.Thread(target=log_listener, args=(log_queue,)) - listener.start() - - with tqdm(**tqdm_kwargs) as pbar: - pool = pool_class( - processes=max_workers, - initializer=initializer, - initargs=(log_queue, task_initializer, logging.getLogger("").level), - ) - try: - results = pool.imap_unordered(task, task_arguments) - for result in results: - if task_finished: - task_finished(result, pbar) - else: - pbar.update() - except KeyboardInterrupt: - # Terminate pool so we exit instantly - pool.terminate() - # Don't try listener.join() here, will deadlock - raise - except Exception: - if not os.environ.get("PYTEST_CURRENT_TEST", ""): - # Unless inside pytest, exit immediately because no one wants - # to wait for child processes to finalize results that will be - # thrown away. Inside pytest, we want child processes to exit - # cleanly so that they output an error messages or coverage data - # we need from them. - pool.terminate() - raise - finally: - # Terminate log listener - log_queue.put_nowait(None) - pool.close() - pool.join() - - listener.join() + _model( + use_threads=use_threads, + max_workers=max_workers, + tqdm_kwargs=tqdm_kwargs, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + ) diff --git a/src/ocrmypdf/_plugin_manager.py b/src/ocrmypdf/_plugin_manager.py index 510ae65b..76e12bd7 100644 --- a/src/ocrmypdf/_plugin_manager.py +++ b/src/ocrmypdf/_plugin_manager.py @@ -32,7 +32,11 @@ class OcrmypdfPluginManager(pluggy.PluginManager): """ def __init__( - self, *args, plugins: List[Union[str, Path]], builtins: bool = True, **kwargs, + self, + *args, + plugins: List[Union[str, Path]], + builtins: bool = True, + **kwargs, ): self.__init_args = args self.__init_kwargs = kwargs @@ -88,7 +92,9 @@ class OcrmypdfPluginManager(pluggy.PluginManager): def get_plugin_manager(plugins: List[Union[str, Path]], builtins=True): pm = OcrmypdfPluginManager( - project_name='ocrmypdf', plugins=plugins, builtins=builtins, + project_name='ocrmypdf', + plugins=plugins, + builtins=builtins, ) return pm diff --git a/src/ocrmypdf/_sync.py b/src/ocrmypdf/_sync.py index af32ad8c..1d78f1d9 100644 --- a/src/ocrmypdf/_sync.py +++ b/src/ocrmypdf/_sync.py @@ -18,7 +18,7 @@ from typing import List, NamedTuple, Optional, Tuple import pikepdf import PIL -from ocrmypdf._concurrent import exec_progress_pool +from ocrmypdf._concurrent import exec_progress_pool, set_execution_model from ocrmypdf._graft import OcrGrafter from ocrmypdf._jobcontext import PageContext, PdfContext, cleanup_working_files from ocrmypdf._logging import PageNumberFilter @@ -279,7 +279,7 @@ def exec_concurrent(context: PdfContext): unit_scale=0.5, disable=not options.progress_bar, ), - task_initializer=partial(worker_init, PIL.Image.MAX_IMAGE_PIXELS), + worker_initializer=partial(worker_init, PIL.Image.MAX_IMAGE_PIXELS), task=exec_page_sync, task_arguments=context.get_page_contexts(), task_finished=update_page, @@ -346,6 +346,8 @@ def run_pipeline(options, *, plugin_manager, api=False): pikepdf_enable_mmap() + set_execution_model(plugin_manager.hook.get_parallel_executor()) + try: check_requested_output_file(options) start_input_file, original_filename = create_input_file(options, work_folder) diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py new file mode 100644 index 00000000..6cf56e42 --- /dev/null +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -0,0 +1,164 @@ +# © 2020 James R. Barlow: github.com/jbarlow83 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +# © 2020 James R. Barlow: github.com/jbarlow83 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + + +import logging +import logging.handlers +import multiprocessing +import os +import queue +import signal +import sys +import threading +from contextlib import suppress +from multiprocessing import Pool as ProcessPool +from multiprocessing.dummy import Pool as ThreadPool +from typing import Callable, Iterable, Optional, Union + +from tqdm import tqdm + +from ocrmypdf import hookimpl +from ocrmypdf.exceptions import InputFileError + +Queue = Union[multiprocessing.Queue, queue.Queue] + + +def log_listener(q: Queue): + """Listen to the worker processes and forward the messages to logging + + For simplicity this is a thread rather than a process. Only one process + should actually write to sys.stderr or whatever we're using, so if this is + made into a process the main application needs to be directed to it. + + See https://docs.python.org/3/howto/logging-cookbook.html#logging-to-a-single-file-from-multiple-processes + """ + + while True: + try: + record = q.get() + if record is None: + break + logger = logging.getLogger(record.name) + logger.handle(record) + except Exception: # pylint: disable=broad-except + import traceback # pylint: disable=import-outside-toplevel + + print("Logging problem", file=sys.stderr) + traceback.print_exc(file=sys.stderr) + + +def process_sigbus(*args): + raise InputFileError("A worker process lost access to an input file") + + +def process_init(q: Queue, user_init: Callable[[], None], loglevel): + """Initialize a process pool worker""" + + # Ignore SIGINT (our parent process will kill us gracefully) + signal.signal(signal.SIGINT, signal.SIG_IGN) + + # Install SIGBUS handler (so our parent process can abort somewhat gracefully) + with suppress(AttributeError): # Windows and Cygwin do not have SIGBUS + # Windows and Cygwin do not have pthread_sigmask or SIGBUS + signal.signal(signal.SIGBUS, process_sigbus) + + # Reconfigure the root logger for this process to send all messages to a queue + h = logging.handlers.QueueHandler(q) + root = logging.getLogger() + root.setLevel(loglevel) + root.handlers = [] + root.addHandler(h) + + user_init() + return + + +def thread_init(_queue: Queue, user_init: Callable[[], None], _loglevel): + # As a thread, block SIGBUS so the main thread deals with it... + with suppress(AttributeError): + signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGBUS}) + + user_init() + return + + +def exec_progress_pool( + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_arguments: Optional[Iterable] = None, + task_finished: Callable, +): + + if use_threads: + log_queue = queue.Queue(-1) + pool_class = ThreadPool + initializer = thread_init + else: + log_queue = multiprocessing.Queue(-1) + pool_class = ProcessPool + initializer = process_init + + if not worker_initializer: + + def _noop(): + return + + worker_initializer = _noop + + # Regardless of whether we use_threads for worker processes, the log_listener + # must be a thread + listener = threading.Thread(target=log_listener, args=(log_queue,)) + listener.start() + + with tqdm(**tqdm_kwargs) as pbar: + pool = pool_class( + processes=max_workers, + initializer=initializer, + initargs=(log_queue, worker_initializer, logging.getLogger("").level), + ) + try: + results = pool.imap_unordered(task, task_arguments) + for result in results: + if task_finished: + task_finished(result, pbar) + else: + pbar.update() + except KeyboardInterrupt: + # Terminate pool so we exit instantly + pool.terminate() + # Don't try listener.join() here, will deadlock + raise + except Exception: + if not os.environ.get("PYTEST_CURRENT_TEST", ""): + # Unless inside pytest, exit immediately because no one wants + # to wait for child processes to finalize results that will be + # thrown away. Inside pytest, we want child processes to exit + # cleanly so that they output an error messages or coverage data + # we need from them. + pool.terminate() + raise + finally: + # Terminate log listener + log_queue.put_nowait(None) + pool.close() + pool.join() + + listener.join() + + +@hookimpl +def get_parallel_executor(): + return exec_progress_pool diff --git a/src/ocrmypdf/pdfinfo/info.py b/src/ocrmypdf/pdfinfo/info.py index ae02fb98..a11e64cf 100644 --- a/src/ocrmypdf/pdfinfo/info.py +++ b/src/ocrmypdf/pdfinfo/info.py @@ -623,7 +623,7 @@ def _pdf_pageinfo_concurrent( tqdm_kwargs=dict( total=total, desc="Scanning contents", unit='page', disable=not progbar ), - task_initializer=partial( + worker_initializer=partial( _pdf_pageinfo_sync_init, infile, logging.getLogger('pdfminer').level ), task=_pdf_pageinfo_sync, diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index 9aee6596..538be7a1 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -9,7 +9,7 @@ from abc import ABC, abstractmethod, abstractstaticmethod from argparse import ArgumentParser, Namespace from collections import namedtuple from pathlib import Path -from typing import TYPE_CHECKING, AbstractSet, List, Optional +from typing import TYPE_CHECKING, AbstractSet, Callable, Iterable, List, Optional import pluggy @@ -62,6 +62,47 @@ def check_options(options: Namespace) -> None: """ +class ParallelExecutor(ABC): + @abstractstaticmethod + def __call__( + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_finished: Callable, + task_arguments: Optional[Iterable] = None, + ): + """ + Args: + use_threads: If False, the workload is the sort that will benefit from + running in a multiprocessing context (for example, it uses Python + heavily, and parallelizing it with threads is not expected to be + performant). + max_workers: The maximum number of workers that should be run. + tdqm_kwargs: Arguments to set up the progress bar. + worker_initializer: Called when the worker is initialized, in the worker's + execution context. Must be possible to marshall to the worker. + task: Called when the worker starts a new task, in the worker's execution + context. Must be possible to marshallable to the worker. + task_finished: Called when a worker finishes a task, in the parent's + context. + task_arguments: An iterable that generates a group of parameters for each + task. This runs in the parent's context, but the parameters must be + marshallable to the worker. + """ + + +@hookspec(firstresult=True) +def get_parallel_executor() -> Callable: + """Called to perform parallel execution + + This may be used to replace OCRmyPDF's default parallel execution system + with a third party alternative. For example, you could make OCRmyPDF run in a + distributed environment. + """ + + @hookspec def validate(pdfinfo: 'PdfInfo', options: Namespace) -> None: """Called to give a plugin an opportunity to review *options* and *pdfinfo*. From 6953f324653ea4b5903e6137a142578733df7c70 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 10 Jan 2021 14:20:28 -0800 Subject: [PATCH 03/27] pdfinfo: remove some messy concurrency handling We can cut down on the use of global variables and save opening an extra copy of the Pdf when threaded. --- src/ocrmypdf/_concurrent.py | 2 +- src/ocrmypdf/pdfinfo/info.py | 68 +++++++++++++++++++++--------------- 2 files changed, 40 insertions(+), 30 deletions(-) diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index 8dc3a6c3..87c9656c 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -11,7 +11,7 @@ def _task_noop(*_args, **_kwargs): return -def _model(**_kwargs): +def _model(**_kwargs) -> None: raise RuntimeError("Parallel executor not set up") diff --git a/src/ocrmypdf/pdfinfo/info.py b/src/ocrmypdf/pdfinfo/info.py index a11e64cf..1ee82cf9 100644 --- a/src/ocrmypdf/pdfinfo/info.py +++ b/src/ocrmypdf/pdfinfo/info.py @@ -6,6 +6,7 @@ # file, You can obtain one at http://mozilla.org/MPL/2.0/. +import atexit import logging import re from collections import defaultdict, namedtuple @@ -571,29 +572,33 @@ def simplify_textboxes(miner, textbox_getter) -> Iterator[TextboxInfo]: worker_pdf = None -def _pdf_pageinfo_sync_init(infile: Path, pdfminer_loglevel): +def _pdf_pageinfo_sync_init(pdf: Pdf, infile: Path, pdfminer_loglevel): global worker_pdf # pylint: disable=global-statement pikepdf_enable_mmap() logging.getLogger('pdfminer').setLevel(pdfminer_loglevel) - # If this function is called as a thread initializer, we need a messy hack - # to close worker_pdf. If called as a process, it will be released when the - # process is terminated. - worker_pdf = pikepdf.open(infile) + # If the pdf is not opened, open a copy for our worker process to use + if pdf is None: + worker_pdf = pikepdf.open(infile) + + def on_process_close(): + worker_pdf.close() + + # Close when this process exits + atexit.register(on_process_close) def _pdf_pageinfo_sync(args): - global worker_pdf # pylint: disable=global-statement - pageno, infile, check_pages, detailed_analysis = args - page = PageInfo(worker_pdf, pageno, infile, check_pages, detailed_analysis) + pageno, thread_pdf, infile, check_pages, detailed_analysis = args + pdf = thread_pdf if thread_pdf is not None else worker_pdf + page = PageInfo(pdf, pageno, infile, check_pages, detailed_analysis) return page def _pdf_pageinfo_concurrent( pdf, infile, progbar, max_workers, check_pages, detailed_analysis=False ): - global worker_pdf # pylint: disable=global-statement pages = [None] * len(pdf.pages) def update_pageinfo(result, pbar): @@ -607,7 +612,6 @@ def _pdf_pageinfo_concurrent( max_workers = available_cpu_count() total = len(pdf.pages) - contexts = ((n, infile, check_pages, detailed_analysis) for n in range(total)) use_threads = False # No performance gain if threaded due to GIL n_workers = min(1 + len(pages) // 4, max_workers) @@ -616,25 +620,31 @@ def _pdf_pageinfo_concurrent( # a separate process. use_threads = True - try: - exec_progress_pool( - use_threads=use_threads, - max_workers=n_workers, - tqdm_kwargs=dict( - total=total, desc="Scanning contents", unit='page', disable=not progbar - ), - worker_initializer=partial( - _pdf_pageinfo_sync_init, infile, logging.getLogger('pdfminer').level - ), - task=_pdf_pageinfo_sync, - task_arguments=contexts, - task_finished=update_pageinfo, - ) - finally: - if worker_pdf and use_threads: - assert n_workers == 1, "Should have only one worker when threaded" - # This is messy, but if we ran in thread, close worker_pdf - worker_pdf.close() + # If we use a thread, we can pass the already-open Pdf for them to use + # If we use processes, we pass a None which tells the init function to open its + # own + initial_pdf = pdf if use_threads else None + + contexts = ( + (n, initial_pdf, infile, check_pages, detailed_analysis) for n in range(total) + ) + assert n_workers == 1 if use_threads else n_workers >= 1, "Not multithreadable" + exec_progress_pool( + use_threads=use_threads, + max_workers=n_workers, + tqdm_kwargs=dict( + total=total, desc="Scanning contents", unit='page', disable=not progbar + ), + worker_initializer=partial( + _pdf_pageinfo_sync_init, + initial_pdf, + infile, + logging.getLogger('pdfminer').level, + ), + task=_pdf_pageinfo_sync, + task_arguments=contexts, + task_finished=update_pageinfo, + ) return pages From 173c0d12740cfc9a6731a427670e1a9590effa28 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 10 Jan 2021 14:22:17 -0800 Subject: [PATCH 04/27] concurrency: lock progress pool For API sanity and to communicate expectations. One progress pool at a time is plenty of complexity. --- src/ocrmypdf/builtin_plugins/concurrency.py | 32 +++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py index 6cf56e42..7d7fbbfc 100644 --- a/src/ocrmypdf/builtin_plugins/concurrency.py +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -21,7 +21,7 @@ import sys import threading from contextlib import suppress from multiprocessing import Pool as ProcessPool -from multiprocessing.dummy import Pool as ThreadPool +from multiprocessing.pool import ThreadPool from typing import Callable, Iterable, Optional, Union from tqdm import tqdm @@ -31,6 +31,8 @@ from ocrmypdf.exceptions import InputFileError Queue = Union[multiprocessing.Queue, queue.Queue] +pool_lock = threading.Lock() + def log_listener(q: Queue): """Listen to the worker processes and forward the messages to logging @@ -96,7 +98,7 @@ def exec_progress_pool( use_threads: bool, max_workers: int, tqdm_kwargs: dict, - worker_initializer: Callable, + worker_initializer: Optional[Callable], task: Callable, task_arguments: Optional[Iterable] = None, task_finished: Callable, @@ -118,6 +120,32 @@ def exec_progress_pool( worker_initializer = _noop + with pool_lock: + _exec_progress_pool( + max_workers=max_workers, + tqdm_kwargs=tqdm_kwargs, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + log_queue=log_queue, + pool_class=pool_class, + initializer=initializer, + ) + + +def _exec_progress_pool( + *, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_arguments: Optional[Iterable] = None, + task_finished: Callable, + log_queue: Queue, + pool_class: Callable, + initializer: Callable, +): # Regardless of whether we use_threads for worker processes, the log_listener # must be a thread listener = threading.Thread(target=log_listener, args=(log_queue,)) From 7bccb8c74844af7b1b7f1376a2abc3bdce7bb466 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Tue, 19 Jan 2021 14:15:07 -0800 Subject: [PATCH 05/27] tests: fix concurrency --- tests/test_validation.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_validation.py b/tests/test_validation.py index f7ce5286..deed9769 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -173,7 +173,7 @@ def test_false_action_store_true(): def test_no_progress_bar(progress_bar, resources): opts = make_opts(progress_bar=progress_bar, input_file=(resources / 'trivial.pdf')) plugin_manager = get_plugin_manager(opts.plugins) - with patch('ocrmypdf._concurrent.tqdm', autospec=True) as tqdmpatch: + with patch('ocrmypdf.builtin_plugins.concurrency.tqdm', autospec=True) as tqdmpatch: vd._check_options(opts, plugin_manager, set()) pdfinfo = PdfInfo(opts.input_file, progbar=opts.progress_bar) assert pdfinfo is not None From 5545bae76f986f8d965c0aa0c4787c71e77d8b30 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 16:19:58 -0800 Subject: [PATCH 06/27] lambda_plugin.py: doesn't work since entry point needs to be in package --- misc/lambda_plugin.py | 192 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 misc/lambda_plugin.py diff --git a/misc/lambda_plugin.py b/misc/lambda_plugin.py new file mode 100644 index 00000000..b39270e4 --- /dev/null +++ b/misc/lambda_plugin.py @@ -0,0 +1,192 @@ +# © 2021 James R Barlow: https://github.com/jbarlow83 +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in all +# copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. + +"""Alternate executor to support OCRmyPDF in AWS Lambda""" + + +import logging +import logging.handlers +import multiprocessing +import os +import queue +import signal +import sys +import threading +from contextlib import suppress +from itertools import islice, repeat, takewhile, zip_longest +from multiprocessing import Pipe, Process, process +from multiprocessing.connection import Connection, wait +from typing import Callable, Iterable, Optional, Union +from unittest.mock import Mock + +from ocrmypdf import hookimpl +from ocrmypdf.exceptions import InputFileError + +pool_lock = threading.Lock() + + +def split_every(n: int, iterable: Iterable): + iterator = iter(iterable) + return takewhile(bool, (list(islice(iterator, n)) for _ in repeat(None))) + + +def log_listener(q): + """Listen to the worker processes and forward the messages to logging + + For simplicity this is a thread rather than a process. Only one process + should actually write to sys.stderr or whatever we're using, so if this is + made into a process the main application needs to be directed to it. + + See https://docs.python.org/3/howto/logging-cookbook.html#logging-to-a-single-file-from-multiple-processes + """ + + while True: + try: + record = q.get() + if record is None: + break + logger = logging.getLogger(record.name) + logger.handle(record) + except Exception: # pylint: disable=broad-except + import traceback # pylint: disable=import-outside-toplevel + + print("Logging problem", file=sys.stderr) + traceback.print_exc(file=sys.stderr) + + +def process_sigbus(*args): + raise InputFileError("A worker process lost access to an input file") + + +def process_loop( + conn: Connection, user_init: Callable[[], None], loglevel, task, task_args +): + """Initialize a process pool worker""" + + # Ignore SIGINT (our parent process will kill us gracefully) + signal.signal(signal.SIGINT, signal.SIG_IGN) + + # Install SIGBUS handler (so our parent process can abort somewhat gracefully) + with suppress(AttributeError): # Windows and Cygwin do not have SIGBUS + # Windows and Cygwin do not have pthread_sigmask or SIGBUS + signal.signal(signal.SIGBUS, process_sigbus) + + # Reconfigure the root logger for this process to send all messages to a queue + # h = logging.handlers.QueueHandler(q) + # root = logging.getLogger() + # root.setLevel(loglevel) + # root.handlers = [] + # root.addHandler(h) + + user_init() + + for args in task_args: + try: + result = task(*args) + except Exception as e: + return # for now + else: + conn.send(result) + + conn.close() + return + + +def exec_progress_pool( + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Optional[Callable], + task: Callable, + task_arguments: Optional[Iterable] = None, + task_finished: Callable, +): + + if not worker_initializer: + + def _noop(): + return + + worker_initializer = _noop + + with pool_lock: + _exec_progress_pool( + max_workers=max_workers, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + ) + + +def _exec_progress_pool( + *, + max_workers: int, + worker_initializer: Callable, + task: Callable, + task_arguments: Optional[Iterable] = None, + task_finished: Callable, +): + pbar = Mock() + + task_arguments = list(task_arguments) + grouped_args = list(zip_longest(*list(split_every(max_workers, task_arguments)))) + + processes = [] + connections = [] + for n in range(max_workers): + parent_conn, child_conn = Pipe() + + worker_args = [args for args in grouped_args[n] if args is not None] + process = Process( + target=process_loop, + args=( + child_conn, + worker_initializer, + logging.getLogger("").level, + task, + worker_args, + ), + ) + process.daemon = True + processes.append(process) + connections.append(parent_conn) + + for process in processes: + process.start() + + while connections: + for r in wait(connections): + try: + msg = r.recv() + except EOFError: + connections.remove(r) + else: + if task_finished: + task_finished(msg, pbar) + + for process in processes: + process.join() + + +@hookimpl +def get_parallel_executor(): + return exec_progress_pool From c6a2716cdbe85ca94cc69216ba94b5b04f29cfcf Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 16:21:07 -0800 Subject: [PATCH 07/27] Temporary move into package --- misc/lambda_plugin.py => src/ocrmypdf/_lambda_plugin.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename misc/lambda_plugin.py => src/ocrmypdf/_lambda_plugin.py (100%) diff --git a/misc/lambda_plugin.py b/src/ocrmypdf/_lambda_plugin.py similarity index 100% rename from misc/lambda_plugin.py rename to src/ocrmypdf/_lambda_plugin.py From 8d23d0b4414ac955483d433edc568c6b7b0c7804 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 19:18:40 -0800 Subject: [PATCH 08/27] Operational lambda executor --- src/ocrmypdf/_lambda_plugin.py | 73 ++++++++++++++++------------------ 1 file changed, 34 insertions(+), 39 deletions(-) diff --git a/src/ocrmypdf/_lambda_plugin.py b/src/ocrmypdf/_lambda_plugin.py index b39270e4..69b709ac 100644 --- a/src/ocrmypdf/_lambda_plugin.py +++ b/src/ocrmypdf/_lambda_plugin.py @@ -31,7 +31,7 @@ import sys import threading from contextlib import suppress from itertools import islice, repeat, takewhile, zip_longest -from multiprocessing import Pipe, Process, process +from multiprocessing import Pipe, Process from multiprocessing.connection import Connection, wait from typing import Callable, Iterable, Optional, Union from unittest.mock import Mock @@ -47,64 +47,48 @@ def split_every(n: int, iterable: Iterable): return takewhile(bool, (list(islice(iterator, n)) for _ in repeat(None))) -def log_listener(q): - """Listen to the worker processes and forward the messages to logging - - For simplicity this is a thread rather than a process. Only one process - should actually write to sys.stderr or whatever we're using, so if this is - made into a process the main application needs to be directed to it. - - See https://docs.python.org/3/howto/logging-cookbook.html#logging-to-a-single-file-from-multiple-processes - """ - - while True: - try: - record = q.get() - if record is None: - break - logger = logging.getLogger(record.name) - logger.handle(record) - except Exception: # pylint: disable=broad-except - import traceback # pylint: disable=import-outside-toplevel - - print("Logging problem", file=sys.stderr) - traceback.print_exc(file=sys.stderr) - - def process_sigbus(*args): raise InputFileError("A worker process lost access to an input file") +class ConnectionLogHandler(logging.handlers.QueueHandler): + def __init__(self, conn: Connection) -> None: + super().__init__(None) + self.conn = conn + + def enqueue(self, record): + self.conn.send(('log', record)) + + def process_loop( conn: Connection, user_init: Callable[[], None], loglevel, task, task_args ): """Initialize a process pool worker""" - # Ignore SIGINT (our parent process will kill us gracefully) - signal.signal(signal.SIGINT, signal.SIG_IGN) - # Install SIGBUS handler (so our parent process can abort somewhat gracefully) with suppress(AttributeError): # Windows and Cygwin do not have SIGBUS # Windows and Cygwin do not have pthread_sigmask or SIGBUS signal.signal(signal.SIGBUS, process_sigbus) # Reconfigure the root logger for this process to send all messages to a queue - # h = logging.handlers.QueueHandler(q) - # root = logging.getLogger() - # root.setLevel(loglevel) - # root.handlers = [] - # root.addHandler(h) + h = ConnectionLogHandler(conn) + root = logging.getLogger() + root.setLevel(loglevel) + root.handlers = [] + root.addHandler(h) user_init() for args in task_args: try: - result = task(*args) + result = task(args) except Exception as e: - return # for now + conn.send(('exception', str(e))) + break else: - conn.send(result) + conn.send(('result', result)) + conn.send(('complete', None)) conn.close() return @@ -149,6 +133,8 @@ def _exec_progress_pool( task_arguments = list(task_arguments) grouped_args = list(zip_longest(*list(split_every(max_workers, task_arguments)))) + if not grouped_args: + return processes = [] connections = [] @@ -176,12 +162,21 @@ def _exec_progress_pool( while connections: for r in wait(connections): try: - msg = r.recv() + msg_type, msg = r.recv() except EOFError: connections.remove(r) else: - if task_finished: - task_finished(msg, pbar) + if msg_type == 'result': + if task_finished: + task_finished(msg, pbar) + elif msg_type == 'log': + record = msg + logger = logging.getLogger(record.name) + logger.handle(record) + elif msg_type == 'exception': + print(msg) + elif msg_type == 'complete': + connections.remove(r) for process in processes: process.join() From c395436ba30bc78d529c147bb75aaf18c741fe92 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 20:03:57 -0800 Subject: [PATCH 09/27] lambda: tidying, special casing use_threads --- src/ocrmypdf/_lambda_plugin.py | 104 ++++++++++++++++++--------------- 1 file changed, 56 insertions(+), 48 deletions(-) diff --git a/src/ocrmypdf/_lambda_plugin.py b/src/ocrmypdf/_lambda_plugin.py index 69b709ac..d6a4e4a3 100644 --- a/src/ocrmypdf/_lambda_plugin.py +++ b/src/ocrmypdf/_lambda_plugin.py @@ -23,17 +23,14 @@ import logging import logging.handlers -import multiprocessing -import os -import queue import signal -import sys import threading from contextlib import suppress +from enum import Enum, auto from itertools import islice, repeat, takewhile, zip_longest from multiprocessing import Pipe, Process from multiprocessing.connection import Connection, wait -from typing import Callable, Iterable, Optional, Union +from typing import Callable, Iterable, Optional from unittest.mock import Mock from ocrmypdf import hookimpl @@ -42,6 +39,12 @@ from ocrmypdf.exceptions import InputFileError pool_lock = threading.Lock() +class MessageType(Enum): + exception = auto() + result = auto() + complete = auto() + + def split_every(n: int, iterable: Iterable): iterator = iter(iterable) return takewhile(bool, (list(islice(iterator, n)) for _ in repeat(None))) @@ -83,47 +86,21 @@ def process_loop( try: result = task(args) except Exception as e: - conn.send(('exception', str(e))) + conn.send((MessageType.exception, str(e))) break else: - conn.send(('result', result)) + conn.send((MessageType.result, result)) - conn.send(('complete', None)) + conn.send((MessageType.complete, None)) conn.close() return -def exec_progress_pool( +def lambda_pool_impl( *, use_threads: bool, max_workers: int, tqdm_kwargs: dict, - worker_initializer: Optional[Callable], - task: Callable, - task_arguments: Optional[Iterable] = None, - task_finished: Callable, -): - - if not worker_initializer: - - def _noop(): - return - - worker_initializer = _noop - - with pool_lock: - _exec_progress_pool( - max_workers=max_workers, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - ) - - -def _exec_progress_pool( - *, - max_workers: int, worker_initializer: Callable, task: Callable, task_arguments: Optional[Iterable] = None, @@ -131,6 +108,32 @@ def _exec_progress_pool( ): pbar = Mock() + if use_threads and max_workers == 1: + for args in task_arguments: + result = task(args) + task_finished(result, pbar) + return + + with pool_lock: + _lambda_pool_impl( + max_workers=max_workers, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + pbar=pbar, + ) + + +def _lambda_pool_impl( + *, + max_workers: int, + worker_initializer: Callable, + task: Callable, + task_arguments: Optional[Iterable] = None, + task_finished: Callable, + pbar, +): task_arguments = list(task_arguments) grouped_args = list(zip_longest(*list(split_every(max_workers, task_arguments)))) if not grouped_args: @@ -165,18 +168,23 @@ def _exec_progress_pool( msg_type, msg = r.recv() except EOFError: connections.remove(r) - else: - if msg_type == 'result': - if task_finished: - task_finished(msg, pbar) - elif msg_type == 'log': - record = msg - logger = logging.getLogger(record.name) - logger.handle(record) - elif msg_type == 'exception': - print(msg) - elif msg_type == 'complete': - connections.remove(r) + continue + + if msg_type == MessageType.result: + if task_finished: + task_finished(msg, pbar) + elif msg_type == 'log': + record = msg + logger = logging.getLogger(record.name) + logger.handle(record) + elif msg_type == MessageType.complete: + connections.remove(r) + elif msg_type == MessageType.exception: + logger = logging.getLogger(__name__) + logger.error(msg) + for process in processes: + process.terminate() + raise RuntimeError("Failed") for process in processes: process.join() @@ -184,4 +192,4 @@ def _exec_progress_pool( @hookimpl def get_parallel_executor(): - return exec_progress_pool + return lambda_pool_impl From 1a3ce59476df8f77a9a7fb297963358833516b1d Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 20:05:14 -0800 Subject: [PATCH 10/27] lambda: Don't be paranoid about exception marshalling It works --- src/ocrmypdf/_lambda_plugin.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/ocrmypdf/_lambda_plugin.py b/src/ocrmypdf/_lambda_plugin.py index d6a4e4a3..eec248cb 100644 --- a/src/ocrmypdf/_lambda_plugin.py +++ b/src/ocrmypdf/_lambda_plugin.py @@ -86,7 +86,7 @@ def process_loop( try: result = task(args) except Exception as e: - conn.send((MessageType.exception, str(e))) + conn.send((MessageType.exception, e)) break else: conn.send((MessageType.result, result)) @@ -180,11 +180,9 @@ def _lambda_pool_impl( elif msg_type == MessageType.complete: connections.remove(r) elif msg_type == MessageType.exception: - logger = logging.getLogger(__name__) - logger.error(msg) for process in processes: process.terminate() - raise RuntimeError("Failed") + raise msg for process in processes: process.join() From 6083b4f0a7e71c7547136aaa91b14c0a0309e87e Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 20:52:41 -0800 Subject: [PATCH 11/27] lambda: don't overrun number of workers needed --- src/ocrmypdf/_lambda_plugin.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/ocrmypdf/_lambda_plugin.py b/src/ocrmypdf/_lambda_plugin.py index eec248cb..1cab2f61 100644 --- a/src/ocrmypdf/_lambda_plugin.py +++ b/src/ocrmypdf/_lambda_plugin.py @@ -141,10 +141,10 @@ def _lambda_pool_impl( processes = [] connections = [] - for n in range(max_workers): + for chunk in grouped_args: parent_conn, child_conn = Pipe() - worker_args = [args for args in grouped_args[n] if args is not None] + worker_args = [args for args in chunk if args is not None] process = Process( target=process_loop, args=( From 6a8dd65aa28ac87ef8ec38167991de29948829fb Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 21:16:11 -0800 Subject: [PATCH 12/27] lambda: more issues related to new executor semantics Now all tests pass, except for: -tests that check the progress bar -tests where xdist may or may not load a _lambda_plugin by running some other test first before a test in optimize --- src/ocrmypdf/_exec/jbig2enc.py | 8 ++++++++ src/ocrmypdf/_exec/pngquant.py | 4 ++++ src/ocrmypdf/optimize.py | 33 +++++++++++++++------------------ 3 files changed, 27 insertions(+), 18 deletions(-) diff --git a/src/ocrmypdf/_exec/jbig2enc.py b/src/ocrmypdf/_exec/jbig2enc.py index 9311ec3f..2e8a058b 100644 --- a/src/ocrmypdf/_exec/jbig2enc.py +++ b/src/ocrmypdf/_exec/jbig2enc.py @@ -41,9 +41,17 @@ def convert_group(*, cwd, infiles, out_prefix): return proc +def convert_group_mp(args): + return convert_group(cwd=args[0], infiles=args[1], out_prefix=args[2]) + + def convert_single(*, cwd, infile, outfile): args = ['jbig2', '-p', infile] with open(outfile, 'wb') as fstdout: proc = run(args, cwd=cwd, stdout=fstdout, stderr=PIPE) proc.check_returncode() return proc + + +def convert_single_mp(args): + return convert_single(cwd=args[0], infile=args[1], outfile=args[2]) diff --git a/src/ocrmypdf/_exec/pngquant.py b/src/ocrmypdf/_exec/pngquant.py index d7e345f2..12dc79bc 100644 --- a/src/ocrmypdf/_exec/pngquant.py +++ b/src/ocrmypdf/_exec/pngquant.py @@ -61,3 +61,7 @@ def quantize(input_file: Path, output_file: Path, quality_min: int, quality_max: if result.returncode == 0: # input_file could be the same as output_file, so we defer the write output_file.write_bytes(result.stdout) + + +def quantize_mp(args): + return quantize(*args) diff --git a/src/ocrmypdf/optimize.py b/src/ocrmypdf/optimize.py index c097e32e..3724f731 100644 --- a/src/ocrmypdf/optimize.py +++ b/src/ocrmypdf/optimize.py @@ -212,7 +212,10 @@ def extract_image_generic( def extract_images( - pike: Pdf, root: Path, options, extract_fn: Callable[..., Optional[XrefExt]], + pike: Pdf, + root: Path, + options, + extract_fn: Callable[..., Optional[XrefExt]], ) -> Iterator[Tuple[int, XrefExt]]: """Extract image using extract_fn @@ -305,10 +308,10 @@ def _produce_jbig2_images( def jbig2_group_args(root: Path, groups: Dict[int, List[XrefExt]]): for group, xref_exts in groups.items(): prefix = f'group{group:08d}' - yield dict( - cwd=fspath(root), - infiles=(img_name(root, xref, ext) for xref, ext in xref_exts), - out_prefix=prefix, + yield ( + fspath(root), # =cwd + (img_name(root, xref, ext) for xref, ext in xref_exts), # =infiles + prefix, # =out_prefix ) def jbig2_single_args(root, groups: Dict[int, List[XrefExt]]): @@ -317,21 +320,18 @@ def _produce_jbig2_images( # Second loop is to ensure multiple images per page are unpacked for n, xref_ext in enumerate(xref_exts): xref, ext = xref_ext - yield dict( - cwd=fspath(root), - infile=img_name(root, xref, ext), - outfile=root / f'{prefix}.{n:04d}', + yield ( + fspath(root), + img_name(root, xref, ext), + root / f'{prefix}.{n:04d}', ) - def convert_generic(fn, kwargs_dict): - return fn(**kwargs_dict) - if options.jbig2_page_group_size > 1: jbig2_args = jbig2_group_args - jbig2_convert = partial(convert_generic, jbig2enc.convert_group) + jbig2_convert = jbig2enc.convert_group_mp else: jbig2_args = jbig2_single_args - jbig2_convert = partial(convert_generic, jbig2enc.convert_single) + jbig2_convert = jbig2enc.convert_single_mp exec_progress_pool( use_threads=True, @@ -476,9 +476,6 @@ def transcode_pngs( ) modified.add(xref) - def pngquant_fn(args): - pngquant.quantize(*args) - exec_progress_pool( use_threads=True, max_workers=options.jobs, @@ -488,7 +485,7 @@ def transcode_pngs( unit='image', disable=not options.progress_bar, ), - task=pngquant_fn, + task=pngquant.quantize_mp, task_arguments=pngquant_args(), ) From 3bd5054634446cec9fa2152e7dc6fa66ed2c4fa8 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 23:40:10 -0800 Subject: [PATCH 13/27] lambda: move to extra_plugins folder --- src/ocrmypdf/extra_plugins/__init__.py | 0 .../awslambda.py} | 23 ++++--------------- 2 files changed, 5 insertions(+), 18 deletions(-) create mode 100644 src/ocrmypdf/extra_plugins/__init__.py rename src/ocrmypdf/{_lambda_plugin.py => extra_plugins/awslambda.py} (80%) diff --git a/src/ocrmypdf/extra_plugins/__init__.py b/src/ocrmypdf/extra_plugins/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/ocrmypdf/_lambda_plugin.py b/src/ocrmypdf/extra_plugins/awslambda.py similarity index 80% rename from src/ocrmypdf/_lambda_plugin.py rename to src/ocrmypdf/extra_plugins/awslambda.py index 1cab2f61..3a5f802d 100644 --- a/src/ocrmypdf/_lambda_plugin.py +++ b/src/ocrmypdf/extra_plugins/awslambda.py @@ -1,22 +1,9 @@ -# © 2021 James R Barlow: https://github.com/jbarlow83 +# © 2021 James R. Barlow: github.com/jbarlow83 # -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files (the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions: -# -# The above copyright notice and this permission notice shall be included in all -# copies or substantial portions of the Software. -# -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -# SOFTWARE. +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + """Alternate executor to support OCRmyPDF in AWS Lambda""" From 386cabff001dabfb8aa3e8644fc02ba6ce7ba083 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 24 Jan 2021 23:56:09 -0800 Subject: [PATCH 14/27] Make progress pool common rather than plugin-specific --- src/ocrmypdf/_concurrent.py | 22 ++++++++++-------- src/ocrmypdf/builtin_plugins/concurrency.py | 25 +++++++++------------ src/ocrmypdf/extra_plugins/awslambda.py | 20 +++++++---------- 3 files changed, 32 insertions(+), 35 deletions(-) diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index 87c9656c..e3fb5b91 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -4,8 +4,11 @@ # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. +import threading from typing import Callable, Iterable, Optional +pool_lock = threading.Lock() + def _task_noop(*_args, **_kwargs): return @@ -35,12 +38,13 @@ def exec_progress_pool( if not task_finished: task_finished = _task_noop - _model( - use_threads=use_threads, - max_workers=max_workers, - tqdm_kwargs=tqdm_kwargs, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - ) + with pool_lock: + _model( + use_threads=use_threads, + max_workers=max_workers, + tqdm_kwargs=tqdm_kwargs, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + ) diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py index 7d7fbbfc..3b3e0d74 100644 --- a/src/ocrmypdf/builtin_plugins/concurrency.py +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -31,8 +31,6 @@ from ocrmypdf.exceptions import InputFileError Queue = Union[multiprocessing.Queue, queue.Queue] -pool_lock = threading.Lock() - def log_listener(q: Queue): """Listen to the worker processes and forward the messages to logging @@ -120,18 +118,17 @@ def exec_progress_pool( worker_initializer = _noop - with pool_lock: - _exec_progress_pool( - max_workers=max_workers, - tqdm_kwargs=tqdm_kwargs, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - log_queue=log_queue, - pool_class=pool_class, - initializer=initializer, - ) + _exec_progress_pool( + max_workers=max_workers, + tqdm_kwargs=tqdm_kwargs, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + log_queue=log_queue, + pool_class=pool_class, + initializer=initializer, + ) def _exec_progress_pool( diff --git a/src/ocrmypdf/extra_plugins/awslambda.py b/src/ocrmypdf/extra_plugins/awslambda.py index 3a5f802d..4f7a7af7 100644 --- a/src/ocrmypdf/extra_plugins/awslambda.py +++ b/src/ocrmypdf/extra_plugins/awslambda.py @@ -11,7 +11,6 @@ import logging import logging.handlers import signal -import threading from contextlib import suppress from enum import Enum, auto from itertools import islice, repeat, takewhile, zip_longest @@ -23,8 +22,6 @@ from unittest.mock import Mock from ocrmypdf import hookimpl from ocrmypdf.exceptions import InputFileError -pool_lock = threading.Lock() - class MessageType(Enum): exception = auto() @@ -101,15 +98,14 @@ def lambda_pool_impl( task_finished(result, pbar) return - with pool_lock: - _lambda_pool_impl( - max_workers=max_workers, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - pbar=pbar, - ) + _lambda_pool_impl( + max_workers=max_workers, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + pbar=pbar, + ) def _lambda_pool_impl( From d274d88929d7d87b0e41313325ca56b1e4739b87 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sat, 30 Jan 2021 17:36:30 -0800 Subject: [PATCH 15/27] Refactor to eliminate global state in _concurrent --- src/ocrmypdf/__init__.py | 1 + src/ocrmypdf/_concurrent.py | 123 ++++++++++---- src/ocrmypdf/_pipeline.py | 8 +- src/ocrmypdf/_sync.py | 19 +-- src/ocrmypdf/builtin_plugins/concurrency.py | 149 +++++++---------- src/ocrmypdf/extra_plugins/awslambda.py | 174 ++++++++++---------- src/ocrmypdf/optimize.py | 29 +++- src/ocrmypdf/pdfinfo/info.py | 23 ++- src/ocrmypdf/pluginspec.py | 37 +---- tests/test_validation.py | 2 +- 10 files changed, 301 insertions(+), 264 deletions(-) diff --git a/src/ocrmypdf/__init__.py b/src/ocrmypdf/__init__.py index 081324e9..80586658 100644 --- a/src/ocrmypdf/__init__.py +++ b/src/ocrmypdf/__init__.py @@ -8,6 +8,7 @@ from pluggy import HookimplMarker as _HookimplMarker from ocrmypdf import helpers, hocrtransform, leptonica, pdfa, pdfinfo +from ocrmypdf._concurrent import Executor from ocrmypdf._jobcontext import PageContext, PdfContext from ocrmypdf._version import PROGRAM_NAME, __version__ from ocrmypdf.api import Verbosity, configure_logging, ocr diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index e3fb5b91..48c13e08 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -4,47 +4,110 @@ # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. +import sys import threading +from abc import ABC, abstractmethod +from functools import partial from typing import Callable, Iterable, Optional -pool_lock = threading.Lock() +from tqdm import tqdm def _task_noop(*_args, **_kwargs): return -def _model(**_kwargs) -> None: - raise RuntimeError("Parallel executor not set up") +class Executor(ABC): + pool_lock = threading.Lock() + + def __call__( + self, + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Optional[Callable] = None, + task: Optional[Callable] = None, + task_arguments: Optional[Iterable] = None, + task_finished: Optional[Callable] = None, + ) -> None: + """ + Args: + use_threads: If False, the workload is the sort that will benefit from + running in a multiprocessing context (for example, it uses Python + heavily, and parallelizing it with threads is not expected to be + performant). + max_workers: The maximum number of workers that should be run. + tdqm_kwargs: Arguments to set up the progress bar. + worker_initializer: Called when the worker is initialized, in the worker's + execution context. Must be possible to marshall to the worker. + task: Called when the worker starts a new task, in the worker's execution + context. Must be possible to marshallable to the worker. + task_finished: Called when a worker finishes a task, in the parent's + context. + task_arguments: An iterable that generates a group of parameters for each + task. This runs in the parent's context, but the parameters must be + marshallable to the worker. + """ + + if not task_arguments: + return # Nothing to do! + if not worker_initializer: + worker_initializer = _task_noop + if not task_finished: + task_finished = _task_noop + if not task: + task = _task_noop + + with self.pool_lock: + self._execute( + use_threads=use_threads, + max_workers=max_workers, + tqdm_kwargs=tqdm_kwargs, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + ) + + @abstractmethod + def _execute( + self, + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_arguments: Iterable, + task_finished: Callable, + ): + """Custom executors should override this method.""" -def set_execution_model(model): - global _model - _model = model +def setup_executor(plugin_manager) -> Executor: + return plugin_manager.hook.get_executor() -def exec_progress_pool( - *, - use_threads: bool, - max_workers: int, - tqdm_kwargs: dict, - worker_initializer: Optional[Callable] = None, - task: Callable, - task_arguments: Optional[Iterable] = None, - task_finished: Optional[Callable] = None, -): - if not worker_initializer: - worker_initializer = _task_noop - if not task_finished: - task_finished = _task_noop +class SerialExecutor(Executor): + """Implements a purely sequential executor using the parallel protocol. - with pool_lock: - _model( - use_threads=use_threads, - max_workers=max_workers, - tqdm_kwargs=tqdm_kwargs, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - ) + The current process/thread will be the worker that executes all tasks + in order. As such, ``worker_initializer`` will never be called. + """ + + def _execute( + self, + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_arguments: Iterable, + task_finished: Callable, + ): + with tqdm(**tqdm_kwargs) as pbar: + for args in task_arguments: + result = task(args) + task_finished(result, pbar) diff --git a/src/ocrmypdf/_pipeline.py b/src/ocrmypdf/_pipeline.py index 6de1f2e9..d4c171d8 100644 --- a/src/ocrmypdf/_pipeline.py +++ b/src/ocrmypdf/_pipeline.py @@ -22,6 +22,7 @@ from PIL import Image, ImageColor, ImageDraw from tqdm import tqdm from ocrmypdf import leptonica +from ocrmypdf._concurrent import Executor from ocrmypdf._exec import unpaper from ocrmypdf._jobcontext import PageContext, PdfContext from ocrmypdf._version import PROGRAM_NAME @@ -146,6 +147,8 @@ def triage(original_filename, input_file, output_file, options): def get_pdfinfo( input_file, + *, + executor: Executor, detailed_analysis=False, progbar=False, max_workers=None, @@ -158,6 +161,7 @@ def get_pdfinfo( progbar=progbar, max_workers=max_workers, check_pages=check_pages, + executor=executor, ) except pikepdf.PasswordError: raise EncryptedPdfError() @@ -792,7 +796,7 @@ def metadata_fixup(working_file: Path, context: PdfContext): return output_file -def optimize_pdf(input_file: Path, context: PdfContext): +def optimize_pdf(input_file: Path, context: PdfContext, executor: Executor): output_file = context.get_path('optimize.pdf') save_settings = dict( compress_streams=True, @@ -800,7 +804,7 @@ def optimize_pdf(input_file: Path, context: PdfContext): object_stream_mode=pikepdf.ObjectStreamMode.generate, linearize=should_linearize(input_file, context), ) - optimize(input_file, output_file, context, save_settings) + optimize(input_file, output_file, context, save_settings, executor) return output_file diff --git a/src/ocrmypdf/_sync.py b/src/ocrmypdf/_sync.py index 1d78f1d9..68577a5d 100644 --- a/src/ocrmypdf/_sync.py +++ b/src/ocrmypdf/_sync.py @@ -15,10 +15,9 @@ from pathlib import Path from tempfile import mkdtemp from typing import List, NamedTuple, Optional, Tuple -import pikepdf import PIL -from ocrmypdf._concurrent import exec_progress_pool, set_execution_model +from ocrmypdf._concurrent import Executor, setup_executor from ocrmypdf._graft import OcrGrafter from ocrmypdf._jobcontext import PageContext, PdfContext, cleanup_working_files from ocrmypdf._logging import PageNumberFilter @@ -224,14 +223,14 @@ def exec_page_sync(page_context: PageContext): ) -def post_process(pdf_file, context: PdfContext): +def post_process(pdf_file, context: PdfContext, executor: Executor): pdf_out = pdf_file if context.options.output_type.startswith('pdfa'): ps_stub_out = generate_postscript_stub(context) pdf_out = convert_to_pdfa(pdf_out, ps_stub_out, context) pdf_out = metadata_fixup(pdf_out, context) - return optimize_pdf(pdf_out, context) + return optimize_pdf(pdf_out, context, executor) def worker_init(max_pixels: int): @@ -242,7 +241,7 @@ def worker_init(max_pixels: int): pikepdf_enable_mmap() -def exec_concurrent(context: PdfContext): +def exec_concurrent(context: PdfContext, executor: Executor): """Execute the pipeline concurrently""" # Run exec_page_sync on every page context @@ -269,7 +268,7 @@ def exec_concurrent(context: PdfContext): finally: tls.pageno = None - exec_progress_pool( + executor( use_threads=options.use_threads, max_workers=max_workers, tqdm_kwargs=dict( @@ -296,7 +295,7 @@ def exec_concurrent(context: PdfContext): # PDF/A and metadata log.info("Postprocessing...") - pdf = post_process(pdf, context) + pdf = post_process(pdf, context, executor) # Copy PDF file to destination copy_final(pdf, options.output_file, context) @@ -346,8 +345,7 @@ def run_pipeline(options, *, plugin_manager, api=False): pikepdf_enable_mmap() - set_execution_model(plugin_manager.hook.get_parallel_executor()) - + executor = setup_executor(plugin_manager) try: check_requested_output_file(options) start_input_file, original_filename = create_input_file(options, work_folder) @@ -360,6 +358,7 @@ def run_pipeline(options, *, plugin_manager, api=False): # Gather pdfinfo and create context pdfinfo = get_pdfinfo( origin_pdf, + executor=executor, detailed_analysis=options.redo_ocr, progbar=options.progress_bar, max_workers=options.jobs if not options.use_threads else 1, # To help debug @@ -372,7 +371,7 @@ def run_pipeline(options, *, plugin_manager, api=False): validate_pdfinfo_options(context) # Execute the pipeline - exec_concurrent(context) + exec_concurrent(context, executor) if options.output_file == '-': log.info("Output sent to stdout") diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py index 3b3e0d74..b61654dc 100644 --- a/src/ocrmypdf/builtin_plugins/concurrency.py +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -26,7 +26,7 @@ from typing import Callable, Iterable, Optional, Union from tqdm import tqdm -from ocrmypdf import hookimpl +from ocrmypdf import Executor, hookimpl from ocrmypdf.exceptions import InputFileError Queue = Union[multiprocessing.Queue, queue.Queue] @@ -91,99 +91,68 @@ def thread_init(_queue: Queue, user_init: Callable[[], None], _loglevel): return -def exec_progress_pool( - *, - use_threads: bool, - max_workers: int, - tqdm_kwargs: dict, - worker_initializer: Optional[Callable], - task: Callable, - task_arguments: Optional[Iterable] = None, - task_finished: Callable, -): +class StandardExecutor(Executor): + def _execute( + self, + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_arguments: Iterable, + task_finished: Callable, + ): + if use_threads: + log_queue = queue.Queue(-1) + pool_class = ThreadPool + initializer = thread_init + else: + log_queue = multiprocessing.Queue(-1) + pool_class = ProcessPool + initializer = process_init - if use_threads: - log_queue = queue.Queue(-1) - pool_class = ThreadPool - initializer = thread_init - else: - log_queue = multiprocessing.Queue(-1) - pool_class = ProcessPool - initializer = process_init + # Regardless of whether we use_threads for worker processes, the log_listener + # must be a thread + listener = threading.Thread(target=log_listener, args=(log_queue,)) + listener.start() - if not worker_initializer: - - def _noop(): - return - - worker_initializer = _noop - - _exec_progress_pool( - max_workers=max_workers, - tqdm_kwargs=tqdm_kwargs, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - log_queue=log_queue, - pool_class=pool_class, - initializer=initializer, - ) - - -def _exec_progress_pool( - *, - max_workers: int, - tqdm_kwargs: dict, - worker_initializer: Callable, - task: Callable, - task_arguments: Optional[Iterable] = None, - task_finished: Callable, - log_queue: Queue, - pool_class: Callable, - initializer: Callable, -): - # Regardless of whether we use_threads for worker processes, the log_listener - # must be a thread - listener = threading.Thread(target=log_listener, args=(log_queue,)) - listener.start() - - with tqdm(**tqdm_kwargs) as pbar: - pool = pool_class( - processes=max_workers, - initializer=initializer, - initargs=(log_queue, worker_initializer, logging.getLogger("").level), - ) - try: - results = pool.imap_unordered(task, task_arguments) - for result in results: - if task_finished: - task_finished(result, pbar) - else: - pbar.update() - except KeyboardInterrupt: - # Terminate pool so we exit instantly - pool.terminate() - # Don't try listener.join() here, will deadlock - raise - except Exception: - if not os.environ.get("PYTEST_CURRENT_TEST", ""): - # Unless inside pytest, exit immediately because no one wants - # to wait for child processes to finalize results that will be - # thrown away. Inside pytest, we want child processes to exit - # cleanly so that they output an error messages or coverage data - # we need from them. + with tqdm(**tqdm_kwargs) as pbar: + pool = pool_class( + processes=max_workers, + initializer=initializer, + initargs=(log_queue, worker_initializer, logging.getLogger("").level), + ) + try: + results = pool.imap_unordered(task, task_arguments) + for result in results: + if task_finished: + task_finished(result, pbar) + else: + pbar.update() + except KeyboardInterrupt: + # Terminate pool so we exit instantly pool.terminate() - raise - finally: - # Terminate log listener - log_queue.put_nowait(None) - pool.close() - pool.join() + # Don't try listener.join() here, will deadlock + raise + except Exception: + if not os.environ.get("PYTEST_CURRENT_TEST", ""): + # Unless inside pytest, exit immediately because no one wants + # to wait for child processes to finalize results that will be + # thrown away. Inside pytest, we want child processes to exit + # cleanly so that they output an error messages or coverage data + # we need from them. + pool.terminate() + raise + finally: + # Terminate log listener + log_queue.put_nowait(None) + pool.close() + pool.join() - listener.join() + listener.join() @hookimpl -def get_parallel_executor(): - return exec_progress_pool +def get_executor(): + return StandardExecutor() diff --git a/src/ocrmypdf/extra_plugins/awslambda.py b/src/ocrmypdf/extra_plugins/awslambda.py index 4f7a7af7..8579683d 100644 --- a/src/ocrmypdf/extra_plugins/awslambda.py +++ b/src/ocrmypdf/extra_plugins/awslambda.py @@ -19,7 +19,7 @@ from multiprocessing.connection import Connection, wait from typing import Callable, Iterable, Optional from unittest.mock import Mock -from ocrmypdf import hookimpl +from ocrmypdf import Executor, hookimpl from ocrmypdf.exceptions import InputFileError @@ -80,97 +80,101 @@ def process_loop( return -def lambda_pool_impl( - *, - use_threads: bool, - max_workers: int, - tqdm_kwargs: dict, - worker_initializer: Callable, - task: Callable, - task_arguments: Optional[Iterable] = None, - task_finished: Callable, -): - pbar = Mock() +class LambdaExecutor(Executor): + def _execute( + self, + *, + use_threads: bool, + max_workers: int, + tqdm_kwargs: dict, + worker_initializer: Callable, + task: Callable, + task_arguments: Iterable, + task_finished: Callable, + ): + pbar = Mock() - if use_threads and max_workers == 1: - for args in task_arguments: - result = task(args) - task_finished(result, pbar) - return + if use_threads and max_workers == 1: + for args in task_arguments: + result = task(args) + task_finished(result, pbar) + return - _lambda_pool_impl( - max_workers=max_workers, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - pbar=pbar, - ) - - -def _lambda_pool_impl( - *, - max_workers: int, - worker_initializer: Callable, - task: Callable, - task_arguments: Optional[Iterable] = None, - task_finished: Callable, - pbar, -): - task_arguments = list(task_arguments) - grouped_args = list(zip_longest(*list(split_every(max_workers, task_arguments)))) - if not grouped_args: - return - - processes = [] - connections = [] - for chunk in grouped_args: - parent_conn, child_conn = Pipe() - - worker_args = [args for args in chunk if args is not None] - process = Process( - target=process_loop, - args=( - child_conn, - worker_initializer, - logging.getLogger("").level, - task, - worker_args, - ), + self._lambda_pool_impl( + max_workers=max_workers, + worker_initializer=worker_initializer, + task=task, + task_arguments=task_arguments, + task_finished=task_finished, + pbar=pbar, ) - process.daemon = True - processes.append(process) - connections.append(parent_conn) - for process in processes: - process.start() + def _lambda_pool_impl( + self, + *, + max_workers: int, + worker_initializer: Callable, + task: Callable, + task_arguments: Iterable, + task_finished: Callable, + pbar, + ): + task_arguments = list(task_arguments) + grouped_args = list( + zip_longest(*list(split_every(max_workers, task_arguments))) + ) + if not grouped_args: + return - while connections: - for r in wait(connections): - try: - msg_type, msg = r.recv() - except EOFError: - connections.remove(r) - continue + processes = [] + connections = [] + for chunk in grouped_args: + parent_conn, child_conn = Pipe() - if msg_type == MessageType.result: - if task_finished: - task_finished(msg, pbar) - elif msg_type == 'log': - record = msg - logger = logging.getLogger(record.name) - logger.handle(record) - elif msg_type == MessageType.complete: - connections.remove(r) - elif msg_type == MessageType.exception: - for process in processes: - process.terminate() - raise msg + worker_args = [args for args in chunk if args is not None] + process = Process( + target=process_loop, + args=( + child_conn, + worker_initializer, + logging.getLogger("").level, + task, + worker_args, + ), + ) + process.daemon = True + processes.append(process) + connections.append(parent_conn) - for process in processes: - process.join() + for process in processes: + process.start() + + while connections: + for r in wait(connections): + try: + msg_type, msg = r.recv() + except EOFError: + connections.remove(r) + continue + + if msg_type == MessageType.result: + if task_finished: + task_finished(msg, pbar) + elif msg_type == 'log': + record = msg + logger = logging.getLogger(record.name) + logger.handle(record) + elif msg_type == MessageType.complete: + connections.remove(r) + elif msg_type == MessageType.exception: + for process in processes: + process.terminate() + raise msg + + for process in processes: + process.join() @hookimpl -def get_parallel_executor(): - return lambda_pool_impl +def get_executor(): + return LambdaExecutor() diff --git a/src/ocrmypdf/optimize.py b/src/ocrmypdf/optimize.py index 3724f731..5993014d 100644 --- a/src/ocrmypdf/optimize.py +++ b/src/ocrmypdf/optimize.py @@ -34,7 +34,7 @@ from PIL import Image from tqdm import tqdm from ocrmypdf import leptonica -from ocrmypdf._concurrent import exec_progress_pool +from ocrmypdf._concurrent import Executor, SerialExecutor from ocrmypdf._exec import jbig2enc, pngquant from ocrmypdf._jobcontext import PdfContext from ocrmypdf.exceptions import OutputFileAccessError @@ -301,7 +301,7 @@ def extract_images_jbig2(pike: Pdf, root: Path, options) -> Dict[int, List[XrefE def _produce_jbig2_images( - jbig2_groups: Dict[int, List[XrefExt]], root: Path, options + jbig2_groups: Dict[int, List[XrefExt]], root: Path, options, executor: Executor ) -> None: """Produce JBIG2 images from their groups""" @@ -333,7 +333,7 @@ def _produce_jbig2_images( jbig2_args = jbig2_single_args jbig2_convert = jbig2enc.convert_single_mp - exec_progress_pool( + executor( use_threads=True, max_workers=options.jobs, tqdm_kwargs=dict( @@ -348,7 +348,11 @@ def _produce_jbig2_images( def convert_to_jbig2( - pike: Pdf, jbig2_groups: Dict[int, List[XrefExt]], root: Path, options + pike: Pdf, + jbig2_groups: Dict[int, List[XrefExt]], + root: Path, + options, + executor: Executor, ) -> None: """Convert images to JBIG2 and insert into PDF. @@ -363,7 +367,7 @@ def convert_to_jbig2( and needs no dictionary. Currently this must be lossless JBIG2. """ - _produce_jbig2_images(jbig2_groups, root, options) + _produce_jbig2_images(jbig2_groups, root, options, executor) for group, xref_exts in jbig2_groups.items(): prefix = f'group{group:08d}' @@ -457,6 +461,7 @@ def transcode_pngs( image_name_fn: Callable[[Path, Xref], Path], root: Path, options, + executor, ) -> None: modified: MutableSet[Xref] = set() if options.optimize >= 2: @@ -476,7 +481,7 @@ def transcode_pngs( ) modified.add(xref) - exec_progress_pool( + executor( use_threads=True, max_workers=options.jobs, tqdm_kwargs=dict( @@ -569,7 +574,13 @@ def rewrite_png(pike: Pdf, im_obj: Object, compdata) -> None: # pragma: no cove im_obj.write(compdata.read(), filter=Name.FlateDecode, decode_parms=dparms) -def optimize(input_file: Path, output_file: Path, context, save_settings) -> None: +def optimize( + input_file: Path, + output_file: Path, + context, + save_settings, + executor: Executor = SerialExecutor(), +) -> None: options = context.options if options.optimize == 0: safe_symlink(input_file, output_file) @@ -591,10 +602,10 @@ def optimize(input_file: Path, output_file: Path, context, save_settings) -> Non # if options.optimize >= 2: # Try pngifying the jpegs # transcode_pngs(pike, jpegs, jpg_name, root, options) - transcode_pngs(pike, pngs, png_name, root, options) + transcode_pngs(pike, pngs, png_name, root, options, executor) jbig2_groups = extract_images_jbig2(pike, root, options) - convert_to_jbig2(pike, jbig2_groups, root, options) + convert_to_jbig2(pike, jbig2_groups, root, options, executor) target_file = output_file.with_suffix('.opt.pdf') pike.remove_unreferenced_resources() diff --git a/src/ocrmypdf/pdfinfo/info.py b/src/ocrmypdf/pdfinfo/info.py index 1ee82cf9..3733feaf 100644 --- a/src/ocrmypdf/pdfinfo/info.py +++ b/src/ocrmypdf/pdfinfo/info.py @@ -10,6 +10,7 @@ import atexit import logging import re from collections import defaultdict, namedtuple +from contextlib import ExitStack from decimal import Decimal from enum import Enum from functools import partial @@ -22,7 +23,7 @@ from warnings import warn import pikepdf from pikepdf import Object, Pdf, PdfMatrix -from ocrmypdf._concurrent import exec_progress_pool +from ocrmypdf._concurrent import Executor, SerialExecutor from ocrmypdf.exceptions import EncryptedPdfError, InputFileError from ocrmypdf.helpers import Resolution, available_cpu_count, pikepdf_enable_mmap from ocrmypdf.pdfinfo.layout import get_page_analysis, get_text_boxes @@ -592,12 +593,21 @@ def _pdf_pageinfo_sync_init(pdf: Pdf, infile: Path, pdfminer_loglevel): def _pdf_pageinfo_sync(args): pageno, thread_pdf, infile, check_pages, detailed_analysis = args pdf = thread_pdf if thread_pdf is not None else worker_pdf - page = PageInfo(pdf, pageno, infile, check_pages, detailed_analysis) - return page + with ExitStack() as stack: + if not pdf: # When called with SerialExecutor + pdf = stack.enter_context(pikepdf.open(infile)) + page = PageInfo(pdf, pageno, infile, check_pages, detailed_analysis) + return page def _pdf_pageinfo_concurrent( - pdf, infile, progbar, max_workers, check_pages, detailed_analysis=False + pdf, + executor: Executor, + infile, + progbar, + max_workers, + check_pages, + detailed_analysis=False, ): pages = [None] * len(pdf.pages) @@ -629,7 +639,7 @@ def _pdf_pageinfo_concurrent( (n, initial_pdf, infile, check_pages, detailed_analysis) for n in range(total) ) assert n_workers == 1 if use_threads else n_workers >= 1, "Not multithreadable" - exec_progress_pool( + executor( use_threads=use_threads, max_workers=n_workers, tqdm_kwargs=dict( @@ -829,10 +839,12 @@ class PdfInfo: def __init__( self, infile, + *, detailed_analysis: bool = False, progbar: bool = False, max_workers: int = None, check_pages=None, + executor: Executor = SerialExecutor(), ): self._infile = infile if check_pages is None: @@ -843,6 +855,7 @@ class PdfInfo: raise EncryptedPdfError() # Triggered by encryption with empty passwd self._pages = _pdf_pageinfo_concurrent( pdf, + executor, infile, progbar, max_workers, diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index 538be7a1..f3d5d6fa 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -13,6 +13,7 @@ from typing import TYPE_CHECKING, AbstractSet, Callable, Iterable, List, Optiona import pluggy +from ocrmypdf._concurrent import Executor from ocrmypdf.helpers import Resolution if TYPE_CHECKING: @@ -62,44 +63,16 @@ def check_options(options: Namespace) -> None: """ -class ParallelExecutor(ABC): - @abstractstaticmethod - def __call__( - use_threads: bool, - max_workers: int, - tqdm_kwargs: dict, - worker_initializer: Callable, - task: Callable, - task_finished: Callable, - task_arguments: Optional[Iterable] = None, - ): - """ - Args: - use_threads: If False, the workload is the sort that will benefit from - running in a multiprocessing context (for example, it uses Python - heavily, and parallelizing it with threads is not expected to be - performant). - max_workers: The maximum number of workers that should be run. - tdqm_kwargs: Arguments to set up the progress bar. - worker_initializer: Called when the worker is initialized, in the worker's - execution context. Must be possible to marshall to the worker. - task: Called when the worker starts a new task, in the worker's execution - context. Must be possible to marshallable to the worker. - task_finished: Called when a worker finishes a task, in the parent's - context. - task_arguments: An iterable that generates a group of parameters for each - task. This runs in the parent's context, but the parameters must be - marshallable to the worker. - """ - - @hookspec(firstresult=True) -def get_parallel_executor() -> Callable: +def get_executor() -> Executor: """Called to perform parallel execution This may be used to replace OCRmyPDF's default parallel execution system with a third party alternative. For example, you could make OCRmyPDF run in a distributed environment. + + OCRmyPDF's executors are analogous to the standard Python executors in + conconcurrent.futures, but they do not work the same way. """ diff --git a/tests/test_validation.py b/tests/test_validation.py index deed9769..f7ce5286 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -173,7 +173,7 @@ def test_false_action_store_true(): def test_no_progress_bar(progress_bar, resources): opts = make_opts(progress_bar=progress_bar, input_file=(resources / 'trivial.pdf')) plugin_manager = get_plugin_manager(opts.plugins) - with patch('ocrmypdf.builtin_plugins.concurrency.tqdm', autospec=True) as tqdmpatch: + with patch('ocrmypdf._concurrent.tqdm', autospec=True) as tqdmpatch: vd._check_options(opts, plugin_manager, set()) pdfinfo = PdfInfo(opts.input_file, progbar=opts.progress_bar) assert pdfinfo is not None From 16bda74974df2802f44f421402ed969b3f119e3b Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sat, 30 Jan 2021 20:42:00 -0800 Subject: [PATCH 16/27] Refactor - decouple progressbar from executor --- docs/plugins.rst | 10 ++++++ src/ocrmypdf/_concurrent.py | 38 ++++++++++++++++----- src/ocrmypdf/builtin_plugins/concurrency.py | 7 +++- src/ocrmypdf/extra_plugins/awslambda.py | 5 +-- src/ocrmypdf/pluginspec.py | 37 ++++++++++++++++++-- tests/test_validation.py | 26 +++++++++----- 6 files changed, 100 insertions(+), 23 deletions(-) diff --git a/docs/plugins.rst b/docs/plugins.rst index 12abf0ab..fb97f7f9 100644 --- a/docs/plugins.rst +++ b/docs/plugins.rst @@ -130,6 +130,16 @@ Custom command line arguments .. autofunction:: ocrmypdf.pluginspec.check_options +Execution and progress reporting +-------------------------------- + +.. autoclass: ocrmypdf.pluginspec.Executor + :members: + +.. autofunction:: ocrmypdf.pluginspec.get_executor + +.. autofunction:: ocrmypdf.pluginspec.get_progress_bar + Applying special behavior before processing ------------------------------------------- diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index 48c13e08..ea975bae 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -10,15 +10,32 @@ from abc import ABC, abstractmethod from functools import partial from typing import Callable, Iterable, Optional -from tqdm import tqdm - def _task_noop(*_args, **_kwargs): return +class NullProgressBar: + def __init__(self, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + return False + + def update(self, _arg=None): + return + + class Executor(ABC): pool_lock = threading.Lock() + pbar_class = NullProgressBar + + def __init__(self, *, pbar_class=None): + if pbar_class: + self.pbar_class = pbar_class def __call__( self, @@ -32,17 +49,21 @@ class Executor(ABC): task_finished: Optional[Callable] = None, ) -> None: """ + Set up parallel execution and progress reporting. + Args: - use_threads: If False, the workload is the sort that will benefit from + use_threads: If ``False``, the workload is the sort that will benefit from running in a multiprocessing context (for example, it uses Python heavily, and parallelizing it with threads is not expected to be performant). max_workers: The maximum number of workers that should be run. tdqm_kwargs: Arguments to set up the progress bar. - worker_initializer: Called when the worker is initialized, in the worker's - execution context. Must be possible to marshall to the worker. + worker_initializer: Called when a worker is initialized, in the worker's + execution context. If the child workers are processes, it must be + possible to marshall/pickle the worker initializer. + ``functools.partial`` can be used to bind parameters. task: Called when the worker starts a new task, in the worker's execution - context. Must be possible to marshallable to the worker. + context. Must be possible to marshall to the worker. task_finished: Called when a worker finishes a task, in the parent's context. task_arguments: An iterable that generates a group of parameters for each @@ -86,7 +107,8 @@ class Executor(ABC): def setup_executor(plugin_manager) -> Executor: - return plugin_manager.hook.get_executor() + pbar_class = plugin_manager.hook.get_progress_bar() + return plugin_manager.hook.get_executor(pbar_class=pbar_class) class SerialExecutor(Executor): @@ -107,7 +129,7 @@ class SerialExecutor(Executor): task_arguments: Iterable, task_finished: Callable, ): - with tqdm(**tqdm_kwargs) as pbar: + with self.pbar_class(**tqdm_kwargs) as pbar: for args in task_arguments: result = task(args) task_finished(result, pbar) diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py index b61654dc..df6d42c6 100644 --- a/src/ocrmypdf/builtin_plugins/concurrency.py +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -117,7 +117,7 @@ class StandardExecutor(Executor): listener = threading.Thread(target=log_listener, args=(log_queue,)) listener.start() - with tqdm(**tqdm_kwargs) as pbar: + with self.pbar_class(**tqdm_kwargs) as pbar: pool = pool_class( processes=max_workers, initializer=initializer, @@ -156,3 +156,8 @@ class StandardExecutor(Executor): @hookimpl def get_executor(): return StandardExecutor() + + +@hookimpl +def get_progress_bar(): + return tqdm diff --git a/src/ocrmypdf/extra_plugins/awslambda.py b/src/ocrmypdf/extra_plugins/awslambda.py index 8579683d..64297f67 100644 --- a/src/ocrmypdf/extra_plugins/awslambda.py +++ b/src/ocrmypdf/extra_plugins/awslambda.py @@ -92,12 +92,10 @@ class LambdaExecutor(Executor): task_arguments: Iterable, task_finished: Callable, ): - pbar = Mock() - if use_threads and max_workers == 1: for args in task_arguments: result = task(args) - task_finished(result, pbar) + task_finished(result, self.pbar_class) return self._lambda_pool_impl( @@ -106,7 +104,6 @@ class LambdaExecutor(Executor): task=task, task_arguments=task_arguments, task_finished=task_finished, - pbar=pbar, ) def _lambda_pool_impl( diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index f3d5d6fa..cf68fd77 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -65,14 +65,47 @@ def check_options(options: Namespace) -> None: @hookspec(firstresult=True) def get_executor() -> Executor: - """Called to perform parallel execution + """Called to obtain an object that manages parallel execution. This may be used to replace OCRmyPDF's default parallel execution system with a third party alternative. For example, you could make OCRmyPDF run in a distributed environment. OCRmyPDF's executors are analogous to the standard Python executors in - conconcurrent.futures, but they do not work the same way. + ``conconcurrent.futures``, but they do not work the same way. + + Should be of type :class:`Executor` or otherwise conforming to the protocol + of that call. + + Note: + This hook will be called from the main process, and may modify global state + before child worker processes are forked. + Note: + This is a :ref:`firstresult hook`. + """ + + +@hookspec(firstresult=True) +def get_progress_bar(): + """Called to obtain a class that can be used to create progress bars. + + The class should follow a tqdm-like protocol. Calling the class should return + a new progress bar object, which is activated with ``__enter__`` and terminated + ``__exit__``. An update method is called whenever the progress bar is updated. + + The progress bar is held in the main process/thread and not updated by child + process/threads. When a child notifies the parent of completed work, the + parent updates the progress bar. + + The arguments are the same as `tqdm `_ accepts. + + Here is how OCRmyPDF will use the progress bar: + + Example: + pbar_class = pm.hook.get_progress_bar() + with pbar_class(**tqdm_kwargs) as pbar: + ... + pbar.update(1) """ diff --git a/tests/test_validation.py b/tests/test_validation.py index f7ce5286..c037fea9 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -6,12 +6,13 @@ import logging -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pikepdf import pytest from ocrmypdf import _validation as vd +from ocrmypdf._concurrent import NullProgressBar, SerialExecutor from ocrmypdf._plugin_manager import get_plugin_manager from ocrmypdf.api import create_options from ocrmypdf.cli import get_parser @@ -173,13 +174,22 @@ def test_false_action_store_true(): def test_no_progress_bar(progress_bar, resources): opts = make_opts(progress_bar=progress_bar, input_file=(resources / 'trivial.pdf')) plugin_manager = get_plugin_manager(opts.plugins) - with patch('ocrmypdf._concurrent.tqdm', autospec=True) as tqdmpatch: - vd._check_options(opts, plugin_manager, set()) - pdfinfo = PdfInfo(opts.input_file, progbar=opts.progress_bar) - assert pdfinfo is not None - assert tqdmpatch.called - _args, kwargs = tqdmpatch.call_args - assert kwargs['disable'] != progress_bar + + vd._check_options(opts, plugin_manager, set()) + + pbar_disabled = None + + class CheckProgressBar(NullProgressBar): + def __init__(self, disable, **kwargs): + nonlocal pbar_disabled + pbar_disabled = disable + super().__init__(disable=disable, **kwargs) + + executor = SerialExecutor(pbar_class=CheckProgressBar) + pdfinfo = PdfInfo(opts.input_file, progbar=opts.progress_bar, executor=executor) + + assert pdfinfo is not None + assert pbar_disabled is not None and pbar_disabled != progress_bar def test_language_warning(caplog): From 42c84531e42d909b1e1a295483f32bceca7b8d37 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 31 Jan 2021 02:18:46 -0800 Subject: [PATCH 17/27] optimize: rewrite JPEG optimize to avoid use of tqdm and parallelize For some reason JPEG optimization was not done in parallel, and was perhaps never done in parallel. Strange oversight. --- src/ocrmypdf/optimize.py | 67 +++++++++++++++++++++++++++------------- 1 file changed, 45 insertions(+), 22 deletions(-) diff --git a/src/ocrmypdf/optimize.py b/src/ocrmypdf/optimize.py index 5993014d..112a26c3 100644 --- a/src/ocrmypdf/optimize.py +++ b/src/ocrmypdf/optimize.py @@ -9,8 +9,6 @@ import logging import sys import tempfile from collections import defaultdict -from functools import partial -from io import BytesIO from os import fspath from pathlib import Path from typing import ( @@ -31,7 +29,6 @@ import img2pdf import pikepdf from pikepdf import Dictionary, Name, Object, Pdf, PdfImage from PIL import Image -from tqdm import tqdm from ocrmypdf import leptonica from ocrmypdf._concurrent import Executor, SerialExecutor @@ -391,27 +388,53 @@ def convert_to_jbig2( ) -def transcode_jpegs(pike: Pdf, jpegs: Sequence[Xref], root: Path, options) -> None: - for xref in tqdm( - jpegs, desc="JPEGs", unit='image', disable=not options.progress_bar - ): - in_jpg = jpg_name(root, xref) - opt_jpg = in_jpg.with_suffix('.opt.jpg') +def _optimize_jpeg(args): + xref, in_jpg, opt_jpg, jpeg_quality = args - # This produces a debug warning from PIL - # DEBUG:PIL.Image:Error closing: 'NoneType' object has no attribute - # 'close'. Seems to be mostly harmless - # https://github.com/python-pillow/Pillow/issues/1144 - with Image.open(in_jpg) as im: - im.save(opt_jpg, optimize=True, quality=options.jpeg_quality) + # This may produce a debug warning from PIL + # DEBUG:PIL.Image:Error closing: 'NoneType' object has no attribute + # 'close'. Seems to be mostly harmless + # https://github.com/python-pillow/Pillow/issues/1144 + with Image.open(in_jpg) as im: + im.save(opt_jpg, optimize=True, quality=jpeg_quality) - if opt_jpg.stat().st_size > in_jpg.stat().st_size: - log.debug("xref %s, jpeg, made larger - skip", xref) - continue + if opt_jpg.stat().st_size > in_jpg.stat().st_size: + log.debug("xref %s, jpeg, made larger - skip", xref) + opt_jpg.unlink() + opt_jpg = None + return xref, opt_jpg - compdata = leptonica.CompressedData.open(opt_jpg) - im_obj = pike.get_object(xref, 0) - im_obj.write(compdata.read(), filter=Name.DCTDecode) + +def transcode_jpegs( + pike: Pdf, jpegs: Sequence[Xref], root: Path, options, executor +) -> None: + def jpeg_args(): + for xref in jpegs: + in_jpg = jpg_name(root, xref) + opt_jpg = in_jpg.with_suffix('.opt.jpg') + yield xref, in_jpg, opt_jpg, options.jpeg_quality + + def finish_jpeg(result, pbar): + xref, opt_jpg = result + if opt_jpg: + compdata = leptonica.CompressedData.open(opt_jpg) + im_obj = pike.get_object(xref, 0) + im_obj.write(compdata.read(), filter=Name.DCTDecode) + pbar.update() + + executor( + use_threads=True, + max_workers=options.jobs, + tqdm_kwargs=dict( + desc="JPEGs", + total=len(jpegs), + unit='image', + disable=not options.progress_bar, + ), + task=_optimize_jpeg, + task_arguments=jpeg_args(), + task_finished=finish_jpeg, + ) def _transcode_png(pike: Pdf, filename: Path, xref: Xref) -> bool: @@ -598,7 +621,7 @@ def optimize( root.mkdir(exist_ok=True) jpegs, pngs = extract_images_generic(pike, root, options) - transcode_jpegs(pike, jpegs, root, options) + transcode_jpegs(pike, jpegs, root, options, executor) # if options.optimize >= 2: # Try pngifying the jpegs # transcode_pngs(pike, jpegs, jpg_name, root, options) From b1da09f141fe1c9af0aa6c7e3b4cc8b39b5c31ed Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 31 Jan 2021 02:21:03 -0800 Subject: [PATCH 18/27] Add plugin for setting logging console So that we are not tied to tqdm. --- docs/plugins.rst | 2 ++ src/ocrmypdf/__main__.py | 5 +++- src/ocrmypdf/_pipeline.py | 7 ++++-- src/ocrmypdf/api.py | 26 ++++++++++++++------- src/ocrmypdf/builtin_plugins/concurrency.py | 6 +++++ src/ocrmypdf/optimize.py | 2 +- src/ocrmypdf/pluginspec.py | 6 +++++ 7 files changed, 42 insertions(+), 12 deletions(-) diff --git a/docs/plugins.rst b/docs/plugins.rst index fb97f7f9..f1c3037c 100644 --- a/docs/plugins.rst +++ b/docs/plugins.rst @@ -136,6 +136,8 @@ Execution and progress reporting .. autoclass: ocrmypdf.pluginspec.Executor :members: +.. autofunction:: ocrmypdf.pluginspec.get_logging_console + .. autofunction:: ocrmypdf.pluginspec.get_executor .. autofunction:: ocrmypdf.pluginspec.get_progress_bar diff --git a/src/ocrmypdf/__main__.py b/src/ocrmypdf/__main__.py index 3c897b72..1046a50c 100755 --- a/src/ocrmypdf/__main__.py +++ b/src/ocrmypdf/__main__.py @@ -47,7 +47,10 @@ def run(args=None): verbosity = Verbosity.quiet options.progress_bar = False configure_logging( - verbosity, progress_bar_friendly=options.progress_bar, manage_root_logger=True + verbosity, + progress_bar_friendly=options.progress_bar, + manage_root_logger=True, + plugin_manager=plugin_manager, ) log.debug('ocrmypdf %s', __version__) try: diff --git a/src/ocrmypdf/_pipeline.py b/src/ocrmypdf/_pipeline.py index d4c171d8..30ab9751 100644 --- a/src/ocrmypdf/_pipeline.py +++ b/src/ocrmypdf/_pipeline.py @@ -19,7 +19,6 @@ import img2pdf import pikepdf from pikepdf.models.metadata import encode_pdf_date from PIL import Image, ImageColor, ImageDraw -from tqdm import tqdm from ocrmypdf import leptonica from ocrmypdf._concurrent import Executor @@ -726,7 +725,11 @@ def convert_to_pdfa(input_pdf: Path, input_ps_stub: Path, context: PdfContext): output_file=output_file, compression=options.pdfa_image_compression, pdfa_part=options.output_type[-1], # is pdfa-1, pdfa-2, or pdfa-3 - progressbar_class=tqdm if options.progress_bar else None, + progressbar_class=( + context.plugin_manager.hook.get_progress_bar() + if options.progress_bar + else None + ), ) return output_file diff --git a/src/ocrmypdf/api.py b/src/ocrmypdf/api.py index 9a37cad6..60de98fe 100644 --- a/src/ocrmypdf/api.py +++ b/src/ocrmypdf/api.py @@ -15,7 +15,10 @@ from pathlib import Path from typing import AnyStr, BinaryIO, Iterable, Optional, Union from warnings import warn -from ocrmypdf._logging import PageNumberFilter, TqdmConsole +from ocrmypdf._logging import ( # pylint: disable=unused-import + PageNumberFilter, + TqdmConsole, +) from ocrmypdf._plugin_manager import get_plugin_manager from ocrmypdf._sync import run_pipeline from ocrmypdf._validation import check_options @@ -47,6 +50,7 @@ def configure_logging( verbosity: Verbosity, progress_bar_friendly: bool = True, manage_root_logger: bool = False, + plugin_manager=None, ): """Set up logging. @@ -74,12 +78,13 @@ def configure_logging( their own debug logging. Args: - verbosity (Verbosity): Verbosity level. - progress_bar_friendly (bool): Install the TqdmConsole log handler, which is + verbosity: Verbosity level. + progress_bar_friendly: Install the TqdmConsole log handler, which is compatible with the tqdm progress bar; without this log messages will - overwrite the progress bar - manage_root_logger (bool): Configure the process's root logger, to ensure + overwrite the progress bar. + manage_root_logger: Configure the process's root logger, to ensure all log output is sent through + plugin_manager: The plugin manager. Returns: The toplevel logger for ocrmypdf (or the root logger, if we are managing it). @@ -90,8 +95,8 @@ def configure_logging( log = logging.getLogger(prefix) log.setLevel(logging.DEBUG) - if progress_bar_friendly: - console = logging.StreamHandler(stream=TqdmConsole(sys.stderr)) + if plugin_manager and progress_bar_friendly: + console = plugin_manager.hook.get_logging_console() else: console = logging.StreamHandler(stream=sys.stderr) @@ -245,6 +250,7 @@ def ocr( # pylint: disable=unused-argument user_patterns: os.PathLike = None, fast_web_view: float = None, plugins: Iterable[StrPath] = None, + plugin_manager=None, keep_temporary_files: bool = None, progress_bar: bool = None, **kwargs, @@ -296,6 +302,9 @@ def ocr( # pylint: disable=unused-argument Returns: :class:`ocrmypdf.ExitCode` """ + if plugins and plugin_manager: + raise ValueError("plugins= and plugin_manager are mutually exclusive") + if not plugins: plugins = [] elif isinstance(plugins, (str, Path)): @@ -315,7 +324,8 @@ def ocr( # pylint: disable=unused-argument # they might install different plugins, and generally speaking we have areas # of code that use global state. - plugin_manager = get_plugin_manager(plugins) + if not plugin_manager: + plugin_manager = get_plugin_manager(plugins) plugin_manager.hook.add_options(parser=parser) # pylint: disable=no-member if 'verbose' in kwargs: diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py index df6d42c6..5bfc05b3 100644 --- a/src/ocrmypdf/builtin_plugins/concurrency.py +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -27,6 +27,7 @@ from typing import Callable, Iterable, Optional, Union from tqdm import tqdm from ocrmypdf import Executor, hookimpl +from ocrmypdf._logging import TqdmConsole from ocrmypdf.exceptions import InputFileError Queue = Union[multiprocessing.Queue, queue.Queue] @@ -161,3 +162,8 @@ def get_executor(): @hookimpl def get_progress_bar(): return tqdm + + +@hookimpl +def get_logging_console(): + return logging.StreamHandler(stream=TqdmConsole(sys.stderr)) diff --git a/src/ocrmypdf/optimize.py b/src/ocrmypdf/optimize.py index 112a26c3..7acb9241 100644 --- a/src/ocrmypdf/optimize.py +++ b/src/ocrmypdf/optimize.py @@ -423,7 +423,7 @@ def transcode_jpegs( pbar.update() executor( - use_threads=True, + use_threads=True, # Processes are significantly slower at this task max_workers=options.jobs, tqdm_kwargs=dict( desc="JPEGs", diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index cf68fd77..ee4e8dc4 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -8,6 +8,7 @@ from abc import ABC, abstractmethod, abstractstaticmethod from argparse import ArgumentParser, Namespace from collections import namedtuple +from logging import Handler from pathlib import Path from typing import TYPE_CHECKING, AbstractSet, Callable, Iterable, List, Optional @@ -27,6 +28,11 @@ hookspec = pluggy.HookspecMarker('ocrmypdf') # pylint: disable=unused-argument +@hookspec +def get_logging_console() -> Handler: + """Returns a logging handler. Should be configured to handle progress bars.""" + + @hookspec def add_options(parser: ArgumentParser) -> None: """Allows the plugin to add its own command line and API arguments. From dccdcfaa913db3c4f4258927a806f7eb36a8ed36 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 31 Jan 2021 02:46:09 -0800 Subject: [PATCH 19/27] leptonica: tidy --- src/ocrmypdf/leptonica.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/ocrmypdf/leptonica.py b/src/ocrmypdf/leptonica.py index 69f6e444..c8759cfa 100644 --- a/src/ocrmypdf/leptonica.py +++ b/src/ocrmypdf/leptonica.py @@ -13,10 +13,8 @@ import argparse import logging import os -import platform import sys import threading -import warnings from collections import deque from collections.abc import Sequence from contextlib import suppress @@ -25,6 +23,7 @@ from functools import lru_cache from io import BytesIO, UnsupportedOperation from os import fspath from tempfile import TemporaryFile +from warnings import warn from ocrmypdf.exceptions import MissingDependencyError from ocrmypdf.lib._leptonica import ffi @@ -390,7 +389,7 @@ class Pix(LeptonicaObject): @classmethod def read(cls, path): - warnings.warn('Use Pix.open() instead', DeprecationWarning) + warn('Use Pix.open() instead', DeprecationWarning) return cls.open(path) @classmethod From 85c6a974ca4ef8335d6e413ad008d959535d951a Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 31 Jan 2021 02:46:44 -0800 Subject: [PATCH 20/27] Fix calls to hook.get_executor --- docs/plugins.rst | 2 +- src/ocrmypdf/_concurrent.py | 4 ++-- src/ocrmypdf/_pipeline.py | 2 +- src/ocrmypdf/builtin_plugins/concurrency.py | 6 +++--- src/ocrmypdf/pluginspec.py | 8 ++++---- 5 files changed, 11 insertions(+), 11 deletions(-) diff --git a/docs/plugins.rst b/docs/plugins.rst index f1c3037c..fad890e7 100644 --- a/docs/plugins.rst +++ b/docs/plugins.rst @@ -140,7 +140,7 @@ Execution and progress reporting .. autofunction:: ocrmypdf.pluginspec.get_executor -.. autofunction:: ocrmypdf.pluginspec.get_progress_bar +.. autofunction:: ocrmypdf.pluginspec.get_progressbar_class Applying special behavior before processing ------------------------------------------- diff --git a/src/ocrmypdf/_concurrent.py b/src/ocrmypdf/_concurrent.py index ea975bae..5882d962 100644 --- a/src/ocrmypdf/_concurrent.py +++ b/src/ocrmypdf/_concurrent.py @@ -107,8 +107,8 @@ class Executor(ABC): def setup_executor(plugin_manager) -> Executor: - pbar_class = plugin_manager.hook.get_progress_bar() - return plugin_manager.hook.get_executor(pbar_class=pbar_class) + pbar_class = plugin_manager.hook.get_progressbar_class() + return plugin_manager.hook.get_executor(progressbar_class=pbar_class) class SerialExecutor(Executor): diff --git a/src/ocrmypdf/_pipeline.py b/src/ocrmypdf/_pipeline.py index 30ab9751..0871aeba 100644 --- a/src/ocrmypdf/_pipeline.py +++ b/src/ocrmypdf/_pipeline.py @@ -726,7 +726,7 @@ def convert_to_pdfa(input_pdf: Path, input_ps_stub: Path, context: PdfContext): compression=options.pdfa_image_compression, pdfa_part=options.output_type[-1], # is pdfa-1, pdfa-2, or pdfa-3 progressbar_class=( - context.plugin_manager.hook.get_progress_bar() + context.plugin_manager.hook.get_progressbar_class() if options.progress_bar else None ), diff --git a/src/ocrmypdf/builtin_plugins/concurrency.py b/src/ocrmypdf/builtin_plugins/concurrency.py index 5bfc05b3..b58211e4 100644 --- a/src/ocrmypdf/builtin_plugins/concurrency.py +++ b/src/ocrmypdf/builtin_plugins/concurrency.py @@ -155,12 +155,12 @@ class StandardExecutor(Executor): @hookimpl -def get_executor(): - return StandardExecutor() +def get_executor(progressbar_class): + return StandardExecutor(pbar_class=progressbar_class) @hookimpl -def get_progress_bar(): +def get_progressbar_class(): return tqdm diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index ee4e8dc4..3ed8c4af 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -28,7 +28,7 @@ hookspec = pluggy.HookspecMarker('ocrmypdf') # pylint: disable=unused-argument -@hookspec +@hookspec(firstresult=True) def get_logging_console() -> Handler: """Returns a logging handler. Should be configured to handle progress bars.""" @@ -70,7 +70,7 @@ def check_options(options: Namespace) -> None: @hookspec(firstresult=True) -def get_executor() -> Executor: +def get_executor(progressbar_class) -> Executor: """Called to obtain an object that manages parallel execution. This may be used to replace OCRmyPDF's default parallel execution system @@ -92,7 +92,7 @@ def get_executor() -> Executor: @hookspec(firstresult=True) -def get_progress_bar(): +def get_progressbar_class(): """Called to obtain a class that can be used to create progress bars. The class should follow a tqdm-like protocol. Calling the class should return @@ -108,7 +108,7 @@ def get_progress_bar(): Here is how OCRmyPDF will use the progress bar: Example: - pbar_class = pm.hook.get_progress_bar() + pbar_class = pm.hook.get_progressbar_class() with pbar_class(**tqdm_kwargs) as pbar: ... pbar.update(1) From 6c8f9223e9e3e18f95ccd9cd60c77e0b6a6d86d0 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 31 Jan 2021 03:00:05 -0800 Subject: [PATCH 21/27] Update awslambda to new pluginspec --- src/ocrmypdf/api.py | 4 +- src/ocrmypdf/extra_plugins/awslambda.py | 74 ++++++++++++------------- 2 files changed, 37 insertions(+), 41 deletions(-) diff --git a/src/ocrmypdf/api.py b/src/ocrmypdf/api.py index 60de98fe..859e4a82 100644 --- a/src/ocrmypdf/api.py +++ b/src/ocrmypdf/api.py @@ -95,9 +95,11 @@ def configure_logging( log = logging.getLogger(prefix) log.setLevel(logging.DEBUG) + console = None if plugin_manager and progress_bar_friendly: console = plugin_manager.hook.get_logging_console() - else: + + if not console: console = logging.StreamHandler(stream=sys.stderr) if verbosity < 0: diff --git a/src/ocrmypdf/extra_plugins/awslambda.py b/src/ocrmypdf/extra_plugins/awslambda.py index 64297f67..23fe9799 100644 --- a/src/ocrmypdf/extra_plugins/awslambda.py +++ b/src/ocrmypdf/extra_plugins/awslambda.py @@ -20,6 +20,7 @@ from typing import Callable, Iterable, Optional from unittest.mock import Mock from ocrmypdf import Executor, hookimpl +from ocrmypdf._concurrent import NullProgressBar from ocrmypdf.exceptions import InputFileError @@ -98,24 +99,6 @@ class LambdaExecutor(Executor): task_finished(result, self.pbar_class) return - self._lambda_pool_impl( - max_workers=max_workers, - worker_initializer=worker_initializer, - task=task, - task_arguments=task_arguments, - task_finished=task_finished, - ) - - def _lambda_pool_impl( - self, - *, - max_workers: int, - worker_initializer: Callable, - task: Callable, - task_arguments: Iterable, - task_finished: Callable, - pbar, - ): task_arguments = list(task_arguments) grouped_args = list( zip_longest(*list(split_every(max_workers, task_arguments))) @@ -146,32 +129,43 @@ class LambdaExecutor(Executor): for process in processes: process.start() - while connections: - for r in wait(connections): - try: - msg_type, msg = r.recv() - except EOFError: - connections.remove(r) - continue + with self.pbar_class(**tqdm_kwargs) as pbar: + while connections: + for r in wait(connections): + try: + msg_type, msg = r.recv() + except EOFError: + connections.remove(r) + continue - if msg_type == MessageType.result: - if task_finished: - task_finished(msg, pbar) - elif msg_type == 'log': - record = msg - logger = logging.getLogger(record.name) - logger.handle(record) - elif msg_type == MessageType.complete: - connections.remove(r) - elif msg_type == MessageType.exception: - for process in processes: - process.terminate() - raise msg + if msg_type == MessageType.result: + if task_finished: + task_finished(msg, pbar) + elif msg_type == 'log': + record = msg + logger = logging.getLogger(record.name) + logger.handle(record) + elif msg_type == MessageType.complete: + connections.remove(r) + elif msg_type == MessageType.exception: + for process in processes: + process.terminate() + raise msg for process in processes: process.join() @hookimpl -def get_executor(): - return LambdaExecutor() +def get_executor(progressbar_class): + return LambdaExecutor(pbar_class=progressbar_class) + + +@hookimpl +def get_logging_console(): + return logging.StreamHandler() + + +@hookimpl +def get_progressbar_class(): + return NullProgressBar From 206c675df68bbca06a7358b6c7a137d5f27d3032 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 31 Jan 2021 19:26:35 -0800 Subject: [PATCH 22/27] docs: api --- src/ocrmypdf/api.py | 17 +++++++++-------- src/ocrmypdf/pluginspec.py | 28 +++++++++++++++++++++++++--- 2 files changed, 34 insertions(+), 11 deletions(-) diff --git a/src/ocrmypdf/api.py b/src/ocrmypdf/api.py index 859e4a82..9bce5352 100644 --- a/src/ocrmypdf/api.py +++ b/src/ocrmypdf/api.py @@ -55,12 +55,15 @@ def configure_logging( """Set up logging. Before calling :func:`ocrmypdf.ocr()`, you can use this function to - configure logging, if you want ocrmypdf's output to look like the ocrmypdf + configure logging if you want ocrmypdf's output to look like the ocrmypdf command line interface. It will register log handlers, log filters, and formatters, configure color logging to standard error, and adjust the log levels of third party libraries. Details of this are fine-tuned and subject to change. The ``verbosity`` argument is equivalent to the argument - ``--verbose`` and applies those settings. + ``--verbose`` and applies those settings. If you have a wrapper + script for ocrmypdf and you want it to be very similar to ocrmypdf, use this + function; if you are using ocrmypdf as part of an application that manages + its own logging, you probably do not want this function. If this function is not called, ocrmypdf will not configure logging, and it is up to the caller of ``ocrmypdf.ocr()`` to set up logging as it wishes using @@ -79,12 +82,10 @@ def configure_logging( Args: verbosity: Verbosity level. - progress_bar_friendly: Install the TqdmConsole log handler, which is - compatible with the tqdm progress bar; without this log messages will - overwrite the progress bar. - manage_root_logger: Configure the process's root logger, to ensure - all log output is sent through - plugin_manager: The plugin manager. + progress_bar_friendly: If True (the default), install a custom log handler + that is compatible with progress bars and colored output. + manage_root_logger: Configure the process's root logger. + plugin_manager: The plugin manager, used for obtaining the custom log handler. Returns: The toplevel logger for ocrmypdf (or the root logger, if we are managing it). diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index 3ed8c4af..d0c10239 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -30,7 +30,14 @@ hookspec = pluggy.HookspecMarker('ocrmypdf') @hookspec(firstresult=True) def get_logging_console() -> Handler: - """Returns a logging handler. Should be configured to handle progress bars.""" + """Returns a custom logging handler. + + Generally this is necessary when both logging output and a progress bar are both + outputting to ``sys.stderr``. + + Note: + This is a :ref:`firstresult hook`. + """ @hookspec @@ -78,11 +85,16 @@ def get_executor(progressbar_class) -> Executor: distributed environment. OCRmyPDF's executors are analogous to the standard Python executors in - ``conconcurrent.futures``, but they do not work the same way. + ``conconcurrent.futures``, but they do not work the same way. Executors may + be reused for different, unrelated batch operations, since all of the context + for a given job are passed to :meth:`Executor.__call__`. Should be of type :class:`Executor` or otherwise conforming to the protocol of that call. + Arguments: + progressbar_class: A progress bar class, which will be created when + Note: This hook will be called from the main process, and may modify global state before child worker processes are forked. @@ -93,11 +105,15 @@ def get_executor(progressbar_class) -> Executor: @hookspec(firstresult=True) def get_progressbar_class(): - """Called to obtain a class that can be used to create progress bars. + """Called to obtain a class that can be used to monitor progress. + + A progress bar is assumed, but this could be used for any type of monitoring. The class should follow a tqdm-like protocol. Calling the class should return a new progress bar object, which is activated with ``__enter__`` and terminated ``__exit__``. An update method is called whenever the progress bar is updated. + Progress bar objects will not be reused; a new one will be created for each + group of tasks. The progress bar is held in the main process/thread and not updated by child process/threads. When a child notifies the parent of completed work, the @@ -105,6 +121,12 @@ def get_progressbar_class(): The arguments are the same as `tqdm `_ accepts. + Progress bars should never write to ``sys.stdout``, or they will corrupt the + output if OCRmyPDF writes a PDF to standard output. + + The type of events that OCRmyPDF reports to a progress bar may change in + minor releases. + Here is how OCRmyPDF will use the progress bar: Example: From a48ca556c7f568d3245096219c0a1c34be184803 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 14 Feb 2021 01:22:33 -0800 Subject: [PATCH 23/27] Add filter_pdf_page hook --- src/ocrmypdf/_pipeline.py | 5 ++++ src/ocrmypdf/pluginspec.py | 54 ++++++++++++++++++++++++++++++++++---- 2 files changed, 54 insertions(+), 5 deletions(-) diff --git a/src/ocrmypdf/_pipeline.py b/src/ocrmypdf/_pipeline.py index 0871aeba..a5b8bc5a 100644 --- a/src/ocrmypdf/_pipeline.py +++ b/src/ocrmypdf/_pipeline.py @@ -610,6 +610,11 @@ def create_pdf_page_from_image(image: Path, page_context: PageContext): imfile, with_pdfrw=False, layout_fun=layout_fun, outputstream=pdf ) log.debug('convert done') + + output_file = page_context.plugin_manager.hook.filter_pdf_page( + page=page_context, image_filename=image, output_pdf=output_file + ) + return output_file diff --git a/src/ocrmypdf/pluginspec.py b/src/ocrmypdf/pluginspec.py index d0c10239..5fa0b3ca 100644 --- a/src/ocrmypdf/pluginspec.py +++ b/src/ocrmypdf/pluginspec.py @@ -238,11 +238,11 @@ def filter_page_image(page: 'PageContext', image_filename: Path) -> Path: will be resized and the OCR layer misaligned. OCRmyPDF does not nothing to enforce these constraints; it is up to the plugin to do sensible things. - OCRmyPDF will create the PDF page based on the image format used. If you - convert the image to a JPEG, the output page will be created as a JPEG, etc. - If you change the colorspace, that change will be kept. Note that the - OCRmyPDF image optimization stage, if enabled, may ultimately chose a - different format. + OCRmyPDF will create the PDF page based on the image format used (unless the + hook is overriden). If you convert the image to a JPEG, the output page will + be created as a JPEG, etc. If you change the colorspace, that change will be + kept. Note that the OCRmyPDF image optimization stage, if enabled, may + ultimately chose a different format. If the return value is a file that does not exist, ``FileNotFoundError`` will occur. The return value should be a path to a file in the same folder @@ -260,6 +260,50 @@ def filter_page_image(page: 'PageContext', image_filename: Path) -> Path: """ +@hookspec(firstresult=True) +def filter_pdf_page( + page: 'PageContext', image_filename: Path, output_pdf: Path +) -> Path: + """Called to convert a filtered whole page image into a PDF. + + A whole page image is only produced when preprocessing command line arguments + are issued or when ``--force-ocr`` is issued. If no whole page is image is + produced for a given page, this function will not be called. This is not + the image that will be shown to OCR. The whole page image is filtered in + the hook above, ``filter_page_image``, then this function is called for + PDF conversion. + + This function will only be called when OCRmyPDF runs in a mode such as + "force OCR" mode where rasterizing of all content is performed. + + Clever things could be done at this stage such as segmenting the page image into + color regions or vector equivalents. + + The provider of the hook implementation is responsible for ensuring that the + OCR text layer is aligned with the PDF produced here, or text misalignment + will result. + + Currently this function must produce a single page PDF or the pipeline will + fail. If the intent is to remove the PDF, then create a single page empty + PDF. + + Args: + page: Context for this page. + image_filename: Filename of the input image used to create output_pdf, + for "reference" if recreating the output_pdf entirely. + output_pdf: The previous created output_pdf. + + Returns: + output_pdf + + Note: + This hook will be called from child processes. Modifying global state + will not affect the main process or other child processes. + Note: + This is a :ref:`firstresult hook`. + """ + + OrientationConfidence = namedtuple('OrientationConfidence', ('angle', 'confidence')) """Expresses an OCR engine's confidence in page rotation. From 18e613657cac71e7c12b0a3aa950747a181299b4 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Sun, 14 Feb 2021 01:23:01 -0800 Subject: [PATCH 24/27] docker-compose: fix typo --- misc/docker-compose.example.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/misc/docker-compose.example.yml b/misc/docker-compose.example.yml index 0db102b9..9668b2b9 100644 --- a/misc/docker-compose.example.yml +++ b/misc/docker-compose.example.yml @@ -9,7 +9,7 @@ services: - "/media/scan:/input" - "/mnt/scan:/output" environment: - - OCR_OUTPUT_DIRECTORY_YEAR_MONT=0 + - OCR_OUTPUT_DIRECTORY_YEAR_MONTH=0 user: ":" entrypoint: python3 command: watcher.py From 873f915212b1253936ae3c79d7dc737f5c0055ba Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Tue, 9 Mar 2021 23:24:10 -0800 Subject: [PATCH 25/27] docs: mention problems with Debian/Ubuntu and other tidying --- docs/installation.rst | 28 ++++++++++++++++++++++++---- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/docs/installation.rst b/docs/installation.rst index 4fc3738a..047d4bb5 100644 --- a/docs/installation.rst +++ b/docs/installation.rst @@ -158,7 +158,7 @@ To install ocrmypdf for the system: .. code-block:: bash - sudo pip3 install ocrmypdf + pip3 install ocrmypdf To install for the current user only: @@ -380,12 +380,16 @@ Install the following dependencies: To install ocrmypdf for the system: +.. code-block:: bash + # As root user pip3 install ocrmypdf ldconfig Or, to install for the current user only: +.. code-block:: bash + export PATH=$HOME/.local/bin:$PATH pip3 install --user ocrmypdf @@ -520,7 +524,7 @@ DLLs or other Windows patches, and may require a reboot. You may then use ``pip`` to install ocrmypdf. (This can performed by a user or Administrator.): -* ``pip install ocrmypdf +* ``pip install ocrmypdf`` Chocolatey automatically selects appropriate versions of these applications. If you are installing them manually, please install 64-bit versions of all applications for @@ -536,8 +540,9 @@ to change the PATH. .. warning:: As of early 2021, users have reported problems with the Microsoft Store version of - Python affected most third party Python packages including OCRmyPDF. Please use - Python downloaded from Python.org or Chocolatey as recommended here. + Python and OCRmyPDF. These issues affect many other third party Python packages. + Please download Python from Python.org or Chocolatey instead, and do not use the + Microsoft Store version. Windows Subsystem for Linux --------------------------- @@ -645,6 +650,21 @@ the latest version. However, PyPI and ``pip`` cannot address the fact that ``ocrmypdf`` depends on certain non-Python system libraries and programs being installed. +.. warning:: + + Debian and Ubuntu users: unfortunately, Debian and Ubuntu customize + Python in non-standard ways, and the nature of these customizations + varies from release to release. This can make for a frustrating + user experience. The instructions below work on almost all platforms that + have Python installed, except for Debian and Ubuntu, where you may need + to take additional steps. For best results on Debian and Ubuntu, use the + ``apt`` packages; or if these are too old, run + ``apt install python3-pip python3-venv``, create a virtual environment, + and install OCRmyPDF in that environment. + + `See here for more inforation on Debian-Python issues + `__. + For best results, first install `your platform's version `__ of ``ocrmypdf``, using the instructions elsewhere in this document. Then From e09ae9c68a3c244fd198cf8b525ad523d81634e3 Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Thu, 1 Apr 2021 16:25:40 -0700 Subject: [PATCH 26/27] Fix test suite failure if filter_pdf_page is missing --- src/ocrmypdf/builtin_plugins/default_filters.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 src/ocrmypdf/builtin_plugins/default_filters.py diff --git a/src/ocrmypdf/builtin_plugins/default_filters.py b/src/ocrmypdf/builtin_plugins/default_filters.py new file mode 100644 index 00000000..83a8f3fc --- /dev/null +++ b/src/ocrmypdf/builtin_plugins/default_filters.py @@ -0,0 +1,12 @@ +# © 2021 James R. Barlow: github.com/jbarlow83 +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +from ocrmypdf import hookimpl + + +@hookimpl +def filter_pdf_page(page, image_filename, output_pdf): + return output_pdf From 2e155c31bf42fe5c0a1efefdad1b49ece04d4fde Mon Sep 17 00:00:00 2001 From: "James R. Barlow" Date: Thu, 1 Apr 2021 16:30:42 -0700 Subject: [PATCH 27/27] Ensure builtin module registration is deterministic --- src/ocrmypdf/_plugin_manager.py | 4 +++- src/ocrmypdf/builtin_plugins/__init__.py | 4 ++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/ocrmypdf/_plugin_manager.py b/src/ocrmypdf/_plugin_manager.py index 76e12bd7..0cd69673 100644 --- a/src/ocrmypdf/_plugin_manager.py +++ b/src/ocrmypdf/_plugin_manager.py @@ -67,7 +67,9 @@ class OcrmypdfPluginManager(pluggy.PluginManager): # 1. Register builtins if self.__builtins: - for module in pkgutil.iter_modules(ocrmypdf.builtin_plugins.__path__): + for module in sorted( + pkgutil.iter_modules(ocrmypdf.builtin_plugins.__path__) + ): name = f'ocrmypdf.builtin_plugins.{module.name}' module = importlib.import_module(name) self.register(module) diff --git a/src/ocrmypdf/builtin_plugins/__init__.py b/src/ocrmypdf/builtin_plugins/__init__.py index 61f9f6d6..05d8c70e 100644 --- a/src/ocrmypdf/builtin_plugins/__init__.py +++ b/src/ocrmypdf/builtin_plugins/__init__.py @@ -3,3 +3,7 @@ # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this # file, You can obtain one at http://mozilla.org/MPL/2.0/. + +# This file exists only mark builtin_plugins as a package. +# The plugin manager will not load it, so anything defined here may not be +# processed as a module.