Optimize set_value with mask caching and combined operations

[KRRT7]

import timeit
from opendbc.can.dbc import Signal


def set_value_original(msg: bytearray, sig: Signal, ival: int) -> None:
  """Original implementation."""
  i = sig.lsb // 8
  bits = sig.size
  if sig.size < 64:
    ival &= (1 << sig.size) - 1
  while 0 <= i < len(msg) and bits > 0:
    shift = sig.lsb % 8 if (sig.lsb // 8) == i else 0
    size = min(bits, 8 - shift)
    mask = ((1 << size) - 1) << shift
    msg[i] &= ~mask
    msg[i] |= (ival & ((1 << size) - 1)) << shift
    bits -= size
    ival >>= size
    i = i + 1 if sig.is_little_endian else i - 1


_MASKS = [(1 << k) - 1 for k in range(9)]


def set_value_optimized(msg: bytearray, sig: Signal, ival: int) -> None:
  """Optimized implementation with mask caching and combined operations."""
  i = sig.lsb // 8
  bits = sig.size
  if sig.size < 64:
    ival &= (1 << sig.size) - 1

  while 0 <= i < len(msg) and bits > 0:
    shift = sig.lsb % 8 if (sig.lsb // 8) == i else 0
    size = min(bits, 8 - shift)
    size_mask = _MASKS[size] if size <= 8 else (1 << size) - 1
    mask = size_mask << shift
    msgval = msg[i]
    msg[i] = (msgval & ~mask) | ((ival & size_mask) << shift)
    bits -= size
    ival >>= size
    i = i + 1 if sig.is_little_endian else i - 1


def benchmark():
  """Run benchmarks for various signal sizes."""
  # Test cases: (lsb, size, is_little_endian, description)
  test_cases = [
    (0, 8, True, "8-bit signal at byte boundary"),
    (4, 12, True, "12-bit signal crossing byte boundary"),
    (0, 16, True, "16-bit signal"),
    (3, 32, True, "32-bit signal with offset"),
    (0, 64, True, "64-bit signal"),
    (0, 32, False, "32-bit big-endian signal"),
  ]

  msg_size = 64
  iterations = 100_000

  print("=" * 80)
  print("Benchmarking set_value: Original vs Optimized")
  print("=" * 80)
  print(f"Iterations per test: {iterations:,}")
  print()

  for lsb, size, is_little, desc in test_cases:
    sig = Signal(name="test", start_bit=lsb, msb=lsb + size - 1, lsb=lsb, size=size, is_signed=False, factor=1.0, offset=0.0, is_little_endian=is_little)
    test_value = (1 << min(size, 63)) - 1 if size < 64 else 0xFFFFFFFFFFFFFFFF

    # Test original
    def test_original(sig=sig, test_value=test_value):
      msg = bytearray(msg_size)
      set_value_original(msg, sig, test_value)

    # Test optimized
    def test_optimized(sig=sig, test_value=test_value):
      msg = bytearray(msg_size)
      set_value_optimized(msg, sig, test_value)

    # Verify both produce same results
    msg_orig = bytearray(msg_size)
    msg_opt = bytearray(msg_size)
    set_value_original(msg_orig, sig, test_value)
    set_value_optimized(msg_opt, sig, test_value)
    assert msg_orig == msg_opt, f"Results differ for {desc}"

    time_original = timeit.timeit(test_original, number=iterations)
    time_optimized = timeit.timeit(test_optimized, number=iterations)

    speedup = (time_original - time_optimized) / time_original * 100
    ratio = time_original / time_optimized

    print(f"{desc}")
    print(f"  Speedup:   {speedup:+.1f}% ({ratio:.2f}x)")
    print()

  print("=" * 80)


if __name__ == "__main__":
  benchmark()

locally I get this:

uv run benchmark_set_value.py
================================================================================
Benchmarking set_value: Original vs Optimized
================================================================================
Iterations per test: 100,000

8-bit signal at byte boundary
  Speedup:   +6.1% (1.06x)

12-bit signal crossing byte boundary
  Speedup:   +9.4% (1.10x)

16-bit signal
  Speedup:   +9.3% (1.10x)

32-bit signal with offset
  Speedup:   +10.1% (1.11x)

64-bit signal
  Speedup:   +12.2% (1.14x)

32-bit big-endian signal
  Speedup:   +6.1% (1.07x)

================================================================================

but with real world data it should be much better

btw this was originally optimized by codeflash: http://codeflash.ai but I wasn't able to open that PR directly due to a bug on our end

[github-actions]

Thanks for contributing to opendbc! In order for us to review your PR as quickly as possible, check the following:

• Convert your PR to a draft unless it's ready to review
• Read the contributing docs
• Before marking as "ready for review", ensure:
  • the goal is clearly stated in the description
  • all the tests are passing
  • include a route or your device' dongle ID if relevant