Add support for --memory-limit= as a Client option on Linux

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))

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):
 

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,15 +1262,19 @@ 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)
         raise
 
     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()

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))

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))