up & down in parallel

Author: reliveyy

images/utils/launcher/node/__init__.py

@@ -13,7 +13,7 @@
 from docker.errors import NotFound
 from docker.types import IPAMConfig, IPAMPool
 
-from launcher.utils import ArgumentParser, yes_or_no
+from launcher.utils import ArgumentParser, yes_or_no, parallel
 from .DockerTemplate import DockerTemplate
 from .arby import Arby
 from .base import Node, ContainerNotFound, NoWaiting
@@ -69,7 +69,8 @@ def execute(self, *args) -> None:
         args = self.parser.parse_args(args)
         name = args.service
         service = self.get_service(name)
-        for line in service.logs(tail=args.tail, since=args.since, until=args.until, follow=args.follow, timestamps=args.timestamps):
+        for line in service.logs(tail=args.tail, since=args.since, until=args.until, follow=args.follow,
+                                 timestamps=args.timestamps):
             print(line)
 
 
@@ -151,6 +152,10 @@ class ServiceNotFound(Exception):
     pass
 
 
+class NetworkNotFound(Exception):
+    pass
+
+
 class NodeManager:
     config: Config
     client: DockerClient
@@ -165,7 +170,7 @@ def __init__(self, config: Config):
         ctx = Context(self.config, self.client, self.image_manager, self)
 
         self.nodes = {name: globals()[name.capitalize()](name, ctx) for name in self.config.nodes}
-        self.docker_network = self.create_docker_network()
+        # self.docker_network = self.create_docker_network()
 
         self.cmd_logs = LogsCommand(self.get_service)
         self.cmd_start = StartCommand(self.get_service)
@@ -197,17 +202,23 @@ def get_network_ipam_pool(self):
         elif self.network == "mainnet":
             return IPAMPool(subnet='10.0.3.0/24', gateway='10.0.3.1')
 
-    def create_docker_network(self):
-        name = self.network_name
-        try:
-            network = self.client.networks.get(name)
-            return network
-        except NotFound:
-            pass
+    def _create_docker_network(self) -> None:
         ipam_pool = self.get_network_ipam_pool()
         ipam_config = IPAMConfig(pool_configs=[ipam_pool])
-        network = self.client.networks.create(name, driver="bridge", ipam=ipam_config)
-        return network
+        network = self.client.networks.create(self.network_name, driver="bridge", ipam=ipam_config)
+        logger.info("Created network: %r", network)
+
+    def _remove_docker_network(self) -> None:
+        network = self.docker_network
+        network.remove()
+        logger.info("Removed network: %r", network)
+
+    @property
+    def docker_network(self):
+        try:
+            return self.client.networks.get(self.network_name)
+        except docker.errors.NotFound as e:
+            raise NetworkNotFound(self.network_name) from e
 
     def get_service(self, name: str) -> Node:
         try:
@@ -216,46 +227,46 @@ def get_service(self, name: str) -> Node:
             raise ServiceNotFound(name) from e
 
     @property
-    def valid_nodes(self):
+    def valid_nodes(self) -> Dict[str, Node]:
         return {name: node for name, node in self.nodes.items() if node.mode == "native" and not node.disabled}
 
-    @property
-    def enabled_nodes(self):
-        return {name: node for name, node in self.nodes.items() if not node.disabled}
-
     def up(self):
-        self.docker_network = self.create_docker_network()
+        nodes = self.valid_nodes
 
-        nodes = self.valid_nodes.values()
+        logger.info("Up services: %s", ", ".join(nodes))
 
-        executor = self.config.executor
-        futs = {executor.submit(node.start): node for node in nodes}
-        done, not_done = wait(futs, 60)
-        if len(not_done) > 0:
-            # TODO retry failed tasks
-            raise RuntimeError("Failed to up")
+        try:
+            _ = self.docker_network
+        except NetworkNotFound:
+            self._create_docker_network()
+
+        def linehead(node):
+            return "starting %s" % node.container_name
+
+        def start(node, stop):
+            node.start()
+
+        nodes = [node for node in nodes.values() if not node.is_running]
+
+        parallel(self.config.executor, nodes, linehead, start)
 
     def down(self):
         nodes = self.valid_nodes
 
-        for name, container in nodes.items():
-            print(f"Stopping {name}...")
-            container.stop()
-        for name, container in nodes.items():
-            print(f"Removing {name}...")
-            container.remove()
+        logger.info("Down services: %s", ", ".join(nodes))
+
+        running_nodes = [node for node in nodes.values() if node.is_running]
+
+        parallel(self.config.executor, running_nodes,
+                 lambda node: "stopping %s" % node.container_name,
+                 lambda node, stop: node.stop())
+
+        parallel(self.config.executor, list(nodes.values()),
+                 lambda node: "removing %s" % node.container_name,
+                 lambda node, stop: node.remove())
+
         print(f"Removing network {self.network_name}")
-        self.docker_network.remove()
-
-    def _display_container_status_text(self, status):
-        if status == "missing":
-            return "create"
-        elif status == "outdated":
-            return "recreate"
-        elif status == "external_with_container":
-            return "remove"
-        elif status == "disabled_with_container":
-            return "remove"
+        self._remove_docker_network()
 
     def check_for_updates(self) -> Dict[Node, str]:
         logger.info("Checking for container updates")
@@ -305,24 +316,35 @@ def update(self) -> None:
         if b1 and b2:
             print("All up-to-date.")
 
-        for container, action in containers.items():
+        def linehead(node):
+            action = containers[node]
             if action == "CREATE":
-                print("Creating %s..." % container.container_name)
-                container.create()
+                return "creating %s" % node.container_name
             elif action == "RECREATE":
-                if container.is_running:
-                    print("Stopping %s..." % container.container_name)
-                    container.stop()
-                print("Removing %s..." % container.container_name)
-                container.remove()
-                print("Creating %s..." % container.container_name)
-                container.create()
+                return "recreating %s" % node.container_name
             elif action == "REMOVE":
-                if container.is_running:
-                    print("Stopping %s..." % container.container_name)
-                    container.stop()
-                print("Removing %s..." % container.container_name)
-                container.remove()
+                return "removing %s" % node.container_name
+
+        def update(node, stop):
+            action = containers[node]
+            if action == "CREATE":
+                node.create()
+            elif action == "RECREATE":
+                if node.is_running:
+                    node.stop()
+                node.remove()
+                node.create()
+            elif action == "REMOVE":
+                if node.is_running:
+                    node.stop()
+                node.remove()
+
+        items = []
+        for container, action in containers.items():
+            if action != "NONE":
+                items.append(container)
+
+        parallel(self.config.executor, items, linehead, update)
 
     def _get_status_nodes(self):
         optional_nodes = ["arby", "boltz", "webui", "proxy"]

images/utils/launcher/node/xud.py

@@ -34,15 +34,19 @@ def getinfo(self):
             raise XudApiError(e.output)
 
 
+class InvalidPassword(Exception):
+    pass
+
+
 class PasswordNotMatch(Exception):
     pass
 
 
-class MnemonicNot24Words(Exception):
+class WrongPassword(Exception):
     pass
 
 
-class InvalidPassword(Exception):
+class MnemonicNot24Words(Exception):
     pass
 
 
@@ -168,6 +172,8 @@ def extract_exception(self, cmd: str, output: str):
         elif cmd.startswith("unlock"):
             if "xud was unlocked succesfully" in output:
                 return
+            elif "password is incorrect" in output:
+                raise WrongPassword
             else:
                 print("^C")
                 raise KeyboardInterrupt()

images/utils/launcher/utils.py

@@ -1,6 +1,11 @@
 import argparse
 import logging
 import os
+import random
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from typing import List, Callable, TypeVar
 
 logger = logging.getLogger(__name__)
 
@@ -79,3 +84,82 @@ def get_percentage(current, total):
     else:
         p = 0
     return "%.2f%% (%d/%d)" % (p, current, total)
+
+
+def color(text: str) -> str:
+    if text == "done":
+        return "\033[32mdone\033[0m"
+    elif text == "error":
+        return "\033[31mfailed\033[0m"
+    else:
+        return text
+
+
+T = TypeVar('T')
+
+
+def parallel(
+        executor: ThreadPoolExecutor,
+        items: List[T],
+        linehead: Callable[[T], str],
+        run: Callable[[T, threading.Event], None]
+):
+    result = {item: None for item in items}
+    stop = threading.Event()
+
+    def animate():
+        nonlocal result
+        nonlocal stop
+
+        lines = []
+        width = 0
+        for item in items:
+            line = linehead(item)
+            line = line.capitalize()
+            if len(line) > width:
+                width = len(line)
+            lines.append(line)
+        fmt = "%-{}s ...".format(width)
+        lines = [fmt % line for line in lines]
+        print("\n".join(lines))
+
+        i = 0
+        while not stop.is_set():
+            print("\033[%dA" % len(items), end="", flush=True)
+            finish = 0
+            for item in items:
+                r = result[item]
+                if r:
+                    suffix = "... " + color(r)
+                    suffix_len = 4 + len(r)
+                    finish += 1
+                else:
+                    suffix = "%-3s" % ("." * abs(3 - i % 6))
+                    suffix_len = 3
+                print("\033[%dC" % (width + 1), end="", flush=True)
+                print(suffix, end="", flush=True)
+                print("\033[%dD\033[1B" % (width + 1 + suffix_len), end="", flush=True)
+            print("\033[K", end="", flush=True)
+            if finish == len(items):
+                break
+            stop.wait(0.5)
+            i += 1
+
+    def wrapper(item):
+        nonlocal result
+        nonlocal stop
+        try:
+            run(item, stop)
+            # time.sleep(random.randint(3, 10))
+            result[item] = "done"
+        except:
+            logger.exception("Failed to %s", linehead(item))
+            result[item] = "error"
+
+    f = executor.submit(animate)
+    try:
+        for item in items:
+            executor.submit(wrapper, item)
+        f.result()
+    finally:
+        stop.set()