Changeset 26 for trunk

Timestamp:

11/11/12 15:06:16 (13 years ago)

Author:

mmamonski

Message:

SupMUC on site fixes: 1. Polling mode 2. Handling missing jobs 3. monitor -> drmaa_monitor 4. force stderr file creation

Location:

trunk

Files:

• ll_drmaa/Makefile.am (modified) (1 diff)
• ll_drmaa/drmaa.c (modified) (1 diff)
• ll_drmaa/job.c (modified) (6 diffs)
• ll_drmaa/job.h (modified) (1 diff)
• ll_drmaa/session.c (modified) (6 diffs)
• m4/ax_ll.m4 (modified) (3 diffs)

trunk/ll_drmaa/Makefile.am

@@ -33 +33 @@
 libdrmaa_la_LDFLAGS = @LL_LDFLAGS@ -version-info @LL_DRMAA_VERSION_INFO@
 
-bin_PROGRAMS = monitor
+bin_PROGRAMS = lldrmaa_monitor
 
-monitor_SOURCES = monitor.c
+lldrmaa_monitor_SOURCES = monitor.c
 
-monitor_CPPFLAGS = -I$(top_srcdir)/drmaa_utils
+lldrmaa_monitor_CPPFLAGS = -I$(top_srcdir)/drmaa_utils
 
 dist_sysconf_DATA = ll_drmaa.conf.example

trunk/ll_drmaa/drmaa.c

@@ -17 +17 @@
  */
 
-#include <sys/timers.h>
 #include <sys/time.h>
 #include <sys/wait.h>

trunk/ll_drmaa/job.c

@@ -35 +35 @@
 
 #include <llapi.h>
+
+#define LL_DRMAA_MAX_MISSING_TIME (60)
 
 static void
@@ -287 +289 @@
                 }
 
-                data = ll_get_objs(job, LL_SCHEDD, NULL, &obj_count, &err_code);
+                data = ll_get_objs(job, LL_CM, NULL, &obj_count, &err_code);
                 if (data == NULL) {
-                        fsd_log_debug(("Code: %d ll_get_objs() returns NULL. %s",lldrmaa_map_get_objs(err_code), lldrmaa_err_get_objs(err_code) ));
-                        /* This error means that there is no info in LL_SCHEDD but this job have probably ended and will be detected by wait_thread. This won't be fsd_log_warning because displays too often during program execution */
+                        if (err_code == LL_GET_OBJS_NO_OBJECTS_ERR)
+                          {
+                                if (self->state != DRMAA_PS_UNDETERMINED) 
+                                  {
+                                        fsd_log_info(("Job %s missing. Assuming finished", self->job_id));
+                                        self->state = DRMAA_PS_DONE;
+                                        self->exit_status = 0;
+                                  }
+                                else if (llself->missing_time == 0)
+                                 {
+                                        llself->missing_time = time(NULL);
+                                        fsd_log_debug(("Job %s missing for the first time", self->job_id)); /* Job may not yet be visible in LL */
+                                 }
+                                else if (time(NULL) - llself->missing_time > LL_DRMAA_MAX_MISSING_TIME)
+                                 {
+                                        fsd_log_error(("Job %s missing for more then %d seconds. Assuming failef", self->job_id, LL_DRMAA_MAX_MISSING_TIME)); 
+                                        /* Job may not yet be visible in LL */
+                                        self->state = DRMAA_PS_FAILED;
+                                        self->exit_status = -1;
+                                 }
+                                else
+                                 {
+                                        fsd_log_debug(("Job %s still missing", self->job_id));
+                                 }
+                          }
+                        else
+                          {
+                                fsd_log_error(("Code: %d,%d ll_get_objs() returns NULL. %s", err_code, lldrmaa_map_get_objs(err_code), lldrmaa_err_get_objs(err_code) ));
+                          }
                 } else {
 
@@ -312 +341 @@
                         }
 
-                        llself->read_job_info( self, step_state, hold_type);
-
-                        fsd_log_info(("LL State: %d -> DRMAA: %s",step_state,drmaa_job_ps_to_str(self->state)));
+                        llself->read_job_info(self, step_state, hold_type);
+
+                        fsd_log_info(("LL State: %d -> DRMAA: %s", step_state, drmaa_job_ps_to_str(self->state)));
                 }
         }
@@ -352 +381 @@
         self->super.control = lldrmaa_job_control;
         self->super.update_status = lldrmaa_job_update_status;
+        self->missing_time = 0;
         self->read_job_info_mon = lldrmaa_job_read_job_info_mon;
         self->read_job_info = lldrmaa_job_read_job_info;
@@ -471 +501 @@
                 }
 
-                TRY
-                {
-                        const char *command = NULL;
-                        char *command_expanded = NULL;
-                        const char *const *i;
-                        int j;
-
-                        /* remote command */
-                        command = jt->get_attr( jt, DRMAA_REMOTE_COMMAND );
-                        if( command == NULL )
-                                fsd_exc_raise_msg(
-                                                FSD_DRMAA_ERRNO_CONFLICTING_ATTRIBUTE_VALUES,
-                                                "drmaa_remote_command not set for job template"
-                                                );
-
-                        command_expanded = expand->expand( expand, fsd_strdup(command), FSD_DRMAA_PH_HD | FSD_DRMAA_PH_WD );
-
-                        fprintf(fd,"# @ executable = %s\n", command_expanded);
-                        fsd_log_debug(("# @ executable = %s\n", command_expanded));
-
-                        fsd_free(command_expanded);
-
-                        /* arguments list */
-                        vector = jt->get_v_attr( jt, DRMAA_V_ARGV );
-
-                        if( vector )
-                        {
-                                fprintf(fd,"# @ arguments =");
-                                fsd_log_debug(("# @ arguments ="));
-
-                                for( i = vector, j = 2;  *i;  i++, j++ )
-                                {
-                                        char *arg_expanded = expand->expand( expand, fsd_strdup(*i), FSD_DRMAA_PH_HD | FSD_DRMAA_PH_WD );
-
-                                        fprintf(fd," '%s'", arg_expanded);
-                                        fsd_log_debug(("%s", arg_expanded));
-
-                                        fsd_free(arg_expanded);
-                                }
-                        }
-
-                        fprintf(fd," \n");
-                }
-                END_TRY
-
                 /* start time */
                 value = jt->get_attr( jt, DRMAA_START_TIME );
@@ -779 +764 @@
                                 fprintf(fd,"# @ queue\n");
                                 fsd_log_debug(("# @ queue"));
+                                
+                                fprintf(fd,"\n");
+                                fprintf(fd,"echo >&2\n"); /* this line forces creation of stderr file */
+
+                                TRY
+                                {
+                                        const char *command = NULL;
+                                        char *command_expanded = NULL;
+                                        const char *const *i;
+                                        int j;
+
+                                        /* remote command */
+                                        command = jt->get_attr( jt, DRMAA_REMOTE_COMMAND );
+                                        if( command == NULL )
+                                                fsd_exc_raise_msg(
+                                                                FSD_DRMAA_ERRNO_CONFLICTING_ATTRIBUTE_VALUES,
+                                                                "drmaa_remote_command not set for job template"
+                                                                );
+
+                                        command_expanded = expand->expand( expand, fsd_strdup(command), FSD_DRMAA_PH_HD | FSD_DRMAA_PH_WD );
+
+                                        fprintf(fd,"%s", command_expanded); /* we put the commmand at the end of script (instead of using @executable keyword
+                                                                                 in roder to avoid coping of the binary */
+                                        fsd_log_debug(("command = %s\n", command_expanded));
+
+                                        fsd_free(command_expanded);
+
+                                        /* arguments list */
+                                        vector = jt->get_v_attr( jt, DRMAA_V_ARGV );
+
+                                        if( vector )
+                                        {
+                                                fsd_log_debug(("arguments = "));
+
+                                                for( i = vector, j = 2;  *i;  i++, j++ )
+                                                {
+                                                        char *arg_expanded = expand->expand( expand, fsd_strdup(*i), FSD_DRMAA_PH_HD | FSD_DRMAA_PH_WD );
+
+                                                        fprintf(fd," '%s'", arg_expanded);
+                                                        fsd_log_debug(("%s", arg_expanded));
+
+                                                        fsd_free(arg_expanded);
+                                                }
+                                        }
+
+                                        fprintf(fd," \n");
+                                }
+                                END_TRY
+
+
                         }
                 }

trunk/ll_drmaa/job.h

@@ -36 +36 @@
 struct lldrmaa_job_s {
         fsd_job_t super;
+        time_t missing_time;
 
         void (*read_job_info_mon)( fsd_job_t *self, const char * state , unsigned status);

trunk/ll_drmaa/session.c

@@ -132 +132 @@
         LL_job job_info;
 
-        char *monitor_program = LL_DRMAA_BIN_DIR"/monitor";
+        char *monitor_program = LL_DRMAA_BIN_DIR"/lldrmaa_monitor";
 
         TRY
@@ -147 +147 @@
 
                 connection_lock = fsd_mutex_lock( &self->drm_connection_mutex );
-                status = llsubmit(cmd_path, monitor_program, llself->unix_socket_name, &job_info, LL_JOB_VERSION);
+                if (self->wait_thread_run_flag) 
+                  {
+                        fsd_log_info(("llsubmit(%s, %s, %s, %p, %d)",cmd_path, monitor_program, llself->unix_socket_name, (void*)&job_info, LL_JOB_VERSION));
+                        status = llsubmit(cmd_path, monitor_program, llself->unix_socket_name, &job_info, LL_JOB_VERSION);
+                  }
+                else
+                  {
+                        fsd_log_info(("llsubmit(%s, NULL, NULL, %p, %d)",cmd_path, (void*)&job_info, LL_JOB_VERSION));
+                        status = llsubmit(cmd_path, NULL, NULL, &job_info, LL_JOB_VERSION);
+                  }
                 connection_lock = fsd_mutex_unlock( &self->drm_connection_mutex );
 
-                if(remove(cmd_path) == -1)
+                if(getenv("LLDRMAA_KEEP_CMD") == NULL && remove(cmd_path) == -1)
                         fsd_log_warning(("Can't delete cmd file: %s", cmd_path));
 
@@ -161 +170 @@
                  }
                 else /* 0 */
+                 {
                         fsd_log_debug(("llsubmit: %s",lldrmaa_err_submit(status)));
+                 }
 
                 if( start != end )
@@ -170 +181 @@
                          {
                                 job_ids[i] = fsd_asprintf("%s.%d.%d", job_info.step_list[i]->id.from_host, job_info.step_list[i]->id.cluster, job_info.step_list[i]->id.proc);
-
+                                fsd_log_info((" new array job id: %s", job_ids[i]));
                                 job = lldrmaa_job_new( fsd_strdup(job_ids[i]) );
                                 job->session = self;
@@ -185 +196 @@
                         job_ids[0] = fsd_asprintf( "%s.%d.0", job_info.step_list[0]->id.from_host, job_info.step_list[0]->id.cluster);
 
+                        fsd_log_info((" new job id: %s", job_ids[0]));
                         job = lldrmaa_job_new( fsd_strdup(job_ids[0]) );
                         job->session = self;
@@ -247 +259 @@
          }
 
-        if ( !self->wait_thread_started )
-                fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR, "DRMAA for LL requires that wait thread is enable. Don't disable it in configuration file!" );
 }
 

trunk/m4/ax_ll.m4

@@ -45 +45 @@
         LL_INCLUDES="-I${with_ll_inc}"
 else
-        LLSUBMIT_PATH=`which llsubmit`
+        LLSUBMIT_PATH=`readlink -f $(which llsubmit)`
         if test x"$LLSUBMIT_PATH" != x; then
                 LLSUBMIT_DIR=`dirname $LLSUBMIT_PATH`
                 LL_HOME=`dirname $LLSUBMIT_DIR`
                 LL_INCLUDES="-I$LL_HOME/include"
+                AC_MSG_NOTICE([no --with-ll-inc given. Using llsubmit path based guess: $LL_INCLUDES])
         else
                 ax_ll_msg="no llsubmit in PATH"
@@ -61 +62 @@
 if test x$with_ll_lib == x; then
         
-        LLSUBMIT_PATH=`which llsubmit`
+        LLSUBMIT_PATH=`readlink -f $(which llsubmit)`
         
         if test x"$LLSUBMIT_PATH" != x; then
@@ -67 +68 @@
                 LL_HOME=`dirname $LLSUBMIT_DIR`
                 with_ll_lib=$LL_HOME/lib
+                AC_MSG_NOTICE([no --with-ll-lib given. Using llsubmit path based guess: $with_ll_lib])
         else
                 ax_ll_msg="no llsubmit in PATH"