fix new scheduler bug

src/ray/raylet/node_manager.cc

@@ -671,6 +671,15 @@ void NodeManager::NodeRemoved(const GcsNodeInfo &node_info) {
  return;
  }
 
+ // Remove the client from the resource map.
+ if (new_scheduler_enabled_) {
+ if (!new_resource_scheduler_->RemoveNode(node_id.Binary())) {
+ RAY_LOG(DEBUG) << "Received NodeRemoved callback for an unknown node: " << node_id
+ << ".";
+ return;
+ }
+ }
+
  // Remove the node manager client.
  const auto client_entry = remote_node_manager_clients_.find(node_id);
  if (client_entry != remote_node_manager_clients_.end()) {

src/ray/raylet/scheduling/cluster_resource_scheduler.cc

@@ -494,6 +494,13 @@ bool ClusterResourceScheduler::RemoveNode(int64_t node_id) {
  }
 }
 
+bool ClusterResourceScheduler::RemoveNode(const std::string &node_id_string) {
+ auto node_id = string_to_int_map_.Get(node_id_string);
+ RAY_CHECK(node_id != -1);
+
+ return RemoveNode(node_id);
+}
+
 int64_t ClusterResourceScheduler::IsSchedulable(const TaskRequest &task_req,
  int64_t node_id,
  const NodeResources &resources) {
@@ -929,7 +936,7 @@ bool ClusterResourceScheduler::AllocateResourceInstances(
  (*allocation)[i] = remaining_demand;
  return true;
  } else {
- (*allocation)[i] = available[i];
+ (*allocation)[i] += available[i];
  remaining_demand -= available[i];
  available[i] = 0;
  }

src/ray/raylet/scheduling/cluster_resource_scheduler.h

@@ -250,6 +250,7 @@ class ClusterResourceScheduler {
  ///
  /// \param ID of the node to be removed.
  bool RemoveNode(int64_t node_id);
+ bool RemoveNode(const std::string &node_id_string);
 
  /// Check whether a task request can be scheduled given a node.
  ///