How do I use atpbar with multiprocessing.Pool?

[thomasdziedzic]

I have the following example code:

import multiprocessing
from atpbar import atpbar, register_reporter, find_reporter, flush
import more_itertools

multiprocessing.set_start_method('fork', force=True)

def f(xs, reporter):
    register_reporter(reporter)

    count = 0

    for x in atpbar(list(xs), name = multiprocessing.current_process().name):
        count += 1

    return count

items = range(80000000)
workloads = more_itertools.divide(8, items)

reporter = find_reporter()

with multiprocessing.Pool(8) as p:
    ret = p.starmap(f, zip(workloads, [reporter, reporter, reporter, reporter, reporter, reporter, reporter, reporter]))
    flush()
    print(sum(ret))

This gives me:

Traceback (most recent call last):
  File "main.py", line 23, in <module>
    ret = p.starmap(f, zip(workloads, [reporter, reporter, reporter, reporter, reporter, reporter, reporter, reporter]))
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/pool.py", line 372, in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/pool.py", line 771, in get
    raise self._value
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/pool.py", line 537, in _handle_tasks
    put(task)
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/reduction.py", line 51, in dumps
    cls(buf, protocol).dump(obj)
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/queues.py", line 58, in __getstate__
    context.assert_spawning(self)
  File "/home/tom/.pyenv/versions/3.8.6/lib/python3.8/multiprocessing/context.py", line 359, in assert_spawning
    raise RuntimeError(
RuntimeError: Queue objects should only be shared between processes through inheritance

[TaiSakuma]

@thomasdziedzic, thank you for your question.

Can you try the following code? It works in my environment.

import multiprocessing
from atpbar import atpbar, register_reporter, find_reporter, flush
import more_itertools

multiprocessing.set_start_method('fork', force=True)

def f(xs):

    count = 0

    for x in atpbar(list(xs), name = multiprocessing.current_process().name):
        count += 1

    return count

items = range(80000000)
workloads = more_itertools.divide(8, items)

reporter = find_reporter()

with multiprocessing.Pool(8, register_reporter, [reporter]) as p:
    ret = p.map(f, workloads)
    flush()
    print(sum(ret))

I see progress bars like below.

  12.87% :::::                                    |  1287216 / 10000000 |:  ForkPoolWorker-8                                                                                                               
  12.98% :::::                                    |  1297716 / 10000000 |:  ForkPoolWorker-1                                                                                                               
  12.91% :::::                                    |  1291065 / 10000000 |:  ForkPoolWorker-3                                                                                                               
  12.96% :::::                                    |  1295941 / 10000000 |:  ForkPoolWorker-6                                                                                                               
  13.05% :::::                                    |  1304935 / 10000000 |:  ForkPoolWorker-5                                                                                                               
  12.73% :::::                                    |  1272576 / 10000000 |:  ForkPoolWorker-2                                                                                                               
  12.69% :::::                                    |  1269152 / 10000000 |:  ForkPoolWorker-4                                                                                                               
  12.82% :::::                                    |  1282320 / 10000000 |:  ForkPoolWorker-7                     

[thomasdziedzic]

Thanks that worked!