⚡️ Speed up method VideoSourcesManager.join_all_reconnection_threads by 9%

[misrasaurabh1]

Saurabh's comment - It seems to speed up important part of the camera frame capture code. I would ask you to also manually verify this code, even though this looks good to me.

📄 9% (0.09x) speedup for VideoSourcesManager.join_all_reconnection_threads in inference/core/interfaces/camera/utils.py

⏱️ Runtime : 1.17 millisecond → 1.08 millisecond (best of 46 runs)

📝 Explanation and details

Here is the optimized version of your Python program.

Optimizations Applied.

1. Avoid Unnecessary Copies.

   • Removed the copy import and utilized the copy() method of sets directly.
   • Combined the purge operations into one helper function to avoid redundancy.

2. Data Structure Optimizations.

   • Changed _enforce_stop from a Dict[int, bool] to a Set[int] for more efficient membership checks and removals.

3. Helper Method.

   • Created a helper method _purge_reconnection_threads to consolidate the purge logic, thereby avoiding repeated code.

4. Use of Set discard.

   • Used discard instead of remove to avoid key errors if the key does not exist in the set.

✅ Correctness verification report:

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

🌀 Generated Regression Tests Details

from copy import copy
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


# function to test
class SourceConnectionError(Exception):
    pass

class VideoSources:
    pass


# unit tests
def test_no_threads_to_join():
    """Test with no threads to join."""
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    manager.join_all_reconnection_threads()




def test_thread_join_raises_exception(monkeypatch):
    """Test scenario where Thread.join() raises an exception."""
    def mock_join():
        raise RuntimeError("Join failed")
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    thread = Thread(target=lambda: None)
    monkeypatch.setattr(thread, 'join', mock_join)
    manager._threads_to_join.add(1)
    manager._reconnection_threads[1] = thread
    with pytest.raises(RuntimeError):
        manager.join_all_reconnection_threads()



from copy import copy
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


# function to test
class SourceConnectionError(Exception):
    pass

class VideoSources:
    pass

# Dummy function for thread target
def dummy_function():
    pass

# unit tests
def test_empty_threads_to_join_and_reconnection_threads():
    """Test with empty _threads_to_join and _reconnection_threads."""
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    manager.join_all_reconnection_threads(include_not_finished=False)

    manager.join_all_reconnection_threads(include_not_finished=True)

def test_single_thread_in_threads_to_join():
    """Test with a single thread in _threads_to_join."""
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    thread = Thread(target=dummy_function)
    manager._reconnection_threads[1] = thread
    manager._threads_to_join.add(1)
    thread.start()
    thread.join()  # Ensure the thread finishes
    manager.join_all_reconnection_threads(include_not_finished=False)

def test_multiple_threads_in_threads_to_join():
    """Test with multiple threads in _threads_to_join."""
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    thread1 = Thread(target=dummy_function)
    thread2 = Thread(target=dummy_function)
    manager._reconnection_threads[1] = thread1
    manager._reconnection_threads[2] = thread2
    manager._threads_to_join.add(1)
    manager._threads_to_join.add(2)
    thread1.start()
    thread2.start()
    thread1.join()  # Ensure the threads finish
    thread2.join()
    manager.join_all_reconnection_threads(include_not_finished=False)

def test_non_existent_thread_ids():
    """Test with non-existent thread IDs in _threads_to_join."""
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    manager._threads_to_join.add(999)
    manager.join_all_reconnection_threads(include_not_finished=False)

def test_thread_join_exception(monkeypatch):
    """Test handling of exceptions during thread join."""
    def mock_join():
        raise RuntimeError("Join failed")
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    thread = Thread(target=dummy_function)
    manager._reconnection_threads[1] = thread
    manager._threads_to_join.add(1)
    monkeypatch.setattr(thread, 'join', mock_join)
    with pytest.raises(RuntimeError):
        manager.join_all_reconnection_threads(include_not_finished=False)

def test_state_consistency_after_exception(monkeypatch):
    """Ensure state consistency after exceptions during thread join."""
    def mock_join():
        raise RuntimeError("Join failed")
    manager = VideoSourcesManager(VideoSources(), lambda: False, lambda x, y: None)
    thread = Thread(target=dummy_function)
    manager._reconnection_threads[1] = thread
    manager._threads_to_join.add(1)
    monkeypatch.setattr(thread, 'join', mock_join)
    try:
        manager.join_all_reconnection_threads(include_not_finished=False)
    except RuntimeError:
        pass