Advanced Thread Pool Control

class ThreadpoolController:
    """
    Controller for all loaded supported thread pool libraries.
    
    Automatically discovers supported libraries when instantiated and provides
    methods for filtering, selecting, and controlling their thread settings.
    
    Attributes:
        lib_controllers: list[LibController] 
            List of individual library controller instances
    """
    
    def __init__(self):
        """
        Initialize controller and discover loaded libraries.
        
        Scans system for loaded shared libraries and creates controllers
        for supported thread pool libraries (BLAS and OpenMP implementations).
        """
        
    def info(self):
        """
        Get information about all controlled libraries.
        
        Returns:
            list[dict]: List of library info dicts (same format as threadpool_info())
        """
        
    def select(self, **kwargs):
        """
        Filter libraries by attributes.
        
        Args:
            **kwargs: Attribute filters where key is attribute name and value
                     is either a single value or list of acceptable values.
                     Common keys: user_api, internal_api, prefix, version, etc.
                     
        Returns:
            ThreadpoolController: New controller with filtered library subset
        """
        
    def limit(self, *, limits=None, user_api=None):
        """
        Create thread limiter for this controller's libraries.
        
        Args:
            limits: int | dict | str | None - Thread limit specification
            user_api: str | None - API type filter
            
        Returns:
            _ThreadpoolLimiter: Context manager for temporary limiting
        """
        
    def wrap(self, *, limits=None, user_api=None):
        """
        Create decorator for this controller's libraries.
        
        Args:
            limits: int | dict | str | None - Thread limit specification  
            user_api: str | None - API type filter
            
        Returns:
            Decorator function for thread limiting
        """
        
    def __len__(self):
        """
        Number of controlled libraries.
        
        Returns:
            int: Count of lib_controllers
        """

from threadpoolctl import ThreadpoolController

# Create controller
controller = ThreadpoolController()
print(f"Found {len(controller)} thread pool libraries")

# Select by API type
blas_controller = controller.select(user_api='blas')
openmp_controller = controller.select(user_api='openmp')

# Select by implementation
openblas_controller = controller.select(internal_api='openblas')
mkl_controller = controller.select(internal_api='mkl')

# Select by multiple criteria (OR logic)
specific_controller = controller.select(
    internal_api=['openblas', 'mkl'],
    prefix='libmkl_rt'
)

# Select by custom attributes (for libraries that have them)
openmp_blas = controller.select(threading_layer='openmp')

from threadpoolctl import ThreadpoolController

controller = ThreadpoolController()

# Limit only BLAS libraries to 1 thread
blas_controller = controller.select(user_api='blas')
with blas_controller.limit(limits=1):
    # Only BLAS libraries limited, OpenMP unchanged
    result = numpy_computation()

# Different limits for different implementations
openblas_controller = controller.select(internal_api='openblas')  
mkl_controller = controller.select(internal_api='mkl')

with openblas_controller.limit(limits=1):
    with mkl_controller.limit(limits=2):
        # OpenBLAS: 1 thread, MKL: 2 threads
        result = mixed_blas_computation()

# Complex filtering and limiting
high_thread_controller = controller.select(
    **{lib.prefix: lib.num_threads for lib in controller.lib_controllers 
       if lib.num_threads > 4}
)
with high_thread_controller.limit(limits=2):
    # Only libraries with >4 threads are limited to 2
    result = computation()

from threadpoolctl import ThreadpoolController

controller = ThreadpoolController()

# Access individual library controllers
for lib_controller in controller.lib_controllers:
    print(f"Library: {lib_controller.internal_api}")
    print(f"  Current threads: {lib_controller.num_threads}")
    print(f"  Prefix: {lib_controller.prefix}")
    print(f"  Version: {lib_controller.version}")
    
    # Get full info dict
    info = lib_controller.info()
    print(f"  Full info: {info}")

# FlexiBLAS-specific method available on FlexiBLAS controller instances
def switch_backend(self, backend):
    """
    Switch the FlexiBLAS backend.
    
    Args:
        backend: str - Backend name or path to shared library.
                      If not loaded, will be loaded first.
                      
    Raises:
        RuntimeError: If backend loading or switching fails
    """

from threadpoolctl import ThreadpoolController

controller = ThreadpoolController()
flexiblas_controller = controller.select(internal_api='flexiblas')

if flexiblas_controller:
    flex_lib = flexiblas_controller.lib_controllers[0]  # Get first FlexiBLAS instance
    
    print(f"Available backends: {flex_lib.available_backends}")
    print(f"Loaded backends: {flex_lib.loaded_backends}")  
    print(f"Current backend: {flex_lib.current_backend}")
    
    # Switch to different backend
    if 'OPENBLAS' in flex_lib.available_backends:
        flex_lib.switch_backend('OPENBLAS')
        print(f"Switched to: {flex_lib.current_backend}")

from threadpoolctl import ThreadpoolController

# Create specialized controllers for different use cases
def create_compute_controller():
    """Controller optimized for compute workloads."""
    controller = ThreadpoolController()
    return controller.select(user_api='blas')

def create_parallel_controller():
    """Controller for parallel processing workloads."""  
    controller = ThreadpoolController()
    return controller.select(user_api='openmp')

# Use in context
compute_controller = create_compute_controller()
parallel_controller = create_parallel_controller()

with compute_controller.limit(limits=1):
    with parallel_controller.limit(limits=4):
        # BLAS: 1 thread, OpenMP: 4 threads
        result = hybrid_computation()

from threadpoolctl import ThreadpoolController
import importlib

# Controller state can change as new libraries are loaded
initial_controller = ThreadpoolController()
print(f"Initial libraries: {len(initial_controller)}")

# Load library that brings in new thread pools
import some_blas_library

# Need new controller to detect newly loaded libraries  
updated_controller = ThreadpoolController()
print(f"After import: {len(updated_controller)}")

# Compare library sets
initial_libs = {lib.filepath for lib in initial_controller.lib_controllers}
updated_libs = {lib.filepath for lib in updated_controller.lib_controllers}
new_libs = updated_libs - initial_libs
print(f"New libraries: {list(new_libs)}")

from threadpoolctl import ThreadpoolController

controller = ThreadpoolController()

# Handle empty selections
blis_controller = controller.select(internal_api='blis')
if len(blis_controller) == 0:
    print("No BLIS libraries found")
else:
    with blis_controller.limit(limits=1):
        result = computation()

# Robust library access
openblas_controllers = controller.select(internal_api='openblas')
if openblas_controllers:
    for lib in openblas_controllers.lib_controllers:
        try:
            # Some old OpenBLAS versions may not support all operations
            threads = lib.num_threads
            lib.set_num_threads(1)
            lib.set_num_threads(threads)  # Restore
        except Exception as e:
            print(f"Library {lib.prefix} error: {e}")