diff --git a/src/ray/common/scheduling/cluster_resource_scheduler.cc b/src/ray/common/scheduling/cluster_resource_scheduler.cc index 65e6ea5ac..f6ad49bc6 100644 --- a/src/ray/common/scheduling/cluster_resource_scheduler.cc +++ b/src/ray/common/scheduling/cluster_resource_scheduler.cc @@ -1,55 +1,18 @@ #include "cluster_resource_scheduler.h" -std::string NodeResources::DebugString() { - std::stringstream buffer; - buffer << " node predefined resources {"; - for (size_t i = 0; i < this->capacities.size(); i++) { - buffer << "(" << this->capacities[i].total << ":" << this->capacities[i].available - << ") "; - } - buffer << "}" << std::endl; - - buffer << " node custom resources {"; - for (auto it = this->custom_resources.begin(); it != this->custom_resources.end(); - ++it) { - buffer << it->first << ":(" << it->second.total << ":" << it->second.available - << ") "; - } - buffer << "}" << std::endl; - return buffer.str(); -} - -std::string TaskRequest::DebugString() { - std::stringstream buffer; - buffer << std::endl << " request predefined resources {"; - for (size_t i = 0; i < this->predefined_resources.size(); i++) { - buffer << "(" << this->predefined_resources[i].demand << ":" - << this->predefined_resources[i].soft << ") "; - } - buffer << "}" << std::endl; - - buffer << " request custom resources {"; - for (size_t i = 0; i < this->custom_resources.size(); i++) { - buffer << this->custom_resources[i].id << ":" - << "(" << this->custom_resources[i].req.demand << ":" - << this->custom_resources[i].req.soft << ") "; - } - buffer << "}" << std::endl; - return buffer.str(); -} - bool NodeResources::operator==(const NodeResources &other) { for (size_t i = 0; i < PredefinedResources_MAX; i++) { - if (this->capacities[i].total != other.capacities[i].total) { + if (this->predefined_resources[i].total != other.predefined_resources[i].total) { return false; } - if (this->capacities[i].available != other.capacities[i].available) { + if (this->predefined_resources[i].available != + other.predefined_resources[i].available) { return false; } } if (this->custom_resources.size() != other.custom_resources.size()) { - return true; + return false; } for (auto it1 = this->custom_resources.begin(); it1 != this->custom_resources.end(); @@ -68,10 +31,175 @@ bool NodeResources::operator==(const NodeResources &other) { return true; } +std::string NodeResources::DebugString() { + std::stringstream buffer; + buffer << " node predefined resources {"; + for (size_t i = 0; i < static_cast(this->predefined_resources.size()); i++) { + buffer << "(" << this->predefined_resources[i].total << ":" + << this->predefined_resources[i].available << ") "; + } + buffer << "}" << std::endl; + + buffer << " node custom resources {"; + for (auto it = this->custom_resources.begin(); it != this->custom_resources.end(); + ++it) { + buffer << it->first << ":(" << it->second.total << ":" << it->second.available + << ") "; + } + buffer << "}" << std::endl; + return buffer.str(); +} + +std::string VectorToString(std::vector &vector) { + std::stringstream buffer; + + buffer << "["; + for (size_t i = 0; i < vector.size(); i++) { + buffer << vector[i]; + if (i < vector.size() - 1) { + buffer << ", "; + } + } + buffer << "]"; + return buffer.str(); +} + +bool NodeResourceInstances::operator==(const NodeResourceInstances &other) { + for (size_t i = 0; i < PredefinedResources_MAX; i++) { + if (!EqualVectors(this->predefined_resources[i].total, + other.predefined_resources[i].total)) { + return false; + } + if (!EqualVectors(this->predefined_resources[i].available, + other.predefined_resources[i].available)) { + return false; + } + } + + if (this->custom_resources.size() != other.custom_resources.size()) { + return false; + } + + for (auto it1 = this->custom_resources.begin(); it1 != this->custom_resources.end(); + ++it1) { + auto it2 = other.custom_resources.find(it1->first); + if (it2 == other.custom_resources.end()) { + return false; + } + if (!EqualVectors(it1->second.total, it2->second.total)) { + return false; + } + if (!EqualVectors(it1->second.available, it2->second.available)) { + return false; + } + } + return true; +} + +std::string NodeResourceInstances::DebugString() { + std::stringstream buffer; + buffer << " node predefined resources {"; + for (size_t i = 0; i < this->predefined_resources.size(); i++) { + buffer << "(" << VectorToString(predefined_resources[i].total) << ":" + << VectorToString(this->predefined_resources[i].available) << ") "; + } + buffer << "}" << std::endl; + + buffer << " node custom resources {"; + for (auto it = this->custom_resources.begin(); it != this->custom_resources.end(); + ++it) { + buffer << it->first << ":(" << VectorToString(it->second.total) << ":" + << VectorToString(it->second.available) << ") "; + } + buffer << "}" << std::endl; + return buffer.str(); +}; + +TaskResourceInstances NodeResourceInstances::GetAvailableResourceInstances() { + TaskResourceInstances task_resources; + task_resources.predefined_resources.resize(PredefinedResources_MAX); + + for (size_t i = 0; i < this->predefined_resources.size(); i++) { + task_resources.predefined_resources[i] = this->predefined_resources[i].available; + } + + for (const auto it : this->custom_resources) { + task_resources.custom_resources.emplace(it.first, it.second.available); + } + + return task_resources; +}; + +std::string TaskRequest::DebugString() { + std::stringstream buffer; + buffer << std::endl << " request predefined resources {"; + for (size_t i = 0; i < this->predefined_resources.size(); i++) { + buffer << "(" << this->predefined_resources[i].demand << ":" + << this->predefined_resources[i].soft << ") "; + } + buffer << "}" << std::endl; + + buffer << " request custom resources {"; + for (size_t i = 0; i < this->custom_resources.size(); i++) { + buffer << this->custom_resources[i].id << ":" + << "(" << this->custom_resources[i].demand << ":" + << this->custom_resources[i].soft << ") "; + } + buffer << "}" << std::endl; + return buffer.str(); +} + +std::string TaskResourceInstances::DebugString() { + std::stringstream buffer; + buffer << std::endl << " task allocation: P {"; + for (size_t i = 0; i < this->predefined_resources.size(); i++) { + buffer << VectorToString(this->predefined_resources[i]); + } + buffer << "}"; + + buffer << " C {"; + for (auto it = this->custom_resources.begin(); it != this->custom_resources.end(); + ++it) { + buffer << it->first << ":" << VectorToString(it->second) << ", "; + } + + buffer << "}" << std::endl; + return buffer.str(); +} + +bool EqualVectors(const std::vector &v1, const std::vector &v2) { + return (v1.size() == v2.size() && std::equal(v1.begin(), v1.end(), v2.begin())); +} + +bool TaskResourceInstances::operator==(const TaskResourceInstances &other) { + for (size_t i = 0; i < PredefinedResources_MAX; i++) { + if (!EqualVectors(this->predefined_resources[i], other.predefined_resources[i])) { + return false; + } + } + + if (this->custom_resources.size() != other.custom_resources.size()) { + return false; + } + + for (auto it1 = this->custom_resources.begin(); it1 != this->custom_resources.end(); + ++it1) { + auto it2 = other.custom_resources.find(it1->first); + if (it2 == other.custom_resources.end()) { + return false; + } + if (!EqualVectors(it1->second, it2->second)) { + return false; + } + } + return true; +} + ClusterResourceScheduler::ClusterResourceScheduler( int64_t local_node_id, const NodeResources &local_node_resources) : local_node_id_(local_node_id) { AddOrUpdateNode(local_node_id_, local_node_resources); + InitLocalResources(local_node_resources); } ClusterResourceScheduler::ClusterResourceScheduler( @@ -93,8 +221,10 @@ void ClusterResourceScheduler::AddOrUpdateNode( void ClusterResourceScheduler::SetPredefinedResources(const NodeResources &new_resources, NodeResources *old_resources) { for (size_t i = 0; i < PredefinedResources_MAX; i++) { - old_resources->capacities[i].total = new_resources.capacities[i].total; - old_resources->capacities[i].available = new_resources.capacities[i].available; + old_resources->predefined_resources[i].total = + new_resources.predefined_resources[i].total; + old_resources->predefined_resources[i].available = + new_resources.predefined_resources[i].available; } } @@ -141,7 +271,8 @@ int64_t ClusterResourceScheduler::IsSchedulable(const TaskRequest &task_req, // First, check predefined resources. for (size_t i = 0; i < PredefinedResources_MAX; i++) { - if (task_req.predefined_resources[i].demand > resources.capacities[i].available) { + if (task_req.predefined_resources[i].demand > + resources.predefined_resources[i].available) { if (task_req.predefined_resources[i].soft) { // A soft constraint has been violated. violations++; @@ -157,15 +288,15 @@ int64_t ClusterResourceScheduler::IsSchedulable(const TaskRequest &task_req, if (it == resources.custom_resources.end()) { // Requested resource doesn't exist at this node. - if (task_req.custom_resources[i].req.soft) { + if (task_req.custom_resources[i].soft) { violations++; } else { return -1; } } else { - if (task_req.custom_resources[i].req.demand > it->second.available) { + if (task_req.custom_resources[i].demand > it->second.available) { // Resource constraint is violated. - if (task_req.custom_resources[i].req.soft) { + if (task_req.custom_resources[i].soft) { violations++; } else { return -1; @@ -261,22 +392,21 @@ bool ClusterResourceScheduler::SubtractNodeAvailableResources( NodeResources &resources = it->second; // Just double check this node can still schedule the task request. - if (IsSchedulable(task_req, local_node_id_, resources) == -1) { + if (IsSchedulable(task_req, node_id, resources) == -1) { return false; } for (size_t i = 0; i < PredefinedResources_MAX; i++) { - resources.capacities[i].available = - std::max(static_cast(0), resources.capacities[i].available - - task_req.predefined_resources[i].demand); + resources.predefined_resources[i].available = + std::max(0., resources.predefined_resources[i].available - + task_req.predefined_resources[i].demand); } for (size_t i = 0; i < task_req.custom_resources.size(); i++) { auto it = resources.custom_resources.find(task_req.custom_resources[i].id); if (it != resources.custom_resources.end()) { it->second.available = - std::max(static_cast(0), - it->second.available - task_req.custom_resources[i].req.demand); + std::max(0., it->second.available - task_req.custom_resources[i].demand); } } return true; @@ -299,15 +429,17 @@ bool ClusterResourceScheduler::AddNodeAvailableResources(int64_t node_id, NodeResources &resources = it->second; for (size_t i = 0; i < PredefinedResources_MAX; i++) { - resources.capacities[i].available = - resources.capacities[i].available + task_req.predefined_resources[i].demand; + resources.predefined_resources[i].available = + std::min(resources.predefined_resources[i].available + + task_req.predefined_resources[i].demand, + resources.predefined_resources[i].total); } for (size_t i = 0; i < task_req.custom_resources.size(); i++) { auto it = resources.custom_resources.find(task_req.custom_resources[i].id); if (it != resources.custom_resources.end()) { - it->second.available = - it->second.available + task_req.custom_resources[i].req.demand; + it->second.available = std::min( + it->second.available + task_req.custom_resources[i].demand, it->second.total); } } return true; @@ -338,9 +470,10 @@ void ClusterResourceScheduler::ResourceMapToNodeResources( const std::unordered_map &resource_map_total, const std::unordered_map &resource_map_available, NodeResources *node_resources) { - node_resources->capacities.resize(PredefinedResources_MAX); + node_resources->predefined_resources.resize(PredefinedResources_MAX); for (size_t i = 0; i < PredefinedResources_MAX; i++) { - node_resources->capacities[i].total = node_resources->capacities[i].available = 0; + node_resources->predefined_resources[i].total = + node_resources->predefined_resources[i].available = 0; } for (auto it = resource_map_total.begin(); it != resource_map_total.end(); ++it) { @@ -353,13 +486,13 @@ void ClusterResourceScheduler::ResourceMapToNodeResources( resource_capacity.available = (int64_t)it2->second; } if (it->first == ray::kCPU_ResourceLabel) { - node_resources->capacities[CPU] = resource_capacity; + node_resources->predefined_resources[CPU] = resource_capacity; } else if (it->first == ray::kGPU_ResourceLabel) { - node_resources->capacities[GPU] = resource_capacity; + node_resources->predefined_resources[GPU] = resource_capacity; } else if (it->first == ray::kTPU_ResourceLabel) { - node_resources->capacities[TPU] = resource_capacity; + node_resources->predefined_resources[TPU] = resource_capacity; } else if (it->first == ray::kMemory_ResourceLabel) { - node_resources->capacities[MEM] = resource_capacity; + node_resources->predefined_resources[MEM] = resource_capacity; } else { // This is a custom resource. node_resources->custom_resources.emplace(string_to_int_map_.Insert(it->first), @@ -391,8 +524,8 @@ void ClusterResourceScheduler::ResourceMapToTaskRequest( task_request->predefined_resources[MEM].demand = it->second; } else { task_request->custom_resources[i].id = string_to_int_map_.Insert(it->first); - task_request->custom_resources[i].req.demand = it->second; - task_request->custom_resources[i].req.soft = false; + task_request->custom_resources[i].demand = it->second; + task_request->custom_resources[i].soft = false; i++; } } @@ -419,14 +552,14 @@ void ClusterResourceScheduler::UpdateResourceCapacity(const std::string &client_ idx = (int)MEM; }; if (idx != -1) { - int64_t diff_capacity = resource_total - it->second.capacities[idx].total; - it->second.capacities[idx].total += diff_capacity; - it->second.capacities[idx].available += diff_capacity; - if (it->second.capacities[idx].available < 0) { - it->second.capacities[idx].available = 0; + int64_t diff_capacity = resource_total - it->second.predefined_resources[idx].total; + it->second.predefined_resources[idx].total += diff_capacity; + it->second.predefined_resources[idx].available += diff_capacity; + if (it->second.predefined_resources[idx].available < 0) { + it->second.predefined_resources[idx].available = 0; } - if (it->second.capacities[idx].total < 0) { - it->second.capacities[idx].total = 0; + if (it->second.predefined_resources[idx].total < 0) { + it->second.predefined_resources[idx].total = 0; } } else { int64_t resource_id = string_to_int_map_.Insert(resource_name); @@ -467,7 +600,7 @@ void ClusterResourceScheduler::DeleteResource(const std::string &client_id_strin idx = (int)MEM; }; if (idx != -1) { - it->second.capacities[idx].total = 0; + it->second.predefined_resources[idx].total = 0; } else { int64_t resource_id = string_to_int_map_.Get(resource_name); auto itr = it->second.custom_resources.find(resource_id); @@ -487,3 +620,237 @@ std::string ClusterResourceScheduler::DebugString(void) { } return buffer.str(); } + +void ClusterResourceScheduler::InitResourceInstances( + double total, bool unit_instances, + ResourceInstanceCapacities *instance_list /* return */) { + if (unit_instances) { + size_t num_instances = static_cast(total); + instance_list->total.resize(num_instances); + instance_list->available.resize(num_instances); + for (size_t i = 0; i < num_instances; i++) { + instance_list->total[i] = instance_list->available[i] = 1.0; + }; + } else { + instance_list->total.resize(1); + instance_list->available.resize(1); + instance_list->total[0] = instance_list->available[0] = total; + } +} + +void ClusterResourceScheduler::InitLocalResources(const NodeResources &node_resources) { + local_resources_.predefined_resources.resize(PredefinedResources_MAX); + + for (size_t i = 0; i < PredefinedResources_MAX; i++) { + if (node_resources.predefined_resources[i].total > 0) { + InitResourceInstances( + node_resources.predefined_resources[i].total, + (UnitInstanceResources.find(i) != UnitInstanceResources.end()), + &local_resources_.predefined_resources[i]); + } + } + + if (node_resources.custom_resources.size() == 0) { + return; + } + + for (auto it = node_resources.custom_resources.begin(); + it != node_resources.custom_resources.end(); ++it) { + if (it->second.total > 0) { + ResourceInstanceCapacities instance_list; + InitResourceInstances(it->second.total, false, &instance_list); + local_resources_.custom_resources.emplace(it->first, instance_list); + } + } +} + +void ClusterResourceScheduler::AddAvailableResourceInstances( + std::vector available, + ResourceInstanceCapacities *resource_instances /* return */) { + for (size_t i = 0; i < available.size(); i++) { + resource_instances->available[i] = std::min( + resource_instances->available[i] + available[i], resource_instances->total[i]); + } +} + +void ClusterResourceScheduler::SubtractAvailableResourceInstances( + std::vector available, + ResourceInstanceCapacities *resource_instances /* return */) { + RAY_CHECK(available.size() == resource_instances->available.size()); + + for (size_t i = 0; i < available.size(); i++) { + resource_instances->available[i] = + std::max(resource_instances->available[i] - available[i], 0.); + } +} + +bool ClusterResourceScheduler::AllocateResourceInstances( + double demand, bool soft, std::vector &available, + std::vector *allocation /* return */) { + allocation->resize(available.size()); + double remaining_demand = demand; + + if (available.size() == 1) { + // This resource has just an instance. + if (available[0] >= remaining_demand) { + available[0] -= remaining_demand; + (*allocation)[0] = remaining_demand; + return true; + } else { + if (soft) { + available[0] = 0; + return true; + } + // Not enough capacity. + return false; + } + } + + // If resources has multiple instances, each instance has total capacity of 1. + // + // If this resource constraint is hard, as long as remaining_demand is greater than 1., + // allocate full unit-capacity instances until the remaining_demand becomes fractional. + // Then try to find the best fit for the fractional remaining_resources. Best fist means + // allocating the resource instance with the smallest available capacity greater than + // remaining_demand + // + // If resource constraint is soft, allocate as many full unit-capacity resources and + // then distribute remaining_demand across remaining instances. Note that in case we can + // overallocate this resource. + if (remaining_demand >= 1.) { + for (size_t i = 0; i < available.size(); i++) { + if (available[i] == 1.) { + // Allocate a full unit-capacity instance. + (*allocation)[i] = 1.; + available[i] = 0; + remaining_demand -= 1.; + } + if (remaining_demand < 1.) { + break; + } + } + } + + if (soft) { + // Just get as many resources as available. + for (size_t i = 0; i < available.size(); i++) { + if (available[i] >= remaining_demand) { + available[i] -= remaining_demand; + (*allocation)[i] = remaining_demand; + return true; + } else { + (*allocation)[i] = available[i]; + remaining_demand -= available[i]; + available[i] = 0; + } + } + return true; + } + + if (remaining_demand >= 1.) { + // Cannot satisfy a demand greater than one if no unit caapcity resource is available. + return false; + } + + // Remaining demand is fractional. Find the best fit, if exists. + if (remaining_demand > 0.) { + int64_t idx_best_fit = -1; + double available_best_fit = 1.; + for (size_t i = 0; i < available.size(); i++) { + if (available[i] >= remaining_demand) { + if (idx_best_fit == -1 || + (available[i] - remaining_demand < available_best_fit)) { + available_best_fit = available[i] - remaining_demand; + idx_best_fit = static_cast(i); + } + } + } + if (idx_best_fit == -1) { + return false; + } else { + (*allocation)[idx_best_fit] = remaining_demand; + available[idx_best_fit] -= remaining_demand; + } + } + return true; +} + +bool ClusterResourceScheduler::AllocateTaskResourceInstances( + const TaskRequest &task_req, TaskResourceInstances *task_allocation /* return */) { + auto it = nodes_.find(local_node_id_); + if (it == nodes_.end()) { + return false; + } + + // Just double check this node can still schedule the task request. + if (IsSchedulable(task_req, local_node_id_, it->second) == -1) { + return false; + } + + task_allocation->predefined_resources.resize(PredefinedResources_MAX); + for (size_t i = 0; i < PredefinedResources_MAX; i++) { + if (task_req.predefined_resources[i].demand > 0) { + if (!AllocateResourceInstances(task_req.predefined_resources[i].demand, + task_req.predefined_resources[i].soft, + local_resources_.predefined_resources[i].available, + &task_allocation->predefined_resources[i])) { + // Allocation failed. Restore node's local resources by freeing the resources + // of the failed allocation. + FreeTaskResourceInstances(*task_allocation); + return false; + } + } + } + + for (size_t i = 0; i < task_req.custom_resources.size(); i++) { + auto it = local_resources_.custom_resources.find(task_req.custom_resources[i].id); + if (it != local_resources_.custom_resources.end()) { + if (task_req.custom_resources[i].demand > 0) { + std::vector allocation; + bool success = AllocateResourceInstances(task_req.custom_resources[i].demand, + task_req.custom_resources[i].soft, + it->second.available, &allocation); + // Even if allocation failed we need to remember partial allocations to correctly + // free resources. + task_allocation->custom_resources.emplace(it->first, allocation); + if (!success) { + // Allocation failed. Restore node's local resources by freeing the resources + // of the failed allocation. + FreeTaskResourceInstances(*task_allocation); + return false; + } + } + } else { + return false; + } + } + return true; +} + +void ClusterResourceScheduler::FreeTaskResourceInstances( + TaskResourceInstances &task_allocation) { + for (size_t i = 0; i < PredefinedResources_MAX; i++) { + AddAvailableResourceInstances(task_allocation.predefined_resources[i], + &local_resources_.predefined_resources[i]); + } + + for (auto it = task_allocation.custom_resources.begin(); + it != task_allocation.custom_resources.end(); it++) { + auto it_local = local_resources_.custom_resources.find(it->first); + if (it_local != local_resources_.custom_resources.end()) { + AddAvailableResourceInstances(it->second, &it_local->second); + } + } +} + +void ClusterResourceScheduler::AddCPUResourceInstances( + std::vector &cpu_instances) { + AddAvailableResourceInstances(cpu_instances, + &local_resources_.predefined_resources[CPU]); +} + +void ClusterResourceScheduler::SubtractCPUResourceInstances( + std::vector &cpu_instances) { + SubtractAvailableResourceInstances(cpu_instances, + &local_resources_.predefined_resources[CPU]); +} diff --git a/src/ray/common/scheduling/cluster_resource_scheduler.h b/src/ray/common/scheduling/cluster_resource_scheduler.h index d04b3e596..29a9cefed 100644 --- a/src/ray/common/scheduling/cluster_resource_scheduler.h +++ b/src/ray/common/scheduling/cluster_resource_scheduler.h @@ -13,15 +13,26 @@ /// List of predefined resources. enum PredefinedResources { CPU, MEM, GPU, TPU, PredefinedResources_MAX }; +// Specify resources that consists of unit-size instances. +static std::unordered_set UnitInstanceResources{CPU, GPU, TPU}; + +// Helper function to compare two vectors with double values. +bool EqualVectors(const std::vector &v1, const std::vector &v2); struct ResourceCapacity { - int64_t total; - int64_t available; + double total; + double available; +}; + +/// Capacities of each instance of a resource. +struct ResourceInstanceCapacities { + std::vector total; + std::vector available; }; struct ResourceRequest { /// Amount of resource being requested. - int64_t demand; + double demand; /// Specify whether the request is soft or hard. /// If hard, the entire request is denied if the demand exceeds the resource /// availability. Otherwise, the request can be still be granted. @@ -30,26 +41,13 @@ struct ResourceRequest { }; /// Resource request, including resource ID. This is used for custom resources. -struct ResourceRequestWithId { +struct ResourceRequestWithId : ResourceRequest { /// Resource ID. int64_t id; - /// Resource request. - ResourceRequest req; }; -struct NodeResources { - /// Available and total capacities for predefined resources. - std::vector capacities; - /// Map containing custom resources. The key of each entry represents the - /// custom resource ID. - absl::flat_hash_map custom_resources; - /// Returns if this equals another node resources. - bool operator==(const NodeResources &other); - /// Returns human-readable string for these resources. - std::string DebugString(); -}; - -struct TaskRequest { +class TaskRequest { + public: /// List of predefined resources required by the task. std::vector predefined_resources; /// List of custom resources required by the tasl. @@ -63,6 +61,51 @@ struct TaskRequest { std::string DebugString(); }; +// Task request specifying instances for each resource. +class TaskResourceInstances { + public: + /// The list of instances of each predifined resource allocated to a task. + std::vector> predefined_resources; + /// The list of instances of each custom resource allocated to a task. + absl::flat_hash_map> custom_resources; + bool operator==(const TaskResourceInstances &other); + /// Get CPU instances only. + std::vector GetCPUInstances() { return this->predefined_resources[CPU]; }; + /// Returns human-readable string for these resources. + std::string DebugString(); +}; + +/// Total and available capacities of each resource of a node. +class NodeResources { + public: + /// Available and total capacities for predefined resources. + std::vector predefined_resources; + /// Map containing custom resources. The key of each entry represents the + /// custom resource ID. + absl::flat_hash_map custom_resources; + /// Returns if this equals another node resources. + bool operator==(const NodeResources &other); + /// Returns human-readable string for these resources. + std::string DebugString(); +}; + +/// Total and available capacities of each resource instance. +/// This is used to describe the resources of the local node. +class NodeResourceInstances { + public: + /// Available and total capacities for each instance of a predefined resource. + std::vector predefined_resources; + /// Map containing custom resources. The key of each entry represents the + /// custom resource ID. + absl::flat_hash_map custom_resources; + /// Extract available resource instances. + TaskResourceInstances GetAvailableResourceInstances(); + /// Returns if this equals another node resources. + bool operator==(const NodeResourceInstances &other); + /// Returns human-readable string for these resources. + std::string DebugString(); +}; + /// Class encapsulating the cluster resources and the logic to assign /// tasks to nodes based on the task's constraints and the available /// resources at those nodes. @@ -70,8 +113,10 @@ class ClusterResourceScheduler { /// List of nodes in the clusters and their resources organized as a map. /// The key of the map is the node ID. absl::flat_hash_map nodes_; - /// ID of local node. + /// Identifier of local node. int64_t local_node_id_; + /// Resources of local node. + NodeResourceInstances local_resources_; /// Keep the mapping between node and resource IDs in string representation /// to integer representation. Used for improving map performance. StringIdMap string_to_int_map_; @@ -90,9 +135,6 @@ class ClusterResourceScheduler { const absl::flat_hash_map &new_custom_resources, absl::flat_hash_map *old_custom_resources); - /// Returns human-readable string for this scheduler. - std::string DebugString(); - public: ClusterResourceScheduler(void){}; @@ -188,7 +230,7 @@ class ClusterResourceScheduler { const std::string &node_id, const std::unordered_map &task_request); - /// Increase available resources of a node when a worker has Finished + /// Increase available resources of a node when a worker has finished /// a task. /// /// \param node_id: ID of node on which request is being scheduled. @@ -227,9 +269,102 @@ class ClusterResourceScheduler { void DeleteResource(const std::string &client_id_string, const std::string &resource_name); - /// Check whether two node resources are identical. - bool EqualNodeResources(const NodeResources &node_resources1, - const NodeResources &node_resources2); + /// Return local resources. + NodeResourceInstances GetLocalResources() { return local_resources_; }; + + /// Create instances for each resource associated with the local node, given + /// the node's resources. + /// + /// \param local_resources: Total resources of the node. + void InitLocalResources(const NodeResources &local_resources); + + /// Initialize the instances of a given resource given the resource's total capacity. + /// If unit_instances is true we split the resources in unit-size instances. For + /// example, if total = 10, then we create 10 instances, each with caoacity 1. + /// Otherwise, we create a single instance of capacity equal to the resource's capacity. + /// + /// \param total: Total resource capacity. + /// \param unit_instances: If true, we split the resource in unit-size instances. + /// If false, we create a single instance of capacity "total". + /// \param instance_list: The list of capacities this resource instances. + void InitResourceInstances(double total, bool unit_instances, + ResourceInstanceCapacities *instance_list); + + /// Allocate enough capacity across the instances of a resource to satisfy "demand". + /// If resource has multiple unit-capacity instance, we consider two cases. + /// + /// 1) If the constraint is hard, allocate full unit-capacity instances until + /// demand becomes fractional, and then satisfy the fractional deman using the + /// instance with the smallest available capacity that can satisfy the fractional + /// demand. For example, assume a resource conisting of 4 instances, with available + /// capacities: (1., 1., .7, 0.5) and deman of 1.2. Then we allocate one full + /// instance and then allocate 0.2 of the 0.5 instance (as this is the instance + /// with the smalest available capacity that can satisfy the remaining demand of 0.2). + /// As a result remaining available capacities will be (0., 1., .7, .2). + /// Thus, if the constraint is hard, we will allocate at most a fractional resource. + /// + /// 2) If the constraint is soft, we can allocate multiple fractional resources, + /// and even overallocate the resource. For example, in the previous case, if we + /// have a demand of 1.8, we can allocate one full instance, the 0.5 instance, and + /// 0.1 from the 0.7 instance. Furthermore, if the demand is 3.5, then we allocate + /// all instances, and return success (true), despite the fact that the total + /// available capacity of the rwsource is 3.2 (= 1. + 1. + .7 + .5), which is less + /// than the demand, 3.5. + /// + /// \param demand: The resource amount to be allocated. + /// \param soft: Specifies whether this demand has soft or hard constraints. + /// \param available: List of available capacities of the instances of the resource. + /// \param allocation: List of instance capacities allocated to satisfy the demand. + /// This is a return parameter. + /// + /// \return true, if allocation successful. In this case, the sum of the elements in + /// "allocation" is equal to "demand". + bool AllocateResourceInstances(double demand, bool soft, std::vector &available, + std::vector *allocation); + + /// Allocate local resources to satisfy a given request (task_req). + /// + /// \param task_req: Resources requested by a task. + /// \param task_allocation: Local resources allocated to satsify task_req demand. + /// This is an output argument. + /// + /// \return true, if allocation successful. If false, the caller needs to free the + /// allocated resources, i.e., task_allocation. + bool AllocateTaskResourceInstances(const TaskRequest &task_req, + TaskResourceInstances *task_allocation); + + /// Free resources which were allocated with a task. The freed resources are + /// added back to the node's local available resources. + /// + /// \param task_allocation: Task's resources to be freed. + void FreeTaskResourceInstances(TaskResourceInstances &task_allocation); + + /// Increase the available capacities of the instances of a given resource. + /// + /// \param available A list of available capacities for resource's instances. + /// \param resource_instances List of the resource instances being updated. + void AddAvailableResourceInstances(std::vector available, + ResourceInstanceCapacities *resource_instances); + + /// Decrease the available capacities of the instances of a given resource. + /// + /// \param free A list of capacities for resource's instances to be freed. + /// \param resource_instances List of the resource instances being updated. + void SubtractAvailableResourceInstances(std::vector free, + ResourceInstanceCapacities *resource_instances); + + /// Increase the available CPU instances of this node. + /// + /// \param cpu_instances CPU instances to be added to available cpus. + void AddCPUResourceInstances(std::vector &cpu_instances); + + /// Decrease the available cpu instances of this node. + /// + /// \param cpu_instances Cpu instances to be removed from available cpus. + void SubtractCPUResourceInstances(std::vector &cpu_instances); + + /// Return human-readable string for this scheduler state. + std::string DebugString(); }; #endif // RAY_COMMON_SCHEDULING_SCHEDULING_H diff --git a/src/ray/common/scheduling/scheduling_test.cc b/src/ray/common/scheduling/scheduling_test.cc index 0731a5267..1edece241 100644 --- a/src/ray/common/scheduling/scheduling_test.cc +++ b/src/ray/common/scheduling/scheduling_test.cc @@ -13,13 +13,14 @@ using namespace std; -/// Used to path empty vector argiuments. +// Used to path empty vector argiuments. vector EmptyIntVector; vector EmptyBoolVector; +vector EmptyDoubleVector; -void initTaskRequest(TaskRequest &tr, vector &pred_demands, +void initTaskRequest(TaskRequest &tr, vector &pred_demands, vector &pred_soft, vector &cust_ids, - vector &cust_demands, vector &cust_soft, + vector &cust_demands, vector &cust_soft, vector &placement_hints) { for (size_t i = 0; i < pred_demands.size(); i++) { ResourceRequest rq; @@ -38,8 +39,8 @@ void initTaskRequest(TaskRequest &tr, vector &pred_demands, for (size_t i = 0; i < cust_ids.size(); i++) { ResourceRequestWithId rq; rq.id = cust_ids[i]; - rq.req.demand = cust_demands[i]; - rq.req.soft = cust_soft[i]; + rq.demand = cust_demands[i]; + rq.soft = cust_soft[i]; tr.custom_resources.push_back(rq); } @@ -48,19 +49,32 @@ void initTaskRequest(TaskRequest &tr, vector &pred_demands, } }; +void addTaskResourceInstances(bool predefined, vector allocation, uint64_t idx, + TaskResourceInstances *task_allocation) { + if (task_allocation->predefined_resources.size() < PredefinedResources_MAX) { + task_allocation->predefined_resources.resize(PredefinedResources_MAX); + } + if (predefined) { + task_allocation->predefined_resources[idx] = allocation; + } else { + task_allocation->custom_resources.insert( + std::pair>(idx, allocation)); + } +}; + void initNodeResources(NodeResources &node, vector &pred_capacities, vector &cust_ids, vector &cust_capacities) { for (size_t i = 0; i < pred_capacities.size(); i++) { ResourceCapacity rc; rc.total = rc.available = pred_capacities[i]; - node.capacities.push_back(rc); + node.predefined_resources.push_back(rc); } if (pred_capacities.size() < PredefinedResources_MAX) { for (int i = pred_capacities.size(); i < PredefinedResources_MAX; i++) { ResourceCapacity rc; rc.total = rc.available = 0; - node.capacities.push_back(rc); + node.predefined_resources.push_back(rc); } } @@ -105,16 +119,17 @@ void initCluster(ClusterResourceScheduler &cluster_resources, int n) { } bool nodeResourcesEqual(const NodeResources &nr1, const NodeResources &nr2) { - if (nr1.capacities.size() != nr2.capacities.size()) { - cout << nr1.capacities.size() << " " << nr2.capacities.size() << endl; + if (nr1.predefined_resources.size() != nr2.predefined_resources.size()) { + cout << nr1.predefined_resources.size() << " " << nr2.predefined_resources.size() + << endl; return false; } - for (size_t i = 0; i < nr1.capacities.size(); i++) { - if (nr1.capacities[i].available != nr2.capacities[i].available) { + for (size_t i = 0; i < nr1.predefined_resources.size(); i++) { + if (nr1.predefined_resources[i].available != nr2.predefined_resources[i].available) { return false; } - if (nr1.capacities[i].total != nr2.capacities[i].total) { + if (nr1.predefined_resources[i].total != nr2.predefined_resources[i].total) { return false; } } @@ -169,7 +184,7 @@ TEST_F(SchedulingTest, SchedulingIdTest) { ASSERT_TRUE(ids.Get(to_string(100)) == -1); - /// Test for handling collision. + // Test for handling collision. StringIdMap short_ids; uint8_t max_id = 8; for (size_t i = 0; i < max_id; i++) { @@ -235,7 +250,7 @@ TEST_F(SchedulingTest, SchedulingModifyClusterNodeTest) { } TEST_F(SchedulingTest, SchedulingUpdateAvailableResourcesTest) { - /// Create cluster resources. + // Create cluster resources. NodeResources node_resources; vector pred_capacities{10, 5, 3}; vector cust_ids{1, 2}; @@ -246,10 +261,10 @@ TEST_F(SchedulingTest, SchedulingUpdateAvailableResourcesTest) { { TaskRequest task_req; #define PRED_CUSTOM_LEN 2 - vector pred_demands{7, 7}; + vector pred_demands{7, 7}; vector pred_soft{false, true}; vector cust_ids{1, 2}; - vector cust_demands{3, 10}; + vector cust_demands{3, 10}; vector cust_soft{false, true}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, EmptyIntVector); @@ -264,9 +279,10 @@ TEST_F(SchedulingTest, SchedulingUpdateAvailableResourcesTest) { ASSERT_TRUE(cluster_resources.GetNodeResources(node_id, &nr2)); for (size_t i = 0; i < PRED_CUSTOM_LEN; i++) { - int64_t t = nr1.capacities[i].available - task_req.predefined_resources[i].demand; + int64_t t = + nr1.predefined_resources[i].available - task_req.predefined_resources[i].demand; if (t < 0) t = 0; - ASSERT_EQ(nr2.capacities[i].available, t); + ASSERT_EQ(nr2.predefined_resources[i].available, t); } for (size_t i = 0; i < PRED_CUSTOM_LEN; i++) { @@ -274,7 +290,7 @@ TEST_F(SchedulingTest, SchedulingUpdateAvailableResourcesTest) { if (it1 != nr1.custom_resources.end()) { auto it2 = nr2.custom_resources.find(task_req.custom_resources[i].id); if (it2 != nr2.custom_resources.end()) { - int64_t t = it1->second.available - task_req.custom_resources[i].req.demand; + int64_t t = it1->second.available - task_req.custom_resources[i].demand; if (t < 0) t = 0; ASSERT_EQ(it2->second.available, t); } @@ -288,7 +304,7 @@ TEST_F(SchedulingTest, SchedulingAddOrUpdateNodeTest) { NodeResources nr, nr_out; int64_t node_id = 1; - /// Add node. + // Add node. { NodeResources node_resources; vector pred_capacities{10, 5, 3}; @@ -299,14 +315,14 @@ TEST_F(SchedulingTest, SchedulingAddOrUpdateNodeTest) { nr = node_resources; } - /// Check whether node resources were correctly added. + // Check whether node resources were correctly added. if (cluster_resources.GetNodeResources(node_id, &nr_out)) { ASSERT_TRUE(nodeResourcesEqual(nr, nr_out)); } else { ASSERT_TRUE(false); } - /// Update node. + // Update node. { NodeResources node_resources; vector pred_capacities{10, 10}; @@ -324,7 +340,7 @@ TEST_F(SchedulingTest, SchedulingAddOrUpdateNodeTest) { } TEST_F(SchedulingTest, SchedulingTaskRequestTest) { - /// Create cluster resources containing local node. + // Create cluster resources containing local node. NodeResources node_resources; vector pred_capacities{5, 5}; vector cust_ids{1}; @@ -332,6 +348,8 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities); ClusterResourceScheduler cluster_resources(0, node_resources); + std::cerr << "XXXXXXXXXXX" << std::endl; + { NodeResources node_resources; vector pred_capacities{10, 2, 3}; @@ -340,23 +358,23 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities); cluster_resources.AddOrUpdateNode(1, node_resources); } - /// Predefined resources, hard constraint violation + // Predefined resources, hard constraint violation { TaskRequest task_req; - vector pred_demands = {11}; + vector pred_demands = {11}; vector pred_soft = {false}; - initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyIntVector, + initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyDoubleVector, EmptyBoolVector, EmptyIntVector); int64_t violations; int64_t node_id = cluster_resources.GetBestSchedulableNode(task_req, &violations); ASSERT_EQ(node_id, -1); } - /// Predefined resources, soft constraint violation + // Predefined resources, soft constraint violation { TaskRequest task_req; - vector pred_demands = {11}; + vector pred_demands = {11}; vector pred_soft = {true}; - initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyIntVector, + initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyDoubleVector, EmptyBoolVector, EmptyIntVector); int64_t violations; int64_t node_id = cluster_resources.GetBestSchedulableNode(task_req, &violations); @@ -364,25 +382,25 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { ASSERT_TRUE(violations > 0); } - /// Predefined resources, no constraint violation. + // Predefined resources, no constraint violation. { TaskRequest task_req; - vector pred_demands = {5}; + vector pred_demands = {5}; vector pred_soft = {false}; - initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyIntVector, + initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyDoubleVector, EmptyBoolVector, EmptyIntVector); int64_t violations; int64_t node_id = cluster_resources.GetBestSchedulableNode(task_req, &violations); ASSERT_TRUE(node_id != -1); ASSERT_TRUE(violations == 0); } - /// Custom resources, hard constraint violation. + // Custom resources, hard constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{1}; - vector cust_demands{11}; + vector cust_demands{11}; vector cust_soft{false}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, EmptyIntVector); @@ -390,13 +408,13 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { int64_t node_id = cluster_resources.GetBestSchedulableNode(task_req, &violations); ASSERT_TRUE(node_id == -1); } - /// Custom resources, soft constraint violation. + // Custom resources, soft constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{1}; - vector cust_demands{11}; + vector cust_demands{11}; vector cust_soft{true}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, EmptyIntVector); @@ -405,13 +423,13 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { ASSERT_TRUE(node_id != -1); ASSERT_TRUE(violations > 0); } - /// Custom resources, no constraint violation. + // Custom resources, no constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{1}; - vector cust_demands{5}; + vector cust_demands{5}; vector cust_soft{false}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, EmptyIntVector); @@ -420,13 +438,13 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { ASSERT_TRUE(node_id != -1); ASSERT_TRUE(violations == 0); } - /// Custom resource missing, hard constraint violation. + // Custom resource missing, hard constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{100}; - vector cust_demands{5}; + vector cust_demands{5}; vector cust_soft{false}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, EmptyIntVector); @@ -434,13 +452,13 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { int64_t node_id = cluster_resources.GetBestSchedulableNode(task_req, &violations); ASSERT_TRUE(node_id == -1); } - /// Custom resource missing, soft constraint violation. + // Custom resource missing, soft constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{100}; - vector cust_demands{5}; + vector cust_demands{5}; vector cust_soft{true}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, EmptyIntVector); @@ -449,13 +467,13 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { ASSERT_TRUE(node_id != -1); ASSERT_TRUE(violations > 0); } - /// Placement_hints, soft constraint violation. + // Placement_hints, soft constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{1}; - vector cust_demands{5}; + vector cust_demands{5}; vector cust_soft{true}; vector placement_hints{2, 3}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, @@ -465,13 +483,13 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { ASSERT_TRUE(node_id != -1); ASSERT_TRUE(violations > 0); } - /// Placement hints, no constraint violation. + // Placement hints, no constraint violation. { TaskRequest task_req; - vector pred_demands{5, 2}; + vector pred_demands{5, 2}; vector pred_soft{false, true}; vector cust_ids{1}; - vector cust_demands{5}; + vector cust_demands{5}; vector cust_soft{true}; vector placement_hints{1, 2, 3}; initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, @@ -483,6 +501,243 @@ TEST_F(SchedulingTest, SchedulingTaskRequestTest) { } } +TEST_F(SchedulingTest, GetLocalAvailableResourcesTest) { + // Create cluster resources containing local node. + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + vector cust_ids{1}; + vector cust_capacities{8}; + initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskResourceInstances available_cluster_resources = + cluster_resources.GetLocalResources().GetAvailableResourceInstances(); + + TaskResourceInstances expected_cluster_resources; + addTaskResourceInstances(true, {1., 1., 1.}, 0, &expected_cluster_resources); + addTaskResourceInstances(true, {4.}, 1, &expected_cluster_resources); + addTaskResourceInstances(true, {1., 1., 1., 1., 1.}, 2, &expected_cluster_resources); + + ASSERT_EQ(expected_cluster_resources == available_cluster_resources, false); + + addTaskResourceInstances(false, {8.}, 1, &expected_cluster_resources); + + ASSERT_EQ(expected_cluster_resources == available_cluster_resources, true); +} + +TEST_F(SchedulingTest, GetCPUInstancesTest) { + TaskResourceInstances task_resources; + addTaskResourceInstances(true, {1., 1., 1.}, CPU, &task_resources); + addTaskResourceInstances(true, {4.}, MEM, &task_resources); + addTaskResourceInstances(true, {1., 1., 1., 1., 1.}, GPU, &task_resources); + + std::vector cpu_instances = task_resources.GetCPUInstances(); + std::vector expected_cpu_instances{1., 1., 1.}; + + ASSERT_EQ(EqualVectors(cpu_instances, expected_cpu_instances), true); +} + +TEST_F(SchedulingTest, AvailableResourceInstancesOpsTest) { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */}; + initNodeResources(node_resources, pred_capacities, EmptyIntVector, EmptyIntVector); + ClusterResourceScheduler cluster(0, node_resources); + + ResourceInstanceCapacities instances; + + instances.total = {6., 6., 6.}; + instances.available = {3., 2., 5.}; + ResourceInstanceCapacities old_instances = instances; + + std::vector a{1., 1., 1.}; + cluster.AddAvailableResourceInstances(a, &instances); + cluster.SubtractAvailableResourceInstances(a, &instances); + + ASSERT_EQ(EqualVectors(instances.available, old_instances.available), true); + + a = {10., 1., 1.}; + cluster.AddAvailableResourceInstances(a, &instances); + std::vector expected_available{6., 3., 6.}; + + ASSERT_EQ(EqualVectors(instances.available, expected_available), true); + + a = {10., 1., 1.}; + cluster.SubtractAvailableResourceInstances(a, &instances); + expected_available = {0., 2., 5.}; + ASSERT_EQ(EqualVectors(instances.available, expected_available), true); +} + +TEST_F(SchedulingTest, TaskResourceInstancesTest) { + // Allocate resources for a task request specifying only predefined resources. + { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + initNodeResources(node_resources, pred_capacities, EmptyIntVector, EmptyIntVector); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskRequest task_req; + vector pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */}; + vector pred_soft = {false}; + initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyDoubleVector, + EmptyBoolVector, EmptyIntVector); + + NodeResourceInstances old_local_resources = cluster_resources.GetLocalResources(); + TaskResourceInstances task_allocation; + bool success = + cluster_resources.AllocateTaskResourceInstances(task_req, &task_allocation); + + ASSERT_EQ(success, true); + + cluster_resources.FreeTaskResourceInstances(task_allocation); + + ASSERT_EQ((cluster_resources.GetLocalResources() == old_local_resources), true); + } + // Try to allocate resources for a task request that overallocates a hard constrained + // resource. + { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + initNodeResources(node_resources, pred_capacities, EmptyIntVector, EmptyIntVector); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskRequest task_req; + vector pred_demands = {4. /* CPU */, 2. /* MEM */, 1.5 /* GPU */}; + vector pred_soft = {false}; // Hard constrained resource. + initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyDoubleVector, + EmptyBoolVector, EmptyIntVector); + + NodeResourceInstances old_local_resources = cluster_resources.GetLocalResources(); + TaskResourceInstances task_allocation; + bool success = + cluster_resources.AllocateTaskResourceInstances(task_req, &task_allocation); + + ASSERT_EQ(success, false); + ASSERT_EQ((cluster_resources.GetLocalResources() == old_local_resources), true); + } + // Allocate resources for a task request that overallocates a soft constrained resource. + { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + initNodeResources(node_resources, pred_capacities, EmptyIntVector, EmptyIntVector); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskRequest task_req; + vector pred_demands = {4. /* CPU */, 2. /* MEM */, 1.5 /* GPU */}; + vector pred_soft = {true}; // Soft constrained resource. + initTaskRequest(task_req, pred_demands, pred_soft, EmptyIntVector, EmptyDoubleVector, + EmptyBoolVector, EmptyIntVector); + + NodeResourceInstances old_local_resources = cluster_resources.GetLocalResources(); + TaskResourceInstances task_allocation; + bool success = + cluster_resources.AllocateTaskResourceInstances(task_req, &task_allocation); + + ASSERT_EQ(success, true); + + TaskResourceInstances expected_task_allocation; + addTaskResourceInstances(true, {0., 0., 0.}, CPU, &expected_task_allocation); + addTaskResourceInstances(true, {2.}, MEM, &expected_task_allocation); + addTaskResourceInstances(true, {0., 0.5, 1., 1., 1.}, GPU, &expected_task_allocation); + + TaskResourceInstances local_available_resources = + cluster_resources.GetLocalResources().GetAvailableResourceInstances(); + + ASSERT_EQ((local_available_resources == expected_task_allocation), true); + } + // Allocate resources for a task request specifying both predefined and custom + // resources. + { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + vector cust_ids{1, 2}; + vector cust_capacities{4, 4}; + initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskRequest task_req; + vector pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */}; + vector pred_soft = {false}; + vector cust_demands{3, 2}; + vector cust_soft{false, false}; + initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, + EmptyIntVector); + + NodeResourceInstances old_local_resources = cluster_resources.GetLocalResources(); + TaskResourceInstances task_allocation; + bool success = + cluster_resources.AllocateTaskResourceInstances(task_req, &task_allocation); + + ASSERT_EQ(success, true); + + cluster_resources.FreeTaskResourceInstances(task_allocation); + + ASSERT_EQ((cluster_resources.GetLocalResources() == old_local_resources), true); + } + // Allocate resources for a task request specifying both predefined and custom + // resources, but overallocates a hard-constrained custom resource. + { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + vector cust_ids{1, 2}; + vector cust_capacities{4, 4}; + initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskRequest task_req; + vector pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */}; + vector pred_soft = {false}; + vector cust_demands{3, 10}; + vector cust_soft{false, false}; + initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, + EmptyIntVector); + + NodeResourceInstances old_local_resources = cluster_resources.GetLocalResources(); + TaskResourceInstances task_allocation; + bool success = + cluster_resources.AllocateTaskResourceInstances(task_req, &task_allocation); + + ASSERT_EQ(success, false); + ASSERT_EQ((cluster_resources.GetLocalResources() == old_local_resources), true); + } + // Allocate resources for a task request specifying both predefined and custom + // resources, but overallocates a soft-constrained custom resource. + { + NodeResources node_resources; + vector pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */}; + vector cust_ids{1, 2}; + vector cust_capacities{4, 4}; + initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities); + ClusterResourceScheduler cluster_resources(0, node_resources); + + TaskRequest task_req; + vector pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */}; + vector pred_soft = {false}; + vector cust_demands{3, 10}; + vector cust_soft{false, true}; + initTaskRequest(task_req, pred_demands, pred_soft, cust_ids, cust_demands, cust_soft, + EmptyIntVector); + + NodeResourceInstances old_local_resources = cluster_resources.GetLocalResources(); + TaskResourceInstances task_allocation; + bool success = + cluster_resources.AllocateTaskResourceInstances(task_req, &task_allocation); + + ASSERT_EQ(success, true); + + TaskResourceInstances expected_task_allocation; + addTaskResourceInstances(true, {0., 0., 0.}, CPU, &expected_task_allocation); + addTaskResourceInstances(true, {2.}, MEM, &expected_task_allocation); + addTaskResourceInstances(true, {0., 0.5, 1., 1., 1.}, GPU, &expected_task_allocation); + addTaskResourceInstances(false, {1.}, 1, &expected_task_allocation); + addTaskResourceInstances(false, {0.}, 2, &expected_task_allocation); + + TaskResourceInstances local_available_resources = + cluster_resources.GetLocalResources().GetAvailableResourceInstances(); + + ASSERT_EQ((local_available_resources == expected_task_allocation), true); + } +} + #ifdef UNORDERED_VS_ABSL_MAPS_EVALUATION TEST_F(SchedulingTest, SchedulingMapPerformanceTest) { size_t map_len = 1000000; @@ -503,10 +758,6 @@ TEST_F(SchedulingTest, SchedulingMapPerformanceTest) { amap_string_key.emplace(to_string(i), i); } - for (size_t i = 0; i < 25; i++) { - cout << "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" << endl; - } - int64_t sum; auto t_start = std::chrono::high_resolution_clock::now();