1
\$\begingroup\$

Tasks are distributed to producer in multiple worker process. Results from workers are sent through Queue to a Consumer at the main process. The consumer uses less time per task than the worker.

The manual counting of tasks seems unsatisfying ... Can imap or anything else make that simpler?

By the way, I guess my rough representation of time for hard disk reading is not quite correct because hard disk reads are buffered.

import multiprocessing as mp
import time
from datetime import datetime


def log(msg):
    d = datetime.utcnow().strftime('%M:%S.%f')[:-3]
    print(d + ' %s' % msg)


def main():
    n_proc = 3
    p = mp.Pool(n_proc)
    m = mp.Manager()
    q = m.Queue()
    size = 20
    c = Consumer(q, size, n_proc)
    for i in range(size):
        log("Main Loop %d" % i)
        # 0.01 sec for (buffered) reading of hard-disk for arrays.
        # values are not 0, 1, 2 ... in the actual program.
        hd_arr = i  # "array from hard disk"
        time.sleep(0.1)
        p.apply_async(producer, args=(hd_arr, q))
        c.invoke()
    log("Processsing remaining tasks")
    p.close()
    p.join()
    c.flush()
    log(c.data)


def producer(x, q):
    time.sleep(1)
    result = (x, x*x)
    q.put(result)


class Consumer():
    def __init__(self, queue, size, n_proc):
        self.queue = queue
        # 1.0 / (0.1 + 0.1) = 5
        self.lb_factor = 5 * n_proc
        self.data = [None] * size
        self.task_count = 0

    def invoke(self, task_increment = 1):
        log("  Consumer checks:")
        self.task_count += task_increment
        if self.task_count == self.lb_factor:
            self.flush()

    def flush(self):
        log("    # Consumer flushes #")
        for i in range(self.task_count):
            x, x2 = self.queue.get()
            self.data[x] = x2
            time.sleep(0.1)
        self.task_count = 0


if __name__ == "__main__":
    main()

Result:

32:45.796 Main Loop 0
32:45.896   Consumer checks:
32:45.897 Main Loop 1
32:45.997   Consumer checks:
32:45.997 Main Loop 2
32:46.098   Consumer checks:
32:46.098 Main Loop 3
32:46.198   Consumer checks:
32:46.199 Main Loop 4
32:46.299   Consumer checks:
32:46.299 Main Loop 5
32:46.400   Consumer checks:
32:46.400 Main Loop 6
32:46.500   Consumer checks:
32:46.500 Main Loop 7
32:46.601   Consumer checks:
32:46.601 Main Loop 8
32:46.701   Consumer checks:
32:46.701 Main Loop 9
32:46.802   Consumer checks:
32:46.802 Main Loop 10
32:46.903   Consumer checks:
32:46.903 Main Loop 11
32:47.003   Consumer checks:
32:47.003 Main Loop 12
32:47.104   Consumer checks:
32:47.104 Main Loop 13
32:47.204   Consumer checks:
32:47.205 Main Loop 14
32:47.305   Consumer checks:
32:47.305     # Consumer flushes #
32:51.243 Main Loop 15
32:51.343   Consumer checks:
32:51.344 Main Loop 16
32:51.444   Consumer checks:
32:51.444 Main Loop 17
32:51.545   Consumer checks:
32:51.545 Main Loop 18
32:51.645   Consumer checks:
32:51.646 Main Loop 19
32:51.746   Consumer checks:
32:51.746 Processsing remaining tasks
32:53.497     # Consumer flushes #
32:54.001 [0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361]

Simulation of Multiple Producer Consumer: Setting the boundary between consumer, producer, and monitor.

\$\endgroup\$
  • \$\begingroup\$ This doesn't work well. Can't rely on manual counting. Have to set limit on queues, and use queues to connect producer, transformer, and consumer. Otherwise, any miscalculation of load balancing can lead to out of memory error. It is better to let one step slightly wait for another step than to have out of memory error. \$\endgroup\$ – hamster on wheels Jul 27 '17 at 3:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.