Объедините Pool.map с Массивом общей памяти в многопроцессорной обработке Python

Если Вы видите RuntimeError: Synchronized objects should only be shared between processes through inheritance ошибка, рассматриваете использование multiprocessing.Manager , поскольку она не имеет этого ограничения. Менеджер работает, полагая, что это, по-видимому, работает в отдельном процессе в целом.

import ctypes
import multiprocessing

manager = multiprocessing.Manager()
counter = manager.Value(ctypes.c_ulonglong, 0)
counter_lock = manager.Lock()  # pylint: disable=no-member

with counter_lock:
    counter.value = count = counter.value + 1

Если данные читаются только для чтения, просто сделайте их переменной в модуле перед вилкой из Pool. Тогда все дочерние процессы должны иметь к нему доступ, и он не будет скопирован, если вы не напишете в него.

import myglobals # anything (empty .py file)
myglobals.data = []

def count_it( key ):
    count = 0
    for c in myglobals.data:
        if c == key:
            count += 1
    return count

if __name__ == '__main__':
myglobals.data = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf"

pool = Pool()
print pool.map( count_it, ["a", "b", "s", "d"] )

Если вы действительно хотите попробовать использовать Array, вы можете попробовать с блокировкой = False аргумент ключевого слова (по умолчанию истинно).

Проблема, которую я вижу, заключается в том, что пул не поддерживает сбор общих данных через свой список аргументов. Это то, что сообщение об ошибке означает, что «объекты должны совместно использоваться между процессами только посредством наследования». Общие данные должны быть унаследованы, т. Е. Глобальными, если вы хотите поделиться ими с помощью класса Pool.

Если вам нужно передать их явно, вам, возможно, придется использовать multiprocessing.Process. Вот ваш переработанный пример:

from multiprocessing import Process, Array, Queue

def count_it( q, arr, key ):
  count = 0
  for c in arr:
    if c == key:
      count += 1
  q.put((key, count))

if __name__ == '__main__':
  testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf"
  # want to share it using shared memory
  toShare = Array('c', testData)

  q = Queue()
  keys = ['a', 'b', 's', 'd']
  workers = [Process(target=count_it, args = (q, toShare, key))
    for key in keys]

  for p in workers:
    p.start()
  for p in workers:
    p.join()
  while not q.empty():
    print q.get(),

Вывод: ('s', 9) ('a', 2) ('b', 3) ('d', 12)

Порядок элементов очереди может варьироваться.

Чтобы сделать это более общим и похожим на Pool, вы можете создать фиксированное количество процессов N, разделив список ключей на N частей, а затем использовать функцию-оболочку в качестве цели процесса, которая будет вызывать count_it для каждого ключа в переданном списке, например:

def wrapper( q, arr, keys ):
  for k in keys:
    count_it(q, arr, k)

The multiprocessing.sharedctypes module provides functions for allocating ctypes objects from shared memory which can be inherited by child processes.

So your usage of sharedctypes is wrong. Do you wish to inherit this array from parent process or you prefer to pass it explicitly? In the former case you have to create a global variable as other answers suggest. But you don't need to use sharedctypes to pass it explicitly, just pass original testData.

BTW, your usage of Pool.map() is wrong. It has the same interface as builtin map() function (did you messed it with starmap()?). Below is working example with, passing array explicitly:

from multiprocessing import Pool

def count_it( (arr, key) ):
    count = 0
    for c in arr:
        if c == key:
            count += 1
    return count

if __name__ == '__main__':
    testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf"
    pool = Pool()
    print pool.map(count_it, [(testData, key) for key in ["a", "b", "s", "d"]])