Spaces:
Running on Zero
Running on Zero
| """Experimental Phase32A selective-Unicode CUDA classifier. | |
| Derived from frozen phase29_gpu.py; requires separately generated frozen | |
| Unicode tables and a new cubin exporting range_selective_unicode_classify. | |
| No production/default integration is included. ASCII controls stay frozen. | |
| Normalizes eligible rows completely; every unsupported row is CPU-owned. | |
| The current native CPU engine and all nine compact fact fields are retained. | |
| """ | |
| from __future__ import annotations | |
| from collections import deque | |
| from concurrent.futures import ThreadPoolExecutor | |
| from dataclasses import dataclass | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| import subprocess | |
| import tempfile | |
| import time | |
| from typing import Callable | |
| import numpy as np | |
| _SOURCE_SHA256_AT_IMPORT = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() | |
| MAX_CANDIDATES = 128 | |
| MAX_CONJUNCTIONS = 32 | |
| MAX_DFA_BYTES = 32 * 1024 * 1024 | |
| FACT_NAMES = ( | |
| "row_index", "article_status", "sp500", "nasdaq100", "russell2000", | |
| "market_wide", "observation_count", "native_candidate", "exact_candidate", | |
| ) | |
| CUDA_SOURCE = r''' | |
| #define MAX_CANDIDATES 128 | |
| #define MAX_CONJUNCTIONS 32 | |
| struct Candidate { int kind, index, start, end; }; | |
| __device__ __forceinline__ bool token(unsigned char c) { | |
| return (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_'; | |
| } | |
| __device__ __forceinline__ bool exact_token(unsigned char c) { | |
| return token(c) || (c >= 'A' && c <= 'Z'); | |
| } | |
| __device__ __forceinline__ unsigned char lower(unsigned char c) { | |
| return c >= 'A' && c <= 'Z' ? c + 32 : c; | |
| } | |
| __device__ __forceinline__ unsigned char normalize(unsigned char c) { | |
| c = lower(c); | |
| if (c == '-' || c == '.' || c == ',' || c == ':' || c == ';' || | |
| c == '(' || c == ')' || c == '[' || c == ']' || c == '{' || | |
| c == '}' || c == '/') return ' '; | |
| return c == '`' ? '\'' : c; | |
| } | |
| __device__ __forceinline__ bool space(unsigned char c) { | |
| return c == ' ' || (c >= 9 && c <= 13) || (c >= 0x1c && c <= 0x1f); | |
| } | |
| // All candidate identities are unique before the fixed-size buffer is used. | |
| // An overflow transfers the whole row to CPU; no partial result is retained. | |
| __device__ bool append_hit( | |
| int key, int start, int end, const int* key_offsets, | |
| const int* key_consumers, const int* consumers, | |
| Candidate* candidates, int* count, unsigned char* conjunctions | |
| ) { | |
| for (int i = key_offsets[key]; i < key_offsets[key + 1]; ++i) { | |
| const int* spec = consumers + key_consumers[i] * 4; | |
| if (spec[0] == 2) { | |
| conjunctions[spec[1]] |= (unsigned char)(1u << spec[2]); | |
| continue; | |
| } | |
| bool duplicate = false; | |
| for (int j = 0; j < *count; ++j) { | |
| Candidate prior = candidates[j]; | |
| if (prior.kind == spec[0] && prior.index == spec[1] && | |
| prior.start == start && prior.end == end) { | |
| duplicate = true; break; | |
| } | |
| } | |
| if (duplicate) continue; | |
| if (*count == MAX_CANDIDATES) return false; | |
| candidates[*count] = Candidate{spec[0], spec[1], start, end}; | |
| ++*count; | |
| } | |
| return true; | |
| } | |
| __device__ int lower_bound_offset(const unsigned int* offsets, int n, int value) { | |
| int left = 0, right = n; | |
| while (left < right) { | |
| int middle = left + (right - left) / 2; | |
| if (offsets[middle] < (unsigned int)value) left = middle + 1; | |
| else right = middle; | |
| } | |
| return left; | |
| } | |
| __device__ bool contains_term( | |
| const unsigned char* text, int left, int right, int term, | |
| const unsigned char* term_bytes, const int* term_offsets | |
| ) { | |
| int first = term_offsets[term], length = term_offsets[term + 1] - first; | |
| if (!length || right - left < length) return false; | |
| for (int start = left; start + length <= right; ++start) { | |
| if (start > left && token(text[start - 1])) continue; | |
| if (start + length < right && token(text[start + length])) continue; | |
| int i = 0; | |
| while (i < length && text[start + i] == term_bytes[first + i]) ++i; | |
| if (i == length) return true; | |
| } | |
| return false; | |
| } | |
| __device__ void context_bounds( | |
| const unsigned char* normalized, const unsigned int* offsets, int norm_length, | |
| int source_length, int start, int end, int width, int* left, int* right | |
| ) { | |
| int source_left = start > width ? start - width : 0; | |
| int source_right = width > source_length - end ? source_length : end + width; | |
| *left = lower_bound_offset(offsets, norm_length, source_left); | |
| *right = lower_bound_offset(offsets, norm_length, source_right); | |
| // Match Phase24's borrowed-window treatment of collapsed leading spaces. | |
| if (*left < *right && normalized[*left] == ' ') ++*left; | |
| } | |
| __device__ bool alias_context_ok( | |
| int context, const int* contexts, const int* clauses, const int* term_ids, | |
| const unsigned char* term_bytes, const int* term_offsets, | |
| const unsigned char* normalized, const unsigned int* offsets, int norm_length, | |
| int source_length, int start, int end | |
| ) { | |
| if (context < 0) return true; | |
| const int* spec = contexts + context * 5; | |
| int left, right; | |
| context_bounds(normalized, offsets, norm_length, source_length, | |
| start, end, spec[0], &left, &right); | |
| for (int i = 0; i < spec[4]; ++i) | |
| if (contains_term(normalized, left, right, term_ids[spec[3] + i], | |
| term_bytes, term_offsets)) return false; | |
| for (int i = 0; i < spec[2]; ++i) { | |
| const int* clause = clauses + 2 * (spec[1] + i); | |
| bool accepted = true; | |
| for (int j = 0; j < clause[1]; ++j) { | |
| if (!contains_term(normalized, left, right, term_ids[clause[0] + j], | |
| term_bytes, term_offsets)) { | |
| accepted = false; break; | |
| } | |
| } | |
| if (accepted) return true; | |
| } | |
| return false; | |
| } | |
| // Strict UTF-8 scalar decoder. Zero width rejects the entire row. | |
| __device__ __forceinline__ int phase32a_decode( | |
| const unsigned char* raw, int length, int position, unsigned int* scalar | |
| ) { | |
| unsigned int first = raw[position]; | |
| if (first < 0x80) { *scalar = first; return 1; } | |
| int width; | |
| unsigned int value, minimum; | |
| if (first >= 0xc2 && first <= 0xdf) { width = 2; value = first & 31; minimum = 0x80; } | |
| else if (first >= 0xe0 && first <= 0xef) { width = 3; value = first & 15; minimum = 0x800; } | |
| else if (first >= 0xf0 && first <= 0xf4) { width = 4; value = first & 7; minimum = 0x10000; } | |
| else return 0; | |
| if (width > length - position) return 0; | |
| for (int index = 1; index < width; ++index) { | |
| unsigned int next = raw[position + index]; | |
| if ((next & 0xc0) != 0x80) return 0; | |
| value = (value << 6) | (next & 63); | |
| } | |
| if (value < minimum || value > 0x10ffff || (value >= 0xd800 && value <= 0xdfff)) return 0; | |
| *scalar = value; | |
| return width; | |
| } | |
| extern "C" __global__ void range_selective_unicode_classify( | |
| const unsigned char* data, const long long* row_offsets, | |
| const long long* timestamps, int rows, int max_row_bytes, | |
| const unsigned int* transitions, const unsigned short* classes, | |
| const int* dfa_meta, const int* output_offsets, const int* output_keys, | |
| const int* key_lengths, const int* key_offsets, const int* key_consumers, | |
| const int* consumers, const long long* aliases, const int* lexical, | |
| const int* contexts, const int* clauses, const int* term_ids, | |
| const unsigned char* term_bytes, const int* term_offsets, | |
| const unsigned short* unicode_directory, const unsigned int* unicode_pages, int conjunction_count, | |
| unsigned char* normalized_data, unsigned int* source_offsets, | |
| Candidate* all_candidates, unsigned char* codes, unsigned int* facts | |
| ) { | |
| int row = blockIdx.x * blockDim.x + threadIdx.x; | |
| if (row >= rows) return; | |
| unsigned int* fact = facts + (long long)row * 8; | |
| for (int i = 0; i < 8; ++i) fact[i] = 0; | |
| codes[row] = 0; // 2=subset reject, 3=candidate overflow, 4=oversize, 5=invalid UTF-8, 6=normalization capacity. | |
| long long base = row_offsets[row], raw_length = row_offsets[row + 1] - base; | |
| if (raw_length > max_row_bytes) { codes[row] = 4; return; } | |
| int length = (int)raw_length; | |
| const unsigned char* raw = data + base; | |
| unsigned char* normalized = normalized_data + base; | |
| unsigned int* offsets = source_offsets + base; | |
| int norm_length = 0, source_length = 0; | |
| bool pending_space = false; | |
| for (int i = 0; i < length;) { | |
| unsigned int scalar; | |
| int width = phase32a_decode(raw, length, i, &scalar); | |
| if (!width) { codes[row] = 5; return; } | |
| unsigned int packed; | |
| if (scalar < 128) packed = 2u | ((unsigned int)normalize((unsigned char)scalar) << 3); | |
| else { | |
| unsigned int page = unicode_directory[scalar >> 8]; | |
| packed = unicode_pages[page * 256 + (scalar & 255)]; | |
| if (!(packed & 7)) { codes[row] = 2; return; } | |
| } | |
| int unit_count = (packed & 7) - 1; | |
| // The table proves unit_count <= source UTF-8 width. Guard corruption | |
| // or a future predicate change; never write outside this row's arena. | |
| if (unit_count > width || unit_count > 4) { codes[row] = 6; return; } | |
| for (int unit = 0; unit < unit_count; ++unit) { | |
| unsigned char c = (packed >> (3 + 7 * unit)) & 127; | |
| if (space(c)) { pending_space = norm_length != 0; continue; } | |
| if (pending_space) { | |
| if (norm_length >= length) { codes[row] = 6; return; } | |
| normalized[norm_length] = ' '; offsets[norm_length++] = source_length; | |
| pending_space = false; | |
| } | |
| if (norm_length >= length) { codes[row] = 6; return; } | |
| normalized[norm_length] = c; offsets[norm_length++] = source_length; | |
| } | |
| i += width; | |
| ++source_length; | |
| } | |
| Candidate* candidates = all_candidates + (long long)row * MAX_CANDIDATES; | |
| int count = 0; | |
| unsigned char conjunctions[MAX_CONJUNCTIONS] = {0}; | |
| bool native_candidate = false, exact_candidate = false; | |
| // Stream 0: normalized. Stream 1: raw case-sensitive. Stream 2: raw ASCII fold. | |
| for (int stream = 0; stream < 3; ++stream) { | |
| const int* meta = dfa_meta + stream * 4; | |
| if (meta[3] == 0) continue; | |
| const unsigned char* text = stream == 0 ? normalized : raw; | |
| int n = stream == 0 ? norm_length : length; | |
| unsigned int state = 0; | |
| int raw_source_position = -1; | |
| for (int pos = 0; pos < n; ++pos) { | |
| // Validated UTF-8; every lead/ASCII byte starts one original scalar. | |
| if (stream != 0 && (text[pos] & 0xc0) != 0x80) ++raw_source_position; | |
| unsigned char c = stream == 2 ? lower(text[pos]) : text[pos]; | |
| state = transitions[meta[0] + state * meta[1] + classes[stream * 256 + c]]; | |
| int first = output_offsets[meta[2] + state]; | |
| int last = output_offsets[meta[2] + state + 1]; | |
| for (int out = first; out < last; ++out) { | |
| int key = output_keys[out]; | |
| int start = pos + 1 - key_lengths[key], end = pos + 1; | |
| bool left_ok = start == 0 || !(stream == 0 ? token(text[start - 1]) : exact_token(text[start - 1])); | |
| bool right_ok = end == n || !(stream == 0 ? token(text[end]) : exact_token(text[end])); | |
| if (!left_ok || !right_ok) continue; | |
| if (stream == 0) native_candidate = true; | |
| else exact_candidate = true; | |
| // Raw keys remain ASCII: a hit spans key_lengths[key] scalars. | |
| // Compatibility mappings never enter either raw matching stream. | |
| int original_end = stream == 0 ? offsets[end - 1] + 1 : raw_source_position + 1; | |
| int original_start = stream == 0 ? offsets[start] : original_end - key_lengths[key]; | |
| if (!append_hit(key, original_start, original_end, key_offsets, | |
| key_consumers, consumers, candidates, &count, conjunctions)) { | |
| codes[row] = 3; return; | |
| } | |
| } | |
| } | |
| } | |
| bool accepted_nonsemantic = false, accepted_evidence = false; | |
| bool unresolved_exposure = false, unresolved_event = false; | |
| unsigned int supported = 0, observations = 0; | |
| for (int i = 0; i < count; ++i) { | |
| Candidate candidate = candidates[i]; | |
| if (candidate.kind == 0) { | |
| const long long* alias = aliases + candidate.index * 8; | |
| bool longest = true; | |
| for (int j = 0; j < count; ++j) { | |
| Candidate other = candidates[j]; | |
| if (other.kind == 0 && other.start == candidate.start && | |
| aliases[other.index * 8] == alias[0] && other.end > candidate.end) { | |
| longest = false; break; | |
| } | |
| } | |
| if (!longest) continue; | |
| ++observations; // Retained observations count before temporal/context checks. | |
| if (timestamps[row] < alias[1] || (alias[3] && timestamps[row] >= alias[2])) continue; | |
| if (!alias_context_ok((int)alias[7], contexts, clauses, term_ids, | |
| term_bytes, term_offsets, normalized, offsets, | |
| norm_length, source_length, candidate.start, candidate.end)) continue; | |
| accepted_nonsemantic = true; | |
| if (alias[4] != -1 && (alias[5] == 0 || alias[5] == 1)) { | |
| accepted_evidence = true; supported |= 1u << alias[4]; | |
| } else if (alias[6]) unresolved_exposure = true; | |
| } else { | |
| const int* rule = lexical + candidate.index * 7; | |
| ++observations; | |
| bool context_ok = rule[2] == 0; | |
| if (!context_ok) { | |
| int left, right; | |
| context_bounds(normalized, offsets, norm_length, source_length, | |
| candidate.start, candidate.end, rule[0], &left, &right); | |
| for (int j = 0; j < rule[2]; ++j) { | |
| if (contains_term(normalized, left, right, term_ids[rule[1] + j], | |
| term_bytes, term_offsets)) { context_ok = true; break; } | |
| } | |
| } | |
| if (rule[3]) { | |
| if (rule[4]) { accepted_evidence = true; supported |= 8; } | |
| else { if (rule[5]) unresolved_exposure = true; if (rule[6]) unresolved_event = true; } | |
| } else { | |
| if (context_ok) accepted_nonsemantic = true; | |
| if (rule[4] && context_ok) { accepted_evidence = true; supported |= 8; } | |
| else if (rule[5]) unresolved_exposure = true; | |
| } | |
| } | |
| } | |
| for (int i = 0; i < conjunction_count; ++i) { | |
| if (conjunctions[i] == 15) { ++observations; accepted_evidence = true; supported |= 8; } | |
| } | |
| if (!observations) return; | |
| fact[0] = accepted_evidence || accepted_nonsemantic || unresolved_event ? 0 : 2; | |
| for (int family = 0; family < 4; ++family) | |
| fact[family + 1] = supported & (1u << family) ? 0 : | |
| (family != 3 && !supported && unresolved_exposure ? 1 : 2); | |
| fact[5] = observations; fact[6] = native_candidate; fact[7] = exact_candidate; | |
| codes[row] = 1; | |
| } | |
| ''' | |
| def build_cubin(output_path, *, nvcc="nvcc", architecture="sm_75", timeout=1200): | |
| """Explicit maintainer build; never called by classifier construction/execution.""" | |
| if not architecture.startswith("sm_") or not architecture[3:].isdigit(): | |
| raise ValueError("architecture must name an explicit SM, for example sm_75") | |
| target = Path(output_path).resolve() | |
| if target.exists(): | |
| raise FileExistsError(f"choose a fresh output path: {target}") | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| with tempfile.TemporaryDirectory(prefix="range-phase32a-") as temporary: | |
| source = Path(temporary) / "range_phase32a.cu" | |
| source.write_text(CUDA_SOURCE, encoding="utf-8") | |
| command = [str(nvcc), "--cubin", "-O3", "--std=c++14", | |
| f"-arch={architecture}", str(source), "-o", str(target)] | |
| completed = subprocess.run(command, check=True, capture_output=True, | |
| text=True, timeout=timeout) | |
| return {"cubin": str(target), "architecture": architecture, | |
| "cubin_sha256": hashlib.sha256(target.read_bytes()).hexdigest(), | |
| "cuda_source_sha256": hashlib.sha256(CUDA_SOURCE.encode()).hexdigest(), | |
| "compiler_stdout": completed.stdout, "compiler_stderr": completed.stderr} | |
| class UnsupportedGpuTaxonomy(ValueError): | |
| """The experimental kernel cannot represent this valid CPU configuration.""" | |
| def _ascii(value): | |
| if not isinstance(value, str) or not value.isascii(): | |
| raise UnsupportedGpuTaxonomy("GPU experiment requires ASCII pattern/context literals") | |
| if not value: | |
| raise UnsupportedGpuTaxonomy("GPU experiment does not support empty literals") | |
| return value.encode("ascii") | |
| def _dfa(patterns): | |
| """Failure-completed byte-class DFA; terminal outputs preserve pattern IDs.""" | |
| children, outputs = [{}], [set()] | |
| alphabet = sorted({byte for pattern, _key in patterns for byte in pattern}) | |
| representatives = list(alphabet) | |
| absent = next((byte for byte in range(256) if byte not in alphabet), None) | |
| if absent is not None: | |
| representatives.append(absent) | |
| if not representatives: | |
| representatives = [0] | |
| class_map = np.full(256, len(representatives) - 1, dtype=np.uint16) | |
| for index, byte in enumerate(alphabet): | |
| class_map[byte] = index | |
| for pattern, key in patterns: | |
| state = 0 | |
| for byte in pattern: | |
| if byte not in children[state]: | |
| children[state][byte] = len(children) | |
| children.append({}) | |
| outputs.append(set()) | |
| state = children[state][byte] | |
| outputs[state].add(key) | |
| if len(children) * len(representatives) * 4 > MAX_DFA_BYTES: | |
| raise UnsupportedGpuTaxonomy("GPU DFA exceeds the 32 MiB experiment limit") | |
| failure = [0] * len(children) | |
| queue = deque(children[0].values()) | |
| order = [] | |
| while queue: | |
| state = queue.popleft() | |
| order.append(state) | |
| for byte, child in children[state].items(): | |
| queue.append(child) | |
| fallback = failure[state] | |
| while fallback and byte not in children[fallback]: | |
| fallback = failure[fallback] | |
| failure[child] = children[fallback].get(byte, 0) | |
| outputs[child].update(outputs[failure[child]]) | |
| dense = np.zeros((len(children), len(representatives)), dtype=np.uint32) | |
| for index, byte in enumerate(representatives): | |
| dense[0, index] = children[0].get(byte, 0) | |
| for state in order: | |
| for index, byte in enumerate(representatives): | |
| dense[state, index] = children[state].get(byte, dense[failure[state], index]) | |
| output_ids, offsets = [], [0] | |
| for values in outputs: | |
| output_ids.extend(sorted(values)) | |
| offsets.append(len(output_ids)) | |
| return dense, class_map, offsets, output_ids | |
| def _pack_specs(specs): | |
| patterns, pattern_consumers, consumers, exact, aliases, lexical, contexts, conjunctions = specs | |
| if len(conjunctions) > MAX_CONJUNCTIONS: | |
| raise UnsupportedGpuTaxonomy(f"GPU supports at most {MAX_CONJUNCTIONS} conjunctions") | |
| if len(patterns) > 512 or sum(len(value) for value in patterns) > 16384: | |
| raise UnsupportedGpuTaxonomy("GPU normalized-pattern budget exceeded") | |
| consumers = [tuple(map(int, value)) for value in consumers] | |
| key_consumers = [list(map(int, value)) for value in pattern_consumers] | |
| keys = [_ascii(value) for value in patterns] | |
| streams = [[(value, index) for index, value in enumerate(keys)], [], []] | |
| raw_keys = {} | |
| for value, mode, kind, index in exact: | |
| encoded = _ascii(value) | |
| if mode == "case_sensitive": | |
| stream = 1 | |
| elif mode == "python_ignorecase_ascii_literal": | |
| stream, encoded = 2, encoded.lower() | |
| else: | |
| raise UnsupportedGpuTaxonomy(f"unsupported exact mode: {mode}") | |
| identity = (stream, encoded) | |
| if identity not in raw_keys: | |
| key = len(keys) | |
| raw_keys[identity] = key | |
| keys.append(encoded) | |
| key_consumers.append([]) | |
| streams[stream].append((encoded, key)) | |
| consumer = len(consumers) | |
| consumers.append((int(kind), int(index), 0, 0)) | |
| key_consumers[raw_keys[identity]].append(consumer) | |
| # Validate every device index/bit operation before copying the config. | |
| for kind, index, field, phrase in consumers: | |
| if kind == 0 and 0 <= index < len(aliases): | |
| continue | |
| if kind == 1 and 0 <= index < len(lexical): | |
| continue | |
| if (kind == 2 and 0 <= index < len(conjunctions) and 0 <= field < 4 | |
| and 0 <= phrase < conjunctions[index][field]): | |
| continue | |
| raise UnsupportedGpuTaxonomy("unsupported consumer index/empty conjunction field") | |
| flat_consumers, consumer_offsets = [], [0] | |
| for values in key_consumers: | |
| flat_consumers.extend(values) | |
| consumer_offsets.append(len(flat_consumers)) | |
| transitions, classes, dfa_meta, output_offsets, output_keys = [], [], [], [], [] | |
| for stream in streams: | |
| dense, class_map, offsets, ids = _dfa(stream) | |
| dfa_meta.extend((len(transitions), dense.shape[1], len(output_offsets), len(stream))) | |
| transitions.extend(dense.reshape(-1).tolist()) | |
| classes.extend(class_map.tolist()) | |
| output_offsets.extend(len(output_keys) + value for value in offsets) | |
| output_keys.extend(ids) | |
| if len(transitions) * 4 > MAX_DFA_BYTES: | |
| raise UnsupportedGpuTaxonomy("combined GPU DFAs exceed 32 MiB") | |
| term_bytes, term_offsets, term_ids, clause_specs = bytearray(), [0], [], [] | |
| term_map = {} | |
| def term(value): | |
| encoded = _ascii(value) | |
| if encoded not in term_map: | |
| term_map[encoded] = len(term_offsets) - 1 | |
| term_bytes.extend(encoded) | |
| term_offsets.append(len(term_bytes)) | |
| return term_map[encoded] | |
| def term_list(values): | |
| start = len(term_ids) | |
| term_ids.extend(term(value) for value in values) | |
| return start, len(values) | |
| context_specs = [] | |
| for window, accepts, rejects in contexts: | |
| first_clause = len(clause_specs) // 2 | |
| for clause in accepts: | |
| clause_specs.extend(term_list(clause)) | |
| reject_start, reject_count = term_list(rejects) | |
| context_specs.extend((window, first_clause, len(accepts), reject_start, reject_count)) | |
| lexical_specs = [] | |
| for window, terms, event, market_wide, unresolved, family in lexical: | |
| first, count = term_list(terms) | |
| lexical_specs.extend((window, first, count, event, market_wide, unresolved, family)) | |
| alias_specs = [] | |
| for group, start, end, family, kind, unresolved, context in aliases: | |
| if family not in (-1, 0, 1, 2, 3) or (context is not None and not 0 <= context < len(contexts)): | |
| raise UnsupportedGpuTaxonomy("unsupported alias family/context") | |
| alias_specs.extend((group, start, end or 0, end is not None, family, | |
| kind, unresolved, -1 if context is None else context)) | |
| if any(window < 0 or window > 2**31 - 1 for window, *_ in (*contexts, *lexical)): | |
| raise UnsupportedGpuTaxonomy("GPU context window must fit a nonnegative int32") | |
| arrays = { | |
| "transitions": np.asarray(transitions, dtype=np.uint32), | |
| "classes": np.asarray(classes, dtype=np.uint16), | |
| "dfa_meta": np.asarray(dfa_meta, dtype=np.int32), | |
| "output_offsets": np.asarray(output_offsets, dtype=np.int32), | |
| "output_keys": np.asarray(output_keys, dtype=np.int32), | |
| "key_lengths": np.asarray([len(value) for value in keys], dtype=np.int32), | |
| "key_offsets": np.asarray(consumer_offsets, dtype=np.int32), | |
| "key_consumers": np.asarray(flat_consumers, dtype=np.int32), | |
| "consumers": np.asarray(consumers, dtype=np.int32).reshape(-1), | |
| "aliases": np.asarray(alias_specs, dtype=np.int64), | |
| "lexical": np.asarray(lexical_specs, dtype=np.int32), | |
| "contexts": np.asarray(context_specs, dtype=np.int32), | |
| "clauses": np.asarray(clause_specs, dtype=np.int32), | |
| "term_ids": np.asarray(term_ids, dtype=np.int32), | |
| "term_bytes": np.frombuffer(bytes(term_bytes), dtype=np.uint8), | |
| "term_offsets": np.asarray(term_offsets, dtype=np.int32), | |
| } | |
| return arrays, len(conjunctions) | |
| def _utf8_buffers(array): | |
| """Same Arrow buffer convention as Phase18, without importing its engine.""" | |
| import pyarrow as pa | |
| if not (pa.types.is_string(array.type) or pa.types.is_large_string(array.type)): | |
| raise TypeError("text must be Arrow string or large_string") | |
| if array.null_count: | |
| raise ValueError("text must not contain nulls") | |
| width = 8 if pa.types.is_large_string(array.type) else 4 | |
| offsets = np.frombuffer(array.buffers()[1], dtype=np.int64 if width == 8 else np.int32, | |
| count=len(array) + 1, offset=array.offset * width).astype(np.int64) | |
| start, end = int(offsets[0]), int(offsets[-1]) | |
| buffer = array.buffers()[2] | |
| data = (np.frombuffer(buffer, dtype=np.uint8, count=end - start, offset=start) | |
| if buffer is not None else np.empty(0, dtype=np.uint8)) | |
| offsets -= start | |
| return data, offsets | |
| def _timestamps(array): | |
| import pyarrow as pa | |
| if not pa.types.is_timestamp(array.type) or array.type.tz is None: | |
| raise ValueError("timestamp must be a timezone-aware Arrow timestamp") | |
| if array.null_count: | |
| raise ValueError("timestamp must not contain nulls") | |
| values = array.cast(pa.int64()).to_numpy(zero_copy_only=False) | |
| unit = array.type.unit | |
| if unit == "us": | |
| return np.ascontiguousarray(values) | |
| if unit == "ns": | |
| # Rust signed integer division truncates toward zero, not minus infinity. | |
| return values // 1000 + ((values < 0) & (values % 1000 != 0)).astype(np.int64) | |
| multiplier = 1000 if unit == "ms" else 1_000_000 | |
| limit = np.iinfo(np.int64) | |
| if len(values) and (int(values.min()) < -((-int(limit.min)) // multiplier) | |
| or int(values.max()) > int(limit.max) // multiplier): | |
| raise ValueError("timestamp value overflows microseconds") | |
| return values * multiplier | |
| def _fact_schema(): | |
| import pyarrow as pa | |
| types = (pa.uint64(), *(pa.uint8() for _ in range(5)), pa.uint32(), pa.bool_(), pa.bool_()) | |
| return pa.schema([pa.field(name, dtype, nullable=False) for name, dtype in zip(FACT_NAMES, types)]) | |
| def _materialize(value): | |
| import pyarrow as pa | |
| if hasattr(value, "to_pyarrow"): | |
| value = value.to_pyarrow() | |
| if not isinstance(value, (pa.Table, pa.RecordBatch)): | |
| value = pa.record_batch(value) | |
| return value | |
| def _record_interval(target, stage, lane, started, ended): | |
| """Append primitive host-clock timestamps; diagnostic callers only.""" | |
| origin = target["_origin"] | |
| target["intervals"].append({"stage": stage, "lane": lane, | |
| "start_seconds": started - origin, | |
| "end_seconds": ended - origin}) | |
| class GpuScreenResult: | |
| row_count: int | |
| facts: object | |
| profile: dict | |
| def to_pyarrow(self): | |
| return self.facts | |
| PHASE32A_PREDICATE_ID = "range-u13-ascii-output-no-byte-growth-exclude-raw4-v1" | |
| PHASE32A_SOURCE_SHA256 = { | |
| "unicode_index.bin": "fe152ac184aa77c40d0f040ba725140b17d845bae9fae255b2b3d3c43bb79c8f", | |
| "unicode_pool.bin": "109a8da1563408ad1da82919a5dfa1ab9fa18870f0f4d22e2818857b5a4fc30f", | |
| "python_whitespace.bin": "6c314375b6c075589b515139e50068d72c0cbd7daf4246e555a9dcfc9748ecc9", | |
| } | |
| def _load_phase32a_tables(directory): | |
| """Validate an explicitly built table artifact before any GPU operation.""" | |
| directory = Path(directory).resolve() | |
| manifest_path = directory / "manifest.json" | |
| if manifest_path.stat().st_size > 65536: | |
| raise ValueError("Phase32A manifest exceeds the bounded metadata size") | |
| manifest_bytes = manifest_path.read_bytes() | |
| manifest = json.loads(manifest_bytes) | |
| if (manifest.get("schema") != 1 or manifest.get("predicate_id") != PHASE32A_PREDICATE_ID | |
| or manifest.get("unicode") != "13.0.0" | |
| or manifest.get("source_sha256") != PHASE32A_SOURCE_SHA256 | |
| or manifest.get("raw_excluded_scalars") != [0x0130, 0x0131, 0x017F, 0x212A] | |
| or manifest.get("page_shift") != 8 or manifest.get("max_output_units") != 4): | |
| raise ValueError("wrong Phase32A Unicode table identity") | |
| names = ("unicode_ascii_directory.bin", "unicode_ascii_pages.bin") | |
| if set(manifest.get("payloads", {})) != set(names): | |
| raise ValueError("unexpected Phase32A payload manifest entries") | |
| pages = manifest["page_count"] | |
| if not isinstance(pages, int) or isinstance(pages, bool) or not 0 < pages <= 4353: | |
| raise ValueError("wrong Phase32A page count") | |
| expected_sizes = {names[0]: 4352 * 2, names[1]: pages * 256 * 4} | |
| payloads = {} | |
| for name in names: | |
| if (directory / name).stat().st_size != expected_sizes[name]: | |
| raise ValueError("wrong bounded Phase32A payload size: " + name) | |
| data = (directory / name).read_bytes() | |
| identity = manifest["payloads"][name] | |
| if len(data) != identity["bytes"] or hashlib.sha256(data).hexdigest() != identity["sha256"]: | |
| raise ValueError("Phase32A table payload mismatch: " + name) | |
| payloads[name] = data | |
| if (not isinstance(pages, int) or isinstance(pages, bool) or not 0 < pages <= 4353 | |
| or len(payloads[names[0]]) != 4352 * 2 or len(payloads[names[1]]) != pages * 256 * 4): | |
| raise ValueError("wrong Phase32A page dimensions") | |
| directory_array = np.frombuffer(payloads[names[0]], dtype="<u2") | |
| pages_array = np.frombuffer(payloads[names[1]], dtype="<u4") | |
| if int(directory_array.max()) >= pages or np.any(pages_array[:256]): | |
| raise ValueError("invalid Phase32A page directory / unsupported page") | |
| # Kernel code depends on this exclusion for original-text exact matching. | |
| for scalar in (0x0130, 0x0131, 0x017F, 0x212A): | |
| if pages_array[int(directory_array[scalar >> 8]) * 256 + (scalar & 255)] != 0: | |
| raise ValueError("raw-ignorecase exception unexpectedly eligible") | |
| return ({"unicode_directory": directory_array, "unicode_pages": pages_array}, | |
| {"predicate_id": PHASE32A_PREDICATE_ID, | |
| "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(), | |
| "payloads": manifest["payloads"], "source_sha256": PHASE32A_SOURCE_SHA256}) | |
| class ExactSelectiveUnicodeCudaClassifier: | |
| """Experimental screen_arrow-compatible adapter; no global backend mutation.""" | |
| def __init__(self, cpu_engine, cubin_path, *, masked_cpu: Callable, unicode_tables, | |
| batch_bytes=64 * 1024 * 1024, batch_rows=65536, max_row_bytes=65536, | |
| overlap_cpu=False, diagnostics=False): | |
| from range_curate._core.accelerated.compact import _build_native_specs | |
| for name, value in (("batch_bytes", batch_bytes), ("batch_rows", batch_rows), | |
| ("max_row_bytes", max_row_bytes)): | |
| if isinstance(value, bool) or not isinstance(value, int) or not 0 < value < 2**31: | |
| raise ValueError(f"{name} must be a positive int32") | |
| if max_row_bytes > batch_bytes: | |
| raise ValueError("max_row_bytes must not exceed batch_bytes") | |
| if not callable(masked_cpu): | |
| raise TypeError("masked_cpu must be the current native masked scanner callable") | |
| if not isinstance(overlap_cpu, bool): | |
| raise TypeError("overlap_cpu must be a boolean") | |
| if not isinstance(diagnostics, bool): | |
| raise TypeError("diagnostics must be a boolean") | |
| self.cpu_engine, self.masked_cpu = cpu_engine, masked_cpu | |
| self.cubin_path = Path(cubin_path).resolve() | |
| self.batch_bytes, self.batch_rows, self.max_row_bytes = batch_bytes, batch_rows, max_row_bytes | |
| self.overlap_cpu = overlap_cpu | |
| self.diagnostics = diagnostics | |
| self._unicode_arrays, self._unicode_identity = _load_phase32a_tables(unicode_tables) | |
| self.unsupported_reason = None | |
| self._arrays, self._conjunction_count = {}, 0 | |
| try: | |
| self._arrays, self._conjunction_count = _pack_specs(_build_native_specs(cpu_engine._screening_indicator)) | |
| except UnsupportedGpuTaxonomy as exc: | |
| self.unsupported_reason = str(exc) | |
| if not self.unsupported_reason: | |
| self._arrays.update(self._unicode_arrays) | |
| self.cp = self._kernel = None | |
| self._device_config, self._workspace = {}, {} | |
| self._capacity_bytes = self._capacity_rows = 0 | |
| self._cubin_sha256 = None | |
| self._cuda_source_sha256 = hashlib.sha256(CUDA_SOURCE.encode()).hexdigest() | |
| self._kernel_attributes = None | |
| self.last_profile = {} | |
| def backend_report(self): | |
| return {"backend": "cpu" if self.unsupported_reason else "experimental_selective_unicode_cuda", | |
| "unsupported_taxonomy_reason": self.unsupported_reason, | |
| "unicode_table_identity": self._unicode_identity, | |
| "unicode_fallback_definition": "outside verified subset; original raw semantics retained", | |
| "cubin": str(self.cubin_path), "batch_bytes": self.batch_bytes, | |
| "batch_rows": self.batch_rows, "max_row_bytes": self.max_row_bytes, | |
| "candidate_limit": MAX_CANDIDATES, "overlap_cpu": self.overlap_cpu, | |
| "execution": "cpu_gpu_chunk_pipeline" if self.overlap_cpu else "synchronous", | |
| "component_times_may_overlap": self.overlap_cpu, | |
| "cubin_sha256": self._cubin_sha256, | |
| "cuda_source_sha256": self._cuda_source_sha256, | |
| "kernel_attributes": self._kernel_attributes} | |
| def _initialize(self): | |
| if self.cp is not None: | |
| return | |
| if not self.cubin_path.is_file(): | |
| raise FileNotFoundError("Build the experimental cubin explicitly before running CUDA: " + str(self.cubin_path)) | |
| import cupy as cp | |
| module = cp.RawModule(path=str(self.cubin_path)) | |
| kernel = module.get_function("range_selective_unicode_classify") | |
| # Keep load-time metadata: later reports must not describe a replacement | |
| # file at the same path or rehash source text inside screening timings. | |
| loaded_cubin_sha256 = hashlib.sha256(self.cubin_path.read_bytes()).hexdigest() | |
| attributes = kernel.attributes | |
| kernel_attributes = { | |
| name: int(attributes[name]) if attributes.get(name) is not None else None | |
| for name in ("num_regs", "local_size_bytes", "shared_size_bytes") | |
| } | |
| config = {name: cp.asarray(value if value.size else np.zeros(1, dtype=value.dtype)) | |
| for name, value in self._arrays.items()} | |
| cp.cuda.Stream.null.synchronize() | |
| self.cp, self._module, self._kernel, self._device_config = cp, module, kernel, config | |
| self._cubin_sha256 = loaded_cubin_sha256 | |
| self._kernel_attributes = kernel_attributes | |
| def diagnostic_unicode_shadow(self, data, *, enabled=False, text_column="text"): | |
| """Derived host replay, explicitly requested AFTER diagnostic timing. | |
| This method is never called by screen_arrow or its CUDA stages. It does | |
| not measure device instructions, memory transactions or service time. | |
| Input must satisfy the normal valid-UTF-8 Arrow contract. It retains at | |
| most one bounded row's decoded text and returns aggregate numbers only. | |
| """ | |
| if enabled is not True: | |
| raise ValueError("Unicode host shadow requires explicit enabled=True outside timing") | |
| import pyarrow as pa | |
| if not isinstance(data, (pa.Table, pa.RecordBatch)): | |
| raise TypeError("Unicode host shadow accepts materialized Arrow input") | |
| if self.unsupported_reason: | |
| return {"scope": "derived host model; no device counters", "applicable": False, | |
| "reason": "unsupported taxonomy routes the complete call to CPU"} | |
| started = time.perf_counter_ns() | |
| counts = dict.fromkeys(("input_rows", "input_bytes", "modeled_decoded_bytes", | |
| "modeled_decoded_scalars", "modeled_supported_scalars", "modeled_prefix_output_units", | |
| "modeled_completed_rows", "modeled_completed_input_bytes", "modeled_completed_source_scalars", | |
| "modeled_completed_output_units", "modeled_unsupported_rows", "modeled_unsupported_row_bytes", | |
| "modeled_oversized_rows", "modeled_oversized_row_bytes"), 0) | |
| mapping_units = {str(value): 0 for value in range(5)} | |
| expansion_rows = {name: 0 for name in ("zero_output", "up_to_1", "up_to_2", "up_to_3", "up_to_4")} | |
| directory, pages = self._unicode_arrays["unicode_directory"], self._unicode_arrays["unicode_pages"] | |
| for batch, _, _ in self._chunks(data, text_column): | |
| raw, offsets = _utf8_buffers(batch.column(batch.schema.get_field_index(text_column))) | |
| for left, right in zip(offsets[:-1], offsets[1:]): | |
| left, right = int(left), int(right) | |
| byte_count = right - left | |
| counts["input_rows"] += 1 | |
| counts["input_bytes"] += byte_count | |
| if byte_count > self.max_row_bytes: | |
| counts["modeled_oversized_rows"] += 1 | |
| counts["modeled_oversized_row_bytes"] += byte_count | |
| continue | |
| text = raw[left:right].tobytes().decode("utf-8", errors="strict") | |
| output = source_scalars = 0 | |
| pending = False | |
| complete = True | |
| for character in text: | |
| scalar = ord(character) | |
| width = 1 if scalar < 128 else 2 if scalar < 2048 else 3 if scalar < 65536 else 4 | |
| counts["modeled_decoded_bytes"] += width | |
| counts["modeled_decoded_scalars"] += 1 | |
| if scalar < 128: | |
| unit = scalar + 32 if 65 <= scalar <= 90 else scalar | |
| if unit in b"-.,:;()[]{}/": | |
| unit = 32 | |
| elif unit == 96: | |
| unit = 39 | |
| packed = 2 | (unit << 3) | |
| else: | |
| packed = int(pages[int(directory[scalar >> 8]) * 256 + (scalar & 255)]) | |
| if not (packed & 7): | |
| counts["modeled_unsupported_rows"] += 1 | |
| counts["modeled_unsupported_row_bytes"] += byte_count | |
| complete = False | |
| break | |
| units = (packed & 7) - 1 | |
| if not 0 <= units <= min(width, 4): | |
| raise ValueError("Unicode table violates the shadow model's capacity predicate") | |
| mapping_units[str(units)] += 1 | |
| counts["modeled_supported_scalars"] += 1 | |
| for index in range(units): | |
| unit = (packed >> (3 + 7 * index)) & 127 | |
| if unit == 32 or 9 <= unit <= 13 or 28 <= unit <= 31: | |
| pending = output != 0 | |
| continue | |
| if pending: | |
| output += 1 | |
| pending = False | |
| output += 1 | |
| source_scalars += 1 | |
| if output > byte_count: | |
| raise ValueError("Unicode shadow output exceeds the declared per-row arena") | |
| counts["modeled_prefix_output_units"] += output | |
| if complete: | |
| counts["modeled_completed_rows"] += 1 | |
| counts["modeled_completed_input_bytes"] += byte_count | |
| counts["modeled_completed_source_scalars"] += source_scalars | |
| counts["modeled_completed_output_units"] += output | |
| bucket = ("zero_output" if not output else "up_to_1" if output <= source_scalars else | |
| "up_to_2" if output <= 2 * source_scalars else | |
| "up_to_3" if output <= 3 * source_scalars else "up_to_4") | |
| expansion_rows[bucket] += 1 | |
| output = counts["modeled_completed_output_units"] | |
| return {"scope": "derived host normalization replay after screen_arrow timer/profiler range; not device-measured", | |
| "applicable": True, "performance_evidence": False, "counts": counts, | |
| "modeled_normalizer_unique_byte_positions_touched": counts["modeled_decoded_bytes"], | |
| "touch_definition": "valid UTF-8 positions through and including the first unsupported scalar; oversized rows touch zero; excludes raw matching/context/repeated loads/cache transactions", | |
| "actual_device_decoded_bytes": None, "actual_device_touched_bytes": None, | |
| "device_internal_service_seconds": None, | |
| "mapping_output_units_per_supported_scalar_histogram": mapping_units, | |
| "completed_row_output_units_per_source_scalar_histogram": expansion_rows, | |
| "modeled_completed_output_units_per_source_scalar": ( | |
| output / counts["modeled_completed_source_scalars"] if counts["modeled_completed_source_scalars"] else None), | |
| "modeled_completed_output_bytes_per_input_byte": ( | |
| output / counts["modeled_completed_input_bytes"] if counts["modeled_completed_input_bytes"] else None), | |
| "normalization_completion_note": "includes rows later assigned CPU fallback for candidate overflow; it is not final eligible-row coverage", | |
| "host_shadow_elapsed_ns": time.perf_counter_ns() - started, | |
| "unicode_table_identity": self._unicode_identity} | |
| def _reserve(self, byte_count, row_count): | |
| cp = self.cp | |
| if byte_count > self._capacity_bytes: | |
| self._capacity_bytes = max(byte_count, self._capacity_bytes) | |
| self._workspace.update(data=cp.empty(self._capacity_bytes, dtype=cp.uint8), | |
| normalized=cp.empty(self._capacity_bytes, dtype=cp.uint8), | |
| source_offsets=cp.empty(self._capacity_bytes, dtype=cp.uint32)) | |
| if row_count > self._capacity_rows: | |
| self._capacity_rows = max(row_count, self._capacity_rows) | |
| self._workspace.update(offsets=cp.empty(self._capacity_rows + 1, dtype=cp.int64), | |
| timestamps=cp.empty(self._capacity_rows, dtype=cp.int64), | |
| candidates=cp.empty(self._capacity_rows * MAX_CANDIDATES * 4, dtype=cp.int32), | |
| codes=cp.empty(self._capacity_rows, dtype=cp.uint8), | |
| facts=cp.empty((self._capacity_rows, 8), dtype=cp.uint32)) | |
| def _gpu_stage(self, batch, *, text_column, timestamp_column, diagnostic=None): | |
| """Complete all CUDA work and return independent, read-only host arrays.""" | |
| cp = self.cp | |
| total_started = time.perf_counter() | |
| data, offsets = _utf8_buffers(batch.column(batch.schema.get_field_index(text_column))) | |
| if diagnostic is not None: | |
| boundary = time.perf_counter() | |
| _record_interval(diagnostic, "utf8_buffer_prepare", "main", total_started, boundary) | |
| timestamp_started = boundary | |
| epochs = _timestamps(batch.column(batch.schema.get_field_index(timestamp_column))) | |
| if diagnostic is not None: | |
| boundary = time.perf_counter() | |
| _record_interval(diagnostic, "timestamp_prepare", "main", timestamp_started, boundary) | |
| reserve_started = boundary | |
| rows, byte_count = batch.num_rows, data.nbytes | |
| self._reserve(max(1, byte_count), rows) | |
| buffers, config = self._workspace, self._device_config | |
| preparation = time.perf_counter() - total_started | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "workspace_reserve", "main", reserve_started, | |
| total_started + preparation) | |
| started = time.perf_counter() | |
| if byte_count: | |
| buffers["data"][:byte_count].set(data) | |
| buffers["offsets"][:rows + 1].set(offsets) | |
| buffers["timestamps"][:rows].set(epochs) | |
| cp.cuda.Stream.null.synchronize() | |
| h2d = time.perf_counter() - started | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "h2d", "main", started, started + h2d) | |
| started = time.perf_counter() | |
| arguments = [buffers["data"], buffers["offsets"], buffers["timestamps"], | |
| np.int32(rows), np.int32(self.max_row_bytes)] | |
| arguments.extend(config[name] for name in self._arrays) | |
| arguments.extend((np.int32(self._conjunction_count), buffers["normalized"], | |
| buffers["source_offsets"], buffers["candidates"], buffers["codes"], buffers["facts"])) | |
| self._kernel(((rows + 127) // 128,), (128,), tuple(arguments)) | |
| cp.cuda.Stream.null.synchronize() | |
| kernel = time.perf_counter() - started | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "kernel_host_interval", "main", started, started + kernel) | |
| started = time.perf_counter() | |
| codes = cp.asnumpy(buffers["codes"][:rows]) | |
| # One fixed 32-byte result slot per row, not a row-by-pattern matrix. | |
| # Host compaction below retains only actual facts in the returned Arrow data. | |
| dense = cp.asnumpy(buffers["facts"][:rows]) | |
| d2h = time.perf_counter() - started | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "d2h", "main", started, started + d2h) | |
| finalize_started = started + d2h | |
| # Neither an outstanding host task nor its Arrow input retains a device | |
| # workspace view. Reusing the single CUDA workspace for the next chunk | |
| # cannot overwrite this task's codes, facts or source-byte offsets. | |
| for values in (codes, dense, offsets): | |
| values.setflags(write=False) | |
| profile = {"rows": rows, "bytes": byte_count, "host_prepare_seconds": preparation, | |
| "h2d_seconds": h2d, "kernel_seconds": kernel, "d2h_seconds": d2h} | |
| if diagnostic is not None: | |
| diagnostic["transfer_payload_bytes"] = { | |
| "h2d": {"text": int(byte_count), "row_offsets": int(offsets.nbytes), | |
| "timestamps": int(epochs.nbytes)}, | |
| "d2h": {"route_codes": int(codes.nbytes), "dense_facts": int(dense.nbytes)}, | |
| } | |
| _record_interval(diagnostic, "gpu_host_finalize", "main", finalize_started, | |
| time.perf_counter()) | |
| return codes, dense, offsets, profile | |
| def _complete_gpu_chunk(self, batch, stage, *, workers, text_column, timestamp_column, | |
| diagnostic=None): | |
| """CPU-only ownership and Arrow assembly; safe in the one host worker.""" | |
| import pyarrow as pa | |
| codes, dense, offsets, gpu_profile = stage | |
| started = time.perf_counter() | |
| fallback_mask = codes >= 2 | |
| fallback = None | |
| if np.any(fallback_mask): | |
| fallback = _materialize(self.masked_cpu( | |
| batch, fallback_mask.astype(np.uint8).tobytes(), workers=workers, | |
| text_column=text_column, timestamp_column=timestamp_column)) | |
| fallback_seconds = time.perf_counter() - started | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "cpu_fallback", diagnostic["completion_lane"], | |
| started, started + fallback_seconds) | |
| started = time.perf_counter() | |
| selected = np.flatnonzero(codes == 1) | |
| values = dense[selected] | |
| schema = _fact_schema() | |
| columns = [pa.array(selected, type=pa.uint64())] | |
| columns.extend(pa.array(values[:, index].astype(np.bool_) if index >= 6 else values[:, index], | |
| type=schema.field(index + 1).type) | |
| for index in range(8)) | |
| gpu_facts = pa.RecordBatch.from_arrays(columns, schema=schema) | |
| tables = [pa.Table.from_batches([gpu_facts])] | |
| if fallback is not None: | |
| tables.append(pa.Table.from_batches([fallback]) if isinstance(fallback, pa.RecordBatch) else fallback) | |
| result = pa.concat_tables(tables).sort_by([("row_index", "ascending")]) | |
| assembly = time.perf_counter() - started | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "chunk_assembly", diagnostic["completion_lane"], | |
| started, started + assembly) | |
| accounting_started = started + assembly | |
| byte_lengths = np.diff(offsets) | |
| profile = {**gpu_profile, "gpu_completed_rows": int(np.count_nonzero(codes < 2)), | |
| "unicode_rows": int(np.count_nonzero(codes == 2)), | |
| "invalid_utf8_rows": int(np.count_nonzero(codes == 5)), | |
| "normalization_capacity_rows": int(np.count_nonzero(codes == 6)), | |
| "overflow_rows": int(np.count_nonzero(codes == 3)), | |
| "oversized_rows": int(np.count_nonzero(codes == 4)), | |
| "fallback_rows": int(np.count_nonzero(fallback_mask)), | |
| "fallback_bytes": int(byte_lengths[fallback_mask].sum()), | |
| "cpu_fallback_seconds": fallback_seconds, "assembly_seconds": assembly} | |
| if diagnostic is not None: | |
| diagnostic["fallback_bytes_by_reason"] = { | |
| "unicode": int(byte_lengths[codes == 2].sum()), | |
| "invalid_utf8": int(byte_lengths[codes == 5].sum()), | |
| "normalization_capacity": int(byte_lengths[codes == 6].sum()), | |
| "candidate_overflow": int(byte_lengths[codes == 3].sum()), | |
| "oversized": int(byte_lengths[codes == 4].sum()), | |
| } | |
| diagnostic["gpu_completed_bytes"] = int(byte_lengths[codes < 2].sum()) | |
| diagnostic["measurements"] = profile.copy() | |
| _record_interval(diagnostic, "completion_accounting", diagnostic["completion_lane"], | |
| accounting_started, time.perf_counter()) | |
| return result, profile | |
| def _gpu_chunk(self, batch, *, workers, text_column, timestamp_column): | |
| """The original synchronous scheduling, retained as the default control.""" | |
| stage = self._gpu_stage(batch, text_column=text_column, timestamp_column=timestamp_column) | |
| return self._complete_gpu_chunk(batch, stage, workers=workers, text_column=text_column, | |
| timestamp_column=timestamp_column) | |
| def _oversized_chunk(self, piece, byte_count, *, workers, text_column, timestamp_column, | |
| diagnostic=None): | |
| import pyarrow as pa | |
| started = time.perf_counter() | |
| facts = _materialize(self.masked_cpu(piece, bytes([1]), workers=workers, | |
| text_column=text_column, timestamp_column=timestamp_column)) | |
| facts = pa.Table.from_batches([facts]) if isinstance(facts, pa.RecordBatch) else facts | |
| profile = {"rows": 1, "bytes": byte_count, "oversized_rows": 1, | |
| "fallback_rows": 1, "fallback_bytes": byte_count, | |
| "cpu_fallback_seconds": time.perf_counter() - started} | |
| if diagnostic is not None: | |
| _record_interval(diagnostic, "cpu_oversized_fallback", diagnostic["completion_lane"], | |
| started, started + profile["cpu_fallback_seconds"]) | |
| diagnostic["fallback_bytes_by_reason"] = { | |
| "unicode": 0, "candidate_overflow": 0, "oversized": int(byte_count), | |
| } | |
| diagnostic["gpu_completed_bytes"] = 0 | |
| diagnostic["transfer_payload_bytes"] = { | |
| "h2d": {"text": 0, "row_offsets": 0, "timestamps": 0}, | |
| "d2h": {"route_codes": 0, "dense_facts": 0}, | |
| } | |
| diagnostic["measurements"] = profile.copy() | |
| _record_interval(diagnostic, "completion_accounting", diagnostic["completion_lane"], | |
| started + profile["cpu_fallback_seconds"], time.perf_counter()) | |
| return facts, profile | |
| def _chunks(self, data, text_column): | |
| import pyarrow as pa | |
| batches = data.to_batches(max_chunksize=self.batch_rows) if isinstance(data, pa.Table) else [data] | |
| row_base = 0 | |
| for batch in batches: | |
| _, offsets = _utf8_buffers(batch.column(batch.schema.get_field_index(text_column))) | |
| start = 0 | |
| while start < batch.num_rows: | |
| end = min(batch.num_rows, start + self.batch_rows, | |
| int(np.searchsorted(offsets, offsets[start] + self.batch_bytes, side="right")) - 1) | |
| end = max(start + 1, end) | |
| yield batch.slice(start, end - start), row_base + start, int(offsets[end] - offsets[start]) | |
| start = end | |
| row_base += batch.num_rows | |
| def screen_arrow(self, data, *, workers=10, row_count=None, text_column="text", timestamp_column="timestamp"): | |
| import pyarrow as pa | |
| if isinstance(workers, bool) or not isinstance(workers, int) or workers < 1: | |
| raise ValueError("workers must be a positive integer") | |
| started = time.perf_counter() | |
| if not isinstance(data, (pa.Table, pa.RecordBatch)): | |
| raise TypeError("this experiment accepts a materialized Arrow Table or RecordBatch") | |
| if row_count is not None and (isinstance(row_count, bool) or not isinstance(row_count, int) or row_count < 0): | |
| raise ValueError("row_count must be a nonnegative integer or None") | |
| if row_count is not None and row_count != data.num_rows: | |
| raise ValueError("row_count differs from Arrow input") | |
| if self.diagnostics: | |
| return self._screen_arrow_diagnostics( | |
| data, workers=workers, text_column=text_column, | |
| timestamp_column=timestamp_column, started=started) | |
| if self.unsupported_reason: | |
| result = self.cpu_engine.screen_arrow(data, workers=workers, row_count=data.num_rows, | |
| text_column=text_column, timestamp_column=timestamp_column) | |
| profile = {**self.backend_report, "rows": data.num_rows, | |
| "total_seconds": time.perf_counter() - started} | |
| self.last_profile = profile | |
| return GpuScreenResult(data.num_rows, _materialize(result), profile) | |
| for name in (text_column, timestamp_column): | |
| if data.schema.get_field_index(name) < 0: | |
| raise ValueError(f"missing or ambiguous Arrow column: {name}") | |
| text_type = data.schema.field(text_column).type | |
| timestamp_type = data.schema.field(timestamp_column).type | |
| if not (pa.types.is_string(text_type) or pa.types.is_large_string(text_type)): | |
| raise TypeError("text must be Arrow string or large_string") | |
| if not pa.types.is_timestamp(timestamp_type) or timestamp_type.tz is None: | |
| raise ValueError("timestamp must be a timezone-aware Arrow timestamp") | |
| if data.column(text_column).null_count or data.column(timestamp_column).null_count: | |
| raise ValueError("text and timestamp must not contain nulls") | |
| self._initialize() | |
| output, aggregate = [], {} | |
| keywords = dict(workers=workers, text_column=text_column, timestamp_column=timestamp_column) | |
| def collect(completed, global_offset): | |
| facts, profile = completed | |
| if facts.num_rows: | |
| indices = facts.column("row_index").to_numpy().astype(np.uint64) + np.uint64(global_offset) | |
| output.append(facts.set_column(0, _fact_schema().field(0), pa.array(indices, type=pa.uint64()))) | |
| for key, value in profile.items(): | |
| aggregate[key] = aggregate.get(key, 0) + value | |
| if not self.overlap_cpu: | |
| for piece, global_offset, byte_count in self._chunks(data, text_column): | |
| if byte_count > self.batch_bytes: | |
| completed = self._oversized_chunk(piece, byte_count, **keywords) | |
| else: | |
| completed = self._gpu_chunk(piece, **keywords) | |
| collect(completed, global_offset) | |
| else: | |
| # At most one host completion is outstanding. The next GPU stage | |
| # uses the same device workspace while that task holds independent | |
| # host copies. No CUDA calls occur in the host worker. | |
| pending, pending_offset = None, 0 | |
| with ThreadPoolExecutor(max_workers=1, thread_name_prefix="range-cuda-fallback") as executor: | |
| try: | |
| for piece, global_offset, byte_count in self._chunks(data, text_column): | |
| stage = None | |
| if byte_count <= self.batch_bytes: | |
| stage = self._gpu_stage(piece, text_column=text_column, | |
| timestamp_column=timestamp_column) | |
| if pending is not None: | |
| collect(pending.result(), pending_offset) | |
| pending = None | |
| if stage is None: | |
| pending = executor.submit(self._oversized_chunk, piece, byte_count, **keywords) | |
| else: | |
| pending = executor.submit(self._complete_gpu_chunk, piece, stage, **keywords) | |
| pending_offset = global_offset | |
| if pending is not None: | |
| collect(pending.result(), pending_offset) | |
| pending = None | |
| except BaseException as foreground_error: | |
| # Drain the bounded host job before leaving this call. If | |
| # GPU/input work and the CPU job both failed, retain both | |
| # errors rather than abandoning the background exception. | |
| if pending is not None: | |
| try: | |
| pending.result() | |
| except BaseException as background_error: | |
| if background_error is not foreground_error: | |
| raise foreground_error from background_error | |
| raise | |
| facts = pa.concat_tables(output).combine_chunks() if output else pa.Table.from_batches([], schema=_fact_schema()) | |
| batch = (facts.to_batches(max_chunksize=max(1, facts.num_rows))[0] if facts.num_rows | |
| else pa.RecordBatch.from_arrays([pa.array([], type=field.type) for field in facts.schema], schema=facts.schema)) | |
| profile = {**self.backend_report, **aggregate, "rows": data.num_rows, | |
| "fact_rows": batch.num_rows, "total_seconds": time.perf_counter() - started, | |
| "device_workspace_bytes": sum(value.nbytes for value in self._workspace.values()), | |
| "device_config_bytes": sum(value.nbytes for value in self._device_config.values())} | |
| self.last_profile = profile | |
| return GpuScreenResult(data.num_rows, batch, profile) | |
| def _screen_arrow_diagnostics(self, data, *, workers, text_column, timestamp_column, started): | |
| """Opt-in host timeline, kept out of the legacy scheduling/timing path. | |
| The kernel and synchronization are unchanged. Intervals describe host | |
| service or coordination, not CUDA-event durations. Each completion job | |
| owns its chunk record; the main thread reads it only after completion. | |
| No article text, Arrow buffers, futures or device views enter the report. | |
| """ | |
| import pyarrow as pa | |
| call_trace = {"_origin": started, "intervals": []} | |
| diagnostic = { | |
| "schema_version": 1, | |
| "clock": "perf_counter_seconds_since_screen_arrow_start", | |
| "device_event_timings": False, | |
| "timing_scope": "extra_instrumented_call_not_headline_timing", | |
| "call_intervals": call_trace["intervals"], | |
| "chunks": [], | |
| } | |
| keywords = dict(workers=workers, text_column=text_column, timestamp_column=timestamp_column) | |
| if self.unsupported_reason: | |
| boundary = time.perf_counter() | |
| _record_interval(call_trace, "input_validation", "main", started, boundary) | |
| result = self.cpu_engine.screen_arrow(data, row_count=data.num_rows, **keywords) | |
| completed = time.perf_counter() | |
| _record_interval(call_trace, "unsupported_taxonomy_cpu", "main", boundary, completed) | |
| facts = _materialize(result) | |
| finalized = time.perf_counter() | |
| _record_interval(call_trace, "final_output_materialization", "main", completed, finalized) | |
| diagnostic["initialization_h2d_config_bytes"] = 0 | |
| diagnostic["transfer_payload_bytes"] = {"h2d": 0, "d2h": 0, "initialization_h2d": 0} | |
| diagnostic["call_total_seconds"] = finalized - started | |
| diagnostic["unsupported_taxonomy_reason"] = self.unsupported_reason | |
| profile = {**self.backend_report, "rows": data.num_rows, | |
| "total_seconds": finalized - started, "diagnostics": diagnostic} | |
| self.last_profile = profile | |
| return GpuScreenResult(data.num_rows, facts, profile) | |
| # Mirror the legacy validation before initializing any CUDA resource. | |
| for name in (text_column, timestamp_column): | |
| if data.schema.get_field_index(name) < 0: | |
| raise ValueError(f"missing or ambiguous Arrow column: {name}") | |
| text_type = data.schema.field(text_column).type | |
| timestamp_type = data.schema.field(timestamp_column).type | |
| if not (pa.types.is_string(text_type) or pa.types.is_large_string(text_type)): | |
| raise TypeError("text must be Arrow string or large_string") | |
| if not pa.types.is_timestamp(timestamp_type) or timestamp_type.tz is None: | |
| raise ValueError("timestamp must be a timezone-aware Arrow timestamp") | |
| if data.column(text_column).null_count or data.column(timestamp_column).null_count: | |
| raise ValueError("text and timestamp must not contain nulls") | |
| boundary = time.perf_counter() | |
| _record_interval(call_trace, "input_validation", "main", started, boundary) | |
| was_initialized = self.cp is not None | |
| self._initialize() | |
| initialized = time.perf_counter() | |
| _record_interval(call_trace, "initialization", "main", boundary, initialized) | |
| diagnostic["initialization_h2d_config_bytes"] = ( | |
| 0 if was_initialized else sum(int(value.nbytes) for value in self._device_config.values())) | |
| output, aggregate = [], {} | |
| def chunks(): | |
| iterator = iter(self._chunks(data, text_column)) | |
| while True: | |
| fetch_started = time.perf_counter() | |
| try: | |
| piece, global_offset, byte_count = next(iterator) | |
| except StopIteration: | |
| _record_interval(call_trace, "chunk_iterator_finish", "main", fetch_started, | |
| time.perf_counter()) | |
| return | |
| fetched = time.perf_counter() | |
| item = {"_origin": started, "intervals": [], | |
| "chunk_index": len(diagnostic["chunks"]), | |
| "row_index_start": int(global_offset), "rows": int(piece.num_rows), | |
| "input_bytes": int(byte_count), | |
| "route": "cpu_oversized" if byte_count > self.batch_bytes else "gpu_then_cpu", | |
| "completion_lane": "cpu_completion" if self.overlap_cpu else "main"} | |
| _record_interval(item, "chunk_fetch", "main", fetch_started, fetched) | |
| diagnostic["chunks"].append(item) | |
| yield piece, global_offset, byte_count, item | |
| def collect(completed, global_offset, item): | |
| collect_started = time.perf_counter() | |
| facts, chunk_profile = completed | |
| if facts.num_rows: | |
| indices = facts.column("row_index").to_numpy().astype(np.uint64) + np.uint64(global_offset) | |
| output.append(facts.set_column(0, _fact_schema().field(0), pa.array(indices, type=pa.uint64()))) | |
| for key, value in chunk_profile.items(): | |
| aggregate[key] = aggregate.get(key, 0) + value | |
| item["fact_rows"] = int(facts.num_rows) | |
| _record_interval(item, "collect", "main", collect_started, time.perf_counter()) | |
| def wait_and_collect(pending, global_offset, item): | |
| wait_started = time.perf_counter() | |
| completed = pending.result() | |
| _record_interval(item, "wait_for_completion", "main", wait_started, time.perf_counter()) | |
| collect(completed, global_offset, item) | |
| if not self.overlap_cpu: | |
| for piece, global_offset, byte_count, item in chunks(): | |
| if byte_count > self.batch_bytes: | |
| completed = self._oversized_chunk(piece, byte_count, diagnostic=item, **keywords) | |
| else: | |
| stage = self._gpu_stage(piece, text_column=text_column, | |
| timestamp_column=timestamp_column, diagnostic=item) | |
| completed = self._complete_gpu_chunk(piece, stage, diagnostic=item, **keywords) | |
| collect(completed, global_offset, item) | |
| else: | |
| pending, pending_offset, pending_item = None, 0, None | |
| executor_started = time.perf_counter() | |
| with ThreadPoolExecutor(max_workers=1, thread_name_prefix="range-cuda-fallback") as executor: | |
| _record_interval(call_trace, "executor_setup", "main", executor_started, time.perf_counter()) | |
| try: | |
| for piece, global_offset, byte_count, item in chunks(): | |
| stage = None | |
| if byte_count <= self.batch_bytes: | |
| stage = self._gpu_stage(piece, text_column=text_column, | |
| timestamp_column=timestamp_column, diagnostic=item) | |
| if pending is not None: | |
| wait_and_collect(pending, pending_offset, pending_item) | |
| pending = None | |
| submit_started = time.perf_counter() | |
| item["submitted_seconds"] = submit_started - started | |
| if stage is None: | |
| pending = executor.submit(self._oversized_chunk, piece, byte_count, | |
| diagnostic=item, **keywords) | |
| else: | |
| pending = executor.submit(self._complete_gpu_chunk, piece, stage, | |
| diagnostic=item, **keywords) | |
| # This main-lane interval is separate from the worker-owned record. | |
| _record_interval(call_trace, "completion_submit", "main", submit_started, | |
| time.perf_counter()) | |
| pending_offset, pending_item = global_offset, item | |
| if pending is not None: | |
| wait_and_collect(pending, pending_offset, pending_item) | |
| pending = None | |
| except BaseException as foreground_error: | |
| if pending is not None: | |
| try: | |
| pending.result() | |
| except BaseException as background_error: | |
| if background_error is not foreground_error: | |
| raise foreground_error from background_error | |
| raise | |
| shutdown_started = time.perf_counter() | |
| _record_interval(call_trace, "executor_shutdown", "main", shutdown_started, time.perf_counter()) | |
| finalize_started = time.perf_counter() | |
| facts = pa.concat_tables(output).combine_chunks() if output else pa.Table.from_batches([], schema=_fact_schema()) | |
| batch = (facts.to_batches(max_chunksize=max(1, facts.num_rows))[0] if facts.num_rows | |
| else pa.RecordBatch.from_arrays([pa.array([], type=field.type) for field in facts.schema], schema=facts.schema)) | |
| finalized = time.perf_counter() | |
| _record_interval(call_trace, "final_output_assembly", "main", finalize_started, finalized) | |
| for item in diagnostic["chunks"]: | |
| del item["_origin"] | |
| diagnostic["fallback_bytes_by_reason"] = { | |
| reason: sum(item["fallback_bytes_by_reason"].get(reason, 0) for item in diagnostic["chunks"]) | |
| for reason in ("unicode", "candidate_overflow", "oversized", "invalid_utf8", "normalization_capacity")} | |
| diagnostic["transfer_payload_bytes"] = { | |
| direction: sum(sum(item["transfer_payload_bytes"][direction].values()) | |
| for item in diagnostic["chunks"]) | |
| for direction in ("h2d", "d2h")} | |
| diagnostic["transfer_payload_bytes"]["initialization_h2d"] = diagnostic["initialization_h2d_config_bytes"] | |
| completed = time.perf_counter() | |
| _record_interval(call_trace, "diagnostic_aggregation", "main", finalized, completed) | |
| diagnostic["call_total_seconds"] = completed - started | |
| profile = {**self.backend_report, **aggregate, "rows": data.num_rows, | |
| "fact_rows": batch.num_rows, "total_seconds": completed - started, | |
| "device_workspace_bytes": sum(value.nbytes for value in self._workspace.values()), | |
| "device_config_bytes": sum(value.nbytes for value in self._device_config.values()), | |
| "diagnostics": diagnostic} | |
| self.last_profile = profile | |
| return GpuScreenResult(data.num_rows, batch, profile) | |
| def adversarial_inputs(): | |
| """Small all-column parity inputs; no execution or GPU initialization on import.""" | |
| texts = [ | |
| "", "nothing matches here", "SPY espionage spy agency", "SPY ETF", | |
| "SPY shares S&P 500", "SPDR S&P 500 ETF Trust", "Invesco QQQ ETF", | |
| "Nasdaq-100", "Nasdaq 100", "rate-cut", "interest-rate cut", | |
| "PowerShares QQQ Trust, Series 1", "CPI consumer price index; GDP growth", | |
| "Fed", "Fed inflation", "SEC proposes rule for public companies", | |
| "SEC proposes rule", "U.S. Securities and Exchange Commission adopts regulation for all exchanges", | |
| "rate-cut; interest-rate cut; Federal Reserve", "federal\x1c\x1dreserve", | |
| "federal\x00reserve", "CPIx xCPI _CPI CPI_", "İ ſ K ı CPI inflation", | |
| "S&P 500 — Federal Reserve", "SPY ETF " + "é", "Federal Reserve " * 140, | |
| "x" * 65537, "ETF " + "x" * 126 + " SPY", "SPY " + "x" * 126 + " ETF", | |
| ] | |
| return [(text, timestamp) for timestamp in ( | |
| "1990-01-01T00:00:00Z", "2018-06-04T03:59:59.999999Z", "2018-06-04T04:00:00Z", | |
| ) for text in texts] | |
| def validate_adversarial_parity(backend, *, workers=10): | |
| """Explicit small GPU check, including both flags omitted by older digests. | |
| This helper is never called automatically. Run only after compiling/loading | |
| the candidate, before accepting any throughput measurement. | |
| """ | |
| from datetime import datetime | |
| import pyarrow as pa | |
| cases = adversarial_inputs() | |
| summaries = [] | |
| for text_type in (pa.string(), pa.large_string()): | |
| # Nonzero Arrow array offsets exercise sliced UTF-8 and timestamp buffers. | |
| texts = ["padding"] + [text for text, _ in cases] + ["padding"] | |
| timestamps = [datetime.fromisoformat("2020-01-01T00:00:00+00:00")] | |
| timestamps.extend(datetime.fromisoformat(value.replace("Z", "+00:00")) for _, value in cases) | |
| timestamps.append(timestamps[0]) | |
| batch = pa.record_batch( | |
| [pa.array(texts, type=text_type), pa.array(timestamps, type=pa.timestamp("us", tz="UTC"))], | |
| names=["text", "timestamp"], | |
| ).slice(1, len(cases)) | |
| expected = _materialize(backend.cpu_engine.screen_arrow(batch, workers=workers, row_count=batch.num_rows)) | |
| observed = backend.screen_arrow(batch, workers=workers, row_count=batch.num_rows) | |
| expected = pa.Table.from_batches([expected]) if isinstance(expected, pa.RecordBatch) else expected | |
| actual = pa.Table.from_batches([observed.to_pyarrow()]) | |
| for name in FACT_NAMES: | |
| if not expected.column(name).equals(actual.column(name)): | |
| raise AssertionError(f"Phase29 adversarial {text_type} parity failed: {name}") | |
| summaries.append({"text_type": str(text_type), "rows": batch.num_rows, | |
| "fact_rows": actual.num_rows, "all_nine_columns_equal": True, | |
| "profile": observed.profile}) | |
| return summaries | |