Bind HIL IRQ observers through private nRF framework copies
This commit is contained in:
mikecarper committed
1 parent
0bcc9e3e9e
commit
7a6a118775
4 files changed
+320
No files matched your search
@@ -820,6 +820,7 @@ jobs:
|
||||
python3 -B test/test_nrf_radio_priority_integration.py -v
|
||||
python3 -B test/test_nrf_radio_usb_irq_guard.py -v
|
||||
python3 -B test/test_nrf_radio_irq_observer.py -v
|
||||
python3 -B test/test_nrf52_hil_irq_binding.py -v
|
||||
python3 -B test/test_nrf_radio_spi_startup_diag.py -v
|
||||
python3 -B test/test_nrf_radio_spi_read_probe.py -v
|
||||
python3 -B test/test_sx126x_nrf_rp_constructors.py -v
|
||||
|
||||
@@ -177,6 +177,7 @@ extra_scripts = create-uf2.py
|
||||
pre:scripts/nrf52_ble_cccd_fix.py
|
||||
pre:scripts/nrf52_ble_dfu_fix.py
|
||||
pre:scripts/nrf52_loop_stack_fix.py
|
||||
pre:scripts/nrf52_hil_irq_observer.py
|
||||
pre:scripts/nrf52_internal_bootloader_link.py
|
||||
; Nordic's builder otherwise overwrites the LTO-aware archive tools.
|
||||
pre:scripts/nrf52_lto_toolchain.py
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
"""Bind the optional HIL IRQ observer in private nRF52840 framework copies.
|
||||
|
||||
The framework's same-object weak vector aliases and its LTO-visible scheduler
|
||||
call bypass GNU --wrap. Only for MESH_HIL_IRQ_TIMING=1, name the wrappers at
|
||||
those seven call sites explicitly. Keep each real handler, SVC/PendSV vector,
|
||||
and the naked port's register/exception-return sequence unchanged. The linker
|
||||
wrap options still resolve the observer's __real_* calls to the real handlers.
|
||||
No installed framework source is modified.
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
IRQ_HANDLERS = (
|
||||
"POWER_CLOCK_IRQHandler", "UARTE0_UART0_IRQHandler", "GPIOTE_IRQHandler",
|
||||
"RTC1_IRQHandler", "USBD_IRQHandler", "UARTE1_IRQHandler",
|
||||
)
|
||||
WRAPPED_SYMBOLS = IRQ_HANDLERS + ("vTaskSwitchContext",)
|
||||
PORT_CALL = ' " bl vTaskSwitchContext \\n"\n'
|
||||
WRAPPED_PORT_CALL = PORT_CALL.replace("bl vTaskSwitchContext", "bl __wrap_vTaskSwitchContext")
|
||||
PORT_DECLARATION = "void xPortPendSVHandler( void ) __attribute__ (( naked ));"
|
||||
PORT_DEFINITION = "void xPortPendSVHandler( void )\n{\n /* This is a naked function. */"
|
||||
|
||||
|
||||
def _error(part):
|
||||
return RuntimeError(
|
||||
"nRF52 HIL IRQ observer: unrecognized " + part + "; review the pinned "
|
||||
"framework before building (shared SDK not modified)")
|
||||
|
||||
|
||||
def patched_startup(source):
|
||||
source = source.replace("\r\n", "\n")
|
||||
pairs = [(" .long " + symbol + "\n",
|
||||
" .long __wrap_" + symbol + "\n") for symbol in IRQ_HANDLERS]
|
||||
# Fail closed on partial patches, duplicate entries, or changed weak aliases.
|
||||
for symbol in IRQ_HANDLERS:
|
||||
if source.count(" IRQ " + symbol + "\n") != 1:
|
||||
raise _error("startup weak IRQ aliases")
|
||||
for symbol in ("SVC_Handler", "PendSV_Handler"):
|
||||
if source.count(" .long " + symbol + "\n") != 1:
|
||||
raise _error("startup exception vectors")
|
||||
original = all(source.count(old) == 1 and new not in source for old, new in pairs)
|
||||
patched = all(source.count(new) == 1 and old not in source for old, new in pairs)
|
||||
if not original and not patched:
|
||||
raise _error("startup IRQ vector entries")
|
||||
for old, new in pairs:
|
||||
source = source.replace(old, new)
|
||||
return source
|
||||
|
||||
|
||||
def patched_port(source):
|
||||
source = source.replace("\r\n", "\n")
|
||||
if (source.count(PORT_DECLARATION) != 1 or source.count(PORT_DEFINITION) != 1):
|
||||
raise _error("GCC naked PendSV definition")
|
||||
if source.count(WRAPPED_PORT_CALL) == 1 and PORT_CALL not in source:
|
||||
return source
|
||||
if source.count(PORT_CALL) != 1 or WRAPPED_PORT_CALL in source:
|
||||
raise _error("GCC PendSV scheduler call")
|
||||
return source.replace(PORT_CALL, WRAPPED_PORT_CALL)
|
||||
|
||||
|
||||
def _private_copy(build_env, node, patch, relative):
|
||||
source = Path(node.srcnode().get_abspath())
|
||||
patched = patch(source.read_text(encoding="utf-8"))
|
||||
destination = Path(build_env.subst("$BUILD_DIR")) / "patched-nrf52-hil-irqs" / relative
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
if not destination.exists() or destination.read_text(encoding="utf-8") != patched:
|
||||
destination.write_text(patched, encoding="utf-8")
|
||||
# Preserve any quoted, source-relative includes after relocation.
|
||||
build_env.AppendUnique(CPPPATH=[str(source.parent)])
|
||||
return build_env.File(str(destination))
|
||||
|
||||
|
||||
def replace_startup(build_env, node):
|
||||
return _private_copy(build_env, node, patched_startup, "gcc_startup_nrf52840.S")
|
||||
|
||||
|
||||
def replace_port(build_env, node):
|
||||
return _private_copy(build_env, node, patched_port, "port.c")
|
||||
|
||||
|
||||
def observer_enabled(flags=(), cppdefines=()):
|
||||
# Reuse the repository's effective -D/-U parser, including split -D values.
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"mesh_nrf52_flash_trim_irq_flags", Path(__file__).with_name("nrf52_flash_trim.py"))
|
||||
parser = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(parser)
|
||||
values = parser.flag_defines(flags, cppdefines)
|
||||
value = values.get("MESH_HIL_IRQ_TIMING")
|
||||
if value is None:
|
||||
return False
|
||||
value = str(value).strip("() ")
|
||||
if value not in ("0", "1"):
|
||||
raise RuntimeError("nRF52 HIL IRQ observer flag must be 0 or 1")
|
||||
return value == "1"
|
||||
|
||||
|
||||
def install(build_env):
|
||||
if not observer_enabled(build_env.get("BUILD_FLAGS", ()), build_env.get("CPPDEFINES", ())):
|
||||
return
|
||||
if (build_env.PioPlatform().name != "nordicnrf52"
|
||||
or build_env.BoardConfig().get("build.mcu", "").lower() != "nrf52840"):
|
||||
raise RuntimeError("nRF52 HIL IRQ observer binding requires the pinned nRF52840 core")
|
||||
build_env.AppendUnique(LINKFLAGS=["-Wl,--wrap=" + name for name in WRAPPED_SYMBOLS])
|
||||
build_env.AddBuildMiddleware(replace_startup, "*cores*nRF5*linker*gcc_startup_nrf52840.S")
|
||||
build_env.AddBuildMiddleware(replace_port, "*cores*nRF5*freertos*portable*GCC*nrf52*port.c")
|
||||
|
||||
|
||||
if "Import" in globals():
|
||||
Import("env")
|
||||
install(env)
|
||||
@@ -0,0 +1,206 @@
|
||||
"""Verify explicit HIL observer bindings without editing the installed core."""
|
||||
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import platform
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SPEC = importlib.util.spec_from_file_location(
|
||||
"nrf52_hil_irq_observer", ROOT / "scripts/nrf52_hil_irq_observer.py")
|
||||
PATCH = importlib.util.module_from_spec(SPEC)
|
||||
SPEC.loader.exec_module(PATCH)
|
||||
|
||||
STARTUP = """ .section .rodata
|
||||
.balign 4
|
||||
.globl test_vectors
|
||||
test_vectors:
|
||||
.long SVC_Handler
|
||||
.long PendSV_Handler
|
||||
""" + "".join(" .long " + symbol + "\n" for symbol in PATCH.IRQ_HANDLERS) + r"""
|
||||
.text
|
||||
.globl Default_Handler
|
||||
Default_Handler:
|
||||
ret
|
||||
.macro IRQ handler
|
||||
.weak \handler
|
||||
.set \handler, Default_Handler
|
||||
.endm
|
||||
""" + "".join(" IRQ " + symbol + "\n" for symbol in PATCH.IRQ_HANDLERS) + """
|
||||
.section .note.GNU-stack,"",@progbits
|
||||
"""
|
||||
PORT = """/* Existing scheduler and exception definitions stay in their object. */
|
||||
""" + PATCH.PORT_DECLARATION + "\n" + PATCH.PORT_DEFINITION + """
|
||||
__asm volatile
|
||||
(
|
||||
" stmdb sp!, {r3} \\n"
|
||||
" msr basepri, r0 \\n"
|
||||
""" + PATCH.PORT_CALL + """ " msr basepri, r0 \\n"
|
||||
" ldmia sp!, {r3} \\n"
|
||||
" bx r14 \\n"
|
||||
);
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
class Node:
|
||||
def __init__(self, path):
|
||||
self.path = path
|
||||
|
||||
def srcnode(self):
|
||||
return self
|
||||
|
||||
def get_abspath(self):
|
||||
return str(self.path)
|
||||
|
||||
|
||||
class BuildEnv(dict):
|
||||
def __init__(self, build_dir, flags=(), defines=(), mcu="nrf52840", name="nordicnrf52"):
|
||||
super().__init__(BUILD_FLAGS=flags, CPPDEFINES=defines)
|
||||
self.build_dir = build_dir
|
||||
self.mcu = mcu
|
||||
self.name = name
|
||||
self.middleware = []
|
||||
|
||||
def PioPlatform(self):
|
||||
return types.SimpleNamespace(name=self.name)
|
||||
|
||||
def BoardConfig(self):
|
||||
return {"build.mcu": self.mcu}
|
||||
|
||||
def AppendUnique(self, **kwargs):
|
||||
for name, values in kwargs.items():
|
||||
self.setdefault(name, [])
|
||||
self[name].extend(value for value in values if value not in self[name])
|
||||
|
||||
def AddBuildMiddleware(self, callback, pattern):
|
||||
self.middleware.append((callback, pattern))
|
||||
|
||||
def subst(self, value):
|
||||
assert value == "$BUILD_DIR"
|
||||
return str(self.build_dir)
|
||||
|
||||
def File(self, value):
|
||||
return Path(value)
|
||||
|
||||
|
||||
class Nrf52HilIrqBindingTest(unittest.TestCase):
|
||||
def test_startup_changes_only_six_vectors_and_preserves_all_real_aliases(self):
|
||||
patched = PATCH.patched_startup(STARTUP.replace("\n", "\r\n"))
|
||||
self.assertEqual(PATCH.patched_startup(patched), patched)
|
||||
expected = STARTUP
|
||||
for symbol in PATCH.IRQ_HANDLERS:
|
||||
expected = expected.replace(" .long " + symbol + "\n",
|
||||
" .long __wrap_" + symbol + "\n")
|
||||
self.assertEqual(patched.count(" IRQ " + symbol + "\n"), 1)
|
||||
self.assertEqual(patched, expected)
|
||||
for invalid in (
|
||||
STARTUP.replace(".long RTC1_IRQHandler", ".word RTC1_IRQHandler"),
|
||||
STARTUP.replace(".long RTC1_IRQHandler", ".long __wrap_RTC1_IRQHandler"),
|
||||
STARTUP + " .long RTC1_IRQHandler\n",
|
||||
STARTUP.replace(" IRQ RTC1_IRQHandler\n", ""),
|
||||
patched.replace(" .long SVC_Handler\n", " .long __wrap_SVC_Handler\n"),
|
||||
):
|
||||
with self.subTest(invalid=invalid), self.assertRaises(RuntimeError):
|
||||
PATCH.patched_startup(invalid)
|
||||
|
||||
def test_port_changes_only_c_call_target_in_existing_naked_assembly(self):
|
||||
patched = PATCH.patched_port(PORT.replace("\n", "\r\n"))
|
||||
self.assertEqual(patched.replace(PATCH.WRAPPED_PORT_CALL, PATCH.PORT_CALL), PORT)
|
||||
self.assertEqual(PATCH.patched_port(patched), patched)
|
||||
for invalid in (
|
||||
PORT.replace(PATCH.PORT_CALL, ""),
|
||||
PORT + PATCH.PORT_CALL,
|
||||
PORT + PATCH.WRAPPED_PORT_CALL,
|
||||
PORT.replace(PATCH.PORT_DECLARATION, ""),
|
||||
PORT.replace("/* This is a naked function. */", "/* Changed port. */"),
|
||||
):
|
||||
with self.subTest(invalid=invalid), self.assertRaises(RuntimeError):
|
||||
PATCH.patched_port(invalid)
|
||||
|
||||
def test_disabled_build_registers_no_hook_link_flag_or_private_copy(self):
|
||||
with tempfile.TemporaryDirectory(prefix="mesh-hil-irq-off-") as directory:
|
||||
for flags, defines in (((), ()), ("-DMESH_HIL_IRQ_TIMING=0", ()),
|
||||
("-UMESH_HIL_IRQ_TIMING", (("MESH_HIL_IRQ_TIMING", 1),))):
|
||||
env = BuildEnv(Path(directory) / "build", flags, defines)
|
||||
PATCH.install(env)
|
||||
self.assertEqual(env.middleware, [])
|
||||
self.assertNotIn("LINKFLAGS", env)
|
||||
self.assertFalse(env.build_dir.exists())
|
||||
|
||||
def test_enabled_build_uses_private_copies_and_wrapped_real_symbol_resolution(self):
|
||||
with tempfile.TemporaryDirectory(prefix="mesh-hil-irq-copy-") as directory:
|
||||
directory = Path(directory)
|
||||
env = BuildEnv(directory / "build", "-D MESH_HIL_IRQ_TIMING=1")
|
||||
PATCH.install(env)
|
||||
self.assertEqual(len(env.middleware), 2)
|
||||
self.assertEqual(env["LINKFLAGS"], ["-Wl,--wrap=" + name for name in PATCH.WRAPPED_SYMBOLS])
|
||||
for filename, text, replace, patch in (
|
||||
("gcc_startup_nrf52840.S", STARTUP, PATCH.replace_startup, PATCH.patched_startup),
|
||||
("port.c", PORT, PATCH.replace_port, PATCH.patched_port),
|
||||
):
|
||||
sdk = directory / "shared-sdk" / filename
|
||||
sdk.parent.mkdir(exist_ok=True)
|
||||
sdk.write_text(text)
|
||||
private = replace(env, Node(sdk))
|
||||
self.assertTrue(private.is_relative_to(env.build_dir))
|
||||
self.assertEqual(private.read_text(), patch(text))
|
||||
self.assertEqual(sdk.read_text(), text)
|
||||
stamp = private.stat().st_mtime_ns
|
||||
self.assertEqual(replace(env, Node(sdk)).stat().st_mtime_ns, stamp)
|
||||
self.assertIn(str(sdk.parent), env["CPPPATH"])
|
||||
for mcu, name in (("nrf52832", "nordicnrf52"), ("nrf52840", "espressif32")):
|
||||
with self.subTest(mcu=mcu, name=name), self.assertRaises(RuntimeError):
|
||||
PATCH.install(BuildEnv(directory, "-DMESH_HIL_IRQ_TIMING=1", mcu=mcu, name=name))
|
||||
with self.assertRaises(RuntimeError):
|
||||
PATCH.observer_enabled("-DMESH_HIL_IRQ_TIMING=2")
|
||||
self.assertTrue(PATCH.observer_enabled(cppdefines=(("MESH_HIL_IRQ_TIMING", 1),)))
|
||||
|
||||
@unittest.skipUnless(shutil.which("c++") and platform.machine() == "x86_64",
|
||||
"native x86_64 GNU toolchain unavailable")
|
||||
def test_real_weak_vector_bypasses_link_wrap_then_explicit_binding_observes_once(self):
|
||||
# The original same-object weak aliases reproduce the final-ELF failure.
|
||||
# Explicit vector targets must call each wrapper and each real handler
|
||||
# exactly once, including when the other C++ translation unit uses LTO.
|
||||
with tempfile.TemporaryDirectory(prefix="mesh-hil-irq-link-", dir="/var/tmp") as directory:
|
||||
directory = Path(directory)
|
||||
source = '#include <cassert>\n#include <cstdint>\n'
|
||||
source += 'extern "C" const uint32_t test_vectors[];\n'
|
||||
source += 'extern "C" void SVC_Handler() {}\nextern "C" void PendSV_Handler() {}\n'
|
||||
source += "static unsigned real_calls[6], wrapped_calls[6];\n"
|
||||
for index, symbol in enumerate(PATCH.IRQ_HANDLERS):
|
||||
source += f'extern "C" void {symbol}() {{ ++real_calls[{index}]; }}\n'
|
||||
source += f'extern "C" void __real_{symbol}();\n'
|
||||
source += (f'extern "C" void __wrap_{symbol}() {{ ++wrapped_calls[{index}]; '
|
||||
f'__real_{symbol}(); }}\n')
|
||||
source += "int main() { for (unsigned i = 0; i < 6; ++i) { "
|
||||
source += "reinterpret_cast<void(*)()>(uintptr_t(test_vectors[i+2]))(); "
|
||||
source += "assert(real_calls[i] == 1); assert(wrapped_calls[i] == EXPECT_WRAPPED); } }\n"
|
||||
(directory / "handlers.cpp").write_text(source)
|
||||
for patched in (False, True):
|
||||
with self.subTest(patched=patched):
|
||||
(directory / "startup.S").write_text(PATCH.patched_startup(STARTUP) if patched else STARTUP)
|
||||
binary = directory / "vectors"
|
||||
built = subprocess.run([
|
||||
"c++", "-std=c++17", "-O2", "-flto", "-fno-semantic-interposition", "-no-pie",
|
||||
f"-DEXPECT_WRAPPED={int(patched)}", str(directory / "handlers.cpp"),
|
||||
str(directory / "startup.S"),
|
||||
*("-Wl,--wrap=" + name for name in PATCH.IRQ_HANDLERS), "-o", str(binary),
|
||||
], capture_output=True, text=True, cwd=directory)
|
||||
self.assertEqual(built.returncode, 0, built.stderr)
|
||||
result = subprocess.run([str(binary)], capture_output=True, text=True)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_nrf52_base_installs_binding_hook_once(self):
|
||||
config = (ROOT / "platformio.ini").read_text()
|
||||
base = config.split("[nrf52_base]", 1)[1].split("\n[", 1)[0]
|
||||
self.assertEqual(base.count("pre:scripts/nrf52_hil_irq_observer.py"), 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user