TODO(nfnt): Have this function return a `Result`.
| 4491 | |
| 4492 | // TODO(nfnt): Have this function return a `Result`. |
| 4493 | void Slave::checkpointResourceState( |
| 4494 | vector<Resource> resources, |
| 4495 | bool changeTotal) |
| 4496 | { |
| 4497 | // TODO(jieyu): Here we assume that CheckpointResourcesMessages are |
| 4498 | // ordered (i.e., slave receives them in the same order master sends |
| 4499 | // them). This should be true in most of the cases because TCP |
| 4500 | // enforces in order delivery per connection. However, the ordering |
| 4501 | // is technically not guaranteed because master creates multiple |
| 4502 | // connections to the slave in some cases (e.g., persistent socket |
| 4503 | // to slave breaks and master uses ephemeral socket). This could |
| 4504 | // potentially be solved by using a version number and rejecting |
| 4505 | // stale messages according to the version number. |
| 4506 | // |
| 4507 | // If CheckpointResourcesMessages are delivered out-of-order, there |
| 4508 | // are two cases to consider: |
| 4509 | // (1) If master does not fail over, it will reconcile the state |
| 4510 | // with the slave if the framework later changes the |
| 4511 | // checkpointed resources. Since master is the source of truth |
| 4512 | // for reservations, the inconsistency is not exposed to |
| 4513 | // frameworks. |
| 4514 | // (2) If master does fail over, the slave will inform the new |
| 4515 | // master about the incorrect checkpointed resources. When that |
| 4516 | // happens, we expect framework to reconcile based on the |
| 4517 | // offers they get. |
| 4518 | |
| 4519 | // An agent with resource providers requires an operation feedback protocol |
| 4520 | // instead of simply checkpointing results by the master. Fail hard here |
| 4521 | // instead of applying an incompatible message. |
| 4522 | const bool checkpointingResourceProviderResources = std::any_of( |
| 4523 | resources.begin(), |
| 4524 | resources.end(), |
| 4525 | [](const Resource& resource) { return resource.has_provider_id(); }); |
| 4526 | |
| 4527 | CHECK(!checkpointingResourceProviderResources) |
| 4528 | << "Resource providers must perform their own checkpointing"; |
| 4529 | |
| 4530 | upgradeResources(&resources); |
| 4531 | |
| 4532 | Resources resourcesToCheckpoint = resources; |
| 4533 | |
| 4534 | // Tests if the given Operation needs to be checkpointed on the agent. |
| 4535 | // |
| 4536 | // The agent checkpoints pending CREATE/DESTROY operations on agent default |
| 4537 | // resources and terminal operations on agent default resources that have |
| 4538 | // unacknowledged status updates. |
| 4539 | auto operationNeedsCheckpointing = [](const Operation& operation) { |
| 4540 | Result<ResourceProviderID> resourceProviderId = |
| 4541 | getResourceProviderId(operation.info()); |
| 4542 | |
| 4543 | CHECK(!resourceProviderId.isError()) |
| 4544 | << "Failed to get resource provider ID: " |
| 4545 | << resourceProviderId.error(); |
| 4546 | |
| 4547 | if (resourceProviderId.isSome()) { |
| 4548 | return false; |
| 4549 | } |
| 4550 |