From 952e2db96d15443fa04013a1bdb0af320c74520b Mon Sep 17 00:00:00 2001 From: ahanden Date: Tue, 30 Jun 2026 21:14:58 -0500 Subject: [PATCH 1/5] Ditched monitor.py in favor of multiprocessing pools --- reditools/tools/analyze/main.py | 69 +++---------- reditools/tools/analyze/monitor.py | 59 ------------ reditools/tools/analyze/redi_thread.py | 128 +++++++++---------------- 3 files changed, 59 insertions(+), 197 deletions(-) delete mode 100644 reditools/tools/analyze/monitor.py diff --git a/reditools/tools/analyze/main.py b/reditools/tools/analyze/main.py index 2a489ba..ece9b23 100644 --- a/reditools/tools/analyze/main.py +++ b/reditools/tools/analyze/main.py @@ -2,14 +2,13 @@ import argparse import sys -from multiprocessing import Process, Queue +import traceback +from multiprocessing import Pool from reditools.logger import Logger -from reditools.region import Region from reditools.tools.analyze.concat_output import concat_output -from reditools.tools.analyze.monitor import monitor from reditools.tools.analyze.parse_args import parse_args -from reditools.tools.analyze.redi_thread import redi_thread +from reditools.tools.analyze.redi_thread import REDIThread from reditools.tools.analyze.region_args import region_args @@ -51,45 +50,6 @@ def setup_logger(options: argparse.Namespace) -> Logger: return Logger(Logger.info_level) return Logger(Logger.silent_level) -def fill_queue(options: argparse.Namespace) -> Queue[tuple[int, Region] | None]: - """ - Fill the input queue with genomic regions to be analyzed. - - Parameters - ---------- - options : argparse.Namespace - The parsed command line options. - - Returns - ------- - Queue[tuple[int, Region] | None] - A queue containing indexed Region objects. - - Raises - ------ - SystemExit - If a required file is not found. - """ - in_queue: Queue[tuple[int, Region] | None] = Queue() - try: - for _ in enumerate(region_args(options)): # noqa: WPS468 - in_queue.put(_) - except FileNotFoundError as exc: - sys.stderr.write(f'[ERROR] {exc}\n') - sys.exit(1) - - # Check thread count - if in_queue.qsize() < options.threads: - sys.stderr.write( - "[WARNING] You have assigned more threads " - f"({options.threads}) than there are genomic ranges " - f"({in_queue.qsize()})\n", - ) - options.threads = in_queue.qsize() - for _ in range(options.threads): - in_queue.put(None) - return in_queue - def main() -> None: """ The main entry point for the REDItools analyze command. @@ -107,19 +67,20 @@ def main() -> None: options.encoding = 'utf-8' - in_queue = fill_queue(options) - - # Start parallel jobs - out_queue: Queue[tuple[int, str]] = Queue() - processes = [] - for _ in range(options.threads): - processes.append(Process( - target=redi_thread, - args=(options, in_queue, out_queue), - )) + regions = region_args(options) + # Re-implement thread count warning here + try: + with Pool(options.threads, REDIThread.init, (options,)) as pool: + imap_iter = pool.imap(REDIThread.analyze, regions, 1) + temp_files = [imap_iter.next() for _ in range(len(regions))] + except Exception as exc: + if options.debug: + traceback.print_exception(*sys.exc_info()) + sys.stderr.write(f'[ERROR] ({type(exc)}) {exc}\n') + sys.exit(1) concat_output( - monitor(processes, out_queue, in_queue.qsize()), + temp_files, options.output_file, 'a' if options.append_file else 'w', options.encoding, diff --git a/reditools/tools/analyze/monitor.py b/reditools/tools/analyze/monitor.py deleted file mode 100644 index f1a8434..0000000 --- a/reditools/tools/analyze/monitor.py +++ /dev/null @@ -1,59 +0,0 @@ -from __future__ import annotations - -import sys -from multiprocessing import Process, Queue -from queue import Empty as EmptyQueueException - - -def check_dead(processes: list[Process]) -> None: - """Check if any of the processes have failed. - - If a process has exited with code 1, all other processes are killed - and the program exits. - - Parameters - ---------- - processes : list[Process] - The list of processes to monitor. - """ - for proc in processes: - if proc.exitcode == 1: - for to_kill in processes: - to_kill.kill() - sys.stderr.write('[ERROR] Killing job\n') - sys.exit(1) - -def monitor( - processes: list[Process], - out_queue: Queue[tuple[int, str]], - chunks: int, -) -> list[str]: - """Monitor progress of parallel analysis processes. - - Parameters - ---------- - processes : list[Process] - The list of worker processes. - out_queue : Queue[tuple[int, str]] - The queue containing analysis result filenames and their indices. - chunks : int - The total number of work chunks. - - Returns - ------- - list[str] - A list of filenames containing analysis results, ordered by chunk index. - """ - tfs = ['' for _ in range(chunks - len(processes))] - - for prc in processes: - prc.start() - - while '' in tfs: - try: - idx, fname = out_queue.get(block=False, timeout=1) - except EmptyQueueException: - check_dead(processes) - else: - tfs[idx] = fname - return tfs diff --git a/reditools/tools/analyze/redi_thread.py b/reditools/tools/analyze/redi_thread.py index 63e7b79..6c39d07 100644 --- a/reditools/tools/analyze/redi_thread.py +++ b/reditools/tools/analyze/redi_thread.py @@ -1,10 +1,5 @@ import argparse -import sys -import traceback -from multiprocessing import Queue -from reditools.alignment_manager import AlignmentManager -from reditools.reditools import REDItools from reditools.region import Region from reditools.tools.analyze.rtchecks import RTChecks from reditools.tools.analyze.setup_alignment_manager import \ @@ -13,82 +8,47 @@ from reditools.tools.analyze.write_results import write_results -def analyze( - options: argparse.Namespace, - rtools: REDItools, - sam_manager: AlignmentManager, - region: Region, - rtqc: RTChecks, -) -> str: - """Analyze a specific genomic region. - - Parameters - ---------- - options : argparse.Namespace - The command-line options. - rtools : REDItools - The REDItools analysis engine. - sam_manager : AlignmentManager - The alignment file manager. - region : Region - The genomic region to analyze. - rtqc : RTChecks - The quality control checks to apply. - - Returns - ------- - str - The path to the temporary file containing the results. - """ - rtresults = rtools.analyze(sam_manager, region) - return write_results( - rtresults, - options.temp_dir, - rtqc, - rtools.log, - ) - -def redi_thread( - options: argparse.Namespace, - in_queue: Queue, - out_queue: Queue, -) -> bool: - """Worker thread function for parallel REDItools analysis. - - Parameters - ---------- - options : argparse.Namespace - The command-line options. - in_queue : Queue - The queue containing genomic regions to analyze. - out_queue : Queue - The queue to put analysis results into. - - Returns - ------- - bool - True when the worker has finished processing all regions. - """ - rtools = setup_rtools(options) - sam_manager = setup_alignment_manager( - options.file, - options.min_read_quality, - options.min_read_length, - options.exclude_reads, - ) - rtqc = RTChecks(options) - while True: - args = in_queue.get() - if args is None: - return True - idx, region = args - try: # noqa: WPS229 - out_queue.put(( - idx, - analyze(options, rtools, sam_manager, region, rtqc), - )) - except Exception as exc: - if options.debug: - traceback.print_exception(*sys.exc_info()) - sys.stderr.write(f'[ERROR] ({type(exc)}) {exc}\n') - sys.exit(1) +class REDIThread: + @classmethod + def init(cls, options: argparse.Namespace) -> None: + """Worker thread function for parallel REDItools analysis. + + Parameters + ---------- + options : argparse.Namespace + The command-line options. + """ + cls.rtools = setup_rtools(options) + cls.sam_manager = setup_alignment_manager( + options.file, + options.min_read_quality, + options.min_read_length, + options.exclude_reads, + ) + cls.rtqc = RTChecks(options) + cls.temp_dir = options.temp_dir + + @classmethod + def analyze( + cls, + region: Region, + ) -> str: + """Analyze a specific genomic region. + + Parameters + ---------- + region : Region + The genomic region to analyze. + + Returns + ------- + str + The path to the temporary file containing the results. + """ + rtresults = cls.rtools.analyze(cls.sam_manager, region) + return write_results( + rtresults, + cls.temp_dir, + cls.rtqc, + cls.rtools.log, + ) From 184fd7859d03bf49037122d4ea1f56d43dedf69e Mon Sep 17 00:00:00 2001 From: ahanden Date: Thu, 2 Jul 2026 22:28:07 -0500 Subject: [PATCH 2/5] Improved error handling and re-added thread count warning --- reditools/tools/analyze/main.py | 29 ++++++++++++++++++++++------- 1 file changed, 22 insertions(+), 7 deletions(-) diff --git a/reditools/tools/analyze/main.py b/reditools/tools/analyze/main.py index ece9b23..ed87ea1 100644 --- a/reditools/tools/analyze/main.py +++ b/reditools/tools/analyze/main.py @@ -1,5 +1,7 @@ from __future__ import annotations +from multiprocessing.context import TimeoutError +from functools import partial import argparse import sys import traceback @@ -50,6 +52,12 @@ def setup_logger(options: argparse.Namespace) -> Logger: return Logger(Logger.info_level) return Logger(Logger.silent_level) +def pool_error(pool, debug, exc): + pool.terminate() + if debug: + raise exc.__cause__ + sys.stderr.write(f'[ERROR] ({type(exc)}) {exc}\n') + def main() -> None: """ The main entry point for the REDItools analyze command. @@ -68,15 +76,22 @@ def main() -> None: options.encoding = 'utf-8' regions = region_args(options) - # Re-implement thread count warning here + + if options.threads > len(regions): + sys.stderr.write( + f"[WARNING] You have assigned {options.threads} threads, " + f"But there are only {len(regions)} genomic range(s). " + "Consider change the value of --window\n" + ) + options.threads = len(regions) try: with Pool(options.threads, REDIThread.init, (options,)) as pool: - imap_iter = pool.imap(REDIThread.analyze, regions, 1) - temp_files = [imap_iter.next() for _ in range(len(regions))] - except Exception as exc: - if options.debug: - traceback.print_exception(*sys.exc_info()) - sys.stderr.write(f'[ERROR] ({type(exc)}) {exc}\n') + kill_pool = partial(pool_error, pool, options.debug) + imap_iter = [pool.apply_async(REDIThread.analyze, args=(region,), error_callback=kill_pool) for region in regions] + pool.close() + pool.join() + temp_files = [_.get(1) for _ in imap_iter] + except TimeoutError: sys.exit(1) concat_output( From 9fe81c6f44be8a19116b922ff235169082e4afb9 Mon Sep 17 00:00:00 2001 From: ahanden Date: Thu, 2 Jul 2026 22:34:45 -0500 Subject: [PATCH 3/5] Added check for single-thread error --- reditools/tools/analyze/main.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/reditools/tools/analyze/main.py b/reditools/tools/analyze/main.py index ed87ea1..a8c6a25 100644 --- a/reditools/tools/analyze/main.py +++ b/reditools/tools/analyze/main.py @@ -1,11 +1,10 @@ from __future__ import annotations -from multiprocessing.context import TimeoutError -from functools import partial import argparse import sys -import traceback +from functools import partial from multiprocessing import Pool +from multiprocessing.context import TimeoutError from reditools.logger import Logger from reditools.tools.analyze.concat_output import concat_output @@ -86,12 +85,17 @@ def main() -> None: options.threads = len(regions) try: with Pool(options.threads, REDIThread.init, (options,)) as pool: - kill_pool = partial(pool_error, pool, options.debug) - imap_iter = [pool.apply_async(REDIThread.analyze, args=(region,), error_callback=kill_pool) for region in regions] + imap_iter = [ + pool.apply_async( + REDIThread.analyze, + args=(region,), + error_callback=partial(pool_error, pool, options.debug), + ) for region in regions + ] pool.close() pool.join() temp_files = [_.get(1) for _ in imap_iter] - except TimeoutError: + except (TimeoutError, IndexError): sys.exit(1) concat_output( From cefe09ad5370c336b0f8ace9b828eb5883496a76 Mon Sep 17 00:00:00 2001 From: ahanden Date: Thu, 2 Jul 2026 22:59:52 -0500 Subject: [PATCH 4/5] Wrapped REDIthread for mypy compliance --- reditools/tools/analyze/main.py | 12 ++++---- reditools/tools/analyze/redi_thread.py | 39 ++++++++++++++++++-------- 2 files changed, 33 insertions(+), 18 deletions(-) diff --git a/reditools/tools/analyze/main.py b/reditools/tools/analyze/main.py index a8c6a25..d4acb88 100644 --- a/reditools/tools/analyze/main.py +++ b/reditools/tools/analyze/main.py @@ -3,13 +3,13 @@ import argparse import sys from functools import partial -from multiprocessing import Pool from multiprocessing.context import TimeoutError +from multiprocessing.pool import Pool from reditools.logger import Logger from reditools.tools.analyze.concat_output import concat_output from reditools.tools.analyze.parse_args import parse_args -from reditools.tools.analyze.redi_thread import REDIThread +from reditools.tools.analyze.redi_thread import REDIThreadManager from reditools.tools.analyze.region_args import region_args @@ -51,10 +51,10 @@ def setup_logger(options: argparse.Namespace) -> Logger: return Logger(Logger.info_level) return Logger(Logger.silent_level) -def pool_error(pool, debug, exc): +def pool_error(pool: Pool, debug: bool, exc: Exception) -> None: pool.terminate() if debug: - raise exc.__cause__ + raise exc.__cause__ # type: ignore[misc] sys.stderr.write(f'[ERROR] ({type(exc)}) {exc}\n') def main() -> None: @@ -84,10 +84,10 @@ def main() -> None: ) options.threads = len(regions) try: - with Pool(options.threads, REDIThread.init, (options,)) as pool: + with Pool(options.threads, REDIThreadManager.init, (options,)) as pool: imap_iter = [ pool.apply_async( - REDIThread.analyze, + REDIThreadManager.analyze, args=(region,), error_callback=partial(pool_error, pool, options.debug), ) for region in regions diff --git a/reditools/tools/analyze/redi_thread.py b/reditools/tools/analyze/redi_thread.py index 6c39d07..b6203fc 100644 --- a/reditools/tools/analyze/redi_thread.py +++ b/reditools/tools/analyze/redi_thread.py @@ -1,5 +1,7 @@ import argparse +from reditools.alignment_manager import AlignmentManager +from reditools.reditools import REDItools from reditools.region import Region from reditools.tools.analyze.rtchecks import RTChecks from reditools.tools.analyze.setup_alignment_manager import \ @@ -9,8 +11,7 @@ class REDIThread: - @classmethod - def init(cls, options: argparse.Namespace) -> None: + def __init__(self, options: argparse.Namespace) -> None: """Worker thread function for parallel REDItools analysis. Parameters @@ -18,19 +19,18 @@ def init(cls, options: argparse.Namespace) -> None: options : argparse.Namespace The command-line options. """ - cls.rtools = setup_rtools(options) - cls.sam_manager = setup_alignment_manager( + self.rtools = setup_rtools(options) + self.sam_manager = setup_alignment_manager( options.file, options.min_read_quality, options.min_read_length, options.exclude_reads, ) - cls.rtqc = RTChecks(options) - cls.temp_dir = options.temp_dir + self.rtqc = RTChecks(options) + self.temp_dir = options.temp_dir - @classmethod def analyze( - cls, + self, region: Region, ) -> str: """Analyze a specific genomic region. @@ -45,10 +45,25 @@ def analyze( str The path to the temporary file containing the results. """ - rtresults = cls.rtools.analyze(cls.sam_manager, region) + rtresults = self.rtools.analyze(self.sam_manager, region) return write_results( rtresults, - cls.temp_dir, - cls.rtqc, - cls.rtools.log, + self.temp_dir, + self.rtqc, + self.rtools.log, ) + +class REDIThreadManager: + thread: REDIThread | None = None + + @classmethod + def init(cls, options: argparse.Namespace) -> None: + cls.thread = REDIThread(options) + + @classmethod + def analyze(cls, region: Region) -> str: + if cls.thread is None: + raise AttributeError('REDIThreadManager not initialized.') + return cls.thread.analyze(region) + + From ec4361e16d8928d349e6f959cdfdbff64c85e24b Mon Sep 17 00:00:00 2001 From: ahanden Date: Thu, 2 Jul 2026 23:05:05 -0500 Subject: [PATCH 5/5] Added documentation --- reditools/tools/analyze/main.py | 18 ++++++++++++++- reditools/tools/analyze/redi_thread.py | 31 +++++++++++++++++++++----- 2 files changed, 42 insertions(+), 7 deletions(-) diff --git a/reditools/tools/analyze/main.py b/reditools/tools/analyze/main.py index d4acb88..fd7a2bc 100644 --- a/reditools/tools/analyze/main.py +++ b/reditools/tools/analyze/main.py @@ -52,6 +52,18 @@ def setup_logger(options: argparse.Namespace) -> Logger: return Logger(Logger.silent_level) def pool_error(pool: Pool, debug: bool, exc: Exception) -> None: + """ + Terminates a multiprocessing Pool. + + Parameters + ---------- + pool : Pool + mutliprocessing Pool to terminate. + debug : bool + If True, raises the exception passed in the third argument. + exc : Exception + Exception responsible for the pool to terminate. + """ pool.terminate() if debug: raise exc.__cause__ # type: ignore[misc] @@ -84,7 +96,11 @@ def main() -> None: ) options.threads = len(regions) try: - with Pool(options.threads, REDIThreadManager.init, (options,)) as pool: + with Pool( + options.threads, + REDIThreadManager.init_thread, + (options,), + ) as pool: imap_iter = [ pool.apply_async( REDIThreadManager.analyze, diff --git a/reditools/tools/analyze/redi_thread.py b/reditools/tools/analyze/redi_thread.py index b6203fc..fd53197 100644 --- a/reditools/tools/analyze/redi_thread.py +++ b/reditools/tools/analyze/redi_thread.py @@ -1,7 +1,5 @@ import argparse -from reditools.alignment_manager import AlignmentManager -from reditools.reditools import REDItools from reditools.region import Region from reditools.tools.analyze.rtchecks import RTChecks from reditools.tools.analyze.setup_alignment_manager import \ @@ -54,16 +52,37 @@ def analyze( ) class REDIThreadManager: - thread: REDIThread | None = None + """Manages a worker thread function for parallel REDItools analysis.""" + + thread: None | REDIThread = None @classmethod - def init(cls, options: argparse.Namespace) -> None: + def init_thread(cls, options: argparse.Namespace) -> None: + """Initialize a REDIThread. + + Parameters + ---------- + options : argparse.Namespace + The command-line options. + """ + cls.thread = REDIThread(options) @classmethod def analyze(cls, region: Region) -> str: + """Instruct thread to analyze a specific genomic region. + + Parameters + ---------- + region : Region + The genomic region to analyze. + + Returns + ------- + str + The path to the temporary file containing the results. + """ + if cls.thread is None: raise AttributeError('REDIThreadManager not initialized.') return cls.thread.analyze(region) - -