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