Repository: Chia-Network/chia-blockchain
File: tools/validate_rpcs.py (Python)

Code before:

        # Set initial values for the loop

        pipeline: set[asyncio.Task[None]] = set()
        completed_requests: int = 0
        # measure time for performance measurement per segment
        cycle_start: float = time.monotonic()
        # also measure time for the whole process
        start_time: float = cycle_start

        def add_tasks_for_height(height: int) -> None:
            block_header_hash = get_block_hash_from_height(height, height_to_hash_bytes)
            # Create tasks for each RPC call based on the flags
            if spends_with_conditions:
                pipeline.add(
                    create_referenced_task(node_spends_with_conditions(node_client, block_header_hash, height))
                )
            if block_spends:
                pipeline.add(create_referenced_task(node_block_spends(node_client, block_header_hash, height)))
            if additions_and_removals:
                pipeline.add(create_referenced_task(node_additions_removals(node_client, block_header_hash, height)))

        for i in range(start_height, end_height + 1):
            add_tasks_for_height(height=i)
            # Make Status Updates.
            if len(pipeline) >= pipeline_depth:
                done, pipeline = await asyncio.wait(pipeline, return_when=asyncio.FIRST_COMPLETED)
                completed_requests += len(done)
                now = time.monotonic()
                if cycle_start + 5 < now:
                    time_taken = now - cycle_start
                    print(
                        f"Processed {completed_requests} RPCs in {time_taken:.2f}s, "
                        f"{time_taken / completed_requests:.4f}s per RPC "
                        f"({i - start_height} Blocks completed out of {end_height - start_height})"
                    )
                    completed_requests = 0
                    cycle_start = now

        # Wait for any remaining tasks to complete
        print(f"Waiting for {len(pipeline)} remaining tasks to complete...")
        if pipeline:
            await asyncio.gather(*pipeline)

        print(f"Finished processing blocks from {start_height} to {end_height} (peak: {peak_height})")
        print(
            f"Time per block for the whole process: "
            f"{(time.monotonic() - start_time) / (end_height - start_height):.4f} seconds"
        )



Review comment (suggestion):
when running this tool for all 3 RPC calls, each iteration adds 3 pending tasks, but only one was popped off. Without this change, `pipeline` is allowed to grow indefinitely.

Code after:

        # Set initial values for the loop

        pipeline: set[asyncio.Task[None]] = set()
        completed_requests: int = 0
        # measure time for performance measurement per segment
        cycle_start: float = time.monotonic()
        # also measure time for the whole process
        start_time: float = cycle_start

        def add_tasks_for_height(height: int) -> None:
            block_header_hash = get_block_hash_for_height(height, height_to_hash_bytes)
            # Create tasks for each RPC call based on the flags
            if spends_with_conditions:
                pipeline.add(
                    create_referenced_task(node_spends_with_conditions(node_client, block_header_hash, height))
                )
            if block_spends:
                pipeline.add(create_referenced_task(node_block_spends(node_client, block_header_hash, height)))
            if additions_and_removals:
                pipeline.add(create_referenced_task(node_additions_removals(node_client, block_header_hash, height)))

        for i in range(start_height, end_height + 1):
            add_tasks_for_height(height=i)
            # Make Status Updates.
            while len(pipeline) >= pipeline_depth:
                done, pipeline = await asyncio.wait(pipeline, return_when=asyncio.FIRST_COMPLETED)
                completed_requests += len(done)
                now = time.monotonic()
                if cycle_start + 5 < now:
                    time_taken = now - cycle_start
                    print(
                        f"Processed {completed_requests} RPCs in {time_taken:.2f}s, "
                        f"{time_taken / completed_requests:.4f}s per RPC "
                        f"({i - start_height} Blocks completed out of {end_height - start_height})"
                    )
                    completed_requests = 0
                    cycle_start = now

        # Wait for any remaining tasks to complete
        print(f"Waiting for {len(pipeline)} remaining tasks to complete...")
        if pipeline:
            await asyncio.gather(*pipeline)

        print(f"Finished processing blocks from {start_height} to {end_height} (peak: {peak_height})")
        print(
            f"Time per block for the whole process: "
            f"{(time.monotonic() - start_time) / (end_height - start_height):.4f} seconds"
        )