为何共享内存set()被多次更新，导致大小异常暴涨？

使用BaseManager创建共享内存set()，供多进程填充后合并。但单进程运行时，本地kmer_set长度为419056，共享set转字符串后的长度却达到14666960，暴涨350倍，导致后续查询函数运行极慢。

代码实现 #

from multiprocessing import Process, BaseManager

class CustomManager(BaseManager):
    pass
CustomManager.register('shared_set', set)
manager = CustomManager()
manager.start()
shared_kmer_set = manager.shared_set()

def classify_reads(fastq_file, bloom_filter, kmer_size, n_consecutive_matches, output, shared_anchor_kmer_set):
    kmer_set = set()
    # populates the kmer_set
    print(len(kmer_set))
    shared_anchor_kmer_set.update(anchor_kmer_set)


processes = [Process(target=classify_reads, args=(i, present_kmers_bf, shared_k_size.value, shared_n_consec_matches.value, shared_output.value, shared_kmer_set)) for i in args["input"]]
for i in range(0, len(processes), args["threads"]):
    for process in processes[i:i + args["threads"]]:
        process.start()
for process in processes:
     process.join()
for process in processes:
    process.terminate()

print(len(str(shared_kmer_set)))

def classify_reads_using_anchor_kmers(shared_kmer_set, file):
    copy_set = str(shared_kmer_set)
    for line in file:
       if line in copy_set:
            print(line)

processes = [Process(target=classify_reads_using_anchor_kmers, args=(i, shared_kmer_set, shared_k_size.value, shared_anchor_proportion_cutoff.value, shared_output.value)) for i in args["input"]]
for i in range(0, len(processes), args["threads"]):
    for process in processes[i:i + args["threads"]]:
        process.start()
for process in processes:
    process.join()
for process in processes:
    process.terminate()

运行输出 #

419056
14666960

核心原因分析 #

1. 共享set是远程代理对象
   通过BaseManager注册的shared_set不是本地set，而是一个跨进程通信的代理对象。调用str(shared_kmer_set)时，得到的不是集合元素的字符串，而是代理对象的内部调试信息，包含大量进程通信的元数据，这才是长度暴涨的根本原因。

2. 错误的元素判断逻辑
   classify_reads_using_anchor_kmers中用str(shared_kmer_set)转成字符串后做in判断，本质是子串匹配而非集合成员判断，既低效又完全不可靠——代理对象的字符串根本不包含真实的k-mer元素。

3. 代码笔误导致集合未正确填充
   classify_reads函数里填充的是kmer_set，但调用update时用了未定义的anchor_kmer_set，应该是shared_anchor_kmer_set.update(kmer_set)，这直接导致共享集合没有被正确填充，后续操作的都是空代理对象的冗余信息。

修复方案 #

1. 修正集合填充逻辑 #

def classify_reads(fastq_file, bloom_filter, kmer_size, n_consecutive_matches, output, shared_anchor_kmer_set):
    kmer_set = set()
    # 填充kmer_set的逻辑
    print(len(kmer_set))
    # 用本地kmer_set更新共享集合
    shared_anchor_kmer_set.update(kmer_set)

2. 正确实现元素查询 #

不要将代理对象转成字符串，直接将共享集合的元素复制到本地集合后操作：

def classify_reads_using_anchor_kmers(file, shared_kmer_set):
    # 复制共享集合到本地，避免跨进程调用开销
    local_set = set(shared_kmer_set)
    for line in file:
        # 去除换行符匹配集合中的k-mer
        kmer = line.strip()
        if kmer in local_set:
            print(line)

3. 优化多进程管理 #

无需额外调用terminate()，join()已等待进程结束，简化启动逻辑：

processes = []
for input_file in args["input"]:
    p = Process(target=classify_reads, args=(input_file, present_kmers_bf, shared_k_size.value, shared_n_consec_matches.value, shared_output.value, shared_kmer_set))
    processes.append(p)
    p.start()

for p in processes:
    p.join()

4. 替代方案：用更高效的共享结构 #

如果k-mer数据量极大，可以直接用multiprocessing.Manager().set()替代自定义BaseManager；或使用生物信息学专用库（如kmerkit）优化性能。

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