🔍 Code Extractor

A guard mechanism that prevents recursive or repeated function calls by tracking active executions and enforcing cooldown periods between calls.

Purpose

ExecutionGuard provides a thread-safe mechanism to prevent functions from being called recursively or too frequently. It tracks which functions are currently executing and enforces a configurable cooldown period between successive calls. This is useful for preventing race conditions, debouncing rapid function calls, and avoiding stack overflow from recursive calls. Common use cases include UI event handlers, API rate limiting, and preventing duplicate background task execution.

Source Code

class ExecutionGuard:
    """Prevent recursive or repeated function calls."""
    
    def __init__(self):
        """Initialize execution guard."""
        self._active_calls = {}
        self._last_calls = {}
    
    def guard(self, func_name, cooldown_ms=500):
        """
        Check if a function should execute.
        
        Parameters:
        - func_name: Unique name for the function
        - cooldown_ms: Minimum time between executions in milliseconds
        
        Returns:
        - Boolean indicating if function should proceed
        """
        import time
        now = time.time() * 1000  # Convert to milliseconds
        
        # Check if function is already running
        if self._active_calls.get(func_name, False):
            logger.debug(f"Skipping repeated call to {func_name} (already active)")
            return False
        
        # Check cooldown period
        last_call = self._last_calls.get(func_name, 0)
        if (now - last_call) < cooldown_ms:
            logger.debug(f"Skipping repeated call to {func_name} (cooldown: {now - last_call}ms < {cooldown_ms}ms)")
            return False
        
        # Record function activity
        self._active_calls[func_name] = True
        self._last_calls[func_name] = now
        return True
    
    def release(self, func_name):
        """Mark a function as no longer active."""
        self._active_calls[func_name] = False

Parameters

Parameter Details

__init__: No parameters required. Initializes internal state tracking dictionaries for active calls and last call timestamps.

guard.func_name: A unique string identifier for the function being guarded. This should be consistent across calls to the same logical function. Typically the function's name or a descriptive identifier.

guard.cooldown_ms: Minimum time in milliseconds that must elapse between successive executions of the function. Default is 500ms. Set to 0 to only prevent recursive calls without cooldown. Must be a non-negative number.

release.func_name: The unique string identifier of the function to mark as no longer active. Must match the func_name used in the corresponding guard() call.

Return Value

The guard() method returns a boolean: True if the function should proceed with execution (not currently active and cooldown period has elapsed), False if the function should be skipped (either already running or called too soon after last execution). The release() method returns None. The __init__ constructor returns an ExecutionGuard instance.

Class Interface

Methods

Attributes

Dependencies

• time
• logging

Required Imports

import time
import logging

Usage Example

# Basic usage pattern
import time
import logging

logger = logging.getLogger(__name__)

# Create an execution guard instance
guard = ExecutionGuard()

def my_function():
    func_name = 'my_function'
    
    # Check if function should execute
    if not guard.guard(func_name, cooldown_ms=1000):
        return  # Skip execution
    
    try:
        # Perform function logic
        print('Executing function')
        time.sleep(0.5)
    finally:
        # Always release the guard
        guard.release(func_name)

# First call executes
my_function()  # Prints: Executing function

# Immediate second call is blocked (cooldown)
my_function()  # Skipped

# After cooldown period, executes again
time.sleep(1.1)
my_function()  # Prints: Executing function

# Decorator pattern example
def guarded_function(guard_instance, cooldown_ms=500):
    def decorator(func):
        def wrapper(*args, **kwargs):
            func_name = func.__name__
            if not guard_instance.guard(func_name, cooldown_ms):
                return None
            try:
                return func(*args, **kwargs)
            finally:
                guard_instance.release(func_name)
        return wrapper
    return decorator

guard = ExecutionGuard()

@guarded_function(guard, cooldown_ms=1000)
def protected_function():
    print('This runs with protection')

protected_function()

Best Practices

• Always call release() in a finally block to ensure the guard is released even if the function raises an exception
• Use consistent func_name strings across guard() and release() calls for the same function
• Consider using a decorator pattern to automatically handle guard() and release() calls
• Set appropriate cooldown_ms values based on your use case: 0 for recursion prevention only, higher values for rate limiting
• Create one ExecutionGuard instance per logical group of functions that need coordinated protection
• The guard is not thread-safe by default; for multi-threaded applications, consider adding locks or using thread-local storage
• func_name should be unique enough to avoid collisions between different functions
• Monitor debug logs to understand when and why function calls are being blocked
• For long-running functions, the cooldown timer starts when guard() is called, not when release() is called
• If a function crashes without calling release(), it will remain marked as active indefinitely; implement timeout mechanisms for critical applications

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function guard_execution 76.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

• class ScriptExecutor 44.3% similar

  A sandboxed Python script executor that safely runs user-provided Python code with timeout controls, security restrictions, and isolated execution environments for data analysis tasks.

  From: /tf/active/vicechatdev/vice_ai/script_executor.py

• class periodic 43.0% 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 ScriptExecutor_v1 41.8% similar

  A sandboxed Python script executor that safely runs user-provided Python code with timeout controls, security restrictions, and isolated execution environments for data analysis tasks.

  From: /tf/active/vicechatdev/full_smartstat/script_executor.py

• function cleanup_old_tasks 41.7% similar

  Removes tasks from the active_tasks dictionary that are older than 1 hour (3600 seconds) based on their creation timestamp.

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