From 5c29ded6acaf09943e395472b6fb2ee095546f43 Mon Sep 17 00:00:00 2001 From: Mazz Mosley Date: Fri, 17 Jul 2015 17:40:14 +0100 Subject: [PATCH 1/3] Parallelise scale Signed-off-by: Mazz Mosley --- compose/service.py | 61 ++++++++++++++++++++++++++-------------------- compose/utils.py | 45 ++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 27 deletions(-) diff --git a/compose/service.py b/compose/service.py index 006696c2..cda68b7f 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 +from .utils import json_hash, parallel_create_execute, parallel_execute log = logging.getLogger(__name__) @@ -162,36 +162,43 @@ class Service(object): 'for this service are created on a single host, the port will clash.' % self.name) - # Create enough containers - containers = self.containers(stopped=True) - while len(containers) < desired_num: - containers.append(self.create_container()) + def create_and_start(number): + container = self.create_container(number=number, quiet=True) + container.start() + return container - running_containers = [] - stopped_containers = [] - for c in containers: - if c.is_running: - running_containers.append(c) - else: - stopped_containers.append(c) - running_containers.sort(key=lambda c: c.number) - stopped_containers.sort(key=lambda c: c.number) + msgs = {'doing': 'Creating', 'done': 'Started'} - # Stop containers - while len(running_containers) > desired_num: - c = running_containers.pop() - log.info("Stopping %s..." % c.name) - c.stop(timeout=timeout) - stopped_containers.append(c) + running_containers = self.containers(stopped=False) + num_running = len(running_containers) - # Start containers - while len(running_containers) < desired_num: - c = stopped_containers.pop(0) - log.info("Starting %s..." % c.name) - self.start_container(c) - running_containers.append(c) + if desired_num == num_running: + # do nothing as we already have the desired number + log.info('Desired container number already achieved') + return - self.remove_stopped() + if desired_num > num_running: + num_to_create = desired_num - num_running + next_number = self._next_container_number() + container_numbers = [ + number for number in range( + next_number, next_number + num_to_create + ) + ] + parallel_create_execute(create_and_start, container_numbers, msgs) + + if desired_num < num_running: + sorted_running_containers = sorted(running_containers, key=attrgetter('number')) + + 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() def remove_stopped(self, **options): for c in self.containers(stopped=True): diff --git a/compose/utils.py b/compose/utils.py index b6ee63d0..af6aa902 100644 --- a/compose/utils.py +++ b/compose/utils.py @@ -12,6 +12,51 @@ from threading import Thread log = logging.getLogger(__name__) +def parallel_create_execute(create_function, container_numbers, msgs={}, **options): + """ + Parallel container creation by calling the create_function for each new container + number passed in. + """ + stream = codecs.getwriter('utf-8')(sys.stdout) + lines = [] + errors = {} + + for number in container_numbers: + write_out_msg(stream, lines, number, msgs['doing']) + + q = Queue() + + def inner_call_function(create_function, number): + try: + container = create_function(number) + except APIError as e: + errors[number] = e.explanation + q.put(container) + + for number in container_numbers: + t = Thread( + target=inner_call_function, + args=(create_function, number), + kwargs=options, + ) + t.daemon = True + t.start() + + done = 0 + total_to_create = len(container_numbers) + while done < total_to_create: + try: + container = q.get(timeout=1) + write_out_msg(stream, lines, container.name, msgs['done']) + done += 1 + except Empty: + pass + + if errors: + for number in errors: + stream.write("ERROR: for {} {} \n".format(number, errors[number])) + + def parallel_execute(command, containers, doing_msg, done_msg, **options): """ Execute a given command upon a list of containers in parallel. From da650e9cfdb8f67f7e14592930bc6e8a904f0d85 Mon Sep 17 00:00:00 2001 From: Mazz Mosley Date: Tue, 21 Jul 2015 11:56:59 +0100 Subject: [PATCH 2/3] 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() From e1c1a4c0aa670f28ef335c31d1d0ace9824a047c Mon Sep 17 00:00:00 2001 From: Mazz Mosley Date: Tue, 21 Jul 2015 14:59:32 +0100 Subject: [PATCH 3/3] Scale restarts stopped containers This is existing behaviour and should be kept. Signed-off-by: Mazz Mosley --- compose/service.py | 44 ++++++++++++++++++++++++++++++++++---------- 1 file changed, 34 insertions(+), 10 deletions(-) diff --git a/compose/service.py b/compose/service.py index abb7536c..8908d4c7 100644 --- a/compose/service.py +++ b/compose/service.py @@ -176,6 +176,30 @@ class Service(object): return if desired_num > num_running: + # we need to start/create until we have desired_num + all_containers = self.containers(stopped=True) + + if num_running != len(all_containers): + # we have some stopped containers, let's start them up again + stopped_containers = sorted([c for c in all_containers if not c.is_running], key=attrgetter('number')) + + num_stopped = len(stopped_containers) + + if num_stopped + num_running > desired_num: + num_to_start = desired_num - num_running + containers_to_start = stopped_containers[:num_to_start] + else: + containers_to_start = stopped_containers + + parallel_execute( + objects=containers_to_start, + obj_callable=lambda c: c.start(), + msg_index=lambda c: c.name, + msg="Starting" + ) + + num_running += len(containers_to_start) + num_to_create = desired_num - num_running next_number = self._next_container_number() container_numbers = [ @@ -202,18 +226,18 @@ class Service(object): 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" - ) + + self.remove_stopped() def remove_stopped(self, **options): - for c in self.containers(stopped=True): - if not c.is_running: - log.info("Removing %s..." % c.name) - c.remove(**options) + containers = [c for c in self.containers(stopped=True) if not c.is_running] + + parallel_execute( + objects=containers, + obj_callable=lambda c: c.remove(**options), + msg_index=lambda c: c.name, + msg="Removing" + ) def create_container(self, one_off=False,