From cb85ac1f0e78425752cf744b2ca2cd268720a337 Mon Sep 17 00:00:00 2001 From: YassineYousfi Date: Tue, 1 Sep 2026 13:59:55 -0700 Subject: [PATCH] amd warp (#38684) * modeld: fuse warp and policy TinyJit * bump tg * fix? * this simple trick... * debug 1 * bump tg * pack all * wips * fix * BIG_INTO_SMALL remove * slower --- openpilot/selfdrive/modeld/SConscript | 24 ++- openpilot/selfdrive/modeld/compile_modeld.py | 147 ++++++++++-------- openpilot/selfdrive/modeld/modeld.py | 43 ++--- .../test/process_replay/model_replay.py | 10 +- 4 files changed, 108 insertions(+), 116 deletions(-) diff --git a/openpilot/selfdrive/modeld/SConscript b/openpilot/selfdrive/modeld/SConscript index f65aebf2f9..5a99aa890c 100644 --- a/openpilot/selfdrive/modeld/SConscript +++ b/openpilot/selfdrive/modeld/SConscript @@ -10,11 +10,6 @@ from openpilot.selfdrive.modeld.constants import ModelConstants from openpilot.selfdrive.modeld.helpers import TG_INPUT_DEVICES_PATH, chestnut_present, modeld_pkl_path -CAMERA_CONFIGS = [ - (_ar_ox_fisheye.width, _ar_ox_fisheye.height), # tici: 1928x1208 - (_os_fisheye.width, _os_fisheye.height), # mici: 1344x760 -] - Import('env', 'arch') chunker_file = File("#openpilot/common/file_chunker.py") lenv = env.Clone() @@ -29,17 +24,17 @@ def estimate_pickle_max_size(onnx_size): return 2.0 * onnx_size + 10 * 1024 * 1024 if arch == 'comma_arm64': + from openpilot.common.hardware import HARDWARE + camera = _os_fisheye if HARDWARE.get_device_type() == "mici" else _ar_ox_fisheye + camera_configs = [(camera.width, camera.height)] tg_backend = 'QCOM' tg_flags = f'DEV={tg_backend} IMAGE=1 FLOAT16=1 NOLOCALS=1 JIT_BATCH_SIZE=0 OPENPILOT_HACKS=1' else: + camera_configs = [(c.width, c.height) for c in (_ar_ox_fisheye, _os_fisheye)] tg_backend = 'CPU' tg_flags = f'DEV=CPU' if arch == 'Darwin' else 'DEV=CPU:LLVM' tg_devices = { # which device to put jit inputs to at runtime - 'openpilot.selfdrive.modeld.modeld': { - 'default': {'WARP_DEV': tg_backend, 'QUEUE_DEV': tg_backend}, - 'chestnut': {'WARP_DEV': tg_backend, 'QUEUE_DEV': 'AMD'} - }, 'openpilot.selfdrive.modeld.dmonitoringmodeld': { 'default': {'DEV': tg_backend} }, @@ -47,7 +42,7 @@ tg_devices = { # which device to put jit inputs to at runtime CHESTNUT = chestnut_present() if CHESTNUT: - chestnut_tg_flags = f'DEBUG=1 DEV=USB+AMD:LLVM WARP_DEV={tg_backend} FLOAT16=1 JIT_BATCH_SIZE=0 GMMU=0 TC_OPT=2' + chestnut_tg_flags = 'DEBUG=1 DEV=USB+AMD:LLVM FRAME_DEV=CPU FLOAT16=1 JIT_BATCH_SIZE=0 GMMU=0 TC_OPT=2 TC_OCCUPANCY_OPT=1' # the USB+AMD GPU takes an exclusive flock; serialize all targets that touch it chestnut_lock = File("models/.chestnut.lock").abspath @@ -77,10 +72,9 @@ frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ for chestnut in [False, True] if CHESTNUT else [False]: target_pkl_path = File(modeld_pkl_path(chestnut)).abspath - # BIG_INTO_SMALL=1 builds the default target from the big model, e.g. to test it without a chestnut - file_prefix, cmd_flags = ('big_', chestnut_tg_flags) if chestnut else ('big_' if os.getenv('BIG_INTO_SMALL') else '', tg_flags) + file_prefix, cmd_flags = ('big_', chestnut_tg_flags) if chestnut else ('', tg_flags) driving_onnx_deps = get_existing_chunks(File(f"models/{file_prefix}driving_supercombo.onnx").abspath) - camera_res_args = ' '.join(f'{cw}x{ch}' for cw, ch in CAMERA_CONFIGS) + camera_res_args = ' '.join(f'{cw}x{ch}' for cw, ch in camera_configs) # CPU 7 is isolated with isolcpus on AGNOS, so explicitly pin the compiler to it. taskset = 'taskset -c 7 ' if arch == 'comma_arm64' else '' cmd = (f'{cmd_flags} {mac_brew_string} {taskset}python3 {modeld_dir}/compile_modeld.py ' @@ -108,7 +102,7 @@ for chestnut in [False, True] if CHESTNUT else [False]: actions = Action(do_compile, " [CHESTNUT] $TARGET") if chestnut else [cmd, Action(do_chunk, " [CHUNK] $TARGET")] node = lenv.Command( chunk_targets, - tinygrad_files + compile_modeld_script + driving_onnx_deps + [Value(chunk_targets), chunker_file], + tinygrad_files + compile_modeld_script + driving_onnx_deps + [Value(camera_res_args), Value(chunk_targets), chunker_file], actions, ) if chestnut: @@ -122,7 +116,7 @@ lenv.Command(fn + "_metadata.pkl", [fn + ".onnx"] + tinygrad_files + script_file dm_w, dm_h = DM_INPUT_SIZE compile_dm_warp_script = [File(f"{modeld_dir}/compile_dm_warp.py")] -for cam_w, cam_h in CAMERA_CONFIGS: +for cam_w, cam_h in camera_configs: dm_pkl_path = File(f"models/dm_warp_{cam_w}x{cam_h}_tinygrad.pkl").abspath cmd = (f'{tg_flags} {mac_brew_string} python3 {modeld_dir}/compile_dm_warp.py ' f'--camera-resolution {cam_w}x{cam_h} --warp-to {dm_w}x{dm_h} ' diff --git a/openpilot/selfdrive/modeld/compile_modeld.py b/openpilot/selfdrive/modeld/compile_modeld.py index f54d09e869..d851c3cc86 100755 --- a/openpilot/selfdrive/modeld/compile_modeld.py +++ b/openpilot/selfdrive/modeld/compile_modeld.py @@ -37,17 +37,12 @@ from tinygrad.engine.jit import TinyJit NV12Frame = namedtuple("NV12Frame", ['width', 'height', 'stride', 'y_height', 'uv_height', 'size']) -WARP_INPUTS = ['tfm', 'big_tfm'] -POLICY_INPUTS = ['img_q', 'big_img_q', 'feat_q', 'desire_q', 'packed_npy_inputs'] - -UV_SCALE_MATRIX = np.array([[0.5, 0, 0], [0, 0.5, 0], [0, 0, 1]], dtype=np.float32) -UV_SCALE_MATRIX_INV = np.linalg.inv(UV_SCALE_MATRIX) - -WARP_DEV = os.getenv('WARP_DEV') +MODELD_INPUTS = ['img_q', 'big_img_q', 'feat_q', 'desire_q', 'packed_npy_inputs'] -def make_random_images(keys, shape, device=None): - return {k: Tensor.randint(shape, low=0, high=256, dtype='uint8', device=device).realize() for k in keys} +def nv12_copy_size(stride: int, y_height: int, uv_height: int) -> int: + # Retain the padded Y and UV plane storage, but skip the trailing kernel/guard allocation. + return stride * (y_height + uv_height) def warp_perspective_tinygrad(src_flat, M_inv, dst_shape, src_shape, stride_pad, border_fill_val=None): @@ -99,7 +94,7 @@ def make_frame_prepare(nv12: NV12Frame, model_w, model_h): def frame_prepare_tinygrad(input_frame, M_inv): # UV_SCALE @ M_inv @ UV_SCALE_INV simplifies to elementwise scaling - M_inv_uv = M_inv * Tensor([[1.0, 1.0, 0.5], [1.0, 1.0, 0.5], [2.0, 2.0, 1.0]], device=WARP_DEV) + M_inv_uv = M_inv * Tensor([[1.0, 1.0, 0.5], [1.0, 1.0, 0.5], [2.0, 2.0, 1.0]], device=Device.DEFAULT) # deinterleave NV12 UV plane (UVUV... -> separate U, V) uv = input_frame[uv_offset:uv_offset + uv_height * stride].reshape(uv_height, stride) with Context(SPLIT_REDUCEOP=0): @@ -118,23 +113,6 @@ def make_frame_prepare(nv12: NV12Frame, model_w, model_h): return frame_prepare_tinygrad -def make_warp_input_queues(vision_input_shapes, frame_skip, device): - img = vision_input_shapes['img'] # (1, 12, 128, 256) - n_frames = img[1] // 6 - img_buf_shape = (frame_skip * (n_frames - 1) + 1, 6, img[2], img[3]) - - npy = { - 'tfm': np.zeros((3, 3), dtype=np.float32), - 'big_tfm': np.zeros((3, 3), dtype=np.float32), - } - input_queues = { - 'img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), - 'big_img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), - **{k: Tensor(v, device='NPY').realize() for k, v in npy.items()}, - } - return input_queues, npy - - def get_policy_npy_shapes(input_shapes): dp = input_shapes['desire_pulse'] # (1, 25, 8) tc = input_shapes['traffic_convention'] # (1, 2) @@ -146,23 +124,32 @@ def get_policy_npy_shapes(input_shapes): return shapes, [math.prod(s) for s in shapes.values()] -def make_input_queues(input_shapes, frame_skip, device): - input_queues, npy = make_warp_input_queues(input_shapes, frame_skip, device) - +def make_input_queues(input_shapes, frame_skip, device, frame_copy_size): + img = input_shapes['img'] # (1, 12, 128, 256) fb = input_shapes['features_buffer'] # (1, T-1, ...), past features only; the model appends the current frame's feature feat_dim = math.prod(fb[2:]) dp = input_shapes['desire_pulse'] # (1, 25, 8) + n_frames = img[1] // 6 + img_buf_shape = (frame_skip * (n_frames - 1) + 1, 6, img[2], img[3]) - shapes, sizes = get_policy_npy_shapes(input_shapes) - packed_npy_inputs = np.zeros(sum(sizes), dtype=np.float32) + policy_shapes, _ = get_policy_npy_shapes(input_shapes) + shapes = {'tfm': (3, 3), 'big_tfm': (3, 3)} | policy_shapes + sizes = [math.prod(s) for s in shapes.values()] + packed_npy_size = sum(sizes) * np.dtype(np.float32).itemsize + packed_input = np.zeros(packed_npy_size + 2 * frame_copy_size, dtype=np.uint8) + packed_npy_inputs = packed_input[:packed_npy_size].view(np.float32) + frames = packed_input[packed_npy_size:] + frame_views = {'img': frames[:frame_copy_size], 'big_img': frames[frame_copy_size:]} # views into the packed inputs, to be refilled at runtime - npy.update({k: v.reshape(s) for (k, s), v in zip(shapes.items(), np.split(packed_npy_inputs, np.cumsum(sizes[:-1])), strict=True)}) - input_queues.update({ + npy = {k: v.reshape(s) for (k, s), v in zip(shapes.items(), np.split(packed_npy_inputs, np.cumsum(sizes[:-1])), strict=True)} + input_queues = { + 'img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), + 'big_img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), 'feat_q': Tensor(np.zeros((frame_skip * fb[1], fb[0], feat_dim), dtype=np.float32), device=device).contiguous().realize(), 'desire_q': Tensor(np.zeros((frame_skip * dp[1], dp[0], dp[2]), dtype=np.float32), device=device).contiguous().realize(), - 'packed_npy_inputs': Tensor(packed_npy_inputs, device='NPY').realize(), - }) - return input_queues, npy + 'packed_npy_inputs': Tensor(packed_input, device='NPY').realize(), + } + return input_queues, npy, frame_views def shift_and_sample(buf, new_val, sample_fn): @@ -178,13 +165,15 @@ def sample_desire(buf, frame_skip): return buf.reshape(-1, frame_skip, *buf.shape[1:]).max(1).flatten(0, 1).unsqueeze(0) -def make_warp(nv12, model_w, model_h, frame_skip): +def make_warp(nv12, model_w, model_h): frame_prepare = make_frame_prepare(nv12, model_w, model_h) def warp(tfm, big_tfm, frame, big_frame): - tfm = tfm.to(WARP_DEV) - big_tfm = big_tfm.to(WARP_DEV) - Tensor.realize(tfm, big_tfm) + tfm = tfm.to(Device.DEFAULT) + big_tfm = big_tfm.to(Device.DEFAULT) + frame = frame.to(Device.DEFAULT) + big_frame = big_frame.to(Device.DEFAULT) + Tensor.realize(tfm, big_tfm, frame, big_frame) warped_frame = frame_prepare(frame, tfm).unsqueeze(0) warped_big_frame = frame_prepare(big_frame, big_tfm).unsqueeze(0) @@ -197,10 +186,10 @@ def make_run_policy(model_runner, model_metadata, frame_skip): sample_desire_fn = partial(sample_desire, frame_skip=frame_skip) sample_skip_fn = partial(sample_skip, frame_skip=frame_skip) npy_shapes, npy_sizes = get_policy_npy_shapes(model_metadata['input_shapes']) + model_input_dtypes = {name: spec.dtype for name, spec in model_runner.graph_inputs.items()} def run_policy(warped, img_q, big_img_q, feat_q, desire_q, packed_npy_inputs): packed_npy_inputs = packed_npy_inputs.to(Device.DEFAULT) - warped = warped.to(Device.DEFAULT) Tensor.realize(packed_npy_inputs, warped) img = shift_and_sample(img_q, warped[0:1], sample_skip_fn) @@ -218,28 +207,45 @@ def make_run_policy(model_runner, model_metadata, frame_skip): 'traffic_convention': traffic_convention, 'action_t': action_t, } + inputs = {name: value.cast(model_input_dtypes[name]) for name, value in inputs.items()} out = next(iter(model_runner(inputs).values())).cast('float32') return out, return run_policy -def compile_jit(jit, make_random_inputs, input_keys, make_queues): - SEED = 42 - def random_inputs_run(fn, seed, test_val=None, test_buffers=None, expect_match=True): - input_queues, npy = make_queues(Device.DEFAULT) - rng = np.random.default_rng(seed) - Tensor.manual_seed(seed) +def make_run_model(warp, run_policy, model_metadata, frame_copy_size): + _, policy_sizes = get_policy_npy_shapes(model_metadata['input_shapes']) + packed_npy_size = (18 + sum(policy_sizes)) * np.dtype(np.float32).itemsize - testing = test_val is not None or test_buffers is not None - n_runs = 1 if testing else 3 + def run_model(img_q, big_img_q, feat_q, desire_q, packed_npy_inputs): + packed_input = packed_npy_inputs.to(Device.DEFAULT) + Tensor.realize(packed_input) + packed_npy_inputs = packed_input[:packed_npy_size].bitcast('float32') + frame = packed_input[packed_npy_size:packed_npy_size + frame_copy_size] + big_frame = packed_input[packed_npy_size + frame_copy_size:] + tfm, big_tfm, policy_inputs = packed_npy_inputs.split([9, 9, sum(policy_sizes)]) + warped = warp(tfm.reshape(3, 3), big_tfm.reshape(3, 3), frame, big_frame) + return run_policy(warped, img_q, big_img_q, feat_q, desire_q, policy_inputs) + return run_model + + +def compile_jit(jit, input_keys, make_queues, benchmark_runs): + if benchmark_runs < 1: + raise ValueError("benchmark_runs must be at least 1") + + SEED = 42 + def random_inputs_run(fn, seed, n_runs, test_val=None, test_buffers=None, expect_match=True): + input_queues, npy, frame_views = make_queues(Device.DEFAULT) + rng = np.random.default_rng(seed) for i in range(n_runs): for v in npy.values(): v[:] = rng.standard_normal(v.shape).astype(v.dtype) + for v in frame_views.values(): + v[:] = rng.integers(0, 256, size=v.shape, dtype=np.uint8) Device.default.synchronize() - random_inputs = make_random_inputs() st = time.perf_counter() - outs = fn(**{k: input_queues[k] for k in input_keys}, **random_inputs) + outs = fn(**{k: input_queues[k] for k in input_keys}) mt = time.perf_counter() Device.default.synchronize() et = time.perf_counter() @@ -258,14 +264,15 @@ def compile_jit(jit, make_random_inputs, input_keys, make_queues): return val, buffers print('capture + replay') - test_val, test_buffers = random_inputs_run(jit, SEED) - print('pickle round trip') + test_val, test_buffers = random_inputs_run(jit, SEED, 3) + print(f'pickle round trip ({benchmark_runs} runs per seed)') with tempfile.TemporaryFile(dir=".") as f: dump_oob(jit, f) f.seek(0) - jit = load_oob(f) - random_inputs_run(jit, SEED, test_val, test_buffers, expect_match=True) - random_inputs_run(jit, SEED+1, test_val, test_buffers, expect_match=False) + loaded_jit = load_oob(f) + random_inputs_run(loaded_jit, SEED, benchmark_runs, test_val, test_buffers, expect_match=True) + random_inputs_run(loaded_jit, SEED+1, benchmark_runs, test_val, test_buffers, expect_match=False) + # Keep the original so per-resolution JITs share model weight buffers in the final pickle. return jit @@ -294,27 +301,31 @@ if __name__ == "__main__": p.add_argument('--onnx', required=True) p.add_argument('--output', required=True) p.add_argument('--frame-skip', type=int, required=True) + p.add_argument('--benchmark-runs', type=int, default=1, + help='timed loaded-JIT runs for each correctness seed') args = p.parse_args() model_path = read_file_chunked_to_disk(args.onnx) model_w, model_h = args.model_size model_runner = OnnxRunner(model_path) - out = {'metadata': make_metadata_dict(model_path)} + out = { + 'metadata': make_metadata_dict(model_path), + 'input_devices': {'model': Device.DEFAULT}, + 'run_model': {}, + } - run_policy_jit = TinyJit(make_run_policy(model_runner, out['metadata'], args.frame_skip), prune=True) - - make_policy_queues = partial(make_input_queues, out['metadata']['input_shapes'], args.frame_skip) - make_random_model_inputs = partial(make_random_images, keys=['warped'], shape=(2, 6, *out['metadata']['input_shapes']['img'][2:]), device=WARP_DEV) - out['run_policy'] = compile_jit(run_policy_jit, make_random_model_inputs, POLICY_INPUTS, - make_policy_queues) + run_policy = make_run_policy(model_runner, out['metadata'], args.frame_skip) for cam_w, cam_h in args.camera_resolutions: nv12 = NV12Frame(cam_w, cam_h, *get_nv12_info(cam_w, cam_h)) - make_random_warp_inputs = partial(make_random_images, keys=['frame', 'big_frame'], shape=nv12.size, device=WARP_DEV) - warp = TinyJit(make_warp(nv12, model_w, model_h, args.frame_skip), prune=True) - make_warp_queues = partial(make_warp_input_queues, out['metadata']['input_shapes'], args.frame_skip) - out[(cam_w,cam_h)] = compile_jit(warp, make_random_warp_inputs, WARP_INPUTS, make_warp_queues) + frame_copy_size = nv12_copy_size(nv12.stride, nv12.y_height, nv12.uv_height) + make_model_queues = partial(make_input_queues, out['metadata']['input_shapes'], args.frame_skip, + frame_copy_size=frame_copy_size) + warp = make_warp(nv12, model_w, model_h) + run_model_jit = TinyJit(make_run_model(warp, run_policy, out['metadata'], frame_copy_size), prune=True) + out['run_model'][(cam_w,cam_h)] = compile_jit(run_model_jit, MODELD_INPUTS, make_model_queues, + args.benchmark_runs) with open(args.output, "wb") as f: dump_oob(out, f) diff --git a/openpilot/selfdrive/modeld/modeld.py b/openpilot/selfdrive/modeld/modeld.py index de66decf5c..040c6b980f 100755 --- a/openpilot/selfdrive/modeld/modeld.py +++ b/openpilot/selfdrive/modeld/modeld.py @@ -4,7 +4,6 @@ import ctypes from functools import cached_property import os os.environ['GMMU'] = '0' # for chestnut fast loading, noop for qcom -from tinygrad.tensor import Tensor from tinygrad.device import Device import usb1 import struct @@ -29,14 +28,13 @@ from openpilot.common.transformations.model import get_warp_matrix from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper from openpilot.selfdrive.controls.lib.drive_helpers import get_accel_from_plan, should_stop, smooth_value, get_curvature_from_plan from openpilot.selfdrive.modeld.parse_model_outputs import Parser -from openpilot.selfdrive.modeld.compile_modeld import make_input_queues, WARP_INPUTS, POLICY_INPUTS +from openpilot.selfdrive.modeld.compile_modeld import make_input_queues, nv12_copy_size, MODELD_INPUTS from openpilot.selfdrive.modeld.fill_model_msg import fill_model_msg, fill_driving_model_data, fill_pose_msg, PublishState from openpilot.common.file_chunker import open_file_chunked from openpilot.common.hardware.usb import CHESTNUT_USB_IDS from openpilot.selfdrive.modeld.constants import ModelConstants, Plan -from openpilot.selfdrive.modeld.helpers import chestnut_present, chestnut_compiled, chestnut_ready, modeld_pkl_path, get_tg_input_devices, load_oob +from openpilot.selfdrive.modeld.helpers import chestnut_present, chestnut_compiled, chestnut_ready, modeld_pkl_path, load_oob -PROCESS_NAME = "openpilot.selfdrive.modeld.modeld" SEND_RAW_PRED = os.getenv('SEND_RAW_PRED') LAT_SMOOTH_SECONDS = 0.0 @@ -177,9 +175,9 @@ class ModelState: prev_desire: np.ndarray # for tracking the rising edge of the pulse def __init__(self, cam_w: int, cam_h: int, chestnut: bool): - input_devices = get_tg_input_devices(PROCESS_NAME, chestnut) - self.WARP_DEV, self.QUEUE_DEV = input_devices['WARP_DEV'], input_devices['QUEUE_DEV'] jits = load_oob(open_file_chunked(modeld_pkl_path(chestnut))) + input_devices = jits['input_devices'] + self.model_device = input_devices['model'] metadata = jits['metadata'] self.input_shapes = metadata['input_shapes'] self.vision_input_names = [k for k in self.input_shapes if 'img' in k] @@ -189,13 +187,11 @@ class ModelState: self.chestnut = chestnut self.frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ - self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV) - self.full_frames: dict[str, Tensor] = {} - self._blob_cache: dict[tuple[str, int], Tensor] = {} + self.frame_copy_size = nv12_copy_size(*get_nv12_info(cam_w, cam_h)[:3]) + self.input_queues, self.npy, self.frame_views = make_input_queues( + self.input_shapes, self.frame_skip, device=self.model_device, frame_copy_size=self.frame_copy_size) self.parser = Parser() - self.frame_buf_params = {k: get_nv12_info(cam_w, cam_h) for k in ('img', 'big_img')} - self.run_policy = jits['run_policy'] - self.warp = jits[(cam_w,cam_h)] + self.run_model = jits['run_model'][(cam_w,cam_h)] def slice_outputs(self, model_outputs: np.ndarray, output_slices: dict[str, slice]) -> dict[str, np.ndarray]: parsed_model_outputs = {k: model_outputs[np.newaxis, v] for k,v in output_slices.items()} @@ -203,14 +199,8 @@ class ModelState: def run(self, bufs: dict[str, VisionBuf], transforms: dict[str, np.ndarray], inputs: dict[str, np.ndarray], after_enqueue: Callable[[], None] | None = None) -> dict[str, np.ndarray]: - for key in bufs.keys(): - ptr = np.frombuffer(bufs[key].data, dtype=np.uint8).ctypes.data - yuv_size = self.frame_buf_params[key][3] - # There is a ringbuffer of imgs, just cache tensors pointing to all of them - cache_key = (key, ptr) - if cache_key not in self._blob_cache: - self._blob_cache[cache_key] = Tensor.from_blob(ptr, (yuv_size,), dtype='uint8', device=self.WARP_DEV) - self.full_frames[key] = self._blob_cache[cache_key] + for key, buf in bufs.items(): + np.copyto(self.frame_views[key], np.frombuffer(buf.data, dtype=np.uint8, count=self.frame_copy_size)) # Model decides when action is completed, so desire input is just a pulse triggered on rising edge inputs['desire_pulse'][0] = 0 @@ -221,11 +211,7 @@ class ModelState: self.npy['tfm'][:,:] = transforms['img'][:,:] self.npy['big_tfm'][:,:] = transforms['big_img'][:,:] - warped = self.warp(**{k: self.input_queues[k] for k in WARP_INPUTS}, frame=self.full_frames['img'], big_frame=self.full_frames['big_img']) - - outs, = self.run_policy( - **{k: self.input_queues[k] for k in POLICY_INPUTS if k in self.input_queues}, warped=warped - ) + outs, = self.run_model(**{k: self.input_queues[k] for k in MODELD_INPUTS}) if after_enqueue is not None: after_enqueue() model_output = outs.numpy()[0] @@ -239,14 +225,13 @@ class ModelState: return outputs_dict def warmup(self) -> None: - dummy_frames = {k: np.zeros(self.frame_buf_params[k][3], dtype=np.uint8) for k in self.vision_input_names} + dummy_frames = {k: np.zeros(self.frame_copy_size, dtype=np.uint8) for k in self.vision_input_names} eye = np.eye(3, dtype=np.float32) dims = {'desire_pulse': ModelConstants.DESIRE_LEN, 'traffic_convention': 2, 'action_t': 2} self.run(dummy_frames, dict.fromkeys(self.vision_input_names, eye), {k: np.zeros(v, dtype=np.float32) for k, v in dims.items()}) - self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV) + self.input_queues, self.npy, self.frame_views = make_input_queues( + self.input_shapes, self.frame_skip, device=self.model_device, frame_copy_size=self.frame_copy_size) self.prev_desire[:] = 0 - self.full_frames.clear() - self._blob_cache.clear() def main(demo=False): diff --git a/openpilot/selfdrive/test/process_replay/model_replay.py b/openpilot/selfdrive/test/process_replay/model_replay.py index ba610c6a2b..927c9b38f1 100755 --- a/openpilot/selfdrive/test/process_replay/model_replay.py +++ b/openpilot/selfdrive/test/process_replay/model_replay.py @@ -33,9 +33,9 @@ MODEL_REPLAY_BUCKET="model_replay_master" GITHUB = GithubUtils(API_TOKEN, DATA_TOKEN) EXEC_TIMINGS = [ - # model, instant max, average max - ("modelV2", 0.05, 0.028), - ("driverStateV2", 0.05, 0.018), + # model, instant max, average max, chestnut average max + ("modelV2", 0.05, 0.03, 0.05), + ("driverStateV2", 0.05, 0.018, 0.018), ] def get_log_fn(test_route, ref="master"): @@ -169,11 +169,13 @@ def model_replay(lr, frs): dmonitoringmodeld_msgs = replay_process(dmonitoringmodeld, dmodeld_logs, frs) msgs = modeld_msgs + dmonitoringmodeld_msgs + chestnut = any(m.modelV2.big for m in modeld_msgs if m.which() == "modelV2") header = ['model', 'max instant', 'max instant allowed', 'average', 'max average allowed', 'test result'] rows = [] timings_ok = True - for (s, instant_max, avg_max) in EXEC_TIMINGS: + for (s, instant_max, avg_max, chestnut_avg_max) in EXEC_TIMINGS: + avg_max = chestnut_avg_max if chestnut else avg_max ts = [getattr(m, s).modelExecutionTime for m in msgs if m.which() == s] # TODO some init can happen in first iteration ts = ts[1:]