Say we split a number into separate domains: Ex: 100 split into: [0, 25] [25, 50] [50, 75] [75, 100]. Then we send each one of those 4 lists out to one of 4 separate process' to be computed and then recombined for an answer as a single split unit of the number 100. We iterate this many times in a row with the need for the process' to work as a unit for 1000's of numbers split into separate domains similar to [0, 25] [25, 50] [50, 75] [75, 100]. An efficiency problem occurs if we have to close the process' to make them act as a single group unit that was processed for an answer. Since windows is crap compared to Unix for running process' we are forced to use the "spawn" method instead of fork. The spawn method is slow at spawning process' so I figured then why not just keep the process' open and pass data to and from them without the need to open and close them for every iteration group of parallel process'. The example code below will do this. It will keep process' open as Class Consumers that will constantly request with run() (in a while loop) for a next_task with the .get() Joinable queue:
import multiprocessing
class Consumer(multiprocessing.Process):
def __init__(self, task_queue, result_queue):
multiprocessing.Process.__init__(self)
self.task_queue = task_queue
self.result_queue = result_queue
def run(self):
while True:
next_task = self.task_queue.get()
if next_task is None:
# Poison pill shutdown of .get() loop with break
self.task_queue.task_done()
break
answer = next_task()
self.task_queue.task_done()
self.result_queue.put(answer)
return
class Task(object):
def __init__(self, a, b):
self.a = a
self.b = b
def __call__(self):
for i in range(self.b):
if self.a % i == 0:
return 0
return 1
if __name__ == '__main__':
# Establish communication queues
tasks = multiprocessing.JoinableQueue()
results = multiprocessing.Queue()
# Number of consumers equal to system cpu_count
num_consumers = multiprocessing.cpu_count()
# Make a list of Consumer object process' ready to be opened.
consumers = [ Consumer(tasks, results) for i in range(num_consumers) ]
for w in consumers:
w.start()
# Enqueue jobs for the Class Consumer process' run() while-loop to .get() a workload:
num_jobs = 10
for i in range(num_jobs):
tasks.put(Task(i, 100)) # Similar jobs would be reiterated before poison pill.
# We start to .get() the results in a different loop-
for _ in range(num_jobs): # -so the above loop enqueues all jobs without-
result = results.get() # -waiting for the previous .put() to .get() first.
# Add a poison pill for each consumer
for i in range(num_consumers): # We only do this when all computation is done.
tasks.put(None) # Here we break all loops of open Consumer enqueue-able process'.
This code is just an example. In other variations of this code: when implementing more iterations of tasks.put() and results.get(), there is need for a way to cause an enqueued Task(object) to return by an outside call before it fully calculates an answer and returns by itself. This would free up resources if you already got your answer from one of the other process' of that single split number group. The __call__ descriptor needs to be present for the Task(object) to work as a function to call tasks.put(Task(i, 100)). I have spent the last 2 weeks trying to figure an efficient way of doing this. Do I need to take a whole different approach? Don't misunderstand my dilemma, I am working with code that works, but not as efficient as I would like on Microsoft Windslows. Any help would be greatly appreciated.
Doesn't a Task(object) exist in the same process as the Consumer() process that enqueued it? If so, couldn't I tell all process' of Class Consumer() Run() to stop their currently running Task(object) without closing their while loops (with the poison pill) so that they can instantly accept another Task() without the need to close and reopen their process again? It really does add up and waste time when you are opening and closing thousands of process' for an iterated calculation. I have tried using Events() Managers() other Queues(). There doesn't seem to be an efficient way to externally intervene on the Task(object) to instantly return to its parent Consumer() so that it doesn't keep wasting resources computing if one of the other Consumers() returned an answer that makes computation of other Consumer() Tasks irrelevant since they are all working as a unified computation of a single number split into groups.