revamp logging for python3.14

Author: andrewprzh

isoquant.py

@@ -47,6 +47,7 @@
 from src.multimap_resolver import MultimapResolvingStrategy
 from src.stats import combine_counts
 from src.barcode_calling import process_single_thread, process_in_parallel, get_umi_length
+from src.common import setup_worker_logging, _get_log_params
 
 
 logger = logging.getLogger('IsoQuant')
@@ -676,38 +677,17 @@ def create_output_dirs(args):
             os.makedirs(sample_aux_dir)
 
 
-def set_logger(args, logger_instance):
-    if "debug" not in args.__dict__ or not args.debug:
-        output_level = logging.INFO
-    else:
-        output_level = logging.DEBUG
-
-    logger_instance.setLevel(output_level)
+def set_logger(args):
+    output_level = logging.DEBUG if args.__dict__.get('debug') else logging.INFO
     log_file = os.path.join(args.output, "isoquant.log")
     if os.path.exists(log_file):
         old_log_file = os.path.join(args.output, "isoquant.log.old")
         with open(old_log_file, "a") as olf:
             olf.write("\n")
             shutil.copyfileobj(open(log_file, "r"), olf)
-
-    f = open(log_file, "w")
-    f.write("Command line: " + args._cmd_line + '\n')
-    f.close()
-    fh = logging.FileHandler(log_file)
-    fh.set_name("isoquant_file_log")
-    fh.setLevel(output_level)
-    ch = logging.StreamHandler(sys.stdout)
-    ch.set_name("isoquant_screen_log")
-    ch.setLevel(logging.INFO)
-
-    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
-    fh.setFormatter(formatter)
-    ch.setFormatter(formatter)
-    if all(fh.get_name() != h.get_name() for h in logger_instance.handlers):
-        logger_instance.addHandler(fh)
-    if all(ch.get_name() != h.get_name() for h in logger_instance.handlers):
-        logger_instance.addHandler(ch)
-
+    with open(log_file, "w") as f:
+        f.write("Command line: " + args._cmd_line + '\n')
+    setup_worker_logging(log_file, output_level)
     logger.info("Running IsoQuant version " + args._version)
 
 
@@ -1013,7 +993,10 @@ def call_barcodes(args):
             # Read chunks are not cleared by the GC in the end of barcode calling, leaving the main
             # IsoQuant process to consume ~2,5 GB even when barcode calling is done.
             # Once 16 child processes are created later, IsoQuant instantly takes threads x 2,5 GB for nothing.
-            with ProcessPoolExecutor(max_workers=1) as proc:
+            log_file, log_level = _get_log_params()
+            with ProcessPoolExecutor(max_workers=1,
+                                     initializer=setup_worker_logging,
+                                     initargs=(log_file, log_level)) as proc:
                 logger.info("Detecting barcodes for %d file(s)" % len(input_files))
                 if bc_threads == 1:
                     future_res = proc.submit(process_single_thread, bc_args)
@@ -1117,7 +1100,7 @@ def main(cmd_args):
     if not cmd_args:
         parser.print_usage()
         sys.exit(IsoQuantExitCode.SUCCESS)
-    set_logger(args, logger)
+    set_logger(args)
     args = check_and_load_args(args, parser)
     create_output_dirs(args)
     set_additional_params(args)

src/barcode_calling/detect_barcodes.py

@@ -27,6 +27,7 @@
 
 from ..modes import IsoQuantMode
 from ..error_codes import IsoQuantExitCode
+from ..common import setup_worker_logging, _get_log_params
 from .common import reverese_complement, load_barcodes
 from . import (
     TenXBarcodeDetector,
@@ -369,8 +370,14 @@ def _process_single_file_in_parallel(input_file, output_tsv, out_fasta, args, ba
     # Clean up parent memory before spawning workers
     gc.collect()
     mp_context = multiprocessing.get_context('spawn')
+    log_file, log_level = _get_log_params()
     # max_tasks_per_child requires Python 3.11+
-    executor_kwargs = {'max_workers': args.threads, 'mp_context': mp_context}
+    executor_kwargs = {
+        'max_workers': args.threads,
+        'mp_context': mp_context,
+        'initializer': setup_worker_logging,
+        'initargs': (log_file, log_level),
+    }
     if sys.version_info >= (3, 11):
         executor_kwargs['max_tasks_per_child'] = 20
     with concurrent.futures.ProcessPoolExecutor(**executor_kwargs) as proc:

src/common.py

@@ -9,13 +9,53 @@
 import os
 import re
 import subprocess
+import sys
 import math
 from collections import defaultdict
 from enum import Enum
 
 logger = logging.getLogger('IsoQuant')
 
 
+def _get_log_params():
+    """Return (log_file, log_level) from the current IsoQuant logger configuration."""
+    w = logging.getLogger('IsoQuant')
+    log_level = w.level or logging.INFO
+    log_file = None
+    for h in w.handlers:
+        if hasattr(h, 'baseFilename'):
+            log_file = h.baseFilename
+            break
+    return log_file, log_level
+
+
+def setup_worker_logging(log_file, log_level):
+    """Initialize IsoQuant logging in a worker process.
+
+    Must be passed as the 'initializer' argument to every ProcessPoolExecutor.
+    Works for both 'fork' (clears broken inherited handlers) and 'spawn'
+    (creates handlers from scratch). Python 3.14 changed the default start
+    method from 'fork' to 'spawn' on Linux, and also changed _at_fork_reinit
+    behaviour so inherited handlers can break even with explicit fork.
+    """
+    w = logging.getLogger('IsoQuant')
+    for h in w.handlers[:]:
+        w.removeHandler(h)
+    w.setLevel(log_level)
+    formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
+    if log_file:
+        fh = logging.FileHandler(log_file, mode='a')
+        fh.set_name("isoquant_file_log")
+        fh.setLevel(log_level)
+        fh.setFormatter(formatter)
+        w.addHandler(fh)
+    ch = logging.StreamHandler(sys.stdout)
+    ch.set_name("isoquant_screen_log")
+    ch.setLevel(logging.INFO)
+    ch.setFormatter(formatter)
+    w.addHandler(ch)
+
+
 class CigarEvent(Enum):
     match = 0
     insertion = 1

src/dataset_processor.py

@@ -22,7 +22,7 @@
 from pyfaidx import Fasta
 
 from .modes import IsoQuantMode
-from .common import proper_plural_form, large_output_enabled
+from .common import proper_plural_form, large_output_enabled, setup_worker_logging, _get_log_params
 from .error_codes import IsoQuantExitCode
 from .serialization import *
 from .stats import EnumStats
@@ -319,7 +319,10 @@ def collect_reads(self, sample):
             # Clean up parent memory before spawning workers
             gc.collect()
             mp_context = multiprocessing.get_context('fork')
-            with ProcessPoolExecutor(max_workers=self.args.threads, mp_context=mp_context) as proc:
+            log_file, log_level = _get_log_params()
+            with ProcessPoolExecutor(max_workers=self.args.threads, mp_context=mp_context,
+                                     initializer=setup_worker_logging,
+                                     initargs=(log_file, log_level)) as proc:
                 results = proc.map(*read_gen, chunksize=1)
         else:
             results = map(*read_gen)
@@ -427,7 +430,10 @@ def process_assigned_reads(self, sample, dump_filename):
             # Clean up parent memory before spawning workers
             gc.collect()
             mp_context = multiprocessing.get_context('fork')
-            with ProcessPoolExecutor(max_workers=self.args.threads, mp_context=mp_context) as proc:
+            log_file, log_level = _get_log_params()
+            with ProcessPoolExecutor(max_workers=self.args.threads, mp_context=mp_context,
+                                     initializer=setup_worker_logging,
+                                     initargs=(log_file, log_level)) as proc:
                 results = proc.map(*model_gen, chunksize=1)
         else:
             results = map(*model_gen)
@@ -527,7 +533,10 @@ def filter_umis(self, sample):
                 # Clean up parent memory before spawning workers
                 gc.collect()
                 mp_context = multiprocessing.get_context('fork')
-                with ProcessPoolExecutor(max_workers=self.args.threads, mp_context=mp_context) as proc:
+                log_file, log_level = _get_log_params()
+                with ProcessPoolExecutor(max_workers=self.args.threads, mp_context=mp_context,
+                                         initializer=setup_worker_logging,
+                                         initargs=(log_file, log_level)) as proc:
                     results = proc.map(*umi_gen, chunksize=1)
             else:
                 results = map(*umi_gen)

src/table_splitter.py

@@ -15,6 +15,7 @@
 
 import pysam
 
+from .common import setup_worker_logging, _get_log_params
 
 logger = logging.getLogger('IsoQuant')
 
@@ -268,7 +269,10 @@ def split_read_table_parallel(sample, input_tsvs, split_reads_file_names, num_th
     # Clean up parent memory before spawning workers
     gc.collect()
     mp_context = multiprocessing.get_context('spawn')
-    with ProcessPoolExecutor(max_workers=num_workers, mp_context=mp_context) as executor:
+    log_file, log_level = _get_log_params()
+    with ProcessPoolExecutor(max_workers=num_workers, mp_context=mp_context,
+                             initializer=setup_worker_logging,
+                             initargs=(log_file, log_level)) as executor:
         futures = []
         for worker_id, my_chrs in enumerate(chr_assignments):
             future = executor.submit(