Repository: codalab/codalab-worksheets

Issue:
CPU job assignment issue
Currently, jobs that were requested to run on CPU workers are still being assigned to a GPU worker even with CPU workers were idle during the time of scheduling.
1. Without any dependencies specified
<img width="1260" alt="Screen Shot 2020-04-13 at 10 25 17 AM" src="https://user-images.githubusercontent.com/1483372/79143135-8850e200-7d71-11ea-8f00-609827b3187f.png">

1. With dependencies specified
<img width="1248" alt="Screen Shot 2020-04-13 at 10 26 05 AM" src="https://user-images.githubusercontent.com/1483372/79143964-fba72380-7d72-11ea-9075-833f305a4eac.png">

We want to avoid dispatching jobs that only require to run on CPU workers to GPU workers as much as possible.

Discussion:

Patch:
diff --git a/codalab/server/bundle_manager.py b/codalab/server/bundle_manager.py
index 83909a22..ee614379 100644
--- a/codalab/server/bundle_manager.py
+++ b/codalab/server/bundle_manager.py
@@ -475,7 +475,7 @@ class BundleManager(object):
             # 6. break ties randomly by a random seed.
             return (
                 not worker['tag_exclusive'],
-                worker['gpus'],
+                worker['gpus'] or worker['has_gpus'],
                 -num_available_deps,
                 worker['cpus'],
                 len(worker['run_uuids']),
diff --git a/codalab/server/worker_info_accessor.py b/codalab/server/worker_info_accessor.py
index 2ab38754..6285fc4a 100644
--- a/codalab/server/worker_info_accessor.py
+++ b/codalab/server/worker_info_accessor.py
@@ -33,6 +33,9 @@ class WorkerInfoAccessor(object):
             for uuid in worker['run_uuids']:
                 self._uuid_to_worker[uuid] = worker
             self._user_id_to_workers[worker['user_id']].append(worker)
+            # 'gpus' field contains the number of free GPUs that comes with each worker. Adding an additional
+            # 'has_gpus' flag here to indicate if the current worker has GPUs or not.
+            worker['has_gpus'] = True if worker['gpus'] > 0 else False
 
     @refresh_cache
     def workers(self):


Test patch:
diff --git a/tests/server/bundle_manager_test.py b/tests/server/bundle_manager_test.py
index c753cd1e..9ea1ace0 100644
--- a/tests/server/bundle_manager_test.py
+++ b/tests/server/bundle_manager_test.py
@@ -49,6 +49,8 @@ class BundleManagerTest(unittest.TestCase):
                 'dependencies': [(1, ''), (2, '')],
                 'shared_file_system': False,
                 'tag_exclusive': False,
+                # An additional flag that is generated from WorkerInfoAccessor class
+                'has_gpus': True,
             },
             {
                 'worker_id': 1,
@@ -60,17 +62,20 @@ class BundleManagerTest(unittest.TestCase):
                 'dependencies': [(1, ''), (2, ''), (3, ''), (4, '')],
                 'shared_file_system': False,
                 'tag_exclusive': False,
+                'has_gpus': True,
             },
+            # the value of GPUs has been deducted to 0 at this point even though this worker has GPUs
             {
                 'worker_id': 2,
                 'cpus': 4,
-                'gpus': 1,
+                'gpus': 0,
                 'memory_bytes': 4 * 1000,
                 'tag': None,
                 'run_uuids': [],
                 'dependencies': [(1, ''), (2, ''), (3, ''), (4, '')],
                 'shared_file_system': False,
                 'tag_exclusive': False,
+                'has_gpus': True,
             },
             {
                 'worker_id': 3,
@@ -82,20 +87,35 @@ class BundleManagerTest(unittest.TestCase):
                 'dependencies': [(1, '')],
                 'shared_file_system': False,
                 'tag_exclusive': False,
+                'has_gpus': False,
             },
             {
                 'worker_id': 4,
                 'cpus': 6,
                 'gpus': 0,
+                'memory_bytes': 4 * 1000,
+                'tag': None,
+                'run_uuids': [1, 2],
+                'dependencies': [(1, '')],
+                'shared_file_system': False,
+                'tag_exclusive': False,
+                'has_gpus': False,
+            },
+            # Tagged workers
+            {
+                'worker_id': 5,
+                'cpus': 6,
+                'gpus': 0,
                 'memory_bytes': 2 * 1000,
                 'tag': 'worker_X',
                 'run_uuids': [],
                 'dependencies': [],
                 'shared_file_system': False,
                 'tag_exclusive': False,
+                'has_gpus': False,
             },
             {
-                'worker_id': 5,
+                'worker_id': 6,
                 'cpus': 6,
                 'gpus': 0,
                 'memory_bytes': 2 * 1000,
@@ -104,31 +124,42 @@ class BundleManagerTest(unittest.TestCase):
                 'dependencies': [],
                 'shared_file_system': False,
                 'tag_exclusive': True,
+                'has_gpus': False,
             },
         ]
         return workers_list
 
-    def test__filter_and_sort_workers_gpus(self):
+    def test_filter_and_sort_workers_gpus(self):
         # Only GPU workers should appear from the returning sorted worker list
         self.bundle_resources.gpus = 1
         sorted_workers_list = self.bundle_manager._filter_and_sort_workers(
             self.workers_list, self.bundle, self.bundle_resources
         )
-        self.assertEqual(len(sorted_workers_list), 3)
-        self.assertEqual(sorted_workers_list[0]['worker_id'], 2)
-        self.assertEqual(sorted_workers_list[1]['worker_id'], 1)
+        self.assertEqual(len(sorted_workers_list), 2)
+        self.assertEqual(sorted_workers_list[0]['worker_id'], 1)
+        self.assertEqual(sorted_workers_list[1]['worker_id'], 0)
+
+    def test_filter_and_sort_workers_cpus(self):
+        # CPU workers should appear on the top of the returning sorted worker list
+        self.bundle_resources.cpus = 1
+        sorted_workers_list = self.bundle_manager._filter_and_sort_workers(
+            self.workers_list, self.bundle, self.bundle_resources
+        )
+        self.assertEqual(len(sorted_workers_list), 6)
+        self.assertEqual(sorted_workers_list[0]['worker_id'], 3)
+        self.assertEqual(sorted_workers_list[1]['worker_id'], 4)
         self.assertEqual(sorted_workers_list[-1]['worker_id'], 0)
 
-    def test__filter_and_sort_workers_tag_exclusive(self):
+    def test_filter_and_sort_workers_tag_exclusive(self):
         # Only non-tag_exclusive workers should appear in the returned sorted worker list.
         sorted_workers_list = self.bundle_manager._filter_and_sort_workers(
             self.workers_list, self.bundle, self.bundle_resources
         )
-        self.assertEqual(len(sorted_workers_list), 5)
+        self.assertEqual(len(sorted_workers_list), 6)
         for worker in sorted_workers_list:
             self.assertEqual(worker['tag_exclusive'], False)
 
-    def test__filter_and_sort_workers_tag_exclusive_priority(self):
+    def test_filter_and_sort_workers_tag_exclusive_priority(self):
         # All other things being equal, tag_exclusive workers
         # should appear in the top from the returned sorted workers list.
         self.bundle.metadata.request_queue = "tag=worker_X"
@@ -136,51 +167,42 @@ class BundleManagerTest(unittest.TestCase):
             self.workers_list, self.bundle, self.bundle_resources
         )
         self.assertEqual(len(sorted_workers_list), 2)
-        self.assertEqual(sorted_workers_list[0]['worker_id'], 5)
-        self.assertEqual(sorted_workers_list[1]['worker_id'], 4)
-
-    def test__filter_and_sort_workers_cpus(self):
-        # CPU workers should appear in the top from the returning sorted worker list
-        self.bundle_resources.cpus = 1
-        sorted_workers_list = self.bundle_manager._filter_and_sort_workers(
-            self.workers_list, self.bundle, self.bundle_resources
-        )
-        self.assertEqual(len(sorted_workers_list), 5)
-        self.assertEqual(sorted_workers_list[0]['worker_id'], 3)
+        self.assertEqual(sorted_workers_list[0]['worker_id'], 6)
+        self.assertEqual(sorted_workers_list[1]['worker_id'], 5)
 
-    def test__get_matched_workers_with_tag(self):
+    def test_get_matched_workers_with_tag(self):
         self.bundle.metadata.request_queue = "tag=worker_X"
         matched_workers = BundleManager._get_matched_workers(
             self.bundle.metadata.request_queue, self.workers_list
         )
         self.assertEqual(len(matched_workers), 2)
-        self.assertEqual(matched_workers[0]['worker_id'], 4)
-        self.assertEqual(matched_workers[1]['worker_id'], 5)
+        self.assertEqual(matched_workers[0]['worker_id'], 5)
+        self.assertEqual(matched_workers[1]['worker_id'], 6)
 
-    def test__get_matched_workers_with_bad_formatted_tag(self):
+    def test_get_matched_workers_with_bad_formatted_tag(self):
         self.bundle.metadata.request_queue = "tag="
         matched_workers = BundleManager._get_matched_workers(
             self.bundle.metadata.request_queue, self.workers_list
         )
         self.assertEqual(len(matched_workers), 0)
 
-    def test__get_matched_workers_without_tag_prefix(self):
+    def test_get_matched_workers_without_tag_prefix(self):
         self.bundle.metadata.request_queue = "worker_X"
         matched_workers = BundleManager._get_matched_workers(
             self.bundle.metadata.request_queue, self.workers_list
         )
         self.assertEqual(len(matched_workers), 2)
-        self.assertEqual(matched_workers[0]['worker_id'], 4)
-        self.assertEqual(matched_workers[1]['worker_id'], 5)
+        self.assertEqual(matched_workers[0]['worker_id'], 5)
+        self.assertEqual(matched_workers[1]['worker_id'], 6)
 
-    def test__get_matched_workers_not_exist_tag(self):
+    def test_get_matched_workers_not_exist_tag(self):
         self.bundle.metadata.request_queue = "worker_Y"
         matched_workers = BundleManager._get_matched_workers(
             self.bundle.metadata.request_queue, self.workers_list
         )
         self.assertEqual(len(matched_workers), 0)
 
-    def test__get_matched_workers_empty_tag(self):
+    def test_get_matched_workers_empty_tag(self):
         self.bundle.metadata.request_queue = ""
         matched_workers = BundleManager._get_matched_workers(
             self.bundle.metadata.request_queue, self.workers_list