diff --git a/Client/bench.py b/Client/bench.py index 33d7d7a4..b4a4eaf7 100644 --- a/Client/bench.py +++ b/Client/bench.py @@ -33,7 +33,9 @@ import queue import re import subprocess -import sys +import threading +import time +import psutil ## Local imports must only use "import x", never "from x import ..." @@ -79,7 +81,7 @@ def single_core_bench(binary, outqueue): except: # Signal an error with (None, None) outqueue.put((None, None)) -def multi_core_bench(binary, threads): +def multi_core_bench(binary, threads, monitor_memory=False): outqueue = multiprocessing.Queue() @@ -92,26 +94,47 @@ def multi_core_bench(binary, threads): for process in processes: process.start() + stop_event = None + monitor = None + monitor_result = {} + + if monitor_memory: + pids = [process.pid for process in processes] + stop_event = threading.Event() + monitor = threading.Thread(target=monitor_peak_memory, args=(pids, stop_event, monitor_result)) + monitor.start() + try: # Every process deposits exactly one result into the Queue - return [outqueue.get(timeout=MAX_BENCH_TIME_SECONDS) for ii in range(threads)] + results = [outqueue.get(timeout=MAX_BENCH_TIME_SECONDS) for _ in range(threads)] except queue.Empty: # Force kill the engine, thus causing the processes to finish utils.kill_process_by_name(binary) raise utils.OpenBenchBadBenchException('[%s] Bench Exceeded Max Duration' % (binary)) finally: # Join everything to avoid zombie processes + if monitor_memory: + stop_event.set() + monitor.join() for process in processes: process.join() -def run_benchmark(binary, threads, sets, expected=None): + return results, monitor_result.get('peak', 0) + +def run_benchmark(binary, threads, sets, expected=None, monitor_memory=False): engine = os.path.basename(binary) + peak_memory_kb = 0 benches, speeds = [], [] - for ii in range(sets): - for bench, speed in multi_core_bench(binary, threads): + for _ in range(sets): + results, peak_memory_run_kb = multi_core_bench(binary, threads, monitor_memory) + + for bench, speed in results: benches.append(bench); speeds.append(speed) + if monitor_memory: + peak_memory_kb = max(peak_memory_kb, peak_memory_run_kb) + if None in benches or None in speeds: raise utils.OpenBenchBadBenchException('[%s] Failed to Execute Benchmark' % (engine)) @@ -121,4 +144,36 @@ def run_benchmark(binary, threads, sets, expected=None): if expected and expected != benches[0]: raise utils.OpenBenchBadBenchException('[%s] Wrong Bench: %d' % (engine, benches[0])) - return sum(speeds) // len(speeds), benches[0] + return sum(speeds) // len(speeds), benches[0], peak_memory_kb + +def sample_engine_memory(workers): + + peak_kb = 0 + for worker in workers: + try: # Engines are direct children of the multiprocessing workers + for engine in worker.children(recursive=True): + try: + info = engine.memory_full_info() + peak_kb += getattr(info, 'pss', info.uss) // 1024 + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + + return peak_kb + +def monitor_peak_memory(worker_pids, stop_event, result): + + MEMORY_SAMPLE_SECONDS = 0.1 + + workers = [] + for pid in worker_pids: + try: workers.append(psutil.Process(pid)) + except psutil.NoSuchProcess: pass + + peak_kb = 0 + while not stop_event.is_set(): + peak_kb = max(peak_kb, sample_engine_memory(workers)) + time.sleep(MEMORY_SAMPLE_SECONDS) + + result['peak'] = max(peak_kb, sample_engine_memory(workers)) diff --git a/Client/utils.py b/Client/utils.py index b2ff2ad3..70d1defa 100644 --- a/Client/utils.py +++ b/Client/utils.py @@ -85,6 +85,11 @@ def __init__(self): self.message = '' super().__init__(self.message) +class OpenBenchInsufficientMemoryException(Exception): + def __init__(self, message): + self.message = message + super().__init__(self.message) + def kill_process_by_name(process_name): diff --git a/Client/worker.py b/Client/worker.py index 7f90ec4d..7a7d9f67 100644 --- a/Client/worker.py +++ b/Client/worker.py @@ -69,7 +69,6 @@ IS_WINDOWS = platform.system() == 'Windows' # Don't touch this IS_LINUX = platform.system() != 'Windows' # Don't touch this - class Configuration: ## Handles configuring the worker with the server. This means collecting @@ -104,16 +103,17 @@ def __init__(self, args): def process_args(self, args): # Extract all of the options - self.username = args.username - self.password = args.password - self.server = args.server - self.threads = int(args.threads) if args.threads != 'auto' else self.physical_cores - self.sockets = int(args.nsockets) - self.identity = args.identity if args.identity else 'None' - self.syzygy_path = args.syzygy if args.syzygy else None - self.fleet = args.fleet if args.fleet else False - self.noisy = args.noisy if args.noisy else False - self.focus = args.focus if args.focus else [] + self.username = args.username + self.password = args.password + self.server = args.server + self.threads = int(args.threads) if args.threads != 'auto' else self.physical_cores + self.sockets = int(args.nsockets) + self.memory_limit = int(args.memory_limit) if args.memory_limit else None + self.identity = args.identity if args.identity else 'None' + self.syzygy_path = args.syzygy if args.syzygy else None + self.fleet = args.fleet if args.fleet else False + self.noisy = args.noisy if args.noisy else False + self.focus = args.focus if args.focus else [] def check_requirements(self): @@ -155,6 +155,7 @@ def init_client(self): def validate_setup(self): + assert self.memory_limit is None or IS_LINUX assert self.threads >= self.sockets assert self.threads % self.sockets == 0 assert min(self.threads, self.sockets) >= 1 @@ -304,6 +305,17 @@ def report_bad_bench(config, error): return ServerReporter.report(config, 'clientBenchError', payload) + @staticmethod + def report_insufficient_memory(config, error): + + payload = { + 'test_id' : config.workload['test']['id'], + 'error' : error, + 'logs' : error, + } + + return ServerReporter.report(config, 'clientSubmitError', payload) + @staticmethod def report_results(config, batches): @@ -869,10 +881,21 @@ def find_pgn_error(reason, command): def determine_scale_factor(config, dev_name, base_name): # Run the benchmarks and compute the scaling NPS value - dev_nps = safe_run_benchmarks(config, 'dev' , dev_name ) - base_nps = safe_run_benchmarks(config, 'base', base_name) + dev_nps , dev_peak = safe_run_benchmarks(config, 'dev' , dev_name ) + base_nps, base_peak = safe_run_benchmarks(config, 'base', base_name) ServerReporter.report_nps(config, dev_nps, base_nps) + if config.memory_limit: + required_mb = estimate_required_memory_kb(config, dev_name, dev_peak, base_name, base_peak) / 1024 + print('\nEstimated memory required: %.2f MB' % required_mb) + + if required_mb > config.memory_limit: + error = '[Error] Insufficient memory to run this Workload (required: %.2f MB, limit: %.2f MB)' % (required_mb, config.memory_limit) + + config.blacklist.append(config.workload['test']['id']) + ServerReporter.report_insufficient_memory(config, error) + raise utils.OpenBenchInsufficientMemoryException(error) + dev_factor = base_factor = None # Scaling is only done relative to the Dev Engine @@ -896,6 +919,40 @@ def determine_scale_factor(config, dev_name, base_name): return factor +def estimate_required_memory_kb(config, dev_name, dev_peak, base_name, base_peak): + + # Reserve 1GB for Python, Fastchess, etc. + MEMORY_RESERVED_KB = 1024**2 + MEMORY_OVERHEAD_FACTOR = 1.25 + + def estimate_memory_per_engine(branch, engine, peak): + bench_hash = get_engine_default_hash(os.path.join('Engines', engine)) + workload_hash = int(re.search(r'Hash=(\d+)', config.workload['test'][branch]['options']).group(1)) + hash_delta = 1024 * max(0, workload_hash - bench_hash) + return peak / config.threads + hash_delta + + runner_cnt = config.workload['distribution']['runner-count'] + concurrency_per = config.workload['distribution']['concurrency-per'] + memory_per_pair_kb = (estimate_memory_per_engine('dev', dev_name, dev_peak) + estimate_memory_per_engine('base', base_name, base_peak)) + + return int(runner_cnt * concurrency_per * memory_per_pair_kb * MEMORY_OVERHEAD_FACTOR + MEMORY_RESERVED_KB) + +def get_engine_default_hash(binary): + + DEFAULT_BENCH_HASH_MB = 16 + + process = Popen(['./%s' % binary], stdin=PIPE, stdout=PIPE, stderr=STDOUT) + + try: + stdout = process.communicate(input=b'uci\nquit\n', timeout=10)[0].decode('utf-8', 'ignore') + match = re.search(r'option name Hash type spin default (\d+)', stdout, re.IGNORECASE) + return int(match.group(1)) + except subprocess.TimeoutExpired: + process.kill() + + print('[Warning] Unable to determine default Hash size for %s' % (binary)) + return DEFAULT_BENCH_HASH_MB + ## Functions interacting with the OpenBench server that establish the initial ## connection and then make simple requests to retrieve Workloads as json objects @@ -1205,7 +1262,7 @@ def safe_run_benchmarks(config, branch, engine): try: print('\nRunning %dx Benchmarks for %s' % (config.threads, name)) - speed, nodes = bench.run_benchmark(binary, config.threads, 1, expected) + speed, nodes, peak_memory_kb = bench.run_benchmark(binary, config.threads, 1, expected, config.memory_limit) except utils.OpenBenchBadBenchException as error: ServerReporter.report_bad_bench(config, error.message) @@ -1213,7 +1270,11 @@ def safe_run_benchmarks(config, branch, engine): print('Bench for %s is %d' % (name, nodes)) print('Speed for %s is %d' % (name, speed)) - return speed + + if config.memory_limit: + print('Peak memory for %s is %.2f MB' % (name, peak_memory_kb / 1024)) + + return speed, peak_memory_kb def build_runner_command(config, dev_cmd, base_cmd, scale_factor, timestamp, runner_idx): @@ -1307,13 +1368,14 @@ def parse_arguments(client_args): ) # Arguments specific to worker.py - p.add_argument('-T', '--threads' , help='Total Threads' , required=True ) - p.add_argument('-N', '--nsockets', help='Number of Sockets' , required=True ) - p.add_argument('-I', '--identity', help='Machine pseudonym' , required=False ) - p.add_argument( '--syzygy' , help='Syzygy WDL' , required=False ) - p.add_argument( '--fleet' , help='Fleet Mode' , action='store_true') - p.add_argument( '--noisy' , help='Reject time-based workloads' , action='store_true') - p.add_argument( '--focus' , help='Prefer certain engine(s)' , nargs='+' ) + p.add_argument('-T', '--threads' , help='Total Threads' , required=True ) + p.add_argument('-N', '--nsockets' , help='Number of Sockets' , required=True ) + p.add_argument('-I', '--identity' , help='Machine pseudonym' , required=False ) + p.add_argument( '--memory-limit', help='Memory limit in MB (Linux only)', required=False ) + p.add_argument( '--syzygy' , help='Syzygy WDL' , required=False ) + p.add_argument( '--fleet' , help='Fleet Mode' , action='store_true') + p.add_argument( '--noisy' , help='Reject time-based workloads' , action='store_true') + p.add_argument( '--focus' , help='Prefer certain engine(s)' , nargs='+' ) # Ignore unknown arguments ( from client ) worker_args, unknown = p.parse_known_args() diff --git a/Scripts/bench_all.py b/Scripts/bench_all.py index 0c247e3c..75826bf7 100755 --- a/Scripts/bench_all.py +++ b/Scripts/bench_all.py @@ -91,6 +91,7 @@ def get_engine(engine, config): parser.add_argument('--engines', help='List of specific engines', nargs='+') parser.add_argument('--threads', help='Concurrent Benchmarks', required=True, type=int) parser.add_argument('--sets' , help='Benchmark Sample Count', required=True, type=int) + parser.add_argument('--memory' , help='Sample peak memory', action='store_true') args = credentialed_cmdline_args(parser) # Get the build info, and default network info, for all applicable engines @@ -121,6 +122,7 @@ def get_engine(engine, config): # Pretty Formatting max_length = max(len(engine) for engine in engines) print_format = '%-' + str(max_length) + 's %8d nps %10d nodes in %6.3f seconds' + print_format_mem = '%-' + str(max_length) + 's %8d nps %10d nodes in %6.3f seconds and %6.2f MB peak memory' for engine in engines: @@ -133,8 +135,11 @@ def get_engine(engine, config): continue try: - nps, nodes = run_benchmark(binary, args.threads, args.sets) - print (print_format % (engine, nps, nodes, nodes / max(1e-6, nps))) + nps, nodes, peak_kb = run_benchmark(binary, args.threads, args.sets, monitor_memory=args.memory) + if args.memory: + print (print_format_mem % (engine, nps, nodes, nodes / max(1e-6, nps), peak_kb / 1024)) + else: + print (print_format % (engine, nps, nodes, nodes / max(1e-6, nps))) except OpenBenchBadBenchException as error: print ('%s: %s' % (engine, error)) diff --git a/Scripts/bench_engine.py b/Scripts/bench_engine.py index 544f3d20..2acaab0c 100755 --- a/Scripts/bench_engine.py +++ b/Scripts/bench_engine.py @@ -36,9 +36,13 @@ p.add_argument('-E', '--engine' , help='Relative path to Binary', required=True) p.add_argument('-T', '--threads' , help='Concurrent Benchmarks', required=True, type=int) p.add_argument('-S', '--sets' , help='Benchmark Sample Count', required=True, type=int) + p.add_argument('-M', '--memory' , help='Sample peak memory', action='store_true') args = p.parse_args() - speed, bench = run_benchmark(args.engine, args.threads, args.sets) + speed, bench, peak_kb = run_benchmark(args.engine, args.threads, args.sets, monitor_memory=args.memory) print('Bench for %s is %d' % (args.engine, bench)) print('Speed for %s is %d' % (args.engine, speed)) + + if args.memory: + print('Peak memory for %s is %.2f MB' % (args.engine, peak_kb / 1024))