🔍 Code Extractor

Executes a function with a specified timeout using threading, raising a TimeoutException if the function exceeds the time limit.

Purpose

This function provides a thread-based timeout mechanism for executing any callable function. It's useful for preventing long-running operations from blocking indefinitely, particularly in scenarios where multiprocessing-based timeouts (like signal.alarm) are not available or suitable. The function captures both successful results and exceptions from the target function, re-raising them appropriately in the calling thread.

Source Code

def run_with_timeout(func, args=(), kwargs=None, timeout_duration=300):
    """
    Run a function with a timeout using threading (works in any thread)
    
    Args:
        func: Function to run
        args: Positional arguments for function
        kwargs: Keyword arguments for function
        timeout_duration: Timeout in seconds
        
    Returns:
        Result from function or raises TimeoutException
    """
    if kwargs is None:
        kwargs = {}
    
    result = [TimeoutException(f"Operation timed out after {timeout_duration} seconds")]
    
    def target():
        try:
            result[0] = func(*args, **kwargs)
        except Exception as e:
            result[0] = e
    
    thread = threading.Thread(target=target)
    thread.daemon = True
    thread.start()
    thread.join(timeout_duration)
    
    if thread.is_alive():
        # Thread is still running - timeout occurred
        logger.warning(f"Operation timed out after {timeout_duration}s")
        raise TimeoutException(f"Operation timed out after {timeout_duration} seconds")
    
    # Check if result is an exception
    if isinstance(result[0], Exception):
        raise result[0]
    
    return result[0]

Parameters

Parameter Details

func: The callable function to execute. Can be any Python function or method that needs timeout protection.

args: A tuple of positional arguments to pass to the function. Defaults to an empty tuple if not provided.

kwargs: A dictionary of keyword arguments to pass to the function. Defaults to None (converted to empty dict internally).

timeout_duration: Maximum time in seconds to allow the function to run before raising TimeoutException. Defaults to 300 seconds (5 minutes).

Return Value

Returns the result value from the executed function if it completes within the timeout period. If the function raises an exception, that exception is re-raised. If the timeout is exceeded, raises a TimeoutException with a descriptive message.

Dependencies

• threading
• logging

Required Imports

import threading
import logging

Usage Example

import threading
import logging

logger = logging.getLogger(__name__)

class TimeoutException(Exception):
    pass

def run_with_timeout(func, args=(), kwargs=None, timeout_duration=300):
    # ... function code ...
    pass

# Example 1: Simple function with timeout
def slow_operation(n):
    import time
    time.sleep(n)
    return f"Completed after {n} seconds"

try:
    result = run_with_timeout(slow_operation, args=(2,), timeout_duration=5)
    print(result)  # Output: Completed after 2 seconds
except TimeoutException as e:
    print(f"Timeout: {e}")

# Example 2: Function that times out
try:
    result = run_with_timeout(slow_operation, args=(10,), timeout_duration=3)
except TimeoutException as e:
    print(f"Timeout: {e}")  # Output: Timeout: Operation timed out after 3 seconds

# Example 3: Function with kwargs
def process_data(data, multiplier=1):
    return [x * multiplier for x in data]

result = run_with_timeout(process_data, args=([1, 2, 3],), kwargs={'multiplier': 2}, timeout_duration=10)
print(result)  # Output: [2, 4, 6]

Best Practices

• Ensure the 'TimeoutException' class is defined before using this function
• Be aware that daemon threads cannot be forcefully stopped; the timed-out function will continue running in the background until completion
• This approach is not suitable for CPU-bound operations that need hard termination; consider multiprocessing for those cases
• The function uses a list to store results to work around Python's closure limitations with mutable variables
• Logging requires a 'logger' object to be configured in the module scope
• Thread-based timeouts work in any thread context, unlike signal-based approaches which only work in the main thread
• Consider the implications of daemon threads: they will be terminated abruptly when the main program exits
• For I/O-bound operations, this is an effective timeout mechanism; for CPU-bound operations, multiprocessing may be more appropriate

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function with_lock 49.9% similar

  A decorator that wraps callback functions (both sync and async) to mark them for execution with a Bokeh document lock, allowing safe modification of Bokeh models.

  From: /tf/active/vicechatdev/patches/server.py

• class periodic 47.2% similar

  A threading class that executes a callback function periodically at specified intervals, with optional count limits and timeout controls.

  From: /tf/active/vicechatdev/patches/util.py

• class TimeoutException 46.4% similar

  A custom exception class that is raised when an operation exceeds its allocated time limit.

  From: /tf/active/vicechatdev/docchat/document_indexer.py

• function guard_execution 44.7% similar

  A decorator factory that prevents rapid repeated execution of a function by enforcing a cooldown period between calls.

  From: /tf/active/vicechatdev/CDocs/__init__.py

• function execute_transaction 41.7% similar

  Executes a database transaction function within a Neo4j session, handling connection management and error logging automatically.

  From: /tf/active/vicechatdev/CDocs/db/db_operations.py