Merge pull request #194 from njzjz/inf-group-size

support infinite group_size

Author: felix5572

doc/examples/shell.md

@@ -14,4 +14,4 @@ The workstation has 48 cores of CPUs and 8 RTX3090 cards. Here we hope each card
 :linenos:
 ```
 
-Note that `group_size` should be set as large as possible to ensure there is only one job and avoid running multiple jobs at the same time.
+Note that `group_size` should be set to `0` (means infinity) to ensure there is only one job and avoid running multiple jobs at the same time.

dpdispatcher/submission.py

@@ -311,11 +311,14 @@ def generate_jobs(self):
         if self.belonging_jobs:
             raise RuntimeError(f'Can not generate jobs when submission.belonging_jobs is not empty. debug:{self}')
         group_size = self.resources.group_size
-        if ( group_size < 1 ) or ( type(group_size) is not int ):
+        if ( group_size < 0 ) or ( type(group_size) is not int ):
             raise RuntimeError('group_size must be a positive number')
         task_num = len(self.belonging_tasks)
         if task_num == 0:
             raise RuntimeError("submission must have at least 1 task")
+        if group_size == 0:
+            # 0 means infinity
+            group_size = task_num
         random.seed(42)
         random_task_index = list(range(task_num))
         random.shuffle(random_task_index)
@@ -815,7 +818,7 @@ def arginfo(detail_kwargs=True):
         doc_cpu_per_node = 'cpu numbers of each node assigned to each job.'
         doc_gpu_per_node = 'gpu numbers of each node assigned to each job.'
         doc_queue_name = 'The queue name of batch job scheduler system.'
-        doc_group_size = 'The number of `tasks` in a `job`.'
+        doc_group_size = 'The number of `tasks` in a `job`. 0 means infinity.'
         doc_custom_flags = 'The extra lines pass to job submitting script header'
         doc_para_deg = 'Decide how many tasks will be run in parallel.'
         doc_source_list = 'The env file to be sourced before the command execution.'

examples/resources/mandu.json

@@ -3,7 +3,7 @@
   "cpu_per_node": 48,
   "gpu_per_node": 8,
   "queue_name": "shell",
-  "group_size": 9999,
+  "group_size": 0,
   "strategy": {
     "if_cuda_multi_devices": true
   },

tests/test_group_size.py

@@ -0,0 +1,42 @@
+"""Test `Submission.generate_jobs` with different group size."""
+
+import os
+import sys
+import json
+from unittest import TestCase
+from pathlib import Path
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+__package__ = 'tests'
+from .context import Machine, Resources, Task, Submission
+
+# 99 tasks in total
+# group_size - expected_ntasks
+group_ntasks_pairs = [
+    (1, 99),
+    (3, 33),
+    (10, 10),
+    (100, 1),
+    (0, 1),
+]
+
+cwd = Path(__file__).parent
+with open(cwd / "jsons" / "machine.json") as f:
+    j_machine = json.load(f)['machine']
+with open(cwd / "jsons" / "resources.json") as f:
+    j_resources = json.load(f)
+with open(cwd / "jsons" / "task.json") as f:
+    j_task = json.load(f)
+
+
+class TestGroupSize(TestCase):
+    def test_works_as_expected(self):
+        for group_size, ntasks in group_ntasks_pairs:
+            with self.subTest(group_size):
+                machine = Machine.load_from_dict(j_machine)
+                j_resources['group_size'] = group_size
+                resources = Resources.load_from_dict(j_resources)
+                tasks = [Task.load_from_dict(j_task) for _ in range(99)]
+                submission = Submission(".", machine, resources, task_list=tasks)
+                submission.generate_jobs()
+                self.assertEqual(len(submission.belonging_jobs), ntasks)