Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Print tx creation timestamp on worker side #7

Closed
wants to merge 1 commit into from
Closed

feat: Print tx creation timestamp on worker side #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 23 additions & 17 deletions benchmark/benchmark/local.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,49 +5,55 @@
from time import sleep

from benchmark.commands import CommandMaker
from benchmark.config import Key, LocalCommittee, NodeParameters, BenchParameters, ConfigError
from benchmark.config import (
Key,
LocalCommittee,
NodeParameters,
BenchParameters,
ConfigError,
)
from benchmark.logs import LogParser, ParseError
from benchmark.utils import Print, BenchError, PathMaker


class LocalBench:
BASE_PORT = 3000
BASE_PORT = 4000

def __init__(self, bench_parameters_dict, node_parameters_dict):
try:
self.bench_parameters = BenchParameters(bench_parameters_dict)
self.node_parameters = NodeParameters(node_parameters_dict)
except ConfigError as e:
raise BenchError('Invalid nodes or bench parameters', e)
raise BenchError("Invalid nodes or bench parameters", e)

def __getattr__(self, attr):
return getattr(self.bench_parameters, attr)

def _background_run(self, command, log_file):
name = splitext(basename(log_file))[0]
cmd = f'{command} 2> {log_file}'
subprocess.run(['tmux', 'new', '-d', '-s', name, cmd], check=True)
cmd = f"{command} 2> {log_file}"
subprocess.run(["tmux", "new", "-d", "-s", name, cmd], check=True)

def _kill_nodes(self):
try:
cmd = CommandMaker.kill().split()
subprocess.run(cmd, stderr=subprocess.DEVNULL)
except subprocess.SubprocessError as e:
raise BenchError('Failed to kill testbed', e)
raise BenchError("Failed to kill testbed", e)

def run(self, debug=False):
assert isinstance(debug, bool)
Print.heading('Starting local benchmark')
Print.heading("Starting local benchmark")

# Kill any previous testbed.
self._kill_nodes()

try:
Print.info('Setting up testbed...')
Print.info("Setting up testbed...")
nodes, rate = self.nodes[0], self.rate[0]

# Cleanup all files.
cmd = f'{CommandMaker.clean_logs()} ; {CommandMaker.cleanup()}'
cmd = f"{CommandMaker.clean_logs()} ; {CommandMaker.cleanup()}"
subprocess.run([cmd], shell=True, stderr=subprocess.DEVNULL)
sleep(0.5) # Removing the store may take time.

Expand Down Expand Up @@ -77,12 +83,12 @@ def run(self, debug=False):
workers_addresses = committee.workers_addresses(self.faults)
rate_share = ceil(rate / committee.workers())
for i, addresses in enumerate(workers_addresses):
for (id, address) in addresses:
for id, address in addresses:
cmd = CommandMaker.run_client(
address,
self.tx_size,
rate_share,
[x for y in workers_addresses for _, x in y]
[x for y in workers_addresses for _, x in y],
)
log_file = PathMaker.client_log_file(i, id)
self._background_run(cmd, log_file)
Expand All @@ -94,34 +100,34 @@ def run(self, debug=False):
PathMaker.committee_file(),
PathMaker.db_path(i),
PathMaker.parameters_file(),
debug=debug
debug=debug,
)
log_file = PathMaker.primary_log_file(i)
self._background_run(cmd, log_file)

# Run the workers (except the faulty ones).
for i, addresses in enumerate(workers_addresses):
for (id, address) in addresses:
for id, address in addresses:
cmd = CommandMaker.run_worker(
PathMaker.key_file(i),
PathMaker.committee_file(),
PathMaker.db_path(i, id),
PathMaker.parameters_file(),
id, # The worker's id.
debug=debug
debug=debug,
)
log_file = PathMaker.worker_log_file(i, id)
self._background_run(cmd, log_file)

# Wait for all transactions to be processed.
Print.info(f'Running benchmark ({self.duration} sec)...')
Print.info(f"Running benchmark ({self.duration} sec)...")
sleep(self.duration)
self._kill_nodes()

# Parse logs and return the parser.
Print.info('Parsing logs...')
Print.info("Parsing logs...")
return LogParser.process(PathMaker.logs_path(), faults=self.faults)

except (subprocess.SubprocessError, ParseError) as e:
self._kill_nodes()
raise BenchError('Failed to run benchmark', e)
raise BenchError("Failed to run benchmark", e)
Loading
Loading