EH: CS-777: Improve qsub -sync so that is support multiple state (like r)

ernst-bablick · ernst-bablick · commit d7a33b850aed · 2025-02-21T17:32:35.000+01:00
diff --git a/doc/markdown/man/man1/submit.include.md b/doc/markdown/man/man1/submit.include.md
@@ -1090,26 +1090,41 @@ defined JSV instances as parameter with the names *global_q_soft* and *global_l_
 Find more information in the sections describing *-q*, *-l* and *-scope*. (see *-jsv* option below or find more
 information concerning JSV in xxqs_name_sxx_jsv(1))
 
-## -sync y\[es\]\|n\[o\]  
+## -sync r|x|n (or y\[es\]\|n\[o\])  
 
 Available for `qsub`.
 
-`-sync y` causes `qsub` to wait for the job to complete before exiting. If the job completes successfully, 
+The `-sync` option *yes* and *no* are deprecated. Use *r*, *x* and *n* instead. `qsub -sync yes` is equivalent to
+`qsub -sync x` and `qsub -sync no` is equivalent to `qsub -sync n`.
+
+`-sync` causes `qsub` to wait for one or multiple job states to be reached before exiting. The option argument
+specifies the job state(s) to wait for. 
+
+If `-sync r` is used `qsub` will wait for the job to leave the *qw* state before it terminates. 
+If `-sync r` is used in conjunction with `-t n[-m[:s]]`, `qsub` will wait for at least one task of the array job to 
+leave the queues-waiting state. A task is considered to leave the queues-waiting state if it enters the running 
+state or if it is deleted before start.
+
+`-sync x` causes `qsub` to wait for the job to complete before exiting. If the job completes successfully, 
 The exit code of `qsub` will be that of the completed job. If the job fails to complete successfully,
 `qsub` will print out an error message indicating why the job failed and will have an exit code of 1. If `qsub` 
-is interrupted, e.g. with CTRL-C, before the job completes, the job will be canceled. With the `-sync n` option, 
-`qsub` will exit with an exit code of 0 as soon as the job is submitted successfully. `-sync n` is default for
-`qsub`.
-
-If `-sync y` is used in conjunction with `-now y`, `qsub` will behave as though only `-now y` were given until the 
-job has been successfully scheduled, after which time `qsub` will behave as though only `-sync y` were given.  
-If `-sync y` is used in conjunction with `-t n[-m[:s]]`, `qsub` will wait for all the job's tasks to complete 
-before exiting. If all the job's tasks complete successfully, The exit code ob `qsub` will be that of the first 
-completed job tasks with a non-zero exit code, or 0 if all job tasks exited with an exit code of 0. If any of the 
-job's tasks fail to complete successfully, `qsub` will print out an error message indicating why the job task(s) 
-failed and will have an exit code of 1. If `qsub` is interrupted, e.g. with CTRL-C, before the job completes, 
+is interrupted, e.g. with CTRL-C, before the job completes, the job will be canceled.
+
+If `-sync x` is used in conjunction with `-t n[-m[:s]]`, `qsub` will wait for all the job's tasks to complete
+before exiting. If all the job's tasks complete successfully, The exit code ob `qsub` will be that of the first
+completed job tasks with a non-zero exit code, or 0 if all job tasks exited with an exit code of 0. If any of the
+job's tasks fail to complete successfully, `qsub` will print out an error message indicating why the job task(s)
+failed and will have an exit code of 1. If `qsub` is interrupted, e.g. with CTRL-C, before the job completes,
 all the job's tasks will be canceled.
 
+With the `-sync n` option (default if nothing else is specified), `qsub` will exit with an exit code of 0 as soon as 
+the job is submitted successfully. 
+
+If `-sync rx` is used, `qsub` will behave as though only `-sync r` were given until the job has been started,
+after which time `qsub` will behave as though only `-sync x` were given. If `-sync ...` is used in conjunction with 
+`-now y`, `qsub` will behave as though only `-now y` were given until the job has been successfully scheduled, 
+after which time `qsub` will behave as though only `-sync ...` were given.  
+
 Information that this switch was specified during submission is not available in the JSV context. (see `-jsv` option 
 above or find more information concerning JSV in xxqs_name_sxx_jsv(1))
 
diff --git a/source/clients/qsub/msg_qsub.h b/source/clients/qsub/msg_qsub.h
@@ -55,5 +55,6 @@
 #define MSG_QSUB_INTERRUPTED                 _MESSAGE(210015, _("Interrupted!"))
 #define MSG_QSUB_TERMINATING                 _MESSAGE(210016, _("Please wait while qsub shuts down."))
 #define MSG_QSUB_COULDNOTREADSCRIPT_S        _MESSAGE(210017, _("Unable to read script file because of error: "))
+#define MSG_QSUB_JOBHASSTARTED_S             _MESSAGE(210018, _("Job " SFN " has started.\n"))
 
 // clang-format on
diff --git a/source/clients/qsub/ocs_qsub.cc b/source/clients/qsub/ocs_qsub.cc
@@ -47,22 +47,24 @@
 #include "sgeobj/cull/sge_all_listsL.h"
 #include "sgeobj/sge_answer.h"
 #include "sgeobj/sge_job.h"
+#include "sgeobj/sge_daemonize.h"
 
 #include "comm/commlib.h"
 
 #include "japi/japi.h"
 #include "japi/japiP.h"
 
 #include "gdi/sge_security.h"
-#include "sgeobj/sge_daemonize.h"
 #include "gdi/sge_gdi.h"
+#include "gdi/ocs_gdi_ClientBase.h"
 
 #include "sig_handlers.h"
 #include "basis_types.h"
 #include "usage.h"
 #include "parse_job_cull.h"
 #include "ocs_client_parse.h"
 #include "ocs_client_job.h"
+#include "parse_qsub.h"
 #include "msg_clients_common.h"
 #include "msg_qsub.h"
 #include "msg_qmaster.h"
@@ -93,7 +95,7 @@ main(int argc, const char **argv)
    int exit_status = 0;
    int just_verify;
    int tmp_ret;
-   int wait_for_job = 0, is_immediate = 0;
+   int is_immediate = 0;
    dstring session_key_out = DSTRING_INIT;
    dstring diag = DSTRING_INIT;
    dstring jobid = DSTRING_INIT;
@@ -208,9 +210,12 @@ main(int argc, const char **argv)
    /* If "-sync y" is set, wait for the job to end. */
    /* Remove all -sync switches since cull_parse_job_parameter()
     * doesn't know what to do with them. */
+   bool wait_for_job = false;
+   u_long32 sync_opt = SYNC_NO;
    while ((ep = lGetElemStrRW(opts_all, SPA_switch_val, "-sync"))) {
-      if (lGetInt(ep, SPA_argval_lIntT) == TRUE) {
-         wait_for_job = 1;
+      sync_opt = lGetUlong(ep, SPA_argval_lUlongT);
+      if (sync_opt != SYNC_NO) {
+         wait_for_job = true;
       }
 
       lRemoveElem(opts_all, &ep);
@@ -416,39 +421,53 @@ main(int argc, const char **argv)
          }
       }
 
+      // We have to wait for certain job states
       if (wait_for_job) {
-         /* Rather than using japi_synchronize on ALL for bulk jobs, we use
-          * japi_wait on ANY num_tasks times because with synchronize, we would
-          * have to wait for all the tasks to finish before we know if any
-          * finished. */
-         for (count = 0; count < num_tasks; count++) {
-            /* Since there's only one running job in the session, we can just
-             * wait for ANY. */
-            if ((tmp_ret = japi_wait(DRMAA_JOB_IDS_SESSION_ANY, &jobid, &stat,
-                          DRMAA_TIMEOUT_WAIT_FOREVER, JAPI_JOB_FINISH, &event,
-                          nullptr, &diag)) != DRMAA_ERRNO_SUCCESS) {
-               if ((tmp_ret != DRMAA_ERRNO_EXIT_TIMEOUT) &&
-                   (tmp_ret != DRMAA_ERRNO_NO_ACTIVE_SESSION)) {
+
+         // JOB START: just wait for the first task to start
+         if ((sync_opt & SYNC_JOB_START) == SYNC_JOB_START) {
+            tmp_ret = japi_wait(DRMAA_JOB_IDS_SESSION_ANY, &jobid, &stat, DRMAA_TIMEOUT_WAIT_FOREVER, JAPI_JOB_START, &event, nullptr, &diag);
+
+            if (tmp_ret != DRMAA_ERRNO_SUCCESS) {
+               if (tmp_ret != DRMAA_ERRNO_EXIT_TIMEOUT && tmp_ret != DRMAA_ERRNO_NO_ACTIVE_SESSION) {
                   fprintf(stderr, "\n");
                   fprintf(stderr, MSG_QSUB_COULDNOTWAITFORJOB_S, sge_dstring_get_string(&diag));
                   fprintf(stderr, "\n");
                }
 
                exit_status = 1;
                goto Error;
+            } else {
+               printf(MSG_QSUB_JOBHASSTARTED_S, sge_dstring_get_string(&jobid));
             }
+         }
 
-            /* report how job finished */
-            /* If the job is an array job, use the first non-zero exit code as
-             * the exit code for qsub. */
-            if (exit_status == 0) {
-               exit_status = report_exit_status(stat,
-                                              sge_dstring_get_string(&jobid));
-            }
-            /* If we've already found a non-zero exit code, just print the exit
-             * info for the task. */
-            else {
-               report_exit_status(stat, sge_dstring_get_string(&jobid));
+         // JOB END: Now wait for the end of *all* tasks
+         if ((sync_opt & SYNC_JOB_END) == SYNC_JOB_END) {
+            for (count = 0; count < num_tasks; count++) {
+               // Rather than using japi_synchronize on ALL for bulk jobs, we use japi_wait on ANY num_tasks times because with synchronize, we would
+               // have to wait for all the tasks to finish before we know if any finished.
+               // Since there's only one running job in the session, we can just  wait for ANY.
+               tmp_ret = japi_wait(DRMAA_JOB_IDS_SESSION_ANY, &jobid, &stat, DRMAA_TIMEOUT_WAIT_FOREVER, JAPI_JOB_FINISH, &event, nullptr, &diag);
+               if (tmp_ret != DRMAA_ERRNO_SUCCESS) {
+                  if (tmp_ret != DRMAA_ERRNO_EXIT_TIMEOUT && tmp_ret != DRMAA_ERRNO_NO_ACTIVE_SESSION) {
+                     fprintf(stderr, "\n");
+                     fprintf(stderr, MSG_QSUB_COULDNOTWAITFORJOB_S, sge_dstring_get_string(&diag));
+                     fprintf(stderr, "\n");
+                  }
+
+                  exit_status = 1;
+                  goto Error;
+               }
+
+               // report how job finished
+               if (exit_status == 0) {
+                  // If the job is an array job, use the first non-zero exit code as the exit code for qsub.
+                  exit_status = report_exit_status(stat, sge_dstring_get_string(&jobid));
+               } else {
+                  // If we've already found a non-zero exit code, just print the exit info for the task.
+                  report_exit_status(stat, sge_dstring_get_string(&jobid));
+               }
             }
          }
       }
diff --git a/source/common/msg_common.h b/source/common/msg_common.h
@@ -851,7 +851,7 @@
 #define MSG_GDI_USAGE_Aattr_OPT                          "[-Aattr obj_nm fname obj_id_list]"
 #define MSG_GDI_UTEXT_Aattr_OPT                          _MESSAGE(23483, _("add to a list attribute of an object")) 
 
-#define MSG_GDI_USAGE_sync_OPT_YN                        "[-sync y[es]|n[o]]"
+#define MSG_GDI_USAGE_sync_OPT_YN                        "[-sync r|x|n]"
 #define MSG_GDI_UTEXT_sync_OPT_YN                        _MESSAGE(23484, _("wait for job to end and return exit code"))
 
 #define MSG_GDI_USAGE_JQ_DEST_OPR                        "job_queue_list"
diff --git a/source/common/parse_qsub.cc b/source/common/parse_qsub.cc
@@ -68,9 +68,8 @@
 
 static int var_list_parse_from_environment(lList **lpp, char **envp);
 static int sge_parse_checkpoint_interval(const char *time_str);
-static int set_yn_option (lList **opts, u_long32 opt, const char *arg, const char *value,
-                          lList **alpp);
-
+static int set_yn_option (lList **opts, u_long32 opt, const char *arg, const char *value, lList **alpp);
+static int ocs_parse_sync_switch(lList **opts, u_long32 opt, const char *arg, const char *value, lList **alpp);
 
 
 /*
@@ -1571,23 +1570,21 @@ DTRACE;
 
 /*----------------------------------------------------------------------------*/
      /*  -sync y[es]|n[o] */
+     /*  -sync r|x|n */
 
-      if(!strcmp("-sync", *sp)) {
+      if (!strcmp("-sync", *sp)) {
          if (lGetElemStr(*pcmdline, SPA_switch_val, *sp)) {
-            answer_list_add_sprintf(&answer, STATUS_EEXIST, ANSWER_QUALITY_WARNING,
-                    MSG_PARSE_XOPTIONALREADYSETOVERWRITINGSETING_S, *sp );
+            answer_list_add_sprintf(&answer, STATUS_EEXIST, ANSWER_QUALITY_WARNING, MSG_PARSE_XOPTIONALREADYSETOVERWRITINGSETING_S, *sp);
          }
-         /* next field is yes/no switch */
          sp++;
          if(!*sp) {
-            answer_list_add_sprintf(&answer, STATUS_ESEMANTIC, ANSWER_QUALITY_ERROR,
-                    MSG_PARSE_XOPTIONMUSTHAVEARGUMENT_S, "-sync" );
+            answer_list_add_sprintf(&answer, STATUS_ESEMANTIC, ANSWER_QUALITY_ERROR, MSG_PARSE_XOPTIONMUSTHAVEARGUMENT_S, "-sync");
             DRETURN(answer);
          }
          
          DPRINTF("\"-sync %s\"\n", *sp);
          
-         if (set_yn_option (pcmdline, sync_OPT, *(sp - 1), *sp, &answer) != STATUS_OK) {
+         if (ocs_parse_sync_switch(pcmdline, sync_OPT, *(sp - 1), *sp, &answer) != STATUS_OK) {
             DRETURN(answer);
          }
 
@@ -2222,6 +2219,74 @@ static int set_yn_option (lList **opts, u_long32 opt, const char *arg, const cha
    return STATUS_OK;
 }
 
+/** @brief Parse the -sync option switch
+ *
+ * Supports the old and new format of the -sync option switch.
+ * Old format: -sync y(es)|n(o)
+ * New format: -sync r|y|x|E|n
+ *
+ *    r - job start
+ *    x or y - job end
+ *    n - no sync
+ *
+ * @param opts list of options
+ * @param opt option code
+ * @param arg option argument
+ * @param value option value
+ * @param alpp answer list
+ *
+ * @return STATUS_OK if success, STATUS_ERROR1 otherwise
+ */
+static int
+ocs_parse_sync_switch(lList **opts, u_long32 opt, const char *arg, const char *value, lList **alpp) {
+   DENTER(TOP_LAYER);
+   u_long32 sync_bits = SYNC_UNINITIALIZED;
+
+   // -sync yes|no (pre OCS format )
+   if (strcasecmp("yes", value) == 0) {
+      sync_bits = SYNC_JOB_END;
+   } else if (strcasecmp ("no", value) == 0) {
+      sync_bits = SYNC_NO;
+   }
+
+   // -sync r|y|x|E|n (new OCS format)
+   if (sync_bits == SYNC_UNINITIALIZED) {
+      const int len = strlen(value);
+      for (int pos = 0; pos < len; pos++) {
+         switch (value[pos]) {
+         case 'r':
+            sync_bits |= SYNC_JOB_START;
+            break;
+         case 'x':
+         case 'y':
+            sync_bits |= SYNC_JOB_END;
+            break;
+         case 'n':
+            sync_bits |= SYNC_NO;
+            break;
+         default:
+            snprintf(SGE_EVENT, SGE_EVENT_SIZE, MSG_PARSE_INVALIDOPTIONARGUMENT_SS, arg, value);
+            answer_list_add(alpp, SGE_EVENT, STATUS_ESYNTAX, ANSWER_QUALITY_ERROR);
+            DRETURN(STATUS_ERROR1);
+         }
+      }
+   }
+
+   // check if SYNC_NO was combined with other bits which might make no sense
+   if ((sync_bits & SYNC_NO) != 0 && sync_bits != SYNC_NO) {
+      snprintf(SGE_EVENT, SGE_EVENT_SIZE, MSG_PARSE_INVALIDOPTIONARGUMENT_SS, arg, value);
+      answer_list_add(alpp, SGE_EVENT, STATUS_ESYNTAX, ANSWER_QUALITY_ERROR);
+      DRETURN(STATUS_ERROR1);
+   }
+
+   // transport bits via opts list
+   lListElem *ep_opt = nullptr;
+   ep_opt = sge_add_arg(opts, opt, lUlongT, arg, value);
+   lSetUlong(ep_opt,SPA_argval_lUlongT, sync_bits);
+
+   DRETURN(STATUS_OK);
+}
+
 /* This method is not thread safe.  Fortunately, it is only used by the
  * -cwd switch which can be forbidden in DRMAA. */
 char *reroot_path(lListElem* pjob, const char *path, lList **alpp) {
diff --git a/source/common/parse_qsub.h b/source/common/parse_qsub.h
@@ -58,6 +58,13 @@
 #define FLG_USE_PSEUDOS 1
 #define FLG_QALTER      2
 
+typedef enum {
+   SYNC_UNINITIALIZED = 0x00000000,
+   SYNC_NO            = 0x00000001,
+   SYNC_JOB_END       = 0x00000002,
+   SYNC_JOB_START     = 0x00000004,
+} sync_switch_t;
+
 /* I've added a -wd option to cull_parse_job_parameter() to deal with the
  * DRMAA_WD attribute.  It makes sense to me that since -wd exists and is
  * handled by cull_parse_job_parameter() that -cwd should just become an alias
