Index: trunk/pbs_drmaa/job.c
===================================================================
--- trunk/pbs_drmaa/job.c	(revision 103)
+++ trunk/pbs_drmaa/job.c	(revision 1)
@@ -1,3 +1,3 @@
-/* $Id$ */
+/* $Id: job.c 370 2010-11-16 09:51:00Z mamonski $ */
 /*
  *  FedStage DRMAA for PBS Pro
@@ -26,5 +26,4 @@
 #include <string.h>
 #include <unistd.h>
-#include <sys/stat.h>
 
 #include <drmaa_utils/drmaa.h>
@@ -34,5 +33,4 @@
 
 #include <pbs_drmaa/job.h>
-#include <pbs_drmaa/log_reader.h>
 #include <pbs_drmaa/pbs_attrib.h>
 #include <pbs_drmaa/session.h>
@@ -44,19 +42,20 @@
 		__attribute__ ((unused))
 #	endif
-	= "$Id$";
+	= "$Id: job.c 370 2010-11-16 09:51:00Z mamonski $";
 #endif
 
 
-static void pbsdrmaa_job_control( fsd_job_t *self, int action );
-
-static void pbsdrmaa_job_update_status( fsd_job_t *self );
-
-static void pbsdrmaa_job_on_missing( fsd_job_t *self );
-
-static void pbsdrmaa_job_on_missing_standard( fsd_job_t *self );
-
-static void pbsdrmaa_job_update( fsd_job_t *self, struct batch_status* );
-
-static int pbsdrmaa_job_read_exit_status( const char *job_id, const char *job_state_dir_prefix);
+static void
+pbsdrmaa_job_control( fsd_job_t *self, int action );
+
+static void
+pbsdrmaa_job_update_status( fsd_job_t *self );
+
+static void
+pbsdrmaa_job_on_missing( fsd_job_t *self );
+
+static void
+pbsdrmaa_job_update( fsd_job_t *self, struct batch_status* );
+
 
 fsd_job_t *
@@ -68,5 +67,4 @@
 	self->super.update_status = pbsdrmaa_job_update_status;
 	self->super.on_missing = pbsdrmaa_job_on_missing;
-	self->missing_time = 0;
 	self->update = pbsdrmaa_job_update;
 	return (fsd_job_t*)self;
@@ -77,40 +75,93 @@
 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;
-
-	fsd_log_enter(( "({job_id=%s}, action=%d)", self->job_id, action ));
-
-	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;
-	 }
+	const char *apicall = NULL;
+	int rc = PBSE_NONE;
+
+	fsd_log_enter(( "({job_id=%s}, action=%d)",
+			self->job_id, action ));
+
+	TRY
+	 {
+		int try_count;
+		const int max_tries = 3;
+
+		conn_lock = fsd_mutex_lock( &self->session->drm_connection_mutex );
+
+		/*TODO reconnect */
+		for( try_count=0;  try_count < max_tries;  try_count++ )
+		 {
+			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 );
+					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 );
+					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 );
+					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 );
+					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 );
+					/* 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;
+			 }
+
+			if( rc == PBSE_NONE )
+				break;
+			else if( rc == PBSE_INTERNAL )
+			 {
+				/*
+				 * In PBS Pro pbs_sigjob raises internal server error (PBSE_INTERNAL)
+				 * when job just changed its state to running.
+				 */
+				fsd_log_debug(( "repeating request (%d of %d)",
+							try_count+2, max_tries ));
+				sleep( 1 );
+			 }
+			else
+				pbsdrmaa_exc_raise_pbs( apicall );
+		 } /* end for */
+	 }
+	FINALLY
+	 {
+		if( conn_lock )
+			conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
+	 }
+	END_TRY
 
 	fsd_log_return((""));
@@ -121,17 +172,53 @@
 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;
 
 	fsd_log_enter(( "({job_id=%s})", self->job_id ));
-	
 	TRY
 	 {
-
+		conn_lock = fsd_mutex_lock( &self->session->drm_connection_mutex );
+retry:
 #ifdef PBS_PROFESSIONAL
-		status = session->pbs_connection->statjob( session->pbs_connection, self->job_id, NULL);
+		status = pbs_statjob( session->pbs_conn, self->job_id, NULL, NULL );
 #else
-		status = session->pbs_connection->statjob( session->pbs_connection, self->job_id, session->status_attrl);
+		status = pbs_statjob( session->pbs_conn, self->job_id, session->status_attrl, NULL );
 #endif
+		fsd_log_debug(( "pbs_statjob(fd=%d, job_id=%s, attribs={...}) =%p",
+				 session->pbs_conn, self->job_id, (void*)status ));
+		if( status == NULL )
+		 {
+#ifndef PBS_PROFESSIONAL
+			fsd_log_error(("pbs_statjob error: %d, %s, %s", pbs_errno, pbse_to_txt(pbs_errno), pbs_strerror(pbs_errno)));
+#else
+			fsd_log_error(("pbs_statjob error: %d, %s", pbs_errno, pbse_to_txt(pbs_errno)));
+#endif
+			switch( pbs_errno )
+			 {
+				case PBSE_UNKJOBID:
+					break;
+				case PBSE_PROTOCOL:
+				case PBSE_EXPIRED:
+					pbs_disconnect( session->pbs_conn );
+					sleep(1);
+					session->pbs_conn = pbs_connect( session->super.contact );
+					if( session->pbs_conn < 0 )
+						pbsdrmaa_exc_raise_pbs( "pbs_connect" );
+					else 
+					 {
+						fsd_log_error(("retry:"));
+						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 )
@@ -139,13 +226,13 @@
 			((pbsdrmaa_job_t*)self)->update( self, status );
 		 }
-		else if( self->state < DRMAA_PS_DONE )
-		 {
+		else
 			self->on_missing( self );
-		 }
 	 }
 	FINALLY
 	 {
+		if( conn_lock )
+			conn_lock = fsd_mutex_unlock( &self->session->drm_connection_mutex );
 		if( status != NULL )
-			session->pbs_connection->statjob_free( session->pbs_connection, status );
+			pbs_statfree( status );
 	 }
 	END_TRY
@@ -161,5 +248,5 @@
 	struct attrl *i = NULL;
 	char pbs_state = 0;
-	int exit_status = -101;
+	int exit_status = -2;
 	const char *cpu_usage = NULL;
 	const char *mem_usage = NULL;
@@ -184,5 +271,5 @@
 				break;
 			case PBSDRMAA_ATTR_EXIT_STATUS:
-				exit_status = fsd_atoi( i->value );
+				exit_status = atoi( i->value );
 				break;
 			case PBSDRMAA_ATTR_RESOURCES_USED:
@@ -204,5 +291,4 @@
 					self->project = fsd_strdup(i->value);
 				break;
-#ifndef PBS_PROFESSIONAL
 			case PBSDRMAA_ATTR_EXECUTION_HOST:
 				if (!self->execution_hosts) {
@@ -211,12 +297,4 @@
 				}
 				break;
-#else
-			case PBSDRMAA_ATTR_EXECUTION_VNODE:
-				if (!self->execution_hosts) {
-					fsd_log_debug(("execution_hosts = %s", i->value));
-					self->execution_hosts = fsd_strdup(i->value);
-				}
-				break;
-#endif
 			case PBSDRMAA_ATTR_START_TIME:
 				{
@@ -236,15 +314,8 @@
 		fsd_log_debug(( "pbs_state: %c", pbs_state ));
 
-	if( exit_status != -101 )
+	if( exit_status != -2 )
 	 {
 		fsd_log_debug(( "exit_status: %d", exit_status ));
 		self->exit_status = exit_status;
-
-		if (self->exit_status < 0)
-		 {
-			self->exit_status = -1;
-			fsd_log_error(("ExitStatus = %d, probably system problem, report job %s to the local administrator", exit_status, self->job_id));
-		 }
-
 	 }
 	if(pbs_state){
@@ -252,10 +323,7 @@
 		 {
 			case 'C': /* Job is completed after having run. */
-#ifdef PBS_PROFESSIONAL
-			case 'F': /* Job is completed after having run. */
-#endif
 				self->flags &= FSD_JOB_TERMINATED_MASK;
 				self->flags |= FSD_JOB_TERMINATED;
-				if (exit_status != -101) { /* has exit code */
+				if (exit_status != -2) { /* has exit code */
 					if( self->exit_status == 0) 
 						self->state = DRMAA_PS_DONE;
@@ -266,12 +334,5 @@
 					self->exit_status = -1;
 				}
-				if (modify_time != 0)
-					self->end_time = modify_time; /* take last modify time as end time */
-				else
-					self->end_time = time(NULL);
-				
-				if (self->start_time == 0)
-					self->start_time = self->end_time;
-
+				self->end_time = modify_time; /* take last modify time as end time */
 				break;
 			case 'E': /* Job is exiting after having run. - MM: ignore exiting state (transient state) - outputs might have not been transfered yet, 
@@ -311,5 +372,6 @@
 		int hours, minutes, seconds;
 		long mem;
-		if( cpu_usage && sscanf( cpu_usage, "%d:%d:%d", &hours, &minutes, &seconds ) == 3 )
+		if( cpu_usage && sscanf( cpu_usage, "%d:%d:%d",
+				&hours, &minutes, &seconds ) == 3 )
 		 {
 			self->cpu_usage = 60*( 60*hours + minutes ) + seconds;
@@ -326,5 +388,6 @@
 			fsd_log_debug(( "vmem_usage: %s=%ldB", vmem_usage, self->vmem_usage ));
 		 }
-		if( walltime && sscanf( walltime, "%d:%d:%d", &hours, &minutes, &seconds ) == 3 )
+		if( walltime && sscanf( walltime, "%d:%d:%d",
+					&hours, &minutes, &seconds ) == 3 )
 		 {
 			self->walltime = 60*( 60*hours + minutes ) + seconds;
@@ -333,98 +396,54 @@
 	 }
 }
+
 
 void
 pbsdrmaa_job_on_missing( fsd_job_t *self )
 {
-	pbsdrmaa_session_t *pbssession = (pbsdrmaa_session_t*)self->session;
-	pbsdrmaa_job_t *pbsself = (pbsdrmaa_job_t *)self;
-	
-	if (!pbsself->missing_time)
-	 {
-		pbsself->missing_time = time(NULL);
-	 }
-
-	fsd_log_info(("pbsdrmaa_job_on_missing: pbs_home=%s, wait_thread_started=%d, submit_time=%d, missing_time=%d", 
-		pbssession->pbs_home, 
-		pbssession->super.wait_thread_started, 
-		(int)self->submit_time, 
-		(int)pbsself->missing_time));
-	
-	#define DRMAA_MAX_MISSING_TIME (30)
-	if( pbssession->pbs_home != NULL && pbssession->super.wait_thread_started && self->submit_time && (time(NULL) - pbsself->missing_time < DRMAA_MAX_MISSING_TIME))
-		fsd_log_info(("Job on missing but WT is running. Skipping...")); /* TODO: try to provide implementation that uses accounting/server log files */
+	fsd_drmaa_session_t *session = self->session;
+	pbsdrmaa_session_t *pbssession = (pbsdrmaa_session_t*)session;
+	if( !pbssession->wait_thread_log){	
+	unsigned missing_mask = 0;
+
+	fsd_log_enter(( "({job_id=%s})", self->job_id ));
+	fsd_log_warning(( "self %s missing from DRM queue", self->job_id ));
+
+	switch( session->missing_jobs )
+	 {
+		case FSD_REVEAL_MISSING_JOBS:         missing_mask = 0;     break;
+		case FSD_IGNORE_MISSING_JOBS:         missing_mask = 0x73;  break;
+		case FSD_IGNORE_QUEUED_MISSING_JOBS:  missing_mask = 0x13;  break;
+	 }
+	fsd_log_debug(( "last job_ps: %s (0x%02x); mask: 0x%02x",
+				drmaa_job_ps_to_str(self->state), self->state, missing_mask ));
+
+	if( self->state < DRMAA_PS_DONE
+			&&  (self->state & ~missing_mask) )
+		fsd_exc_raise_fmt(
+				FSD_ERRNO_INTERNAL_ERROR,
+				"self %s missing from queue", self->job_id
+				);
+
+	if( (self->flags & FSD_JOB_TERMINATED_MASK) == 0 )
+	 {
+		self->flags &= FSD_JOB_TERMINATED_MASK;
+		self->flags |= FSD_JOB_TERMINATED;
+	 }
+
+	if( (self->flags & FSD_JOB_ABORTED) == 0
+			&&  session->missing_jobs == FSD_IGNORE_MISSING_JOBS )
+	 { /* assume everthing was ok */
+		self->state = DRMAA_PS_DONE;
+		self->exit_status = 0;
+	 }
 	else
-		pbsdrmaa_job_on_missing_standard( self );	
-}
-
-void
-pbsdrmaa_job_on_missing_standard( fsd_job_t *self )
-{
-	fsd_drmaa_session_t *session = self->session;
-	pbsdrmaa_session_t *pbssession = (pbsdrmaa_session_t *)session;
-	int exit_status = -1;
-	
-	fsd_log_enter(( "({job_id=%s})", self->job_id ));
-	fsd_log_warning(( "Job %s missing from DRM queue", self->job_id ));
-
-	fsd_log_info(( "job_on_missing: last job_ps: %s (0x%02x)", drmaa_job_ps_to_str(self->state), self->state));
-
-	if( (exit_status = pbsdrmaa_job_read_exit_status(self->job_id, pbssession->job_exit_status_file_prefix)) == 0 )
-	{
-		self->state = DRMAA_PS_DONE;
-		self->exit_status = exit_status;
+	 { /* job aborted */
+		self->state = DRMAA_PS_FAILED;
+		self->exit_status = -1;
+	 }
+
+	fsd_log_return(( "; job_ps=%s, exit_status=%d",
+				drmaa_job_ps_to_str(self->state), self->exit_status ));
 	}
-	else
-	{
-		self->state = DRMAA_PS_FAILED;
-		self->exit_status = exit_status;
-	}
-        fsd_log_info(("job_on_missing evaluation result: state=%d exit_status=%d", self->state, self->exit_status));
-
-	fsd_cond_broadcast( &self->status_cond);
-	fsd_cond_broadcast( &self->session->wait_condition );
-
-	fsd_log_return(( "; job_ps=%s, exit_status=%d", drmaa_job_ps_to_str(self->state), self->exit_status ));
-}
-
-int
-pbsdrmaa_job_read_exit_status( const char *job_id, const char *job_state_dir_prefix)
-{
-	char *status_file = NULL, *start_file = NULL;
-	FILE *fhandle = NULL;
-	int exit_status = -1;
-
-	fsd_log_enter(("({job_id=%s, job_state_dir_prefix=%s})", job_id, job_state_dir_prefix));
-
-	status_file = fsd_asprintf("%s/%s.exitcode", job_state_dir_prefix, job_id);
-	start_file = fsd_asprintf("%s/%s.started", job_state_dir_prefix, job_id);
-
-	if ((fhandle = fopen(status_file, "r")) == NULL)
-	 {
-		struct stat tmpstat;
-
-		fsd_log_error(("Failed to open job status file: %s", status_file));
-		if (stat(start_file, &tmpstat) == 0 && (tmpstat.st_mode & S_IFREG))
-		 {
-			exit_status = 143; /* SIGTERM */
-			fsd_log_info(("But start file exist %s. Assuming that job was killed (exit_status=%d).", start_file, exit_status));
-		 }
-		else
-		 {
-			fsd_log_error(("Start file not found: %s", start_file));
-		 }
-	
-
-	 }
-	else
-	 {
-		(void)fscanf(fhandle, "%d", &exit_status); /*on error exit_status == -1 */
-		fclose(fhandle);
-	 }
-
-	fsd_free(status_file);
-	fsd_free(start_file);
-
-	return exit_status;
-}
-
+}
+
