Changeset 48 for trunk/pbs_drmaa

Timestamp:

12/07/11 18:00:32 (13 years ago)

Author:

mmamonski

Message:

node_properties and custom_resources support. Now one can request resources via env variable

Location:

trunk/pbs_drmaa

Files:

• log_reader.c (modified) (9 diffs)
• log_reader.h (modified) (2 diffs)
• pbs_attrib.gperf (modified) (4 diffs)
• pbs_attrib.h (modified) (3 diffs)
• submit.c (modified) (9 diffs)

trunk/pbs_drmaa/log_reader.c

@@ -72 +72 @@
 static void pbsdrmaa_read_log();
 
-static void pbsdrmaa_select_file_wait_thread( pbsdrmaa_log_reader_t * self);
-
-char *pbsdrmaa_read_line_wait_thread( pbsdrmaa_log_reader_t * self);
+static void pbsdrmaa_select_file( pbsdrmaa_log_reader_t * self);
+
+static void pbsdrmaa_close_log( pbsdrmaa_log_reader_t * self);
+
+static void pbsdrmaa_reopen_log( pbsdrmaa_log_reader_t * self);
 
 static time_t pbsdrmaa_parse_log_timestamp(const char *timestamp, char *unixtime_str, size_t size);
@@ -141 +143 @@
                 self->session = session;
 
-                self->select_file = pbsdrmaa_select_file_wait_thread;
+                self->select_file = pbsdrmaa_select_file;
                 self->read_log = pbsdrmaa_read_log;     
+                self->close = pbsdrmaa_close_log;
+                self->reopen = pbsdrmaa_reopen_log;
                 
                 self->run_flag = true;
@@ -148 +152 @@
                 self->date_changed = true;
                 self->first_open = true;
+                self->log_path = NULL;
+                self->current_offset = 0;
                 
         }
@@ -458 +464 @@
                                 fsd_mutex_unlock( &self->session->mutex );
 
+                                /* close */
+                                self->close(self);
+
                                 sleep(((pbsdrmaa_session_t *)self->session)->wait_thread_sleep_time);
+
+                                /* and reopen log file */
+                                self->reopen(self);
 
                                 fsd_mutex_lock( &self->session->mutex );
@@ -490 +502 @@
 
 void
-pbsdrmaa_select_file_wait_thread ( pbsdrmaa_log_reader_t * self )
+pbsdrmaa_select_file( pbsdrmaa_log_reader_t * self )
 {
         pbsdrmaa_session_t *pbssession = (pbsdrmaa_session_t*) self->session;
@@ -496 +508 @@
         if (self->date_changed)
          {
-                char * log_path = NULL;
                 int num_tries = 0;
                 struct tm tm; 
@@ -511 +522 @@
                 #define DRMAA_WAIT_THREAD_MAX_TRIES (12)
                 /* generate new date, close file and open new */
-                log_path = fsd_asprintf("%s/server_logs/%04d%02d%02d", pbssession->pbs_home, tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday);
+                if (self->log_path)
+                        fsd_free(self->log_path);
+
+                self->log_path = fsd_asprintf("%s/server_logs/%04d%02d%02d", pbssession->pbs_home, tm.tm_year + 1900, tm.tm_mon + 1, tm.tm_mday);
 
                 if(self->fhandle)
                         fclose(self->fhandle);
 
-                fsd_log_info(("Opening log file: %s",log_path));
+                fsd_log_info(("Opening log file: %s",self->log_path));
                                 
         retry:
-                if ((self->fhandle = fopen(log_path,"r")) == NULL && (num_tries > DRMAA_WAIT_THREAD_MAX_TRIES || self->first_open))
+                if ((self->fhandle = fopen(self->log_path,"r")) == NULL && (num_tries > DRMAA_WAIT_THREAD_MAX_TRIES || self->first_open))
                  {
-                        fsd_log_error(("Can't open log file. Verify pbs_home. Running standard wait_thread."));
+                        fsd_log_error(("Can't open log file: %s. Verify pbs_home. Running standard wait_thread.", self->log_path));
                         fsd_log_error(("Remember that without keep_completed set the standard wait_thread won't provide information about job exit status"));
                         /*pbssession->super.enable_wait_thread = false;*/ /* run not wait_thread */
@@ -530 +544 @@
                 else if ( self->fhandle == NULL )
                  { /* Torque seems not to create a new file immediately after the old one is closed */
-                        fsd_log_warning(("Can't open log file: %s. Retries count: %d", log_path, num_tries));
+                        fsd_log_warning(("Can't open log file: %s. Retries count: %d", self->log_path, num_tries));
                         num_tries++;
                         sleep(2 * num_tries);
                         goto retry;
                  }
-
-                fsd_free(log_path);
 
                 fsd_log_debug(("Log file opened"));
@@ -653 +665 @@
 }
 
-
+void
+pbsdrmaa_close_log( pbsdrmaa_log_reader_t * self )
+{
+
+        self->current_offset = ftello(self->fhandle);
+
+        fclose(self->fhandle);
+
+        self->fhandle = NULL;
+}
+
+void
+pbsdrmaa_reopen_log( pbsdrmaa_log_reader_t * self )
+{
+        if ((self->fhandle = fopen(self->log_path,"r")) == NULL)
+         {
+                fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"Failed to reopen log file");
+         }
+
+        if(fseek(self->fhandle, self->current_offset, SEEK_SET) == (off_t) -1)
+         {
+                fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"fseek error");
+         }
+}
+

trunk/pbs_drmaa/log_reader.h

@@ -45 +45 @@
         void (*select_file) ( pbsdrmaa_log_reader_t * self );
         
+        void (*close) ( pbsdrmaa_log_reader_t * self );
+
+        void (*reopen) ( pbsdrmaa_log_reader_t * self );
+
+
         /* determines if function should run */
         bool run_flag;
@@ -59 +64 @@
         /* for wait_thread - log file first open */
         bool volatile first_open;       
+
+        char *volatile log_path;
+
+        off_t volatile current_offset;
 };
 

trunk/pbs_drmaa/pbs_attrib.gperf

@@ -68 +68 @@
 Resource_List.host, t(PBSDRMAA_ATTR_HOST)
 Resource_List.nodes, t(PBSDRMAA_ATTR_NODES)
-Resource_List.software, t(PBSDRMAA_ATTR_SOFTWARE)
 Resource_List.procs, t(PBSDRMAA_ATTR_PROCS)
 Resource_List.ncpus, t(PBSDRMAA_ATTR_NCPUS)
@@ -103 +102 @@
 submit_args, t(PBSDRMAA_ATTR_SUBMIT_ARGS)
 mtime, t(PBSDRMAA_ATTR_MTIME)
-# reservation attributes:
-# reserve_start,       t(PBSDRMAA_ATTR_RESERVATION_START)
-# reserve_end,         t(PBSDRMAA_ATTR_RESERVATION_END)
-# reserve_duration,    t(PBSDRMAA_ATTR_RESERVATION_DURATION)
-# reserve_state,       t(PBSDRMAA_ATTR_RESERVATION_STATE)
-# reserve_substate,    t(PBSDRMAA_ATTR_RESERVATION_SUBSTATE)
+pbsdrmaa.node_properties, t(PBSDRMAA_ATTR_NODE_PROPERTIES)
+pbsdrmaa.custom_resources, t(PBSDRMAA_ATTR_CUSTOM_RESOURCES)
 %%
 #undef t
@@ -187 +182 @@
         { "Resource_List.host", PBSDRMAA_ATTR_HOST, false },
         { "Resource_List.nodes", PBSDRMAA_ATTR_NODES, false },
-        { "Resource_List.software", PBSDRMAA_ATTR_SOFTWARE, false },
         { "Resource_List.procs", PBSDRMAA_ATTR_PROCS, false },
         { "Resource_List.ncpus", PBSDRMAA_ATTR_NCPUS, false },
@@ -221 +215 @@
         { "submit_args", PBSDRMAA_ATTR_SUBMIT_ARGS, false },
         { "mtime", PBSDRMAA_ATTR_MTIME, false },
+        { "pbsdrmaa.node_properties", PBSDRMAA_ATTR_NODE_PROPERTIES, false },
+        { "pbsdrmaa.custom_resources", PBSDRMAA_ATTR_CUSTOM_RESOURCES, false },
+        
 };
 

trunk/pbs_drmaa/pbs_attrib.h

@@ -53 +53 @@
 #define PBSDRMAA_PROCS                  "Resource_List.procs"
 #define PBSDRMAA_NCPUS                  "Resource_List.ncpus"
-#define PBSDRMAA_SOFTWARE               "Resource_List.software"
 #define PBSDRMAA_MAIL_POINTS            "Mail_Points"
 #define PBSDRMAA_OUTPUT_PATH            "Output_Path"
@@ -85 +84 @@
 #define PBSDRMAA_SUBMIT_ARGS            "submit_args"
 #define PBSDRMAA_MTIME                  "mtime"
-
+#define PBSDRMAA_NODE_PROPERTIES        "pbsdrmaa.node_properties"
+#define PBSDRMAA_CUSTOM_RESOURCES       "pbsdrmaa.custom_resources"
 
 typedef enum {
@@ -142 +142 @@
         PBSDRMAA_ATTR_SUBMIT_ARGS,
         PBSDRMAA_ATTR_MTIME,
+        PBSDRMAA_ATTR_NODE_PROPERTIES,
+        PBSDRMAA_ATTR_CUSTOM_RESOURCES,
 
         PBSDRMAA_N_PBS_ATTRIBUTES

trunk/pbs_drmaa/submit.c

@@ -49 +49 @@
 #endif
 
-static void
-pbsdrmaa_submit_destroy( pbsdrmaa_submit_t *self );
-
-static char *
-pbsdrmaa_submit_submit( pbsdrmaa_submit_t *self );
-
-static void
-pbsdrmaa_submit_eval( pbsdrmaa_submit_t *self );
-
-
-static void
-pbsdrmaa_submit_set( pbsdrmaa_submit_t *self, const char *pbs_attr,
-                char *value, unsigned placeholders );
+static void pbsdrmaa_submit_destroy( pbsdrmaa_submit_t *self );
+
+static char *pbsdrmaa_submit_submit( pbsdrmaa_submit_t *self );
+
+static void pbsdrmaa_submit_eval( pbsdrmaa_submit_t *self );
+
+static void pbsdrmaa_submit_set( pbsdrmaa_submit_t *self, const char *pbs_attr, char *value, unsigned placeholders );
 
 static void pbsdrmaa_submit_apply_defaults( pbsdrmaa_submit_t *self );
@@ -101 +95 @@
                 self->apply_job_environment = pbsdrmaa_submit_apply_job_environment;
                 self->apply_email_notification = pbsdrmaa_submit_apply_email_notification;
-                self->apply_native_specification =
-                        pbsdrmaa_submit_apply_native_specification;
+                self->apply_native_specification = pbsdrmaa_submit_apply_native_specification;
 
                 self->pbs_job_attributes = pbsdrmaa_pbs_template_new();
@@ -143 +136 @@
         TRY
          {
-                const fsd_template_t *pbs_tmpl = self->pbs_job_attributes;
-                unsigned i;
+                fsd_template_t *pbs_tmpl = self->pbs_job_attributes;
+                int i;
                 int tries_left = ((pbsdrmaa_session_t *)self->session)->max_retries_count;
                 int sleep_time = 1;
 
-                for( i = 0;  i < PBSDRMAA_N_PBS_ATTRIBUTES;  i++ )
+                for( i = PBSDRMAA_N_PBS_ATTRIBUTES - 1; i >= 0; i-- ) /* down loop -> start with custom resources */
                  {
                         const char *name = pbs_tmpl->by_code( pbs_tmpl, i )->name;
-                        if( name  &&  name[0] != '!' && pbs_tmpl->get_attr( pbs_tmpl, name ) )
+                        const char *value = pbs_tmpl->get_attr( pbs_tmpl, name );
+
+                        if (!value)
+                                continue;
+
+                        if ( i == PBSDRMAA_ATTR_CUSTOM_RESOURCES)
                          {
-                                const char *value;
-
-                                value = pbs_tmpl->get_attr( pbs_tmpl, name );
+                                char *value_copy = fsd_strdup(value);
+                                char *tok_comma_ctx = NULL;
+                                char *res_token = NULL;
+                                /* matlab:2,simulink:1 */
+
+                                for (res_token = strtok_r(value_copy, ",", &tok_comma_ctx); res_token; res_token = strtok_r(NULL, ",", &tok_comma_ctx))
+                                 {
+                                        char *value_p = strstr(res_token, ":");
+
+                                        if (value_p)
+                                         {
+                                                char *name_p = NULL;
+                                                *value_p = '\0';
+                                                value_p++;
+                                                name_p = fsd_asprintf("Resource_List.%s",res_token);
+                                                pbs_attr = pbsdrmaa_add_attr( pbs_attr, name_p, value_p );
+                                                fsd_free(name_p);
+                                         }
+                                        else
+                                         {
+                                                fsd_exc_raise_code( FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE );
+                                         }
+                                 }
+
+                                fsd_free(value_copy);
+                         }
+                        else if (i == PBSDRMAA_ATTR_NODE_PROPERTIES)
+                         {
+                                const char *nodes_value = pbs_tmpl->get_attr( pbs_tmpl, PBSDRMAA_NODES );
+                                char *final_value = NULL;
+
+                                if (!nodes_value)
+                                 {
+                                        final_value = fsd_asprintf("%s:%s",nodes_value, value);
+                                 }
+                                else
+                                 {
+                                        final_value = fsd_asprintf("1:%s", value);
+                                 }
+
+                                pbs_tmpl->set_attr( pbs_tmpl, PBSDRMAA_NODES, final_value);
+                                fsd_free(final_value);
+                         }
+                        else
+                         {
                                 pbs_attr = pbsdrmaa_add_attr( pbs_attr, name, value );
                          }
@@ -448 +488 @@
 }
 
+
+
 void
 pbsdrmaa_submit_apply_job_environment( pbsdrmaa_submit_t *self )
@@ -471 +513 @@
 
         if (env_v)
-        {
+         {
                 ii = 0;
-                while (env_v[ii]) {
+                while (env_v[ii])
+                 {
                         len += strlen(env_v[ii]) + 1;
                         ii++;
-                }
-        }
+                 }
+         }
         
         len+= strlen("PBS_O_WORKDIR=") + strlen(wd);
@@ -531 +574 @@
 }
 
-static void parse_resources(fsd_template_t *pbs_attr,const char *resources)
+static const char *get_job_env(pbsdrmaa_submit_t *self, const char *env_name)
+{
+        const fsd_template_t *jt = self->job_template;
+        const char *const *env_v = jt->get_v_attr( jt, DRMAA_V_ENV);
+        int ii = 0;
+
+        while (env_v[ii])
+         {
+                char *eq_p = strstr(env_v[ii], "=");
+
+                if ((eq_p) && (strncmp(env_v[ii], env_name, eq_p - env_v[ii]) == 0))
+                                return ++eq_p;
+
+                ii++;
+         }
+
+        return NULL;
+}
+
+static void parse_resources(pbsdrmaa_submit_t *self, fsd_template_t *pbs_attr,const char *resources)
 {
         char * volatile name = NULL;
@@ -550 +612 @@
                                 name = fsd_asprintf("Resource_List.%s", arg);
                                 value = ++psep;
-                                pbs_attr->set_attr( pbs_attr, name , value );
+                                if (value[0] == '$' && get_job_env(self, value + 1))
+                                        pbs_attr->set_attr( pbs_attr, name , get_job_env(self, value + 1) ); /*get value from job env variable */
+                                else
+                                        pbs_attr->set_attr( pbs_attr, name , value );
                                 fsd_free(name);
                                 name = NULL;
@@ -600 +665 @@
                 const char *native_specification )
 {
-        fsd_log_enter(( "({native_specification=%s})", native_specification ));
+        fsd_log_enter(( "({native_specification=%s})", native_specification ));
 
         if( native_specification == NULL )
@@ -700 +765 @@
                                                         break;
                                                 case 'l' :
-                                                        parse_resources(pbs_attr, arg);
+                                                        parse_resources( self, pbs_attr, arg);
                                                         break;                                                  
                                                 default :