### Start and Stop Event Loop Hang Inspection

Source: https://context7.com/quantlane/libs/llms.txt

Starts a background thread to monitor event loop progress. If a hang is detected, it writes stack traces to a specified directory. Ensure `stop_wait` is called with the returned dumper object for clean shutdown.

```python
import asyncio
import os
import tempfile
import aiodebug.hang_inspection

async def main():
    output_dir = tempfile.mkdtemp()
    print(f"Stack traces will be written to: {output_dir}")

    # Start hang detection with a 0.25 s (default) hang threshold
    dumper = aiodebug.hang_inspection.start(output_dir, interval=0.25)

    # --- Simulate a hang: block the event loop for 1 second ---
    import time
    time.sleep(1.0)  # This will trigger 3 stack trace files

    # Allow the event loop to recover briefly
    await asyncio.sleep(0.5)

    # Clean shutdown
    await aiodebug.hang_inspection.stop_wait(dumper)

    # Inspect output
    files = sorted(os.listdir(output_dir))
    print(f"Generated {len(files)} stack trace file(s):")
    for f in files:
        print(f"  {f}")
    # Example output:
    #   stacktrace-20240104-094154.197418-0.txt
    #   stacktrace-20240104-094155.206574-1.txt
    #   stacktrace-20240104-094156.211781-2.txt

    if files:
        with open(os.path.join(output_dir, files[0])) as fh:
            print("\n--- First stack trace (excerpt) ---")
            print(fh.read()[:500])

asyncio.run(main())
```

--------------------------------

### aiodebug.hang_inspection.start / stop_wait

Source: https://context7.com/quantlane/libs/llms.txt

Starts a background thread to monitor event loop progress and dumps stack traces if a hang is detected. The `stop_wait` function is used for clean shutdown.

```APIDOC
## `aiodebug.hang_inspection.start` / `stop_wait` — Dump stack traces on event loop hangs

Starts a background daemon thread (`TraceDumper`) that monitors whether the event loop is still making progress. If the loop fails to update its heartbeat within `interval` seconds, the dumper writes three consecutive stack-trace files (1 second apart) to `stack_output_dir`, capturing all thread frames at the time of the hang. Returns a `(TraceDumper, asyncio.Task)` tuple that must be passed to `stop_wait` for clean shutdown.

```python
import asyncio
import os
import tempfile
import aiodebug.hang_inspection

async def main():
    output_dir = tempfile.mkdtemp()
    print(f"Stack traces will be written to: {output_dir}")

    # Start hang detection with a 0.25 s (default) hang threshold
    dumper = aiodebug.hang_inspection.start(output_dir, interval=0.25)

    # --- Simulate a hang: block the event loop for 1 second ---
    import time
    time.sleep(1.0)  # This will trigger 3 stack trace files

    # Allow the event loop to recover briefly
    await asyncio.sleep(0.5)

    # Clean shutdown
    await aiodebug.hang_inspection.stop_wait(dumper)

    # Inspect output
    files = sorted(os.listdir(output_dir))
    print(f"Generated {len(files)} stack trace file(s):")
    for f in files:
        print(f"  {f}")
    # Example output:
    #   stacktrace-20240104-094154.197418-0.txt
    #   stacktrace-20240104-094155.206574-1.txt
    #   stacktrace-20240104-094156.211781-2.txt

    if files:
        with open(os.path.join(output_dir, files[0])) as fh:
            print("\n--- First stack trace (excerpt) ---")
            print(fh.read()[:500])

asyncio.run(main())
```
```

--------------------------------

### Start Hang Inspection in aiodebug

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Initiate hang inspection to dump stack traces of all threads when the event loop hangs for too long. Specify an output directory and an optional interval for detection.

```python
import aiodebug.hang_inspection

dumper = aiodebug.hang_inspection.start('/path/to/output/directory', interval = 0.25)  # 0.25 is the default
```

--------------------------------

### Set Up Time Dilated Loop for Testing

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Replace the default asyncio event loop with TimeDilatedLoop for testing purposes. This allows controlling the passage of time within the event loop.

```python
import aiodebug.testing.time_dilated_loop

loop = aiodebug.testing.time_dilated_loop.TimeDilatedLoop()
asyncio.set_event_loop(loop)
```

--------------------------------

### Enable logging for slow event loop callbacks

Source: https://context7.com/quantlane/libs/llms.txt

Patches asyncio to log warnings for callbacks exceeding a specified duration. Supports default logging or custom handlers for structured output.

```python
import asyncio
import json
import logging
import aiodebug.log_slow_callbacks

logging.basicConfig(level=logging.WARNING)

# --- Option 1: Default logging (WARNING level) ---
# Logs: "Executing <Task ...> took 0.082 seconds"
aiodebug.log_slow_callbacks.enable(0.05)  # threshold: 50 ms

# --- Option 2: Custom structured handler ---
logger = logging.getLogger("myapp")

def on_slow(task_name: str, duration: float) -> None:
    logger.warning(
        "Slow callback detected",
        extra={"task_name": task_name, "duration_seconds": round(duration, 4)},
    )

aiodebug.log_slow_callbacks.enable(0.05, on_slow_callback=on_slow)

# --- Demo: a callback that intentionally blocks ---
async def slow_task():
    import time
    time.sleep(0.1)  # blocks the event loop for 100 ms — will trigger the warning

async def main():
    await slow_task()

asyncio.run(main())
# Expected output (default handler):
# WARNING:aiodebug.log_slow_callbacks:Executing <coroutine slow_task> took 0.100 seconds
```

--------------------------------

### aiodebug.monitor_loop_lag.enable

Source: https://context7.com/quantlane/libs/llms.txt

Tracks event loop lag by measuring the difference between scheduled and actual wake-up times, sending the lag in milliseconds to a StatsD client.

```APIDOC
## aiodebug.monitor_loop_lag.enable — Track event loop lag via StatsD

Creates an asyncio monitoring task that wakes every `interval` seconds, measures how much later than scheduled it actually woke (i.e., event loop lag), and sends the lag in milliseconds to a StatsD client under the metric `aiodebug.monitor_loop_lag`. High or growing lag indicates callbacks are blocking the loop.

### Parameters
- **statsd_client** (object) - Required - An object with a `timing(metric: str, value: float)` method, conforming to `StatsdClientProtocol`.
- **interval** (float) - Optional - The interval in seconds at which to measure and report loop lag. Defaults to 0.25 seconds.
- **loop** (asyncio.AbstractEventLoop) - Optional - The event loop to monitor. Defaults to the current running event loop.

### Request Example
```python
import asyncio
import aiodebug.monitor_loop_lag

# Minimal StatsD client interface required by aiodebug (StatsdClientProtocol):
# def timing(self, metric: str, value: float) -> None

class MockStatsd:
    def timing(self, metric: str, value: float) -> None:
        print(f"[StatsD] {metric} = {value:.3f} ms")

# --- Basic usage with default 0.25 s interval ---
async def main():
    statsd = MockStatsd()
    aiodebug.monitor_loop_lag.enable(statsd)           # sends lag every 0.25 s
    await asyncio.sleep(1)                             # let a few measurements accumulate

asyncio.run(main())

# --- Custom interval and explicit loop ---
async def main_custom():
    loop = asyncio.get_event_loop()
    statsd = MockStatsd()
    aiodebug.monitor_loop_lag.enable(statsd, interval=0.5, loop=loop)
    await asyncio.sleep(2)

asyncio.run(main_custom())
```

### Response Example
```
[StatsD] aiodebug.monitor_loop_lag = 0.021 ms
[StatsD] aiodebug.monitor_loop_lag = 0.018 ms
[StatsD] aiodebug.monitor_loop_lag = 0.025 ms
[StatsD] aiodebug.monitor_loop_lag = 0.019 ms
```
```

--------------------------------

### Track event loop lag with StatsD

Source: https://context7.com/quantlane/libs/llms.txt

Monitors asyncio event loop lag and sends metrics to a StatsD client. Useful for detecting blocking callbacks. Requires a StatsdClientProtocol compatible client.

```python
import asyncio
import aiodebug.monitor_loop_lag

# Minimal StatsD client interface required by aiodebug (StatsdClientProtocol):
# def timing(self, metric: str, value: float) -> None

class MockStatsd:
    def timing(self, metric: str, value: float) -> None:
        print(f"[StatsD] {metric} = {value:.3f} ms")

# --- Basic usage with default 0.25 s interval ---
async def main():
    statsd = MockStatsd()
    aiodebug.monitor_loop_lag.enable(statsd)           # sends lag every 0.25 s
    await asyncio.sleep(1)                             # let a few measurements accumulate

asyncio.run(main())
# Expected output (approximate):
# [StatsD] aiodebug.monitor_loop_lag = 0.021 ms
# [StatsD] aiodebug.monitor_loop_lag = 0.018 ms
# [StatsD] aiodebug.monitor_loop_lag = 0.025 ms
# [StatsD] aiodebug.monitor_loop_lag = 0.019 ms

# --- Custom interval and explicit loop ---
async def main_custom():
    loop = asyncio.get_event_loop()
    statsd = MockStatsd()
    aiodebug.monitor_loop_lag.enable(statsd, interval=0.5, loop=loop)
    await asyncio.sleep(2)

asyncio.run(main_custom())
```

--------------------------------

### Enable Event Loop Lag Monitoring with StatsD

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Monitor and report event loop lags to StatsD using aiodebug. This helps in understanding and diagnosing delays in scheduled asynchronous tasks.

```python
import aiodebug.monitor_loop_lag

aiodebug.monitor_loop_lag.enable(statsd_client)
```

--------------------------------

### Enable Slow Callback Logging in aiodebug

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Enable logging for callbacks that block the event loop for longer than a specified duration. This helps identify performance bottlenecks without enabling full debug mode.

```python
import aiodebug.log_slow_callbacks

aiodebug.log_slow_callbacks.enable(0.05)
```

--------------------------------

### Custom Callback for Slow Callbacks in aiodebug

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Configure aiodebug to use a custom callback function when a slow callback is detected. This allows for custom logging, such as structured JSON logging, instead of default WARNING logs.

```python
import aiodebug.log_slow_callbacks

aiodebug.log_slow_callbacks.enable(
	0.05,
	on_slow_callback = lambda task_name, duration: json_logger.warning(
		'Task blocked async loop for too long',
		extra = {'task_name': task_name, 'duration': duration}
	)
)
```

--------------------------------

### aiodebug.log_slow_callbacks.enable

Source: https://context7.com/quantlane/libs/llms.txt

Detects and logs event loop callbacks that exceed a specified duration threshold. It can use default logging or a custom handler for structured output.

```APIDOC
## aiodebug.log_slow_callbacks.enable — Detect and log slow event loop callbacks

Patches `asyncio.events.Handle._run` globally so that any callback exceeding the given `slow_duration` threshold (in seconds) triggers a warning. By default it logs via `logwood` (if installed) or the standard `logging` module, but a custom `on_slow_callback` handler can be supplied for structured logging, metrics, or alerting.

### Parameters
- **slow_duration** (float) - Required - The duration threshold in seconds to consider a callback slow.
- **on_slow_callback** (callable) - Optional - A function to call when a slow callback is detected. It receives `task_name` (str) and `duration` (float) as arguments.

### Request Example
```python
import asyncio
import json
import logging
import aiodebug.log_slow_callbacks

logging.basicConfig(level=logging.WARNING)

# --- Option 1: Default logging (WARNING level) ---
# Logs: "Executing <Task ...> took 0.082 seconds"
aiodebug.log_slow_callbacks.enable(0.05)  # threshold: 50 ms

# --- Option 2: Custom structured handler ---
logger = logging.getLogger("myapp")

def on_slow(task_name: str, duration: float) -> None:
    logger.warning(
        "Slow callback detected",
        extra={"task_name": task_name, "duration_seconds": round(duration, 4)},
    )

aiodebug.log_slow_callbacks.enable(0.05, on_slow_callback=on_slow)

# --- Demo: a callback that intentionally blocks ---
async def slow_task():
    import time
    time.sleep(0.1)  # blocks the event loop for 100 ms — will trigger the warning

async def main():
    await slow_task()

asyncio.run(main())
```

### Response Example
```
WARNING:aiodebug.log_slow_callbacks:Executing <coroutine slow_task> took 0.100 seconds
```
```

--------------------------------

### aiodebug.testing.time_dilated_loop.DilationMeter

Source: https://context7.com/quantlane/libs/llms.txt

A helper class to compute the current time dilation factor by comparing external subjective timestamps against real monotonic time. Useful for driving `TimeDilatedLoop.time_dilation` from external sources.

```APIDOC
## `aiodebug.testing.time_dilated_loop.DilationMeter` — Compute time dilation from external subjective timestamps

A helper class that calculates the current time dilation factor by comparing externally-provided "subjective" timestamps against real monotonic time. Useful when driving `TimeDilatedLoop.time_dilation` from an external clock source (e.g., a simulation engine or replay feed).

```python
import asyncio
import time
import aiodebug.testing.time_dilated_loop

meter = aiodebug.testing.time_dilated_loop.DilationMeter()
loop = aiodebug.testing.time_dilated_loop.TimeDilatedLoop()
asyncio.set_event_loop(loop)

async def adjust_from_feed():
    # Simulate receiving subjective timestamps from an external source
    subjective_clock = [1.0, 2.0, 3.0, 5.0, 7.0]  # seconds in "sim time"
    for t in subjective_clock:
        meter.set_subjective_time(t)
        loop.time_dilation = meter.dilation
        print(f"Subjective time={t:.1f}s  →  dilation={meter.dilation:.2f}×")
        await asyncio.sleep(0.1)  # real 0.1 s between feed updates

loop.run_until_complete(adjust_from_feed())
loop.close()
# Example output:
# Subjective time=1.0s  →  dilation=1.00×
# Subjective time=2.0s  →  dilation=9.87×   (1 subjective second in ~0.1 real seconds)
# Subjective time=3.0s  →  dilation=9.91×
# Subjective time=5.0s  →  dilation=19.74×  (2 subjective seconds in ~0.1 real seconds)
# Subjective time=7.0s  →  dilation=19.80×
```
```

--------------------------------

### Control Event Loop Time with TimeDilatedLoop

Source: https://context7.com/quantlane/libs/llms.txt

Subclasses the asyncio event loop to decouple its internal clock from real time using a `time_dilation` multiplier. Useful for speeding up or slowing down `asyncio.sleep` calls during testing.

```python
import asyncio
import time
import aiodebug.testing.time_dilated_loop

async def run_scenario():
    loop = asyncio.get_event_loop()

    # --- 3× speed: asyncio.sleep(1) completes in ~0.333 real seconds ---
    loop.time_dilation = 3.0
    t0 = time.monotonic()
    await asyncio.sleep(1)
    elapsed = time.monotonic() - t0
    print(f"[3× dilation] asyncio.sleep(1) took {elapsed:.2f}s real time")  # ~0.33s

    # --- 0.5× speed: asyncio.sleep(1) completes in ~2 real seconds ---
    loop.time_dilation = 0.5
    t0 = time.monotonic()
    await asyncio.sleep(1)
    elapsed = time.monotonic() - t0
    print(f"[0.5× dilation] asyncio.sleep(1) took {elapsed:.2f}s real time")  # ~2.00s

    # --- Restore normal speed ---
    loop.time_dilation = 1.0

loop = aiodebug.testing.time_dilated_loop.TimeDilatedLoop()
asyncio.set_event_loop(loop)
loop.run_until_complete(run_scenario())
loop.close()
```

--------------------------------

### aiodebug.testing.time_dilated_loop.TimeDilatedLoop

Source: https://context7.com/quantlane/libs/llms.txt

A subclass of asyncio's event loop that allows decoupling the loop's internal clock from wall-clock time using a `time_dilation` multiplier. Useful for speeding up or slowing down time-sensitive tests.

```APIDOC
## `aiodebug.testing.time_dilated_loop.TimeDilatedLoop` — Speed up or slow down event loop time

A drop-in `asyncio` event loop subclass that decouples the loop's internal clock from wall-clock time via a `time_dilation` multiplier. Setting `time_dilation > 1` makes the loop's `asyncio.sleep` calls resolve faster than real time; setting it `< 1` slows them down. This is invaluable for testing time-sensitive async code without long-running test suites.

```python
import asyncio
import time
import aiodebug.testing.time_dilated_loop

async def run_scenario():
    loop = asyncio.get_event_loop()

    # --- 3× speed: asyncio.sleep(1) completes in ~0.333 real seconds ---
    loop.time_dilation = 3.0
    t0 = time.monotonic()
    await asyncio.sleep(1)
    elapsed = time.monotonic() - t0
    print(f"[3× dilation] asyncio.sleep(1) took {elapsed:.2f}s real time")  # ~0.33s

    # --- 0.5× speed: asyncio.sleep(1) completes in ~2 real seconds ---
    loop.time_dilation = 0.5
    t0 = time.monotonic()
    await asyncio.sleep(1)
    elapsed = time.monotonic() - t0
    print(f"[0.5× dilation] asyncio.sleep(1) took {elapsed:.2f}s real time")  # ~2.00s

    # --- Restore normal speed ---
    loop.time_dilation = 1.0

loop = aiodebug.testing.time_dilated_loop.TimeDilatedLoop()
asyncio.set_event_loop(loop)
loop.run_until_complete(run_scenario())
loop.close()
# Output:
# [3× dilation] asyncio.sleep(1) took 0.33s real time
# [0.5× dilation] asyncio.sleep(1) took 2.00s real time
```
```

--------------------------------

### Stop and Wait for Hang Inspection in aiodebug

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Stop the hang inspection process and wait for any ongoing dump operations to complete. This should be called when the hang inspection is no longer needed.

```python
await aiodebug.hang_inspection.stop_wait(dumper)
```

--------------------------------

### Compute Time Dilation with DilationMeter

Source: https://context7.com/quantlane/libs/llms.txt

Calculates the current time dilation factor by comparing external subjective timestamps with real monotonic time. Use this to dynamically adjust `TimeDilatedLoop.time_dilation` based on an external clock source.

```python
import asyncio
import time
import aiodebug.testing.time_dilated_loop

meter = aiodebug.testing.time_dilated_loop.DilationMeter()
loop = aiodebug.testing.time_dilated_loop.TimeDilatedLoop()
asyncio.set_event_loop(loop)

async def adjust_from_feed():
    # Simulate receiving subjective timestamps from an external source
    subjective_clock = [1.0, 2.0, 3.0, 5.0, 7.0]  # seconds in "sim time"
    for t in subjective_clock:
        meter.set_subjective_time(t)
        loop.time_dilation = meter.dilation
        print(f"Subjective time={t:.1f}s  →  dilation={meter.dilation:.2f}×")
        await asyncio.sleep(0.1)  # real 0.1 s between feed updates

loop.run_until_complete(adjust_from_feed())
loop.close()
```

--------------------------------

### Adjust Time Dilation in aiodebug

Source: https://gitlab.com/quantlane/libs/aiodebug/-/blob/main/README.rst

Modify the time_dilation attribute of the TimeDilatedLoop to speed up or slow down time. A value greater than 1 speeds up time, while a value less than 1 slows it down.

```python
loop.time_dilation = 3
await asyncio.sleep(1)  # Takes 0.333s of real time

loop.time_dilation = 0.1
await asyncio.sleep(1)  # Takes 10s of real time
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.