Merge branch 'optimistic-sync' into main

Author: Alberto Sonnino

benchmark/aws/instance.py

@@ -163,7 +163,7 @@ def create_instances(self, instances):
                 client.run_instances(
                     ImageId=self._get_ami(client),
                     InstanceType=self.settings.instance_type,
-                    KeyName='aws',
+                    KeyName=self.settings.key_name,
                     MaxCount=instances,
                     MinCount=instances,
                     SecurityGroups=[self.SECURITY_GROUP_NAME],

benchmark/aws/remote.py

@@ -75,12 +75,11 @@ def install(self):
         hosts = self.manager.hosts(flat=True)
         try:
             g = Group(*hosts, user='ubuntu', connect_kwargs=self.connect)
-            output = g.run(' && '.join(cmd), hide=True)
-            self._check_stderr(output)
+            g.run(' && '.join(cmd), hide=True)
             Print.heading(f'Initialized testbed of {len(hosts)} nodes')
-        except GroupException as e:
-            error = FabricError(e)
-            raise BenchError('Failed to install repo on testbed', error)
+        except (GroupException, ExecutionError) as e:
+            e = FabricError(e) if isinstance(e, GroupException) else e
+            raise BenchError('Failed to install repo on testbed', e)
 
     def kill(self, hosts=[], delete_logs=False):
         assert isinstance(hosts, list)

benchmark/aws/settings.py

@@ -6,17 +6,22 @@ class SettingsError(Exception):
 
 
 class Settings:
-    def __init__(self, key_path, consensus_port, mempool_port, front_port, repo_name, 
+    def __init__(self, key_name, key_path, consensus_port, mempool_port, front_port, repo_name,
                  repo_url, branch, instance_type, aws_regions):
-        regions = aws_regions if isinstance(aws_regions, list) else [aws_regions]
-        inputs_str = [key_path, repo_name, repo_url, branch, instance_type] + regions
+        regions = aws_regions if isinstance(
+            aws_regions, list) else [aws_regions]
+        inputs_str = [
+            key_name, key_path, repo_name, repo_url, branch, instance_type
+        ]
+        inputs_str += regions
         inputs_int = [consensus_port, mempool_port, front_port]
         ok = all(isinstance(x, str) for x in inputs_str)
         ok &= all(isinstance(x, int) for x in inputs_int)
         ok &= len(regions) > 0
         if not ok:
             raise SettingsError('Invalid settings types')
 
+        self.key_name = key_name
         self.key_path = key_path
 
         self.consensus_port = consensus_port
@@ -37,7 +42,8 @@ def load(cls, filename):
                 data = load(f)
 
             return cls(
-                data['key_path'],
+                data['key']['name'],
+                data['key']['path'],
                 data['ports']['consensus'],
                 data['ports']['mempool'],
                 data['ports']['front'],
@@ -51,4 +57,4 @@ def load(cls, filename):
             raise SettingsError(str(e))
 
         except KeyError as e:
-            raise SettingsError(f'Malformed settings: missing key {e}')
+            raise SettingsError(f'Malformed settings: missing key {e}')

benchmark/benchmark/aggregator.py

@@ -14,12 +14,14 @@ def __init__(self, nodes, rate, tx_size):
         self.nodes = nodes
         self.rate = rate
         self.tx_size = tx_size
+        self.max_latency = 'any'
 
     def __str__(self):
         return (
             f' Committee size: {self.nodes} nodes\n'
-            f' Input rate: {self.rate} txs\n'
+            f' Input rate: {self.rate} tx/s\n'
             f' Transaction size: {self.tx_size} B\n'
+            f' Max latency: {self.max_latency} ms\n'
         )
 
     def __eq__(self, other):
@@ -104,7 +106,10 @@ def print(self):
                     '-----------------------------------------\n'
                 )
                 filename = PathMaker.agg_file(
-                    setup.nodes, setup.rate, setup.tx_size
+                    setup.nodes, 
+                    setup.rate, 
+                    setup.tx_size, 
+                    max_latency=setup.max_latency
                 )
                 with open(filename, 'w') as f:
                     f.write(string)
@@ -123,23 +128,26 @@ def _print_latency(self):
 
         return organized
 
-    def _print_tps(self, max_latency=4000):
+    def _print_tps(self, max_latencies=[2_000, 5_000]):
         records = deepcopy(self.records)
         organized = defaultdict(list)
-        for setup, result in records.items():
-            if result.mean_latency <= max_latency:
-                nodes = setup.nodes
-                setup.nodes = 'x'
-                setup.rate = 'any'
-
-                new_point = all(nodes != x[0] for x in organized[setup])
-                highest_tps = False
-                for w, r in organized[setup]:
-                    if result.mean_tps > r.mean_tps and nodes == w:
-                        organized[setup].remove((w, r))
-                        highest_tps = True
-                if new_point or highest_tps:
-                    organized[setup] += [(nodes, result)]
+        for max_latency in max_latencies:
+            for setup, result in records.items():
+                setup = deepcopy(setup)
+                if result.mean_latency <= max_latency:
+                    nodes = setup.nodes
+                    setup.nodes = 'x'
+                    setup.rate = 'any'
+                    setup.max_latency = max_latency
+
+                    new_point = all(nodes != x[0] for x in organized[setup])
+                    highest_tps = False
+                    for w, r in organized[setup]:
+                        if result.mean_tps > r.mean_tps and nodes == w:
+                            organized[setup].remove((w, r))
+                            highest_tps = True
+                    if new_point or highest_tps:
+                        organized[setup] += [(nodes, result)]
 
         [v.sort(key=lambda x: x[0]) for v in organized.values()]
         return organized

benchmark/benchmark/config.py

@@ -95,6 +95,7 @@ def __init__(self, json):
             inputs += [json['consensus']['max_payload_size']]
             inputs += [json['consensus']['min_block_delay']]
             inputs += [json['mempool']['queue_capacity']]
+            inputs += [json['consensus']['sync_retry_delay']]
             inputs += [json['mempool']['max_payload_size']]
             inputs += [json['mempool']['min_block_delay']]
         except KeyError as e:

benchmark/benchmark/logs.py

@@ -1,10 +1,9 @@
 from datetime import datetime
 from glob import glob
-from itertools import repeat
 from multiprocessing import Pool
 from os.path import join
 from re import findall, search
-from statistics import mean, median_grouped, stdev
+from statistics import mean
 
 from benchmark.utils import Print
 

benchmark/benchmark/plot.py

@@ -1,4 +1,4 @@
-from re import findall, search
+from re import findall, search, split
 import matplotlib.pyplot as plt
 from matplotlib.ticker import StrMethodFormatter
 from glob import glob
@@ -15,6 +15,7 @@ def __init__(self, filenames):
         if not filenames:
             raise PlotError('No data to plot')
 
+        filenames.sort(key=self._natural_keys)
         self.results = []
         try:
             for filename in filenames:
@@ -23,6 +24,10 @@ def __init__(self, filenames):
         except OSError as e:
             raise PlotError(f'Failed to load log files: {e}')
 
+    def _natural_keys(self, text):
+        def try_cast(text): return int(text) if text.isdigit() else text
+        return [try_cast(c) for c in split('(\d+)', text)]
+
     def _tps(self, data):
         values = findall(r' TPS: (\d+) \+/- (\d+)', data)
         values = [(int(x), int(y)) for x, y in values]
@@ -46,7 +51,7 @@ def _bps2tps(self, x):
         size = int(search(r'Transaction size: (\d+)', data).group(1))
         return x * 10**6 / size
 
-    def _plot(self, x_label, y_label, y_axis, z_axis, filename):
+    def _plot(self, x_label, y_label, y_axis, z_axis, type):
         plt.figure()
         for result in self.results:
             y_values, y_err = y_axis(result)
@@ -58,12 +63,14 @@ def _plot(self, x_label, y_label, y_axis, z_axis, filename):
                 x_values, y_values, yerr=y_err,  # uplims=True, lolims=True,
                 marker='o', label=z_axis(result), linestyle='dotted'
             )
+            # if type == 'latency':
+            #    plt.yscale('log')
 
         plt.xlim(xmin=0)
         plt.ylim(bottom=0)
         plt.xlabel(x_label)
         plt.ylabel(y_label[0])
-        plt.legend(loc='upper right')
+        plt.legend(loc='upper left')
         ax = plt.gca()
         ax.xaxis.set_major_formatter(StrMethodFormatter('{x:,.0f}'))
         ax.yaxis.set_major_formatter(StrMethodFormatter('{x:,.0f}'))
@@ -75,7 +82,7 @@ def _plot(self, x_label, y_label, y_axis, z_axis, filename):
             secaxy.yaxis.set_major_formatter(StrMethodFormatter('{x:,.0f}'))
 
         for x in ['pdf', 'png']:
-            plt.savefig(PathMaker.plot_file(filename, x), bbox_inches='tight')
+            plt.savefig(PathMaker.plot_file(type, x), bbox_inches='tight')
 
     @staticmethod
     def nodes(data):
@@ -86,13 +93,18 @@ def nodes(data):
     def tx_size(data):
         return search(r'Transaction size: .*', data).group(0)
 
+    @staticmethod
+    def max_latency(data):
+        x = search(r'Max latency: (\d+)', data).group(1)
+        return f'Max latency: {float(x) / 1000:,.0f} s'
+
     @classmethod
     def plot_robustness(cls, z_axis):
         assert hasattr(z_axis, '__call__')
         x_label = 'Input rate (tx/s)'
         y_label = ['Throughput (tx/s)', 'Throughput (MB/s)']
 
-        files = glob(PathMaker.agg_file(r'[0-9]*', 'x', r'*'))
+        files = glob(PathMaker.agg_file(r'[0-9]*', 'x', r'*', 'any'))
         ploter = cls(files)
         ploter._plot(x_label, y_label, ploter._tps, z_axis, 'robustness')
 
@@ -102,7 +114,7 @@ def plot_latency(cls, z_axis):
         x_label = 'Throughput (tx/s)'
         y_label = ['Latency (ms)']
 
-        files = glob(PathMaker.agg_file(r'[0-9]*', 'any', r'*'))
+        files = glob(PathMaker.agg_file(r'[0-9]*', 'any', r'*', 'any'))
         ploter = cls(files)
         ploter._plot(x_label, y_label, ploter._latency, z_axis, 'latency')
 
@@ -112,6 +124,6 @@ def plot_tps(cls, z_axis):
         x_label = 'Committee size'
         y_label = ['Throughput (tx/s)', 'Throughput (MB/s)']
 
-        files = glob(PathMaker.agg_file('x', 'any', r'*'))
+        files = glob(PathMaker.agg_file('x', 'any', r'*', r'*'))
         ploter = cls(files)
         ploter._plot(x_label, y_label, ploter._tps, z_axis, 'tps')

benchmark/benchmark/utils.py

@@ -1,3 +1,6 @@
+from os.path import join
+
+
 class BenchError(Exception):
     def __init__(self, message, error):
         assert isinstance(error, Exception)
@@ -9,11 +12,11 @@ def __init__(self, message, error):
 class PathMaker:
     @staticmethod
     def binary_path():
-        return '../target/release/'
+        return join('..', 'target', 'release')
 
     @staticmethod
     def node_crate_path():
-        return '../node'
+        return join('..', 'node')
 
     @staticmethod
     def committee_file():
@@ -35,37 +38,42 @@ def db_path(i):
 
     @staticmethod
     def logs_path():
-        return './logs'
+        return 'logs'
 
     @staticmethod
     def node_log_file(i):
         assert isinstance(i, int) and i >= 0
-        return f'{PathMaker.logs_path()}/node-{i}.log'
+        return join(PathMaker.logs_path(), f'node-{i}.log')
 
     @staticmethod
     def client_log_file(i):
         assert isinstance(i, int) and i >= 0
-        return f'{PathMaker.logs_path()}/client-{i}.log'
+        return join(PathMaker.logs_path(), f'client-{i}.log')
 
     @staticmethod
     def results_path():
-        return './results'
+        return 'results'
 
     @staticmethod
     def result_file(nodes, rate, tx_size):
-        return f'{PathMaker.results_path()}/bench-{nodes}-{rate}-{tx_size}.txt'
+        return join(
+            PathMaker.results_path(), f'bench-{nodes}-{rate}-{tx_size}.txt'
+        )
 
     @staticmethod
     def plots_path():
-        return './plots'
+        return 'plots'
 
     @staticmethod
-    def agg_file(nodes, rate, tx_size):
-        return f'{PathMaker.plots_path()}/agg-{nodes}-{rate}-{tx_size}.txt'
+    def agg_file(nodes, rate, tx_size, max_latency):
+        return join(
+            PathMaker.plots_path(),
+            f'agg-{nodes}-{rate}-{tx_size}-{max_latency}.txt'
+        )
 
     @staticmethod
     def plot_file(name, ext):
-        return f'{PathMaker.plots_path()}/{name}.{ext}'
+        return join(PathMaker.plots_path(), f'{name}.{ext}')
 
 
 class Color:
@@ -110,6 +118,7 @@ def error(e):
 
 def progress_bar(iterable, prefix='', suffix='', decimals=1, length=30, fill='█', print_end='\r'):
     total = len(iterable)
+
     def printProgressBar(iteration):
         formatter = '{0:.'+str(decimals)+'f}'
         percent = formatter.format(100 * (iteration / float(total)))