fix: 移除 _local_kriging 内部 multiprocessing.Pool 避免 Windows 嵌套 spawn 死锁
根因: step11 用 ProcessPoolExecutor 派发 CSV 到子进程, 子进程内 ContentMapper.process_data → _local_kriging 又创建 multiprocessing.Pool。Windows spawn 模式下嵌套 Pool 死锁。 修复: _local_kriging 内部改为顺序执行 16 个块。 每块 ~12s (650K 网格点 × 50 近邻), 总计 ~3 分钟, 完全可接受。 step11 层的 ProcessPoolExecutor 仍提供 CSV 级并行。
This commit is contained in:
@ -597,14 +597,11 @@ class ContentMapper:
|
||||
grid_y = grid_yy[:, 0]
|
||||
total_cells = len(grid_x) * len(grid_y)
|
||||
|
||||
_n_workers = min(os.cpu_count() or 4, 8)
|
||||
print(f"正在使用 局部克里金 (自适应分块 + 40% 重叠缓冲):"
|
||||
f"网格={total_cells:,} 点, "
|
||||
f"{_n_workers} workers")
|
||||
f"网格={total_cells:,} 点")
|
||||
|
||||
grid_content = self._local_kriging(
|
||||
points, values, grid_x, grid_y,
|
||||
n_workers=_n_workers,
|
||||
n_closest_points=50,
|
||||
)
|
||||
|
||||
@ -692,7 +689,7 @@ class ContentMapper:
|
||||
return grid_content
|
||||
|
||||
def _local_kriging(self, points, values, grid_x, grid_y,
|
||||
n_workers=4, n_closest_points=50):
|
||||
n_closest_points=50):
|
||||
"""局部克里金:自适应分块 + 重叠缓冲区 + 保护性近邻限制
|
||||
|
||||
1. 自适应块大小: 根据 extent 自动切分为 ~4×4 块 (16~25块)
|
||||
@ -700,7 +697,6 @@ class ContentMapper:
|
||||
3. 保护性近邻: n_closest_points=50,稀释极端异常值
|
||||
4. 网格点仅使用严格不重叠的块范围(Buffer 仅用于筛选采样点)
|
||||
"""
|
||||
import multiprocessing
|
||||
|
||||
x_min, x_max = float(grid_x[0]), float(grid_x[-1])
|
||||
y_min, y_max = float(grid_y[0]), float(grid_y[-1])
|
||||
@ -775,9 +771,15 @@ class ContentMapper:
|
||||
if len(tasks) <= 1:
|
||||
return self._local_krige_block(*tasks[0])
|
||||
|
||||
print(f" 启动 {min(n_workers, len(tasks))} 个 worker 进程...")
|
||||
with multiprocessing.Pool(processes=min(n_workers, len(tasks))) as pool:
|
||||
results = pool.map(_local_krige_block_worker, tasks)
|
||||
# 在 ProcessPoolExecutor 的子进程内(step11 批量模式)顺序执行,
|
||||
# 避免 Windows spawn 模式下嵌套 multiprocessing 死锁。
|
||||
# 16 个块顺序跑 ~1-2 分钟,完全可接受。
|
||||
print(f" 顺序执行 {len(tasks)} 个局部 Kriging 块...")
|
||||
results = []
|
||||
for i, task in enumerate(tasks):
|
||||
if i % max(1, len(tasks) // 4) == 0 or i == len(tasks) - 1:
|
||||
print(f" [LocalKrige] {i+1}/{len(tasks)} ...")
|
||||
results.append(_local_krige_block_worker(task))
|
||||
|
||||
# 拼接:全 NaN 数组,逐块填回
|
||||
grid_full = np.full((len(grid_y), len(grid_x)), np.nan, dtype=np.float64)
|
||||
|
||||
Reference in New Issue
Block a user