⚡️ Speed up method VideoSourcesManager.all_sources_ended by 25%

[codeflash-ai]

📄 25% (0.25x) speedup for VideoSourcesManager.all_sources_ended in inference/core/interfaces/camera/utils.py

⏱️ Runtime : 25.4 microseconds → 20.4 microseconds (best of 52 runs)

📝 Explanation and details

The optimization caches the length of video_sources.all_sources during initialization instead of computing it on every call to all_sources_ended(). This simple change delivers a 24% speedup by eliminating redundant len() calls.

Key optimization:

• Added self._all_sources_len = len(self._video_sources.all_sources) in __init__
• Changed comparison from len(self._video_sources.all_sources) to self._all_sources_len

Why this works:
The len() function has overhead when called repeatedly, even on simple containers. By pre-computing the length once during initialization, each call to all_sources_ended() saves the cost of:

1. Attribute lookup (self._video_sources.all_sources)
2. Function call overhead (len())
3. Potential iteration if all_sources is not a simple list/tuple

Performance characteristics:

• Best gains on large source counts: 42-51% faster for 1000+ sources
• Consistent improvements across all test cases: 6-51% speedup range
• Diminishing returns for very small source counts but still positive gains

Assumptions:
This optimization assumes video_sources.all_sources remains static after VideoSourcesManager initialization, which is typical for video source management where the set of cameras/streams is configured upfront. If sources can be dynamically added/removed, the cache would need invalidation logic.

The optimization is particularly valuable if all_sources_ended() is called frequently in monitoring loops or status checks, which is common in video processing pipelines.

✅ Correctness verification report:

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

🌀 Generated Regression Tests and Runtime

from datetime import datetime
from threading import Thread
from typing import Callable, Dict, Optional, Set

# imports
import pytest  # used for our unit tests
from inference.core.interfaces.camera.utils import VideoSourcesManager


# Minimal stub for SourceConnectionError, as it's not relevant for these tests
class SourceConnectionError(Exception):
    pass


# Minimal stub for VideoSources, just enough for our tests
class VideoSources:
    def __init__(self, all_sources):
        self.all_sources = all_sources


from inference.core.interfaces.camera.utils import VideoSourcesManager

# unit tests

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


def test_no_sources():
    # Scenario: No sources at all
    sources = VideoSources([])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    # With no sources, all_sources_ended should be True
    codeflash_output = manager.all_sources_ended()  # 551ns -> 371ns (48.5% faster)


def test_one_source_not_ended():
    # Scenario: One source, not ended
    sources = VideoSources([1])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    # No sources ended
    codeflash_output = manager.all_sources_ended()  # 480ns -> 423ns (13.5% faster)


def test_one_source_ended():
    # Scenario: One source, ended
    sources = VideoSources([1])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.add(1)
    codeflash_output = manager.all_sources_ended()  # 520ns -> 411ns (26.5% faster)


def test_multiple_sources_none_ended():
    # Scenario: Multiple sources, none ended
    sources = VideoSources([1, 2, 3])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    codeflash_output = manager.all_sources_ended()  # 561ns -> 398ns (41.0% faster)


def test_multiple_sources_some_ended():
    # Scenario: Multiple sources, some ended
    sources = VideoSources([1, 2, 3])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.add(1)
    codeflash_output = manager.all_sources_ended()  # 550ns -> 444ns (23.9% faster)
    manager._ended_sources.add(2)
    codeflash_output = manager.all_sources_ended()  # 311ns -> 236ns (31.8% faster)


def test_multiple_sources_all_ended():
    # Scenario: Multiple sources, all ended
    sources = VideoSources([1, 2, 3])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update([1, 2, 3])
    codeflash_output = manager.all_sources_ended()  # 523ns -> 408ns (28.2% faster)


def test_multiple_sources_more_ended_than_exist():
    # Scenario: Ended sources contains more than all_sources (should still return True)
    sources = VideoSources([1, 2, 3])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update([1, 2, 3, 4, 5])
    codeflash_output = manager.all_sources_ended()  # 514ns -> 435ns (18.2% faster)


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


def test_sources_with_non_integer_ids():
    # Scenario: Source IDs are strings
    sources = VideoSources(["camA", "camB", "camC"])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.add("camA")
    codeflash_output = manager.all_sources_ended()  # 531ns -> 445ns (19.3% faster)
    manager._ended_sources.update(["camB", "camC"])
    codeflash_output = manager.all_sources_ended()  # 321ns -> 294ns (9.18% faster)


def test_sources_with_mixed_types():
    # Scenario: Source IDs are mixed types
    sources = VideoSources([1, "camB", (3,)])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update([1, "camB"])
    codeflash_output = manager.all_sources_ended()  # 509ns -> 424ns (20.0% faster)
    manager._ended_sources.add((3,))
    codeflash_output = manager.all_sources_ended()  # 310ns -> 270ns (14.8% faster)


def test_ended_sources_with_nonexistent_ids():
    # Scenario: Ended sources contains IDs not present in all_sources
    sources = VideoSources([1, 2])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update([3, 4])
    # Should be False, since none of the actual sources are ended
    codeflash_output = manager.all_sources_ended()  # 478ns -> 417ns (14.6% faster)
    manager._ended_sources.add(1)
    codeflash_output = manager.all_sources_ended()  # 353ns -> 280ns (26.1% faster)
    manager._ended_sources.add(2)
    codeflash_output = manager.all_sources_ended()  # 215ns -> 187ns (15.0% faster)


def test_duplicate_source_ids():
    # Scenario: Duplicate IDs in all_sources (should count unique only)
    sources = VideoSources([1, 1, 2])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.add(1)
    # Only one unique ended source
    codeflash_output = manager.all_sources_ended()  # 545ns -> 443ns (23.0% faster)
    manager._ended_sources.add(2)
    # Now both unique sources are ended
    codeflash_output = manager.all_sources_ended()  # 273ns -> 241ns (13.3% faster)


def test_empty_ended_sources():
    # Scenario: ended_sources is empty, but all_sources is not
    sources = VideoSources([1, 2, 3])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    codeflash_output = manager.all_sources_ended()  # 498ns -> 421ns (18.3% faster)


def test_large_source_ids():
    # Scenario: Large integer IDs
    sources = VideoSources([999999999, 888888888])
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.add(999999999)
    codeflash_output = manager.all_sources_ended()  # 527ns -> 388ns (35.8% faster)
    manager._ended_sources.add(888888888)
    codeflash_output = manager.all_sources_ended()  # 331ns -> 257ns (28.8% faster)


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


def test_large_number_of_sources_none_ended():
    # Scenario: 1000 sources, none ended
    sources = VideoSources(list(range(1000)))
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    codeflash_output = manager.all_sources_ended()  # 630ns -> 416ns (51.4% faster)


def test_large_number_of_sources_half_ended():
    # Scenario: 1000 sources, half ended
    sources = VideoSources(list(range(1000)))
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update(range(500))
    codeflash_output = manager.all_sources_ended()  # 657ns -> 499ns (31.7% faster)


def test_large_number_of_sources_all_ended():
    # Scenario: 1000 sources, all ended
    sources = VideoSources(list(range(1000)))
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update(range(1000))
    codeflash_output = manager.all_sources_ended()  # 660ns -> 539ns (22.4% faster)


def test_large_number_of_sources_more_ended_than_exist():
    # Scenario: 1000 sources, ended sources contains more than 1000 unique IDs
    sources = VideoSources(list(range(1000)))
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update(range(1000))
    codeflash_output = manager.all_sources_ended()  # 639ns -> 502ns (27.3% faster)


def test_large_number_of_sources_with_nonexistent_ended():
    # Scenario: 1000 sources, ended sources only contains IDs not in all_sources
    sources = VideoSources(list(range(1000)))
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update(range(1000, 1100))
    codeflash_output = manager.all_sources_ended()  # 593ns -> 456ns (30.0% faster)


def test_large_number_of_sources_with_some_nonexistent_ended():
    # Scenario: 1000 sources, ended sources contains all actual sources and some extra
    sources = VideoSources(list(range(1000)))
    manager = VideoSourcesManager(sources, lambda: False, lambda i, e: None)
    manager._ended_sources.update(list(range(1000)) + list(range(1000, 1100)))
    codeflash_output = manager.all_sources_ended()  # 667ns -> 503ns (32.6% faster)


# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

from datetime import datetime
from threading import Thread
from typing import Callable, Dict, Optional, Set

# imports
import pytest
from inference.core.interfaces.camera.utils import VideoSourcesManager


# Dummy exception for compatibility with the provided code
class SourceConnectionError(Exception):
    pass


# Dummy VideoSources class for testing
class VideoSources:
    def __init__(self, all_sources):
        # all_sources: list of int (source ids)
        self.all_sources = all_sources


from inference.core.interfaces.camera.utils import VideoSourcesManager

# unit tests


# Helper function to create a dummy should_stop and on_reconnection_error
def dummy_should_stop():
    return False


def dummy_on_reconnection_error(source_id, error):
    pass


# ------------------ Basic Test Cases ------------------


def test_all_sources_ended_empty_sources():
    # Scenario: No sources at all
    # Expect: all_sources_ended should return True (since there are no sources to end)
    manager = VideoSourcesManager(
        video_sources=VideoSources([]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    codeflash_output = manager.all_sources_ended()  # 730ns -> 688ns (6.10% faster)


def test_all_sources_ended_no_sources_ended():
    # Scenario: Some sources, none ended
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, 2, 3]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    codeflash_output = manager.all_sources_ended()  # 594ns -> 450ns (32.0% faster)


def test_all_sources_ended_some_sources_ended():
    # Scenario: Some sources, one ended
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, 2, 3]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.add(1)
    codeflash_output = manager.all_sources_ended()  # 584ns -> 467ns (25.1% faster)


def test_all_sources_ended_all_sources_ended():
    # Scenario: All sources ended
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, 2, 3]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update([1, 2, 3])
    codeflash_output = manager.all_sources_ended()  # 580ns -> 455ns (27.5% faster)


def test_all_sources_ended_more_ended_than_sources():
    # Scenario: _ended_sources has more entries than all_sources (should still return True)
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, 2]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update([1, 2, 3, 4])
    codeflash_output = manager.all_sources_ended()  # 580ns -> 417ns (39.1% faster)


# ------------------ Edge Test Cases ------------------


def test_all_sources_ended_duplicate_source_ids():
    # Scenario: Duplicate source IDs in all_sources (should treat as unique by set logic)
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, 1, 2]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    # Only one entry for '1' should be counted
    manager._ended_sources.update([1, 2])
    codeflash_output = manager.all_sources_ended()  # 581ns -> 478ns (21.5% faster)


def test_all_sources_ended_negative_and_zero_source_ids():
    # Scenario: Negative and zero source IDs
    manager = VideoSourcesManager(
        video_sources=VideoSources([0, -1, 2]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update([0, -1])
    codeflash_output = manager.all_sources_ended()  # 579ns -> 444ns (30.4% faster)
    manager._ended_sources.add(2)
    codeflash_output = manager.all_sources_ended()  # 310ns -> 278ns (11.5% faster)


def test_all_sources_ended_non_integer_source_ids():
    # Scenario: Non-integer source IDs (e.g., string, tuple)
    manager = VideoSourcesManager(
        video_sources=VideoSources(["a", (1, 2), 3.5]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update(["a", (1, 2)])
    codeflash_output = manager.all_sources_ended()  # 543ns -> 434ns (25.1% faster)
    manager._ended_sources.add(3.5)
    codeflash_output = manager.all_sources_ended()  # 322ns -> 296ns (8.78% faster)


def test_all_sources_ended_empty_ended_sources():
    # Scenario: _ended_sources is empty, but all_sources is not
    manager = VideoSourcesManager(
        video_sources=VideoSources([10, 20, 30]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    codeflash_output = manager.all_sources_ended()  # 489ns -> 440ns (11.1% faster)


def test_all_sources_ended_large_source_id_values():
    # Scenario: Very large integer source IDs
    manager = VideoSourcesManager(
        video_sources=VideoSources([999999999, 888888888]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.add(999999999)
    codeflash_output = manager.all_sources_ended()  # 522ns -> 423ns (23.4% faster)
    manager._ended_sources.add(888888888)
    codeflash_output = manager.all_sources_ended()  # 332ns -> 286ns (16.1% faster)


# ------------------ Large Scale Test Cases ------------------


def test_all_sources_ended_large_number_of_sources_none_ended():
    # Scenario: 1000 sources, none ended
    sources = list(range(1000))
    manager = VideoSourcesManager(
        video_sources=VideoSources(sources),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    codeflash_output = manager.all_sources_ended()  # 586ns -> 411ns (42.6% faster)


def test_all_sources_ended_large_number_of_sources_some_ended():
    # Scenario: 1000 sources, half ended
    sources = list(range(1000))
    manager = VideoSourcesManager(
        video_sources=VideoSources(sources),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update(sources[:500])
    codeflash_output = manager.all_sources_ended()  # 684ns -> 500ns (36.8% faster)


def test_all_sources_ended_large_number_of_sources_all_ended():
    # Scenario: 1000 sources, all ended
    sources = list(range(1000))
    manager = VideoSourcesManager(
        video_sources=VideoSources(sources),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update(sources)
    codeflash_output = manager.all_sources_ended()  # 629ns -> 508ns (23.8% faster)


def test_all_sources_ended_large_number_of_sources_extra_ended():
    # Scenario: 1000 sources, all ended plus extra ended sources
    sources = list(range(1000))
    manager = VideoSourcesManager(
        video_sources=VideoSources(sources),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update(sources + [1001, 1002])
    codeflash_output = manager.all_sources_ended()  # 634ns -> 489ns (29.7% faster)


def test_all_sources_ended_large_number_of_sources_empty_sources():
    # Scenario: 0 sources, no ended sources
    manager = VideoSourcesManager(
        video_sources=VideoSources([]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    codeflash_output = manager.all_sources_ended()  # 549ns -> 416ns (32.0% faster)


# ------------------ Mutation-sensitive Test ------------------


def test_all_sources_ended_mutation_sensitivity():
    # Changing the comparison to > instead of >= would break this test!
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, 2]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update([1, 2])
    codeflash_output = manager.all_sources_ended()  # 553ns -> 445ns (24.3% faster)
    # Remove one ended source
    manager._ended_sources.remove(2)
    codeflash_output = manager.all_sources_ended()  # 355ns -> 310ns (14.5% faster)
    # Add a third ended source not present in all_sources
    manager._ended_sources.add(3)
    codeflash_output = manager.all_sources_ended()  # 234ns -> 196ns (19.4% faster)


# ------------------ Additional Edge Cases ------------------


def test_all_sources_ended_sources_with_none_id():
    # Scenario: Source IDs include None
    manager = VideoSourcesManager(
        video_sources=VideoSources([None, 1]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.add(1)
    codeflash_output = manager.all_sources_ended()  # 530ns -> 440ns (20.5% faster)
    manager._ended_sources.add(None)
    codeflash_output = manager.all_sources_ended()  # 326ns -> 267ns (22.1% faster)


def test_all_sources_ended_sources_with_mixed_types():
    # Scenario: Source IDs with mixed types
    manager = VideoSourcesManager(
        video_sources=VideoSources([1, "2", (3, 4)]),
        should_stop=dummy_should_stop,
        on_reconnection_error=dummy_on_reconnection_error,
    )
    manager._ended_sources.update([1, "2"])
    codeflash_output = manager.all_sources_ended()  # 517ns -> 439ns (17.8% faster)
    manager._ended_sources.add((3, 4))
    codeflash_output = manager.all_sources_ended()  # 333ns -> 276ns (20.7% faster)


# 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-VideoSourcesManager.all_sources_ended-miqvkdp1 and push.