⚡️ Speed up function get_async_inline_code by 33% in PR #1518 (proper-async)

[codeflash-ai]

⚡️ This pull request contains optimizations for PR #1518

If you approve this dependent PR, these changes will be merged into the original PR branch proper-async.

This PR will be automatically closed if the original PR is merged.

---

📄 33% (0.33x) speedup for get_async_inline_code in codeflash/code_utils/instrument_existing_tests.py

⏱️ Runtime : 515 microseconds → 388 microseconds (best of 152 runs)

📝 Explanation and details

The optimization achieves a 32% runtime improvement by eliminating redundant work on every function call through two key changes:

Primary Optimization: Module-Level Constants + Caching

What changed:

1. Module-level string constants: The large inline code strings (1000+ characters each) are now defined once as module-level constants (_BEHAVIOR_ASYNC_INLINE_CODE, _PERFORMANCE_ASYNC_INLINE_CODE, _CONCURRENCY_ASYNC_INLINE_CODE) instead of being reconstructed as string literals on every function call.

2. Cached dispatcher with dictionary lookup: The get_async_inline_code() function is decorated with @cache and uses a pre-built dictionary (_INLINE_CODE_MAP) for O(1) mode lookups, replacing the sequential if-statement chain.

Why this is faster:

• Eliminates string allocation overhead: In the original code, Python had to allocate and construct the multi-line string literal every time a function was called. String literals in function bodies are not automatically interned, so each call created a new string object. The optimized version references the same string object stored at module initialization.

• Reduces CPU instruction count: The original sequential if-checks required evaluating up to 2 enum comparisons per call. The optimized dictionary lookup is a single hash table access (~O(1)) that's even faster with @cache memoization—subsequent calls with the same TestingMode return the cached result immediately without any dictionary lookup.

• Caching multiplier effect: The @cache decorator means the first call with each TestingMode performs the dictionary lookup once, then all subsequent calls with that mode are nearly instant pointer returns from the cache.

How this impacts real workloads:
Based on the function_references, get_async_inline_code() is called during test instrumentation in hot paths like test_async_bubble_sort_behavior_results(), test_async_function_performance_mode(), and test_async_function_error_handling(). These test setup functions likely run many times during development and CI/CD pipelines. The optimization means:

• Test instrumentation is faster: Setting up async decorators for behavior/performance testing completes 32% faster, reducing overall test suite setup time.
• Scales with test volume: The annotated tests show improvements compound in loops—test_mass_compilation_of_generated_codes_varied_modes runs 38.6% faster (329μs → 237μs) when calling the function 1000 times.
• Best for repeated mode access: Tests that call the same mode multiple times benefit most from caching (e.g., test_get_async_inline_code_called_multiple_times_performance shows 44.1% speedup for 100 calls).

The optimization trades a negligible increase in module initialization time and memory (storing three strings at module level) for substantial per-call speedup, making it particularly effective for test instrumentation workflows that repeatedly access the same testing mode configurations.

✅ Correctness verification report:

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

🌀 Click to see Generated Regression Tests

import pytest  # used for our unit tests
from codeflash.code_utils.instrument_existing_tests import \
    get_async_inline_code
from codeflash.models.models import TestingMode

def test_behavior_mode_returns_behavior_code_and_is_valid_python():
    # Call the function with the BEHAVIOR enum value.
    codeflash_output = get_async_inline_code(TestingMode.BEHAVIOR); code = codeflash_output # 782ns -> 1.31μs (40.4% slower)
    # Ensure the returned string is syntactically valid python code by compiling it.
    compile(code, "<behavior-inline>", "exec")

def test_concurrency_mode_returns_concurrency_code_and_contains_expected_parts():
    # Call for concurrency mode.
    codeflash_output = get_async_inline_code(TestingMode.CONCURRENCY); code = codeflash_output # 1.00μs -> 1.29μs (22.5% slower)
    # Ensure the string compiles as Python source.
    compile(code, "<concurrency-inline>", "exec")

def test_performance_mode_returns_performance_code_and_is_valid_python():
    # Call for performance mode.
    codeflash_output = get_async_inline_code(TestingMode.PERFORMANCE); code = codeflash_output # 1.00μs -> 1.29μs (22.5% slower)
    # Compile to ensure valid Python.
    compile(code, "<performance-inline>", "exec")

def test_line_profile_mode_maps_to_performance_code():
    # LINE_PROFILE is not explicitly handled; function should return performance code for that enum.
    codeflash_output = get_async_inline_code(TestingMode.LINE_PROFILE); code_line_profile = codeflash_output # 992ns -> 1.27μs (22.0% slower)
    codeflash_output = get_async_inline_code(TestingMode.PERFORMANCE); code_performance = codeflash_output # 441ns -> 701ns (37.1% slower)
    # Compile the result to ensure syntactic validity.
    compile(code_line_profile, "<line-profile-inline>", "exec")

def test_unexpected_input_types_fall_back_to_performance_code():
    # Passing None (not of TestingMode) should hit the fallback branch and return performance code.
    codeflash_output = get_async_inline_code(None); code_none = codeflash_output # 1.18μs -> 841ns (40.5% faster)
    codeflash_output = get_async_inline_code(TestingMode.PERFORMANCE); code_performance = codeflash_output # 501ns -> 782ns (35.9% slower)
    # Passing an integer also should fall back to performance code.
    codeflash_output = get_async_inline_code(0); code_int = codeflash_output # 431ns -> 330ns (30.6% faster)
    # All compiled successfully.
    compile(code_none, "<none-inline>", "exec")
    compile(code_int, "<int-inline>", "exec")

def test_all_enum_values_return_non_empty_strings_and_compile():
    # Iterate through all enum members to ensure each produces a non-empty string and valid code.
    for mode in TestingMode:
        codeflash_output = get_async_inline_code(mode); code = codeflash_output # 3.64μs -> 5.35μs (32.0% slower)
        # Attempt to compile each to ensure no syntax errors across modes.
        compile(code, f"<compile-{mode.name}>", "exec")

def test_repeated_calls_are_consistent_and_compile_many_times():
    # For performance reasons but still meeting spec, do 1000 iterations.
    iterations = 1000
    # Collect results for each mode to assert consistency.
    results_behavior = [get_async_inline_code(TestingMode.BEHAVIOR) for _ in range(iterations)]
    results_concurrency = [get_async_inline_code(TestingMode.CONCURRENCY) for _ in range(iterations)]
    results_performance = [get_async_inline_code(TestingMode.PERFORMANCE) for _ in range(iterations)]
    # Compile each unique code once to ensure validity (avoid repeated compilations).
    compile(results_behavior[0], "<behavior-large>", "exec")
    compile(results_concurrency[0], "<concurrency-large>", "exec")
    compile(results_performance[0], "<performance-large>", "exec")

def test_mass_compilation_of_generated_codes_varied_modes():
    # Generate 1000 codes cycling through the three handled branches and compile each one.
    codes = []
    modes = [TestingMode.BEHAVIOR, TestingMode.CONCURRENCY, TestingMode.PERFORMANCE]
    for i in range(1000):
        mode = modes[i % len(modes)]
        codes.append(get_async_inline_code(mode)) # 329μs -> 237μs (38.6% faster)
    for index, code in enumerate(codes):
        # Each compiled in its own name to ease debugging if compilation fails.
        compile(code, f"<mass-compile-{index}>", "exec")
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import pytest
from codeflash.code_utils.instrument_existing_tests import (
    get_async_inline_code, get_behavior_async_inline_code,
    get_concurrency_async_inline_code, get_performance_async_inline_code)
from codeflash.models.models import TestingMode

def test_get_async_inline_code_behavior_mode():
    """Test that get_async_inline_code returns behavior code for BEHAVIOR mode."""
    codeflash_output = get_async_inline_code(TestingMode.BEHAVIOR); result = codeflash_output # 771ns -> 1.21μs (36.4% slower)

def test_get_async_inline_code_performance_mode():
    """Test that get_async_inline_code returns performance code for PERFORMANCE mode."""
    codeflash_output = get_async_inline_code(TestingMode.PERFORMANCE); result = codeflash_output # 931ns -> 1.19μs (21.9% slower)

def test_get_async_inline_code_concurrency_mode():
    """Test that get_async_inline_code returns concurrency code for CONCURRENCY mode."""
    codeflash_output = get_async_inline_code(TestingMode.CONCURRENCY); result = codeflash_output # 872ns -> 1.26μs (30.9% slower)

def test_get_async_inline_code_line_profile_mode_defaults_to_performance():
    """Test that LINE_PROFILE mode returns performance code (default behavior)."""
    codeflash_output = get_async_inline_code(TestingMode.LINE_PROFILE); result = codeflash_output # 862ns -> 1.18μs (27.1% slower)

def test_behavior_async_inline_code_contains_imports():
    """Test that behavior code contains necessary imports."""
    result = get_behavior_async_inline_code()

def test_performance_async_inline_code_contains_imports():
    """Test that performance code contains necessary imports."""
    result = get_performance_async_inline_code()

def test_concurrency_async_inline_code_contains_imports():
    """Test that concurrency code contains necessary imports."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_contains_function_wrapper():
    """Test that behavior code defines the wrapper function."""
    result = get_behavior_async_inline_code()

def test_performance_code_contains_function_wrapper():
    """Test that performance code defines the wrapper function."""
    result = get_performance_async_inline_code()

def test_concurrency_code_contains_function_wrapper():
    """Test that concurrency code defines the wrapper function."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_contains_database_operations():
    """Test that behavior code contains sqlite3 database operations."""
    result = get_behavior_async_inline_code()

def test_behavior_code_contains_pickle_operations():
    """Test that behavior code uses pickle for serialization."""
    result = get_behavior_async_inline_code()

def test_performance_code_contains_timing_logic():
    """Test that performance code contains timing measurements."""
    result = get_performance_async_inline_code()

def test_concurrency_code_contains_gather_logic():
    """Test that concurrency code uses asyncio.gather."""
    result = get_concurrency_async_inline_code()

def test_all_modes_return_different_strings():
    """Test that different modes return different code strings."""
    codeflash_output = get_async_inline_code(TestingMode.BEHAVIOR); behavior = codeflash_output # 791ns -> 1.31μs (39.7% slower)
    codeflash_output = get_async_inline_code(TestingMode.PERFORMANCE); performance = codeflash_output # 621ns -> 702ns (11.5% slower)
    codeflash_output = get_async_inline_code(TestingMode.CONCURRENCY); concurrency = codeflash_output # 460ns -> 551ns (16.5% slower)

def test_get_async_inline_code_with_enum_member():
    """Test that function works with actual enum members."""
    for mode in TestingMode:
        codeflash_output = get_async_inline_code(mode); result = codeflash_output # 2.12μs -> 2.94μs (28.2% slower)

def test_behavior_code_not_empty():
    """Test that behavior code is not empty."""
    result = get_behavior_async_inline_code()

def test_performance_code_not_empty():
    """Test that performance code is not empty."""
    result = get_performance_async_inline_code()

def test_concurrency_code_not_empty():
    """Test that concurrency code is not empty."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_has_complete_decorator():
    """Test that behavior code has a complete decorator structure."""
    result = get_behavior_async_inline_code()

def test_performance_code_has_complete_decorator():
    """Test that performance code has a complete decorator structure."""
    result = get_performance_async_inline_code()

def test_concurrency_code_has_complete_decorator():
    """Test that concurrency code has a complete decorator structure."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_handles_exceptions():
    """Test that behavior code contains exception handling."""
    result = get_behavior_async_inline_code()

def test_performance_code_handles_exceptions():
    """Test that performance code contains exception handling."""
    result = get_performance_async_inline_code()

def test_concurrency_code_handles_exceptions():
    """Test that concurrency code contains exception handling."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_disables_garbage_collection():
    """Test that behavior code disables and re-enables garbage collection."""
    result = get_behavior_async_inline_code()

def test_performance_code_disables_garbage_collection():
    """Test that performance code disables and re-enables garbage collection."""
    result = get_performance_async_inline_code()

def test_concurrency_code_disables_garbage_collection():
    """Test that concurrency code disables and re-enables garbage collection."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_uses_environment_variables():
    """Test that behavior code reads required environment variables."""
    result = get_behavior_async_inline_code()

def test_performance_code_uses_environment_variables():
    """Test that performance code reads required environment variables."""
    result = get_performance_async_inline_code()

def test_concurrency_code_uses_environment_variables():
    """Test that concurrency code reads environment variables."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_extracts_test_context():
    """Test that behavior code extracts test context from environment."""
    result = get_behavior_async_inline_code()

def test_performance_code_extracts_test_context():
    """Test that performance code extracts test context from environment."""
    result = get_performance_async_inline_code()

def test_behavior_code_uses_asyncio_loop():
    """Test that behavior code uses asyncio loop."""
    result = get_behavior_async_inline_code()

def test_performance_code_uses_asyncio_loop():
    """Test that performance code uses asyncio loop."""
    result = get_performance_async_inline_code()

def test_concurrency_code_uses_asyncio_gather():
    """Test that concurrency code properly uses asyncio.gather."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_prints_output_tags():
    """Test that behavior code prints test output tags."""
    result = get_behavior_async_inline_code()

def test_performance_code_prints_output_tags():
    """Test that performance code prints test output tags."""
    result = get_performance_async_inline_code()

def test_concurrency_code_prints_output_tags():
    """Test that concurrency code prints output tags."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_creates_temp_file():
    """Test that behavior code creates temporary files."""
    result = get_behavior_async_inline_code()

def test_behavior_code_handles_test_iteration():
    """Test that behavior code handles test iterations."""
    result = get_behavior_async_inline_code()

def test_all_codes_are_valid_python():
    """Test that all generated code snippets are syntactically valid Python."""
    codes = [
        get_behavior_async_inline_code(),
        get_performance_async_inline_code(),
        get_concurrency_async_inline_code(),
    ]
    for code in codes:
        # This will raise SyntaxError if code is invalid
        compile(code, '<string>', 'exec')

def test_get_async_inline_code_called_multiple_times_behavior():
    """Test that get_async_inline_code can be called many times with BEHAVIOR mode."""
    results = []
    for _ in range(100):
        codeflash_output = get_async_inline_code(TestingMode.BEHAVIOR); result = codeflash_output # 26.6μs -> 24.7μs (7.93% faster)
        results.append(result)

def test_get_async_inline_code_called_multiple_times_performance():
    """Test that get_async_inline_code can be called many times with PERFORMANCE mode."""
    results = []
    for _ in range(100):
        codeflash_output = get_async_inline_code(TestingMode.PERFORMANCE); result = codeflash_output # 35.5μs -> 24.6μs (44.1% faster)
        results.append(result)

def test_get_async_inline_code_called_multiple_times_concurrency():
    """Test that get_async_inline_code can be called many times with CONCURRENCY mode."""
    results = []
    for _ in range(100):
        codeflash_output = get_async_inline_code(TestingMode.CONCURRENCY); result = codeflash_output # 34.6μs -> 26.4μs (30.9% faster)
        results.append(result)

def test_get_async_inline_code_called_multiple_times_all_modes():
    """Test that get_async_inline_code can be called many times across all modes."""
    mode_results = {mode: [] for mode in TestingMode}
    for _ in range(50):
        for mode in TestingMode:
            codeflash_output = get_async_inline_code(mode); result = codeflash_output
            mode_results[mode].append(result)
    # Verify consistency within each mode
    for mode in TestingMode:
        pass

def test_behavior_code_string_length():
    """Test that behavior code has reasonable length."""
    result = get_behavior_async_inline_code()

def test_performance_code_string_length():
    """Test that performance code has reasonable length."""
    result = get_performance_async_inline_code()

def test_concurrency_code_string_length():
    """Test that concurrency code has reasonable length."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_contains_all_required_functions():
    """Test that behavior code defines all required helper functions."""
    result = get_behavior_async_inline_code()

def test_performance_code_contains_all_required_functions():
    """Test that performance code defines all required helper functions."""
    result = get_performance_async_inline_code()

def test_concurrency_code_contains_all_required_functions():
    """Test that concurrency code defines all required function."""
    result = get_concurrency_async_inline_code()

def test_behavior_code_variable_initialization():
    """Test that behavior code properly initializes variables."""
    result = get_behavior_async_inline_code()

def test_behavior_code_test_id_creation():
    """Test that behavior code creates test IDs correctly."""
    result = get_behavior_async_inline_code()

def test_performance_code_test_id_creation():
    """Test that performance code creates test IDs correctly."""
    result = get_performance_async_inline_code()

def test_behavior_code_with_large_repetitions():
    """Test behavior code string remains consistent across many iterations."""
    first_call = get_behavior_async_inline_code()
    for _ in range(500):
        current_call = get_behavior_async_inline_code()

def test_performance_code_with_large_repetitions():
    """Test performance code string remains consistent across many iterations."""
    first_call = get_performance_async_inline_code()
    for _ in range(500):
        current_call = get_performance_async_inline_code()

def test_concurrency_code_with_large_repetitions():
    """Test concurrency code string remains consistent across many iterations."""
    first_call = get_concurrency_async_inline_code()
    for _ in range(500):
        current_call = get_concurrency_async_inline_code()

def test_behavior_code_line_count():
    """Test that behavior code has reasonable number of lines."""
    result = get_behavior_async_inline_code()
    lines = result.split('\n')

def test_performance_code_line_count():
    """Test that performance code has reasonable number of lines."""
    result = get_performance_async_inline_code()
    lines = result.split('\n')

def test_concurrency_code_line_count():
    """Test that concurrency code has reasonable number of lines."""
    result = get_concurrency_async_inline_code()
    lines = result.split('\n')

def test_behavior_code_indentation_consistency():
    """Test that behavior code has consistent indentation."""
    result = get_behavior_async_inline_code()
    lines = result.split('\n')
    # Check that most lines start with 0, 4, 8, 12 spaces (standard Python)
    for line in lines:
        if line.strip():
            indent = len(line) - len(line.lstrip())

def test_performance_code_indentation_consistency():
    """Test that performance code has consistent indentation."""
    result = get_performance_async_inline_code()
    lines = result.split('\n')
    for line in lines:
        if line.strip():
            indent = len(line) - len(line.lstrip())

def test_concurrency_code_indentation_consistency():
    """Test that concurrency code has consistent indentation."""
    result = get_concurrency_async_inline_code()
    lines = result.split('\n')
    for line in lines:
        if line.strip():
            indent = len(line) - len(line.lstrip())
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-pr1518-2026-02-18T06.52.51 and push.

[claude]

Minor: Dead code / string duplication

The get_async_inline_code function now uses _INLINE_CODE_MAP (module-level constants), but the three original functions (get_behavior_async_inline_code, get_performance_async_inline_code, get_concurrency_async_inline_code) still exist and return identical strings. This means each inline code string is stored twice in memory.

Consider either:

• Removing the old functions and keeping only the module-level constants, or
• Having the old functions reference the constants (e.g., return _BEHAVIOR_ASYNC_INLINE_CODE)

Not a bug — just unnecessary duplication.

[claude]

PR Review Summary

Prek Checks

• Fixed: Unsorted imports (ruff I001) in codeflash/code_utils/instrument_existing_tests.py — from functools import cache moved to correct alphabetical position among stdlib imports.
• Status: All prek checks now pass (committed and pushed as 867d67d3).

Mypy

• 0 new errors introduced by this PR. All 83 mypy errors in the changed files are pre-existing on the base branch.

Code Review

No critical bugs, security vulnerabilities, or breaking API changes found.

This PR optimizes get_async_inline_code() by:

1. Moving inline code strings to module-level constants (_BEHAVIOR_ASYNC_INLINE_CODE, etc.)
2. Using a dictionary lookup (_INLINE_CODE_MAP) instead of if/elif chain
3. Adding @cache memoization on get_async_inline_code()

The string constants were verified to be byte-identical to the original function return values.

Minor observation (inline comment posted): The three original get_behavior_async_inline_code(), get_performance_async_inline_code(), and get_concurrency_async_inline_code() functions are now dead code — they're defined but never called. This causes each inline code string to be stored twice in the module. Not a bug, but unnecessary duplication.

Test Coverage

File	Stmts	Miss	Coverage
codeflash/code_utils/instrument_existing_tests.py	441	44	90%

• Test results: 2375 passed, 8 failed (pre-existing failures unrelated to this PR), 56 skipped
• Base branch (proper-async) coverage comparison was not possible due to CI environment constraints, but since this is a pure performance optimization with no logic changes and identical string content, coverage is not expected to regress.
• Coverage at 90% exceeds the 75% threshold for modified files.

Optimization PRs (codeflash-ai[bot])

Checked CI status for all 29 open codeflash optimization PRs. No PRs were merged — all have at least some failing checks. The closest candidates (PRs #1344, #1346, #1350, #1359) pass all substantive CI checks but fail on code/snyk (quota limit) and claude-review (external bot), which may not be required checks.

---

Last updated: 2026-02-18T07:15Z