(#13) Threadsafe I/O support

Author: ray-chew

runs/icon_etopo_global.py

@@ -543,20 +543,23 @@ def parallel_wrapper(grid, params, reader, writer, chunk_output_dir, clat_rad, c
             'total_cores': 12,  # Conservative: use 12 of 16 threads
             'total_memory_gb': 12.0,
             'netcdf_chunk_size': 100,
+            'threads_per_worker': 1,  # Set to None for auto-compute
             'memory_per_cpu_mb': None,  # Will calculate dynamically
             'description': 'Generic laptop (16 threads, 16GB RAM)'
         },
         'dkrz_hpc': {
             'total_cores': 250,
             'total_memory_gb': 240.0,
             'netcdf_chunk_size': 100,
+            'threads_per_worker': None,  # Auto-compute based on worker memory
             'memory_per_cpu_mb': None,  # SLURM quota on interactive partition
             'description': 'DKRZ HPC interactive partition (standard memory node)'
         },
         'laptop_performance': {
             'total_cores': 20,  # Use 20 of 24 threads (leave 4 for background)
             'total_memory_gb': 80.0,
             'netcdf_chunk_size': 100,
+            'threads_per_worker': None,  # Auto-compute based on worker memory
             'memory_per_cpu_mb': None,  # Will calculate dynamically
             'description': 'AMD Ryzen AI 9 HX 370 (24 threads, 94GB RAM)'
         }
@@ -654,11 +657,15 @@ def parallel_wrapper(grid, params, reader, writer, chunk_output_dir, clat_rad, c
     logger.info(f"  Available cores: {total_cores}")
     logger.info(f"  Available memory: {total_memory_gb} GB")
     logger.info(f"  NetCDF chunk size: {netcdf_chunk_size} cells")
+
+    # Threading configuration display
+    if config['threads_per_worker'] is not None:
+        logger.info(f"  Threading mode: MANUAL (threads_per_worker = {config['threads_per_worker']})")
+    else:
+        logger.info(f"  Threading mode: AUTO (will compute based on worker count)")
+
     if config['memory_per_cpu_mb'] is not None:
         logger.info(f"  SLURM quota: {config['memory_per_cpu_mb']} MB per CPU")
-        logger.info(f"  Mode: HPC (threads scale with worker memory)")
-    else:
-        logger.info(f"  Mode: Laptop (threads distributed evenly)")
     logger.info("=" * 80)
 
     # Group cells by memory requirements for dynamic worker allocation
@@ -744,15 +751,29 @@ def parallel_wrapper(grid, params, reader, writer, chunk_output_dir, clat_rad, c
                 n_workers = batch_config['n_workers']
                 memory_per_worker = f"{int(batch_config['memory_per_worker_gb'])}GB"
 
-                # CRITICAL: threads_per_worker MUST be 1 because HDF5 is not thread-safe
-                # HDF5 was not compiled with --enable-threadsafe on this system.
-                # Even opening different NetCDF files from different threads causes crashes.
-                # Use more workers instead of threads for parallelism.
-                threads_per_worker = 1
+                # ============================================================
+                # THREADS PER WORKER CONFIGURATION
+                # ============================================================
+                # If threads_per_worker is explicitly set in config, use that value
+                # Otherwise, auto-compute based on available cores and workers
+                if config['threads_per_worker'] is not None:
+                    threads_per_worker = config['threads_per_worker']
+                    logger.info(f"\n  Using manual threads_per_worker: {threads_per_worker}")
+                else:
+                    # Auto-compute: distribute available cores among workers
+                    # Reserve at least 1 thread per worker, and cap at reasonable maximum
+                    threads_per_worker = max(1, min(4, total_cores // n_workers))
+                    logger.info(f"\n  Auto-computed threads_per_worker: {threads_per_worker}")
+                    logger.info(f"    (Based on {total_cores} cores / {n_workers} workers)")
+
+                # Note: Thread-safe HDF5 is required for threads_per_worker > 1
+                # Verify with: python3 -c "import netCDF4; print(netCDF4.__hdf5libversion__)"
+                # ============================================================
 
                 logger.info(f"\n  Starting Dask client for memory batch {mem_batch_idx}:")
                 logger.info(f"    Workers: {n_workers} × {memory_per_worker}")
                 logger.info(f"    Threads per worker: {threads_per_worker}")
+                logger.info(f"    Total parallel threads: {n_workers * threads_per_worker}")
                 logger.info(f"    Expected memory per cell: {batch_config['memory_per_cell_gb']:.1f} GB")
 
                 client = Client(