Handle global restart requests from any worker (global manager only). Updates the global restart count, fans out RestartReplicaTraining to all replicas, and returns the current global restart count.
(
self, request: manager_pb2.RequestGlobalRestartRequest, context
)
| 457 | raise |
| 458 | |
| 459 | def RequestGlobalRestart( |
| 460 | self, request: manager_pb2.RequestGlobalRestartRequest, context |
| 461 | ) -> manager_pb2.RequestGlobalRestartResponse: |
| 462 | """Handle global restart requests from any worker (global manager only). |
| 463 | |
| 464 | Updates the global restart count, fans out RestartReplicaTraining to all |
| 465 | replicas, and returns the current global restart count. |
| 466 | """ |
| 467 | try: |
| 468 | self._validate_role("global_manager", "handle global restart requests") |
| 469 | |
| 470 | worker_identity = request.worker_identity |
| 471 | logging.warning( |
| 472 | "Global restart requested by worker %s (replica=%d, worker=%d): " |
| 473 | "reason='%s', worker_restart_count=%d", |
| 474 | worker_identity.hostname, |
| 475 | worker_identity.replica_id, |
| 476 | worker_identity.worker_id, |
| 477 | request.reason, |
| 478 | request.restart_count, |
| 479 | ) |
| 480 | |
| 481 | # Only fan out restart if this is a new restart generation. |
| 482 | # If another worker already triggered a restart for this or a later |
| 483 | # generation, skip the fan-out to avoid redundant restarts. |
| 484 | should_fan_out_restart = False |
| 485 | with self._registry_lock: |
| 486 | if request.restart_count + 1 > self._global_restart_count: |
| 487 | self._global_restart_count = request.restart_count + 1 |
| 488 | should_fan_out_restart = True |
| 489 | current_global_restart_count = self._global_restart_count |
| 490 | |
| 491 | if should_fan_out_restart: |
| 492 | restart_reason = ( |
| 493 | f"Coordinated restart: {request.reason} " |
| 494 | f"(requested by replica {worker_identity.replica_id} " |
| 495 | f"worker {worker_identity.worker_id})" |
| 496 | ) |
| 497 | self._restart_all_replicas( |
| 498 | restart_reason, restart_count=current_global_restart_count |
| 499 | ) |
| 500 | else: |
| 501 | logging.info( |
| 502 | "FT Monitor: Skipping redundant restart fan-out from %s " |
| 503 | "(replica=%d, worker=%d, restart_count=%d, global_restart_count=%d)", |
| 504 | worker_identity.hostname, |
| 505 | worker_identity.replica_id, |
| 506 | worker_identity.worker_id, |
| 507 | request.restart_count, |
| 508 | current_global_restart_count, |
| 509 | ) |
| 510 | |
| 511 | return self._create_success_response( |
| 512 | manager_pb2.RequestGlobalRestartResponse, |
| 513 | message=f"Global restart initiated, global_restart_count=" |
| 514 | f"{current_global_restart_count}", |
| 515 | global_restart_count=current_global_restart_count, |
| 516 | ) |
nothing calls this directly
no test coverage detected