Tracks a minimum RTT within the last 10 RTT samples.
| 145 | |
| 146 | |
| 147 | class MovingMinimum: |
| 148 | """Tracks a minimum RTT within the last 10 RTT samples.""" |
| 149 | |
| 150 | samples: Deque[float] |
| 151 | |
| 152 | def __init__(self) -> None: |
| 153 | self.samples = deque(maxlen=_MAX_RTT_SAMPLES) |
| 154 | |
| 155 | def add_sample(self, sample: float) -> None: |
| 156 | if sample < 0: |
| 157 | raise ValueError(f"duration cannot be negative {sample}") |
| 158 | self.samples.append(sample) |
| 159 | |
| 160 | def get(self) -> float: |
| 161 | """Get the min, or 0.0 if there aren't enough samples yet.""" |
| 162 | if len(self.samples) >= _MIN_RTT_SAMPLES: |
| 163 | return min(self.samples) |
| 164 | return 0.0 |
| 165 | |
| 166 | def reset(self) -> None: |
| 167 | self.samples.clear() |