diff --git a/plmap/plmapp.py b/plmap/plmapp.py index ac9fbe0..3714b7b 100644 --- a/plmap/plmapp.py +++ b/plmap/plmapp.py @@ -31,9 +31,11 @@ # The output will look like this : # [8, 17, 23, 31] - +from __future__ import division from multiprocessing import Queue from multiprocessing import Process +from multiprocessing import Value +from multiprocessing import Lock from pprint import pprint import datetime import sys @@ -41,10 +43,35 @@ import time import signal import types +from progress_bar import generate_loading_string + + + +PROGRESS_BAR_POLL_TIME = 1 + +def display_progress_bar(progress_details): + """ 50%[----------> ]5/10 + """ + completed_tasks = progress_details['completed_tasks'].value + total_tasks = progress_details['total_tasks'] + + while completed_tasks != total_tasks: + time.sleep(PROGRESS_BAR_POLL_TIME) + completed_tasks = progress_details['completed_tasks'].value + print generate_loading_string(completed_tasks, total_tasks) + sys.stdout.write("\033[F") + + # Display 100% completion + completed_tasks = progress_details['completed_tasks'].value + print generate_loading_string(completed_tasks, total_tasks) + sys.exit(0) + + + class FunctionTimeoutException(Exception): """ This exception will be raised when a function takes too long to complete """ @@ -58,99 +85,106 @@ def receive_signal(signum, stack): -def queue_exec(input_queue, output_queue, log_queue, individual_timeout=20.0, default_output=None): +def queue_exec(input_queue, output_queue, p_id, progress_details, default_output, individual_timeout=300.0): """ This function will be executed by all the child Processes spawned by the Parent process """ + if p_id != -1: # Not a progress bar process + signal.signal(signal.SIGALRM, receive_signal) - signal.signal(signal.SIGALRM, receive_signal) - - while True: - try: - id, function, args, kwargs = input_queue.get_nowait() - except : - break - - signal.alarm(int(individual_timeout)) - - try: - output = function(*args, **kwargs) - output_queue.put({ 'id' : id, 'output' : output, 'error_code' : '0', 'err_desc' : None }) - except FunctionTimeoutException, details: - output_queue.put({ 'id' : id, 'output' : default_output, 'error_code' : '1', 'err_desc' : str(details) }) - log_queue.put("state:MAJOR, id:{}, output:{}, err:{}, desc:{}".format(id, default_output, '1', str(details))) - except Exception, details: - log_queue.put("state:MAJOR, id:{}, output:{}, err:{}, desc:{}".format(id, default_output, '1', str(details))) - output_queue.put({ 'id' : id, 'output' : default_output, 'error_code' : '1', 'err_desc' : str(details) }) - pass - # Log the error here - signal.alarm(0) + while True: + try: + loc, function, args, kwargs = input_queue.get_nowait() + except: + break + + signal.alarm(int(individual_timeout)) - sys.exit(0) + try: + output = function(*args, **kwargs) + output_queue.put((loc, (0, None), output)) + except FunctionTimeoutException, details: + output_queue.put((loc, (1, details), default_output)) + except Exception, details: + output_queue.put((loc, (2, details), default_output)) + finally: + signal.alarm(0) + with progress_details['process_lock']: + progress_details['completed_tasks'].value += 1 + sys.exit(0) + else: + display_progress_bar(progress_details) -def plmapp(func, args=[], kwargs=[], processes=10, default_output=None, individual_timeout=20): +def plmapp(func, args=[], kwargs=[], processes=10, progress_bar=False, default_output=None, individual_timeout=20): """ """ - input_queue = Queue() - output_queue = Queue() - log_queue = Queue() - bigger_array = args if len(args) > len(kwargs) else kwargs big_len = len(bigger_array) ag = lambda array, index, default : default if index >= len(array) else array[index] + input_queue = Queue() + output_queue = Queue() + error_queue = Queue() + + + progress_details = {'completed_tasks' : Value('i', 0), + 'total_tasks' : big_len, + 'process_lock' : Lock()} + # Load the input queue with tasks - for _ in xrange(big_len): - task = ( _ , func, ag(args, _, ()), ag(kwargs, _, {}) ) + for i in xrange(big_len): + task = (i, func, ag(args, i, ()), ag(kwargs, i, {})) input_queue.put(task) # Create the processes processes = [] - for _ in xrange(min(processes, 10)): - p = Process(target=queue_exec, args=(input_queue, output_queue, log_queue, individual_timeout, default_output)) + end = min(processes, big_len) + start = -1 if progress_bar else 0 + + + for i in xrange(start, end): + p = Process(target=queue_exec, args=(input_queue, output_queue, i, progress_details, default_output, individual_timeout)) processes.append(p) p.start() # Wait for the processes to complete - for _ in processes: - _.join() - - # Get all the output dictionaries from the output queue - output_dicts = [] - for output_dict in xrange(big_len): - output_dicts.append( output_queue.get_nowait() ) + for process in processes: + process.join() - # Sort the output based on the id - output_sorted_dicts = sorted(output_dicts, key=lambda x : x['id']) + output_errors = [ None for _ in xrange(big_len)] + output_values = [ None for _ in xrange(big_len)] - error_array = [ (_['error_code'], _['err_desc']) for _ in output_sorted_dicts ] - output_array = [ _['output'] for _ in output_sorted_dicts ] - - return (error_array, output_array) + while True: + try: + loc, error_value = error_queue.get_nowait() + output_errors[loc] = error_value + except: + break + return output_errors, output_values if __name__ == "__main__": - def do_it(a, b=8, c=10): - time.sleep(1) - return a + b + c + def do_it(a): + time.sleep(a) + return a # args - args = [ (_,) for _ in xrange(10) ] + args = [[i] for i in range(10)] kwargs = [] func = do_it threads = 8 # user input ends here print "Start time : ", datetime.datetime.now() - error, output = plmapp(func, args, kwargs, threads, default_output=None) + error, output = plmapp(func, args, kwargs, threads, True) print "End time : " , datetime.datetime.now() print "Errors : " pprint(error) diff --git a/plmap/plmapt.py b/plmap/plmapt.py index 4e1140d..849d07a 100644 --- a/plmap/plmapt.py +++ b/plmap/plmapt.py @@ -33,48 +33,24 @@ def add ( a, b ) : from __future__ import division -from threading import Thread -from Queue import Queue -import sys -import time -from pprint import pprint +import sys +from Queue import Queue +from threading import Lock, Thread +from progress_bar import generate_loading_string PROGRESS_BAR_POLL_TIME = 1 -PROGRESS_BAR_LENGTH = 20 - class Worker(Thread): """ Worker thread is the thread that actually executes the function with the given arguments """ - def __init__(self, thread_id, input_queue, default_output=None, progress_details={}, progress_bar_thread=False): - super(Worker, self).__init__(name='%s' %(thread_id)) + def __init__(self, thread_id, input_queue, progress_details, default_output): + super(Worker, self).__init__(name='%s' % (thread_id)) self.input_queue = input_queue self.default_output = default_output self.progress_details = progress_details - self.progress_bar_thread = progress_bar_thread - - - def generate_loading_string(self, completed_tasks, total_tasks): - """ % [< -- based on percentage completion>] Completed/Total - """ - try: - fraction_completed = ( completed_tasks / total_tasks) - except: - fraction_completed = 1 # To avoid division by Zero - - percentage_complete = fraction_completed * 100 - - dashes = int(PROGRESS_BAR_LENGTH * fraction_completed) - blanks = PROGRESS_BAR_LENGTH - dashes - - bar = "[" + "-" * dashes + ">" + " " * blanks + "]" - fraction_display = "%s/%s" %(completed_tasks, total_tasks) - - loading_string = "%s%% %s %s" %(percentage_complete, bar, fraction_display) - - return loading_string + self.progress_bar_thread = (thread_id == -1) def display_progress_bar(self): @@ -82,27 +58,22 @@ def display_progress_bar(self): """ completed_tasks = self.progress_details['completed_tasks'] total_tasks = self.progress_details['total_tasks'] - while completed_tasks != total_tasks: - time.sleep(PROGRESS_BAR_POLL_TIME) - completed_tasks = self.progress_details['completed_tasks'] total_tasks = self.progress_details['total_tasks'] - - print self.generate_loading_string(completed_tasks, total_tasks) + print generate_loading_string(completed_tasks, total_tasks) sys.stdout.write("\033[F") - # Display 100% completion completed_tasks = self.progress_details['completed_tasks'] total_tasks = self.progress_details['total_tasks'] - print self.generate_loading_string(completed_tasks, total_tasks) - + print generate_loading_string(completed_tasks, total_tasks) - # run method will be called when the thread is started using thread.start() , Hence overriding run method + # run method will be called when the thread is started using + # thread.start() , Hence overriding run method def run(self): - """ Pick up the task from the queue and execute it - """ + """ Pick up the task from the queue and execute it + """ if not self.progress_bar_thread: while True: try: @@ -110,35 +81,38 @@ def run(self): except: break - func, args, kwargs, error_loc, output_loc = task - + func, loc, args, kwargs, errors, outputs = task + try: output = func(*args, **kwargs) - error_loc.append(0) # Indicates Success - output_loc.append(output) + errors[loc], outputs[loc] = 0, output except Exception, details: - error_loc.append(1) # Indicates Failure - output_loc.append(self.default_output) + errors[loc], outputs[loc] = details, self.default_output finally: - self.progress_details['completed_tasks'] += 1 + with self.progress_details['thread_lock']: + self.progress_details['completed_tasks'] += 1 else: self.display_progress_bar() -def plmapt(func, args=[], kwargs=[], threads=10, default_output=None, sort_output=True, progress_bar=False): + + +def plmapt(func, args=[], kwargs=[], threads=10, default_output=None, progress_bar=False): """ Create the workers and get them to work. Returns the error and output array - Sorting the output according to the input order is optional () Displaying the Progress bar is optional """ - bigger_array = args if len(args) > len(kwargs) else kwargs + bigger_array = args if len(args) > len(kwargs) else kwargs big_len = len(bigger_array) - # Get the element of an array given the index, return default if index out of range - ag = lambda array, index, default : default if index >= len(array) else array[index] + # Get the element of an array given the index, return default if index out + # of range + ag = lambda array, index, default: default if index >= len(array) else array[ + index] - progress_details = {'total_tasks' : big_len , 'completed_tasks' : 0} + progress_details = {'total_tasks': big_len, + 'completed_tasks': 0, 'thread_lock': Lock()} input_queue = Queue() output_errors = [[] for _ in xrange(big_len)] @@ -146,24 +120,21 @@ def plmapt(func, args=[], kwargs=[], threads=10, default_output=None, sort_outpu # Load the input_queue ( big_len = the number of tasks ) for i in xrange(big_len): - task = [func, ag(args, i, ()), ag(kwargs, i, {}), output_errors[i], output_values[i]] + task = [func, i, ag(args, i, ()), ag(kwargs, i, {}), + output_errors, output_values] input_queue.put(task) # Create workers and start them list_of_workers = [] - for i in xrange(min(threads, big_len)): - worker = Worker(i, input_queue, default_output, progress_details) + end = min(threads, big_len) + start = -1 if progress_bar else 0 + + for i in xrange(start, end): + worker = Worker(i, input_queue, progress_details, default_output) worker.setDaemon(True) list_of_workers.append(worker) worker.start() - # If the progress bar needs to be displayed - if progress_bar: - progress_worker = Worker("-1", input_queue, default_output, progress_details, True) - progress_worker.setDaemon(True) - list_of_workers.append(progress_worker) - progress_worker.start() - # Wait till the threads complete for _ in list_of_workers: _.join() @@ -171,36 +142,39 @@ def plmapt(func, args=[], kwargs=[], threads=10, default_output=None, sort_outpu return output_errors, output_values - if __name__ == "__main__": + import time + from pprint import pprint - def add ( a, b ) : - time.sleep(a) + def add(a, b): + time.sleep(1) if a == 3: 1 / 0 - return (a, b,a + b) + return (a, b, a + b) inputs = [ - [3, 5], - [8, 9] , - [4, 12], - [1, 16], - [1, 16], - [1, 16], - [1, 16], - [1, 16], - [3, 16], - [1, 16], + [3, 5], + [8, 9], + [4, 12], + [1, 16], + [1, 16], + [1, 16], + [1, 16], + [1, 16], + [3, 16], + [1, 16], ] - error, output = plmapt(add, inputs , [], 4, default_output=None, progress_bar=True) - print output - error, output = plmapt(add, inputs , [], 8, default_output=None, sort_output=False, progress_bar=True) - print output - error, output = plmapt(add, [] , [], 4, default_output=None, progress_bar=True) - print output - error, output = plmapt(add, inputs , [], 8, default_output=None, sort_output=False, progress_bar=False) - print output - print error - - + error, output = plmapt( + add, inputs, [], 1, default_output=None, progress_bar=True) + pprint(output) + error, output = plmapt( + add, inputs, [], 2, default_output=None, progress_bar=True) + pprint(output) + error, output = plmapt( + add, [], [], 4, default_output=None, progress_bar=True) + pprint(output) + error, output = plmapt( + add, inputs, [], 8, default_output=None, progress_bar=False) + pprint(output) + pprint(error) diff --git a/plmap/progress_bar.py b/plmap/progress_bar.py new file mode 100644 index 0000000..cef78a3 --- /dev/null +++ b/plmap/progress_bar.py @@ -0,0 +1,21 @@ +from __future__ import division + + + +PROGRESS_BAR_LENGTH = 40 + + +def generate_loading_string(completed_tasks, total_tasks): + """ % [< -- based on percentage completion>] Completed/Total + """ + try: + fraction_completed = (completed_tasks / total_tasks) + except: + fraction_completed = 1 # To avoid division by Zero + percentage_complete = fraction_completed * 100 + dashes = int(PROGRESS_BAR_LENGTH * fraction_completed) + blanks = PROGRESS_BAR_LENGTH - dashes + bar = "[" + "-" * dashes + ">" + " " * blanks + "]" + fraction_display = "%s/%s" % (completed_tasks, total_tasks) + loading_string = "%s%% %s %s" % (percentage_complete, bar, fraction_display) + return loading_string diff --git a/setup.py b/setup.py index 77ff741..85d0f38 100644 --- a/setup.py +++ b/setup.py @@ -3,7 +3,7 @@ from setuptools import setup setup(name='plmap', - version='1.1.0', + version='2.0.0', description='Parallel map using multi-threading and multi-processing', author='Lohith B R', packages=['plmap'],