[BUG] Fast path optimization for min_bytes_per_latency allows bandwidth escape

[Prgrmman]

In io_model.py, the AbstractDevice class has an optimization to avoid using its _bandwidth_manager() function.

The point of the bandwidth manager is to maintain a set of active requests, predict the next completion time, and credit each request in the active queue with processed time.

This largely simulates PS queue system, where each request is served in parallel by a factor of max concurrency that the user specifies in their OPAL config.

There is an optimization to avoid the bandwidth manager: if the the request is small enough such that the product of the bandwidth and the device latency is bigger than this it means that the request will be gated by latency, so it skips the bandwidth manager. This quantity is called min_bytes_per_latency and expresses the amount of bytes you could move in a latency period assuming you had access to the device's full bandwidth.

In AbstractDevice._process_request_io

        request.arrival_time = self.simpy_env.now
        # self.log.debug(f"{self.simpy_env.now} IORequest.{request.id} arrives (size={request.size})")

        # Wait for a concurrency slot
        with self.concurrency.request() as req_slot:
            # self.log.debug(f"{self.simpy_env.now} IORequest[{request.id}] queued")
            yield req_slot  # WAIT IN QUEUE

            # Enforce minimum latency and check if request can complete within latency period
            if self.latency_per_request_sec > 0:
                # self.log.debug(f"{self.simpy_env.now} IORequest[{request.id}] starts latency {self.latency_per_request_sec}")
                yield self.simpy_env.timeout(self.latency_per_request_sec)

If you receive a concurrency slot, and your request is individually small, you get to skip the bandwidth manager.
The issue is that this result starts to break down if there are multiple requests in the same latency window.

For a single request, this works out well, because the model assumes that the single request occupies all of the device bandwidth over the latency window.
There are a few cases where this produces the wrong result:

1. Assume that initially all concurrency slots are empty. Assume that a batch size of B requests such that B <= C where is C is concurrency setting are enqueued. Assume that each request in the batch is bigger than the min_bytes_per_latency. The code will yield B completion time events of simpy_env.now + latency_per_request_sec. This means it is possible for up to min_bytes_per_latency * C requests or C * bandwidth * latency_per_request_sec to compete in a time of latency_per_request_sec. This effectively allows you to achieve a max bandwidth of C * bandwidth (the latency_per_request_sec terms cancel), thus giving you C times more bandwidth than you actually configured!
2. Assume that the device has C-1 concurrency slots occupied, making your request the Cth request, and assume that the previous C-1 requests were slightly larger than min_bytes_per_latency (let us say 1 byte more), arrived roughly all at the same time, and that they are in the bandwidth manager. The problem with this situation is that the Cth request will given a sojourn time as if it had access to all of the bandwidth in the latency period because it skips the bandwidth manager code path. The Cth request finishes at time now + latency_per_request_sec, whereas the other requests all finish at (C-1) *min_bytes_per_latency / B = (C-1) * latency_per_request_sec. Add these parts up, and you get C * min_bytes_per_latency * latency_per_request_sec all finish in that latency_per_request_sec window, so the result of getting C* bandwidth holds true in both cases.

I had Claude code produce some stubs with custom abstract devices to show that the C * bandwidth result is achievable in real code.

There is another flavor of bug here, and that is the Bandwidth manager gives new arrivals retroactive credit.
Let's say request is interrupted and added self.active_requests.
The issue occurs from this block of code in _bandwidth_manager

            for req_id, req_data in self.active_requests.items():
                req_data["remaining"] -= bytes_processed_per_req

                # Check if request completed
                if req_data["remaining"] <= 0:
                    request = req_data["request"]
                    request.finish_time = self.simpy_env.now
                    request.event.succeed(request)
                    completed_requests.append(req_id)
                    # self.log.debug(f"{self.simpy_env.now} IORequest[{req_id}] completed")

This code block runs after the device is interrupted or the first request completes, and bytes_processed_per_req is computed prior to the interruption. Prior to that interruption, the new request is inserted into self.active_requests`
Thus, it would be possible for the request to arrive just as the next period would complete, and the actual average bandwidth observed would exceed configured bandwidth.

Normally we have bw_per_req = self.bytes_per_sec / num_active. Let's say there are C-1 requests active that can be interrupted, which means bw_per_req = B / (C-1).

We have all of the C-1 requests completing at time T. If that final request interrupts after time T, C * B / (C-1) * T bytes moved in time T.

Without the bug, this should have been (C-1) * B / (C-1) * T) which simplifies to B * T bytes moved over time T.
Because of the retroactive credit, Bandwidth in that time becomes (C) / (C-1) * B, which is strictly greater than B. Thus, like the fast path bug above, configured bandwidth can be exceeded.

[Prgrmman]

This test shows the overshoot factor by C=1000

import pytest
import simpy
from opal.infra.io_model import AbstractDevice, OpalIORequest
from opal.core.environment import OpalSimulatorEnvironment

NANO_SECONDS = 1e9
GIB = 2**30
MIB= 2**20
BW = 100 * GIB # unit in bytes per second
LATENCY_SEC = 20 / 1000 # 20 ms in seconds
CAPACITY = 1000 * GIB
CONCURRENCY = 1000
REQUEST_COUNT = 1000
REQUEST_SIZE = 80 * MIB
#REQUEST_SIZE = BW * LATENCY_SEC # NOTE: this has the largest effect

@pytest.fixture
def fake_env():
    class MockEnv:
        def __init__(self):
            self.simpy_env = simpy.Environment()
        def get_config(self): return {}
        def are_we_done(self): return False
    return MockEnv()

class TestAbstractDevice:
    def test_measured_bw(self, fake_env:OpalSimulatorEnvironment):
        """
        Test that measured bandwidth meets configured bandwidth
        """
        dev = AbstractDevice(fake_env, "fake_device", CAPACITY, BW, LATENCY_SEC, CONCURRENCY)
        requests = [OpalIORequest(REQUEST_SIZE) for _ in range(REQUEST_COUNT)]
        dev.process_requests(requests)
        env = fake_env.simpy_env
        env.run(1000) # arbitrary
        for r in range(10):
            print(requests[r].finish_time)
        # compute max finish time to get duration
        duration = max(r.finish_time for r in requests)
        measured_bw = REQUEST_COUNT * REQUEST_SIZE / duration
        print(f"measured bw and config bw should have a ratio <= 1: {measured_bw / BW}")
        assert measured_bw / BW <= 1

[Prgrmman]

This test case can get 2x the bandwidth (when number of requests = 2 = C) for the retro-active bandwidth bug:

    def test_interrupt_bw(self, fake_env:OpalSimulatorEnvironment):
        """
        Test the interruption path of the bandwidth manager, and make sure
        that measured bandwdith meets configured bandwidth
        """

        dev = AbstractDevice(fake_env, "fake_device", CAPACITY, BW, LATENCY_SEC, CONCURRENCY)

        # For the problem to appear, we need < CONCURRENCY requests to be in the system first
        # allow that request to run for a few ms, and then insert the next request
        # the next request will be given fake credit, allowing us to measure a faster than configured bw. 

        # the resident request has to still be transferring when the second one
        # registers, so size it to outlast interrupt_delay
        interrupt_delay = 5 / 1000 # 5 ms in seconds
        resident_size = 4 * GIB

        env = fake_env.simpy_env
        resident_request = OpalIORequest(resident_size)
        interrupt_request = OpalIORequest(REQUEST_SIZE)

        def submit_after(request, delay):
            """Hand a request to the device `delay` seconds into the run"""
            yield env.timeout(delay)
            dev.process_one_request(request)

        dev.process_one_request(resident_request)
        env.process(submit_after(interrupt_request, interrupt_delay))

        env.run(1000) # arbitrary

        # service time is latency plus transfer; the transfer alone can never
        # beat size / BW no matter what else is in flight
        transfer_time = interrupt_request.finish_time - interrupt_request.arrival_time - LATENCY_SEC
        fastest_possible = REQUEST_SIZE / BW
        print(f"transfer took {transfer_time*1000:.4f} ms, floor is {fastest_possible*1000:.4f} ms")
        print(interrupt_request.arrival_time, interrupt_request.finish_time)
        assert transfer_time >= fastest_possible

[animeshtrivedi]

Thanks for reporting this bug. Indeed the issue that you report can happen. My expectation was that this would be a corner case, not something that will be pertinent in daily benchmarking. But absolutely - the bug exists, and should not be the case.

Thanks for the PR also @Prgrmman