How to loop in parallel to alter objects inside a dictionary in Python?

user_12994944 0
Viewed 361

Here is a minimum example of my problem:

import concurrent.futures
from functools import partial

# Object class
class obj:
    def __init__(self,tup):
        self.tup = tup

# Function includes attributes in objects of class above
def new(fi,fdic):
    fdic[fi].new = 'work_'+ str(fdic[fi].tup)
    
# Dictionary full of instances of obj above    
dic = {'a':obj(1),
       'b':obj(2),
       'c':obj(3),
       'd':obj(4),
       'e':obj(5),
       'f':obj(6),
      }

partial_new = partial(new, fdic=dic)

Now I want to multiprocess all the objects in the dictionary (because I have too many in reality). The code below runs. But it does not "work", because I actually need ProcessPool (I think? Because I want to process things in parallel).

with concurrent.futures.ThreadPoolExecutor() as executor:
    for _ in executor.map(partial_new, dic.keys()):
        pass
print(dic['b'].new)

This one does not run:

with concurrent.futures.ProcessPoolExecutor() as executor:
    for _ in executor.map(partial_new, dic.keys()):
        pass
print(dic['b'].new)

My question is: How do I make this work?

I just need to use the function to modify all the objects inside the dictionary in parallel. Later I wills save the full dictionary, but the function that I apply does not return anything (if this makes things easier).

object for-loop dictionary multiprocessing python
2 Answers

You can use ThreadPool from the multiprocess module in the following way:

  1. Create a list of the dict keys (ls = [a for a in dict.keys())
  2. Define a function that given a pointer to a dict and a key does the alteration you desire
  3. use ThreadPool's starmap() method to run that function on the list you created and the dict
  4. join and close the thread pool
user_126903130

Is the issue that it takes a long time to calculate the new value?

def get_new_value(dictionary_item):
    key, value = dictionary_item
    return key, 'work_' + str(value.tup)

with concurrent.futures.ProcessPoolExecutor() as executor:
    for key, new_value in executor.map(get_new_value, dic.items()):
        dic[key].new = new_value

You can only have one thread modifying dic. But you can pass key and value to a thread, have the thread return the key and the new value, and then have the original thread do the work of updating the dictionary.

You'll probably want to specify a chunksize to map

=== edited ===

As promised, my complete file.

import concurrent.futures

# Object class
class obj:
    def __init__(self, tup):
        self.tup = tup


# Dictionary full of instances of obj above
dic = {'a': obj(1),
       'b': obj(2),
       'c': obj(3),
       'd': obj(4),
       'e': obj(5),
       'f': obj(6),
       }


def get_new_value(dictionary_item):
    key, value = dictionary_item
    return key, 'work_' + str(value.tup)

def go():
  with concurrent.futures.ProcessPoolExecutor() as executor:
    for key, new_value in executor.map(get_new_value, dic.items()):
        dic[key].new = new_value
  # Make sure it really worked!
  for key, value in dic.items():
      print(key, value.new)


if __name__ == '__main__':
    go()
user_64574070
Related