Python multitraitement l'écriture dans un fichier

J'essaye de résoudre un gros problème numérique qui implique beaucoup de sous-problèmes, et je suis en utilisant Python module multiprocessing (plus précisément de la Piscine.carte) pour séparer les différents sous-problèmes indépendants sur différents cœurs. Chaque subproblem implique le calcul de beaucoup de sous-sous-problèmes, et je suis en train de efficacement memoize ces résultats en les stockant dans un fichier si ils n'ont pas été calculé par un processus encore, sinon passer le calcul et il suffit de lire les résultats à partir du fichier.

Je vais avoir des problèmes de concurrence avec les fichiers: différents processus parfois vérifier pour voir si un sous-subproblem a été calculée pour l'instant (en recherchant le fichier où les résultats seraient stockées), de voir qu'il n'a pas, lancer le calcul, puis essayez d'écrire les résultats dans le même fichier en même temps. Comment puis-je éviter d'écrire des collisions de ce genre?

Découvrez un exemple tiré de la documentation de l'aide de multiprocessing.Lock pour synchroniser plusieurs processus.
Vous pourriez avoir un seul processus de l'écriture des résultats, avec une File d'attente en entrée qui pourrait être alimenté par les autres processus de travail. Je crois qu'il serait plus sûr d'avoir tous les processus de travail en lecture seule.
Je devrais avoir mentionné que, pour compliquer encore les choses, je suis en cours d'exécution à plusieurs gros problèmes principaux dans le même temps sur un cluster, avec chacun l'écriture des résultats à la sous-sous-problèmes sur le même réseau système de fichiers. Donc je peux obtenir des collisions de processus s'exécutant sur des machines distinctes entièrement (donc je ne pense pas que les solutions en utilisant des choses comme le multitraitement.Verrouillage de travail).
Est le problème que vous rencontrez avec l'écriture des fichiers de collisions, ou est-il juste que vous ne voulez pas reproduire les travaux dans les situations où un travailleur commence la résolution d'un sous-subproblem tandis qu'un autre est déjà au travail sur elle? Ce dernier est un peu plus difficile à résoudre (plus de synchronisation est nécessaire).
Bien à l'origine, j'avais l'écriture des fichiers de collisions, mais je trouve que la vérification du fichier de l'existence immédiatement avant l'écriture (au lieu de compter sur la case je faire avant de commencer le calcul de la sous-subproblem) a pris soin de cela. Maintenant c'est plus le dernier; je voudrais éviter un double travail, si possible (et peut imaginer d'autres personnes dans la même situation).
Si votre réseau système de fichiers prend en charge le verrouillage de fichier, vous pouvez utiliser le système d'exploitation de fichier spécifique à la méthode de création exclusivement à créer le fichier et maintenez un verrou exclusif sur elle jusqu'à ce que les résultats sont prêts, puis fermez-le. Tout processus qui n'a pas à "gagner" le créer des la course essayais de l'ouvrir et de ré-essayer (avec un délai) jusqu'à ce que l'ont été en mesure de l'ouvrir, puis ils peuvent lire les résultats.
Ah, merci JimP! Qui ressemble exactement à ce dont j'ai besoin. Je vais le regarder.
Vous êtes essentiellement de la programmation d'un serveur de base de données ici. Avez-vous envisagé d'utiliser une existante?

InformationsquelleAutor Big Dogg | 2012-11-19

io multiprocessing mutex python

98

@GP89 mentionné une bonne solution. Utiliser une file d'attente pour envoyer les tâches en écriture à un processus spécifique qui a seul l'accès en écriture au fichier. Tous les autres travailleurs ont accès en lecture seule. Cela permettra d'éliminer les collisions. Voici un exemple qui utilise apply_async, mais il faudra travailler avec la carte trop:
```
import multiprocessing as mp
import time

fn = 'c:/temp/temp.txt'

def worker(arg, q):
    '''stupidly simulates long running process'''
    start = time.clock()
    s = 'this is a test'
    txt = s
    for i in xrange(200000):
        txt += s 
    done = time.clock() - start
    with open(fn, 'rb') as f:
        size = len(f.read())
    res = 'Process' + str(arg), str(size), done
    q.put(res)
    return res

def listener(q):
    '''listens for messages on the q, writes to file. '''

    f = open(fn, 'wb') 
    while 1:
        m = q.get()
        if m == 'kill':
            f.write('killed')
            break
        f.write(str(m) + '\n')
        f.flush()
    f.close()

def main():
    #must use Manager queue here, or will not work
    manager = mp.Manager()
    q = manager.Queue()    
    pool = mp.Pool(mp.cpu_count() + 2)

    #put listener to work first
    watcher = pool.apply_async(listener, (q,))

    #fire off workers
    jobs = []
    for i in range(80):
        job = pool.apply_async(worker, (i, q))
        jobs.append(job)

    # collect results from the workers through the pool result queue
    for job in jobs: 
        job.get()

    #now we are done, kill the listener
    q.put('kill')
    pool.close()

if __name__ == "__main__":
   main()
```
- Hey Mike, merci pour la réponse. Je pense que ce serait travailler pour la question comme je l'ai formulé, mais je ne suis pas si sûr que ça va régler le problème comme indiqué dans les commentaires à la question de savoir comment, concrètement, j'ai plusieurs principaux programmes en cours d'exécution à travers plusieurs machines sur un réseau de système de fichiers, ce qui peut avoir des processus qui va essayer d'écrire dans le même fichier. (FWIW, je suis autour de mon problème personnel dans un hacky façon il y a un moment mais je suis en commentant dans le cas d'autres ont des problèmes similaires.)
- Je tiens vraiment à upvote ce à plusieurs reprises. Cela a été utile, donc beaucoup de temps pour moi. Une fois de plus aujourd'hui.
- Merci Mike - j'avais été aux prises avec la façon d'utiliser les MP des Files d'attente. Votre exemple est très clair et simple.
- J'ai dû ajouter un pool.join() ci-dessous pool.close(). Sinon, ma travailleurs devraient se terminer avant l'auditeur et le processus serait de tout arrêter.
- Merci beaucoup pour cette! Notez que j'ai dû inclure herrherr la suggestion de la peur, il peut provoquer une difficile à détecter un défaut dans au moins mon scénario.
- Qu'en est-il de la consommation est largement dépassés en nombre et les causes des problèmes de mémoire? Comment voulez-vous mettre en œuvre plusieurs consommateurs ont tous écrit pour le même fichier?
- pourquoi mp.cpu_count() + 2 lors de la définition de nombre de processus?
- Après l'adoption de ce code, mon programme se ferme avant que l'auditeur a terminé son travail, comment pourrais-je résoudre ce problème?
- Cela fonctionne très bien, sauf qu'il met mes sorties dans un ordre aléatoire sur le disque, au lieu de le faire dans l'ordre que j'ai envoyer des données à travers. Je suis l'aide de la carte plutôt que async pour les threads de travail. Savez pas comment résoudre ce problème.
- Testé sur Linux, besoin de changer f = open(fn, 'wb') à f = open(fn, 'w') pour stocker le résultat, l'autre sage le fichier de sortie sera vide pendant que le code peut fonctionner comme un charme.
InformationsquelleAutor MikeHunter

Il me semble que vous devez utiliser Manager pour enregistrer temporairement vos résultats à une liste, puis d'écrire les résultats de la liste dans un fichier. Aussi, l'utilisation starmap à passer de l'objet que vous souhaitez traiter et de la gestion de la liste. La première étape consiste à construire le paramètre à passer à starmap, qui comprend la liste.

from multiprocessing import Manager
from multiprocessing import Pool  
import pandas as pd

def worker(row, param):
    # do something here and then append it to row
    x = param**2
    row.append(x)

if __name__ == '__main__':
    pool_parameter = [] # list of objects to process
    with Manager() as mgr:
        row = mgr.list([])

        # build list of parameters to send to starmap
        for param in pool_parameter:
            params.append([row,param])

        with Pool() as p:
            p.starmap(worker, params)

À partir de ce point, vous devez décider comment vous allez gérer la liste. Si vous avez des tonnes de RAM et une énorme quantité de données hésitez pas à concaténer à l'aide de pandas. Ensuite, vous pouvez enregistrer des fichiers très facilement au format csv ou un cornichon.

        df = pd.concat(row, ignore_index=True)

        df.to_pickle('data.pickle')
        df.to_csv('data.csv')

InformationsquelleAutor fizix137

Vous devez vous connecter pour publier un commentaire.