[Core] Allow more PENDING jobs to be scheduled concurrently (1.4x faster)

sky/skylet/job_lib.py

@@ -33,6 +33,8 @@
 # the same pid is reused by a different process.
 JOB_CMD_IDENTIFIER = 'echo "SKYPILOT_JOB_ID <{}>"'
 
+_MAX_PENDING_SUBMIT = 2
+
 
 def _get_lock_path(job_id: int) -> str:
     lock_path = os.path.expanduser(_JOB_STATUS_LOCK.format(job_id))
@@ -246,6 +248,7 @@ def schedule_step(self, force_update_jobs: bool = False) -> None:
         # TODO(zhwu, mraheja): One optimization can be allowing more than one
         # job staying in the pending state after ray job submit, so that to be
         # faster to schedule a large amount of jobs.
+        pending_submit_cnt = 0
         for job_id in pending_job_ids:
             with filelock.FileLock(_get_lock_path(job_id)):
                 pending_job = _get_pending_job(job_id)
@@ -268,11 +271,15 @@ def schedule_step(self, force_update_jobs: bool = False) -> None:
                     # before the last reboot.
                     self.remove_job_no_lock(job_id)
                     continue
+
                 if submit:
-                    # Next job waiting for resources
+                    pending_submit_cnt += 1
+                if pending_submit_cnt >= _MAX_PENDING_SUBMIT:
+                    # There are too many pending jobs, wait for a while.
                     return
-                self._run_job(job_id, run_cmd)
-                return
+                if not submit:
+                    self._run_job(job_id, run_cmd)
+                    pending_submit_cnt += 1
 
     def _get_pending_job_ids(self) -> List[int]:
         """Returns the job ids in the pending jobs table