multitraitement: partager un gros objet en lecture seule entre les processus?

Les processus enfants générés via le multitraitement partagent-ils des objets créés précédemment dans le programme?

J'ai la configuration suivante:

do_some_processing(filename):
    for line in file(filename):
        if line.split(',')[0] in big_lookup_object:
            # something here

if __name__ == '__main__':
    big_lookup_object = marshal.load('file.bin')
    pool = Pool(processes=4)
    print pool.map(do_some_processing, glob.glob('*.data'))

Je charge un gros objet en mémoire, puis je crée un pool de travailleurs qui doivent utiliser ce gros objet. Le gros objet est accessible en lecture seule, je n'ai pas besoin de passer des modifications entre les processus.

Ma question est la suivante: le gros objet est-il chargé dans la mémoire partagée, comme ce serait le cas si je créais un processus sous unix / c, ou chaque processus charge-t-il sa propre copie du gros objet?

Mise à jour: pour clarifier davantage - big_lookup_object est un objet de recherche partagé. Je n'ai pas besoin de diviser cela et de le traiter séparément. Je dois en garder un seul exemplaire. Le travail dont j'ai besoin pour le fractionner consiste à lire de nombreux autres fichiers volumineux et à rechercher les éléments de ces fichiers volumineux par rapport à l'objet de recherche.

Mise à jour supplémentaire: la base de données est une bonne solution, memcached pourrait être une meilleure solution et le fichier sur disque (shelve ou dbm) pourrait être encore meilleur. Dans cette question, j'étais particulièrement intéressé par une solution en mémoire. Pour la solution finale, j'utiliserai hadoop, mais je voulais voir si je pouvais également avoir une version locale en mémoire.

"Les processus enfants générés via le multi-traitement partagent-ils des objets créés plus tôt dans le programme?"

Non (python avant la version 3.8), et Oui dans la version 3.8 ( https://docs.python.org/3/library/multiprocessing.shared_memory.html#module-multiprocessing.shared_memory )

Les processus ont un espace mémoire indépendant.

Solution 1

Pour tirer le meilleur parti d'une grande structure avec beaucoup de travailleurs, procédez comme suit.

1. Ecrire chaque travailleur comme un "filtre" - lit les résultats intermédiaires de stdin, fonctionne, écrit les résultats intermédiaires sur stdout.

2. Connectez tous les travailleurs en tant que pipeline:

   process1 <source | process2 | process3 | ... | processn >result

Chaque processus lit, travaille et écrit.

Ceci est remarquablement efficace car tous les processus sont exécutés simultanément. Les écritures et les lectures passent directement par des tampons partagés entre les processus.

Solution 2

Dans certains cas, vous avez une structure plus complexe - souvent une structure «en éventail». Dans ce cas, vous avez un parent avec plusieurs enfants.

1. Parent ouvre les données source. Le parent fourche un certain nombre d'enfants.

2. Parent lit la source, ferme des parties de la source sur chaque enfant exécuté simultanément.

3. Lorsque le parent atteint la fin, fermez le tuyau. L'enfant obtient la fin du fichier et se termine normalement.

Les parties enfants sont agréables à écrire car chaque enfant lit simplement sys.stdin.

Le parent a un peu de fantaisie pour engendrer tous les enfants et retenir correctement les tuyaux, mais ce n'est pas trop mal.

Fan-in est la structure opposée. Un certain nombre de processus exécutés indépendamment doivent entrelacer leurs entrées dans un processus commun. Le collecteur n'est pas aussi facile à écrire, car il doit lire à partir de nombreuses sources.

La lecture de nombreux tubes nommés est souvent effectuée à l'aide du selectmodule pour voir quels tubes ont une entrée en attente.

Solution 3

La recherche partagée est la définition d'une base de données.

Solution 3A - charger une base de données. Laissez les travailleurs traiter les données de la base de données.

Solution 3B - créez un serveur très simple en utilisant werkzeug (ou similaire) pour fournir des applications WSGI qui répondent à HTTP GET afin que les travailleurs puissent interroger le serveur.

Solution 4

Objet du système de fichiers partagé. Unix OS propose des objets de mémoire partagée. Ce ne sont que des fichiers qui sont mappés à la mémoire afin que l'échange d'E / S soit effectué au lieu de lectures tamponnées plus conventionnelles.

Vous pouvez le faire à partir d'un contexte Python de plusieurs manières

1. Écrivez un programme de démarrage qui (1) divise votre objet gigantesque original en objets plus petits, et (2) démarre les travailleurs, chacun avec un objet plus petit. Les objets plus petits pourraient être des objets Python marinés pour économiser un tout petit peu de temps de lecture de fichier.

2. Écrivez un programme de démarrage qui (1) lit votre objet gigantesque original et écrit un fichier structuré en page et codé en octets en utilisant des seekopérations pour vous assurer que les sections individuelles sont faciles à trouver avec de simples recherches. C'est ce que fait un moteur de base de données - diviser les données en pages, rendre chaque page facile à localiser via un fichier seek.

   Créez des ouvriers ayant accès à ce grand fichier structuré en page. Chaque travailleur peut rechercher les pièces pertinentes et y effectuer son travail.

Les processus enfants générés via le multitraitement partagent-ils des objets créés précédemment dans le programme?

Ça dépend. Pour les variables globales en lecture seule, cela peut souvent être considéré comme tel (à part la mémoire consommée), sinon il ne devrait pas.

La documentation du multitraitement dit:

Better to inherit than pickle/unpickle

Sous Windows, de nombreux types issus du multitraitement doivent être sélectionnables pour que les processus enfants puissent les utiliser. Cependant, il faut généralement éviter d'envoyer des objets partagés à d'autres processus en utilisant des canaux ou des files d'attente. Au lieu de cela, vous devez organiser le programme de sorte qu'un processus qui a besoin d'accéder à une ressource partagée créée ailleurs puisse l'hériter d'un processus ancêtre.

Explicitly pass resources to child processes

Sous Unix, un processus enfant peut utiliser une ressource partagée créée dans un processus parent à l'aide d'une ressource globale. Cependant, il est préférable de transmettre l'objet en tant qu'argument au constructeur du processus enfant.

En plus de rendre le code (potentiellement) compatible avec Windows, cela garantit également que tant que le processus enfant est toujours actif, l'objet ne sera pas récupéré dans le processus parent. Cela peut être important si une ressource est libérée lorsque l'objet est récupéré dans le processus parent.

Global variables

Gardez à l'esprit que si le code exécuté dans un processus enfant tente d'accéder à une variable globale, alors la valeur qu'il voit (le cas échéant) peut ne pas être la même que la valeur du processus parent au moment où Process.start () a été appelé .

Exemple

Sous Windows (processeur unique):

#!/usr/bin/env python
import os, sys, time
from multiprocessing import Pool

x = 23000 # replace `23` due to small integers share representation
z = []    # integers are immutable, let's try mutable object

def printx(y):
    global x
    if y == 3:
       x = -x
    z.append(y)
    print os.getpid(), x, id(x), z, id(z) 
    print y
    if len(sys.argv) == 2 and sys.argv[1] == "sleep":
       time.sleep(.1) # should make more apparant the effect

if __name__ == '__main__':
    pool = Pool(processes=4)
    pool.map(printx, (1,2,3,4))

Avec sleep:

$ python26 test_share.py sleep
2504 23000 11639492 [1] 10774408
1
2564 23000 11639492 [2] 10774408
2
2504 -23000 11639384 [1, 3] 10774408
3
4084 23000 11639492 [4] 10774408
4

Sans sleep:

$ python26 test_share.py
1148 23000 11639492 [1] 10774408
1
1148 23000 11639492 [1, 2] 10774408
2
1148 -23000 11639324 [1, 2, 3] 10774408
3
1148 -23000 11639324 [1, 2, 3, 4] 10774408
4

S.Lott a raison. Les raccourcis multitraitement de Python vous donnent effectivement un morceau de mémoire séparé et dupliqué.

Sur la plupart des systèmes * nix, utiliser un appel à un niveau inférieur os.fork()vous donnera en fait une mémoire de copie sur écriture, ce que vous pensez peut-être. AFAIK, en théorie, dans le programme le plus simpliste possible, vous pouvez lire ces données sans les dupliquer.

Cependant, les choses ne sont pas aussi simples dans l'interpréteur Python. Les données d'objet et les méta-données sont stockées dans le même segment de mémoire, donc même si l'objet ne change jamais, quelque chose comme un compteur de référence pour cet objet incrémenté provoquera une écriture mémoire, et donc une copie. Presque tous les programmes Python qui font plus que "imprimer" bonjour "" entraîneront des incréments du nombre de références, donc vous ne réaliserez probablement jamais l'avantage de la copie sur écriture.

Même si quelqu'un réussissait à pirater une solution de mémoire partagée en Python, essayer de coordonner le ramassage des ordures entre les processus serait probablement assez pénible.

Si vous utilisez Unix, ils peuvent partager le même objet, en raison du fonctionnement de fork (c'est-à-dire que les processus enfants ont une mémoire séparée mais c'est une copie en écriture, donc elle peut être partagée tant que personne ne la modifie). J'ai essayé ce qui suit:

import multiprocessing

x = 23

def printx(y):
    print x, id(x)
    print y

if __name__ == '__main__':
    pool = multiprocessing.Pool(processes=4)
    pool.map(printx, (1,2,3,4))

et a obtenu la sortie suivante:

$ ./mtest.py
23 22995656
1
23 22995656
2
23 22995656
3
23 22995656
4

Bien sûr, cela ne prouve pas qu'une copie n'a pas été faite, mais vous devriez pouvoir le vérifier dans votre situation en regardant la sortie de pspour voir la quantité de mémoire réelle utilisée par chaque sous-processus.

Différents processus ont un espace d'adressage différent. Comme exécuter différentes instances de l'interpréteur. C'est à cela que sert l'IPC (communication interprocessus).

Vous pouvez utiliser des files d'attente ou des tuyaux à cette fin. Vous pouvez également utiliser rpc sur tcp si vous souhaitez distribuer les processus sur un réseau ultérieurement.

http://docs.python.org/dev/library/multiprocessing.html#exchanging-objects-between-processes

Pas directement lié au multitraitement en soi, mais d'après votre exemple, il semblerait que vous puissiez simplement utiliser le module shelve ou quelque chose comme ça. Le "big_lookup_object" doit-il vraiment être complètement en mémoire?

Non, mais vous pouvez charger vos données en tant que processus enfant et lui permettre de partager ses données avec d'autres enfants. voir ci-dessous.

import time
import multiprocessing

def load_data( queue_load, n_processes )

    ... load data here into some_variable

    """
    Store multiple copies of the data into
    the data queue. There needs to be enough
    copies available for each process to access. 
    """

    for i in range(n_processes):
        queue_load.put(some_variable)


def work_with_data( queue_data, queue_load ):

    # Wait for load_data() to complete
    while queue_load.empty():
        time.sleep(1)

    some_variable = queue_load.get()

    """
    ! Tuples can also be used here
    if you have multiple data files
    you wish to keep seperate.  
    a,b = queue_load.get()
    """

    ... do some stuff, resulting in new_data

    # store it in the queue
    queue_data.put(new_data)


def start_multiprocess():

    n_processes = 5

    processes = []
    stored_data = []

    # Create two Queues
    queue_load = multiprocessing.Queue()
    queue_data = multiprocessing.Queue()

    for i in range(n_processes):

        if i == 0:
            # Your big data file will be loaded here...
            p = multiprocessing.Process(target = load_data,
            args=(queue_load, n_processes))

            processes.append(p)
            p.start()   

        # ... and then it will be used here with each process
        p = multiprocessing.Process(target = work_with_data,
        args=(queue_data, queue_load))

        processes.append(p)
        p.start()

    for i in range(n_processes)
        new_data = queue_data.get()
        stored_data.append(new_data)    

    for p in processes:
        p.join()
    print(processes)    

Pour la plate-forme Linux / Unix / MacOS, forkmap est une solution rapide et sale.