Verschachteltes DictProxy-Objekt mit Mehrfachverarbeitung lässt die Verwendung von .items() nicht zu ⇐ Python

[Anonymous]

Wie in diesem Beitrag angedeutet, erlaubt Manager().dict() jeweils nur die Ausführung einer Instanz des Multiprocessing-Objekts. Aus diesem Grund versuche ich, das verschachtelte Wörterbuch v von for k,v in sharedDict.items() in ein Diktatobjekt zu konvertieren. Aber jedes Mal, wenn ich versuche, das Manager-DictProxy-Objekt mit .items oder .get aufzurufen, stürzt das Programm ab. In meinem Fall ist letzteres, .get, 1 zu 5 pro Chance, wenn ich auf „Programm ausführen“ klicke.

Der Zweck des Wörterbuchs besteht darin, den Überblick darüber zu behalten, welche Befehle ausgeführt wurden und in welcher Warteschlange sie gerade ausgeführt werden, sobald sie mit der Ausführung begonnen haben (normalerweise eine von vier Multiprocessing-Warteschlangen). Der Einfachheit halber ist der Wert des verschachtelten Schlüssels vorerst nur ein Zeichen statt einer Liste, um den Schuldigen dieses Problems zu diagnostizieren.

Typischer Fehler ist Datei „“, Zeile 2, in Schlüsseln und BrokenPipeError: [Errno 32] Broken Pipe
Der Fehler beginnt bei der Methode, def traverseRootAndNestedKey(self, nestedKey):, der dabei hilft, den Wert des Befehlsstatus zu aktualisieren. Ich habe das vollständige Programm eingefügt.
import multiprocessing
import multiprocessing.managers
import os, sys
import pprint

class ProgressDictionary:
def __init__(self):
self.dictionary = {}
# E.g. k[0] = {'Root Key A': Manager().dict({'Nested key A_1': [1, 0],
# 'Nested key A_2': [1, 0],
# 'Nested key A_3': [1, 0],
# 'Nested key A_4': [1, 0]})}
self.mprocDictionary = multiprocessing.Manager().dict()
# Will need to be rebuilt each time self.dictionary is .updated()
# E.g. mprocDictionary.update())
# = Manager().dict{'Root Key A': Manager().dict({'Nested key A_1': [1, 0],
# ...
# 'Nested key A_4': [1, 0]}),
# 'Root Key B': Manager().dict({'Nested key B_1': [1, 0],
# ...
# 'Nested key B_4': [1, 0]})}
self.container = multiprocessing.Manager().dict() # Not used

def buildMprocDict(self, rootKey, listOfNestedKeys):
self.dictionary.update({rootKey: self.addNest(listOfNestedKeys)})
# Debug
# pprint.PrettyPrinter(width=128, sort_dicts=False).pprint(self.dictionary)
# No braces as intended -> 'Root Key A':
self.mprocDictionary.update(self.dictionary)
# Debug
# print(type(self.mprocDictionary))
# {'Root Key A': ,
# 'Root Key B': }
# type(self.mprocDictionary):

# Not used
self.container.update({i : multiprocessing.Manager().dict()
for i in self.mprocDictionary.keys()})
# Debug
# for k, v in self.mprocDictionary.items():
# print(f'{k} ---:--- {v}')
# print(f'K43: {[i for i in self.mprocDictionary.keys()]}')
# ['Root Key A', 'Root Key B']
# print('L26: class function exit')
def addNest(self, listOfNestedKeys):
# Encapsulate nested dicts
return multiprocessing.Manager().dict({i: 'A' for i in listOfNestedKeys})
def traverseNestedKeyOnly(self, rootKey, nestedKey):
for k, v in dict(self.mprocDictionary.get(rootKey, {}).items()):
if k == nestedKey:
return k
def traverseRootAndNestedKey(self, nestedKey):

for k in self.mprocDictionary.keys():
print(f'L56 {self.mprocDictionary.get(k).items()}') # Works
print(f'L57 {type(dict(self.mprocDictionary.get(k)))}') # Will occasionally generate error (1 in 5 chance)
# multiprocessing.managers.DictProxy'> w/out(dict())
print(f'L58 {(dict(self.mprocDictionary.get(k))).items()}') # File "", line 2, in keys
sys.exit()
for k2, v in dict(self.mprocDictionary.get(k)).items(): # crashes, File "", line 2, in keys
print(f'L64 {k2} -:- {v}')
sys.exit()
# temp_dict = {k: dict(v) if isinstance(v, multiprocessing.managers.DictProxy) else v
# for k, v in self.mprocDictionary.items()}
#pprint.PrettyPrinter(width=96, sort_dicts=False).pprint(temp_dict)
sys.exit()
for k, v in self.mprocDictionary.items(): #File "", line 2, in __getitem__
for k2, v2 in v.items(): # We only know value, but not key
# In pyCharm, v.items() gives warning: Unresolved attribute reference 'items' for class 'str'
if k2 == nestedKey:
return [k, k2]
#if nested key is not found due to being deleted earlier for whatever reason
return None
def setMemAddress(self, nestedKey, pid, debug_helper):
# *args used, incase root key is never supplied.
print(f'L64 argument show: {debug_helper}') # No show
k, k2 = self.traverseRootAndNestedKey(nestedKey) # for k, v in self.mprocDictionary.items():
# File "", line 2, in items
# Sometimes will cut off here, and won't show the following statements
# Converting dictProxy to dict: for k, v in dict(self.mprocDictionary).items(),
# ...In traverseRootAndNestedKey() method .did not resolve this
print(f'L65 key: {k}, val: {k2}') # display correct values
#print(f'L68 {self.mprocDictionary[k]}') #
#print(f'L69 {dict(self.mprocDictionary[k])}') # File "", line 2, in __getitem__
if k is not None:
# error here, nothing is shown
temp = self.mprocDictionary[k][k2]
print(f'L93 {temp}') # Will sometimes show None, other times 'A' w/Multiproc
temp = pid
# Alternative method
self.mprocDictionary[k][k2].update(temp)
def getMemAddress(self, rootKey, nestedKey):
k = self.traverseNestedKeyOnly(rootKey, nestedKey)
try:
return self.mprocDictionary[rootKey][k]
except IndexError:
return None
def printDictionary(self):
pprint.PrettyPrinter(width=96, sort_dicts=False).pprint(self.dictionary)
def printMprocDictionary(self):
for k, v in self.mprocDictionary.items():
for k2, v2 in v.items():
print(f'L78 {k2} : {v2}')

splitCMDqueue = multiprocessing.Queue()

qList = [multiprocessing.Queue() for i in range(2)]
listCMDs = []
listCMDs.append(['''system cmd a_1''',
'''system cmd a_2''',
'''system cmd a_3''',
'''system cmd a_4'''])
listCMDs.append(['''system cmd b_1''',
'''system cmd b_2''',
'''system cmd b_3''',
'''system cmd b_4'''])
listCMDcondition = ['''condition_system_cmd_1''',
'''condition_system_cmd_2''']

# {'Root Key A': {
# 'Nested key A_1': [1, 0],
# 'Nested key A_2': [1, 0],
# 'Nested key A_3': [1, 0],
# 'Nested key A_4': [1, 0]},
# 'Root Key B': {
# 'Nested key B_1': [1, 0],
# 'Nested key B_2': [1, 0],
# 'Nested key B_3': [1, 0],
# 'Nested key B_4': [1, 0]}}
def worker(sharedClass, nK):
sharedClass.setMemAddress(nK, '0x12345FA', 'Call from Line 110')
#print("L48") # Shows when previous statement is commented out
print(f'L130, {sharedClass.getMemAddress(listCMDcondition[0], nK)}')

jobs = []
runtimeCommandProgress = ProgressDictionary()

if __name__ == '__main__':

for idx, i in enumerate(listCMDs):
runtimeCommandProgress.buildMprocDict(listCMDcondition[idx], i) # works/success
for j in i:
qList[idx].put(j)

example_cmd = listCMDs[0][1] # Nested key A_2, system cmd a_2
# Both work
# runtimeCommandProgress.setMemAddress(example_cmd, '0x12345FA', 'L136')
# print(f'L133 {runtimeCommandProgress.getMemAddress(listCMDcondition[0], example_cmd)}')

p = multiprocessing.Process(target=worker, args=(runtimeCommandProgress, example_cmd))

jobs.append(p)
for p in jobs:
p.start()