Numba List并行创建填充致解释器崩溃，求可行解决方案

问题原因 #

你遇到的double free or corruption (!prev)崩溃确实是因为Numba typed.List的append方法并非线程安全，在prange并行循环中，多个线程同时调用append会触发内存竞争，导致解释器崩溃。

而预填充列表时的报错，是因为[None] * len(input)生成的是元素类型为none的列表，Numba的强类型系统不允许将(Dict, int)元组赋值给none类型的元素，因此找不到对应的setitem实现。

解决办法 #

方案1：预填充类型匹配的列表（推荐） #

创建类型明确的typed.List，预填充对应类型的占位元素后，再通过并行循环赋值，避免线程竞争：

import numba
from numba import prange
from numba.typed import Dict, List
from numba.types import Tuple, DictType, int64

@numba.njit(parallel=True)
def create_list_of_dicts(input: List[int]):
    # 定义列表元素的类型：元组（键值为int64的字典，int64整数）
    element_type = Tuple((DictType(int64, int64), int64))
    # 创建指定类型的空列表
    output = List.empty_list(element_type)
    
    # 预填充列表到目标长度（必须用对应类型的占位元素，不能用None）
    temp_dict = Dict()
    temp_dict[1] = 2
    for _ in range(len(input)):
        output.append((temp_dict, 0))
    
    # 并行循环赋值，每个线程创建独立的字典
    for i in prange(len(input)):
        new_dict = Dict()
        new_dict[1] = 2
        output[i] = (new_dict, input[i])
    
    return output

# 测试代码
input_list = List([i for i in range(1000)])
result = create_list_of_dicts(input_list)
print(len(result))  # 输出1000

关键点说明 #

1. 明确类型：用List.empty_list指定列表元素的具体类型，让Numba能正确推断类型并生成有效代码。
2. 预填充正确类型的元素：不能用None占位，必须使用与最终元素类型一致的对象（哪怕是临时对象），避免类型不兼容的报错。
3. 线程独立创建字典：每个线程生成自己的Dict实例，避免共享对象带来的潜在线程安全问题。

替代方案：线程局部结果合并 #

如果预填充不方便，也可以让每个线程生成局部列表，最后合并所有结果（适合结果顺序不敏感的场景）：

import numba
from numba import prange
from numba.typed import Dict, List

@numba.njit(parallel=True)
def create_list_of_dicts(input: List[int]):
    # 获取并行线程数
    num_threads = numba.get_num_threads()
    # 为每个线程创建局部列表
    local_lists = [List() for _ in range(num_threads)]
    
    for i in prange(len(input)):
        thread_id = numba.get_thread_id()
        new_dict = Dict()
        new_dict[1] = 2
        local_lists[thread_id].append((new_dict, input[i]))
    
    # 合并所有局部列表
    output = List()
    for lst in local_lists:
        output.extend(lst)
    
    return output

# 测试代码
input_list = List([i for i in range(1000)])
result = create_list_of_dicts(input_list)
print(len(result))  # 输出1000

这个方案通过线程局部存储避免了直接共享列表的竞争，最后合并结果，但注意结果的顺序会被打乱（因为线程执行顺序不确定）。

内容的提问来源于stack exchange，提问作者Bananach