Profiler Module

Core profiler implementation with multi-track support and runtime control.

Profiler Class

stichotrope.profiler.Profiler

Main profiler class with multi-track support and runtime enable/disable.

Example

profiler = Profiler("MyApp")

@profiler.track(0, "process_data") def process_data(data): return transform(data)

def complex_function(): with profiler.block(1, "database_query"): result = query_database() return result

results = profiler.get_results()

Source code in stichotrope/profiler.py

class Profiler:
    """
    Main profiler class with multi-track support and runtime enable/disable.

    Example:
        profiler = Profiler("MyApp")

        @profiler.track(0, "process_data")
        def process_data(data):
            return transform(data)

        def complex_function():
            with profiler.block(1, "database_query"):
                result = query_database()
            return result

        results = profiler.get_results()
    """

    def __init__(self, name: str = "Profiler"):
        """
        Initialize a new profiler instance.

        Args:
            name: Human-readable name for this profiler
        """
        global _NEXT_PROFILER_ID

        # Register profiler with lock protection
        with _REGISTRY_LOCK:
            self._profiler_id = _NEXT_PROFILER_ID
            _NEXT_PROFILER_ID += 1
            _PROFILER_REGISTRY[self._profiler_id] = self

        self._name = name

        # Thread-local storage for per-thread profiling data
        self._thread_local = threading.local()

        # Global lock protects _all_thread_data registry
        self._global_lock = threading.RLock()

        # Registry of all thread-local data: thread_id -> thread_local
        self._all_thread_data: dict[int, Any] = {}

        # Instance enable/disable flag
        self._started = True  # Profiler starts enabled by default

    def start(self) -> None:
        """Start profiling (resume data collection)."""
        self._started = True

    def stop(self) -> None:
        """Stop profiling (pause data collection)."""
        self._started = False

    def is_started(self) -> bool:
        """Check if profiler is started."""
        return self._started

    def _get_thread_data(self) -> Any:
        """
        Get or initialize thread-local data for the current thread.

        Uses hasattr pattern to avoid AttributeError on first access.
        Registers thread in global registry on first access.
        Caches thread_data reference in thread-local storage for fast access.

        Returns:
            Thread-local data object with tracks, track_enabled, next_block_idx
        """
        if not hasattr(self._thread_local, 'data'):
            # First access from this thread - initialize thread-local storage
            thread_id = threading.get_ident()
            thread_name = threading.current_thread().name

            # Register thread in global registry (LOCK REQUIRED)
            # Create a data object to hold this thread's profiling data
            with self._global_lock:
                if thread_id not in self._all_thread_data:
                    # Create a simple object to hold thread data
                    class ThreadData:
                        pass

                    thread_data = ThreadData()
                    thread_data.tracks = {}
                    thread_data.track_enabled = {}
                    thread_data.next_block_idx = {}
                    thread_data.thread_id = thread_id
                    thread_data.thread_name = thread_name

                    self._all_thread_data[thread_id] = thread_data

                else:
                    # Thread was already registered (e.g., after clear())
                    thread_data = self._all_thread_data[thread_id]

            # Cache thread_data reference in thread-local storage for fast access
            # This eliminates dict lookup on every _get_thread_data() call
            self._thread_local.data = thread_data

        # Return cached thread_data reference (fast - no dict lookup)
        return self._thread_local.data

    def set_track_enabled(self, track_idx: int, enabled: bool) -> None:
        """
        Enable or disable a specific track.

        Args:
            track_idx: Track index
            enabled: True to enable, False to disable
        """
        thread_data = self._get_thread_data()
        thread_data.track_enabled[track_idx] = enabled

    def is_track_enabled(self, track_idx: int) -> bool:
        """
        Check if a specific track is enabled.

        Args:
            track_idx: Track index

        Returns:
            True if track is enabled (default: True)
        """
        thread_data = self._get_thread_data()
        return thread_data.track_enabled.get(track_idx, True)

    def set_track_name(self, track_idx: int, name: str) -> None:
        """
        Set a human-readable name for a track.

        Args:
            track_idx: Track index
            name: Track name
        """
        thread_data = self._get_thread_data()
        track = self._get_or_create_track(thread_data, track_idx)
        track.track_name = name

    def _get_or_create_track(self, thread_data: Any, track_idx: int) -> ProfileTrack:
        """
        Get or create a track by index in thread-local storage.

        Args:
            thread_data: Thread-local data object
            track_idx: Track index

        Returns:
            ProfileTrack instance
        """
        if track_idx not in thread_data.tracks:
            thread_data.tracks[track_idx] = ProfileTrack(track_idx=track_idx)
            thread_data.next_block_idx[track_idx] = 0
        return thread_data.tracks[track_idx]

    def _register_block(self, thread_data: Any, track_idx: int, name: str, file: str, line: int) -> int:
        """
        Register a new profiling block and return its index.

        Args:
            thread_data: Thread-local data object
            track_idx: Track index
            name: Block name
            file: Source file
            line: Line number

        Returns:
            Block index within the track
        """
        track = self._get_or_create_track(thread_data, track_idx)
        block_idx = thread_data.next_block_idx[track_idx]
        thread_data.next_block_idx[track_idx] += 1

        track.add_block(block_idx, name, file, line)
        return block_idx

    def _record_block_time(self, track_idx: int, block_idx: int, elapsed_ns: int) -> None:
        """
        Record execution time for a block.

        HOT PATH - NO LOCKS. Uses thread-local data only.

        Args:
            track_idx: Track index
            block_idx: Block index
            elapsed_ns: Elapsed time in nanoseconds
        """
        thread_data = self._get_thread_data()
        track = thread_data.tracks.get(track_idx)
        if track is None:
            return

        block = track.get_block(block_idx)
        if block is not None:
            block.record_time(elapsed_ns)

    def _aggregate_results(self) -> ProfilerResults:
        """
        Aggregate profiling data from all threads.

        Uses sequential merge algorithm (GIL-friendly).
        Acquires _global_lock to safely iterate _all_thread_data.

        Returns:
            ProfilerResults with aggregated data from all threads
        """
        aggregated_tracks: dict[int, ProfileTrack] = {}

        # Acquire lock to safely iterate all thread data
        with self._global_lock:
            # Iterate all threads and merge their data
            for thread_id, thread_local in self._all_thread_data.items():
                # Merge each track from this thread
                for track_idx, source_track in thread_local.tracks.items():
                    # Get or create aggregated track
                    if track_idx not in aggregated_tracks:
                        aggregated_tracks[track_idx] = ProfileTrack(
                            track_idx=track_idx,
                            track_name=source_track.track_name
                        )

                    aggregated_track = aggregated_tracks[track_idx]

                    # Merge all blocks from source track
                    for block_idx, source_block in source_track.blocks.items():
                        self._merge_block(aggregated_track, block_idx, source_block)

        results = ProfilerResults(profiler_name=self._name)
        results.tracks = aggregated_tracks
        return results

    def _merge_block(self, aggregated_track: ProfileTrack, block_idx: int, source_block: Any) -> None:
        """
        Merge a source block into the aggregated track.

        Args:
            aggregated_track: Target track for aggregation
            block_idx: Block index
            source_block: Source ProfileBlock to merge
        """
        if block_idx not in aggregated_track.blocks:
            # First occurrence of this block - add it
            aggregated_track.add_block(
                block_idx,
                source_block.name,
                source_block.file,
                source_block.line
            )
            target_block = aggregated_track.blocks[block_idx]
            target_block.hit_count = source_block.hit_count
            target_block.total_time_ns = source_block.total_time_ns
            target_block.min_time_ns = source_block.min_time_ns
            target_block.max_time_ns = source_block.max_time_ns
        else:
            # Block already exists - merge statistics
            target_block = aggregated_track.blocks[block_idx]
            target_block.hit_count += source_block.hit_count
            target_block.total_time_ns += source_block.total_time_ns
            target_block.min_time_ns = min(target_block.min_time_ns, source_block.min_time_ns)
            target_block.max_time_ns = max(target_block.max_time_ns, source_block.max_time_ns)

    def get_results(self) -> ProfilerResults:
        """
        Get profiling results aggregated from all threads.

        Returns:
            ProfilerResults containing all tracks and blocks from all threads
        """
        return self._aggregate_results()

    def clear(self) -> None:
        """Clear all profiling data from all threads."""
        # Clear global thread data registry
        with self._global_lock:
            self._all_thread_data.clear()

        # Invalidate cached thread_data reference in current thread
        # This ensures the thread re-initializes on next access
        if hasattr(self._thread_local, 'data'):
            delattr(self._thread_local, 'data')

    def track(self, track_idx: int, name: Optional[str] = None) -> Callable:
        """
        Decorator for profiling functions.

        Example:
            @profiler.track(0, "process_data")
            def process_data(data):
                return transform(data)

            # Auto-detect function name
            @profiler.track(0)
            def compute():
                return result

        Args:
            track_idx: Track index for this function
            name: Optional name (defaults to function.__name__)

        Returns:
            Decorator function
        """
        # Level 1: Global enable/disable (zero overhead when disabled)
        if not _PROFILER_ENABLED:
            return lambda func: func  # Identity decorator - ZERO overhead

        def decorator(func: Callable) -> Callable:
            # Use function name if not provided
            block_name = name if name is not None else func.__name__

            # Get call-site information
            frame = inspect.currentframe()
            if frame and frame.f_back:
                file = frame.f_back.f_code.co_filename
                line = frame.f_back.f_lineno
            else:
                file = "<unknown>"
                line = 0

            # Check call-site cache with lock protection
            # We only cache the block_idx here, not register the block yet
            # Block registration happens in each thread when wrapper is first executed
            cache_key = (track_idx, file, line, block_name)
            block_idx_container = [None]  # Use list to allow modification in wrapper

            with _GLOBAL_CACHE_LOCK:
                if cache_key in _CALL_SITE_CACHE:
                    profiler_id, block_idx = _CALL_SITE_CACHE[cache_key]
                    block_idx_container[0] = block_idx
                else:
                    # Allocate a block_idx without registering the block yet
                    # We'll use a counter to allocate unique block indices
                    # For now, use the cache size as the block_idx
                    block_idx = len(_CALL_SITE_CACHE)
                    _CALL_SITE_CACHE[cache_key] = (self._profiler_id, block_idx)
                    block_idx_container[0] = block_idx

            # Store block_idx in function attribute for fast access
            @functools.wraps(func)
            def wrapper(*args: Any, **kwargs: Any) -> Any:
                # Level 2: Per-track enable/disable (fast guard)
                if not self.is_track_enabled(track_idx):
                    return func(*args, **kwargs)

                # Level 3: Instance start/stop
                if not self._started:
                    return func(*args, **kwargs)

                # Get the cached block_idx
                block_idx = block_idx_container[0]

                # Ensure block exists in current thread's storage
                thread_data = self._get_thread_data()
                track = thread_data.tracks.get(track_idx)
                if track is None or block_idx not in track.blocks:
                    # Block not yet registered in this thread - register it
                    track = self._get_or_create_track(thread_data, track_idx)
                    if block_idx not in track.blocks:
                        track.add_block(block_idx, block_name, file, line)

                # Profile the function
                start = get_time_ns()
                try:
                    result = func(*args, **kwargs)
                    return result
                finally:
                    end = get_time_ns()
                    elapsed = end - start
                    self._record_block_time(track_idx, block_idx, elapsed)

            return wrapper

        return decorator

    @contextmanager
    def block(self, track_idx: int, name: str) -> Generator[None, None, None]:
        """
        Context manager for profiling code blocks.

        Example:
            with profiler.block(1, "database_query"):
                result = query_database()

        Args:
            track_idx: Track index for this block
            name: Block name (required)

        Yields:
            None
        """
        # Level 1: Global enable/disable
        if not _PROFILER_ENABLED:
            yield
            return

        # Level 2: Per-track enable/disable
        if not self.is_track_enabled(track_idx):
            yield
            return

        # Level 3: Instance start/stop
        if not self._started:
            yield
            return

        # Get call-site information
        frame = inspect.currentframe()
        if frame and frame.f_back:
            file = frame.f_back.f_code.co_filename
            line = frame.f_back.f_lineno
        else:
            file = "<unknown>"
            line = 0

        # Check call-site cache with lock protection
        # We only cache the block_idx here, not register the block yet
        cache_key = (track_idx, file, line, name)
        with _GLOBAL_CACHE_LOCK:
            if cache_key in _CALL_SITE_CACHE:
                profiler_id, block_idx = _CALL_SITE_CACHE[cache_key]
            else:
                # Allocate a block_idx without registering the block yet
                block_idx = len(_CALL_SITE_CACHE)
                _CALL_SITE_CACHE[cache_key] = (self._profiler_id, block_idx)

        # Ensure block exists in current thread's storage
        thread_data = self._get_thread_data()
        track = thread_data.tracks.get(track_idx)
        if track is None or block_idx not in track.blocks:
            # Block not yet registered in this thread - register it
            track = self._get_or_create_track(thread_data, track_idx)
            if block_idx not in track.blocks:
                track.add_block(block_idx, name, file, line)

        # Profile the block
        start = get_time_ns()
        try:
            yield
        finally:
            end = get_time_ns()
            elapsed = end - start
            self._record_block_time(track_idx, block_idx, elapsed)

    def export_csv(self, filename: str) -> None:
        """
        Export profiling results to CSV file.

        Args:
            filename: Output CSV filename
        """
        from stichotrope.export import export_csv

        results = self.get_results()
        with open(filename, "w", newline="") as f:
            export_csv(results, f)

    def export_json(self, filename: str, indent: int = 2) -> None:
        """
        Export profiling results to JSON file.

        Args:
            filename: Output JSON filename
            indent: JSON indentation level
        """
        from stichotrope.export import export_json

        results = self.get_results()
        with open(filename, "w") as f:
            export_json(results, f, indent=indent)

    def print_results(self) -> None:
        """Print profiling results to console in a formatted table."""
        from stichotrope.export import print_results

        results = self.get_results()
        print_results(results)

    def __repr__(self) -> str:
        # Count unique tracks across all threads
        track_indices = set()
        thread_count = 0

        with self._global_lock:
            thread_count = len(self._all_thread_data)
            for thread_data in self._all_thread_data.values():
                track_indices.update(thread_data.tracks.keys())

        return (
            f"Profiler(name={self._name!r}, tracks={len(track_indices)}, "
            f"threads={thread_count}, started={self._started})"
        )

Functions

__init__(name='Profiler')

Initialize a new profiler instance.

Parameters:

Name	Type	Description	Default
name	str	Human-readable name for this profiler	'Profiler'

Source code in stichotrope/profiler.py

def __init__(self, name: str = "Profiler"):
    """
    Initialize a new profiler instance.

    Args:
        name: Human-readable name for this profiler
    """
    global _NEXT_PROFILER_ID

    # Register profiler with lock protection
    with _REGISTRY_LOCK:
        self._profiler_id = _NEXT_PROFILER_ID
        _NEXT_PROFILER_ID += 1
        _PROFILER_REGISTRY[self._profiler_id] = self

    self._name = name

    # Thread-local storage for per-thread profiling data
    self._thread_local = threading.local()

    # Global lock protects _all_thread_data registry
    self._global_lock = threading.RLock()

    # Registry of all thread-local data: thread_id -> thread_local
    self._all_thread_data: dict[int, Any] = {}

    # Instance enable/disable flag
    self._started = True  # Profiler starts enabled by default

block(track_idx, name)

Context manager for profiling code blocks.

Example

with profiler.block(1, "database_query"): result = query_database()

Parameters:

Name	Type	Description	Default
track_idx	int	Track index for this block	required
name	str	Block name (required)	required

Yields:

Type	Description
None	None

Source code in stichotrope/profiler.py

@contextmanager
def block(self, track_idx: int, name: str) -> Generator[None, None, None]:
    """
    Context manager for profiling code blocks.

    Example:
        with profiler.block(1, "database_query"):
            result = query_database()

    Args:
        track_idx: Track index for this block
        name: Block name (required)

    Yields:
        None
    """
    # Level 1: Global enable/disable
    if not _PROFILER_ENABLED:
        yield
        return

    # Level 2: Per-track enable/disable
    if not self.is_track_enabled(track_idx):
        yield
        return

    # Level 3: Instance start/stop
    if not self._started:
        yield
        return

    # Get call-site information
    frame = inspect.currentframe()
    if frame and frame.f_back:
        file = frame.f_back.f_code.co_filename
        line = frame.f_back.f_lineno
    else:
        file = "<unknown>"
        line = 0

    # Check call-site cache with lock protection
    # We only cache the block_idx here, not register the block yet
    cache_key = (track_idx, file, line, name)
    with _GLOBAL_CACHE_LOCK:
        if cache_key in _CALL_SITE_CACHE:
            profiler_id, block_idx = _CALL_SITE_CACHE[cache_key]
        else:
            # Allocate a block_idx without registering the block yet
            block_idx = len(_CALL_SITE_CACHE)
            _CALL_SITE_CACHE[cache_key] = (self._profiler_id, block_idx)

    # Ensure block exists in current thread's storage
    thread_data = self._get_thread_data()
    track = thread_data.tracks.get(track_idx)
    if track is None or block_idx not in track.blocks:
        # Block not yet registered in this thread - register it
        track = self._get_or_create_track(thread_data, track_idx)
        if block_idx not in track.blocks:
            track.add_block(block_idx, name, file, line)

    # Profile the block
    start = get_time_ns()
    try:
        yield
    finally:
        end = get_time_ns()
        elapsed = end - start
        self._record_block_time(track_idx, block_idx, elapsed)

clear()

Clear all profiling data from all threads.

Source code in stichotrope/profiler.py

def clear(self) -> None:
    """Clear all profiling data from all threads."""
    # Clear global thread data registry
    with self._global_lock:
        self._all_thread_data.clear()

    # Invalidate cached thread_data reference in current thread
    # This ensures the thread re-initializes on next access
    if hasattr(self._thread_local, 'data'):
        delattr(self._thread_local, 'data')

export_csv(filename)

Export profiling results to CSV file.

Parameters:

Name	Type	Description	Default
filename	str	Output CSV filename	required

Source code in stichotrope/profiler.py

def export_csv(self, filename: str) -> None:
    """
    Export profiling results to CSV file.

    Args:
        filename: Output CSV filename
    """
    from stichotrope.export import export_csv

    results = self.get_results()
    with open(filename, "w", newline="") as f:
        export_csv(results, f)

export_json(filename, indent=2)

Export profiling results to JSON file.

Parameters:

Name	Type	Description	Default
filename	str	Output JSON filename	required
indent	int	JSON indentation level	2

Source code in stichotrope/profiler.py

def export_json(self, filename: str, indent: int = 2) -> None:
    """
    Export profiling results to JSON file.

    Args:
        filename: Output JSON filename
        indent: JSON indentation level
    """
    from stichotrope.export import export_json

    results = self.get_results()
    with open(filename, "w") as f:
        export_json(results, f, indent=indent)

get_results()

Get profiling results aggregated from all threads.

Returns:

Type	Description
ProfilerResults	ProfilerResults containing all tracks and blocks from all threads

Source code in stichotrope/profiler.py

def get_results(self) -> ProfilerResults:
    """
    Get profiling results aggregated from all threads.

    Returns:
        ProfilerResults containing all tracks and blocks from all threads
    """
    return self._aggregate_results()

is_started()

Check if profiler is started.

Source code in stichotrope/profiler.py

def is_started(self) -> bool:
    """Check if profiler is started."""
    return self._started

is_track_enabled(track_idx)

Check if a specific track is enabled.

Parameters:

Name	Type	Description	Default
track_idx	int	Track index	required

Returns:

Type	Description
bool	True if track is enabled (default: True)

Source code in stichotrope/profiler.py

def is_track_enabled(self, track_idx: int) -> bool:
    """
    Check if a specific track is enabled.

    Args:
        track_idx: Track index

    Returns:
        True if track is enabled (default: True)
    """
    thread_data = self._get_thread_data()
    return thread_data.track_enabled.get(track_idx, True)

print_results()

Print profiling results to console in a formatted table.

Source code in stichotrope/profiler.py

def print_results(self) -> None:
    """Print profiling results to console in a formatted table."""
    from stichotrope.export import print_results

    results = self.get_results()
    print_results(results)

set_track_enabled(track_idx, enabled)

Enable or disable a specific track.

Parameters:

Name	Type	Description	Default
track_idx	int	Track index	required
enabled	bool	True to enable, False to disable	required

Source code in stichotrope/profiler.py

def set_track_enabled(self, track_idx: int, enabled: bool) -> None:
    """
    Enable or disable a specific track.

    Args:
        track_idx: Track index
        enabled: True to enable, False to disable
    """
    thread_data = self._get_thread_data()
    thread_data.track_enabled[track_idx] = enabled

set_track_name(track_idx, name)

Set a human-readable name for a track.

Parameters:

Name	Type	Description	Default
track_idx	int	Track index	required
name	str	Track name	required

Source code in stichotrope/profiler.py

def set_track_name(self, track_idx: int, name: str) -> None:
    """
    Set a human-readable name for a track.

    Args:
        track_idx: Track index
        name: Track name
    """
    thread_data = self._get_thread_data()
    track = self._get_or_create_track(thread_data, track_idx)
    track.track_name = name

start()

Start profiling (resume data collection).

Source code in stichotrope/profiler.py

def start(self) -> None:
    """Start profiling (resume data collection)."""
    self._started = True

stop()

Stop profiling (pause data collection).

Source code in stichotrope/profiler.py

def stop(self) -> None:
    """Stop profiling (pause data collection)."""
    self._started = False

track(track_idx, name=None)

Decorator for profiling functions.

Example

@profiler.track(0, "process_data") def process_data(data): return transform(data)

Auto-detect function name

@profiler.track(0) def compute(): return result

Parameters:

Name	Type	Description	Default
track_idx	int	Track index for this function	required
name	Optional[str]	Optional name (defaults to function.name)	None

Returns:

Type	Description
Callable	Decorator function

Source code in stichotrope/profiler.py

def track(self, track_idx: int, name: Optional[str] = None) -> Callable:
    """
    Decorator for profiling functions.

    Example:
        @profiler.track(0, "process_data")
        def process_data(data):
            return transform(data)

        # Auto-detect function name
        @profiler.track(0)
        def compute():
            return result

    Args:
        track_idx: Track index for this function
        name: Optional name (defaults to function.__name__)

    Returns:
        Decorator function
    """
    # Level 1: Global enable/disable (zero overhead when disabled)
    if not _PROFILER_ENABLED:
        return lambda func: func  # Identity decorator - ZERO overhead

    def decorator(func: Callable) -> Callable:
        # Use function name if not provided
        block_name = name if name is not None else func.__name__

        # Get call-site information
        frame = inspect.currentframe()
        if frame and frame.f_back:
            file = frame.f_back.f_code.co_filename
            line = frame.f_back.f_lineno
        else:
            file = "<unknown>"
            line = 0

        # Check call-site cache with lock protection
        # We only cache the block_idx here, not register the block yet
        # Block registration happens in each thread when wrapper is first executed
        cache_key = (track_idx, file, line, block_name)
        block_idx_container = [None]  # Use list to allow modification in wrapper

        with _GLOBAL_CACHE_LOCK:
            if cache_key in _CALL_SITE_CACHE:
                profiler_id, block_idx = _CALL_SITE_CACHE[cache_key]
                block_idx_container[0] = block_idx
            else:
                # Allocate a block_idx without registering the block yet
                # We'll use a counter to allocate unique block indices
                # For now, use the cache size as the block_idx
                block_idx = len(_CALL_SITE_CACHE)
                _CALL_SITE_CACHE[cache_key] = (self._profiler_id, block_idx)
                block_idx_container[0] = block_idx

        # Store block_idx in function attribute for fast access
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            # Level 2: Per-track enable/disable (fast guard)
            if not self.is_track_enabled(track_idx):
                return func(*args, **kwargs)

            # Level 3: Instance start/stop
            if not self._started:
                return func(*args, **kwargs)

            # Get the cached block_idx
            block_idx = block_idx_container[0]

            # Ensure block exists in current thread's storage
            thread_data = self._get_thread_data()
            track = thread_data.tracks.get(track_idx)
            if track is None or block_idx not in track.blocks:
                # Block not yet registered in this thread - register it
                track = self._get_or_create_track(thread_data, track_idx)
                if block_idx not in track.blocks:
                    track.add_block(block_idx, block_name, file, line)

            # Profile the function
            start = get_time_ns()
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                end = get_time_ns()
                elapsed = end - start
                self._record_block_time(track_idx, block_idx, elapsed)

        return wrapper

    return decorator

Global Functions

set_global_enabled

stichotrope.profiler.set_global_enabled(enabled)

Enable or disable profiling globally across all profiler instances.

When disabled, decorators return identity functions (zero overhead).

Parameters:

Name	Type	Description	Default
enabled	bool	True to enable profiling, False to disable	required

Source code in stichotrope/profiler.py

def set_global_enabled(enabled: bool) -> None:
    """
    Enable or disable profiling globally across all profiler instances.

    When disabled, decorators return identity functions (zero overhead).

    Args:
        enabled: True to enable profiling, False to disable
    """
    global _PROFILER_ENABLED
    _PROFILER_ENABLED = enabled

is_global_enabled

stichotrope.profiler.is_global_enabled()

Check if profiling is globally enabled.

Source code in stichotrope/profiler.py

def is_global_enabled() -> bool:
    """Check if profiling is globally enabled."""
    return _PROFILER_ENABLED

Usage Examples

Basic Profiler Usage

from stichotrope import Profiler

# Create a profiler instance
profiler = Profiler("MyApplication")

# Profile a function with decorator
@profiler.track(0, "process_data")
def process_data(data):
    return transform(data)

# Profile a code block with context manager
def complex_function():
    with profiler.block(1, "database_query"):
        result = query_database()
    return result

# Get results
results = profiler.get_results()
profiler.print_results()

Runtime Control

from stichotrope import Profiler, set_global_enabled

profiler = Profiler("MyApp")

# Per-profiler control
profiler.stop()  # Pause profiling
profiler.start()  # Resume profiling

# Per-track control
profiler.set_track_enabled(0, False)  # Disable track 0
profiler.set_track_enabled(0, True)   # Re-enable track 0

# Global control (affects all profilers)
set_global_enabled(False)  # Disable all profiling (zero overhead)
set_global_enabled(True)   # Re-enable profiling

Multi-Track Organization

from stichotrope import Profiler

profiler = Profiler("WebServer")

# Track 0: Request handling
@profiler.track(0, "handle_request")
def handle_request(request):
    return process_request(request)

# Track 1: Database operations
@profiler.track(1, "db_query")
def query_database(query):
    return execute_query(query)

# Track 2: Cache operations
@profiler.track(2, "cache_lookup")
def check_cache(key):
    return cache.get(key)

# Results are organized by track
results = profiler.get_results()
for track in results.tracks:
    print(f"Track {track.track_idx}: {len(track.blocks)} blocks")

Nested Profiling

from stichotrope import Profiler

profiler = Profiler("DataPipeline")

def process_pipeline(data):
    # Outer block
    with profiler.block(0, "full_pipeline"):
        # Inner blocks
        with profiler.block(1, "load_data"):
            loaded = load(data)

        with profiler.block(1, "transform_data"):
            transformed = transform(loaded)

        with profiler.block(1, "save_data"):
            save(transformed)

    return transformed

Multi-Threaded Usage

from stichotrope import Profiler
import threading

profiler = Profiler("MultiThreadApp")

@profiler.track(0, "worker_task")
def worker_task(task_id):
    # Profiling in thread
    with profiler.block(1, "task_processing"):
        process(task_id)

# Start multiple threads
threads = [
    threading.Thread(target=worker_task, args=(i,))
    for i in range(5)
]

for t in threads:
    t.start()
for t in threads:
    t.join()

# Retrieve results from all threads
all_results = profiler.get_all_thread_data()
for thread_id, thread_results in all_results.items():
    print(f"Thread {thread_id}: {thread_results}")

Implementation Details

Call-Site Caching

The profiler uses call-site caching to minimize overhead. Each unique call site (file, line number, function name) is cached, so subsequent calls to the same profiled function have minimal overhead.

Thread Safety

The Profiler is thread-safe by design in v0.2.0+. Each thread maintains its own profiling data independently, with a thread-safe aggregation mechanism for retrieving cross-thread results.

Per-thread operations (lock-free): - track() decorator on functions - block() context manager - get_results() for current thread's data

Cross-thread operations (synchronized): - get_all_thread_data() for aggregating results across threads

For multi-threaded applications, use get_all_thread_data() to retrieve and aggregate profiling results from all threads.

Performance Characteristics

• Overhead when enabled: ≤1% for blocks over 1ms. The raw overhead is typically 4µs so profiling anything within the same order of magnitude will result in significant timing distortions.
• Overhead when disabled: Zero overhead (decorators return identity functions)

For detailed performance analysis and benchmarking methodology, see Performance Documentation.

See Also

• Export Module - Exporting and displaying results
• Types Module - Data structure definitions
• Getting Started - Usage guide