"""Bounded read-only native ownership evidence for the supported frozen runtime. Raw dependent-handle totals remain visible. Only exact live ArrayPool TLS roots with a same-T singleton CWT native-slot match can be classified as runtime-owned. This reader is used exclusively by the sustained lifecycle profile, never by first-paint or responsiveness performance measurements. """ import hashlib import json from pathlib import Path import re import subprocess import time MAX_SEGMENTS = 128 MAX_WORDS_PER_READ = 4096 MAX_TLS_ARRAY_LENGTH = 4096 MAX_THREADS = 256 MAX_READ_COMMANDS_PER_PASS = 4096 U64_MAX = (1 << 64) - 1 PINNED_CORELIB_SHA256 = '613426f7e28a74de30be95bc56dcba8991286ed396379a585ced38dbd57e493e' PINNED_COSMOS_ORIGINAL_SHA256 = '5d63cc0687931c69f0cb24740f02172e86b9e7664074d2f7b08fcf1a405c062e' def signed32(value): value &= 0xffffffff return value - (1 << 32) if value & (1 << 31) else value def checked_address(address, count=1): assert isinstance(address, int) and isinstance(count, int) assert 1 <= count <= MAX_WORDS_PER_READ assert 0xffff800000000000 <= address <= U64_MAX - count * 8 + 1 assert address % 8 == 0 return address def parse_words(text, address, count): checked_address(address, count) words = {} for line in text.splitlines(): row = re.match(r'^\s*(?:0x)?([0-9a-fA-F]{1,16}):\s*(.*)$', line) if row is None: continue first = int(row[1], 16) for index, value in enumerate(re.findall(r'0x([0-9a-fA-F]{1,16})\b', row[2])): at = first + index * 8 if address <= at < address + count * 8 and (at - address) % 8 == 0: parsed = int(value, 16) assert at not in words or words[at] == parsed, 'Conflicting repeated HMP word' words[at] = parsed assert len(words) == count, ('Incomplete bounded HMP read', hex(address), count, len(words), text[-200:]) return [words[address + index * 8] for index in range(count)] def decode_segment(address, words): assert address % 4096 == 0 and len(words) == 512 slots = [] unexpected = [] for index in range(170): target, dependent, tag = words[2 + index * 3:5 + index * 3] kind = tag & 0xffffffff if kind == 6: slots.append({'SlotAddress': address + 16 + index * 24, 'Target': target, 'Dependent': dependent, 'Type': kind, 'Padding': tag >> 32}) elif kind != 0xffffffff: unexpected.append({'SlotIndex': index, 'Type': kind}) # The actual Pack/Unpack implementation puts Alive in the low sixteen bits. # Its source comment incorrectly describes these two fields in reverse. return {'Address': address, 'Next': words[0], 'FreeHeadPacked': words[1], 'Alive': words[1] & 0xffff, 'FreeListHead': (words[1] >> 16) & 0xffff, 'Version': words[1] >> 32, 'Slots': slots, 'UnexpectedTypes': unexpected, 'AliveMatchesTypedSlots': (words[1] & 0xffff) == len(slots)} class _SnapshotRead: def __init__(self, qmp, folder, pass_name, symbols, layout): self.qmp = qmp self.layout = layout self.folder = folder self.pass_name = pass_name self.symbols = symbols self.reads = [] self.errors = [] self.headers = {} self.object_words = {} self.word_cache = {} self.start = time.monotonic() def read(self, address, count=1): checked_address(address, count) assert len(self.reads) < MAX_READ_COMMANDS_PER_PASS command = f'x/{count}gx {address:#x}' assert re.fullmatch(r'x/[1-9][0-9]*gx 0x[0-9a-f]+', command) dump = self.qmp('human-monitor-command', {'command-line': command}) assert isinstance(dump, str) self.reads.append({'Command': command, 'Address': address, 'Count': count, 'Raw': dump}) words = parse_words(dump, address, count) self.word_cache.update({address + index * 8: value for index, value in enumerate(words)}) return words def cached(self, address, count=1): checked_address(address, count) if all(address + index * 8 in self.word_cache for index in range(count)): return [self.word_cache[address + index * 8] for index in range(count)] return self.read(address, count) def many_cached(self, address, count): assert 0 <= count <= MAX_TLS_ARRAY_LENGTH * 4 result = [] while count: take = min(count, MAX_WORDS_PER_READ) result.extend(self.cached(address, take)) address += take * 8 count -= take return result def batch_words(self, addresses): pages = {} for address in set(addresses): checked_address(address) if address not in self.word_cache: pages.setdefault(address & ~4095, []).append(address) for addresses in pages.values(): first, last = min(addresses), max(addresses) self.cached(first, (last - first) // 8 + 1) def array(self, address, stride_words=1, maximum=MAX_TLS_ARRAY_LENGTH, used=None): checked_address(address, 2) header = self.cached(address, 2) length = header[1] & 0xffffffff assert 0 <= length <= maximum, ('Bounded managed array length', hex(address), length, maximum) count = length if used is None else used assert 0 <= count <= length return {'Address': address, 'HeaderWords': header, 'Length': length, 'Used': count, 'StrideWords': stride_words, 'DataWords': self.many_cached(address + 16, count * stride_words)} def runtime_pool_snapshot(self, slots): descriptors, pool_cells = {}, {} for address, names in self.symbols.items(): for name in names: prefix = '__TypeThreadStaticIndexS_P_CoreLib_System_Buffers_SharedArrayPool_1<' pool_prefix = '__GCSTATICSS_P_CoreLib_System_Buffers_ArrayPool_1<' if name.startswith(prefix) and name.endswith('>'): descriptors[name[len(prefix):-1]] = (address, name) if name.startswith(pool_prefix) and name.endswith('>'): pool_cells[name[len(pool_prefix):-1]] = (address, name) assert descriptors and set(descriptors) == set(pool_cells) and len(descriptors) <= 256 self.batch_words([address + offset for address, _ in descriptors.values() for offset in (0, 8)] + [address for address, _ in pool_cells.values()] + [self.layout['SchedulerGcCell'], self.layout['RuntimeThreadGcCell']]) descriptor_rows = [] managers = [] for generic, (address, name) in sorted(descriptors.items()): words = self.cached(address, 2) row = {'ClosedT': generic, 'Descriptor': address, 'DescriptorSymbol': name, 'DescriptorWords': words, 'Manager': words[0], 'TypeIndex': signed32(words[1] & 0xffffffff), 'PoolGcCell': pool_cells[generic][0], 'PoolGcCellSymbol': pool_cells[generic][1]} descriptor_rows.append(row) if row['Manager'] and row['TypeIndex'] >= 0: managers.append(row['Manager'] + 8) self.batch_words(managers) for row in descriptor_rows: row['ModuleIndex'] = signed32(self.cached(row['Manager'] + 8)[0] & 0xffffffff) if row['Manager'] and row['TypeIndex'] >= 0 else None scheduler_base = self.cached(self.layout['SchedulerGcCell'])[0] assert scheduler_base registry = self.cached(scheduler_base + 0x18)[0] registry_array = self.array(registry, maximum=MAX_THREADS) threads = [] self.batch_words([thread + offset for thread in registry_array['DataWords'] if thread for offset in (0x10, 0x78)]) for index, thread in enumerate(registry_array['DataWords']): if thread: threads.append({'RegistryIndex': index, 'ThreadAddress': thread, 'ThreadId': signed32(self.cached(thread + 0x78)[0] & 0xffffffff), 'TlsRoot': self.cached(thread + 0x10)[0]}) runtime_base = self.cached(self.layout['RuntimeThreadGcCell'])[0] bootstrap = self.cached(runtime_base + 8)[0] if runtime_base else 0 thread_roots = [dict(thread, RootKind='SchedulerThread') for thread in threads] thread_roots.append({'ThreadId': None, 'RootKind': 'BootstrapFallback', 'TlsRoot': bootstrap}) root_snapshots, owners = [], [] for thread in thread_roots: item = dict(thread) item['PoolFields'] = [] if thread['TlsRoot']: modules = self.array(thread['TlsRoot']) item['ModuleStorage'] = modules rows = {} for descriptor in descriptor_rows: module, index = descriptor['ModuleIndex'], descriptor['TypeIndex'] if module is None or module < 0 or index < 0 or module >= modules['Length']: continue row_address = modules['DataWords'][module] if not row_address: continue if module not in rows: rows[module] = self.array(row_address) row = rows[module] if index >= row['Length'] or not row['DataWords'][index]: continue type_base = row['DataWords'][index] type_words = self.cached(type_base, 2) target = type_words[1] field = {'ClosedT': descriptor['ClosedT'], 'ModuleIndex': module, 'TypeIndex': index, 'TypeBase': type_base, 'TypeBaseHeaderAndField': type_words, 'TlsFieldAddress': type_base + 8, 'Target': target} item['PoolFields'].append(field) if target: owners.append(dict(field, ThreadId=thread['ThreadId'], RootKind=thread['RootKind'], ThreadAddress=thread.get('ThreadAddress'), RegistryIndex=thread.get('RegistryIndex'))) item['TypeStorageRows'] = rows root_snapshots.append(item) slot_map = {slot['SlotAddress']: slot for slot in slots} pool_registries = [] reverse = [] for descriptor in descriptor_rows: item = {'ClosedT': descriptor['ClosedT'], 'GcCell': descriptor['PoolGcCell'], 'GcCellSymbol': descriptor['PoolGcCellSymbol']} blob = self.cached(item['GcCell'])[0] item['GcStaticBlob'] = blob if blob: singleton = self.cached(blob + 8)[0] item['PoolSingleton'] = singleton if singleton: cwt = self.cached(singleton + 8)[0] item['AllTlsBucketsCwt'] = cwt if cwt: container = self.cached(cwt + 0x10)[0] item['CwtContainer'] = container if container: entries = self.cached(container + 0x18)[0] first_free = signed32(self.cached(container + 0x28)[0] & 0xffffffff) item['FirstFreeEntry'] = first_free assert first_free >= 0 array = self.array(entries, stride_words=2, used=first_free) item['CwtEntries'] = array decoded = [] for index in range(first_free): packed, handle = array['DataWords'][index * 2:index * 2 + 2] entry = {'Index': index, 'HashCode': signed32(packed & 0xffffffff), 'Next': signed32(packed >> 32), 'DependentHandle': handle} if handle in slot_map: slot = slot_map[handle] entry.update({'SlotTarget': slot['Target'], 'SlotDependent': slot['Dependent']}) reverse.append(dict(entry, ClosedT=descriptor['ClosedT'])) decoded.append(entry) item['DecodedUsedEntries'] = decoded pool_registries.append(item) tls_slots = [slot for slot in slots if slot['TargetObject'].get('EEType') == self.layout['TlsArrayEEType']] bodies = [] for slot in tls_slots: body = self.array(slot['Target'], stride_words=2, maximum=27) assert body['Length'] == 27 buckets = [{'Index': index, 'PooledArray': body['DataWords'][index * 2], 'TimestampAndPadding': body['DataWords'][index * 2 + 1]} for index in range(27)] self.preload_object_headers([bucket['PooledArray'] for bucket in buckets]) for bucket in buckets: bucket['PooledArrayObject'] = self.describe_object(bucket['PooledArray']) bodies.append({'SlotAddress': slot['SlotAddress'], 'TlsArray': slot['Target'], 'Body': body, 'Buckets': buckets, 'ThreadOwners': [owner for owner in owners if owner['Target'] == slot['Target']], 'PoolRegistryOwners': [owner for owner in reverse if owner['DependentHandle'] == slot['SlotAddress']]}) return {'DescriptorCount': len(descriptor_rows), 'Descriptors': descriptor_rows, 'SchedulerBase': scheduler_base, 'SchedulerRegistry': registry_array, 'RegisteredThreads': threads, 'RuntimeThreadBase': runtime_base, 'BootstrapFallbackStorage': bootstrap, 'ThreadRootSnapshots': root_snapshots, 'PoolRegistries': pool_registries, 'TlsArrayBodies': bodies} def preload_object_headers(self, addresses): pages = {} for address in set(addresses): if not address or address in self.object_words: continue try: checked_address(address) pages.setdefault(address & ~4095, []).append(address) except AssertionError: pass # describe_object records the bounded failure later. for page, objects in sorted(pages.items()): first, last = min(objects), max(objects) # Batch only the interval within one mapped object page, never # speculative intervening pages; at most512words/4096bytes. try: words = self.read(first, (last - first) // 8 + 1) for address in objects: self.object_words[address] = words[(address - first) // 8] except Exception: pass # An individual bounded retry is recorded below. def describe_object(self, address): if address == 0: return {'ObjectAddress': 0, 'Null': True} if address in self.headers: return self.headers[address] try: word = self.object_words[address] if address in self.object_words else self.read(address)[0] mt = word & ~1 result = {'ObjectAddress': address, 'HeaderWord': word, 'EEType': mt, 'EETypeSymbols': self.symbols.get(mt, []), 'ResolvedExactSymbol': mt in self.symbols} except Exception as error: result = {'ObjectAddress': address, 'ReadError': str(error)} self.errors.append(result) self.headers[address] = result return result def run(self): before = self.read(self.layout['DependentStore'], 3) head, tail, committed = before assert head and tail and committed % 4096 == 0 assert 4096 <= committed <= MAX_SEGMENTS * 4096 segments, seen, current = [], set(), head while current: assert len(segments) < MAX_SEGMENTS and current not in seen, 'Overlong or cyclic depstore chain' checked_address(current, 512) assert current % 4096 == 0 seen.add(current) segment = decode_segment(current, self.read(current, 512)) segment['RawWords'] = self.cached(current, 512) segments.append(segment) current = segment['Next'] assert segments[-1]['Address'] == tail and len(segments) * 4096 == committed slots = [slot for segment in segments for slot in segment['Slots']] self.preload_object_headers([address for slot in slots for address in (slot['Target'], slot['Dependent'])]) for slot in slots: slot['TargetObject'] = self.describe_object(slot['Target']) slot['DependentObject'] = self.describe_object(slot['Dependent']) slot['ClearedTombstone'] = slot['Target'] == 0 and slot['Dependent'] == 0 runtime = self.runtime_pool_snapshot(slots) after = self.read(self.layout['DependentStore'], 3) return {'ReadOnly': True, 'StoreBefore': before, 'StoreAfter': after, 'SupportedTlsArrayEEType': self.layout['TlsArrayEEType'], 'StoreStableWithinPass': before == after, 'Segments': segments, 'AllocatedTypedSlots': len(slots), 'AliveCounterSum': sum(segment['Alive'] for segment in segments), 'AliveCountersMatchTypedSlots': all(segment['AliveMatchesTypedSlots'] and not segment['UnexpectedTypes'] for segment in segments), 'Slots': slots, 'RuntimeTlsPools': runtime, 'ObjectHeaderReads': self.headers, 'ObjectReadErrors': self.errors, 'ReadCommandCount': len(self.reads), 'HostElapsedSeconds': time.monotonic() - self.start} def save_reads(self): target = self.folder / (self.pass_name + '-Raw.txt') target.write_text(''.join(f"READ {index + 1}: {item['Command']}\n{item['Raw']}\n" for index, item in enumerate(self.reads)), encoding='utf-8') return {'Path': str(target), 'Sha256': hashlib.sha256(target.read_bytes()).hexdigest(), 'ReadCommandCount': len(self.reads), 'MaxWordsInAnyRead': max((item['Count'] for item in self.reads), default=0)} def parse_symbols(text): symbols = {} for line in text.splitlines(): match = re.match(r'^([0-9a-fA-F]{16})\s+[A-Za-z]\s+(.+)$', line) if match: symbols.setdefault(int(match[1], 16), []).append(match[2]) return symbols def symbol_address(symbols, name): matches = [address for address, names in symbols.items() if name in names] assert len(matches) == 1, ('Missing or ambiguous supported native symbol', name) return matches[0] def require_layout_shapes(disassembly, gc_static): """Reject unknown machine-code field shapes, never infer a new ABI layout.""" checks = { 'DependentStoreB0': bool(re.search(r'addq\s+\$0xb0,', disassembly['Count'])) and 'GCHandleSegmentStore__get_Count>' in disassembly['Count'] and bool(re.search(r'#\s+0x' + format(gc_static, 'x') + r'\b', disassembly['Count'])), 'PoolTlsField8': '__GetThreadStaticBase_' in disassembly['Pool'] and bool(re.search(r'leaq\s+0x8\(%\w+\)', disassembly['Pool'])) and bool(re.search(r'movq\s+0x8\(%\w+\)', disassembly['Pool'])) and 'ConditionalWeakTable_2__Add>' in disassembly['Pool'], 'CwtContainer10': bool(re.search(r'0x10\(%\w+\)', disassembly['CwtAdd'])), 'CwtEntries18Count28Stride16Handle8': all(token in disassembly['CwtEntry'] for token in ('0x18(', '0x28(', 'shlq\t$0x4', 'leaq\t0x10(', 'RhHandleAllocDependent>')) and bool(re.search(r'movq\s+%\w+, 0x8\(%\w+\)', disassembly['CwtEntry'])), 'ThreadTls10': bool(re.search(r'movq\s+0x10\(%\w+\), %\w+', disassembly['ThreadTls'])), 'ThreadId78': bool(re.search(r'movl\s+0x78\(%\w+\), %\w+', disassembly['ThreadId'])), 'SchedulerRegistry18': '__GCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Scheduler_SchedulerManager>' in disassembly['Registry'] and bool(re.search(r'movq\s+0x18\(%\w+\)', disassembly['Registry'])), } assert all(checks.values()), ('Unsupported native ownership field layout', checks) return checks def ownership_fingerprint(value): return {'Store': value['StoreBefore'], 'Segments': [segment['RawWords'] for segment in value['Segments']], 'RuntimeTlsPools': value['RuntimeTlsPools'], 'ObjectHeaders': {int(key): item.get('HeaderWord') for key, item in value['ObjectHeaderReads'].items()}} class NativeDependentSlotReader: """Supported-runtime reader; every guest command is bounded read-only HMP x/.""" def __init__(self, qmp, elf, expected_kernel_sha, folder, corelib, gc_patch_record, llvm_directory=None): self.qmp, self.elf, self.folder = qmp, Path(elf), Path(folder) self.folder.mkdir(parents=True, exist_ok=True) assert hashlib.sha256(self.elf.read_bytes()).hexdigest() == expected_kernel_sha assert hashlib.sha256(Path(corelib).read_bytes()).hexdigest() == PINNED_CORELIB_SHA256, 'Unsupported CoreLib10.0.12 image' patch_path = Path(gc_patch_record['Path']) patch_raw = patch_path.read_bytes() assert hashlib.sha256(patch_raw).hexdigest() == gc_patch_record['Sha256'] patch = json.loads(patch_raw.decode('utf-8-sig')) assert patch['PinnedVersion'] == '3.0.94' and patch['OriginalPinnedSha256'] == PINNED_COSMOS_ORIGINAL_SHA256 assert patch['AfterSha256'] == gc_patch_record['RuntimeSha256'] assert patch['DependentHandleCountReadOnlyExport'] is True and patch['GlobalPackageCacheModified'] is False llvm = Path(llvm_directory) if llvm_directory else Path(__file__).resolve().parents[1] / '.tools/llvm/bin' nm = subprocess.run([str(llvm / 'llvm-nm.exe'), '--defined-only', '--numeric-sort', str(self.elf)], check=True, capture_output=True, text=True, encoding='utf-8', timeout=30).stdout self.symbols = parse_symbols(nm) gc_static = symbol_address(self.symbols, '__NONGCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Memory_GarbageCollector_GarbageCollector') self.layout = {'DependentStore': gc_static + 0xb0, 'SchedulerGcCell': symbol_address(self.symbols, '__GCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Scheduler_SchedulerManager'), 'RuntimeThreadGcCell': symbol_address(self.symbols, '__GCSTATICSCosmos_Kernel_Core_Cosmos_Kernel_Core_Runtime_Thread'), 'TlsArrayEEType': symbol_address(self.symbols, '_ZTV67__Array')} methods = { 'Count': 'XfeRuntimeDependentHandleCount', 'Pool': 'S_P_CoreLib_System_Buffers_SharedArrayPool_1__InitializeTlsBucketsAndTrimming', 'CwtAdd': 'S_P_CoreLib_System_Runtime_CompilerServices_ConditionalWeakTable_2__Add', 'CwtEntry': 'S_P_CoreLib_System_Runtime_CompilerServices_ConditionalWeakTable_2_Container__CreateEntryNoResize', 'ThreadTls': 'RhGetThreadStaticStorage', 'ThreadId': 'XfeRuntimeCurrentSchedulerThreadId', 'Registry': 'Cosmos_Kernel_Core_Cosmos_Kernel_Core_Scheduler_SchedulerManager__RegisterThread'} disassembly = {} for label, name in methods.items(): symbol_address(self.symbols, name) disassembly[label] = subprocess.run([str(llvm / 'llvm-objdump.exe'), '--no-show-raw-insn', '-d', '--disassemble-symbols=' + name, str(self.elf)], check=True, capture_output=True, text=True, encoding='utf-8', timeout=30).stdout checks = require_layout_shapes(disassembly, gc_static) dump = self.folder / 'SupportedLayoutDisassembly.txt' dump.write_text('\n'.join(disassembly.values()), encoding='utf-8') self.provenance = {'Kind': 'Cosmos3.0.94-CoreLib10.0.12-NativeOwnershipLayoutV1', 'KernelSha256': expected_kernel_sha, 'CoreLibSha256': PINNED_CORELIB_SHA256, 'GcPatchReceiptSha256': gc_patch_record['Sha256'], 'GcRuntimeSha256': patch['AfterSha256'], 'SymbolTableSha256': hashlib.sha256(nm.encode()).hexdigest(), 'ResolvedLayout': self.layout, 'MachineCodeShapeChecks': checks, 'DisassemblyPath': str(dump.resolve()), 'DisassemblySha256': hashlib.sha256(dump.read_bytes()).hexdigest(), 'OnlyGuestCommandsAdded': 'human-monitor-command x/<1..4096>gx ', 'MaxSegments': MAX_SEGMENTS, 'MaxThreads': MAX_THREADS, 'MaxArrayLength': MAX_TLS_ARRAY_LENGTH, 'VmPaused': False, 'ForcedGc': False, 'InjectedCodeOrGuestWrites': False, 'PerformanceEvidenceClaimed': False} (self.folder / 'LayoutProvenance.json').write_text(json.dumps(self.provenance, indent=2), encoding='utf-8') def capture(self, cycle, observation): folder = self.folder / f'Closed-{cycle:02d}' folder.mkdir() passes, files = [], [] for name in ('A', 'B'): reader = _SnapshotRead(self.qmp, folder, name, self.symbols, self.layout) try: value = reader.run() target = folder / (name + '.json') target.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding='utf-8') files.append({'Path': str(target.resolve()), 'Sha256': hashlib.sha256(target.read_bytes()).hexdigest()}) passes.append(value) finally: raw = reader.save_reads() files.append(raw) first, second = passes stable = ownership_fingerprint(first) == ownership_fingerprint(second) valid = stable and all(value['StoreStableWithinPass'] and value['AliveCountersMatchTypedSlots'] and not value['ObjectReadErrors'] and value['AllocatedTypedSlots'] == observation['AllocatedDependentSlots'] and value['AliveCounterSum'] == observation['AllocatedDependentSlots'] for value in passes) record = {'Cycle': cycle, 'StableRequiredFieldsInTwoPasses': stable, 'CountsAgreeWithClosedPair': all(value['AllocatedTypedSlots'] == observation['AllocatedDependentSlots'] for value in passes), 'QualifiedNativeRead': valid, 'EvidenceFiles': files, 'ComparisonScope': 'Complete depstore headers/slots, descriptors, live-thread root chains, TLS bodies and exact pool CWT entries; unrelated batched-read gap bytes excluded.'} (folder / 'Capture.json').write_text(json.dumps(record, indent=2), encoding='utf-8') assert valid, ('Unstable or inconsistent closed native ownership capture', record) return second, record def _root_identity(root): return (root['RootKind'], root.get('ThreadAddress'), root.get('ThreadId'), root.get('RegistryIndex')) def _row(root, module): return root.get('TypeStorageRows', {}).get(module, root.get('TypeStorageRows', {}).get(str(module))) def _pool_field(root, descriptor): module, index = descriptor['ModuleIndex'], descriptor['TypeIndex'] if module is None or module < 0 or index < 0: return {'TypeBase': 0, 'Target': 0} row = _row(root, module) if row is None or index >= row['Length'] or row['DataWords'][index] == 0: return {'TypeBase': 0, 'Target': 0} fields = [field for field in root['PoolFields'] if field['ClosedT'] == descriptor['ClosedT']] assert len(fields) == 1 and fields[0]['TypeBase'] == row['DataWords'][index] return fields[0] class ClosedCycleOwnershipTracker: """Fixed non-TLS baseline with exact first-use runtime TLS ownership checks.""" def __init__(self): self.previous = None self.baseline_non_tls = None self.baseline_tombstones = None self.known_tls = {} def qualify(self, value): runtime = value['RuntimeTlsPools'] descriptors = {entry['ClosedT']: entry for entry in runtime['Descriptors']} assert descriptors and len(descriptors) == runtime['DescriptorCount'] roots = {_root_identity(root): root for root in runtime['ThreadRootSnapshots']} assert len(roots) == len(runtime['ThreadRootSnapshots']), 'Duplicate live-thread identity' slots = {slot['SlotAddress']: slot for slot in value['Slots']} assert len(slots) == value['AllocatedTypedSlots'] == value['AliveCounterSum'] current = {} for body in runtime['TlsArrayBodies']: slot = slots[body['SlotAddress']] assert slot['TargetObject'].get('EEType') == value['SupportedTlsArrayEEType'], 'Fake TLS object type' assert slot['Target'] == body['TlsArray'] and slot['Target'] and slot['Dependent'] == 0 assert not slot['ClearedTombstone'] and body['Body']['Length'] == 27 assert len(body['ThreadOwners']) == len(body['PoolRegistryOwners']) == 1, 'TLS owner missing or aliased' owner, pool = body['ThreadOwners'][0], body['PoolRegistryOwners'][0] assert owner['ClosedT'] in descriptors and owner['ClosedT'] == pool['ClosedT'], 'Wrong closed-T pool ownership' identity = _root_identity(owner) assert identity in roots, 'TLS owner is not a fixed live thread/bootstrap root' descriptor = descriptors[owner['ClosedT']] field = _pool_field(roots[identity], descriptor) assert field['Target'] == owner['Target'] == slot['Target'] assert field['TypeBase'] == owner['TypeBase'] assert descriptor['ModuleIndex'] == owner['ModuleIndex'] and descriptor['TypeIndex'] == owner['TypeIndex'] assert pool['DependentHandle'] == slot['SlotAddress'] and pool['SlotTarget'] == slot['Target'] and pool['SlotDependent'] == 0 assert pool['HashCode'] >= 0, 'Free CWT entry cannot establish runtime ownership' key = (owner['ClosedT'], *identity) assert key not in current, 'More than one TLS array for the same live T/thread' current[key] = {'SlotAddress': slot['SlotAddress'], 'Target': slot['Target'], 'TypeBase': owner['TypeBase'], 'ClosedT': owner['ClosedT'], 'RootIdentity': identity} tls_addresses = {entry['SlotAddress'] for entry in current.values()} assert len(tls_addresses) == len(current), 'One runtime TLS slot is aliased by multiple owners' assert len({entry['Target'] for entry in current.values()}) == len(current), 'A runtime TLS target is aliased by multiple owners' non_tls = {address: (slot['Target'], slot['Dependent']) for address, slot in slots.items() if address not in tls_addresses} tombstones = {address for address, slot in slots.items() if slot['ClearedTombstone']} initialized = [] if self.previous is None: self.baseline_non_tls = non_tls self.baseline_tombstones = tombstones else: old_runtime = self.previous['RuntimeTlsPools'] old_descriptors = {entry['ClosedT']: entry for entry in old_runtime['Descriptors']} assert descriptors == old_descriptors, 'Compiled native pool descriptors changed' old_roots = {_root_identity(root): root for root in old_runtime['ThreadRootSnapshots']} assert set(roots) == set(old_roots), 'A fixed live thread exited, appeared or changed identity' assert non_tls == self.baseline_non_tls, 'Unexplained non-TLS native slot, deletion or pair change' assert tombstones == self.baseline_tombstones, 'A native tombstone appeared or changed' for key, previous in self.known_tls.items(): assert current.get(key) == previous, 'Existing TLS root/slot was lost or replaced' for identity, root in roots.items(): old = old_roots[identity] old_modules, modules = old.get('ModuleStorage'), root.get('ModuleStorage') if old_modules: assert modules and modules['Length'] >= old_modules['Length'], 'TLS module storage shrank or disappeared' for module, old_address in enumerate(old_modules['DataWords']): new_address = modules['DataWords'][module] if old_address != new_address and old_address: old_row, new_row = _row(old, module), _row(root, module) assert old_row and new_row and new_row['Length'] >= old_row['Length'] assert new_row['DataWords'][:old_row['Length']] == old_row['DataWords'], 'TLS row resize did not preserve every prior field' for generic, descriptor in descriptors.items(): before, after = _pool_field(old, descriptor), _pool_field(root, descriptor) if before['TypeBase']: assert after['TypeBase'] == before['TypeBase'], 'Existing compiled TLS type-base was replaced' if before['Target']: assert after['Target'] == before['Target'], 'Existing compiled TLS field was replaced or cleared' for key, entry in current.items(): if key in self.known_tls: continue identity = tuple(entry['RootIdentity']) before = _pool_field(old_roots[identity], descriptors[entry['ClosedT']]) assert before['Target'] == 0, 'Additional TLS slot aliases an already initialized runtime root' initialized.append({'ClosedT': entry['ClosedT'], 'RootIdentity': identity, 'PreviousTypeBase': before['TypeBase'], 'PreviousTarget': 0, 'NewTypeBase': entry['TypeBase'], 'NewTarget': entry['Target'], 'ExactNativeSlot': entry['SlotAddress']}) self.known_tls.update(current) self.previous = value return {'Passed': True, 'AllocatedGlobalDependentSlots': value['AllocatedTypedSlots'], 'ExcludedProvenLiveRuntimeTlsSlots': len(current), 'FixedNonTlsSlots': len(non_tls), 'FixedTombstones': len(tombstones), 'ProvenFirstUseRuntimeTlsInitializations': initialized, 'AllowedUnknownGrowth': 0, 'WarmBaselineRolled': False}