wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-22 13:00:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

New scheduler local node (#6913)

Browse Source 
Co-authored-by: Edward Oakes <ed.nmi.oakes@gmail.com>
This commit is contained in:
Ion
2020-02-07 12:57:59 -08:00
committed by GitHub
co-authored by Edward Oakes
parent 676a011784
commit f9885b8710
3 changed files with 910 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/common/scheduling/cluster_resource_scheduler.cc
+440 -73
View File
@@ -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<size_t>(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<double> &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<double> &v1, const std::vector<double> &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<int64_t>(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<int64_t>(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<std::string, double> &resource_map_total,
const std::unordered_map<std::string, double> &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<size_t>(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<double> 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<double> 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<double> &available,
std::vector<double> *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<int64_t>(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<double> 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<double> &cpu_instances) {
AddAvailableResourceInstances(cpu_instances,
&local_resources_.predefined_resources[CPU]);
}
void ClusterResourceScheduler::SubtractCPUResourceInstances(
std::vector<double> &cpu_instances) {
SubtractAvailableResourceInstances(cpu_instances,
&local_resources_.predefined_resources[CPU]);
}
src/ray/common/scheduling/cluster_resource_scheduler.h
+162 -27
View File
@@ -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<int64_t> UnitInstanceResources{CPU, GPU, TPU};
// Helper function to compare two vectors with double values.
bool EqualVectors(const std::vector<double> &v1, const std::vector<double> &v2);
struct ResourceCapacity {
int64_t total;
int64_t available;
double total;
double available;
};
/// Capacities of each instance of a resource.
struct ResourceInstanceCapacities {
std::vector<double> total;
std::vector<double> 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<ResourceCapacity> capacities;
/// Map containing custom resources. The key of each entry represents the
/// custom resource ID.
absl::flat_hash_map<int64_t, ResourceCapacity> 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<ResourceRequest> 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<std::vector<double>> predefined_resources;
/// The list of instances of each custom resource allocated to a task.
absl::flat_hash_map<int64_t, std::vector<double>> custom_resources;
bool operator==(const TaskResourceInstances &other);
/// Get CPU instances only.
std::vector<double> 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<ResourceCapacity> predefined_resources;
/// Map containing custom resources. The key of each entry represents the
/// custom resource ID.
absl::flat_hash_map<int64_t, ResourceCapacity> 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<ResourceInstanceCapacities> predefined_resources;
/// Map containing custom resources. The key of each entry represents the
/// custom resource ID.
absl::flat_hash_map<int64_t, ResourceInstanceCapacities> 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<int64_t, NodeResources> 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<int64_t, ResourceCapacity> &new_custom_resources,
absl::flat_hash_map<int64_t, ResourceCapacity> *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<std::string, double> &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<double> &available,
std::vector<double> *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<double> 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<double> 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<double> &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<double> &cpu_instances);
/// Return human-readable string for this scheduler state.
std::string DebugString();
};
#endif // RAY_COMMON_SCHEDULING_SCHEDULING_H
src/ray/common/scheduling/scheduling_test.cc
+308 -57
View File
@@ -13,13 +13,14 @@
using namespace std;
/// Used to path empty vector argiuments.
// Used to path empty vector argiuments.
vector<int64_t> EmptyIntVector;
vector<bool> EmptyBoolVector;
vector<double> EmptyDoubleVector;
void initTaskRequest(TaskRequest &tr, vector<int64_t> &pred_demands,
void initTaskRequest(TaskRequest &tr, vector<double> &pred_demands,
vector<bool> &pred_soft, vector<int64_t> &cust_ids,
vector<int64_t> &cust_demands, vector<bool> &cust_soft,
vector<double> &cust_demands, vector<bool> &cust_soft,
vector<int64_t> &placement_hints) {
for (size_t i = 0; i < pred_demands.size(); i++) {
ResourceRequest rq;
@@ -38,8 +39,8 @@ void initTaskRequest(TaskRequest &tr, vector<int64_t> &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<int64_t> &pred_demands,
}
};
void addTaskResourceInstances(bool predefined, vector<double> 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<int64_t, vector<double>>(idx, allocation));
}
};
void initNodeResources(NodeResources &node, vector<int64_t> &pred_capacities,
vector<int64_t> &cust_ids, vector<int64_t> &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<int64_t> pred_capacities{10, 5, 3};
vector<int64_t> cust_ids{1, 2};
@@ -246,10 +261,10 @@ TEST_F(SchedulingTest, SchedulingUpdateAvailableResourcesTest) {
{
TaskRequest task_req;
#define PRED_CUSTOM_LEN 2
vector<int64_t> pred_demands{7, 7};
vector<double> pred_demands{7, 7};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{1, 2};
vector<int64_t> cust_demands{3, 10};
vector<double> cust_demands{3, 10};
vector<bool> 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<int64_t> 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<int64_t> 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<int64_t> pred_capacities{5, 5};
vector<int64_t> 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<int64_t> 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<int64_t> pred_demands = {11};
vector<double> pred_demands = {11};
vector<bool> 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<int64_t> pred_demands = {11};
vector<double> pred_demands = {11};
vector<bool> 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<int64_t> pred_demands = {5};
vector<double> pred_demands = {5};
vector<bool> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{1};
vector<int64_t> cust_demands{11};
vector<double> cust_demands{11};
vector<bool> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{1};
vector<int64_t> cust_demands{11};
vector<double> cust_demands{11};
vector<bool> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{1};
vector<int64_t> cust_demands{5};
vector<double> cust_demands{5};
vector<bool> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{100};
vector<int64_t> cust_demands{5};
vector<double> cust_demands{5};
vector<bool> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{100};
vector<int64_t> cust_demands{5};
vector<double> cust_demands{5};
vector<bool> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{1};
vector<int64_t> cust_demands{5};
vector<double> cust_demands{5};
vector<bool> cust_soft{true};
vector<int64_t> 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<int64_t> pred_demands{5, 2};
vector<double> pred_demands{5, 2};
vector<bool> pred_soft{false, true};
vector<int64_t> cust_ids{1};
vector<int64_t> cust_demands{5};
vector<double> cust_demands{5};
vector<bool> cust_soft{true};
vector<int64_t> 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<int64_t> pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */};
vector<int64_t> cust_ids{1};
vector<int64_t> 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<double> cpu_instances = task_resources.GetCPUInstances();
std::vector<double> expected_cpu_instances{1., 1., 1.};
ASSERT_EQ(EqualVectors(cpu_instances, expected_cpu_instances), true);
}
TEST_F(SchedulingTest, AvailableResourceInstancesOpsTest) {
NodeResources node_resources;
vector<int64_t> 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<double> 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<double> 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<int64_t> 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<double> pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */};
vector<bool> 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<int64_t> 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<double> pred_demands = {4. /* CPU */, 2. /* MEM */, 1.5 /* GPU */};
vector<bool> 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<int64_t> 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<double> pred_demands = {4. /* CPU */, 2. /* MEM */, 1.5 /* GPU */};
vector<bool> 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<int64_t> pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */};
vector<int64_t> cust_ids{1, 2};
vector<int64_t> cust_capacities{4, 4};
initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities);
ClusterResourceScheduler cluster_resources(0, node_resources);
TaskRequest task_req;
vector<double> pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */};
vector<bool> pred_soft = {false};
vector<double> cust_demands{3, 2};
vector<bool> 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<int64_t> pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */};
vector<int64_t> cust_ids{1, 2};
vector<int64_t> cust_capacities{4, 4};
initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities);
ClusterResourceScheduler cluster_resources(0, node_resources);
TaskRequest task_req;
vector<double> pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */};
vector<bool> pred_soft = {false};
vector<double> cust_demands{3, 10};
vector<bool> 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<int64_t> pred_capacities{3 /* CPU */, 4 /* MEM */, 5 /* GPU */};
vector<int64_t> cust_ids{1, 2};
vector<int64_t> cust_capacities{4, 4};
initNodeResources(node_resources, pred_capacities, cust_ids, cust_capacities);
ClusterResourceScheduler cluster_resources(0, node_resources);
TaskRequest task_req;
vector<double> pred_demands = {3. /* CPU */, 2. /* MEM */, 1.5 /* GPU */};
vector<bool> pred_soft = {false};
vector<double> cust_demands{3, 10};
vector<bool> 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();