Building basic Python sample profiler

Recently I applied for a vacancy and saw this test task:

Implement a Python tracing utility that measures the execution time of specific functions in a Python program. You should be able to:
• Enable profiling at an arbitrary moment providing a set of functions to trace
• Disable profiling and get the results

Bonus Features
• Allow toggling function list dynamically (e.g. without disabling it).
• Allow to get intermediate results while profiling is running

Requirements
• Implement the tracing without modifying the target functions directly.
• Provide a simple way to specify which functions to track (e.g., via a configuration file or a decorator).
• Ensure minimal overhead when tracing is disabled.
• Format and display results in a clear manner (e.g., as a table in the console).

I haven’t ever created a profiler before, so it was an interesting and challenging task. In this article, I want to share my approach to help beginners understand what a profiler is and how we can create a basic implementation of it.

If you’re interested in the source code, feel free to check out my GitHub repository.

What is a profiler?

A profiler is a tool used to analyze program execution, primarily to identify performance bottlenecks. It collects metrics like execution time and resource usage for parts of your code. Think of it as a diagnostic tool that helps you understand where your program spends its time.

There are two main types of profilers:

1. Instrumentation Profilers: These modify the target code by inserting measurement code. They provide detailed and accurate information about each function but add significant overhead.
   For example, in Python, you can create a profiler of this type with sys.setprofile function. It will handle events like function calls, returns, and exception raising. Basically on each function call you can save the function name and then build statistics with this data — it’s already a profiler :)
2. Sampling Profilers: These periodically sample the program’s state (call stack) to capture the currently running operation. They have less overhead but provide statistical approximations rather than exact measurements. Also, they have a chance to miss brief function calls between sampling intervals.

Since sampling profilers allow the target program to run at near full speed when they are disabled and don’t need to change the source code of the target application, I decided to implement this type of profiler despite the fact that they show less precise timing information.

Sampling Profiler Approach

As I said before, sampling profilers work by periodically checking what a program is doing (one time in δt seconds), usually by examining the call stack at regular intervals. Each sample represents a snapshot of the program’s state at that moment. From the call stack, we can get the name of the currently running function (or in which function is located currently executing line/operation).

For our profiler, we needed to:

1. Profile a running Python process without modifying its code
2. Selectively trace specific functions
3. Allow dynamic updating of traced functions
4. Provide clear results during and after profiling

To access a running process’s call stack, we need to use a ptrace-like system function. Since each OS has its own implementation, I used Docker to create a consistent environment (because I didn’t want and hadn’t time to get in cross-platform staff😅) and LLDB debugger with the cpython-lldb extension that provides utilities for tracing Python thread stack (basically debugger is an add-on to the functionality of the ‘general’ profiler — under the hood it’s also trace and manipulate stack of another process).
Yeah, I could use something more specific like python-ptrace library, but given limited time, my first thought seemed more rational.

So, what about steps 2 and 3? We need to dynamically operate a list of traced functions. It’ll be easier just to save all samples and then select functions that we need (also in this way we can output results anytime). We get a count of times a specific function was running. Just multiplying it with δt, we have an approximate run-time of it.
I found this explanation of how to get measurement error on Wikipedia, we can calculate it too:

“The actual amount of error is usually more than one sampling period. In fact, if a value is n times the sampling period, the expected error in it is the square root of n sampling periods.”

Enough words, let’s get to the realization.

Profiler Architecture

The profiler operates as three threads:

1. Main Thread: Handles the CLI interface and user commands
2. Sampling Thread: Connects to the target process via LLDB and collects samples
3. Watcher Thread: Monitors the sampling thread and handles when it stops

This design allows users to interact with the profiler while it’s collecting data.

Get Tuman Area’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

General file structure:

pythor_profiler_with_docker/
├── profiler/
│   ├── controller.py       # Process profiler commands (Main + Watcher threads)
│   └── sampler.py          # Sample target process, save snapshots (Sampler thread)
└── cli.py                  # Handle user commands (Main thread)

Command Line Interface

For pretty UI I use prompt_toolkit and argparse for parsing command line arguments.
I guess everything is easy here:

# cli.py

"""Command-line interface for launching and interacting with the profiler."""

from prompt_toolkit import PromptSession
from prompt_toolkit.patch_stdout import patch_stdout
from prompt_toolkit.completion import WordCompleter

import argparse
import sys

from profiler.controller import ProfilerController
from profiler.sampler import DEFAULT_SAMPLING_TIMEOUT
from profiler.ui import pprint, prompt_session


# All supported profiler commands
CLI_COMMANDS = [
    "start", "stop", "add", "remove", "results", "status", "exit"
]


# Create general parser that holds subparsers for each command
def build_args_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Profiler CLI")
    subparsers = parser.add_subparsers(dest="command")

    start_parser = subparsers.add_parser("start",
                                         help="Run profiler with certain "
                                         "arguments.")
    start_parser.add_argument("-p", "--pid", type=int, required=True,
                              help="PID of process you want to profile.")
    start_parser.add_argument("-f", "--func", nargs="+", required=True,
                              help="List of functions you want to profile "
                              "in selected Python process.")
    start_parser.add_argument("-t", "--timeout", type=float,
                              help="Custom sampling timeout. Default is "
                              f"{DEFAULT_SAMPLING_TIMEOUT} s")

    stop_parser = subparsers.add_parser("stop",
                                        help="Stop profiler.")

    add_parser = subparsers.add_parser("add",
                                       help="Add functions to profiling.")
    add_parser.add_argument("-f", "--func", nargs="+", required=True,
                            help="List of functions you want to profile "
                            "in selected Python process.")

    remove_parser = subparsers.add_parser("remove",
                                          help="Remove functions from "
                                          "profiling.")
    remove_parser.add_argument("-f", "--func", nargs="+", required=True,
                               help="List of functions you want to remove "
                               "from profiling in selected Python process.")

    results_parser = subparsers.add_parser("results",
                                           help="Get intermediate profiling "
                                           "results.")

    status_parser = subparsers.add_parser("status",
                                          help="Get profiler status.")

    exit_parser = subparsers.add_parser("exit",
                                        help="Exit Profiler CLI.")

    return parser


# Launch the profiler main thread that handles and processes user commands
def run_cli_loop() -> None:
    """Handle and process profiler commands in infinite interaction loop."""
    profiler = ProfilerController()
    parser = build_args_parser()

    command_completer = WordCompleter(CLI_COMMANDS, ignore_case=True)
    session = PromptSession(completer=command_completer)

    pprint("<ansicyan>\n🚀 Welcome to <b>Sample Profiler CLI</b>! "
           "Type <b>help</b> or <b>exit</b> to quit.\n</ansicyan>")

    # Process arguments on profiler startup
    if len(sys.argv) > 1:
        args = parser.parse_args()
        profiler.process_command(args)

    # Obtain commands in an infinite loop 
    while True:
        try:
            with patch_stdout():
                user_input = prompt_session(
                    "<arrow>➤ </arrow><prompt>Profiler</prompt>"
                    "<arrow> > </arrow>",
                    session
                ).strip()

            if not user_input:
                continue

            args = parser.parse_args(user_input.split())
            profiler.process_command(args)
        except SystemExit as e:
            if e.code != 0:
                pprint("<error>⚠️ Invalid command.</error>")
            else:
                pprint("\n<info>👋 Exiting Profiler CLI. Bye!</info>")
                return
        except KeyboardInterrupt:
            pprint("<error>\n🛑 Interrupted. Use 'exit' or Ctrl+D to quit."
                   "</error>")
        except EOFError:
            pprint("\n<info>👋 Exiting Profiler CLI. Bye!</info>")
            return
        except Exception as e:
            pprint(f"<error>💥 Error: {e}</error>")


if __name__ == "__main__":
    run_cli_loop()

Profiler Controller

This file contains an interface (launched in the main thread) responsible for command logic. It will run Sampler and Watcher threads on start command.
In my opinion, the most important moment here is to use threading.Lock for getting samples from the Sampler thread (this way we prevent race conditions):

# profiler/controller.py

"""Main profiler interface that manages tracing logic and handles commands
from the CLI.
"""

import argparse
from math import sqrt
from tabulate import tabulate
import threading
from typing import List, Set, Optional
import psutil
import sys

from .sampler import Sampler, DEFAULT_SAMPLING_TIMEOUT


class ProfilerController:
    def __init__(self):
        self.running = False
        self.functions_to_profile: Set[str] = set()
        self.pid_to_trace: int = 0
        self.sampler_instance: Sampler = None
        self.sampling_thread: threading.Thread = None
        self.sampling_timeout: float = DEFAULT_SAMPLING_TIMEOUT
        self.controller_lock = threading.Lock()

    def process_command(self, args: argparse.Namespace) -> None:
        if args.command == "start":
            self.start(pid=args.pid,
                       functions_to_trace=args.func,
                       sampling_timeout=args.timeout)

        elif args.command == "stop":
            self.stop()

        elif args.command == "add":
            self.add_functions_to_profile(args.func)

        elif args.command == "remove":
            self.remove_functions_from_profile(args.func)

        elif args.command == "results":
            self.print_results()

        elif args.command == "status":
            self.print_status()

        elif args.command == "exit":
            if self.running:
                self.stop()
            sys.exit(0)

        else:
            print("Invalid command.")

    def start(self, pid: int, functions_to_trace: List[str],
              sampling_timeout: Optional[float] = None) -> None:
        """
        Launch a profiler for the specified process and set of functions with
        selected sampling timeout.

        Args:
            pid (int): PID of process you want to profile.
            functions_to_trace (List[str]): List of functions you want to
                profile in selected Python process.
            sampling_timeout (float): Sampling timeout in seconds. Profiler
                will take a sample of selected Python process one time in
                *sampling_timeout* seconds.
        """
        if self.running:
            print("Profiler is already running!")
            return

        if not psutil.pid_exists(pid):
            print(f"Process PID {pid} not found.")
            sys.exit(0)

        self.running = True
        self.pid_to_trace = pid

        if sampling_timeout is not None:
            self.sampling_timeout = sampling_timeout

        self.add_functions_to_profile(functions_to_trace)
        print(
            f"Starting profiling process {pid} "
            f"for functions: {self.functions_to_profile}"
        )

        # Start collecting samples in a separate thread
        self.sampler_instance = Sampler(
            self.pid_to_trace,
            self.sampling_timeout
        )
        self.sampling_thread = threading.Thread(
            target=self.sampler_instance.start_sample_loop
        )
        self.sampling_thread.daemon = True
        self.sampling_thread.start()

        watcher = threading.Thread(target=self._watch_sampler)
        watcher.daemon = True
        watcher.start()

    def _watch_sampler(self):
        """Wait until sampler is stopped and then stop profiler."""
        self.sampling_thread.join()

        self.stop()

    def add_functions_to_profile(self, func_to_add: List[str]) -> None:
        self.functions_to_profile.update(func_to_add)
        print("Sampling functions:", self.functions_to_profile)

    def remove_functions_from_profile(self, func_to_remove: List[str]) -> None:
        self.functions_to_profile.difference_update(func_to_remove)
        print("Sampling functions:", self.functions_to_profile)

    def print_results(self) -> None:
        """Get profiling results and print them in console."""
        if not self.sampler_instance:
            print("No sampling data available. Start profiler first.")
            return

        function_counts = {}

        function_names = list(self.functions_to_profile)
        function_names.sort()
        function_execution_times = ["No Data Available"] * len(function_names)

        for entry in self.sampler_instance.get_samples():
            fn = entry["function"]
            if fn in self.functions_to_profile:
                function_counts[fn] = function_counts.get(fn, 0) + 1

        for fn, count in function_counts.items():
            total_time = count * self.sampling_timeout
            error = sqrt(count) * self.sampling_timeout
            time_with_error = f"{round(total_time, 4)} ± {round(error, 4)}"

            function_execution_times[
                function_names.index(fn)
            ] = time_with_error

        results = {"Function Name": function_names,
                   "Approximate execution time (s)": function_execution_times}
        print(tabulate(results, headers="keys", tablefmt="rounded_outline"))

    def print_status(self) -> None:
        """
        Print the following profiler info:
        - Is profiler running.
        - Is PID to trace exists.
        - *self.sampling_thread*
        - *self.sampling_timeout*
        - *self.functions_to_profile*
        """
        print("===Profiler Status===")
        print("Profiler is running." if self.running else
              "Profiler is stopped.")
        print(f"PID {self.pid_to_trace} exists:",
              psutil.pid_exists(self.pid_to_trace))
        print("Sampling thread:", self.sampling_thread)
        print("Sampling timeout:", self.sampling_timeout)
        print("Sampling functions:", self.functions_to_profile)
        print("=====================")

    def stop(self) -> None:
        """Stop profiler and print profiling results."""
        with self.controller_lock:
            if not self.running:
                return
            self.running = False
            print("Profiler stopped.")

        self.print_results()

After the target process is ended, you will get a pretty-formatted result:

Profiler stopped.
╭───────────────────┬──────────────────────────────────╮
│ Function Name     │ Approximate execution time (s)   │
├───────────────────┼──────────────────────────────────┤
│ bar               │ No Data Available                │
│ disk_io           │ 1.5 ± 0.1732                     │
│ foo               │ No Data Available                │
│ heavy_computation │ 0.9 ± 0.1342                     │
│ network_request   │ 1.52 ± 0.1744                    │
╰───────────────────┴──────────────────────────────────╯

Sampler

First of all, we create a subprocess that operates LLDB instance. With it, we can write commands to its stdin and read its stdout (collect call stack). After that, we launch an infinite sampling loop.

At each sampling interval:

1. The profiler pauses the target process with process signal SIGINT command (aka Ctrl+C)
2. Captures the current Python stack frame with py-bt command from cpython-lldb extension
3. Store the sample in internal storage. Now it’s just a list, but you can modify it to any database if you need
   It’s also important to use threading.Lock for writing data here
4. Resumes the process with process continue command

The sampling continues at regular intervals (default 20 ms, can be changed), building a statistical profile of function execution times.
After the target process is ended, we just detach and exit.

One annoying moment here: we should read stdout of LLDB process after we enter process continue command, not after py-bt. It happens because of buffering. Nothing we can do with this in this implementation.
¯\_(ツ)_/¯

# profiler/sampler.py

"""Sampler interface that handles sampling of a running Python process using
LLDB to collect call stack traces.
"""

import time
from typing import List, Dict, Union
import psutil
import re
import subprocess
import threading


DEFAULT_SAMPLING_TIMEOUT: float = 0.02  # In seconds


class Sampler:
    def __init__(self, pid_to_trace: int, sampling_timeout: float):
        self.pid_to_trace = pid_to_trace
        self.sampling_timeout = sampling_timeout
        self.is_running = False
        self.samples: List[Dict[float, str]] = []
        self.samples_lock = threading.Lock()
        self.lldb_instance = None

    def start_sample_loop(self) -> None:
        """
        Start main sampling loop. Getting a sample every
        *self.sampling_timeout* seconds.

        Loop looks like this:
        1. A permanent LLDB session is started.
        2. while running:
            - Check if tracing process exists.
            - Suspend thread via LLDB.
            - Capture current running Python function.
            - Resume thread.
            - Sleep for *self.sampling_timeout* milliseconds
        3. Exit LLDB
        """
        self.is_running = True
        self.start_debugger_session()

        while self.is_running:
            if not psutil.pid_exists(self.pid_to_trace):
                break

            sample = self.get_name_of_running_function()
            with self.samples_lock:
                self.samples.append(sample)

            time.sleep(self.sampling_timeout)

        self.lldb_instance.stdin.write("detach\n")
        self.lldb_instance.stdin.write("exit\n")
        self.lldb_instance.stdin.flush()

        return

    def start_debugger_session(self) -> None:
        # Run LLDB and attach it to the target process.
        # We use subprocess.Popen to create a permanent interactive session.
        self.lldb_instance = subprocess.Popen(
            ["lldb", "-p", str(self.pid_to_trace), "--local-lldbinit"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        # Waiting for the LLDB to be ready to obtain new command.
        while True:
            line = self.lldb_instance.stdout.readline()
            if "(lldb)" in line:
                break

    # Function for getting one sample
    def get_name_of_running_function(self) -> Dict[str, Union[float, str]]:
        """
        Capture Python process stack frame and get last function from it (this
        function is currently running).

        1. Interrupt the execution of the target process with
            "process signal SIGINT" command (aka. Ctrl-C).
        2. Execute the `py-bt` command to retrieve the Python stack.
        3. Continue the process execution with the `process continue` command.

        Returns:
            (Dict[str, Union[float, str]]): A dictionary with:
                "timestamp": timestamp of sampling (float)
                "function": name of the currently running function (str)
        """
        timestamp = time.time()

        try:
            # Stop program via LLDB
            self.lldb_instance.stdin.write("process signal SIGINT\n")
            self.lldb_instance.stdin.flush()
            while True:
                line = self.lldb_instance.stdout.readline()
                if "(lldb)" in line:
                    break

            # Get call trace
            self.lldb_instance.stdin.write("py-bt\n")
            self.lldb_instance.stdin.flush()
            output_lines = []
            while True:
                line = self.lldb_instance.stdout.readline()
                if "(lldb)" in line:
                    break

            self.lldb_instance.stdin.write("process continue\n")
            self.lldb_instance.stdin.flush()
            while True:
                line = self.lldb_instance.stdout.readline()
                if "(lldb)" in line:
                    break
                output_lines.append(line)
            decoded = "".join(output_lines)
            function_name = self.parse_python_stack(decoded)
        except Exception as e:
            print(f"Error capturing sample with lldb: {e}")
            function_name = "Unknown"

        return {"timestamp": timestamp, "function": function_name}

    def parse_python_stack(self, py_bt_output: str) -> str:
        """
        Parses the LLDB output of the py-bt command and returns the
        function name, in which the execution is currently in progress.

        Args:
            py_bt_output (str): Output of py-bt program.

        Returns:
            str: Name of Python function that currently running.
        """
        pattern = re.compile(r'^\s*File "([^"]+)", line \d+, in (.+)$')
        matches = []

        for line in py_bt_output.splitlines():
            m = pattern.search(line)
            if m:
                func = m.group(2).strip()
                matches.append(func)

        if matches:
            return matches[-1]
        return "No function detected."

    def get_samples(self) -> List[Dict[float, str]]:
        with self.samples_lock:
            return self.samples.copy()

This is the main part of project that covers all its functions. Of course, there are other files, but I find them not so important.

Conclusion

So, it was fun… All task requirements were satisfied and I successfully got to the technical interview …and successfully failed it :)
Whatever
Hope it was helpful, until next time 🙃