| 437 | |
| 438 | |
| 439 | class _RttMonitor(MonitorBase): |
| 440 | def __init__(self, topology: Topology, topology_settings: TopologySettings, pool: Pool): |
| 441 | """Maintain round trip times for a server. |
| 442 | |
| 443 | The Topology is weakly referenced. |
| 444 | """ |
| 445 | super().__init__( |
| 446 | topology, |
| 447 | "pymongo_server_rtt_task", |
| 448 | topology_settings.heartbeat_frequency, |
| 449 | common.MIN_HEARTBEAT_INTERVAL, |
| 450 | ) |
| 451 | |
| 452 | self._pool = pool |
| 453 | self._moving_average = MovingAverage() |
| 454 | self._moving_min = MovingMinimum() |
| 455 | self._lock = _async_create_lock() |
| 456 | |
| 457 | async def close(self) -> None: |
| 458 | self.gc_safe_close() |
| 459 | # Increment the generation and maybe close the socket. If the executor |
| 460 | # thread has the socket checked out, it will be closed when checked in. |
| 461 | await self._pool.reset() |
| 462 | |
| 463 | async def add_sample(self, sample: float) -> None: |
| 464 | """Add a RTT sample.""" |
| 465 | async with self._lock: |
| 466 | self._moving_average.add_sample(sample) |
| 467 | self._moving_min.add_sample(sample) |
| 468 | |
| 469 | async def get(self) -> tuple[Optional[float], float]: |
| 470 | """Get the calculated average, or None if no samples yet and the min.""" |
| 471 | async with self._lock: |
| 472 | return self._moving_average.get(), self._moving_min.get() |
| 473 | |
| 474 | async def reset(self) -> None: |
| 475 | """Reset the average RTT.""" |
| 476 | async with self._lock: |
| 477 | self._moving_average.reset() |
| 478 | self._moving_min.reset() |
| 479 | |
| 480 | async def _run(self) -> None: |
| 481 | try: |
| 482 | # NOTE: This thread is only run when using the streaming |
| 483 | # heartbeat protocol (MongoDB 4.4+). |
| 484 | # XXX: Skip check if the server is unknown? |
| 485 | rtt = await self._ping() |
| 486 | await self.add_sample(rtt) |
| 487 | except ReferenceError: |
| 488 | # Topology was garbage-collected. |
| 489 | await self.close() |
| 490 | except Exception: |
| 491 | await self._pool.reset() |
| 492 | |
| 493 | async def _ping(self) -> float: |
| 494 | """Run a "hello" command and return the RTT.""" |
| 495 | async with self._pool.checkout() as conn: |
| 496 | if self._executor._stopped: |