Skip to content

Instantly share code, notes, and snippets.

@cloudnull
Last active June 26, 2021 05:34
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save cloudnull/411a4090e0438039a6d3c2071458705b to your computer and use it in GitHub Desktop.
Save cloudnull/411a4090e0438039a6d3c2071458705b to your computer and use it in GitHub Desktop.
Showing a global dictionary not being shared across processes. Only test-with-manager.py passes the assertion.
import multiprocessing
ITEM_CONTAINER = dict(a=0, b=1,)
CONTROL_CONTAINER = dict(a=0, b=1, c=2, d=3, e=4, f=5)
def _acquire_in_thread_inheritance(test, d):
d[test] = len(d.keys())
print(d, type(d), id(d))
def main_inheritance(d):
processes = [
multiprocessing.Process(target=_acquire_in_thread_inheritance, args=(item, d,))
for item in ["c", "d", "e", "f"]
]
for t in processes:
t.daemon = True
t.start()
for t in processes:
t.join()
print("Test inheritance")
print(type(ITEM_CONTAINER), id(ITEM_CONTAINER))
print(ITEM_CONTAINER)
main_inheritance(d=ITEM_CONTAINER)
#Fails
assert CONTROL_CONTAINER == ITEM_CONTAINER
import multiprocessing
ITEM_CONTAINER = dict(a=0, b=1,)
CONTROL_CONTAINER = dict(a=0, b=1, c=2, d=3, e=4, f=5)
def _acquire_in_thread_global(test):
ITEM_CONTAINER[test] = len(ITEM_CONTAINER.keys())
print(ITEM_CONTAINER, type(ITEM_CONTAINER), id(ITEM_CONTAINER))
def main_global():
processes = [
multiprocessing.Process(target=_acquire_in_thread_global, args=(item,))
for item in ["c", "d", "e", "f"]
]
for t in processes:
t.daemon = True
t.start()
for t in processes:
t.join()
print("Test global")
print(type(ITEM_CONTAINER), id(ITEM_CONTAINER))
print(ITEM_CONTAINER)
main_global()
#Fails
assert CONTROL_CONTAINER == ITEM_CONTAINER
import multiprocessing
ITEM_MANAGER_CONTAINER = multiprocessing.Manager().dict(a=0, b=1,)
CONTROL_CONTAINER = dict(a=0, b=1, c=2, d=3, e=4, f=5)
def _acquire_in_thread_global_manager(test):
ITEM_MANAGER_CONTAINER[test] = len(ITEM_MANAGER_CONTAINER.keys())
print(ITEM_MANAGER_CONTAINER, type(ITEM_MANAGER_CONTAINER), id(ITEM_MANAGER_CONTAINER))
def main_global_manager():
processes = [
multiprocessing.Process(target=_acquire_in_thread_global_manager, args=(item,))
for item in ["c", "d", "e", "f"]
]
for t in processes:
t.daemon = True
t.start()
for t in processes:
t.join()
print("test with manager")
print(type(ITEM_MANAGER_CONTAINER), id(ITEM_MANAGER_CONTAINER))
print(ITEM_MANAGER_CONTAINER)
main_global_manager()
#Passes
assert CONTROL_CONTAINER == dict(ITEM_MANAGER_CONTAINER)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment