deepcopy eines objekts mit shelve instanz

[Boa]

Hi,

Statt mein Objekt einem anderen Prozess zu übergeben möchte ich davon eine Kopie erstellen. Soweit ich es verstanden habe wir das automatisch gemacht, da das Objekt beim erstellen eines Prozesses serialisiert/deserialisiert wird, über das pickle Modul. Dabei werden die Methoden __setattribute__ und __getattribute__ aufgerufen. Diese möchte ich nun überschreiben, da das Objekt eine Instanz von shelve hat, welche man nicht per deepcopy kopieren kann. Wie kann ich die beiden Methoden überschreiben, sodass z.B. beim "kopieren" die shelve Instanz mit einem dict ersetzt wird, dass die gleichen Werte wie das shelve hat?

MfG,
Boa

[BlackJack]

@Boa: Ich denke Du verwechselt da Methoden und meintest `__getstate__()` und `__setstate__()`. Du musst nur `__getstate__()` überschreiben und dort ein Wörterbuch (`dict`) mit allen Attributen/Werten des Objekts zurück geben. Und beim `shelve`-Objekt in diesem Wörterbuch das `shelve`-Objekt durch ein Wörterbuch ersetzen in dass Du den Inhalt des `shelve`\s kopierst. Ungetestet:

Code: Alles auswählen

    def __getstate__(self):
        result = dict((n, getattr(self, n)) for n in dir(self))
        result['the_shelve'] = dict(result['the_shelve'])
        return result

[Boa]

Ich denke Du verwechselt da Methoden und meintest `__getstate__()` und `__setstate__()`.

Ja, das stimmt. Anscheinend wird `__getstate__()` nicht aufgerufen, wenn ich die Variable an den Prozess übergebe. Über welchen Mechanismus wird die Kopie denn sonst erstellt? Denn eine geteilte Resource ist das Objekt ja nicht, wenn sich Änderungen im Prozess nicht auf das Objekt im Elternprozess auswirken. Ich könnte auch mit dem copy Modul vorher eine Kopie anfertigen und diese Kopie als Parameter übergeben; aber dann habe ich die Kopie sowohl im Eltern als auch im Kind Prozess.
Beispielcode:

Code: Alles auswählen

from multiprocessing import Process
import shelve, random

def f(name):
    print 'hello', name
    name.set_name('alice')
    
class Name(object):
    def __init__(self, name):
        self.myshelve = {}     #shelve.open("/tmp/"+str(random.randint(100000,1000000)))
        self.myshelve['name'] = name
    def set_name(self, name):
        self.myshelve['name'] = name
    def __str__(self):
        return self.myshelve['name']
    def __getstate__(self):
        print "getstate"
        return super(Name, self).__getstate__() 

if __name__ == '__main__':
    name = Name('bob')
    p = Process(target=f, args=(name,))
    p.start()
    p.join()
    print name

[BlackJack]

@Boa: Wie es aussieht wird da, zumindest unter Unices, gar nichts besonderes getan, denn die Kopie entsteht ja im Speicher schon durch den `fork()`-Aufruf. In dem Zusammenhang wirft also noch nicht einmal das `shelve`-Objekt ein direktes Problem auf. Eventuell aber ein Indirektes, weil dann zwei Prozesse die gleiche darunter liegende Datei verwenden. Kommt also darauf an ob die Datenbank die konkret für das `shelve` verwendet wird, mit parallelen Schreibzugriffen klar kommt. Sofern man so etwas in den Prozessen dann macht.

[Boa]

Hallo BlackJack,

Danke für die Erklärung. Ich wusste nicht, dass die Prozesse mit fork implementiert sind, auch wenn es im Nachhinein naheliegend scheint. Es wäre gut, wenn das explizit in der Dokumentation erwähnt wird. Ich dachte das funktioniert über Reflection
Das Problem mit shelve ist wie du gesagt hast, dass es nicht für Nebenläufigkeit entworfen wurde. Meine Lösung nutzt die Methode __deepcopy__() um eine handgefertigte Kopie meines Objektes anzulegen. Ich verwende statt shelve nun das Nebenläufigkeits unterstützende dictionary dict() von multiprocessing.Manager, welches von allen Kopien geteilt wird. Um die Daten beim Neustart zu erhalten, speichere ich diese per atexit in ein shelve ab. Das klappt in dem Fall, weil mein dict nicht allzu groß wird. Ansonsten müsste man mit einer Datenbank arbeiten, die Nebenläufigkeit unterstützt (z.B. Berkeley DB).
Hier ein Beispiel:

Code: Alles auswählen

from multiprocessing import Process
import shelve, random
from copy import deepcopy
from multiprocessing import Manager
import atexit

def rename(person):
    print 'hello', person
    person.set_name('alice')
    print person.is_copy
    print 'renamed to', person
   
class Person(object):
    def __init__(self, name):
        self.is_copy = False
        manager = Manager()
        self.name_db = manager.dict()
        self.name_db['name'] = name
        self.name_db_dir = "/tmp"
        self.name_db_path = self.name_db_dir+"/person.db"
        try:
            last_session_revisions = shelve.open(self.name_db_path)
            self.name_db.update(last_session_revisions)
        except:
            print "Database from last session could not be loaded."
        atexit.register( lambda : self._close() )
    def set_name(self, name):
        self.name_db['name'] = name
    def __str__(self):
        return self.name_db['name']
    def __getstate__(self):
        print "getstate"
        return super(Person, self).__getstate__()        
    def __deepcopy__(self, memo):
        cls = self.__class__
        result = cls.__new__(cls)
        memo[id(self)] = result
        for k, v in self.__dict__.items():
            if k == 'name_db':
                print "do not deepcopy shared db"
                setattr(result, k, self.name_db)
            elif k == 'is_copy':
                setattr(result, k, True)
            else:
                setattr(result, k, deepcopy(v, memo))
        return result        
    def _close(self):
        print "closing"
        if not self.is_copy:
            try:
                os.makedirs(self.name_db_dir)
            except:
                pass
            self.name_db = shelve.open(self.name_db_path)
            self.name_db.update(self._revisions)
            self.name_db.close()

if __name__ == '__main__':
    person = Person('bob')
    p = Process(target=rename, args=(deepcopy(person),))
    p.start()
    p.join()
    print "hello", person