⚡️ Speed up method BCDataStream.read_uint16 by 36%

[codeflash-ai]

📄 36% (0.36x) speedup for BCDataStream.read_uint16 in electrum/transaction.py

⏱️ Runtime : 1.22 milliseconds → 895 microseconds (best of 10 runs)

📝 Explanation and details

The optimized code achieves a 36% speedup by inlining the read_uint16 method and eliminating the struct.calcsize() call.

Key optimizations:

1. Function call elimination: The original code delegates to _read_num('<H'), requiring an additional function call with parameter passing. The optimized version inlines this logic directly in read_uint16, removing the function call overhead.

2. Hardcoded size calculation: Instead of calling struct.calcsize('<H') every time (which always returns 2 for a uint16), the optimized code directly increments the cursor by 2. This eliminates a repeated calculation that the Python interpreter must perform on every invocation.

Performance impact analysis:

• The line profiler shows read_uint16 total time decreased from 15.8ms to 5.1ms (68% improvement)
• The cursor increment operation (self.read_cursor += 2) is 11% faster than the original struct.calcsize approach (1.35ms vs 1.52ms)
• Exception handling remains identical, preserving all error conditions

Test case performance:
The optimization is particularly effective for:

• High-frequency reads: Large buffer tests show 35-40% improvements when reading 500-1000 uint16 values
• Sequential operations: Multiple reads in the same stream benefit from reduced per-call overhead
• All input types: Consistent speedups across bytes, bytearray, and memoryview inputs (22-49% faster)

This optimization is especially valuable in Bitcoin transaction parsing where read_uint16 is likely called frequently during deserialization of binary data streams.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 4145 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import struct

# imports
import pytest
from electrum.transaction import BCDataStream


# function to test
class SerializationError(Exception):
    pass

# unit tests

# -------------------- BASIC TEST CASES --------------------

def test_read_uint16_basic_little_endian():
    # 0x1234 in little-endian is b'\x34\x12'
    ds = BCDataStream()
    ds.input = b'\x34\x12'
    ds.read_cursor = 0
    codeflash_output = ds.read_uint16() # 2.73μs -> 1.85μs (47.3% faster)

def test_read_uint16_zero():
    # 0x0000
    ds = BCDataStream()
    ds.input = b'\x00\x00'
    ds.read_cursor = 0
    codeflash_output = ds.read_uint16() # 1.65μs -> 1.38μs (19.1% faster)

def test_read_uint16_max_value():
    # 0xFFFF
    ds = BCDataStream()
    ds.input = b'\xFF\xFF'
    ds.read_cursor = 0
    codeflash_output = ds.read_uint16() # 1.70μs -> 1.20μs (41.4% faster)

def test_read_uint16_multiple_reads():
    # Read two uint16 values sequentially
    ds = BCDataStream()
    ds.input = b'\x01\x00\xFF\xFF'
    ds.read_cursor = 0
    codeflash_output = ds.read_uint16() # 1.60μs -> 1.20μs (33.3% faster)
    codeflash_output = ds.read_uint16() # 725ns -> 559ns (29.7% faster)

def test_read_uint16_cursor_advance():
    # Ensure cursor advances by 2 bytes per read
    ds = BCDataStream()
    ds.input = b'\x01\x02\x03\x04'
    ds.read_cursor = 0
    ds.read_uint16() # 1.48μs -> 1.25μs (17.8% faster)
    ds.read_uint16() # 670ns -> 526ns (27.4% faster)

# -------------------- EDGE TEST CASES --------------------

def test_read_uint16_insufficient_bytes():
    # Only one byte provided, should raise SerializationError
    ds = BCDataStream()
    ds.input = b'\x01'
    ds.read_cursor = 0
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_empty_input():
    # No bytes provided
    ds = BCDataStream()
    ds.input = b''
    ds.read_cursor = 0
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_cursor_at_end():
    # Cursor is at the end of input, nothing left to read
    ds = BCDataStream()
    ds.input = b'\x12\x34'
    ds.read_cursor = 2
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_cursor_one_before_end():
    # Cursor is one byte before end, not enough bytes for uint16
    ds = BCDataStream()
    ds.input = b'\x12\x34'
    ds.read_cursor = 1
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_non_bytes_input():
    # input is not bytes or bytearray, e.g. None
    ds = BCDataStream()
    ds.input = None
    ds.read_cursor = 0
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_negative_cursor():
    # Negative cursor should raise SerializationError
    ds = BCDataStream()
    ds.input = b'\x12\x34'
    ds.read_cursor = -1
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_large_cursor():
    # Cursor is way beyond input length
    ds = BCDataStream()
    ds.input = b'\x12\x34'
    ds.read_cursor = 100
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_input_bytearray():
    # Accepts bytearray as input
    ds = BCDataStream()
    ds.input = bytearray([0x78, 0x56])
    ds.read_cursor = 0
    codeflash_output = ds.read_uint16() # 2.80μs -> 1.96μs (43.4% faster)

def test_read_uint16_input_memoryview():
    # Accepts memoryview as input (struct.unpack_from supports it)
    ds = BCDataStream()
    ds.input = memoryview(b'\xCD\xAB')
    ds.read_cursor = 0
    codeflash_output = ds.read_uint16() # 1.71μs -> 1.37μs (24.3% faster)

# -------------------- LARGE SCALE TEST CASES --------------------

def test_read_uint16_large_buffer():
    # Read 500 uint16 values from a buffer of 1000 bytes
    values = [i for i in range(500)]
    # Pack as little-endian uint16
    buf = b''.join(struct.pack('<H', v) for v in values)
    ds = BCDataStream()
    ds.input = buf
    ds.read_cursor = 0
    for expected in values:
        codeflash_output = ds.read_uint16() # 231μs -> 169μs (36.8% faster)

def test_read_uint16_large_buffer_with_offset():
    # Start reading from a non-zero cursor in a large buffer
    values = [0xAAAA, 0xBBBB, 0xCCCC, 0xDDDD, 0xEEEE]
    buf = b'\x00\x00' * 10 + b''.join(struct.pack('<H', v) for v in values)
    ds = BCDataStream()
    ds.input = buf
    ds.read_cursor = 20  # Skip the first 10 uint16s
    for expected in values:
        codeflash_output = ds.read_uint16() # 3.70μs -> 2.53μs (46.0% faster)

def test_read_uint16_large_buffer_incomplete_final():
    # Buffer is one byte short at the end, last read should fail
    values = [0x1111] * 499
    buf = b''.join(struct.pack('<H', v) for v in values) + b'\xFF'  # 999 bytes
    ds = BCDataStream()
    ds.input = buf
    ds.read_cursor = 0
    for _ in range(499):
        codeflash_output = ds.read_uint16()
    # Last read should fail, only 1 byte left
    with pytest.raises(SerializationError):
        ds.read_uint16()

def test_read_uint16_large_buffer_all_zeros():
    # Large buffer of zeros
    buf = b'\x00\x00' * 1000
    ds = BCDataStream()
    ds.input = buf
    ds.read_cursor = 0
    for _ in range(1000):
        codeflash_output = ds.read_uint16() # 459μs -> 335μs (36.9% faster)

# -------------------- MISCELLANEOUS TESTS --------------------

def test_read_uint16_does_not_mutate_input():
    # Ensure input buffer is not mutated by read_uint16
    buf = bytearray(b'\x01\x02')
    ds = BCDataStream()
    ds.input = buf
    ds.read_cursor = 0
    before = bytes(ds.input)
    ds.read_uint16() # 2.26μs -> 1.65μs (36.7% faster)
    after = bytes(ds.input)

def test_read_uint16_multiple_streams_independent():
    # Two BCDataStream instances should not interfere with each other
    ds1 = BCDataStream()
    ds2 = BCDataStream()
    ds1.input = b'\x01\x00\x02\x00'
    ds2.input = b'\xFF\xFF\x00\x00'
    ds1.read_cursor = 0
    ds2.read_cursor = 0
    codeflash_output = ds1.read_uint16() # 1.56μs -> 1.26μs (24.4% faster)
    codeflash_output = ds2.read_uint16() # 647ns -> 482ns (34.2% faster)
    codeflash_output = ds1.read_uint16() # 578ns -> 471ns (22.7% faster)
    codeflash_output = ds2.read_uint16() # 487ns -> 322ns (51.2% faster)

def test_read_uint16_struct_format_enforcement():
    # Should not accept other sizes (e.g. float or int32)
    ds = BCDataStream()
    ds.input = b'\x01\x00\x00\x00'
    ds.read_cursor = 0
    # read_uint16 should only read two bytes, not four
    codeflash_output = ds.read_uint16() # 1.56μs -> 1.04μs (49.8% faster)
    # Next two bytes are both zero
    codeflash_output = ds.read_uint16() # 658ns -> 503ns (30.8% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import struct

# imports
import pytest  # used for our unit tests
from electrum.transaction import BCDataStream


# Define a custom exception used in BCDataStream
class SerializationError(Exception):
    pass

# unit tests

# --- Basic Test Cases ---

def test_read_uint16_basic_values():
    # Test reading minimum value (0)
    stream = BCDataStream()
    stream.input = b'\x00\x00'
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 1.50μs -> 1.15μs (30.8% faster)

    # Test reading maximum value (65535)
    stream = BCDataStream()
    stream.input = b'\xff\xff'
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 684ns -> 559ns (22.4% faster)

    # Test reading a typical value (e.g. 1, 256, 4660)
    stream = BCDataStream()
    stream.input = b'\x01\x00'  # 1
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 513ns -> 384ns (33.6% faster)

    stream = BCDataStream()
    stream.input = b'\x00\x01'  # 256
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 493ns -> 354ns (39.3% faster)

    stream = BCDataStream()
    stream.input = b'\x34\x12'  # 0x1234 = 4660
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 455ns -> 355ns (28.2% faster)

def test_read_uint16_multiple_reads():
    # Test reading two values sequentially
    stream = BCDataStream()
    stream.input = b'\x01\x00\xff\xff'
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 1.53μs -> 1.12μs (36.7% faster)
    codeflash_output = stream.read_uint16() # 665ns -> 560ns (18.8% faster)

# --- Edge Test Cases ---

def test_read_uint16_empty_input():
    # Test reading from empty input
    stream = BCDataStream()
    stream.input = b''
    stream.read_cursor = 0
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_insufficient_bytes():
    # Test reading with only 1 byte available
    stream = BCDataStream()
    stream.input = b'\x01'
    stream.read_cursor = 0
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_cursor_at_end():
    # Test reading when cursor is at the very end
    stream = BCDataStream()
    stream.input = b'\x01\x00'
    stream.read_cursor = 2  # cursor at end
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_cursor_near_end():
    # Test reading when cursor is at the penultimate byte
    stream = BCDataStream()
    stream.input = b'\x01\x00'
    stream.read_cursor = 1  # only one byte left
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_non_byte_input():
    # Test reading when input is not bytes/bytearray
    stream = BCDataStream()
    stream.input = 'not bytes'
    stream.read_cursor = 0
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_none_input():
    # Test reading when input is None
    stream = BCDataStream()
    stream.input = None
    stream.read_cursor = 0
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_cursor_out_of_bounds():
    # Test reading when cursor is negative
    stream = BCDataStream()
    stream.input = b'\x01\x00'
    stream.read_cursor = -1
    with pytest.raises(SerializationError):
        stream.read_uint16()

    # Test reading when cursor is far beyond input
    stream = BCDataStream()
    stream.input = b'\x01\x00'
    stream.read_cursor = 100
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_cursor_advances_correctly():
    # Test that cursor advances exactly 2 bytes per read
    stream = BCDataStream()
    stream.input = b'\x01\x00\x02\x00\x03\x00'
    stream.read_cursor = 0
    codeflash_output = stream.read_uint16() # 2.73μs -> 1.98μs (38.2% faster)
    codeflash_output = stream.read_uint16() # 862ns -> 514ns (67.7% faster)
    codeflash_output = stream.read_uint16() # 516ns -> 347ns (48.7% faster)

# --- Large Scale Test Cases ---

def test_read_uint16_large_input():
    # Test reading many uint16 values from a large input
    N = 1000
    # Create input: [0, 1, 2, ..., 999] as uint16 little-endian
    values = list(range(N))
    input_bytes = b''.join(struct.pack('<H', v) for v in values)
    stream = BCDataStream()
    stream.input = input_bytes
    stream.read_cursor = 0
    for expected in values:
        codeflash_output = stream.read_uint16() # 458μs -> 339μs (35.1% faster)

def test_read_uint16_large_input_partial_read():
    # Test reading part of a large input, then stopping
    N = 1000
    values = list(range(N))
    input_bytes = b''.join(struct.pack('<H', v) for v in values)
    stream = BCDataStream()
    stream.input = input_bytes
    stream.read_cursor = 0
    # Read first 10 values
    for expected in values[:10]:
        codeflash_output = stream.read_uint16() # 6.18μs -> 4.41μs (40.1% faster)

def test_read_uint16_large_input_insufficient_bytes():
    # Test large input with last value incomplete
    N = 999
    values = list(range(N))
    input_bytes = b''.join(struct.pack('<H', v) for v in values) + b'\x01'  # 1 byte extra
    stream = BCDataStream()
    stream.input = input_bytes
    stream.read_cursor = 0
    for expected in values:
        codeflash_output = stream.read_uint16()
    # Now there is 1 byte left, which is insufficient for another uint16
    with pytest.raises(SerializationError):
        stream.read_uint16()

def test_read_uint16_large_input_cursor_offset():
    # Test reading from a large input with cursor offset
    N = 100
    values = list(range(N))
    input_bytes = b''.join(struct.pack('<H', v) for v in values)
    stream = BCDataStream()
    stream.input = input_bytes
    stream.read_cursor = 2 * 50  # start at value 50
    for expected in values[50:]:
        codeflash_output = stream.read_uint16() # 25.2μs -> 18.4μs (37.3% faster)
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
from electrum.transaction import BCDataStream
import pytest

def test_BCDataStream_read_uint16():
    with pytest.raises(SerializationError, match="a\\ bytes\\-like\\ object\\ is\\ required,\\ not\\ 'NoneType'"):
        BCDataStream.read_uint16(BCDataStream())

To edit these changes git checkout codeflash/optimize-BCDataStream.read_uint16-mhokukgc and push.