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

Пытаюсь снова, потому что я только что увидел награду;)

В основном я думаю, что сообщение об ошибке означает то, о чем говорилось - многопроцессорные массивы разделяемой памяти не могут быть переданы в качестве аргументов (путем травления). Нет смысла сериализовать данные - дело в том, что данные являются разделяемой памятью. Итак, вам нужно сделать общий массив глобальным. Я думаю, что лучше использовать его как атрибут модуля, как в моем первом ответе, но просто оставить его как глобальную переменную в вашем примере также хорошо. Принимая во внимание вашу точку зрения о нежелании устанавливать данные перед форком, вот модифицированный пример. Если вы хотите иметь более одного возможного общего массива (и именно поэтому вы хотели передать toShare в качестве аргумента), вы могли бы аналогичным образом создать глобальный список общих массивов и просто передать индекс в count_it (который стал бы for c in toShare[i]:).

from sys import stdin
from multiprocessing import Pool, Array, Process

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

if __name__ == '__main__':
  # allocate shared array - want lock=False in this case since we 
  # aren't writing to it and want to allow multiple processes to access
  # at the same time - I think with lock=True there would be little or 
  # no speedup
  maxLength = 50
  toShare = Array('c', maxLength, lock=False)

  # fork
  pool = Pool()

  # can set data after fork
  testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf"
  if len(testData) > maxLength:
      raise ValueError, "Shared array too small to hold data"
  toShare[:len(testData)] = testData

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

[РЕДАКТИРОВАТЬ: Вышеупомянутое не работает в Windows из-за того, что не используется fork. Однако приведенное ниже работает в Windows, все еще используя Pool, поэтому я думаю, что это наиболее близко к тому, что вы хотите:

from sys import stdin
from multiprocessing import Pool, Array, Process
import mymodule

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

def initProcess(share):
  mymodule.toShare = share

if __name__ == '__main__':
  # allocate shared array - want lock=False in this case since we 
  # aren't writing to it and want to allow multiple processes to access
  # at the same time - I think with lock=True there would be little or 
  # no speedup
  maxLength = 50
  toShare = Array('c', maxLength, lock=False)

  # fork
  pool = Pool(initializer=initProcess,initargs=(toShare,))

  # can set data after fork
  testData = "abcabcs bsdfsdf gdfg dffdgdfg sdfsdfsd sdfdsfsdf"
  if len(testData) > maxLength:
      raise ValueError, "Shared array too small to hold data"
  toShare[:len(testData)] = testData

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

Не уверен, почему карта не обрабатывает массив, а процесс и пул - я думаю, возможно, он был передан в точке инициализации подпроцесса в Windows. Обратите внимание, что данные все еще устанавливаются после вилки.

Если данные доступны только для чтения, просто сделайте их переменной в модуле перед вилкой из 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, вы можете попробовать с аргументом ключевого слова lock=False (по умолчанию это правда).

Проблема, которую я вижу, заключается в том, что Pool не поддерживает сбор общих данных через свой список аргументов. Это то, что сообщение об ошибке означает, что «объекты должны совместно использоваться между процессами только посредством наследования». Общие данные должны быть унаследованными, то есть глобальными, если вы хотите поделиться ими с помощью класса 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)

Если вы видите:

RuntimeError: синхронизированные объекты должны совместно использоваться между процессами только через наследование.

Рассмотрите возможность использования multiprocessing.Manager, поскольку это не так. есть это ограничение. Менеджер работает, считая, что он предположительно работает в отдельном процессе.

import ctypes
import multiprocessing

# Put this in a method or function, otherwise it will run on import from each module:
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