ControlPort Monitoring

ControlPort is GNU Radio’s binary protocol for advanced runtime control. It provides richer functionality than XML-RPC: native type support, performance counters, knob metadata, and PMT message injection.

ControlPort vs XML-RPC

Feature	XML-RPC	ControlPort
Protocol	HTTP (text)	Thrift (binary)
Performance	Slower	Faster
Type Support	Basic Python	Complex, vectors, PMT
Metadata	None	Units, min/max, hints
Perf Counters	No	Yes
Message Ports	No	Yes

Use ControlPort when:

• You need performance profiling (throughput, timing, buffer utilization)
• Working with complex data types (complex numbers, vectors)
• You want parameter metadata (units, valid ranges)
• You need to send PMT messages to blocks

Enable ControlPort

Launch with ControlPort enabled:

enable_runtime_mode()

launch_flowgraph(
    flowgraph_path="/tmp/fm_receiver.py",
    name="fm-profiled",
    enable_controlport=True,       # Enable ControlPort
    controlport_port=9090,         # Port (default 9090)
    enable_perf_counters=True      # Enable performance counters
)

Note

The base runtime image includes GNU Radio’s Thrift support. No additional configuration is needed.

Connect via ControlPort

By Container Name

connect_to_container_controlport(name="fm-profiled")

By Host/Port

connect_controlport(host="127.0.0.1", port=9090)

Working with Knobs

ControlPort exposes block parameters as “knobs” with the naming pattern: block_alias::parameter

Get Knobs

# Get all knobs
get_knobs(pattern="")
# Returns: [KnobModel(name="osmosdr_source_0::freq_corr", value=0.0, ...), ...]

# Filter by regex
get_knobs(pattern=".*::freq.*")
# Returns: [KnobModel(name="osmosdr_source_0::freq_corr", ...), ...]

# Get knobs for a specific block
get_knobs(pattern="osmosdr_source_0::.*")

Set Knobs

# Set multiple knobs atomically
set_knobs({
    "sig_source_0::frequency": 1500.0,
    "sig_source_0::amplitude": 0.8
})

Get Knob Properties

get_knob_properties(names=["sig_source_0::frequency"])
# Returns: [KnobPropertiesModel(
#   name="sig_source_0::frequency",
#   units="Hz",
#   min=0.0,
#   max=1e9,
#   description="Output signal frequency"
# )]

Performance Counters

Monitor block performance in real-time:

get_performance_counters()
# Returns: [PerfCounterModel(
#   block="low_pass_filter_0",
#   nproduced=1048576,
#   nconsumed=10485760,
#   avg_work_time_us=45.3,
#   avg_throughput=23.2e6,
#   pc_input_buffers_full=0.85,
#   pc_output_buffers_full=0.12
# ), ...]

# Filter by block
get_performance_counters(block="low_pass_filter_0")

Performance Metrics Explained

Metric	Description
nproduced	Items produced by block
nconsumed	Items consumed by block
avg_work_time_us	Average time in work() function
avg_throughput	Items per second
pc_input_buffers_full	Input buffer utilization (0-1)
pc_output_buffers_full	Output buffer utilization (0-1)

Tip

High pc_input_buffers_full indicates downstream blocks are slow. High pc_output_buffers_full indicates the block itself is a bottleneck.

PMT Message Injection

Send messages to block message ports:

# Send a simple string message
post_message(
    block="pdu_sink_0",
    port="pdus",
    message="hello"
)

# Send a dict (converted to PMT dict)
post_message(
    block="command_handler_0",
    port="command",
    message={"freq": 1e6, "gain": 40}
)

Example: Performance Monitor

#!/usr/bin/env python3
"""Monitor flowgraph performance and identify bottlenecks."""

import asyncio
import time
from fastmcp import Client

async def monitor_performance():
    async with Client("gr-mcp") as client:
        await client.call_tool("enable_runtime_mode", {})

        # Launch with ControlPort
        await client.call_tool("launch_flowgraph", {
            "flowgraph_path": "/tmp/signal_chain.py",
            "name": "perf-test",
            "enable_controlport": True,
            "enable_perf_counters": True
        })

        time.sleep(5)  # Let it warm up

        # Connect via ControlPort
        await client.call_tool("connect_to_container_controlport", {
            "name": "perf-test"
        })

        # Sample performance 10 times
        for i in range(10):
            result = await client.call_tool("get_performance_counters", {})

            print(f"\n--- Sample {i+1} ---")
            for counter in result.data:
                buffer_status = "OK"
                if counter.pc_input_buffers_full > 0.9:
                    buffer_status = "INPUT BOTTLENECK"
                elif counter.pc_output_buffers_full > 0.9:
                    buffer_status = "BLOCK BOTTLENECK"

                print(f"{counter.block}: {counter.avg_throughput/1e6:.2f} MSps, "
                      f"work={counter.avg_work_time_us:.1f}us [{buffer_status}]")

            time.sleep(1)

        # Cleanup
        await client.call_tool("disconnect_controlport", {})
        await client.call_tool("stop_flowgraph", {"name": "perf-test"})
        await client.call_tool("remove_flowgraph", {"name": "perf-test"})

if __name__ == "__main__":
    asyncio.run(monitor_performance())

Using Both XML-RPC and ControlPort

You can have both connections active simultaneously:

# Launch with both
launch_flowgraph(
    flowgraph_path="...",
    name="dual-control",
    xmlrpc_port=8080,
    enable_controlport=True
)

# Connect to XML-RPC for variable control
connect_to_container(name="dual-control")

# Connect to ControlPort for performance monitoring
connect_to_container_controlport(name="dual-control")

# Use XML-RPC for simple variable changes
set_variable(name="freq", value=101.1e6)

# Use ControlPort for performance data
get_performance_counters()

# Disconnect both
disconnect()  # Disconnects both XML-RPC and ControlPort

Disconnect ControlPort

# Disconnect only ControlPort (keep XML-RPC)
disconnect_controlport()

# Or disconnect everything
disconnect()

Next Steps

• Runtime Communication Concepts — Deep dive into protocol details
• Code Coverage — Collect test coverage from runtime