Wie kombiniere ich Pool.map mit Array (Shared Memory) in Python Multiprocessing?

Ich habe eine sehr große (nur Lesen) array mit Daten, die ich will verarbeitet werden, die mehrere Prozesse parallel.

Ich mag den Pool.map-Funktion auf und möchte es verwenden, um zu berechnen Funktionen auf, die die Daten parallel.

Sah ich, dass man verwenden kann, die den Wert oder die Array-Klasse verwenden shared memory, Daten zwischen Prozessen. Aber wenn ich versuche, mit dieser bekomme ich einen Absturz mit Laufzeitfehler: 'SynchronizedString Objekte sollten nur dann geteilt werden zwischen Prozessen, die durch Vererbung, wenn Sie den Pool.map-Funktion:

Hier ist ein Vereinfachtes Beispiel, was ich versuche zu tun:

from sys import stdin
from multiprocessing import Pool, Array

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"
  # want to share it using shared memory
  toShare = Array('c', testData)

  # this works
  print count_it( toShare, "a" )

  pool = Pool()

  # RuntimeError here
  print pool.map( count_it, [(toShare,key) for key in ["a", "b", "s", "d"]] )

Kann mir jemand sagen was mache ich hier falsch?

Also, was ich möchte zu tun ist, übergeben Sie Informationen über einen neu erstellten freigegebenen Speicher-array, um die Prozesse, nachdem Sie erstellt wurden in der Prozess-pool.

InformationsquelleAutor der Frage Jeroen Dirks | 2009-11-04

Versuchen wieder, wie ich gerade gesehen habe, dass die bounty 😉

Grundsätzlich finde ich die Fehlermeldung bedeutet, was es sagt - multiprocessing-shared-memory-Arrays können nicht übergeben werden als Argumente (durch Beizen). Es hat keinen Sinn, serialise die Daten - der Punkt ist, die Daten shared memory. So haben Sie, um das freigegebene array global. Ich denke, es ist übersichtlicher, es zu setzen als das Attribut eines Moduls, wie in meiner ersten Antwort, ist aber nur verlassen, es ist wie eine Globale variable in deinem Beispiel funktioniert auch gut. An Bord nehmen Sie Ihren Punkt nicht wollen, um die Daten vor der Gabel, hier ist ein modifiziertes Beispiel. Wenn Sie wollte mehr als einen möglichen gemeinsamen array (und das ist, warum Sie überholen wollte toShare als argument), könnten Sie ähnlich wie eine Globale Liste der freigegebenen arrays und übergeben Sie den index, um count_it (die würden sich 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"] )

[EDIT: Der obige funktioniert leider nicht unter windows, weil Sie nicht mit der Gabel. Doch die unter funktioniert auf Windows, noch mit Pool, so dass ich denke, das ist das nächste, was Sie wollen:

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"] )

Nicht sicher, warum die Karte nicht, Gurke die array-sondern Prozess-und Pool - ich denke, vielleicht hat es geht in dem Punkt der Teilprozess der Initialisierung unter windows. Beachten Sie, dass die Daten noch nach der Gabel obwohl.

InformationsquelleAutor der Antwort robince

4

Das problem, das ich sehe, ist, dass der Pool nicht unterstützt Beizen freigegebenen Daten durch seiner Liste von Argumenten. Das ist es, was die Fehlermeldung bedeutet, durch "Objekte sollte nur genutzt werden, Prozesse, die durch Vererbung". Die freigegebenen Daten werden vererbt, d.h., "global", wenn Sie wollen, es zu teilen mit dem Pool. Klasse.

Wenn Sie müssen Sie explizit übergeben, können Sie verwenden, multiprocessing.Prozess. Hier ist Ihre überarbeitete Beispiel:
```
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(),
```
Ausgabe: ('s', 9) ('a', 2) ('b', 3)
('d', 12)

Die Reihenfolge der Elemente der queue variieren.

Machen dies mehr generische und ähnlich wie Pool, können Sie eine Feste Zahl N der Prozesse, teilen Sie die Liste der Schlüssel in N Teile, und verwenden Sie dann eine wrapper Funktion als Prozess Ziel, die call count_it für jede Taste in der Liste weitergegeben werden, wie:
```
def wrapper( q, arr, keys ):
  for k in keys:
    count_it(q, arr, k)
```
InformationsquelleAutor der Antwort jwilson
2

Wenn die Daten nur gelesen, dass es einfach eine variable in einem Modul vor die Gabel aus dem Pool. Dann werden alle Kind-Prozesse sollten in der Lage sein, darauf zuzugreifen und es kann nicht kopiert werden, vorausgesetzt, dass Sie nicht schreiben.
```
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"] )
```
Wenn Sie versuchen möchten, verwenden Sie Array-obwohl Sie könnten versuchen, mit der lock=False keyword argument (Standardwert ist true).

InformationsquelleAutor der Antwort robince
-1

Die multiprocessing.sharedctypes - Modul bietet Funktionen für die Zuweisung von ctypes Objekte aus dem gemeinsamen Speicher, die vererbt werden können, indem Sie die untergeordneten Prozesse.

So Ihre Nutzung von sharedctypes ist falsch. Möchten Sie Erben dieses array aus übergeordneten Prozess oder Sie es vorziehen zu übergeben, die es explizit? Im ersten Fall erstellen Sie eine Globale variable als andere Antworten vorschlagen. Aber Sie brauchen nicht zu verwenden sharedctypes zu pass es sich ausdrücklich vor, geben Sie nur original -testData.

BTW, die Nutzung Pool.map() ist falsch. Es hat die gleiche Schnittstelle wie builtin map() Funktion (hast du versaut es mit starmap()?). Unten ist Beispiel mit, vorbei array explizit:
```
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"]])
```
InformationsquelleAutor der Antwort Denis Otkidach

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.