Support max pods allowed per node

Author: islinwb

src/base/resource_desc.proto

@@ -62,4 +62,6 @@ message ResourceDescriptor {
   uint64 trace_machine_id = 21;
   // Resource labels
   repeated Label labels = 32;
+  // Max pods allowed per node
+  uint64 max_pods = 33;
 }

src/scheduling/flow/cpu_cost_model.cc

@@ -82,7 +82,10 @@ ArcDescriptor CpuCostModel::ResourceNodeToResourceNode(
 }
 
 ArcDescriptor CpuCostModel::LeafResourceNodeToSink(ResourceID_t resource_id) {
-  return ArcDescriptor(0LL, FLAGS_max_tasks_per_pu, 0ULL);
+  //ResourceID_t machine_res_id = MachineResIDForResource(resource_id);
+  ResourceStatus* rs = FindPtrOrNull(*resource_map_, resource_id);
+  ResourceTopologyNodeDescriptor* rtnd = rs->mutable_topology_node();
+  return ArcDescriptor(0LL, rtnd->resource_desc().num_slots_below(), 0ULL);
 }
 
 ArcDescriptor CpuCostModel::TaskContinuation(TaskID_t task_id) {
@@ -822,7 +825,7 @@ void CpuCostModel::AddMachine(ResourceTopologyNodeDescriptor* rtnd_ptr) {
   CHECK(rd.type() == ResourceDescriptor::RESOURCE_MACHINE);
   ResourceID_t res_id = ResourceIDFromString(rd.uuid());
   vector<EquivClass_t> machine_ecs;
-  for (uint64_t index = 0; index < FLAGS_max_multi_arcs_for_cpu; ++index) {
+  for (uint64_t index = 0; index < rd.max_pods(); ++index) {
     EquivClass_t multi_machine_ec = GetMachineEC(rd.friendly_name(), index);
     machine_ecs.push_back(multi_machine_ec);
     CHECK(InsertIfNotPresent(&ec_to_index_, multi_machine_ec, index));
@@ -898,7 +901,9 @@ FlowGraphNode* CpuCostModel::GatherStats(FlowGraphNode* accumulator,
       }
       // Running/idle task count
       rd_ptr->set_num_running_tasks_below(rd_ptr->current_running_tasks_size());
-      rd_ptr->set_num_slots_below(FLAGS_max_tasks_per_pu);
+      ResourceStatus* m_rs = FindPtrOrNull(*resource_map_, machine_res_id);
+      ResourceTopologyNodeDescriptor* m_rtnd = m_rs->mutable_topology_node();
+      rd_ptr->set_num_slots_below(m_rtnd->resource_desc().max_pods());
       return accumulator;
     }
   } else if (accumulator->type_ == FlowNodeType::MACHINE) {