Orchestrata Ultimate › Plans › codemod-engine-briefs › BRIEF-W5
BRIEF-W5: C Extension (Qwen Max Agent)
Source: plans/codemod-engine-briefs/BRIEF-W5.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-07-27 · 718 lines.
Prose pending.
Plan text
BRIEF-W5: C Extension (Qwen Max Agent)
Wave Overview
Wave: W5 (C Extension)
Agent: Qwen Max (C systems programming, CPython C-API expertise)
Duration: ~20 minutes
Dependencies: W1 (CSR index Python implementation for parity reference)
Parallel Execution: No (single agent)
Objective
Implement the _codemod_fast C extension for performance-critical operations in the codemod engine. The extension provides:
- CSR build + BFS in C (10x faster than Python
array.array) - xxhash3 wrapper (5 GB/s vs SHA-256's 500 MB/s)
- mmap file loading (zero-copy parse source)
The C extension is an optional optimization. The engine MUST work without it via pure-Python fallbacks.
Implements: REQ-110-129 (Py_mod_gil = Py_MOD_GIL_NOT_USED), REQ-110-130 (GIL probe fast-fail), design decision D12 (C extension follows janus-swi vendored-fork pattern).
Scope
Files to Create
-
scripts/_codemod_fast/pyproject.toml(~30 LOC)- setuptools build-backend
- Package metadata for
codemod-fast
-
scripts/_codemod_fast/setup.py(~40 LOC)Extension('codemod_fast._native', ...)with C source files- Compiler flags:
-O3 -march=native -fPIC - xxhash3 bundled or linked
-
scripts/_codemod_fast/codemod_fast/__init__.py(~60 LOC)- Python fallback pattern:
try import _native, fall back to pure Python - Public API:
csr_build,csr_bfs,xxhash3_64,mmap_load
- Python fallback pattern:
-
scripts/_codemod_fast/codemod_fast/csr_index.c(~250 LOC)csr_build(node_count, edges)→(offsets_array, neighbors_array)csr_bfs(offsets, neighbors, start, max_depth)→set[int]- CSR construction in C with direct
int[]manipulation - BFS with visited bitmap (no Python set overhead)
-
scripts/_codemod_fast/codemod_fast/content_hash.c(~120 LOC)xxhash3_64(data: bytes) -> str(16-char hex digest)- Wraps xxhash3's
XXH3_64bits()fromxxhash.h - Accepts
Py_bufferfor zero-copy from Python bytes
-
scripts/_codemod_fast/codemod_fast/mmap_loader.c(~100 LOC)mmap_load(path: str) -> bytes- Uses
mmap()withMAP_PRIVATE | MAP_POPULATEfor read-ahead - Returns
PyBytesfrom mapped region (zero-copy where possible)
-
scripts/_codemod_fast/codemod_fast/module.c(~80 LOC)PyModuleDefwith method table- CRITICAL:
Py_mod_gil = Py_MOD_GIL_NOT_USEDdeclaration - Module initialization via
PyModuleDef_Init
-
tests/refactor/codemod/test_c_extension.py(~200 LOC)- Parity tests: C and Python produce identical results
- GIL probe: verify
Py_MOD_GIL_NOT_USEDis effective - Performance measurement: C is measurably faster
Files to Modify
pyproject.toml(root)- Add
[tool.uv.sources]entry:codemod-fast = { path = "scripts/_codemod_fast" } - Add to
no-build-isolation-package:["janus-swi", "codemod-fast"]
- Add
Instructions
Step 1: Create Sub-Package Structure
Follow the janus-swi vendored-fork pattern from the orchestrata-ultimate project. The sub-package lives under scripts/_codemod_fast/ and is wired into the root pyproject.toml via [tool.uv.sources].
mkdir -p scripts/_codemod_fast/codemod_fast
Create the directory structure:
scripts/_codemod_fast/
pyproject.toml
setup.py
codemod_fast/
__init__.py
csr_index.c
content_hash.c
mmap_loader.c
module.c
Step 2: Create scripts/_codemod_fast/pyproject.toml
[build-system]
requires = ["setuptools>=70", "wheel>=0.43"]
build-backend = "setuptools.build_meta"
[project]
name = "codemod-fast"
version = "0.1.0"
description = "C extension for codemod engine performance-critical operations"
requires-python = ">=3.14"
[tool.setuptools.packages.find]
include = ["codemod_fast*"]
Step 3: Create scripts/_codemod_fast/setup.py
from setuptools import setup, Extension
native = Extension(
"codemod_fast._native",
sources=[
"codemod_fast/module.c",
"codemod_fast/csr_index.c",
"codemod_fast/content_hash.c",
"codemod_fast/mmap_loader.c",
],
extra_compile_args=["-O3", "-march=native", "-fPIC"],
include_dirs=["codemod_fast"],
)
setup(
ext_modules=[native],
)
Note: If xxhash3 is not available as a system library, bundle xxhash.h and xxhash.c (single-header mode: #define XXH_INLINE_ALL) in the codemod_fast/ directory and add xxhash.c to the sources list. The xxhash library is MIT-licensed.
Step 4: Implement csr_index.c
This provides two functions:
csr_build(node_count, edges): Takes an integer node count and a list of (src, dst) tuples. Returns a tuple of two Python array.array('i') objects: (offsets, neighbors) for the forward CSR.
Key implementation details:
- Allocate C
int[]arrays for degree counting, offsets, and neighbor indices - Count out-degree per node in a single pass over edges
- Build prefix-sum offsets array
- Scatter destination node IDs into neighbor array using cursor array
- Wrap final C arrays into Python
array.arrayobjects viaPyByteArray_FromStringAndSizeor direct buffer copy - Free all C allocations before returning
csr_bfs(offsets, neighbors, start, max_depth): Takes CSR arrays, a start node ID, and a max depth (-1 for unlimited). Returns a Python set of reachable node IDs.
Key implementation details:
- Use a C
int[]queue (pre-allocated tonode_count) - Use a C
char[]visited bitmap (1 byte per node, faster than bit packing) - Process BFS level-by-level using two pointers (current level start/end)
- Stop when
max_depthreached or queue empty - Build Python
setfrom visited nodes at the end
/* csr_index.c - CSR build and BFS in C */
#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <stdlib.h>
#include <string.h>
/* csr_build(node_count: int, edges: list[tuple[int, int]]) -> tuple[array, array] */
static PyObject* csr_build(PyObject* self, PyObject* args) {
int node_count;
PyObject* edges_list;
if (!PyArg_ParseTuple(args, "iO", &node_count, &edges_list))
return NULL;
Py_ssize_t edge_count = PyList_Size(edges_list);
/* Allocate degree counters and offsets */
int* out_degree = (int*)calloc(node_count, sizeof(int));
int* offsets = (int*)calloc(node_count + 1, sizeof(int));
int* neighbors = (int*)malloc(edge_count * sizeof(int));
int* cursor = (int*)malloc(node_count * sizeof(int));
/* Count out-degrees */
for (Py_ssize_t i = 0; i < edge_count; i++) {
PyObject* edge = PyList_GetItem(edges_list, i);
int src = (int)PyLong_AsLong(PyTuple_GetItem(edge, 0));
out_degree[src]++;
}
/* Build prefix-sum offsets */
for (int i = 0; i < node_count; i++) {
offsets[i + 1] = offsets[i] + out_degree[i];
cursor[i] = offsets[i];
}
/* Scatter neighbors */
for (Py_ssize_t i = 0; i < edge_count; i++) {
PyObject* edge = PyList_GetItem(edges_list, i);
int src = (int)PyLong_AsLong(PyTuple_GetItem(edge, 0));
int dst = (int)PyLong_AsLong(PyTuple_GetItem(edge, 1));
neighbors[cursor[src]++] = dst;
}
/* Build Python array.array objects */
/* ... wrap offsets and neighbors into array.array('i') ... */
free(out_degree);
free(cursor);
/* offsets and neighbors are transferred to Python objects */
/* Return tuple */
/* ... */
}
/* csr_bfs(offsets, neighbors, start, max_depth) -> set[int] */
static PyObject* csr_bfs(PyObject* self, PyObject* args) {
/* Parse CSR array.array buffers, run BFS with visited bitmap */
/* ... */
}
Step 5: Implement content_hash.c
Wraps xxhash3's XXH3_64bits() for high-speed content hashing.
/* content_hash.c - xxhash3 wrapper */
#define PY_SSIZE_T_CLEAN
#include <Python.h>
/* If xxhash is bundled inline: */
#define XXH_INLINE_ALL
#include "xxhash.h"
/* xxhash3_64(data: bytes) -> str (16-char hex) */
static PyObject* content_hash_xxhash3_64(PyObject* self, PyObject* args) {
Py_buffer buf;
if (!PyArg_ParseTuple(args, "y*", &buf))
return NULL;
XXH64_hash_t hash = XXH3_64bits(buf.buf, buf.len);
PyBuffer_Release(&buf);
char hex[17];
snprintf(hex, sizeof(hex), "%016llx", (unsigned long long)hash);
return PyUnicode_FromString(hex);
}
Performance target: 5 GB/s throughput on large files (vs SHA-256 at ~500 MB/s).
Step 6: Implement mmap_loader.c
Uses mmap() for zero-copy file loading.
/* mmap_loader.c - mmap file loading */
#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
/* mmap_load(path: str) -> bytes */
static PyObject* mmap_loader_load(PyObject* self, PyObject* args) {
const char* path;
if (!PyArg_ParseTuple(args, "s", &path))
return NULL;
int fd = open(path, O_RDONLY);
if (fd < 0) {
PyErr_SetFromErrnoWithFilename(PyExc_OSError, path);
return NULL;
}
struct stat st;
if (fstat(fd, &st) < 0) {
close(fd);
PyErr_SetFromErrnoWithFilename(PyExc_OSError, path);
return NULL;
}
size_t size = st.st_size;
if (size == 0) {
close(fd);
return PyBytes_FromStringAndSize("", 0);
}
void* mapped = mmap(NULL, size, PROT_READ, MAP_PRIVATE | MAP_POPULATE, fd, 0);
close(fd); /* fd no longer needed after mmap */
if (mapped == MAP_FAILED) {
PyErr_SetFromErrnoWithFilename(PyExc_OSError, path);
return NULL;
}
PyObject* result = PyBytes_FromStringAndSize((char*)mapped, size);
munmap(mapped, size);
return result;
}
Note: MAP_POPULATE triggers read-ahead on Linux, warming the page cache. On non-Linux platforms, fall back to MAP_PRIVATE only.
Step 7: Implement module.c (CRITICAL)
This is the module definition file. It MUST declare Py_MOD_GIL_NOT_USED or the entire process loses free-threading capability.
/* module.c - PyModuleDef with GIL-free declaration */
#define PY_SSIZE_T_CLEAN
#include <Python.h>
/* Method table - declared in other .c files */
extern PyObject* csr_build(PyObject*, PyObject*);
extern PyObject* csr_bfs(PyObject*, PyObject*);
extern PyObject* content_hash_xxhash3_64(PyObject*, PyObject*);
extern PyObject* mmap_loader_load(PyObject*, PyObject*);
static PyMethodDef module_methods[] = {
{"csr_build", csr_build, METH_VARARGS,
"Build CSR adjacency arrays from edge list."},
{"csr_bfs", csr_bfs, METH_VARARGS,
"BFS transitive closure on CSR index."},
{"xxhash3_64", content_hash_xxhash3_64, METH_VARARGS,
"Compute xxhash3 64-bit hash of data."},
{"mmap_load", mmap_loader_load, METH_VARARGS,
"Load file contents via mmap."},
{NULL, NULL, 0, NULL}
};
static struct PyModuleDef_Slot module_slots[] = {
{Py_mod_gil, Py_MOD_GIL_NOT_USED}, /* CRITICAL: free-threading compat */
{0, NULL}
};
static struct PyModuleDef module_def = {
PyModuleDef_HEAD_INIT,
.m_name = "codemod_fast._native",
.m_doc = "C extension for codemod engine (CSR, xxhash3, mmap)",
.m_size = 0,
.m_methods = module_methods,
.m_slots = module_slots,
};
PyMODINIT_FUNC PyInit__native(void) {
return PyModuleDef_Init(&module_def);
}
CRITICAL: The {Py_mod_gil, Py_MOD_GIL_NOT_USED} slot entry is what tells CPython 3.13+ that this extension is safe to use without the GIL. Without it, importing this module will re-enable the GIL for the entire process, defeating free-threading. This is enforced by tests/conftest.py's GIL probe (REQ-110-130).
Step 8: Create codemod_fast/__init__.py (Python Fallback)
"""codemod_fast: C-accelerated operations with pure-Python fallback."""
from __future__ import annotations
from pathlib import Path
import array
from collections import deque
try:
from codemod_fast._native import (
csr_build as _native_csr_build,
csr_bfs as _native_csr_bfs,
xxhash3_64 as _native_xxhash3_64,
mmap_load as _native_mmap_load,
)
_HAS_NATIVE = True
except ImportError:
_HAS_NATIVE = False
def csr_build(
node_count: int, edges: list[tuple[int, int]]
) -> tuple[array.array, array.array]:
if _HAS_NATIVE:
return _native_csr_build(node_count, edges)
return _py_csr_build(node_count, edges)
def csr_bfs(
offsets: array.array,
neighbors: array.array,
start: int,
max_depth: int = -1,
) -> set[int]:
if _HAS_NATIVE:
return _native_csr_bfs(offsets, neighbors, start, max_depth)
return _py_csr_bfs(offsets, neighbors, start, max_depth)
def xxhash3_64(data: bytes) -> str:
if _HAS_NATIVE:
return _native_xxhash3_64(data)
return _py_xxhash3_64(data)
def mmap_load(path: str | Path) -> bytes:
if _HAS_NATIVE:
return _native_mmap_load(str(path))
return _py_mmap_load(path)
def _py_csr_build(
node_count: int, edges: list[tuple[int, int]]
) -> tuple[array.array, array.array]:
out_degree = [0] * node_count
for src, _ in edges:
out_degree[src] += 1
offsets = array.array("i", [0] * (node_count + 1))
for i in range(node_count):
offsets[i + 1] = offsets[i] + out_degree[i]
neighbors = array.array("i", [0] * len(edges))
cursor = list(offsets[:-1])
for src, dst in edges:
neighbors[cursor[src]] = dst
cursor[src] += 1
return offsets, neighbors
def _py_csr_bfs(
offsets: array.array,
neighbors: array.array,
start: int,
max_depth: int = -1,
) -> set[int]:
visited: set[int] = set()
queue: deque[tuple[int, int]] = deque([(start, 0)])
while queue:
node, depth = queue.popleft()
if node in visited:
continue
visited.add(node)
if 0 <= max_depth <= depth:
continue
lo, hi = offsets[node], offsets[node + 1]
for i in range(lo, hi):
nb = neighbors[i]
if nb not in visited:
queue.append((nb, depth + 1))
return visited
def _py_xxhash3_64(data: bytes) -> str:
try:
import xxhash
return xxhash.xxh3_64(data).hexdigest()
except ImportError:
import hashlib
return hashlib.sha256(data).hexdigest()[:16]
def _py_mmap_load(path: str | Path) -> bytes:
return Path(path).read_bytes()
Step 9: Wire in Root pyproject.toml
Add to the root project's pyproject.toml:
[tool.uv.sources]
codemod-fast = { path = "scripts/_codemod_fast" }
[tool.uv]
no-build-isolation-package = ["janus-swi", "codemod-fast"]
The no-build-isolation-package setting ensures the C extension is compiled against the project's free-threaded Python 3.14t interpreter (with Py_GIL_DISABLED), not an isolated build environment that might lack free-threading support.
After editing pyproject.toml, run:
uv sync --dev
This builds the C extension and installs it into the project's virtual environment.
Step 10: Write Parity Tests
Create tests/refactor/codemod/test_c_extension.py:
"""Parity tests: verify C and Python implementations produce identical results.
These tests import both the native C functions and the pure-Python fallbacks
and verify they produce identical output for the same input.
"""
from __future__ import annotations
import array
import time
from pathlib import Path
import pytest
def _get_native_and_python():
"""Import native C functions and Python fallbacks."""
try:
from codemod_fast._native import (
csr_build as native_csr_build,
csr_bfs as native_csr_bfs,
xxhash3_64 as native_xxhash3_64,
mmap_load as native_mmap_load,
)
has_native = True
except ImportError:
has_native = False
from codemod_fast import (
_py_csr_build,
_py_csr_bfs,
_py_xxhash3_64,
_py_mmap_load,
)
return has_native, {
"native": {
"csr_build": native_csr_build if has_native else None,
"csr_bfs": native_csr_bfs if has_native else None,
"xxhash3_64": native_xxhash3_64 if has_native else None,
"mmap_load": native_mmap_load if has_native else None,
},
"python": {
"csr_build": _py_csr_build,
"csr_bfs": _py_csr_bfs,
"xxhash3_64": _py_xxhash3_64,
"mmap_load": _py_mmap_load,
},
}
@pytest.mark.skipif(
not _get_native_and_python()[0],
reason="C extension not built",
)
class TestCSRParity:
def test_csr_build_empty(self):
has_native, impls = _get_native_and_python()
n_off, n_nb = impls["native"]["csr_build"](5, [])
p_off, p_nb = impls["python"]["csr_build"](5, [])
assert list(n_off) == list(p_off)
assert list(n_nb) == list(p_nb)
def test_csr_build_simple_graph(self):
has_native, impls = _get_native_and_python()
edges = [(0, 1), (0, 2), (1, 3), (2, 3), (3, 4)]
n_off, n_nb = impls["native"]["csr_build"](5, edges)
p_off, p_nb = impls["python"]["csr_build"](5, edges)
assert list(n_off) == list(p_off)
assert sorted(list(n_nb)) == sorted(list(p_nb))
def test_csr_bfs_full_graph(self):
has_native, impls = _get_native_and_python()
edges = [(0, 1), (0, 2), (1, 3), (2, 3), (3, 4)]
off, nb = impls["python"]["csr_build"](5, edges)
n_result = impls["native"]["csr_bfs"](off, nb, 0, -1)
p_result = impls["python"]["csr_bfs"](off, nb, 0, -1)
assert n_result == p_result
def test_csr_bfs_depth_limited(self):
has_native, impls = _get_native_and_python()
edges = [(0, 1), (1, 2), (2, 3), (3, 4)]
off, nb = impls["python"]["csr_build"](5, edges)
n_result = impls["native"]["csr_bfs"](off, nb, 0, 2)
p_result = impls["python"]["csr_bfs"](off, nb, 0, 2)
assert n_result == p_result
class TestHashParity:
def test_xxhash3_64_empty(self):
has_native, impls = _get_native_and_python()
if not has_native:
pytest.skip("C extension not built")
assert impls["native"]["xxhash3_64"](b"") == impls["python"]["xxhash3_64"](b"")
def test_xxhash3_64_data(self):
has_native, impls = _get_native_and_python()
if not has_native:
pytest.skip("C extension not built")
data = b"Hello, codemod engine!"
assert impls["native"]["xxhash3_64"](data) == impls["python"]["xxhash3_64"](data)
class TestMmapParity:
def test_mmap_load_file(self, tmp_path: Path):
has_native, impls = _get_native_and_python()
if not has_native:
pytest.skip("C extension not built")
test_file = tmp_path / "test.py"
test_file.write_text("x = 42\n")
assert impls["native"]["mmap_load"](str(test_file)) == impls["python"]["mmap_load"](test_file)
class TestGILDeclaration:
def test_py_mod_gil_not_used(self):
"""REQ-110-129: C extension declares Py_mod_gil = Py_MOD_GIL_NOT_USED."""
try:
import codemod_fast._native as native
except ImportError:
pytest.skip("C extension not built")
# If we got here on a free-threaded build without the GIL probe
# fast-failing, the declaration is effective.
import sys
if hasattr(sys, "_is_gil_enabled") and sys._is_gil_enabled():
pytest.fail("GIL is enabled despite Py_MOD_GIL_NOT_USED declaration")
class TestPerformance:
def test_csr_build_faster(self):
has_native, impls = _get_native_and_python()
if not has_native:
pytest.skip("C extension not built")
edges = [(i % 100, (i + 1) % 100) for i in range(10000)]
start = time.perf_counter()
for _ in range(100):
impls["native"]["csr_build"](100, edges)
native_time = time.perf_counter() - start
start = time.perf_counter()
for _ in range(100):
impls["python"]["csr_build"](100, edges)
python_time = time.perf_counter() - start
# C should be at least 2x faster (target is 10x)
assert native_time < python_time, (
f"C ({native_time:.3f}s) not faster than Python ({python_time:.3f}s)"
)
Step 11: Verify the Build
# Build the C extension
uv sync --dev
# Verify the extension loads
uv run python -c "from codemod_fast._native import csr_build, csr_bfs, xxhash3_64, mmap_load; print('Native OK')"
# Verify fallback works when native is unavailable
uv run python -c "from codemod_fast import csr_build, xxhash3_64; print('Fallback OK')"
# Verify GIL declaration
uv run python -c "
import sys
import codemod_fast._native
if hasattr(sys, '_is_gil_enabled') and sys._is_gil_enabled():
print('FAIL: GIL is enabled')
sys.exit(1)
else:
print('GIL-free OK')
"
# Run parity tests
uv run pytest tests/refactor/codemod/test_c_extension.py -v
Success Criteria
- [ ] C extension builds successfully via
uv sync --dev - [ ]
Py_mod_gil = Py_MOD_GIL_NOT_USEDdeclared inmodule.c(REQ-110-129) - [ ] GIL probe does not fast-fail after importing
_native(REQ-110-130) - [ ] All parity tests pass (C and Python produce identical results)
- [ ] CSR build in C is measurably faster than Python (target: 10x, minimum: 2x)
- [ ] xxhash3 produces correct 16-char hex digests
- [ ] mmap_load reads files correctly (including empty files)
- [ ] Python fallback works when C extension is not available
- [ ] Root
pyproject.tomlwired with[tool.uv.sources]andno-build-isolation-package - [ ] No compiler warnings with
-Wall -Wextra
Deliverables
scripts/_codemod_fast/pyproject.toml— sub-package build configscripts/_codemod_fast/setup.py— C extension definitionscripts/_codemod_fast/codemod_fast/__init__.py— Python fallback layerscripts/_codemod_fast/codemod_fast/csr_index.c— CSR build + BFS in Cscripts/_codemod_fast/codemod_fast/content_hash.c— xxhash3 wrapperscripts/_codemod_fast/codemod_fast/mmap_loader.c— mmap file loadingscripts/_codemod_fast/codemod_fast/module.c— PyModuleDef with GIL-free declarationtests/refactor/codemod/test_c_extension.py— parity + GIL + performance testspyproject.toml(root, modified) — uv sources and no-build-isolation wiring
Handoff to W6
W6 (Quality) will:
- Run all gates (
ruff,black,standards,check-deps,test-review,pyright_ratchet) - Verify coverage ≥90% for
codemod_fast/__init__.py(the Python fallback layer) - Run the full test suite via
python scripts/run_tests.py tests/refactor/codemod/ - Verify the C extension does not break any existing tests
- Confirm GIL probe passes in the test conftest
Notes
- Single agent: This wave uses one agent (Qwen Max) because C extension development requires tight coordination between the C source files and the module definition.
- Optional optimization: The C extension is a performance enhancement, not a functional requirement. The engine works correctly with pure-Python fallbacks. If the C extension fails to build on any platform, the engine continues to function.
- Linux only: The
mmap_loader.cuses Linux-specific flags (MAP_POPULATE). On macOS/BSD, these flags are absent and the code should#ifdefthem out. The target platform for this project is Linux (Incus containers). - xxhash3 bundling: If the system does not have xxhash installed, bundle the single-header
xxhash.hwithXXH_INLINE_ALLdefined. The xxhash library is MIT-licensed and ~3,000 LOC in single-header mode. - Free-threading is non-negotiable: The
Py_MOD_GIL_NOT_USEDdeclaration is the single most important line in this wave. Without it, importing the C extension re-enables the GIL for the entire process, negating all free-threading benefits. Thetests/conftest.pyGIL probe will fast-fail the test suite if this is missing. - Compiler flags: Use
-O3 -march=native -fPIC. The-march=nativeflag enables SIMD instructions specific to the build machine, which benefits xxhash3's AVX2/SSE paths. - No
except: pass: The fallbacktry/except ImportErrorin__init__.pyis intentional and correct — it must catch onlyImportError, not all exceptions.