Python多进程大数据集评估问题：共享变量未更新

处理大量数据时改用multiprocessing实现，但共享变量max_value与best_combination始终无法更新，锁机制似乎未生效。脚本逻辑为：生成并筛选组合、按1000条分块、多进程评估，当组合value总和大于当前最大值时更新共享变量。

import itertools
from timeit import default_timer as timer
from math import factorial
from multiprocessing import Process, Value, Pool, Manager, Lock

def total_combo(n, r):
    return factorial(n) // factorial(r) // factorial(n-r)

def iterator_slice(iterator, length):
    iterator = iter(iterator)
    while True:
        res = tuple(itertools.islice(iterator, length))
        if not res:
            break
        yield res

def evaluate_combination(combinations, best_combination, max_value, lock):
           
    for combination in combinations:
        #total_index = sum(people['index'] for people in combination)
        total_value = sum(people['value'] for people in combination)
        
        # Check if the combination meets the constraints
        if total_value > max_value.value:
            with lock:
                best_combination = list(combination)
                max_value.value = total_value

def generate_filtered_combinations(players,r):
    return itertools.filterfalse(lambda y: sum(x['index'] for x in y) > 30, itertools.combinations(players, r))

if __name__ == "__main__":
    
    start = timer()
    
    # Example usage
    players = [
        {"people":"John Doe1","index":8,"value":15},
        {"people":"John Doe2","index":7,"value":28},
        {"people":"John Doe3","index":6,"value":13},
        {"people":"John Doe4","index":7,"value":11},
        {"people":"John Smith1","index":7,"value":11},
        {"people":"John Smith2","index":6,"value":9},
        {"people":"John Smith3","index":6,"value":10},
        # .... just some data for example
    ];
    
    manager = Manager()
        
    max_value = manager.Value('i', 0)
    best_combination = manager.list()
    lock = Lock()
       
    for r in range(6,8):
        print(r, total_combo(len(players),r)) # max possible items in the combination for r with players
        combos = generate_filtered_combinations(players,r) # generate all combinations filtered
              
        pool = Pool(4)  # lets use 4 workers
        cursor_iterator = iterator_slice(combos, 1000)  # slice it
        queue = []  # a queue for our current worker async results, a deque would be faster
        
        while cursor_iterator or queue:  # while we have anything to do...
            try:
                # add our next slice to the pool:
                queue.append(pool.apply_async(evaluate_combination, [next(cursor_iterator), best_combination, max_value, lock]))
            except (StopIteration, TypeError):  # no more data, clear out the slice iterator
                cursor_iterator = None
            # wait for a free worker or until all remaining finish
            while queue and (len(queue) >= pool._processes or not cursor_iterator):
                process = queue.pop(0)  # grab a process response from the top
                if not process.ready():  # a sub-process has not finished execution
                    queue.append(process)  # add it back to the queue
                else:
                    # you can use process.get() to get the result if needed
                    pass
        pool.close()
        
    # Print the best combination
    for people in best_combination:
        print(people['people'],people['index'])
    
    print(max_value.value)

    end = timer()
    print(end - start)

1. best_combination赋值错误 #

原代码中best_combination = list(combination)是给函数内的本地变量重新赋值，并未修改Manager创建的共享列表。正确做法是清空共享列表后替换为新组合：

with lock:
    best_combination[:] = list(combination)
    max_value.value = total_value

2. 锁未跨进程生效 #

普通Lock()创建的锁仅在当前进程内有效，Pool的子进程无法共享该锁。需用Manager.Lock()创建跨进程锁：

lock = manager.Lock()  # 替换原有的lock = Lock()

3. 存在竞态条件 #

原代码先判断total_value > max_value.value再加锁，可能在判断与加锁之间，其他进程已更新max_value，导致错误覆盖。应在加锁后重新检查：

for combination in combinations:
    total_value = sum(people['value'] for people in combination)
    with lock:
        if total_value > max_value.value:
            best_combination[:] = list(combination)
            max_value.value = total_value

import itertools
from timeit import default_timer as timer
from math import factorial
from multiprocessing import Pool, Manager

def total_combo(n, r):
    return factorial(n) // factorial(r) // factorial(n-r)

def iterator_slice(iterator, length):
    iterator = iter(iterator)
    while True:
        res = tuple(itertools.islice(iterator, length))
        if not res:
            break
        yield res

def evaluate_combination(combinations, best_combination, max_value, lock):
    for combination in combinations:
        total_value = sum(people['value'] for people in combination)
        # 加锁后检查并更新，避免竞态条件
        with lock:
            if total_value > max_value.value:
                best_combination[:] = list(combination)
                max_value.value = total_value

def generate_filtered_combinations(players,r):
    return itertools.filterfalse(lambda y: sum(x['index'] for x in y) > 30, itertools.combinations(players, r))

if __name__ == "__main__":
    start = timer()
    
    players = [
        {"people":"John Doe1","index":8,"value":15},
        {"people":"John Doe2","index":7,"value":28},
        {"people":"John Doe3","index":6,"value":13},
        {"people":"John Doe4","index":7,"value":11},
        {"people":"John Smith1","index":7,"value":11},
        {"people":"John Smith2","index":6,"value":9},
        {"people":"John Smith3","index":6,"value":10},
        # .... just some data for example
    ]
    
    manager = Manager()
    max_value = manager.Value('i', 0)
    best_combination = manager.list()
    # 使用Manager创建跨进程锁
    lock = manager.Lock()
       
    for r in range(6,8):
        print(r, total_combo(len(players),r))
        combos = generate_filtered_combinations(players,r)
              
        pool = Pool(4)
        cursor_iterator = iterator_slice(combos, 1000)
        queue = []
        
        while cursor_iterator or queue:
            try:
                queue.append(pool.apply_async(evaluate_combination, [next(cursor_iterator), best_combination, max_value, lock]))
            except (StopIteration, TypeError):
                cursor_iterator = None
            while queue and (len(queue) >= pool._processes or not cursor_iterator):
                process = queue.pop(0)
                if not process.ready():
                    queue.append(process)
                else:
                    pass
        pool.close()
        pool.join()  # 等待进程池所有任务完成
        
    # 输出结果
    print("最佳组合：")
    for people in best_combination:
        print(people['people'], people['index'])
    print(f"最大value总和：{max_value.value}")
    
    end = timer()
    print(f"耗时：{end - start:.2f}秒")

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