如何在多处理包中锁定特定的数组索引？答案

【问题标题】：How does one lock specific array indices in multiprocessing package?如何在多处理包中锁定特定的数组索引？
【发布时间】：2017-04-12 15:56:29
【问题描述】：

我有一个multiprocessing.manager.Array 对象，该对象将由多个工作人员共享以统计观察到的事件：数组中的每个元素都包含不同事件类型的统计。增加一个计数需要读取和写入操作，所以我认为为了避免竞争条件，每个工作人员都需要请求一个涵盖两个阶段的锁，例如

with lock:
    my_array[event_type_index] += 1

我的直觉是应该可以在特定的数组元素上放置一个锁。使用这种类型的锁，worker #1 可以在 worker #2 递增元素 2 的同时递增元素 1。这对我的应用程序（n-gram 计数）特别有用，其中数组长度很大，很少发生冲突。

但是，我不知道如何为数组请求元素锁定。 multiprocessing中是否存在这种情况，或者有解决方法吗？

关于更多上下文，我在下面包含了我当前的实现：

import multiprocessing as mp
from queue import Empty

def count_ngrams_in_sentence(n, ngram_counts, char_to_idx_dict, sentence_queue, lock):
    while True:
        try:
            my_sentence_str = sentence_queue.get_nowait()
            my_sentence_indices = [char_to_idx_dict[i] for i in my_sentence_str]
            my_n = n.value
            for i in range(len(my_sentence_indices) - my_n + 1):
                my_index = int(sum([my_sentence_indices[i+j]*(27**(my_n - j - 1)) \
                                    for j in range(my_n)]))
                with lock:  # lock the whole array?
                    ngram_counts[my_index] += 1
            sentence_queue.task_done()
        except Empty:
            break
    return

if __name__ == '__main__':
    n = 4
    num_ngrams = 27**n
    num_workers = 2
    sentences = [ ... list of sentences in lowercase ASCII + spaces ... ]

    manager = mp.Manager()
    sentence_queue = manager.JoinableQueue()
    for sentence in sentences:
        sentence_queue.put(sentence)
    n = manager.Value('i', value=n, lock=False)
    char_to_idx_dict = manager.dict([(i,ord(i)-97) for i in string.ascii_lowercase] + [(' ', 26)],
                                    lock=False)
    lock = manager.Lock()
    ngram_counts = manager.Array('l', [0]*num_ngrams, lock=lock)
''
    workers = [mp.Process(target=count_ngrams_in_sentence,
                          args=[n,
                                ngram_counts,
                                char_to_idx_dict,
                                sentence_queue,
                                lock]) for i in range(num_workers)]
    for worker in workers:
        worker.start()
    sentence_queue.join()

【问题讨论】：

天啊，我也有类似的问题。 @mewahl 你有什么进展吗？
很抱歉，我从来没有解决过这个问题，@mozboz -- 祝你好运！
感谢您的回复。我没有大量的数组元素，所以只使用与我的数据数组具有相同 ID 的单独的锁数组。不知道这是否有效，但还没有到那个阶段。

标签： python-3.x multiprocessing python-multiprocessing

【解决方案1】：

Multiprocessing.manager.Array 带有内置锁。必须切换到 RawArray。

有一个锁列表。在修改索引之前，获取数组的锁。然后释放。

locks[i].acquire()
array[i,:]=0
locks[i].release()

正如我所说，如果数组是 MultiProcessing.RawArray 或类似数组，则多个进程可以同时读取或写入。对于一些类型的数组，对数组的读/写本质上是原子的——锁本质上是内置的。在继续之前仔细研究这一点。

在性能方面，在 Python 中，对列表的索引是纳秒级的，而获取和释放锁则是微秒级的。这不是一个大问题。

【讨论】：