From 57d0a58855bcebf29caeb915489dcf83bbebc993 Mon Sep 17 00:00:00 2001 From: Mitchell Goff Date: Mon, 4 May 2026 20:11:03 -0700 Subject: [PATCH] Round trip through bytes to save 30x on migration memory usage (#37963) --- selfdrive/test/process_replay/migration.py | 44 ++++++++++++---------- 1 file changed, 24 insertions(+), 20 deletions(-) diff --git a/selfdrive/test/process_replay/migration.py b/selfdrive/test/process_replay/migration.py index 12a9664a5..578dd4462 100644 --- a/selfdrive/test/process_replay/migration.py +++ b/selfdrive/test/process_replay/migration.py @@ -87,6 +87,10 @@ def migrate(lr: LogIterable, migration_funcs: list[MigrationFunc]): return lr +def as_reader(builder) -> capnp.lib.capnp._DynamicStructReader: + return log.Event.from_bytes(builder.to_bytes()).__enter__() # round-trip through bytes, 2x faster and 30x less memory than builder.as_reader() + + def migration(inputs: list[str], product: str|None=None): def decorator(func): @functools.wraps(func) @@ -124,7 +128,7 @@ def migrate_longitudinalPlan(msgs): new_msg = msg.as_builder() a_target, should_stop = get_accel_from_plan(msg.longitudinalPlan.speeds, msg.longitudinalPlan.accels, CONTROL_N_T_IDX) new_msg.longitudinalPlan.aTarget, new_msg.longitudinalPlan.shouldStop = float(a_target), bool(should_stop) - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -133,7 +137,7 @@ def migrate_driverAssistance(msgs): add_ops = [] for _, msg in msgs: new_msg = messaging.new_message('driverAssistance', valid=True, logMonoTime=msg.logMonoTime) - add_ops.append(new_msg.as_reader()) + add_ops.append(as_reader(new_msg)) return [], add_ops, [] @@ -150,7 +154,7 @@ def migrate_drivingModelData(msgs): fill_lane_line_meta(dmd.drivingModelData.laneLineMeta, msg.modelV2.laneLines, msg.modelV2.laneLineProbs) if all(len(a) for a in [msg.modelV2.position.x, msg.modelV2.position.y, msg.modelV2.position.z]): fill_xyz_poly(dmd.drivingModelData.path, ModelConstants.POLY_PATH_DEGREE, msg.modelV2.position.x, msg.modelV2.position.y, msg.modelV2.position.z) - add_ops.append( dmd.as_reader()) + add_ops.append(as_reader(dmd)) return [], add_ops, [] @@ -175,7 +179,7 @@ def migrate_liveTracks(msgs): pts.append(pt) new_msg.liveTracks.points = pts - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -195,7 +199,7 @@ def migrate_liveLocationKalman(msgs): lp_field.valid = llk_field.valid for flag in ["inputsOK", "posenetOK", "sensorsOK"]: setattr(m.livePose, flag, getattr(msg.liveLocationKalmanDEPRECATED, flag)) - ops.append((index, m.as_reader())) + ops.append((index, as_reader(m))) return ops, [], [] @@ -210,7 +214,7 @@ def migrate_livePose(msgs): if msg.which() == "livePose": new_msg = msg.as_builder() new_msg.livePose.timestamp = msg.logMonoTime - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -226,7 +230,7 @@ def migrate_controlsState(msgs): "alertStatus", "alertSize", "alertType", "experimentalMode", "personality"): setattr(ss, field, getattr(msg.controlsState.deprecated, field)) - add_ops.append(m.as_reader()) + add_ops.append(as_reader(m)) return [], add_ops, [] @@ -242,7 +246,7 @@ def migrate_carState(msgs): msg = msg.as_builder() msg.carState.vCruise = last_cs.controlsState.deprecated.vCruise msg.carState.vCruiseCluster = last_cs.controlsState.deprecated.vCruiseCluster - ops.append((index, msg.as_reader())) + ops.append((index, as_reader(msg))) return ops, [], [] @@ -252,7 +256,7 @@ def migrate_managerState(msgs): for index, msg in msgs: new_msg = msg.as_builder() new_msg.managerState.processes = [{'name': name, 'running': True} for name in managed_processes] - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -265,7 +269,7 @@ def migrate_gpsLocation(msgs): # hasFix is a newer field if not g.hasFix and g.flags == 1: g.hasFix = True - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -281,7 +285,7 @@ def migrate_deviceState(msgs): if msg.which() == 'deviceState': n = msg.as_builder() n.deviceState.deviceType = init_data.deviceType - ops.append((i, n.as_reader())) + ops.append((i, as_reader(n))) return ops, [], [] @@ -293,7 +297,7 @@ def migrate_carOutput(msgs): co.valid = msg.valid co.logMonoTime = msg.logMonoTime co.carOutput.actuatorsOutput = msg.carControl.actuatorsOutputDEPRECATED - add_ops.append(co.as_reader()) + add_ops.append(as_reader(co)) return [], add_ops, [] @@ -331,13 +335,13 @@ def migrate_pandaStates(msgs): new_msg.logMonoTime = msg.logMonoTime new_msg.pandaStates[0] = msg.pandaStateDEPRECATED new_msg.pandaStates[0].safetyParam = safety_param - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) elif msg.which() == 'pandaStates': new_msg = msg.as_builder() new_msg.pandaStates[-1].safetyParam = safety_param # Clear DISABLE_DISENGAGE_ON_GAS bit to fix controls mismatch new_msg.pandaStates[-1].alternativeExperience &= ~1 - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -352,7 +356,7 @@ def migrate_peripheralState(msgs): new_msg = messaging.new_message("peripheralState") new_msg.valid = msg.valid new_msg.logMonoTime = msg.logMonoTime - add_ops.append(new_msg.as_reader()) + add_ops.append(as_reader(new_msg)) return [], add_ops, [] @@ -407,7 +411,7 @@ def migrate_cameraStates(msgs): new_msg.valid = msg.valid del_ops.append(index) - add_ops.append(new_msg.as_reader()) + add_ops.append(as_reader(new_msg)) return [], add_ops, del_ops @@ -419,7 +423,7 @@ def migrate_carParams(msgs): CP.carParams.carFingerprint = MIGRATION.get(CP.carParams.carFingerprint, CP.carParams.carFingerprint) for car_fw in CP.carParams.carFw: car_fw.brand = CP.carParams.brand - ops.append((index, CP.as_reader())) + ops.append((index, as_reader(CP))) return ops, [], [] @@ -451,7 +455,7 @@ def migrate_sensorEvents(msgs): m_dat.timestamp = evt.timestamp setattr(m_dat, evt.which(), getattr(evt, evt.which())) - add_ops.append(m.as_reader()) + add_ops.append(as_reader(m)) del_ops.append(index) return [], add_ops, del_ops @@ -474,7 +478,7 @@ def migrate_onroadEvents(msgs): new_msg.valid = msg.valid new_msg.logMonoTime = msg.logMonoTime new_msg.onroadEvents = onroadEvents - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], [] @@ -514,6 +518,6 @@ def migrate_driverMonitoringState(msgs): dm.visionPolicyState.pose.calibrated = old.posePitchValidCount >= 600 and old.poseYawValidCount >= 600 dm.wheeltouchPolicyState.awarenessPercent = int(max(0, min(100, (old.awarenessPassive if old.isActiveMode else old.awarenessStatus) * 100))) dm.wheeltouchPolicyState.awarenessStep = 0. if old.isActiveMode else old.stepChange - ops.append((index, new_msg.as_reader())) + ops.append((index, as_reader(new_msg))) return ops, [], []