Changeset 85 for trunk

Timestamp:

01/17/13 18:44:15 (11 years ago)

Author:

mmamonski

Message:

PBS DRMAA autoclose connection

Location:

trunk/pbs_drmaa

Files:

• job.c (modified) (4 diffs)
• log_reader.c (modified) (2 diffs)
• pbs_conn.c (modified) (32 diffs)
• pbs_conn.h (modified) (3 diffs)
• session.c (modified) (9 diffs)
• session.h (modified) (3 diffs)
• submit.c (modified) (5 diffs)
• util.c (modified) (2 diffs)
• util.h (modified) (1 diff)

trunk/pbs_drmaa/job.c

@@ -77 +77 @@
 pbsdrmaa_job_control( fsd_job_t *self, int action )
 {
-        volatile bool conn_lock = false;
         pbsdrmaa_session_t *session = (pbsdrmaa_session_t*)self->session;
         const char *job_id = self->job_id;
-        const char *apicall = NULL;
-        int rc = PBSE_NONE;
-
-        fsd_log_enter(( "({job_id=%s}, action=%d)",
-                        self->job_id, action ));
-
-        TRY
-         {
-                int tries_left = session->max_retries_count;
-                int sleep_time = 1;
-
-                conn_lock = fsd_mutex_lock( &self->session->drm_connection_mutex );
-
-                /*TODO reconnect */
-                while ( true )
-                 {
-                        switch( action )
-                         {
-                                /*
-                                 * We cannot know whether we did suspend job
-                                 * in other way than remembering this inside DRMAA session.
-                                 */
-                                case DRMAA_CONTROL_SUSPEND:
-                                        apicall = "pbs_sigjob";
-                                        rc = pbs_sigjob( session->pbs_conn, (char*)job_id,
-                                                        "SIGSTOP", NULL );
-                                        fsd_log_info(("pbs_sigjob(%s, SIGSTOP) =%d", job_id, rc));
-                                        if( rc == PBSE_NONE )
-                                                self->flags |= FSD_JOB_SUSPENDED;
-                                        break;
-                                case DRMAA_CONTROL_RESUME:
-                                        apicall = "pbs_sigjob";
-                                        rc = pbs_sigjob( session->pbs_conn, (char*)job_id,
-                                                        "SIGCONT", NULL );
-                                        fsd_log_info(("pbs_sigjob(%s, SIGCONT) =%d", job_id, rc));
-                                        if( rc == PBSE_NONE )
-                                                self->flags &= ~FSD_JOB_SUSPENDED;
-                                        break;
-                                case DRMAA_CONTROL_HOLD:
-                                        apicall = "pbs_holdjob";
-                                        rc = pbs_holdjob( session->pbs_conn, (char*)job_id,
-                                                        USER_HOLD, NULL );
-                                        fsd_log_info(("pbs_sigjob(%s, SIGHOLD) =%d", job_id, rc));
-                                        if( rc == PBSE_NONE )
-                                                self->flags |= FSD_JOB_HOLD;
-                                        break;
-                                case DRMAA_CONTROL_RELEASE:
-                                        apicall = "pbs_rlsjob";
-                                        rc = pbs_rlsjob( session->pbs_conn, (char*)job_id,
-                                                        USER_HOLD, NULL );
-                                        fsd_log_info(("pbs_rlsjob(%s) =%d", job_id, rc));
-                                        if( rc == PBSE_NONE )
-                                                self->flags &= FSD_JOB_HOLD;
-                                        break;
-                                case DRMAA_CONTROL_TERMINATE:
-                                        apicall = "pbs_deljob";
-                                        rc = pbs_deljob( session->pbs_conn, (char*)job_id, NULL );
-                                        fsd_log_info(("pbs_deljob(%s) =%d", job_id, rc));
-                                        /* Torque:
-                                         * deldelay=N -- delay between SIGTERM and SIGKILL (default 0) */
-                                        if( rc == PBSE_NONE )
-                                         {
-                                                self->flags &= FSD_JOB_TERMINATED_MASK;
-                                                if( (self->flags & FSD_JOB_TERMINATED) == 0 )
-                                                        self->flags |= FSD_JOB_TERMINATED | FSD_JOB_ABORTED;
-                                         }
-                                        break;
-                         }
-
-retry_connect:
-                        if ( rc == PBSE_NONE )
-                                break;
-                        else if (( rc == PBSE_INTERNAL || rc == PBSE_PROTOCOL || rc == PBSOLDE_PROTOCOL || rc == PBSE_EXPIRED || rc == PBSOLDE_EXPIRED) && (tries_left--))
-                         {
-                                if (rc == PBSE_PROTOCOL || rc == PBSE_EXPIRED || rc == PBSOLDE_PROTOCOL || rc == PBSOLDE_EXPIRED)
-                                 {
-                                        if ( session->pbs_conn >= 0)
-                                                pbs_disconnect( session->pbs_conn );
-
-                                        sleep( sleep_time++ );
-
-                                        session->pbs_conn = pbs_connect( session->super.contact );
-
-                                        if (session->pbs_conn < 0)
-                                                goto retry_connect;
-
-                                        fsd_log_info(( "pbs_connect(%s) =%d", session->super.contact, session->pbs_conn ));
-                                 }
-                                else /* PBSE_INTERNAL */
-                                 {
-                                        /*
-                                         * In PBS Pro pbs_sigjob raises internal server error (PBSE_INTERNAL)
-                                         * when job just changed its state to running.
-                                         */
-                                        sleep( sleep_time++ );
-                                 }
-                                fsd_log_debug(( "repeating request (%d of %d)", tries_left, session->max_retries_count));
-                         }
-                        else
-                                pbsdrmaa_exc_raise_pbs( apicall );
-                 } /* end while */
-         }
-        FINALLY
-         {
-                if( conn_lock )
-                        conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
-         }
-        END_TRY
+
+        fsd_log_enter(( "({job_id=%s}, action=%d)", self->job_id, action ));
+
+        while ( true )
+         {
+                switch( action )
+                 {
+                        /*
+                         * We cannot know whether we did suspend job
+                         * in other way than remembering this inside DRMAA session.
+                         */
+                        case DRMAA_CONTROL_SUSPEND:
+                                session->pbs_connection->sigjob( session->pbs_connection, (char*)job_id, "SIGSTOP");
+                                self->flags |= FSD_JOB_SUSPENDED;
+                                break;
+                        case DRMAA_CONTROL_RESUME:
+                                session->pbs_connection->sigjob( session->pbs_connection, (char*)job_id, "SIGCONT");
+                                self->flags &= ~FSD_JOB_SUSPENDED;
+                                break;
+                        case DRMAA_CONTROL_HOLD:
+                                session->pbs_connection->holdjob( session->pbs_connection, (char*)job_id );
+                                self->flags |= FSD_JOB_HOLD;
+                                break;
+                        case DRMAA_CONTROL_RELEASE:
+                                session->pbs_connection->rlsjob( session->pbs_connection, (char*)job_id );
+                                self->flags &= ~FSD_JOB_HOLD;
+                                break;
+                        case DRMAA_CONTROL_TERMINATE:
+                                session->pbs_connection->deljob( session->pbs_connection, (char*)job_id );
+                                /* TODO: make deldelay configurable ???:
+                                 * deldelay=N -- delay between SIGTERM and SIGKILL (default 0) */
+                                self->flags &= FSD_JOB_TERMINATED_MASK;
+                                if( (self->flags & FSD_JOB_TERMINATED) == 0 )
+                                        self->flags |= FSD_JOB_TERMINATED | FSD_JOB_ABORTED;
+                                break;
+                 }
+         }
 
         fsd_log_return((""));
@@ -196 +124 @@
 pbsdrmaa_job_update_status( fsd_job_t *self )
 {
-        volatile bool conn_lock = false;
         struct batch_status *volatile status = NULL;
         pbsdrmaa_session_t *session = (pbsdrmaa_session_t*)self->session;
-        int tries_left = session->max_retries_count;
-        int sleep_time = 1;
 
         fsd_log_enter(( "({job_id=%s})", self->job_id ));
@@ -206 +131 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock( &self->session->drm_connection_mutex );
-retry:
-                if (session->pbs_conn < 0) {
-                        fsd_log_info(("No connection with pbs. Reconnecting"));
-                        goto retry_connect;
-                }
-
 
 #ifdef PBS_PROFESSIONAL
-                status = pbs_statjob( session->pbs_conn, self->job_id, NULL, NULL );
+                status = session->pbs_connection->statjob( session->pbs_connection, self->job_id, NULL);
 #else
-                status = pbs_statjob( session->pbs_conn, self->job_id, session->status_attrl, NULL );
+                status = session->pbs_connection->statjob( session->pbs_connection, self->job_id, session->status_attrl);
 #endif
-                fsd_log_info(( "pbs_statjob(fd=%d, job_id=%s, attribs={...}) =%p",
-                                 session->pbs_conn, self->job_id, (void*)status ));
-                if( status == NULL )
-                 {
-
-#ifndef PBS_PROFESSIONAL
-                        if ( pbs_errno != PBSE_UNKJOBID )
-                                fsd_log_error(("pbs_statjob error: %d, %s, %s", pbs_errno, pbse_to_txt(pbs_errno), pbs_strerror(pbs_errno)));
-                        else
-                                fsd_log_debug(("pbs_statjob error: %d, %s, %s", pbs_errno, pbse_to_txt(pbs_errno), pbs_strerror(pbs_errno)));
-#else
-#  ifndef PBS_PROFESSIONAL_NO_LOG
-                        if ( pbs_errno != PBSE_UNKJOBID && pbs_errno != PBSE_HISTJOBID )
-                                fsd_log_error(("pbs_statjob error: %d, %s", pbs_errno, pbse_to_txt(pbs_errno)));
-                        else
-                                fsd_log_debug(("pbs_statjob error: %d, %s", pbs_errno, pbse_to_txt(pbs_errno)));
-#  else
-                        if ( pbs_errno != PBSE_UNKJOBID && pbs_errno != PBSE_HISTJOBID )
-                                fsd_log_error(("pbs_statjob error: %d", pbs_errno));
-                        else
-                                fsd_log_debug(("pbs_statjob error: %d", pbs_errno));
-#  endif
-#endif
-
-                        switch( pbs_errno )
-                         {
-                                case PBSE_UNKJOBID:
-#ifdef PBS_PROFESSIONAL
-                                case PBSE_HISTJOBID:
-#endif
-                                        break;
-                                case PBSE_PROTOCOL:
-#if PBSOLDE_PROTOCOL != PBSE_PROTOCOL
-                                case PBSOLDE_PROTOCOL:
-#endif
-                                case PBSE_EXPIRED:
-#if PBSOLDE_EXPIRED != PBSE_EXPIRED
-                                case PBSOLDE_EXPIRED:
-#endif
-                                        if ( session->pbs_conn >= 0 )
-                                                pbs_disconnect( session->pbs_conn );
-                                        fsd_log_info(("Protocol error. Reconnecting..."));
-retry_connect:
-                                        sleep(sleep_time++);
-                                        session->pbs_conn = pbs_connect( session->super.contact );
-                                        if( session->pbs_conn < 0 )
-                                         {
-                                                if (tries_left--) {
-                                                        fsd_log_info(("Retrying... (%d tries left)", tries_left));
-                                                        goto retry_connect;
-                                                } else {
-                                                        fsd_log_error(("No more tries left... Throwing exception"));
-                                                        pbsdrmaa_exc_raise_pbs( "pbs_connect" );
-                                                }
-                                         }
-                                        else
-                                         {
-                                                goto retry;
-                                         }
-                                default:
-                                        pbsdrmaa_exc_raise_pbs( "pbs_statjob" );
-                                        break;
-                                case 0:  /* ? */
-                                        fsd_exc_raise_code( FSD_ERRNO_INTERNAL_ERROR );
-                                        break;
-                         }
-
-                 }
-
-                conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
-
 
                 if( status != NULL )
@@ -302 +149 @@
         FINALLY
          {
-                if( conn_lock )
-                        conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
                 if( status != NULL )
-                        pbs_statfree( status );
+                        session->pbs_connection->statjob_free( session->pbs_connection, status );
          }
         END_TRY

trunk/pbs_drmaa/log_reader.c

@@ -693 +693 @@
         self->current_offset = ftello(self->fhandle);
         
-        fsd_log_debug(("Closing log  file (offset=%d)", self->current_offset));  
+        fsd_log_debug(("Closing log  file (offset=%d)", (int)self->current_offset));
 
         fclose(self->fhandle);
@@ -703 +703 @@
 pbsdrmaa_reopen_log( pbsdrmaa_log_reader_t * self )
 {
-        fsd_log_debug(("Reopening log file: %s (offset=%d)", self->log_path, self->current_offset));  
+        fsd_log_debug(("Reopening log file: %s (offset=%d)", self->log_path, (int)self->current_offset));
 
         if ((self->fhandle = fopen(self->log_path,"r")) == NULL)

trunk/pbs_drmaa/pbs_conn.c

@@ -28 +28 @@
 #include <drmaa_utils/iter.h>
 #include <drmaa_utils/conf.h>
-#include <drmaa_utils/session.h>
 #include <drmaa_utils/datetime.h>
 
+#include <pbs_drmaa/session.h>
 #include <pbs_drmaa/pbs_conn.h>
 #include <pbs_drmaa/util.h>
@@ -53 +53 @@
 static void pbsdrmaa_pbs_holdjob( pbsdrmaa_pbs_conn_t *self,  char *job_id );
 
-static void pbsdrmaa_pbs_reconnect_internal( pbsdrmaa_pbs_conn_t *self, bool reconnect);
-
-static void pbsdrmaa_pbs_check_connect_internal( pbsdrmaa_pbs_conn_t *self, bool reconnect);
-
+static void pbsdrmaa_pbs_connection_autoclose_thread_loop( pbsdrmaa_pbs_conn_t *self, bool reconnect);
+
+
+static void check_reconnect( pbsdrmaa_pbs_conn_t *self, bool reconnect);
+
+static void start_autoclose_thread( pbsdrmaa_pbs_conn_t *self );
+
+static void stop_autoclose_thread( pbsdrmaa_pbs_conn_t *self );
+
+
+#if defined PBS_PROFESSIONAL && defined PBSE_HISTJOBID
+        #define IS_MISSING_JOB (pbs_errno == PBSE_UNKJOBID || pbs_errno == PBSE_HISTJOBID)
+#else
+        #define IS_MISSING_JOB (pbs_errno == PBSE_UNKJOBID)
+#endif
 #define IS_TRANSIENT_ERROR (pbs_errno == PBSE_PROTOCOL || pbs_errno == PBSE_EXPIRED || pbs_errno == PBSOLDE_PROTOCOL || pbs_errno == PBSOLDE_EXPIRED)
 
-        
 pbsdrmaa_pbs_conn_t * 
-pbsdrmaa_pbs_conn_new( pbsdrmaa_session_t *session, char *server )
+pbsdrmaa_pbs_conn_new( fsd_drmaa_session_t *session, const char *server )
 {
         pbsdrmaa_pbs_conn_t *volatile self = NULL;
@@ -84 +94 @@
 
                 self->connection_fd = -1;
-                self->last_usage = time(NULL);
 
                 /*ignore SIGPIPE - otherwise pbs_disconnect cause the program to exit */
                 signal(SIGPIPE, SIG_IGN);       
 
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                check_reconnect(self, false);
           }
         EXCEPT_DEFAULT
@@ -100 +109 @@
                         if (self->connection_fd != -1)
                                 pbs_disconnect(self->connection_fd);
+                        stop_autoclose_thread(self);
                   }
                         
@@ -110 +120 @@
         return self;
 }
-
 
 void
@@ -148 +157 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock(&self->session->super.drm_connection_mutex);
-
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                conn_lock = fsd_mutex_lock(&self->session->drm_connection_mutex);
+
+                check_reconnect(self, false);
 
 retry:
@@ -162 +171 @@
                         if (IS_TRANSIENT_ERROR && first_try)
                          {
-                                pbsdrmaa_pbs_reconnect_internal(self, true);
+                                check_reconnect(self, true);
                                 first_try = false;
                                 goto retry;
@@ -168 +177 @@
                         else
                          {
-                                pbsdrmaa_exc_raise_pbs( "pbs_submit");
+                                pbsdrmaa_exc_raise_pbs( "pbs_submit", self->connection_fd);
                          }
                  }
@@ -180 +189 @@
          {
                 if(conn_lock)
-                        conn_lock = fsd_mutex_unlock(&self->session->super.drm_connection_mutex);
+                        conn_lock = fsd_mutex_unlock(&self->session->drm_connection_mutex);
          }
         END_TRY
@@ -202 +211 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock(&self->session->super.drm_connection_mutex);
-
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                conn_lock = fsd_mutex_lock(&self->session->drm_connection_mutex);
+
+                check_reconnect(self, false);
 
 retry:
@@ -213 +222 @@
                 if(status == NULL)
                  {
-                        fsd_log_error(( "pbs_statjob failed, pbs_errno = %d", pbs_errno ));
-                        if (IS_TRANSIENT_ERROR && first_try)
-                         {
-                                pbsdrmaa_pbs_reconnect_internal(self, true);
+                        if (IS_MISSING_JOB)
+                         {
+                                fsd_log_info(( "missing job = %s (code=%d)", job_id, pbs_errno ));
+                         }
+                        else if (IS_TRANSIENT_ERROR && first_try)
+                         {
+                                fsd_log_error(( "pbs_statjob failed, pbs_errno = %d", pbs_errno ));
+                                check_reconnect(self, true);
                                 first_try = false;
                                 goto retry;
@@ -222 +235 @@
                         else
                          {
-                                pbsdrmaa_exc_raise_pbs( "pbs_statjob");
+                                pbsdrmaa_exc_raise_pbs( "pbs_statjob", self->connection_fd);
                          }
                  }
@@ -236 +249 @@
          {
                 if(conn_lock)
-                        conn_lock = fsd_mutex_unlock(&self->session->super.drm_connection_mutex);
+                        conn_lock = fsd_mutex_unlock(&self->session->drm_connection_mutex);
          }
         END_TRY
@@ -266 +279 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock(&self->session->super.drm_connection_mutex);
-
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                conn_lock = fsd_mutex_lock(&self->session->drm_connection_mutex);
+
+                check_reconnect(self, false);
 
 retry:
@@ -280 +293 @@
                         if (IS_TRANSIENT_ERROR && first_try)
                          {
-                                pbsdrmaa_pbs_reconnect_internal(self, true);
+                                check_reconnect(self, true);
                                 first_try = false;
                                 goto retry;
@@ -286 +299 @@
                         else
                          {
-                                pbsdrmaa_exc_raise_pbs( "pbs_sigjob");
+                                pbsdrmaa_exc_raise_pbs( "pbs_sigjob", self->connection_fd);
                          }
                  }
@@ -297 +310 @@
          {
                 if(conn_lock)
-                        conn_lock = fsd_mutex_unlock(&self->session->super.drm_connection_mutex);
+                        conn_lock = fsd_mutex_unlock(&self->session->drm_connection_mutex);
          }
         END_TRY
@@ -318 +331 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock(&self->session->super.drm_connection_mutex);
-
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                conn_lock = fsd_mutex_lock(&self->session->drm_connection_mutex);
+
+                check_reconnect(self, false);
 
 retry:
@@ -332 +345 @@
                         if (IS_TRANSIENT_ERROR && first_try)
                          {
-                                pbsdrmaa_pbs_reconnect_internal(self, true);
+                                check_reconnect(self, true);
                                 first_try = false;
                                 goto retry;
@@ -338 +351 @@
                         else
                          {
-                                pbsdrmaa_exc_raise_pbs( "pbs_deljob");
+                                pbsdrmaa_exc_raise_pbs( "pbs_deljob", self->connection_fd);
                          }
                  }
@@ -349 +362 @@
          {
                 if(conn_lock)
-                        conn_lock = fsd_mutex_unlock(&self->session->super.drm_connection_mutex);
+                        conn_lock = fsd_mutex_unlock(&self->session->drm_connection_mutex);
          }
         END_TRY
@@ -369 +382 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock(&self->session->super.drm_connection_mutex);
-
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                conn_lock = fsd_mutex_lock(&self->session->drm_connection_mutex);
+
+                check_reconnect(self, false);
 
 retry:
@@ -383 +396 @@
                         if (IS_TRANSIENT_ERROR && first_try)
                          {
-                                pbsdrmaa_pbs_reconnect_internal(self, true);
+                                check_reconnect(self, true);
                                 first_try = false;
                                 goto retry;
@@ -389 +402 @@
                         else
                          {
-                                pbsdrmaa_exc_raise_pbs( "pbs_rlsjob");
+                                pbsdrmaa_exc_raise_pbs( "pbs_rlsjob", self->connection_fd);
                          }
                  }
@@ -400 +413 @@
          {
                 if(conn_lock)
-                        conn_lock = fsd_mutex_unlock(&self->session->super.drm_connection_mutex);
+                        conn_lock = fsd_mutex_unlock(&self->session->drm_connection_mutex);
          }
         END_TRY
@@ -420 +433 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock(&self->session->super.drm_connection_mutex);
-
-                pbsdrmaa_pbs_reconnect_internal(self, false);
+                conn_lock = fsd_mutex_lock(&self->session->drm_connection_mutex);
+
+                check_reconnect(self, false);
 
 retry:
@@ -434 +447 @@
                         if (IS_TRANSIENT_ERROR && first_try)
                          {
-                                pbsdrmaa_pbs_reconnect_internal(self, true);
+                                check_reconnect(self, true);
                                 first_try = false;
                                 goto retry;
@@ -440 +453 @@
                         else
                          {
-                                pbsdrmaa_exc_raise_pbs( "pbs_holdjob");
+                                pbsdrmaa_exc_raise_pbs( "pbs_holdjob", self->connection_fd);
                          }
                  }
@@ -451 +464 @@
          {
                 if(conn_lock)
-                        conn_lock = fsd_mutex_unlock(&self->session->super.drm_connection_mutex);
+                        conn_lock = fsd_mutex_unlock(&self->session->drm_connection_mutex);
          }
         END_TRY
@@ -460 +473 @@
 
 void 
-pbsdrmaa_pbs_reconnect_internal( pbsdrmaa_pbs_conn_t *self, bool force_reconnect)
-{
-        int tries_left = self->session->max_retries_count;
+check_reconnect( pbsdrmaa_pbs_conn_t *self, bool force_reconnect)
+{
+        int tries_left = ((pbsdrmaa_session_t *)self->session)->max_retries_count;
         int sleep_time = 1;
 
@@ -476 +489 @@
                 else
                  {
+                        stop_autoclose_thread(self);
                         pbs_disconnect(self->connection_fd);
                         self->connection_fd = -1;
                  }
           }
+
+
 
 retry_connect: /* Life... */
@@ -492 +508 @@
         
         if( self->connection_fd < 0 )
-                pbsdrmaa_exc_raise_pbs( "pbs_connect" );
+                pbsdrmaa_exc_raise_pbs( "pbs_connect", self->connection_fd );
         
         fsd_log_return(("(%d)", self->connection_fd));
 }
 
+
+static void start_autoclose_thread( pbsdrmaa_pbs_conn_t *self )
+{
+
+
+}
+
+static void stop_autoclose_thread( pbsdrmaa_pbs_conn_t *self )
+{
+
+
+}
+

trunk/pbs_drmaa/pbs_conn.h

@@ -29 +29 @@
 #include <drmaa_utils/job.h>
 #include <drmaa_utils/session.h>
+#include <drmaa_utils/thread.h>
 
-#include <session.h>
 
 #include <pbs_ifl.h>
@@ -36 +36 @@
 typedef struct pbsdrmaa_pbs_conn_s pbsdrmaa_pbs_conn_t;
 
-pbsdrmaa_pbs_conn_t * pbsdrmaa_pbs_conn_new ( pbsdrmaa_session_t * session, char *server);
+pbsdrmaa_pbs_conn_t *
+pbsdrmaa_pbs_conn_new(
+                fsd_drmaa_session_t * session,
+                const char *server);
 
-void
-pbsdrmaa_pbs_conn_destroy ( pbsdrmaa_pbs_conn_t * self );
+void pbsdrmaa_pbs_conn_destroy ( pbsdrmaa_pbs_conn_t * self );
 
 struct pbsdrmaa_pbs_conn_s {
-        pbsdrmaa_session_t *volatile session;
+        fsd_drmaa_session_t *volatile session;
 
         char* (*submit) ( pbsdrmaa_pbs_conn_t *self, struct attropl *attrib, char *script, char *destination );
@@ -63 +65 @@
         int connection_fd;
         
-        /* timestamp of last usage */
-        time_t last_usage;      
+        /* timestamp of last connect time */
+        time_t last_connect_time;
+
+        fsd_cond_t autoclose_cond;
+        fsd_mutex_t autoclose_mutex;
+        bool close_connection;
 };
 

trunk/pbs_drmaa/session.c

@@ -98 +98 @@
 
                 self->log_file_initial_size = 0;
-                self->pbs_conn = -1;
                 self->pbs_home = NULL;
 
@@ -123 +122 @@
                 self->super.missing_jobs = FSD_IGNORE_MISSING_JOBS;
 
-                 {
-                        int tries_left = self->max_retries_count;
-                        int sleep_time = 1;
-                        /*ignore SIGPIPE - otheriwse pbs_disconnect cause the program to exit */
-                        signal(SIGPIPE, SIG_IGN);
-retry_connect: /* Life... */
-                        self->pbs_conn = pbs_connect( self->super.contact );
-                        fsd_log_info(( "pbs_connect(%s) =%d", self->super.contact, self->pbs_conn ));
-                        if( self->pbs_conn < 0 && tries_left-- )
-                         {
-                                sleep(sleep_time++);
-                                goto retry_connect;
-                         }
-
-                        if( self->pbs_conn < 0 )
-                                pbsdrmaa_exc_raise_pbs( "pbs_connect" );
-                 }
+                self->pbs_connection = pbsdrmaa_pbs_conn_new( (fsd_drmaa_session_t *)self, contact );
+                self->connection_max_lifetime =  30; /* 30 seconds */
 
          }
@@ -162 +146 @@
         pbsdrmaa_session_t *pbsself = (pbsdrmaa_session_t*)self;
         self->stop_wait_thread( self );
-        if( pbsself->pbs_conn >= 0 )
-                pbs_disconnect( pbsself->pbs_conn );
+        pbsdrmaa_pbs_conn_destroy(pbsself->pbs_connection);
         fsd_free( pbsself->status_attrl );
         fsd_free( pbsself->job_exit_status_file_prefix );
@@ -229 +212 @@
         fsd_conf_option_t *max_retries_count = NULL;
         fsd_conf_option_t *user_state_dir = NULL;
+        fsd_conf_option_t *connection_max_lifetime = NULL;
+
 
         pbs_home = fsd_conf_dict_get(self->configuration, "pbs_home" );
@@ -234 +219 @@
         max_retries_count = fsd_conf_dict_get(self->configuration, "max_retries_count" );
         user_state_dir = fsd_conf_dict_get(self->configuration, "user_state_dir" );
+        connection_max_lifetime = fsd_conf_dict_get(self->configuration, "connection_max_lifetime");
 
         if( pbs_home && pbs_home->type == FSD_CONF_STRING )
@@ -274 +260 @@
           }
 
+        if ( connection_max_lifetime && connection_max_lifetime->type == FSD_CONF_INTEGER)
+          {
+                pbsself->connection_max_lifetime = connection_max_lifetime->val.integer;
+                fsd_log_info(("Max connection lifetime: %d", pbsself->connection_max_lifetime));
+          }
+
         if ( wait_thread_sleep_time && wait_thread_sleep_time->type == FSD_CONF_INTEGER)
           {
@@ -315 +307 @@
 pbsdrmaa_session_update_all_jobs_status( fsd_drmaa_session_t *self )
 {
-        volatile bool conn_lock = false;
         volatile bool jobs_lock = false;
         pbsdrmaa_session_t *pbsself = (pbsdrmaa_session_t*)self;
         fsd_job_set_t *jobs = self->jobs;
         struct batch_status *volatile status = NULL;
-        volatile int tries_left = pbsself->max_retries_count;
-        volatile int sleep_time = 1;
 
         fsd_log_enter((""));
@@ -327 +316 @@
         TRY
          {
-                conn_lock = fsd_mutex_lock( &self->drm_connection_mutex );
-retry:
+
 /* TODO: query only for user's jobs pbs_selstat + ATTR_u */
 #ifdef PBS_PROFESSIONAL
-                status = pbs_statjob( pbsself->pbs_conn, NULL, NULL, NULL );
+                status = pbsself->pbs_connection->statjob(pbsself->pbs_connection, NULL, NULL);
 #else
-                status = pbs_statjob( pbsself->pbs_conn, NULL, pbsself->status_attrl, NULL );
+                status = pbsself->pbs_connection->statjob(pbsself->pbs_connection, NULL, pbsself->status_attrl);
 #endif
-                fsd_log_info(( "pbs_statjob( fd=%d, job_id=NULL, attribs={...} ) =%p", pbsself->pbs_conn, (void*)status ));
-                if( status == NULL  &&  pbs_errno != 0 )
-                 {
-                        if (pbs_errno == PBSE_PROTOCOL || pbs_errno == PBSE_EXPIRED || pbs_errno == PBSOLDE_PROTOCOL || pbs_errno == PBSOLDE_EXPIRED)
-                         {
-                                if ( pbsself->pbs_conn >= 0)
-                                        pbs_disconnect( pbsself->pbs_conn );
-retry_connect:
-                                sleep(sleep_time++);
-                                pbsself->pbs_conn = pbs_connect( pbsself->super.contact );
-                                if( pbsself->pbs_conn < 0)
-                                 {
-                                        if (tries_left--)
-                                                goto retry_connect;
-                                        else
-                                                pbsdrmaa_exc_raise_pbs( "pbs_connect" );
-                                 }
-                                else
-                                        goto retry;
-                         }
-                        else
-                         {
-                                pbsdrmaa_exc_raise_pbs( "pbs_statjob" );
-                         }
-                 }
-                conn_lock = fsd_mutex_unlock( &self->drm_connection_mutex );
 
                  {
@@ -421 +383 @@
          {
                 if( status != NULL )
-                        pbs_statfree( status );
-                if( conn_lock )
-                        conn_lock = fsd_mutex_unlock( &self->drm_connection_mutex );
+                         pbsself->pbs_connection->statjob_free(pbsself->pbs_connection, status );
                 if( jobs_lock )
                         jobs_lock = fsd_mutex_unlock( &jobs->mutex );

trunk/pbs_drmaa/session.h

@@ -27 +27 @@
 #include <drmaa_utils/session.h>
 
+#include <pbs_drmaa/pbs_conn.h>
+
 typedef struct pbsdrmaa_session_s pbsdrmaa_session_t;
 
-fsd_drmaa_session_t *
-pbsdrmaa_session_new( const char *contact );
+fsd_drmaa_session_t *pbsdrmaa_session_new( const char *contact );
 
 struct pbsdrmaa_session_s {
@@ -45 +46 @@
 
         /*
-         * PBS connection (or -1 if not connected).
-         * A descriptor of socket conencted to PBS server.
+         * PBS connection handle
          */
-        int pbs_conn;
+        pbsdrmaa_pbs_conn_t *pbs_connection;
 
         /*
@@ -91 +91 @@
 
         /*
-         * Whether to cache PBS Connection
+         * PBS Max connection time
          */
-        bool cache_connection;
+        int connection_max_lifetime;
 };
 

trunk/pbs_drmaa/submit.c

@@ -136 +136 @@
 pbsdrmaa_submit_submit( pbsdrmaa_submit_t *self )
 {
-        volatile bool conn_lock = false;
         struct attrl *volatile pbs_attr = NULL;
         char *volatile job_id = NULL;
@@ -143 +142 @@
                 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 = PBSDRMAA_N_PBS_ATTRIBUTES - 1; i >= 0; i-- ) /* down loop -> start with custom resources */
@@ -208 +205 @@
                 pbs_attr = pbsdrmaa_submit_filter(pbs_attr);
 
-                conn_lock = fsd_mutex_lock( &self->session->drm_connection_mutex );
-retry:
-                job_id = pbs_submit( ((pbsdrmaa_session_t*)self->session)->pbs_conn,
-                                (struct attropl*)pbs_attr, self->script_filename,
-                                self->destination_queue, NULL );
+                job_id = ((pbsdrmaa_session_t *)self->session)->pbs_connection->submit( ((pbsdrmaa_session_t *)self->session)->pbs_connection, (struct attropl*)pbs_attr, self->script_filename, self->destination_queue);
 
                 fsd_log_info(("pbs_submit(%s, %s) =%s", self->script_filename, self->destination_queue, job_id));
 
-                if( job_id == NULL )
-                {
-                        fsd_log_error(( "pbs_submit failed, pbs_errno = %d", pbs_errno ));
-                        if (pbs_errno == PBSE_PROTOCOL || pbs_errno == PBSE_EXPIRED || pbs_errno == PBSOLDE_PROTOCOL || pbs_errno == PBSOLDE_EXPIRED)
-                         {
-                                pbsdrmaa_session_t *pbsself = (pbsdrmaa_session_t*)self->session;
-
-                                fsd_log_error(( "Protocol error. Retrying..." ));
-
-                                if (pbsself->pbs_conn >= 0 )
-                                        pbs_disconnect( pbsself->pbs_conn );
-retry_connect:
-                                sleep(sleep_time++);
-                                pbsself->pbs_conn = pbs_connect( pbsself->super.contact );
-                                if( pbsself->pbs_conn < 0) 
-                                 {
-                                        if (tries_left--)
-                                                goto retry_connect;
-                                        else
-                                                pbsdrmaa_exc_raise_pbs( "pbs_connect" );
-                                 }
-                                else
-                                 {
-                                        if (tries_left--)
-                                                goto retry;
-                                        else
-                                                pbsdrmaa_exc_raise_pbs( "pbs_submit" );
-                                 }
-                         }
-                        else
-                         {
-                                pbsdrmaa_exc_raise_pbs( "pbs_submit" );
-                         }
-                }
-                conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
          }
         EXCEPT_DEFAULT
@@ -259 +217 @@
         FINALLY
          {
-                if( conn_lock )
-                        conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
                 if( pbs_attr )
                         pbsdrmaa_free_attrl( pbs_attr );
@@ -894 +850 @@
                         if (!attr_value)
                           {
-                                fsd_exc_raise_fmt(FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_FORMAT, "Invalid output line of submit filter:", output_line);
+                                fsd_exc_raise_fmt(FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_FORMAT, "Invalid output line of submit filter: %s", output_line);
                           }
                         else

trunk/pbs_drmaa/util.c

@@ -114 +114 @@
 
 void
-pbsdrmaa_exc_raise_pbs( const char *function )
+pbsdrmaa_exc_raise_pbs( const char *function, int connection )
 {
         int _pbs_errno;
         int fsd_errno;
         const char *message = NULL;
+        const char *extended_message = NULL;
 
         _pbs_errno = pbs_errno;
@@ -128 +129 @@
 #endif
 
+        if ( connection != -1 )
+         {
+                extended_message = pbs_geterrmsg(connection);
+         }
+
         fsd_errno = pbsdrmaa_map_pbs_errno( _pbs_errno );
-        fsd_log_error((
-                                "call to %s returned with error %d:%s mapped to %d:%s",
-                                function,
-                                _pbs_errno, message,
-                                fsd_errno, fsd_strerror(fsd_errno)
-                                ));
-        fsd_exc_raise_fmt( fsd_errno, " %s", function, message );
-}
-
+
+        fsd_log_error(( "call to %s returned with error %d:%s(%s) mapped to %d:%s",
+                                        function,
+                                        _pbs_errno, message, extended_message,
+                                        fsd_errno, fsd_strerror(fsd_errno)
+                        ));
+
+        if (extended_message)
+                fsd_exc_raise_fmt(fsd_errno, "%s: %s ", message, extended_message);
+        else
+                fsd_exc_raise_fmt(fsd_errno, "%s", message);
+}
 
 /** Maps PBS error code into DMRAA code. */

trunk/pbs_drmaa/util.h

@@ -30 +30 @@
 
 
-void pbsdrmaa_exc_raise_pbs( const char *function );
+void pbsdrmaa_exc_raise_pbs( const char *function, int connection_fd );
 int pbsdrmaa_map_pbs_errno( int _pbs_errno );