concurrent.futures.ProcessPoolExecutor() not calling method inside class

user_11447688 0
Viewed 693

I am tring to use concurrent.futures.ProcessPoolExecutor() inside a class. but method operation_data which is responsible to call method operation is not calling it. "inside concurrent insert" and "do some operations" is not getting printed.

import concurrent.futures

class ConcurrentTest:

    def operation(self, chunk):
        print("inside concurrent insert")
        print("do some operations")

    def operation_data(self):
        chuck_list = [chunk for chunk in self.chunks([millions_of_data_in_this_list], 10000)]
        print('chunk list done')
        with concurrent.futures.ProcessPoolExecutor() as executor:
            for c in chuck_list:
                executor.submit(operation, c)
        print("done")

    @staticmethod
    def chunks(l, n):

        for i in range(0, len(l), n):  
            yield l[i:i + n]


concurrent_test = ConcurrentTest()
concurrent_test.operation_data()

Am I doing anything wrong? Please suggest.

Thanks in advance!

python-3.x multithreading multiprocessing python
1 Answers

change executor.submit(operation, c) to

executor.submit(self.operation, c)

or use Executor.map() instead

executor.map(self.operation, chunk_list)

Note, using map() allows you to submit the original list in chunks, allowing you to further simplify your code by passing your big list directly

executor.map(self.operation, your_big_list, chunksize=n)

EDIT: As noted in the comments by @Booboo you will not get parallelization by calling future.result() immediately after submit, so I edited my answer accordingly.

EDIT2: Here is full example with executor.map whit all unnecessary code removed and using chunksize argument

import concurrent.futures

class ConcurrentTest:
    def operation(self, chunk):
        print(f"{chunk} inside concurrent insert")
        print(f"do some operations with {chunk}")

    def operation_data(self):
        with concurrent.futures.ProcessPoolExecutor() as executor:
            executor.map(self.operation, range(20), chunksize=3)
        print("done")

concurrent_test = ConcurrentTest()
concurrent_test.operation_data()

output (yours may differ):

0 inside concurrent insert
do some operations with 0
1 inside concurrent insert
do some operations with 1

--- skipped for brevity ---

13 inside concurrent insert
do some operations with 13
14 inside concurrent insert
do some operations with 14
18 inside concurrent insert
do some operations with 18
19 inside concurrent insert
do some operations with 19
15 inside concurrent insert
11 inside concurrent insert
do some operations with 15
16 inside concurrent insert
do some operations with 16
do some operations with 11
17 inside concurrent insert
do some operations with 17
done
user_40466320
Related