From da650e9cfdb8f67f7e14592930bc6e8a904f0d85 Mon Sep 17 00:00:00 2001 From: Mazz Mosley Date: Tue, 21 Jul 2015 11:56:59 +0100 Subject: [PATCH] Refactor parallel execute Refactored parallel execute and execute create into a single function parallel_execute that can now handle both cases. This helps untangle it from being so tightly coupled to the container. Updated all the relevant operations to use the refactored function. Signed-off-by: Mazz Mosley --- compose/project.py | 21 ++++++++-- compose/service.py | 39 +++++++++++------- compose/utils.py | 99 ++++++++++++++-------------------------------- 3 files changed, 72 insertions(+), 87 deletions(-) diff --git a/compose/project.py b/compose/project.py index 541ff3ff..c5028492 100644 --- a/compose/project.py +++ b/compose/project.py @@ -198,15 +198,30 @@ class Project(object): service.start(**options) def stop(self, service_names=None, **options): - parallel_execute("stop", self.containers(service_names), "Stopping", "Stopped", **options) + parallel_execute( + objects=self.containers(service_names), + obj_callable=lambda c: c.stop(**options), + msg_index=lambda c: c.name, + msg="Stopping" + ) def kill(self, service_names=None, **options): - parallel_execute("kill", self.containers(service_names), "Killing", "Killed", **options) + parallel_execute( + objects=self.containers(service_names), + obj_callable=lambda c: c.kill(**options), + msg_index=lambda c: c.name, + msg="Killing" + ) def remove_stopped(self, service_names=None, **options): all_containers = self.containers(service_names, stopped=True) stopped_containers = [c for c in all_containers if not c.is_running] - parallel_execute("remove", stopped_containers, "Removing", "Removed", **options) + parallel_execute( + objects=stopped_containers, + obj_callable=lambda c: c.remove(**options), + msg_index=lambda c: c.name, + msg="Removing" + ) def restart(self, service_names=None, **options): for service in self.get_services(service_names): diff --git a/compose/service.py b/compose/service.py index cda68b7f..abb7536c 100644 --- a/compose/service.py +++ b/compose/service.py @@ -24,7 +24,7 @@ from .const import ( from .container import Container from .legacy import check_for_legacy_containers from .progress_stream import stream_output, StreamOutputError -from .utils import json_hash, parallel_create_execute, parallel_execute +from .utils import json_hash, parallel_execute log = logging.getLogger(__name__) @@ -162,13 +162,11 @@ class Service(object): 'for this service are created on a single host, the port will clash.' % self.name) - def create_and_start(number): - container = self.create_container(number=number, quiet=True) + def create_and_start(service, number): + container = service.create_container(number=number, quiet=True) container.start() return container - msgs = {'doing': 'Creating', 'done': 'Started'} - running_containers = self.containers(stopped=False) num_running = len(running_containers) @@ -185,20 +183,31 @@ class Service(object): next_number, next_number + num_to_create ) ] - parallel_create_execute(create_and_start, container_numbers, msgs) + + parallel_execute( + objects=container_numbers, + obj_callable=lambda n: create_and_start(service=self, number=n), + msg_index=lambda n: n, + msg="Creating and starting" + ) if desired_num < num_running: + num_to_stop = num_running - desired_num sorted_running_containers = sorted(running_containers, key=attrgetter('number')) + containers_to_stop = sorted_running_containers[-num_to_stop:] - if desired_num < num_running: - # count number of running containers. - num_to_stop = num_running - desired_num - - containers_to_stop = sorted_running_containers[-num_to_stop:] - # TODO: refactor these out? - parallel_execute("stop", containers_to_stop, "Stopping", "Stopped") - parallel_execute("remove", containers_to_stop, "Removing", "Removed") - # self.remove_stopped() + parallel_execute( + objects=containers_to_stop, + obj_callable=lambda c: c.stop(timeout=timeout), + msg_index=lambda c: c.name, + msg="Stopping" + ) + parallel_execute( + objects=containers_to_stop, + obj_callable=lambda c: c.remove(), + msg_index=lambda c: c.name, + msg="Removing" + ) def remove_stopped(self, **options): for c in self.containers(stopped=True): diff --git a/compose/utils.py b/compose/utils.py index af6aa902..ff3096fd 100644 --- a/compose/utils.py +++ b/compose/utils.py @@ -12,114 +12,75 @@ from threading import Thread log = logging.getLogger(__name__) -def parallel_create_execute(create_function, container_numbers, msgs={}, **options): +def parallel_execute(objects, obj_callable, msg_index, msg): """ - Parallel container creation by calling the create_function for each new container - number passed in. + For a given list of objects, call the callable passing in the first + object we give it. """ stream = codecs.getwriter('utf-8')(sys.stdout) lines = [] errors = {} - for number in container_numbers: - write_out_msg(stream, lines, number, msgs['doing']) + for obj in objects: + write_out_msg(stream, lines, msg_index(obj), msg) q = Queue() - def inner_call_function(create_function, number): + def inner_execute_function(an_callable, parameter, msg_index): try: - container = create_function(number) + result = an_callable(parameter) except APIError as e: - errors[number] = e.explanation - q.put(container) + errors[msg_index] = e.explanation + result = "error" + q.put((msg_index, result)) - for number in container_numbers: + for an_object in objects: t = Thread( - target=inner_call_function, - args=(create_function, number), - kwargs=options, + target=inner_execute_function, + args=(obj_callable, an_object, msg_index(an_object)), ) t.daemon = True t.start() done = 0 - total_to_create = len(container_numbers) - while done < total_to_create: + total_to_execute = len(objects) + + while done < total_to_execute: try: - container = q.get(timeout=1) - write_out_msg(stream, lines, container.name, msgs['done']) + msg_index, result = q.get(timeout=1) + if result == 'error': + write_out_msg(stream, lines, msg_index, msg, status='error') + else: + write_out_msg(stream, lines, msg_index, msg) done += 1 except Empty: pass if errors: - for number in errors: - stream.write("ERROR: for {} {} \n".format(number, errors[number])) + for error in errors: + stream.write("ERROR: for {} {} \n".format(error, errors[error])) -def parallel_execute(command, containers, doing_msg, done_msg, **options): - """ - Execute a given command upon a list of containers in parallel. - """ - stream = codecs.getwriter('utf-8')(sys.stdout) - lines = [] - errors = {} - - for container in containers: - write_out_msg(stream, lines, container.name, doing_msg) - - q = Queue() - - def container_command_execute(container, command, **options): - try: - getattr(container, command)(**options) - except APIError as e: - errors[container.name] = e.explanation - q.put(container) - - for container in containers: - t = Thread( - target=container_command_execute, - args=(container, command), - kwargs=options, - ) - t.daemon = True - t.start() - - done = 0 - - while done < len(containers): - try: - container = q.get(timeout=1) - write_out_msg(stream, lines, container.name, done_msg) - done += 1 - except Empty: - pass - - if errors: - for container in errors: - stream.write("ERROR: for {} {} \n".format(container, errors[container])) - - -def write_out_msg(stream, lines, container_name, msg): +def write_out_msg(stream, lines, msg_index, msg, status="done"): """ Using special ANSI code characters we can write out the msg over the top of a previous status message, if it exists. """ - if container_name in lines: - position = lines.index(container_name) + obj_index = msg_index + if msg_index in lines: + position = lines.index(obj_index) diff = len(lines) - position # move up stream.write("%c[%dA" % (27, diff)) # erase stream.write("%c[2K\r" % 27) - stream.write("{}: {} \n".format(container_name, msg)) + stream.write("{} {}... {}\n".format(msg, obj_index, status)) # move back down stream.write("%c[%dB" % (27, diff)) else: diff = 0 - lines.append(container_name) - stream.write("{}: {}... \r\n".format(container_name, msg)) + lines.append(obj_index) + stream.write("{} {}... \r\n".format(msg, obj_index)) stream.flush()