"""Deploy isolated native peers from an external, identity-pinned SSH profile. Controllers collect evidence only. All transfers, byte checks and drain/resource assertions run inside hardware_peer; this is not a performance benchmark. """ import argparse import hashlib import json from pathlib import Path import shlex import subprocess import time import uuid import cleanup import installed import gdr import staging def restart_plan(case): mode = case.get('mode') if mode == 'installed-memory-cleanup': installed.validate(case) return None if mode in ('installed-restart', 'installed-interruption', 'installed-lost-completion', 'installed-cleanup-failure'): installed.validate(case) return case['fault_role'], case['generations'] if mode in ('large-cq-busy', 'large-pd-busy', 'large-qp-cleanup-error', 'large-channel-busy'): role = case.get('fault_role') if role not in ('source', 'requester'): raise ValueError('native cleanup faults require an explicit source/requester fault_role') return role, 2 if mode in ('live-reconnect', 'mr-cleanup-error', 'mw-cleanup-error', 'monitor-recovery'): role = case.get('fault_role') if (role not in ('source', 'requester') or (mode == 'mw-cleanup-error' and role != 'source') or (mode == 'live-reconnect' and role != 'requester')): raise ValueError('lifecycle cases require an explicit supported fault_role') return None if 'fault_role' in case: raise ValueError('fault_role is reserved for native cleanup modes') if mode in ('large-peer-exit', 'large-peer-restart'): return 'source', 3 if mode.endswith('restart') else 2 if mode in ('large-requester-exit', 'large-requester-restart'): return 'requester', 3 if mode.endswith('restart') else 2 return None def native_peer_test(profile): runner = profile.get('runner') if runner == 'orbitte-multirail': return 'transport::rdma::cq_lane::tests::hardware::multirail::multirail_peer' if runner == 'orbitte-connected': return 'transport::rdma::cq_lane::tests::hardware::connected::connected_endpoint_peer' if runner == 'orbitte-shared-cq': return 'transport::rdma::cq_lane::tests::hardware::shared_cq_peer' if runner == 'orbitte-window': return 'transport::rdma::cq_lane::tests::hardware::source_window_peer' if runner == 'orbitkv-peer-host': return 'peer::orbitte::tests::rdma_host::registered_host_rdma_peer' if runner in (None, 'orbitte-installed-host'): return None raise ValueError('unsupported hardware runner') def validate_peer_listing(test, listing): matches = [line.strip() for line in listing.splitlines() if line.strip().endswith((': test', ': benchmark'))] if matches != [test + ': test']: raise ValueError('native peer selector must resolve exactly one test: ' + repr(matches)) class Controller: def __init__(self, profile, binary, output): self.profile, self.binary, self.output = profile, binary, output self.sha256 = hashlib.sha256(binary.read_bytes()).hexdigest() self.remote = '/var/tmp/orbitte-runs/' + output.name + '-' + uuid.uuid4().hex[:12] self.commands = [] self.started = [] self.connections = [] self.fault_library = None self.inventory_tool = None def command(self, command, *, allow_failure=False): result = subprocess.run(command, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=60) self.commands.append({'command': command, 'exit_code': result.returncode, 'stdout': result.stdout, 'stderr': result.stderr}) (self.output / 'commands.json').write_text(json.dumps(self.commands, indent=2) + '\n') if result.returncode and not allow_failure: raise RuntimeError(f'command failed: {command!r}: {result.stderr}') return result.stdout def ssh(self, node, arguments): return self.command(['ssh', '-F', self.profile['ssh_config'], self.profile['nodes'][node]['alias'], shlex.join(arguments)]) def copy(self, node, local, remote): # SFTP/scp reads only the explicitly named runner/config files. SSH keys # remain in the controller's existing configuration and never in bundles. self.command(['scp', '-F', self.profile['ssh_config'], str(local), f'{self.profile["nodes"][node]["alias"]}:{remote}']) def deploy(self, node): identity = self.ssh(node, ['python3', '-c', "import json,socket;from pathlib import Path;print(json.dumps([socket.gethostname(),Path('/proc/sys/kernel/random/boot_id').read_text().strip()]))"]) expected = self.profile['nodes'][node] if json.loads(identity) != [expected['hostname'], expected['boot_id']]: raise RuntimeError('host/boot identity changed before deployment') self.ssh(node, ['mkdir', '-m', '700', '-p', self.remote]) if self.fault_library is not None: self.copy(node, self.fault_library, self.remote + '/liborbitte_cq_fault.so') if self.inventory_tool is not None: self.copy(node, self.inventory_tool, self.remote + '/discovery_inventory.py') self.copy(node, self.binary, self.remote + '/hardware_peer') self.copy(node, Path(__file__).with_name('supervise.py'), self.remote + '/supervise.py') self.ssh(node, ['chmod', '700', self.remote + '/hardware_peer']) self.ssh(node, ['ldd', self.remote + '/hardware_peer']) def collect(self, node, root): names = ['inventory-before.json', 'inventory-after.json', 'running.json', 'stdout.log', 'stderr.log', 'result.json', 'supervisor.log', 'maps.txt', 'restart.json', 'cuda-staging.jsonl'] names += ['cq-fault-' + str(g) + '.jsonl' for g in range(3)] names += ['maps-' + str(g) + '.txt' for g in range(3)] names += ['restart-' + kind + '-' + str(g) + '.txt' for kind in ('state', 'request', 'peer-state', 'peer', 'gate') for g in range(3)] script = "import json;from pathlib import Path;p=Path(" + repr(root) + ");print(json.dumps({n:(p/n).read_text() if (p/n).exists() else None for n in " + repr(names) + "}))" return json.loads(self.ssh(node, ['python3', '-c', script])) def start(self, node, root, kind, arguments, transport, environment=None): target = self.profile['nodes'][node] locks = [] if transport == 'rdma': if self.profile.get('runner') in ('orbitte-multirail', 'orbitte-installed-host') or self.profile.get('rails'): locks.extend(self.rail_locks(node)) else: locks.append(target['nic_lock']) environment = {**target.get('env', {}), **(environment or {})} if kind != 'cpu': locks.append(target['gpu_lock']) environment['CUDA_VISIBLE_DEVICES'] = target['gpu_uuid'] config = {'hostname': target['hostname'], 'boot_id': target['boot_id'], 'binary_sha256': self.sha256, 'arguments': arguments, 'locks': locks, 'env': environment, 'timeout_seconds': 55} if environment.get('ORBITTE_HOST_DISCOVERY') == 'auto': if self.inventory_tool is None: raise ValueError('automatic discovery requires its packaged inventory tool') config['inventory_tool_sha256'] = hashlib.sha256(self.inventory_tool.read_bytes()).hexdigest() if 'ORBITTE_RESTART_GENERATIONS' in environment: config.update(restart_role=environment['ORBITTE_SHARED_ROLE'], restart_kill_role=environment['ORBITTE_RESTART_KILL_ROLE'], restart_generations=int(environment['ORBITTE_RESTART_GENERATIONS']), restart_live_peer=environment.get('ORBITTE_HOST_MODE') in ('installed-restart', 'installed-interruption', 'installed-lost-completion', 'installed-cleanup-failure')) if environment.get('ORBITTE_STAGING_OBSERVE') == '1': if self.fault_library is None: raise ValueError('staging gate requires its pinned CUDA observer') environment.update(LD_PRELOAD=self.remote + '/liborbitte_cq_fault.so', ORBITTE_IPC_LOG=root + '/cuda-staging.jsonl') config['fault_library_sha256'] = hashlib.sha256(self.fault_library.read_bytes()).hexdigest() if environment.get('ORBITTE_GDR_OBSERVE') == '1': if self.fault_library is None: raise ValueError('GPU direct gate requires its pinned native observer') environment['LD_PRELOAD'] = self.remote + '/liborbitte_cq_fault.so' config['fault_library_sha256'] = hashlib.sha256(self.fault_library.read_bytes()).hexdigest() if (environment.get('ORBITTE_HOST_MODE') in ('installed-lost-completion', 'installed-cleanup-failure', 'installed-inflight-dma', 'installed-rail-accounting') and environment['ORBITTE_SHARED_ROLE'] == 'requester') or ( environment.get('ORBITTE_HOST_MODE') == 'installed-memory-cleanup' and environment['ORBITTE_SHARED_ROLE'] == environment['ORBITTE_CLEANUP_ROLE']): if self.fault_library is None: raise ValueError('native fault requires its pinned observer library') environment['LD_PRELOAD'] = self.remote + '/liborbitte_cq_fault.so' config['fault_library_sha256'] = hashlib.sha256(self.fault_library.read_bytes()).hexdigest() local = self.output / (Path(root).name + '.json') local.write_text(json.dumps(config, indent=2) + '\n') self.ssh(node, ['mkdir', '-m', '700', '-p', root]) self.copy(node, local, root + '/config.json') self.started.append((node, root)) command = ['ssh', '-F', self.profile['ssh_config'], target['alias'], shlex.join(['python3', self.remote + '/supervise.py', 'start', root + '/config.json'])] stdout = self.output / (Path(root).name + '-starter.stdout') stderr = self.output / (Path(root).name + '-starter.stderr') with stdout.open('wb') as output, stderr.open('wb') as errors: process = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=output, stderr=errors) self.connections.append((process, command, stdout, stderr)) def settle(self): pending = list(self.started) evidence = {} deadline = time.monotonic() + 65 while pending and time.monotonic() < deadline: for node, root in pending[:]: try: record = self.collect(node, root) evidence[node + ':' + root] = record if record['result.json'] is not None: result = json.loads(record['result.json']) if result.get('child_reaped') and result.get('resource_locks_released'): pending.remove((node, root)) except Exception as error: evidence[node + ':' + root] = {'collection_error': str(error)} if pending: time.sleep(0.5) connections = [] for process, command, stdout, stderr in self.connections: try: code = process.wait(timeout=max(0.1, deadline - time.monotonic())) except subprocess.TimeoutExpired: code = None connections.append({'command': command, 'local_ssh_pid': process.pid, 'exit_code': code, 'stdout': str(stdout), 'stderr': str(stderr)}) reaped = not pending and all(connection['exit_code'] == 0 for connection in connections) result = {'all_supervised_children_reaped': reaped, 'unconfirmed': pending, 'evidence': evidence, 'starter_connections': connections} (self.output / 'cleanup.json').write_text(json.dumps(result, indent=2) + '\n') return reaped def rail_locks(self, node): rails = self.profile.get('rails') if (not isinstance(rails, list) or len(rails) != 2 or any(not isinstance(name, str) or not name or any(not (char.isascii() and (char.isalnum() or char in '_.-')) for char in name) for name in rails) or len(set(rails)) != 2): raise ValueError('multirail requires two distinct explicit RNIC names') locks = self.profile['nodes'][node].get('nic_locks') if (not isinstance(locks, dict) or set(locks) != set(rails) or any(not isinstance(path, str) or not path.startswith('/') for path in locks.values()) or len(set(locks.values())) != 2): raise ValueError('each multirail RNIC requires its own absolute lock path') return [locks[name] for name in rails] def case(self, index, case): if self.profile.get('runner') == 'orbitte-installed-host': installed.validate(case) for node in (case['source'], case['requester']): self.rail_locks(node) return self.run_case(index, case, lambda node, role, kind, root, readiness=None: installed.command(self.profile, case, node, role, root, readiness)) if self.profile.get('runner') == 'orbitte-multirail': return self.multirail_case(index, case) source, requester = case['source'], case['requester'] source_root = self.remote + f'/{index:02}-source' requester_root = self.remote + f'/{index:02}-requester' consumer = self.profile.get('runner') == 'orbitkv-peer-host' window_probe = self.profile.get('runner') == 'orbitte-window' shared = self.profile.get('runner') in ['orbitte-shared-cq', 'orbitte-window', 'orbitte-connected'] if self.profile.get('runner') not in [None, 'orbitkv-peer-host', 'orbitte-shared-cq', 'orbitte-window', 'orbitte-connected']: raise ValueError('unsupported hardware runner') if consumer and (case['transport'] != 'rdma' or case['source_kind'] != 'cpu' or case['requester_kind'] != 'cpu'): raise ValueError('OrbitKV registered-read component gate accepts host RDMA only') discovery = case.get('rdma_auto', False) if type(discovery) is not bool or (discovery and case['transport'] != 'rdma'): raise ValueError('rdma_auto requires a boolean RDMA profile') cq_events = case.get('rdma_events', False) if type(cq_events) is not bool or (cq_events and (case['transport'] != 'rdma' or consumer)): raise ValueError('rdma_events requires a boolean native RDMA profile') multi_context = case.get('rdma_multi_context', False) if type(multi_context) is not bool or (multi_context and ( self.profile.get('runner') != 'orbitte-shared-cq' or case.get('duplex') or case['transport'] != 'rdma' or case['source_kind'] != 'cpu' or case['requester_kind'] != 'cpu')): raise ValueError('rdma_multi_context requires a boolean host shared-CQ profile') shared_qp = case.get("shared_qp") if shared_qp is not None and (self.profile.get("runner") != "orbitte-shared-cq" or shared_qp not in ("healthy", "lost-receive") or case.get("operation") not in ("read", "write") or case.get("transport") != "rdma" or case.get("source_kind") != "cpu" or case.get("requester_kind") != "cpu" or type(case.get("payload")) is not int or not 4096 <= case["payload"] <= 8 << 20 or case.get("duplex") or multi_context or case.get("rdma_max_cqes") != 4 or case.get("rdma_max_send_wr") != 4): raise ValueError("invalid shared QP grant profile") def command(node, role, kind, root, readiness=None): if shared: if case['transport'] != 'rdma' or kind != 'cpu': raise ValueError('shared CQ gate accepts host RDMA only') capacity, window = case['rdma_max_cqes'], case['rdma_max_send_wr'] fragmented = window_probe and case.get('window_probe') in ['fragment-read', 'fragment-write', 'partial-read', 'partial-write'] if type(capacity) is not int or not 1 <= capacity <= 4 or type(window) is not int or not 1 <= window <= (4 if shared_qp else 3 if fragmented else 2): raise ValueError('invalid shared CQ/SQ profile') if fragmented and (capacity != 3 or window != 3 or type(case['payload']) is not int or case['payload'] != 12288): raise ValueError('fragmented fault gate requires three 4096-byte WRs') arguments = [native_peer_test(self.profile), '--exact', '--ignored', '--test-threads=1', '--nocapture'] environment = { 'ORBITTE_SHARED_ROLE': role, 'ORBITTE_SHARED_BIND': self.profile['nodes'][node]['bind'], 'ORBITTE_SHARED_DEVICE': self.profile['rnic'], 'ORBITTE_SHARED_GID': str(self.profile['gid_index']), 'ORBITTE_SHARED_PAYLOAD': str(case['payload']), 'ORBITTE_SHARED_SEED': str(case['seed']), 'ORBITTE_SHARED_CQES': str(capacity), 'ORBITTE_SHARED_WINDOW': str(window), 'ORBITTE_SHARED_MAPS': root + '/maps.txt', } environment['ORBITTE_RDMA_EVENTS'] = str(int(cq_events)) if shared_qp: environment["ORBITTE_SHARED_QP"] = shared_qp environment["ORBITTE_SHARED_OPERATION"] = case["operation"] if multi_context: environment['ORBITTE_RDMA_MULTI_CONTEXT'] = '1' if discovery: environment['ORBITTE_RDMA_DISCOVERY'] = '1' if case.get('duplex'): environment['ORBITTE_SHARED_DUPLEX'] = '1' role_capacity = case.get(role + '_rdma_max_cqes', capacity) if type(role_capacity) is not int or not 1 <= role_capacity <= 4: raise ValueError('invalid duplex CQ capacity') environment['ORBITTE_SHARED_CQES'] = str(role_capacity) if window_probe: if case.get('window_probe') not in ['revoked', 'parent', 'remote', 'range', 'fragment-read', 'fragment-write', 'partial-read', 'partial-write']: raise ValueError('invalid MW probe') environment['ORBITTE_WINDOW_PROBE'] = case['window_probe'] if readiness is not None: if len(readiness) != 1: raise ValueError('invalid shared CQ readiness') environment['ORBITTE_SHARED_PEER'] = readiness[0] return arguments, environment if consumer: arguments = [native_peer_test(self.profile), '--exact', '--ignored', '--test-threads=1', '--nocapture'] environment = { 'ORBITKV_ORBITTE_ROLE': role, 'ORBITKV_ORBITTE_BIND': self.profile['nodes'][node]['bind'], 'ORBITKV_ORBITTE_RDMA_DEVICE': self.profile['rnic'], 'ORBITKV_ORBITTE_RDMA_PORT': str(self.profile.get('rdma_port', 1)), 'ORBITKV_ORBITTE_RDMA_GID_INDEX': str(self.profile['gid_index']), 'ORBITKV_ORBITTE_PAYLOAD': str(case['payload']), 'ORBITKV_ORBITTE_SEED': str(case['seed']), 'ORBITKV_ORBITTE_MAPS': root + '/maps.txt', } if readiness is not None: if len(readiness) != 4 or not readiness[0].startswith('orbitte+rdma://'): raise ValueError('invalid OrbitKV RDMA readiness') environment['ORBITKV_ORBITTE_REMOTE'] = json.dumps(readiness) return arguments, environment arguments = [role, self.profile['nodes'][node]['bind'], self.profile['rnic'], str(self.profile['gid_index']), kind, str(case['payload']), str(case['seed']), case['transport']] host_memory = case.get(role + '_host_memory', case.get('host_memory', 'direct')) if host_memory not in ['direct', 'staged']: raise ValueError('RDMA host memory must be direct or staged') environment = { 'ORBITTE_HARDWARE_MAPS': root + '/maps.txt', 'ORBITTE_RDMA_HOST_MEMORY': host_memory, 'ORBITTE_EXPECT_QP_REUSE': str(int(case['transport'] == 'rdma' and all( case.get(r + '_host_memory', case.get('host_memory', 'direct')) == 'direct' and (case[r + '_kind'] == 'cpu' or case[r + '_kind'].startswith('pinned:')) for r in ['source', 'requester']))), } if 'staging_chunk_size' in case: staging.validate_case(case) environment.update(ORBITTE_SHARED_ROLE=role, ORBITTE_STAGING_OBSERVE='1', ORBITTE_STAGING_CHUNK_SIZE=str(staging.configured_chunk(case, role)), ORBITTE_STAGING_CHUNK_COUNT=str(staging.configured_depth(case, role)), ORBITTE_STAGING_PIPELINE_DEPTH=str(staging.configured_depth(case, role)), ORBITTE_EXPECT_STAGE_DEPTH=str(staging.stage_width(case, role)), ORBITTE_EXPECT_STAGE_CHUNK_SIZE=str(staging.transfer_chunk(case))) if case.get('remote_staging', False): staging.validate_case(case) length = case.get('slice_bytes', case['payload'] // 4 - 23) chunk = staging.transfer_chunk(case) environment.update(ORBITTE_REMOTE_STAGING='1', ORBITTE_EXPECT_QP_REUSE=str(int(case['transport'] == 'rdma')), ORBITTE_EXPECT_GRANTS=str(6 * ((length + chunk - 1) // chunk))) cuda_memory = case.get('cuda_memory', 'staged') if cuda_memory not in ('staged', 'direct'): raise ValueError('CUDA RDMA memory policy must be staged or direct') if cuda_memory == 'direct': if shared or consumer or case['transport'] != 'rdma': raise ValueError('CUDA direct gate requires the public native peer') environment.update(ORBITTE_RDMA_CUDA_MEMORY='direct', MLX5_SCATTER_TO_CQE='0', ORBITTE_GDR_OBSERVE='1', ORBITTE_SHARED_ROLE=role) environment['ORBITTE_EXPECT_QP_REUSE'] = str(int( not case['source_kind'].startswith('cuda:') and all( case.get(r + '_host_memory', case.get('host_memory', 'direct')) == 'direct' for r in ('source', 'requester')))) if self.profile.get('rails'): if cuda_memory != 'direct': raise ValueError('public multirail profile requires explicit CUDA direct policy') environment['ORBITTE_RDMA_RAILS'] = ','.join(self.profile['rails']) count = case.get('expected_grants') if type(count) is not int or not 6 <= count <= 4096: raise ValueError('multirail profile must pin its expected physical grant count') environment['ORBITTE_EXPECT_GRANTS'] = str(count) if 'slice_bytes' in case: length = case['slice_bytes'] if type(length) is not int or not 1 <= length <= case['payload'] // 4 - 23: raise ValueError('explicit slice length exceeds the payload') environment['ORBITTE_RDMA_SLICE_BYTES'] = str(length) for key, variable, limit in [('rdma_block_size', 'ORBITTE_RDMA_BLOCK_SIZE', 64 << 20), ('rdma_max_send_wr', 'ORBITTE_RDMA_MAX_SEND_WR', 64), ('rdma_max_cqes', 'ORBITTE_RDMA_MAX_CQES', 4096), ('rdma_max_endpoints', 'ORBITTE_RDMA_MAX_ENDPOINTS', 4096)]: value = case.get(key) if value is not None: if type(value) is not int or not 1 <= value <= limit: raise ValueError('invalid ' + key) environment[variable] = str(value) environment['ORBITTE_RDMA_EVENTS'] = str(int(cq_events)) if discovery: environment['ORBITTE_RDMA_DISCOVERY'] = '1' return arguments + (readiness[:3] if readiness else []), environment return self.run_case(index, case, command) def multirail_case(self, index, case): source, requester = case['source'], case['requester'] for node in [source, requester]: self.rail_locks(node) if (case.get('transport') != 'rdma' or case.get('source_kind') != 'cpu' or case.get('requester_kind') != 'cpu' or type(case.get('rdma_events')) is not bool or case.get('mode') not in ['healthy', 'cq-error', 'partial-post', 'paused-close', 'mr-rollback', 'concurrent-quota', 'concurrent-grants', 'concurrent-close', 'concurrent-cq-error', 'rq-pressure', 'rq-close', 'rq-cq-error', 'large-quota', 'large-cold', 'large-close', 'large-cq-error', 'folded-quota', 'folded-close', 'folded-cq-error', 'quarter-quota', 'large-qp-error', 'folded-qp-error', 'large-lost-send-cqe', 'folded-lost-send-cqe', 'large-lost-recv-cqe', 'folded-lost-recv-cqe', 'large-peer-exit', 'large-peer-restart', 'large-requester-exit', 'large-requester-restart', 'large-cq-busy', 'large-pd-busy', 'large-qp-cleanup-error', 'large-channel-busy', 'live-reconnect', 'mr-cleanup-error', 'mw-cleanup-error', 'monitor-recovery'] or case.get('operation') not in ['read', 'write'] or type(case.get('fault_rail')) is not int or case['fault_rail'] not in [0, 1] or type(case.get('mask')) is not int or case['mask'] not in [1, 2, 3]): raise ValueError('invalid explicit multirail profile') if case['mode'] != 'healthy' and case['mask'] != 3: raise ValueError('multirail fault controls require both RNICs') if 'num_workers' in case: raise ValueError('use num_lanes for unified worker and CQ parallelism') workers = case.get('num_lanes', 1) if type(workers) is not int or not 1 <= workers <= 64: raise ValueError('num_lanes must be an integer in 1..64') if case.get('mode') in ('cq-error', 'partial-post', 'paused-close') and workers < 2: raise ValueError('native blocking sibling-post controls require at least two workers') restart = restart_plan(case) if case['mode'] == 'large-channel-busy' and not case['rdma_events']: raise ValueError('completion-channel cleanup requires event-driven progress') shared_qp = case.get("shared_qp") if shared_qp is not None and (self.profile.get("runner") != "orbitte-shared-cq" or shared_qp not in ("healthy", "lost-receive") or case.get("operation") not in ("read", "write") or case.get("transport") != "rdma" or case.get("source_kind") != "cpu" or case.get("requester_kind") != "cpu" or type(case.get("payload")) is not int or not 4096 <= case["payload"] <= 8 << 20 or case.get("duplex") or multi_context or case.get("rdma_max_cqes") != 4 or case.get("rdma_max_send_wr") != 4): raise ValueError("invalid shared QP grant profile") def command(node, role, kind, root, readiness=None): arguments = [native_peer_test(self.profile), '--exact', '--ignored', '--test-threads=1', '--nocapture'] environment = { 'ORBITTE_SHARED_ROLE': role, 'ORBITTE_SHARED_BIND': self.profile['nodes'][node]['bind'], 'ORBITTE_SHARED_RAILS': ','.join(self.profile['rails']), 'ORBITTE_SHARED_GID': str(self.profile['gid_index']), 'ORBITTE_SHARED_MAPS': root + '/maps.txt', 'ORBITTE_RDMA_EVENTS': str(int(case['rdma_events'])), 'ORBITTE_RDMA_LANES': str(workers), 'ORBITTE_RAIL_MODE': case['mode'], 'ORBITTE_RAIL_OPERATION': case['operation'], 'ORBITTE_RAIL_FAULT': str(case['fault_rail']), 'ORBITTE_RAIL_MASK': str(case['mask']), 'ORBITTE_RAIL_FAULT_ROLE': str(case.get('fault_role', 'source')), } if restart is not None: environment['ORBITTE_RESTART_KILL_ROLE'] = restart[0] environment['ORBITTE_RESTART_GENERATIONS'] = str(restart[1]) if readiness is not None: if len(readiness) != 1: raise ValueError('invalid multirail readiness') environment['ORBITTE_SHARED_PEER'] = readiness[0] return arguments, environment return self.run_case(index, case, command) def run_case(self, index, case, command): source, requester = case['source'], case['requester'] source_root = self.remote + f'/{index:02}-source' requester_root = self.remote + f'/{index:02}-requester' arguments, environment = command(source, 'source', case['source_kind'], source_root) self.start(source, source_root, case['source_kind'], arguments, case['transport'], environment) started = time.monotonic() readiness = None while time.monotonic() - started < 25: evidence = self.collect(source, source_root) if evidence['result.json'] is not None: raise RuntimeError('source exited before readiness: ' + json.dumps(evidence)) for line in (evidence['stdout.log'] or '').splitlines(): if line.startswith('ORBITTE_READY '): readiness = line.split()[1:] if readiness: break time.sleep(0.2) if readiness is None: raise RuntimeError('source readiness timeout; supervisor will collect and reap its own child') arguments, environment = command(requester, 'requester', case['requester_kind'], requester_root, readiness) self.start(requester, requester_root, case['requester_kind'], arguments, case['transport'], environment) forwarded = set() if 'num_workers' in case: raise ValueError('use num_lanes for unified worker and CQ parallelism') workers = case.get('num_lanes', 1) if type(workers) is not int or not 1 <= workers <= 64: raise ValueError('num_lanes must be an integer in 1..64') if case.get('mode') in ('cq-error', 'partial-post', 'paused-close') and workers < 2: raise ValueError('native blocking sibling-post controls require at least two workers') restart = restart_plan(case) while time.monotonic() - started < 65: evidence = {role: self.collect(node, root) for role, node, root in [('source', source, source_root), ('requester', requester, requester_root)]} if restart is not None: running = json.loads(evidence['source']['running.json'] or '{}') ready = [line.split()[1] for line in (evidence['source']['stdout.log'] or '').splitlines() if line.startswith('ORBITTE_READY ')] generation = (len(ready) - 1 if case.get('mode') in ('installed-restart', 'installed-interruption', 'installed-lost-completion', 'installed-cleanup-failure') else running.get('generation')) if (type(generation) is int and 0 < generation < restart[1] and generation not in forwarded and len(ready) > generation): destination = requester_root + '/restart-peer-' + str(generation) + '.txt' script = "from pathlib import Path;p=Path(" + repr(destination) + ");t=p.with_suffix('.tmp');t.write_text(" + repr(ready[generation]) + ");t.replace(p)" self.ssh(requester, ['python3', '-c', script]) forwarded.add(generation) if all(record['result.json'] is not None for record in evidence.values()): break time.sleep(0.3) passed = all(record['result.json'] is not None and json.loads(record['result.json'])['status'] == 'passed' and record['maps.txt'] is not None for record in evidence.values()) validation_error = None if passed and self.profile.get('runner') == 'orbitte-installed-host': try: installed.verify(self.profile, case, evidence) except (ValueError, KeyError, TypeError) as error: passed = False validation_error = str(error) if passed and case.get('cuda_memory') == 'direct': try: gdr.verify(self.profile, case, evidence) except (ValueError, KeyError, TypeError) as error: passed = False validation_error = str(error) if passed and 'staging_chunk_size' in case: try: staging.verify(case, evidence) except (ValueError, KeyError, TypeError) as error: passed = False validation_error = str(error) if passed and case.get('mode') in cleanup.MODES: try: cleanup.verify(case, evidence) except (KeyError, TypeError, ValueError) as error: passed = False validation_error = str(error) # Preserve every successful/failed run verbatim, including supervisors. (self.output / f'case-{index:02}.json').write_text(json.dumps({'case': case, 'status': 'passed' if passed else 'failed', 'remote_root': self.remote, 'restart_ready_forwarded': sorted(forwarded), 'validation_error': validation_error, 'evidence': evidence}, indent=2) + '\n') if not passed: raise RuntimeError(f'case {index} failed; retained evidence in {self.output}') print(json.dumps({'case': index, 'status': 'passed', **case}), flush=True) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--profile', required=True, type=Path) parser.add_argument('--binary', required=True, type=Path) parser.add_argument('--output', required=True, type=Path) parser.add_argument('--installation', type=Path, help='build_installed.py manifest for the public host/GPU consumer') parser.add_argument('--gdr-observer', type=Path, help='pinned passive native GPU MR/WR observer') parser.add_argument('--staging-observer', type=Path, help='packaged passive CUDA allocation/copy observer') args = parser.parse_args() args.output.mkdir(parents=True, exist_ok=False) profile = json.loads(args.profile.read_text()) controller = Controller(profile, args.binary.resolve(), args.output.resolve()) manifest = {'profile': profile, 'binary_sha256': controller.sha256, 'remote_root': controller.remote, 'started_unix': time.time(), 'status': 'running', 'performance_measurement': False} try: if args.staging_observer: if args.gdr_observer or args.installation is None or profile.get('runner') is not None: raise ValueError('staging gate requires only the installed public GPU consumer') if len({profile['nodes'][node]['hostname'] for node in profile['nodes']}) != 2 or len({profile['nodes'][node]['boot_id'] for node in profile['nodes']}) != 2: raise ValueError('staging gate requires two physical hosts') for case in profile['cases']: staging.validate_case(case) controller.fault_library = args.staging_observer.resolve() manifest['cuda_observer_sha256'] = hashlib.sha256(controller.fault_library.read_bytes()).hexdigest() if args.gdr_observer: if profile.get('runner') is not None or not all(case.get('cuda_memory') == 'direct' for case in profile['cases']): raise ValueError('GPU observer requires explicit public direct CUDA cases') gdr.validate_profile(profile) controller.fault_library = args.gdr_observer.resolve() manifest['gdr_observer_sha256'] = hashlib.sha256(controller.fault_library.read_bytes()).hexdigest() if profile.get('runner') == 'orbitte-installed-host': installed.validate_nodes(profile) if args.installation is None: raise ValueError('installed host requires its frozen installation manifest') installation = json.loads(args.installation.read_text()) if (installation.get('status') != 'passed' or installation.get('dirty') or installation.get('binary_sha256') != controller.sha256 or installation.get('protocol') != installed.PROTOCOL): raise ValueError('installation is dirty, incomplete or belongs to another binary') if any(case.get('mode') in ('installed-lost-completion', 'installed-cleanup-failure', 'installed-inflight-dma', 'installed-memory-cleanup', 'installed-rail-accounting') for case in profile['cases']): controller.fault_library = Path(installation['fault_library']) if hashlib.sha256(controller.fault_library.read_bytes()).hexdigest() != installation['fault_library_sha256']: raise ValueError('native fault library differs from the packaged artifact') if any(case.get('discovery') == 'auto' for case in profile['cases']): controller.inventory_tool = Path(installation['inventory_tool']) if hashlib.sha256(controller.inventory_tool.read_bytes()).hexdigest() != installation['inventory_tool_sha256']: raise ValueError('discovery inventory differs from the packaged artifact') manifest['installation'] = installation manifest['installation_manifest_sha256'] = hashlib.sha256(args.installation.read_bytes()).hexdigest() if controller.command([str(args.binary.resolve()), '--protocol']).strip() != installed.PROTOCOL: raise ValueError('unexpected installed binary protocol') elif args.installation is not None: installation = json.loads(args.installation.read_text()) if (not (args.gdr_observer or args.staging_observer) or installation.get('status') != 'passed' or installation.get('dirty') or installation.get('consumer') != 'gpu' or installation.get('binary_sha256') != controller.sha256 or installation.get('protocol') != 'orbitte-hardware-peer-v1' or (args.gdr_observer and installation.get('fault_library_sha256') != manifest['gdr_observer_sha256']) or (args.staging_observer and installation.get('cuda_observer_sha256') != manifest['cuda_observer_sha256'])): raise ValueError('GPU installation is dirty, incomplete or belongs to other inputs') manifest['installation'] = installation manifest['installation_manifest_sha256'] = hashlib.sha256(args.installation.read_bytes()).hexdigest() if controller.command([str(args.binary.resolve()), '--protocol']).strip() != installation['protocol']: raise ValueError('unexpected installed GPU binary protocol') native_test = native_peer_test(profile) if native_test is not None: listing = controller.command([str(args.binary.resolve()), native_test, '--exact', '--list']) validate_peer_listing(native_test, listing) manifest['native_test_selection'] = {'test': native_test, 'listing': listing, 'matches': 1} for node in sorted({case[role] for case in profile['cases'] for role in ['source', 'requester']}): controller.deploy(node) for index, case in enumerate(profile['cases']): controller.case(index, case) manifest['status'] = 'passed' except Exception as error: manifest.update(status='failed', error=str(error)) raise finally: manifest['all_supervised_children_reaped'] = controller.settle() if not manifest['all_supervised_children_reaped']: manifest['status'] = 'failed' manifest['finished_unix'] = time.time() (args.output / 'manifest.json').write_text(json.dumps(manifest, indent=2) + '\n') if manifest['status'] != 'passed': raise SystemExit('hardware evidence or cleanup did not pass') if __name__ == '__main__': main()