Wie über einen dict-Proxy in Python iterieren?

Question

Ich benutze Pythons multiprocessing.Manager, um den Zugriff auf ein Dataset zu teilen, das ein Prozess generiert und andere anzeigen. Allerdings renne ich in das Problem, dass ein von manager.dict() zurückgegebener dict-Proxy iteritems() nicht unterstützt.

Ich könnte über items() iterieren, aber das bedeutet, ein neues Tupel aller Elemente im Dict zu konstruieren, die eine große Zahl ist. Gibt es eine Möglichkeit, dies zu tun, ohne eine Zwischenliste/Tupel zu konstruieren, so dass nur eine konstante Menge an zusätzlichem Speicher verwendet wird?

Hinweis: Es ist OK, wenn die Lösung erfordert, dass der Generierungsprozess für die Iteration pausiert.

Answer 1

Sie könnten über keys() iterieren, um Ihren Speicherbedarf zu reduzieren. Du musst dich davor schützen, dass Schlüssel gelöscht werden.

Ansonsten ist hier ein Beispiel mit zwei verschiedenen Möglichkeiten, mit denen Sie die Elemente in einem Diktat durchlaufen können. Die Methode iteritems() in diesem Beispiel funktioniert nur vom Prozess, der das Manager-Objekt und den untergeordneten Prozess erstellt, die das Manager-Objekt erstellt. Dies liegt daran, dass das Manager-Objekt zum Erstellen neuer Proxys benötigt wird und andere Prozesse keinen Zugriff darauf haben. Die Methode iteritems2() funktioniert von anderen Prozessen, da sie nicht darauf angewiesen ist, in diesen Prozessen einen neuen Proxy zu erstellen.

import multiprocessing as mp 
import multiprocessing.managers 

class mydict(dict): 
    def __init__(self, *args, **kwargs): 
     dict.__init__(self, *args, **kwargs) 
     self.iters = {} 

    def iteritems(self): 
     print "iteritems", mp.current_process() 
     return dict.iteritems(self) 

    def _iteritems_start(self): 
     print "_iteritems_start", mp.current_process() 
     i = dict.iteritems(self) 
     self.iters[id(i)] = i 
     return id(i) 

    def _iteritems_next(self, iter_id): 
     try: 
      return self.iters[iter_id].next() 
     except StopIteration: 
      del self.iters[iter_id] 
      return None 

class mydict_proxy(mp.managers.DictProxy): 
    def iteritems(self): 
     print "iteritems proxy", mp.current_process() 
     return self._callmethod("iteritems") 

    def iteritems2(self): 
     print "iteritems2 proxy", mp.current_process() 
     iter_id = self._callmethod("_iteritems_start") 
     def generator(): 
      while True: 
       a = self._callmethod("_iteritems_next", 
          (iter_id,)) 
       if a == None: 
        return 
       yield a 
     return generator() 

    _method_to_typeid_ = { "iteritems": "Iterator" } 
    _exposed_ = mp.managers.DictProxy._exposed_ 
    _exposed_ += ("iteritems", "_iteritems_start", "_iteritems_next") 

class mymanager(mp.managers.BaseManager): 
    pass 
mymanager.register("mydict", mydict, mydict_proxy) 
mymanager.register("Iterator", proxytype = mp.managers.IteratorProxy, 
      create_method = False) 

def other(d): 
    for k, v in d.iteritems2(): 
     d[k] = v.lower() 
    for k, v in d.iteritems(): 
     d[k] = ord(v) 

def main(): 
    manager = mymanager() 
    manager.start() 
    d = manager.mydict(list(enumerate("ABCDEFGHIJKLMNOP"))) 
    for (k, v) in d.iteritems(): 
     print k, v 
    proc = mp.Process(target = other, args = (d,)) 
    proc.start() 
    proc.join() 
    for (k, v) in d.iteritems(): 
     print k, v 

if __name__ == "__main__": 
    main() 

Beachten Sie, dass während dieser Code mehr Speicher effizient sein kann, es ist wahrscheinlich ein Heck viel langsamer zu gehen.

Answer 2

iteritems() ist für eine Liste dict. Sie könnten eine for-Schleife verwenden. Oder Sie könnten sagen: sorted(), die Schlüssel in einer sortierten Liste zurückgeben und dann über diese Liste iterieren und dict[key] tun. Ich hoffe, das hilft. Wenn es einen besseren Weg gibt. Teile mit mir. Ich möchte unbedingt wissen.

Answer 3

Sie können die Klasse SyncManager verwenden, um Ihre eigenen Typen zu registrieren. Dann können Sie Methoden für diesen Typ implementieren, z. um nur eine begrenzte Anzahl von Gegenständen von einem Diktat zu erhalten.

Hier ist ein Beispiel für den Einstieg:

import multiprocessing 
from multiprocessing import managers 


class TakerDict(dict): 
    """Like a dict, but allows taking a limited number of items.""" 

    def take(self, items=1): 
     """Take the first `items` items.""" 
     return [item for _, item in zip(range(items), self.items())] 


# NOTE: add other dict methods to the tuple if you need them. 
TakerProxy = managers.MakeProxyType('TakerProxy', ('take',)) 

managers.SyncManager.register('taker', TakerDict, TakerProxy) 


if __name__ == '__main__': 
    manager = multiprocessing.Manager() 
    taker = manager.taker() 
    # in other processes, use e.g. taker.take(5) 

Somit Speicherverbrauch zu begrenzen, würden Sie den Manager-Prozess aufrufen müssen, um immer wieder die nächste Charge von Elementen zu erhalten.

Um dies zu tun, müsste Ihr Diktat die Indizierung unterstützen (Sie können also von einem bestimmten Offset aus fortfahren). Da Sie keinen Zugriff auf die zugrunde liegende Reihenfolge der Elemente in einem Dict haben, wäre es wahrscheinlich besser, stattdessen eine Liste zu verwenden (z. B. manager.list()). Dann fragen Sie in Ihren Subprozessen nach der len() der Liste, und indexieren Sie durch eine Scheibe, um einen Stapel der passenden Größe zu erhalten - Sie müssen keinen Proxy-Typ dafür registrieren.