We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[autoscaler] Inconsistent reporting of worker node types
On master, it seems I have a cluster with a cpu_4_ondemand head node and cpu_64_ondemand worker (68 CPUs in total), but somehow the debug string is reporting 2 workers of cpu_64_ondemand (which is wrong, since there is only 1 worker node, and the head node is cpu_4_ondemand).

YAML: https://gist.github.com/ericl/f9c32b8bbbe4e1aed43899832730f70d (switching the Ray version to latest wheel)

```
 - ResourceUsage: 0.0/68.0 CPU, 0.0/1.0 Custom1, 0.0 GiB/198.65 GiB memory, 0.0
GiB/59.19 GiB object_store_memory
 - TimeSinceLastHeartbeat: Min=0 Mean=0 Max=0
Worker node types:
 - cpu_64_ondemand: 2

```

cc @wuisawesome 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/autoscaler/resource_demand_scheduler.py`
Content:
```
1 """Implements multi-node-type autoscaling.
2 
3 This file implements an autoscaling algorithm that is aware of multiple node
4 types (e.g., example-multi-node-type.yaml). The Ray autoscaler will pass in
5 a vector of resource shape demands, and the resource demand scheduler will
6 return a list of node types that can satisfy the demands given constraints
7 (i.e., reverse bin packing).
8 """
9 
10 import copy
11 import numpy as np
12 import logging
13 import collections
14 from typing import List, Dict, Tuple
15 
16 from ray.autoscaler.node_provider import NodeProvider
17 from ray.autoscaler.tags import TAG_RAY_USER_NODE_TYPE, NODE_KIND_UNMANAGED
18 
19 logger = logging.getLogger(__name__)
20 
21 # e.g., m4.16xlarge.
22 NodeType = str
23 
24 # e.g., {"resources": ..., "max_workers": ...}.
25 NodeTypeConfigDict = str
26 
27 # e.g., {"GPU": 1}.
28 ResourceDict = str
29 
30 # e.g., IP address of the node.
31 NodeID = str
32 
33 
34 class ResourceDemandScheduler:
35     def __init__(self, provider: NodeProvider,
36                  node_types: Dict[NodeType, NodeTypeConfigDict],
37                  max_workers: int):
38         self.provider = provider
39         self.node_types = node_types
40         self.max_workers = max_workers
41 
42     def get_nodes_to_launch(self, nodes: List[NodeID],
43                             pending_nodes: Dict[NodeType, int],
44                             resource_demands: List[ResourceDict],
45                             usage_by_ip: Dict[str, ResourceDict]
46                             ) -> List[Tuple[NodeType, int]]:
47         """Given resource demands, return node types to add to the cluster.
48 
49         This method:
50             (1) calculates the resources present in the cluster.
51             (2) calculates the unfulfilled resource bundles.
52             (3) calculates which nodes need to be launched to fulfill all
53                 the bundle requests, subject to max_worker constraints.
54 
55         Args:
56             nodes: List of existing nodes in the cluster.
57             pending_nodes: Summary of node types currently being launched.
58             resource_demands: Vector of resource demands from the scheduler.
59         """
60 
61         if resource_demands is None:
62             logger.info("No resource demands")
63             return []
64 
65         node_resources, node_type_counts = self.calculate_node_resources(
66             nodes, pending_nodes, usage_by_ip)
67 
68         logger.info("Cluster resources: {}".format(node_resources))
69         logger.info("Node counts: {}".format(node_type_counts))
70 
71         unfulfilled = get_bin_pack_residual(node_resources, resource_demands)
72         logger.info("Resource demands: {}".format(resource_demands))
73         logger.info("Unfulfilled demands: {}".format(unfulfilled))
74 
75         nodes = get_nodes_for(
76             self.node_types, node_type_counts,
77             self.max_workers - len(nodes) - sum(pending_nodes.values()),
78             unfulfilled)
79         logger.info("Node requests: {}".format(nodes))
80         return nodes
81 
82     def calculate_node_resources(
83             self, nodes: List[NodeID], pending_nodes: Dict[NodeID, int],
84             usage_by_ip: Dict[str, ResourceDict]
85     ) -> (List[ResourceDict], Dict[NodeType, int]):
86         """Returns node resource list and node type counts."""
87 
88         node_resources = []
89         node_type_counts = collections.defaultdict(int)
90 
91         def add_node(node_type, available_resources=None):
92             if node_type not in self.node_types:
93                 logger.warn(
94                     f"Missing entry for node_type {node_type} in "
95                     f"cluster config: {self.node_types} under entry "
96                     f"available_node_types. This node's resources will be "
97                     f"ignored. If you are using an unmanaged node, manually "
98                     f"set the user_node_type tag to \"{NODE_KIND_UNMANAGED}\""
99                     f"in your cloud provider's management console.")
100                 return None
101             # Careful not to include the same dict object multiple times.
102             available = copy.deepcopy(self.node_types[node_type]["resources"])
103             # If available_resources is None this might be because the node is
104             # no longer pending, but the raylet hasn't sent a heartbeat to gcs
105             # yet.
106             if available_resources is not None:
107                 available = copy.deepcopy(available_resources)
108 
109             node_resources.append(available)
110             node_type_counts[node_type] += 1
111 
112         for node_id in nodes:
113             tags = self.provider.node_tags(node_id)
114             if TAG_RAY_USER_NODE_TYPE in tags:
115                 node_type = tags[TAG_RAY_USER_NODE_TYPE]
116                 node_type_counts[node_type] += 1
117                 ip = self.provider.internal_ip(node_id)
118                 available_resources = usage_by_ip.get(ip)
119                 add_node(node_type, available_resources)
120 
121         for node_type, count in pending_nodes.items():
122             for _ in range(count):
123                 add_node(node_type)
124 
125         return node_resources, node_type_counts
126 
127     def debug_string(self, nodes: List[NodeID],
128                      pending_nodes: Dict[NodeID, int],
129                      usage_by_ip: Dict[str, ResourceDict]) -> str:
130         print(f"{usage_by_ip}")
131         node_resources, node_type_counts = self.calculate_node_resources(
132             nodes, pending_nodes, usage_by_ip)
133 
134         out = "Worker node types:"
135         for node_type, count in node_type_counts.items():
136             out += "\n - {}: {}".format(node_type, count)
137             if pending_nodes.get(node_type):
138                 out += " ({} pending)".format(pending_nodes[node_type])
139 
140         return out
141 
142 
143 def get_nodes_for(node_types: Dict[NodeType, NodeTypeConfigDict],
144                   existing_nodes: Dict[NodeType, int], max_to_add: int,
145                   resources: List[ResourceDict]) -> List[Tuple[NodeType, int]]:
146     """Determine nodes to add given resource demands and constraints.
147 
148     Args:
149         node_types: node types config.
150         existing_nodes: counts of existing nodes already launched.
151             This sets constraints on the number of new nodes to add.
152         max_to_add: global constraint on nodes to add.
153         resources: resource demands to fulfill.
154 
155     Returns:
156         List of nodes types and count to add.
157     """
158     nodes_to_add = collections.defaultdict(int)
159     allocated_resources = []
160 
161     while resources and sum(nodes_to_add.values()) < max_to_add:
162         utilization_scores = []
163         for node_type in node_types:
164             if (existing_nodes.get(node_type, 0) + nodes_to_add.get(
165                     node_type, 0) >= node_types[node_type]["max_workers"]):
166                 continue
167             node_resources = node_types[node_type]["resources"]
168             score = _utilization_score(node_resources, resources)
169             if score is not None:
170                 utilization_scores.append((score, node_type))
171 
172         # Give up, no feasible node.
173         if not utilization_scores:
174             logger.info(
175                 "No feasible node type to add for {}".format(resources))
176             break
177 
178         utilization_scores = sorted(utilization_scores, reverse=True)
179         best_node_type = utilization_scores[0][1]
180         nodes_to_add[best_node_type] += 1
181         allocated_resources.append(node_types[best_node_type]["resources"])
182         residual = get_bin_pack_residual(allocated_resources[-1:], resources)
183         assert len(residual) < len(resources), (resources, residual)
184         resources = residual
185 
186     return list(nodes_to_add.items())
187 
188 
189 def _utilization_score(node_resources: ResourceDict,
190                        resources: ResourceDict) -> float:
191     remaining = copy.deepcopy(node_resources)
192 
193     fittable = []
194     for r in resources:
195         if _fits(remaining, r):
196             fittable.append(r)
197             _inplace_subtract(remaining, r)
198     if not fittable:
199         return None
200 
201     util_by_resources = []
202     for k, v in node_resources.items():
203         util = (v - remaining[k]) / v
204         util_by_resources.append(v * (util**3))
205 
206     # Prioritize using all resources first, then prioritize overall balance
207     # of multiple resources.
208     return (min(util_by_resources), np.mean(util_by_resources))
209 
210 
211 def get_bin_pack_residual(
212         node_resources: List[ResourceDict],
213         resource_demands: List[ResourceDict]) -> List[ResourceDict]:
214     """Return a subset of resource_demands that cannot fit in the cluster.
215 
216     TODO(ekl): this currently does not guarantee the resources will be packed
217     correctly by the Ray scheduler. This is only possible once the Ray backend
218     supports a placement groups API.
219 
220     Args:
221         node_resources (List[ResourceDict]): List of resources per node.
222         resource_demands (List[ResourceDict]): List of resource bundles that
223             need to be bin packed onto the nodes.
224 
225     Returns:
226         List[ResourceDict] the residual list resources that do not fit.
227     """
228 
229     unfulfilled = []
230 
231     # A most naive bin packing algorithm.
232     nodes = copy.deepcopy(node_resources)
233     for demand in resource_demands:
234         found = False
235         for node in nodes:
236             if _fits(node, demand):
237                 _inplace_subtract(node, demand)
238                 found = True
239                 break
240         if not found:
241             unfulfilled.append(demand)
242 
243     return unfulfilled
244 
245 
246 def _fits(node: ResourceDict, resources: ResourceDict) -> bool:
247     for k, v in resources.items():
248         if v > node.get(k, 0.0):
249             return False
250     return True
251 
252 
253 def _inplace_subtract(node: ResourceDict, resources: ResourceDict) -> None:
254     for k, v in resources.items():
255         assert k in node, (k, node)
256         node[k] -= v
257         assert node[k] >= 0.0, (node, k, v)
258
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/ray/autoscaler/resource_demand_scheduler.py b/python/ray/autoscaler/resource_demand_scheduler.py
--- a/python/ray/autoscaler/resource_demand_scheduler.py
+++ b/python/ray/autoscaler/resource_demand_scheduler.py
@@ -56,6 +56,7 @@
             nodes: List of existing nodes in the cluster.
             pending_nodes: Summary of node types currently being launched.
             resource_demands: Vector of resource demands from the scheduler.
+            usage_by_ip: Mapping from ip to available resources.
         """
 
         if resource_demands is None:
@@ -113,7 +114,6 @@
             tags = self.provider.node_tags(node_id)
             if TAG_RAY_USER_NODE_TYPE in tags:
                 node_type = tags[TAG_RAY_USER_NODE_TYPE]
-                node_type_counts[node_type] += 1
                 ip = self.provider.internal_ip(node_id)
                 available_resources = usage_by_ip.get(ip)
                 add_node(node_type, available_resources)

{"golden_diff": "diff --git a/python/ray/autoscaler/resource_demand_scheduler.py b/python/ray/autoscaler/resource_demand_scheduler.py\n--- a/python/ray/autoscaler/resource_demand_scheduler.py\n+++ b/python/ray/autoscaler/resource_demand_scheduler.py\n@@ -56,6 +56,7 @@\n             nodes: List of existing nodes in the cluster.\n             pending_nodes: Summary of node types currently being launched.\n             resource_demands: Vector of resource demands from the scheduler.\n+            usage_by_ip: Mapping from ip to available resources.\n         \"\"\"\n \n         if resource_demands is None:\n@@ -113,7 +114,6 @@\n             tags = self.provider.node_tags(node_id)\n             if TAG_RAY_USER_NODE_TYPE in tags:\n                 node_type = tags[TAG_RAY_USER_NODE_TYPE]\n-                node_type_counts[node_type] += 1\n                 ip = self.provider.internal_ip(node_id)\n                 available_resources = usage_by_ip.get(ip)\n                 add_node(node_type, available_resources)\n", "issue": "[autoscaler] Inconsistent reporting of worker node types\nOn master, it seems I have a cluster with a cpu_4_ondemand head node and cpu_64_ondemand worker (68 CPUs in total), but somehow the debug string is reporting 2 workers of cpu_64_ondemand (which is wrong, since there is only 1 worker node, and the head node is cpu_4_ondemand).\r\n\r\nYAML: https://gist.github.com/ericl/f9c32b8bbbe4e1aed43899832730f70d (switching the Ray version to latest wheel)\r\n\r\n```\r\n - ResourceUsage: 0.0/68.0 CPU, 0.0/1.0 Custom1, 0.0 GiB/198.65 GiB memory, 0.0\r\nGiB/59.19 GiB object_store_memory\r\n - TimeSinceLastHeartbeat: Min=0 Mean=0 Max=0\r\nWorker node types:\r\n - cpu_64_ondemand: 2\r\n\r\n```\r\n\r\ncc @wuisawesome \n", "before_files": [{"content": "\"\"\"Implements multi-node-type autoscaling.\n\nThis file implements an autoscaling algorithm that is aware of multiple node\ntypes (e.g., example-multi-node-type.yaml). The Ray autoscaler will pass in\na vector of resource shape demands, and the resource demand scheduler will\nreturn a list of node types that can satisfy the demands given constraints\n(i.e., reverse bin packing).\n\"\"\"\n\nimport copy\nimport numpy as np\nimport logging\nimport collections\nfrom typing import List, Dict, Tuple\n\nfrom ray.autoscaler.node_provider import NodeProvider\nfrom ray.autoscaler.tags import TAG_RAY_USER_NODE_TYPE, NODE_KIND_UNMANAGED\n\nlogger = logging.getLogger(__name__)\n\n# e.g., m4.16xlarge.\nNodeType = str\n\n# e.g., {\"resources\": ..., \"max_workers\": ...}.\nNodeTypeConfigDict = str\n\n# e.g., {\"GPU\": 1}.\nResourceDict = str\n\n# e.g., IP address of the node.\nNodeID = str\n\n\nclass ResourceDemandScheduler:\n    def __init__(self, provider: NodeProvider,\n                 node_types: Dict[NodeType, NodeTypeConfigDict],\n                 max_workers: int):\n        self.provider = provider\n        self.node_types = node_types\n        self.max_workers = max_workers\n\n    def get_nodes_to_launch(self, nodes: List[NodeID],\n                            pending_nodes: Dict[NodeType, int],\n                            resource_demands: List[ResourceDict],\n                            usage_by_ip: Dict[str, ResourceDict]\n                            ) -> List[Tuple[NodeType, int]]:\n        \"\"\"Given resource demands, return node types to add to the cluster.\n\n        This method:\n            (1) calculates the resources present in the cluster.\n            (2) calculates the unfulfilled resource bundles.\n            (3) calculates which nodes need to be launched to fulfill all\n                the bundle requests, subject to max_worker constraints.\n\n        Args:\n            nodes: List of existing nodes in the cluster.\n            pending_nodes: Summary of node types currently being launched.\n            resource_demands: Vector of resource demands from the scheduler.\n        \"\"\"\n\n        if resource_demands is None:\n            logger.info(\"No resource demands\")\n            return []\n\n        node_resources, node_type_counts = self.calculate_node_resources(\n            nodes, pending_nodes, usage_by_ip)\n\n        logger.info(\"Cluster resources: {}\".format(node_resources))\n        logger.info(\"Node counts: {}\".format(node_type_counts))\n\n        unfulfilled = get_bin_pack_residual(node_resources, resource_demands)\n        logger.info(\"Resource demands: {}\".format(resource_demands))\n        logger.info(\"Unfulfilled demands: {}\".format(unfulfilled))\n\n        nodes = get_nodes_for(\n            self.node_types, node_type_counts,\n            self.max_workers - len(nodes) - sum(pending_nodes.values()),\n            unfulfilled)\n        logger.info(\"Node requests: {}\".format(nodes))\n        return nodes\n\n    def calculate_node_resources(\n            self, nodes: List[NodeID], pending_nodes: Dict[NodeID, int],\n            usage_by_ip: Dict[str, ResourceDict]\n    ) -> (List[ResourceDict], Dict[NodeType, int]):\n        \"\"\"Returns node resource list and node type counts.\"\"\"\n\n        node_resources = []\n        node_type_counts = collections.defaultdict(int)\n\n        def add_node(node_type, available_resources=None):\n            if node_type not in self.node_types:\n                logger.warn(\n                    f\"Missing entry for node_type {node_type} in \"\n                    f\"cluster config: {self.node_types} under entry \"\n                    f\"available_node_types. This node's resources will be \"\n                    f\"ignored. If you are using an unmanaged node, manually \"\n                    f\"set the user_node_type tag to \\\"{NODE_KIND_UNMANAGED}\\\"\"\n                    f\"in your cloud provider's management console.\")\n                return None\n            # Careful not to include the same dict object multiple times.\n            available = copy.deepcopy(self.node_types[node_type][\"resources\"])\n            # If available_resources is None this might be because the node is\n            # no longer pending, but the raylet hasn't sent a heartbeat to gcs\n            # yet.\n            if available_resources is not None:\n                available = copy.deepcopy(available_resources)\n\n            node_resources.append(available)\n            node_type_counts[node_type] += 1\n\n        for node_id in nodes:\n            tags = self.provider.node_tags(node_id)\n            if TAG_RAY_USER_NODE_TYPE in tags:\n                node_type = tags[TAG_RAY_USER_NODE_TYPE]\n                node_type_counts[node_type] += 1\n                ip = self.provider.internal_ip(node_id)\n                available_resources = usage_by_ip.get(ip)\n                add_node(node_type, available_resources)\n\n        for node_type, count in pending_nodes.items():\n            for _ in range(count):\n                add_node(node_type)\n\n        return node_resources, node_type_counts\n\n    def debug_string(self, nodes: List[NodeID],\n                     pending_nodes: Dict[NodeID, int],\n                     usage_by_ip: Dict[str, ResourceDict]) -> str:\n        print(f\"{usage_by_ip}\")\n        node_resources, node_type_counts = self.calculate_node_resources(\n            nodes, pending_nodes, usage_by_ip)\n\n        out = \"Worker node types:\"\n        for node_type, count in node_type_counts.items():\n            out += \"\\n - {}: {}\".format(node_type, count)\n            if pending_nodes.get(node_type):\n                out += \" ({} pending)\".format(pending_nodes[node_type])\n\n        return out\n\n\ndef get_nodes_for(node_types: Dict[NodeType, NodeTypeConfigDict],\n                  existing_nodes: Dict[NodeType, int], max_to_add: int,\n                  resources: List[ResourceDict]) -> List[Tuple[NodeType, int]]:\n    \"\"\"Determine nodes to add given resource demands and constraints.\n\n    Args:\n        node_types: node types config.\n        existing_nodes: counts of existing nodes already launched.\n            This sets constraints on the number of new nodes to add.\n        max_to_add: global constraint on nodes to add.\n        resources: resource demands to fulfill.\n\n    Returns:\n        List of nodes types and count to add.\n    \"\"\"\n    nodes_to_add = collections.defaultdict(int)\n    allocated_resources = []\n\n    while resources and sum(nodes_to_add.values()) < max_to_add:\n        utilization_scores = []\n        for node_type in node_types:\n            if (existing_nodes.get(node_type, 0) + nodes_to_add.get(\n                    node_type, 0) >= node_types[node_type][\"max_workers\"]):\n                continue\n            node_resources = node_types[node_type][\"resources\"]\n            score = _utilization_score(node_resources, resources)\n            if score is not None:\n                utilization_scores.append((score, node_type))\n\n        # Give up, no feasible node.\n        if not utilization_scores:\n            logger.info(\n                \"No feasible node type to add for {}\".format(resources))\n            break\n\n        utilization_scores = sorted(utilization_scores, reverse=True)\n        best_node_type = utilization_scores[0][1]\n        nodes_to_add[best_node_type] += 1\n        allocated_resources.append(node_types[best_node_type][\"resources\"])\n        residual = get_bin_pack_residual(allocated_resources[-1:], resources)\n        assert len(residual) < len(resources), (resources, residual)\n        resources = residual\n\n    return list(nodes_to_add.items())\n\n\ndef _utilization_score(node_resources: ResourceDict,\n                       resources: ResourceDict) -> float:\n    remaining = copy.deepcopy(node_resources)\n\n    fittable = []\n    for r in resources:\n        if _fits(remaining, r):\n            fittable.append(r)\n            _inplace_subtract(remaining, r)\n    if not fittable:\n        return None\n\n    util_by_resources = []\n    for k, v in node_resources.items():\n        util = (v - remaining[k]) / v\n        util_by_resources.append(v * (util**3))\n\n    # Prioritize using all resources first, then prioritize overall balance\n    # of multiple resources.\n    return (min(util_by_resources), np.mean(util_by_resources))\n\n\ndef get_bin_pack_residual(\n        node_resources: List[ResourceDict],\n        resource_demands: List[ResourceDict]) -> List[ResourceDict]:\n    \"\"\"Return a subset of resource_demands that cannot fit in the cluster.\n\n    TODO(ekl): this currently does not guarantee the resources will be packed\n    correctly by the Ray scheduler. This is only possible once the Ray backend\n    supports a placement groups API.\n\n    Args:\n        node_resources (List[ResourceDict]): List of resources per node.\n        resource_demands (List[ResourceDict]): List of resource bundles that\n            need to be bin packed onto the nodes.\n\n    Returns:\n        List[ResourceDict] the residual list resources that do not fit.\n    \"\"\"\n\n    unfulfilled = []\n\n    # A most naive bin packing algorithm.\n    nodes = copy.deepcopy(node_resources)\n    for demand in resource_demands:\n        found = False\n        for node in nodes:\n            if _fits(node, demand):\n                _inplace_subtract(node, demand)\n                found = True\n                break\n        if not found:\n            unfulfilled.append(demand)\n\n    return unfulfilled\n\n\ndef _fits(node: ResourceDict, resources: ResourceDict) -> bool:\n    for k, v in resources.items():\n        if v > node.get(k, 0.0):\n            return False\n    return True\n\n\ndef _inplace_subtract(node: ResourceDict, resources: ResourceDict) -> None:\n    for k, v in resources.items():\n        assert k in node, (k, node)\n        node[k] -= v\n        assert node[k] >= 0.0, (node, k, v)\n", "path": "python/ray/autoscaler/resource_demand_scheduler.py"}], "after_files": [{"content": "\"\"\"Implements multi-node-type autoscaling.\n\nThis file implements an autoscaling algorithm that is aware of multiple node\ntypes (e.g., example-multi-node-type.yaml). The Ray autoscaler will pass in\na vector of resource shape demands, and the resource demand scheduler will\nreturn a list of node types that can satisfy the demands given constraints\n(i.e., reverse bin packing).\n\"\"\"\n\nimport copy\nimport numpy as np\nimport logging\nimport collections\nfrom typing import List, Dict, Tuple\n\nfrom ray.autoscaler.node_provider import NodeProvider\nfrom ray.autoscaler.tags import TAG_RAY_USER_NODE_TYPE, NODE_KIND_UNMANAGED\n\nlogger = logging.getLogger(__name__)\n\n# e.g., m4.16xlarge.\nNodeType = str\n\n# e.g., {\"resources\": ..., \"max_workers\": ...}.\nNodeTypeConfigDict = str\n\n# e.g., {\"GPU\": 1}.\nResourceDict = str\n\n# e.g., IP address of the node.\nNodeID = str\n\n\nclass ResourceDemandScheduler:\n    def __init__(self, provider: NodeProvider,\n                 node_types: Dict[NodeType, NodeTypeConfigDict],\n                 max_workers: int):\n        self.provider = provider\n        self.node_types = node_types\n        self.max_workers = max_workers\n\n    def get_nodes_to_launch(self, nodes: List[NodeID],\n                            pending_nodes: Dict[NodeType, int],\n                            resource_demands: List[ResourceDict],\n                            usage_by_ip: Dict[str, ResourceDict]\n                            ) -> List[Tuple[NodeType, int]]:\n        \"\"\"Given resource demands, return node types to add to the cluster.\n\n        This method:\n            (1) calculates the resources present in the cluster.\n            (2) calculates the unfulfilled resource bundles.\n            (3) calculates which nodes need to be launched to fulfill all\n                the bundle requests, subject to max_worker constraints.\n\n        Args:\n            nodes: List of existing nodes in the cluster.\n            pending_nodes: Summary of node types currently being launched.\n            resource_demands: Vector of resource demands from the scheduler.\n            usage_by_ip: Mapping from ip to available resources.\n        \"\"\"\n\n        if resource_demands is None:\n            logger.info(\"No resource demands\")\n            return []\n\n        node_resources, node_type_counts = self.calculate_node_resources(\n            nodes, pending_nodes, usage_by_ip)\n\n        logger.info(\"Cluster resources: {}\".format(node_resources))\n        logger.info(\"Node counts: {}\".format(node_type_counts))\n\n        unfulfilled = get_bin_pack_residual(node_resources, resource_demands)\n        logger.info(\"Resource demands: {}\".format(resource_demands))\n        logger.info(\"Unfulfilled demands: {}\".format(unfulfilled))\n\n        nodes = get_nodes_for(\n            self.node_types, node_type_counts,\n            self.max_workers - len(nodes) - sum(pending_nodes.values()),\n            unfulfilled)\n        logger.info(\"Node requests: {}\".format(nodes))\n        return nodes\n\n    def calculate_node_resources(\n            self, nodes: List[NodeID], pending_nodes: Dict[NodeID, int],\n            usage_by_ip: Dict[str, ResourceDict]\n    ) -> (List[ResourceDict], Dict[NodeType, int]):\n        \"\"\"Returns node resource list and node type counts.\"\"\"\n\n        node_resources = []\n        node_type_counts = collections.defaultdict(int)\n\n        def add_node(node_type, available_resources=None):\n            if node_type not in self.node_types:\n                logger.warn(\n                    f\"Missing entry for node_type {node_type} in \"\n                    f\"cluster config: {self.node_types} under entry \"\n                    f\"available_node_types. This node's resources will be \"\n                    f\"ignored. If you are using an unmanaged node, manually \"\n                    f\"set the user_node_type tag to \\\"{NODE_KIND_UNMANAGED}\\\"\"\n                    f\"in your cloud provider's management console.\")\n                return None\n            # Careful not to include the same dict object multiple times.\n            available = copy.deepcopy(self.node_types[node_type][\"resources\"])\n            # If available_resources is None this might be because the node is\n            # no longer pending, but the raylet hasn't sent a heartbeat to gcs\n            # yet.\n            if available_resources is not None:\n                available = copy.deepcopy(available_resources)\n\n            node_resources.append(available)\n            node_type_counts[node_type] += 1\n\n        for node_id in nodes:\n            tags = self.provider.node_tags(node_id)\n            if TAG_RAY_USER_NODE_TYPE in tags:\n                node_type = tags[TAG_RAY_USER_NODE_TYPE]\n                ip = self.provider.internal_ip(node_id)\n                available_resources = usage_by_ip.get(ip)\n                add_node(node_type, available_resources)\n\n        for node_type, count in pending_nodes.items():\n            for _ in range(count):\n                add_node(node_type)\n\n        return node_resources, node_type_counts\n\n    def debug_string(self, nodes: List[NodeID],\n                     pending_nodes: Dict[NodeID, int],\n                     usage_by_ip: Dict[str, ResourceDict]) -> str:\n        print(f\"{usage_by_ip}\")\n        node_resources, node_type_counts = self.calculate_node_resources(\n            nodes, pending_nodes, usage_by_ip)\n\n        out = \"Worker node types:\"\n        for node_type, count in node_type_counts.items():\n            out += \"\\n - {}: {}\".format(node_type, count)\n            if pending_nodes.get(node_type):\n                out += \" ({} pending)\".format(pending_nodes[node_type])\n\n        return out\n\n\ndef get_nodes_for(node_types: Dict[NodeType, NodeTypeConfigDict],\n                  existing_nodes: Dict[NodeType, int], max_to_add: int,\n                  resources: List[ResourceDict]) -> List[Tuple[NodeType, int]]:\n    \"\"\"Determine nodes to add given resource demands and constraints.\n\n    Args:\n        node_types: node types config.\n        existing_nodes: counts of existing nodes already launched.\n            This sets constraints on the number of new nodes to add.\n        max_to_add: global constraint on nodes to add.\n        resources: resource demands to fulfill.\n\n    Returns:\n        List of nodes types and count to add.\n    \"\"\"\n    nodes_to_add = collections.defaultdict(int)\n    allocated_resources = []\n\n    while resources and sum(nodes_to_add.values()) < max_to_add:\n        utilization_scores = []\n        for node_type in node_types:\n            if (existing_nodes.get(node_type, 0) + nodes_to_add.get(\n                    node_type, 0) >= node_types[node_type][\"max_workers\"]):\n                continue\n            node_resources = node_types[node_type][\"resources\"]\n            score = _utilization_score(node_resources, resources)\n            if score is not None:\n                utilization_scores.append((score, node_type))\n\n        # Give up, no feasible node.\n        if not utilization_scores:\n            logger.info(\n                \"No feasible node type to add for {}\".format(resources))\n            break\n\n        utilization_scores = sorted(utilization_scores, reverse=True)\n        best_node_type = utilization_scores[0][1]\n        nodes_to_add[best_node_type] += 1\n        allocated_resources.append(node_types[best_node_type][\"resources\"])\n        residual = get_bin_pack_residual(allocated_resources[-1:], resources)\n        assert len(residual) < len(resources), (resources, residual)\n        resources = residual\n\n    return list(nodes_to_add.items())\n\n\ndef _utilization_score(node_resources: ResourceDict,\n                       resources: ResourceDict) -> float:\n    remaining = copy.deepcopy(node_resources)\n\n    fittable = []\n    for r in resources:\n        if _fits(remaining, r):\n            fittable.append(r)\n            _inplace_subtract(remaining, r)\n    if not fittable:\n        return None\n\n    util_by_resources = []\n    for k, v in node_resources.items():\n        util = (v - remaining[k]) / v\n        util_by_resources.append(v * (util**3))\n\n    # Prioritize using all resources first, then prioritize overall balance\n    # of multiple resources.\n    return (min(util_by_resources), np.mean(util_by_resources))\n\n\ndef get_bin_pack_residual(\n        node_resources: List[ResourceDict],\n        resource_demands: List[ResourceDict]) -> List[ResourceDict]:\n    \"\"\"Return a subset of resource_demands that cannot fit in the cluster.\n\n    TODO(ekl): this currently does not guarantee the resources will be packed\n    correctly by the Ray scheduler. This is only possible once the Ray backend\n    supports a placement groups API.\n\n    Args:\n        node_resources (List[ResourceDict]): List of resources per node.\n        resource_demands (List[ResourceDict]): List of resource bundles that\n            need to be bin packed onto the nodes.\n\n    Returns:\n        List[ResourceDict] the residual list resources that do not fit.\n    \"\"\"\n\n    unfulfilled = []\n\n    # A most naive bin packing algorithm.\n    nodes = copy.deepcopy(node_resources)\n    for demand in resource_demands:\n        found = False\n        for node in nodes:\n            if _fits(node, demand):\n                _inplace_subtract(node, demand)\n                found = True\n                break\n        if not found:\n            unfulfilled.append(demand)\n\n    return unfulfilled\n\n\ndef _fits(node: ResourceDict, resources: ResourceDict) -> bool:\n    for k, v in resources.items():\n        if v > node.get(k, 0.0):\n            return False\n    return True\n\n\ndef _inplace_subtract(node: ResourceDict, resources: ResourceDict) -> None:\n    for k, v in resources.items():\n        assert k in node, (k, node)\n        node[k] -= v\n        assert node[k] >= 0.0, (node, k, v)\n", "path": "python/ray/autoscaler/resource_demand_scheduler.py"}]}