"""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 mixed_gpu 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 == 'orbite-multirail': return 'transport::rdma::cq_lane::tests::hardware::multirail::multirail_peer' if runner == 'orbite-connected': return 'transport::rdma::cq_lane::tests::hardware::connected::connected_endpoint_peer' if runner == 'orbite-shared-cq': return 'transport::rdma::cq_lane::tests::hardware::shared_cq_peer' if runner == 'orbite-window': return 'transport::rdma::cq_lane::tests::hardware::source_window_peer' if runner == 'orbitkv-peer-host': return 'peer::orbite::tests::rdma_host::registered_host_rdma_peer' if runner in (None, 'orbite-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/orbite-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 + '/liborbite_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 ('orbite-multirail', 'orbite-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('ORBITE_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 'ORBITE_RESTART_GENERATIONS' in environment: config.update(restart_role=environment['ORBITE_SHARED_ROLE'], restart_kill_role=environment['ORBITE_RESTART_KILL_ROLE'], restart_generations=int(environment['ORBITE_RESTART_GENERATIONS']), restart_live_peer=environment.get('ORBITE_HOST_MODE') in ('installed-restart', 'installed-interruption', 'installed-lost-completion', 'installed-cleanup-failure')) if environment.get('ORBITE_STAGING_OBSERVE') == '1': if self.fault_library is None: raise ValueError('staging gate requires its pinned CUDA observer') environment.update(LD_PRELOAD=self.remote + '/liborbite_cq_fault.so', ORBITE_IPC_LOG=root + '/cuda-staging.jsonl') config['fault_library_sha256'] = hashlib.sha256(self.fault_library.read_bytes()).hexdigest() if environment.get('ORBITE_GDR_OBSERVE') == '1': if self.fault_library is None: raise ValueError('GPU direct gate requires its pinned native observer') environment['LD_PRELOAD'] = self.remote + '/liborbite_cq_fault.so' config['fault_library_sha256'] = hashlib.sha256(self.fault_library.read_bytes()).hexdigest() if (environment.get('ORBITE_HOST_MODE') in ('installed-lost-completion', 'installed-cleanup-failure', 'installed-inflight-dma', 'installed-rail-accounting') and environment['ORBITE_SHARED_ROLE'] == 'requester') or ( environment.get('ORBITE_HOST_MODE') == 'installed-memory-cleanup' and environment['ORBITE_SHARED_ROLE'] == environment['ORBITE_CLEANUP_ROLE']): if self.fault_library is None: raise ValueError('native fault requires its pinned observer library') environment['LD_PRELOAD'] = self.remote + '/liborbite_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') == 'orbite-mixed-gpu': return self.run_case(index, case, lambda node, role, kind, root, readiness=None: mixed_gpu.command(self.profile, case, node, role, root, readiness)) if self.profile.get('runner') == 'orbite-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') == 'orbite-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') == 'orbite-window' shared = self.profile.get('runner') in ['orbite-shared-cq', 'orbite-window', 'orbite-connected'] if self.profile.get('runner') not in [None, 'orbitkv-peer-host', 'orbite-shared-cq', 'orbite-window', 'orbite-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') != 'orbite-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") != "orbite-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 = { 'ORBITE_SHARED_ROLE': role, 'ORBITE_SHARED_BIND': self.profile['nodes'][node]['bind'], 'ORBITE_SHARED_DEVICE': self.profile['rnic'], 'ORBITE_SHARED_GID': str(self.profile['gid_index']), 'ORBITE_SHARED_PAYLOAD': str(case['payload']), 'ORBITE_SHARED_SEED': str(case['seed']), 'ORBITE_SHARED_CQES': str(capacity), 'ORBITE_SHARED_WINDOW': str(window), 'ORBITE_SHARED_MAPS': root + '/maps.txt', } environment['ORBITE_RDMA_EVENTS'] = str(int(cq_events)) if shared_qp: environment["ORBITE_SHARED_QP"] = shared_qp environment["ORBITE_SHARED_OPERATION"] = case["operation"] if multi_context: environment['ORBITE_RDMA_MULTI_CONTEXT'] = '1' if discovery: environment['ORBITE_RDMA_DISCOVERY'] = '1' if case.get('duplex'): environment['ORBITE_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['ORBITE_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['ORBITE_WINDOW_PROBE'] = case['window_probe'] if readiness is not None: if len(readiness) != 1: raise ValueError('invalid shared CQ readiness') environment['ORBITE_SHARED_PEER'] = readiness[0] return arguments, environment if consumer: arguments = [native_peer_test(self.profile), '--exact', '--ignored', '--test-threads=1', '--nocapture'] environment = { 'ORBITKV_ORBITE_ROLE': role, 'ORBITKV_ORBITE_BIND': self.profile['nodes'][node]['bind'], 'ORBITKV_ORBITE_RDMA_DEVICE': self.profile['rnic'], 'ORBITKV_ORBITE_RDMA_PORT': str(self.profile.get('rdma_port', 1)), 'ORBITKV_ORBITE_RDMA_GID_INDEX': str(self.profile['gid_index']), 'ORBITKV_ORBITE_PAYLOAD': str(case['payload']), 'ORBITKV_ORBITE_SEED': str(case['seed']), 'ORBITKV_ORBITE_MAPS': root + '/maps.txt', } if readiness is not None: if len(readiness) != 4 or not readiness[0].startswith('orbite+rdma://'): raise ValueError('invalid OrbitKV RDMA readiness') environment['ORBITKV_ORBITE_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 = { 'ORBITE_HARDWARE_MAPS': root + '/maps.txt', 'ORBITE_RDMA_HOST_MEMORY': host_memory, 'ORBITE_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(ORBITE_SHARED_ROLE=role, ORBITE_STAGING_OBSERVE='1', ORBITE_STAGING_CHUNK_SIZE=str(staging.configured_chunk(case, role)), ORBITE_STAGING_CHUNK_COUNT=str(staging.configured_depth(case, role)), ORBITE_STAGING_PIPELINE_DEPTH=str(staging.configured_depth(case, role)), ORBITE_EXPECT_STAGE_DEPTH=str(staging.stage_width(case, role)), ORBITE_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(ORBITE_REMOTE_STAGING='1', ORBITE_EXPECT_QP_REUSE=str(int(case['transport'] == 'rdma')), ORBITE_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(ORBITE_RDMA_CUDA_MEMORY='direct', MLX5_SCATTER_TO_CQE='0', ORBITE_GDR_OBSERVE='1', ORBITE_SHARED_ROLE=role) reuse = case.get('gpu_source_reuse', False) environment['ORBITE_EXPECT_GPU_KEY_REUSE'] = str(int(reuse and role == 'source')) environment['ORBITE_EXPECT_QP_REUSE'] = str(int( (reuse or 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['ORBITE_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['ORBITE_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['ORBITE_RDMA_SLICE_BYTES'] = str(length) for key, variable, limit in [('rdma_block_size', 'ORBITE_RDMA_BLOCK_SIZE', 64 << 20), ('rdma_max_send_wr', 'ORBITE_RDMA_MAX_SEND_WR', 64), ('rdma_max_cqes', 'ORBITE_RDMA_MAX_CQES', 4096), ('rdma_max_endpoints', 'ORBITE_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['ORBITE_RDMA_EVENTS'] = str(int(cq_events)) if discovery: environment['ORBITE_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") != "orbite-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 = { 'ORBITE_SHARED_ROLE': role, 'ORBITE_SHARED_BIND': self.profile['nodes'][node]['bind'], 'ORBITE_SHARED_RAILS': ','.join(self.profile['rails']), 'ORBITE_SHARED_GID': str(self.profile['gid_index']), 'ORBITE_SHARED_MAPS': root + '/maps.txt', 'ORBITE_RDMA_EVENTS': str(int(case['rdma_events'])), 'ORBITE_RDMA_LANES': str(workers), 'ORBITE_RAIL_MODE': case['mode'], 'ORBITE_RAIL_OPERATION': case['operation'], 'ORBITE_RAIL_FAULT': str(case['fault_rail']), 'ORBITE_RAIL_MASK': str(case['mask']), 'ORBITE_RAIL_FAULT_ROLE': str(case.get('fault_role', 'source')), } if restart is not None: environment['ORBITE_RESTART_KILL_ROLE'] = restart[0] environment['ORBITE_RESTART_GENERATIONS'] = str(restart[1]) if readiness is not None: if len(readiness) != 1: raise ValueError('invalid multirail readiness') environment['ORBITE_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('ORBITE_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('ORBITE_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') in ('orbite-installed-host', 'orbite-mixed-gpu'): try: if self.profile['runner'] == 'orbite-mixed-gpu': mixed_gpu.verify(case, evidence) else: 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('--require-gpu-source-reuse', action='store_true', help='require the positive GPU key profile in every case') 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() if args.require_gpu_source_reuse and not args.gdr_observer: parser.error('--require-gpu-source-reuse requires --gdr-observer') 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, require_reuse=args.require_gpu_source_reuse) controller.fault_library = args.gdr_observer.resolve() manifest['gdr_observer_sha256'] = hashlib.sha256(controller.fault_library.read_bytes()).hexdigest() if profile.get('runner') in ('orbite-installed-host', 'orbite-mixed-gpu'): if profile['runner'] == 'orbite-mixed-gpu': mixed_gpu.validate(profile) protocol = mixed_gpu.PROTOCOL else: installed.validate_nodes(profile) protocol = installed.PROTOCOL if args.installation is None: raise ValueError('installed gate 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') != protocol): raise ValueError('installation is dirty, incomplete or belongs to another binary') if profile['runner'] == 'orbite-mixed-gpu': for name in ('mixed_gpu.py', 'run.py', 'supervise.py'): driver = Path(__file__).with_name(name) if hashlib.sha256(driver.read_bytes()).hexdigest() != installation['fixture_files_sha256']['tests/hardware/' + name]: raise ValueError('mixed GPU driver differs from its packaged source') 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() != 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') != 'orbite-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()