Parallel manager2

[lolyu]

Description of PR

Summary:
Fixes # (issue)

Type of change

• Bug fix
• Testbed and Framework(new/improvement)
• New Test case
  • Skipped for non-supported platforms
• Test case improvement

Back port request

• 202205
• 202305
• 202311
• 202405
• 202411
• 202505
• 202511

Approach

What is the motivation for this PR?

How did you do it?

How did you verify/test it?

Any platform specific information?

Supported testbed topology if it's a new test case?

Documentation

[lolyu]

@greptile

[greptile-apps]

Greptile Overview

Greptile Summary

This PR introduces a parallel fixture execution framework to optimize test execution time by parallelizing fixture setup/teardown operations. The implementation includes:

• Parallel Fixture Manager: A new pytest plugin (parallel_fixture) that uses ThreadPoolExecutor to execute fixture setup/teardown tasks concurrently across different scopes (SESSION, MODULE, CLASS, FUNCTION)
• Scope-based barriers: Automatic synchronization points that ensure all parallel tasks complete before proceeding to the next test scope
• Deadlock prevention: Fork-safety mechanisms using os.register_at_fork to prevent logging and Ansible display deadlocks in multi-threaded/multi-process environments
• Closure bug fixes: Corrected several __getattr__ implementations in base.py, duthosts.py, and multi_asic.py where module names were incorrectly captured by reference instead of by value
• Docker capability: Added SYS_PTRACE capability to enable thread interruption for timeout handling

The architecture is well-documented and addresses real concurrency challenges in the SONiC test infrastructure. However, there are several logical issues that need attention:

• Missing return value check in PyThreadState_SetAsyncExc could cause silent failures when interrupting threads
• Exception re-raising loses original traceback context, making debugging harder
• The reset() method doesn't clear the terminated flag, potentially preventing the manager from being reused
• Potential AttributeError when checking handler.lock that could be None

Confidence Score: 3/5

• This PR has logical issues that could cause runtime failures in edge cases but is generally sound in design
• The PR implements a complex parallel execution framework with good architecture and documentation. However, several critical logical issues were found: (1) missing return value checks for thread injection API that could silently fail, (2) exception handling that loses traceback context, (3) the reset mechanism doesn't clear the terminated flag which breaks reusability, and (4) potential None reference on handler.lock. These issues should be addressed before merging to ensure reliability in production.
• tests/common/plugins/parallel_fixture/__init__.py and tests/common/helpers/parallel.py need attention for the logical issues mentioned

Important Files Changed

Filename	Overview
tests/common/devices/base.py	Fixed closure bug where module_name was captured incorrectly by returning a wrapper function
tests/common/devices/duthosts.py	Fixed closure bug and added explicit initialization of _nodes attributes to prevent __getattr__ recursion
tests/common/devices/multi_asic.py	Fixed closure bug where multi_asic_attr was captured incorrectly by returning a wrapper function
tests/common/helpers/parallel.py	Added deadlock prevention for fork operations with logging locks, patched Ansible worker processes
tests/common/plugins/parallel_fixture/init.py	Implemented parallel fixture execution framework with thread pool, scope-based barriers, and exception handling

Sequence Diagram

sequenceDiagram
    participant Pytest
    participant ParallelManager
    participant ThreadPool
    participant MonitorThread
    participant Fixture
    participant Barrier

    Note over Pytest,Barrier: Session Setup Phase
    Pytest->>ParallelManager: Create (session scope)
    ParallelManager->>ThreadPool: Initialize worker threads
    ParallelManager->>MonitorThread: Start monitoring
    
    Pytest->>Fixture: Execute fixture (session scope)
    Fixture->>ParallelManager: submit_setup_task(SESSION, func)
    ParallelManager->>ThreadPool: Submit task to thread pool
    ThreadPool-->>ParallelManager: Return future
    Fixture-->>Pytest: Yield immediately
    
    Pytest->>Barrier: setup_barrier_session
    Barrier->>ParallelManager: wait_for_setup_tasks(SESSION)
    ParallelManager->>ThreadPool: Wait for all session tasks
    ThreadPool-->>ParallelManager: Tasks complete
    
    Note over Pytest,Barrier: Module/Class/Function Scopes
    Pytest->>Fixture: Execute fixture (module scope)
    Fixture->>ParallelManager: submit_setup_task(MODULE, func)
    ParallelManager->>ThreadPool: Submit to pool
    Fixture-->>Pytest: Yield
    
    Pytest->>Barrier: setup_barrier_module
    Barrier->>ParallelManager: wait_for_setup_tasks(MODULE)
    ParallelManager->>ThreadPool: Wait for module tasks
    
    Note over Pytest,Barrier: Test Execution
    Pytest->>ParallelManager: pytest_runtest_call hook
    ParallelManager->>ThreadPool: Ensure all tasks complete
    ParallelManager->>ThreadPool: Terminate executor
    Pytest->>Pytest: Run test function
    
    Note over Pytest,Barrier: Teardown Phase
    Pytest->>ParallelManager: pytest_runtest_teardown hook
    ParallelManager->>ThreadPool: Reset and create new executor
    ParallelManager->>MonitorThread: Restart monitoring
    
    Pytest->>Fixture: Teardown fixture
    Fixture->>ParallelManager: submit_teardown_task(scope, func)
    ParallelManager->>ThreadPool: Submit teardown task
    Fixture-->>Pytest: Return
    
    Pytest->>Barrier: teardown_barrier_function
    Barrier->>ParallelManager: wait_for_teardown_tasks(FUNCTION)
    ParallelManager->>ThreadPool: Wait for function teardowns
    
    Pytest->>ParallelManager: pytest_runtest_logreport hook
    ParallelManager->>ThreadPool: Terminate executor
    ParallelManager->>MonitorThread: Stop monitoring

Loading

[greptile-apps]

_{5 files reviewed, 5 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

The PyThreadState_SetAsyncExc call doesn't check the return value. According to Python docs, it returns the number of thread states modified. If it returns 0, the thread ID is invalid. If it returns >1, multiple thread states were affected (shouldn't happen but indicates a serious issue).

Suggested change

	def raise_async_exception(tid, exc_type):
	"""Injects an exception into the specified thread."""
	if not isinstance(tid, int):
	raise TypeError("Thread ID must be an integer")
	ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid),
	ctypes.py_object(exc_type))
	def raise_async_exception(tid, exc_type):
	"""Injects an exception into the specified thread."""
	if not isinstance(tid, int):
	raise TypeError("Thread ID must be an integer")
	ret = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid),
	ctypes.py_object(exc_type))
	if ret == 0:
	logging.warning("Thread ID %s not found when injecting exception", tid)
	elif ret > 1:
	# Revert if we affected multiple threads
	ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None)
	raise SystemError(f"PyThreadState_SetAsyncExc affected {ret} threads")

Prompt To Fix With AI

This is a comment left during a code review.
Path: tests/common/plugins/parallel_fixture/__init__.py
Line: 38:44

Comment:
The `PyThreadState_SetAsyncExc` call doesn't check the return value. According to Python docs, it returns the number of thread states modified. If it returns 0, the thread ID is invalid. If it returns >1, multiple thread states were affected (shouldn't happen but indicates a serious issue).

```suggestion
def raise_async_exception(tid, exc_type):
    """Injects an exception into the specified thread."""
    if not isinstance(tid, int):
        raise TypeError("Thread ID must be an integer")

    ret = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid),
                                                      ctypes.py_object(exc_type))
    if ret == 0:
        logging.warning("Thread ID %s not found when injecting exception", tid)
    elif ret > 1:
        # Revert if we affected multiple threads
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None)
        raise SystemError(f"PyThreadState_SetAsyncExc affected {ret} threads")
```

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

The exception re-raising logic uses deprecated sys.exc_info()[0] and loses the original exception chain. This makes debugging harder and may not properly preserve the exception context.

Suggested change

	raise sys.exc_info()[0](traceback.format_exc())
	raise

Prompt To Fix With AI

This is a comment left during a code review.
Path: tests/common/plugins/parallel_fixture/__init__.py
Line: 266:266

Comment:
The exception re-raising logic uses deprecated `sys.exc_info()[0]` and loses the original exception chain. This makes debugging harder and may not properly preserve the exception context.

```suggestion
                raise
```

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

Clearing the futures before termination check creates a race condition. If terminate() hasn't been called yet, clearing these sets will lose track of active tasks. The terminated flag should be checked before clearing.

Suggested change

	self.active_futures.clear()
	self.done_futures.clear()
	if not self.terminated:
	raise RuntimeError("Cannot reset a running parallel fixture manager.")
	self.active_futures.clear()
	self.done_futures.clear()

Prompt To Fix With AI

This is a comment left during a code review.
Path: tests/common/plugins/parallel_fixture/__init__.py
Line: 393:394

Comment:
Clearing the futures before termination check creates a race condition. If `terminate()` hasn't been called yet, clearing these sets will lose track of active tasks. The `terminated` flag should be checked before clearing.

```suggestion
        if not self.terminated:
            raise RuntimeError("Cannot reset a running parallel fixture manager.")
        
        self.active_futures.clear()
        self.done_futures.clear()
```

How can I resolve this? If you propose a fix, please make it concise.

[greptile-apps]

reset() doesn't set self.terminated = False after resetting the executor. This means once terminated, the manager can never be used again because terminate() will immediately return at line 335 due to the terminated check. Either reset should set terminated = False, or the design should clarify if reset is only for internal reuse.

Prompt To Fix With AI

This is a comment left during a code review.
Path: tests/common/plugins/parallel_fixture/__init__.py
Line: 388:398

Comment:
`reset()` doesn't set `self.terminated = False` after resetting the executor. This means once terminated, the manager can never be used again because `terminate()` will immediately return at line 335 due to the terminated check. Either reset should set `terminated = False`, or the design should clarify if reset is only for internal reuse.

How can I resolve this? If you propose a fix, please make it concise.