schedule_throughput_test
Schedule network throughput tests between hosts to measure data transfer capacity using pScheduler. Configure source, destination, and duration parameters for performance analysis.
Instructions
Schedule a throughput test using pScheduler.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| source | No | Source host (optional) | |
| dest | Yes | Destination host | |
| duration | No | Test duration (e.g., PT30S) |
Implementation Reference
- FastMCP tool definition for schedule_throughput_test, which calls the underlying pScheduler client.
async def schedule_throughput_test( dest: str, source: Optional[str] = None, duration: str = "PT30S", slip: str = "PT10M", ) -> str: """Schedule a throughput test using pScheduler. Args: dest: Destination host for the test source: Optional source host (if not specified, uses pScheduler host) duration: Test duration in ISO 8601 format (e.g., 'PT30S' for 30 seconds) slip: Schedule slip time in ISO 8601 format (e.g., 'PT10M' for 10 minutes) Returns: JSON string with test details including run URL for status checks """ result = await pscheduler_client.schedule_throughput_test(source, dest, duration, slip) return json.dumps(result.model_dump(), indent=2) - src/perfsonar_mcp/pscheduler.py:88-130 (handler)Actual implementation of the throughput test scheduling logic in the pScheduler client.
async def schedule_throughput_test( self, source: Optional[str], dest: str, duration: str = "PT30S", slip: str = "PT10M", ) -> PSchedulerTaskResponse: """ Schedule a throughput test Args: source: Source host (None for local) dest: Destination host duration: Test duration in ISO 8601 format (e.g., PT30S for 30 seconds) slip: Schedule slip time in ISO 8601 format (e.g., PT10M for 10 minutes) Returns: Task response """ logger.info( f"Scheduling throughput test from {source or 'local'} to {dest} with duration {duration}" ) # Determine which node to schedule on (prefer source if available) scheduler_node = source or dest scheduler_url = f"https://{scheduler_node}/pscheduler" logger.info(f"Using pScheduler at: {scheduler_url}") test_spec = ThroughputTestSpec(source=source, dest=dest, duration=duration) task_request = PSchedulerTaskRequest( test=PSchedulerTestSpec( type="throughput", spec=test_spec.model_dump(exclude_none=True) ), schedule={"slip": slip}, ) # Create a temporary client for this specific scheduler client = PSchedulerClient(scheduler_url) try: return await client.create_task(task_request) finally: await client.close()