<commit_before>/***************************Copyright-DO-NOT-REMOVE-THIS-LINE**
 * CONDOR Copyright Notice
 *
 * See LICENSE.TXT for additional notices and disclaimers.
 *
 * Copyright (c)1990-1998 CONDOR Team, Computer Sciences Department, 
 * University of Wisconsin-Madison, Madison, WI.  All Rights Reserved.  
 * No use of the CONDOR Software Program Source Code is authorized 
 * without the express consent of the CONDOR Team.  For more information 
 * contact: CONDOR Team, Attention: Prof<NAME>, 
 * 7367 Computer Sciences, 1210 W. Dayton St., Madison, WI 53706-1685, 
 * (608) 262-0856 or <EMAIL>.
 *
 * U.S. Government Rights Restrictions: Use, duplication, or disclosure 
 * by the U.S. Government is subject to restrictions as set forth in 
 * subparagraph (c)(1)(ii) of The Rights in Technical Data and Computer 
 * Software clause at DFARS 252.227-7013 or subparagraphs (c)(1) and 
 * (2) of Commercial Computer Software-Restricted Rights at 48 CFR 
 * 52.227-19, as applicable, CONDOR Team, Attention: Prof<NAME> 
 * Livny, 7367 Computer Sciences, 1210 W. Dayton St., Madison, 
 * WI 53706-1685, (608) 262-0856 or <EMAIL>.
****************************Copyright-DO-NOT-REMOVE-THIS-LINE**/


#include "condor_common.h"
#include "condor_attributes.h"
#include "condor_debug.h"
#include "environ.h"  // for Environ object

#include "globus_gram_client.h"
#include "globus_gram_error.h"

#include "gridmanager.h"
#include "globusjob.h"


#define JOB_PROBE_INTERVAL		300

// timer id values that indicates the timer is not registered
#define TIMER_UNSET		-1

// GridManager job states
#define GM_INIT					0
#define GM_REGISTER				1
#define GM_STDIO_UPDATE			2
#define GM_UNSUBMITTED			3
#define GM_SUBMIT_SAVE			4
#define GM_SUBMIT_COMMIT		5
#define GM_SUBMITTED			6
#define GM_CHECK_OUTPUT			7
#define GM_DONE_SAVE			8
#define GM_DONE_COMMIT			9
#define GM_STOP_AND_RESTART		10
#define GM_RESTART				11
#define GM_RESTART_SAVE			12
#define GM_RESTART_COMMIT		13
#define GM_CANCEL				14
#define GM_CANCEL_WAIT			15
#define GM_FAILED				16
#define GM_DELETE				17
#define GM_CLEAR_REQUEST		18

char *GMStateNames[] = {
	"GM_INIT",
	"GM_REGISTER",
	"GM_STDIO_UPDATE",
	"GM_UNSUBMITTED",
	"GM_SUBMIT_SAVE",
	"GM_SUBMIT_COMMIT",
	"GM_SUBMITTED",
	"GM_CHECK_OUTPUT",
	"GM_DONE_SAVE",
	"GM_DONE_COMMIT",
	"GM_STOP_AND_RESTART",
	"GM_RESTART",
	"GM_RESTART_SAVE",
	"GM_RESTART_COMMIT",
	"GM_CANCEL",
	"GM_CANCEL_WAIT",
	"GM_FAILED",
	"GM_DELETE",
	"GM_CLEAR_REQUEST"
};

#define LOG_GLOBUS_ERROR(func,error) \
    dprintf(D_ALWAYS, \
			"gmState %s, globusState %d: %s return Globus error %d\n", \
            GMStateNames[gmState],globusState,func,error)

GlobusJob::GlobusJob( GlobusJob& copy )
{
	dprintf(D_ALWAYS, "GlobusJob copy constructor called. This is a No-No!\n");
	ASSERT( 0 );
/*
	removedByUser = copy.removedByUser;
	callbackRegistered = copy.callbackRegistered;
	ignore_callbacks = copy.ignore_callbacks;
	procID = copy.procID;
	exitValue = copy.exitValue;
	jobContact = (copy.jobContact == NULL) ? NULL : strdup( copy.jobContact );
	RSL = (copy.RSL == NULL) ? NULL : strdup( copy.RSL );
	localOutput = (copy.localOutput == NULL) ? NULL : strdup( copy.localOutput );
	localError = (copy.localError == NULL) ? NULL : strdup( copy.localError );
	globusError = copy.globusError;
	jmFailureCode = copy.jmFailureCode;
	userLogFile = (copy.userLogFile == NULL) ? NULL : strdup( copy.userLogFile );
	executeLogged = copy.executeLogged;
	stateChanged = copy.stateChanged;
	newJM = copy.newJM;
	restartingJM = copy.restartingJM;
	restartWhen = copy.restartWhen;
	resourceDown = copy.resourceDown;
	condorState = copy.condorState;
	gmState = copy.gmState;
	globusState = copy.globusState;
	jmUnreachable = false;
	evaluateStateTid = TIMER_UNSET;

	myResource = copy.myResource;
	myResource->RegisterJob( this );
*/
}

GlobusJob::GlobusJob( ClassAd *classad, GlobusResource *resource )
{
	char buf[4096];
	bool full_rsl_given = false;

	removedByUser = false;
	callbackRegistered = false;
	ignore_callbacks = false;
	classad->LookupInteger( ATTR_CLUSTER_ID, procID.cluster );
	classad->LookupInteger( ATTR_PROC_ID, procID.proc );
	jobContact = NULL;
	RSL = NULL;
	localOutput = NULL;
	localError = NULL;
	userLogFile = NULL;
	globusError = 0;
	jmFailureCode = 0;
	exitValue = 0;
	submitLogged = false;
	executeLogged = false;
	stateChanged = false;
	newJM = false;
	restartingJM = false;
	restartWhen = 0;
	gmState = GM_INIT;
	globusState = GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED;
	jmUnreachable = false;
	evaluateStateTid = TIMER_UNSET;
	lastProbeTime = 0;
	enteredCurrentGmState = time(NULL);
	enteredCurrentGlobusState = time(NULL);
	numSubmitAttempts = 0;
	// ad = NULL;

	myResource = resource;
	myResource->RegisterJob( this );
	resourceDown = myResource->IsDown();

	buf[0] = '\0';
	classad->LookupString( ATTR_GLOBUS_RSL, buf );
	if ( buf[0] == '&' ) {
		full_rsl_given = true;
		RSL = strdup( buf );
	}

	buf[0] = '\0';
	classad->LookupString( ATTR_GLOBUS_CONTACT_STRING, buf );
	if ( strcmp( buf, NULL_JOB_CONTACT ) != 0 ) {
		jobContact = strdup( buf );
	}

	classad->LookupInteger( ATTR_GLOBUS_STATUS, globusState );
	classad->LookupInteger( ATTR_JOB_STATUS, condorState );

	globusError = GLOBUS_SUCCESS;

	buf[0] = '\0';
	classad->LookupString( ATTR_ULOG_FILE, buf );
	if ( buf[0] != '\0' ) {
		userLogFile = strdup( buf );
	}

	if ( !full_rsl_given ) {
		RSL = buildRSL( classad );
	}
}

GlobusJob::~GlobusJob()
{
	if ( evaluateStateTid != TIMER_UNSET ) {
		daemonCore->Cancel_Timer( evaluateStateTid );
	}
	if ( myResource != NULL ) {
		myResource->UnregisterJob( this );
	}
	if ( jobContact ) {
		free( jobContact );
	}
	if ( RSL ) {
		free( RSL );
	}
	if ( localOutput ) {
		free( localOutput );
	}
	if ( localError ) {
		free( localError );
	}
	if ( userLogFile ) {
		free( userLogFile );
	}
	if ( myResource ) {
		myResource->UnregisterJob( this );
	}
//	if ( ad ) {
//		delete ad;
//	}
}

void GlobusJob::SetEvaluateState()
{
	if ( evaluateStateTid == TIMER_UNSET ) {
		evaluateStateTid = daemonCore->Register_Timer( 0,
					(TimerHandlercpp)&GlobusJob::doEvaluateState,
					"doEvaluateState", (Service*) this );
	} else {
		daemonCore->Reset_Timer( evaluateStateTid, 0 );
	}
}

int GlobusJob::doEvaluateState()
{
	bool connect_failure = false;
	int old_gm_state;
	int old_globus_state;
	bool reevaluate_state = true;

	int rc;
	int status;
	int error;

	evaluateStateTid = TIMER_UNSET;

	if ( jmUnreachable || resourceDown ) {
		globus_gahp_set_new_call( false );
	} else {
		globus_gahp_set_new_call( true );
	}

	do {
		reevaluate_state = false;
		old_gm_state = gmState;
		old_globus_state = globusState;

		switch ( gmState ) {
		case GM_INIT:
			if ( jobContact == NULL ) {
				gmState = GM_UNSUBMITTED;
			} else {
				if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
					submitLogged = true;
				}
				if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
					executeLogged = true;
				}

				rc = globus_gram_client_job_signal( jobContact,
										(globus_gram_client_job_signal_t)0,
										NULL, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_UNKNOWN_SIGNAL_TYPE ) {
					newJM = true;
					gmState = GM_REGISTER;
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_INVALID_JOB_QUERY ||
							rc == GLOBUS_GRAM_PROTOCOL_ERROR_JOB_QUERY_DENIAL ) {
					newJM = false;
					if ( localOutput || localError ) {
						gmState = GM_CANCEL;
					} else {
						gmState = GM_REGISTER;
					}
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					gmState = GM_RESTART;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(0)", rc );
					gmState = GM_STOP_AND_RESTART
				}
			}
			break;
		case GM_REGISTER:
			rc = globus_gram_client_job_callback_register( jobContact,
										GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
										gramCallbackContact, &status,
										&error );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_callback_register()", rc );
				gmState = GM_STOP_AND_RESTART;
				break;
			}
			callbackRegistered = true;
			UpdateGlobusState( status, error );
/*
			globusState = status;
			globusError = error;
			addScheddUpdateAction( this, UA_UPDATE_GLOBUS_STATE );
*/
			if ( newJM ) {
				gmState = GM_STDIO_UPDATE;
			} else {
				gmState = GM_SUBMITTED;
			}
			break;
		case GM_STDIO_UPDATE:
			char rsl[4096];
			char buffer[1024];
			struct stat file_status;

			sprintf( rsl, "&(remote_io_url=%s)", gassServerUrl );
			if ( localOutput ) {
				rc = stat( localOutput, &file_status );
				if ( rc < 0 ) {
					file_status.st_size = 0;
				}
				sprintf( buffer, "(stdout=%s%s)(stdout_position=%d)",
						 gassServerUrl, localOutput, file_status.st_size );
				strcat( rsl, buffer );
			}
			if ( localError ) {
				rc = stat( localError, &file_status );
				if ( rc < 0 ) {
					file_status.st_size = 0;
				}
				sprintf( buffer, "(stderr=%s%s)(stderr_position=%d)",
						 gassServerUrl, localError, file_status.st_size );
				strcat( rsl, buffer );
			}
			rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_STDIO_UPDATE,
								rsl, &status, &error, new_cal );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(STDIO_UPDATE)", rc );
				gmState = GM_STOP_AND_RESTART;
				break;
			}
			if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED ) {
				gmState = GM_SUBMIT_COMMIT;
			} else {
				gmState = GM_SUBMITTED;
			}
			break;
		case GM_UNSUBMITTED:
			if ( condorState == REMOVED ) {
				gmState = GM_DELETE;
			} else if ( condorState == HELD ) {
				DeleteJob( this );
			} else {
				char *job_contact;
				rc = globus_gram_client_job_request( rmContact, rsl,
										GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
										gramCallbackContact, &job_contact );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONNECTION_FAILED ) {
					connect_failure = true;
					break;
				}
				numSubmitAttempts++;
				if ( rc == GLOBUS_SUCCESS ) {
					newJM = false;
					callbackRegistered = true;
					jobContact = strdup( job_contact );
					globus_gram_client_job_contact_free( job_contact );
					gmState = GM_SUBMIT_SAVE;
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_WAITING_FOR_COMMIT ) {
					newJM = true;
					callbackRegistered = true;
					jobContact = strdup( job_contact );
					globus_gram_client_job_contact_free( job_contact );
					gmState = GM_SUBMIT_SAVE;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_request()", rc );
					globusError = rc;
					WriteGlobusSubmitFailedToUserLog( this );
					gmState = GM_UNSUBMITTED;
					reevaluate_state = true;
				}
			}
			break;
		case GM_SUBMIT_SAVE:
			if ( condorState == REMOVED || condorState == HELD ) {
				gmState = GM_CANCEL;
			} else {
				rc = addScheddUpdateAction( this, UA_UPDATE_CONTACT_STRING,
											GM_SUBMIT_SAVE );
				if ( rc == PENDING ) {
					break;
				}
				if ( newJM ) {
					gmState = GM_SUBMIT_COMMIT;
				} else {
					gmState = GM_SUBMITTED;
				}
			}
			break;
		case GM_SUBMIT_COMMIT:
			if ( condorState == REMOVED || condorState == HELD ) {
				gmState = GM_CANCEL;
			} else {
				rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_REQUEST,
								NULL, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc != GLOBUS_SUCCESS ) {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_REQUEST)", rc );
					WriteGlobusSubmitFailedToUserLog( this );
					gmState = GM_CANCEL;
				} else {
					gmState = GM_SUBMITTED;
				}
			}
			break;
		case GM_SUBMITTED:
			if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
				if ( newJM ) {
					gmState = GM_CHECK_OUTPUT;
				} else {
					gmState = GM_DONE_SAVE;
				}
			} else if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
				gmState = GM_FAILED;
			} else {
				if ( condorState == REMOVED || condorState == HELD ) {
					gmState = GM_CANCEL;
				} else {
					time_t now = time(NULL);
					if ( lastProbeTime < enteredCurrentGmState ) {
						lastProbeTime = enteredCurrentGmState;
					}
					if ( now >= lastProbeTime + JOB_PROBE_INTERVAL ) {
						rc = globus_gram_client_job_status( jobContact,
															&status, &error );
						if ( rc == WOULD_CALL || rc == PENDING ) {
							break;
						}
						if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
							connect_failure = true;
							break;
						}
						if ( rc != GLOBUS_SUCCESS ) {
							// unhandled error
							LOG_GLOBUS_ERROR( "globus_gram_client_job_status()", rc );
							gmState = GM_STOP_AND_RESTART;
							break;
						}
						if ( globusState != state ) {
							globusState = state;
							enteredCurrentGlobusState = now;
						}
						globusError = error;
						lastProbeTime = now;
					}
					evaluateStateTid = daemonCore->Register_Timer(
								lastProbeTime + JOB_PROBE_INTERVAL,
								(TimerHandlercpp)&GlobusJob::doEvaluateState,
								"doEvaluateState", (Service*) this );
				}
			}
			break;
		case CHECK_OUTPUT:
			if ( condorState == REMOVED || condorState == HELD ) {
				gmState = GM_DONE_COMMIT;
			} else {
				if ( localOutput == NULL && localError == NULL ) {
					gmState = GM_DONE_SAVE;
					break;
				}
				int output_size = -1;
				int error_size = -1;
				struct stat file_status;
				char size_str[50];
				if ( localOutput ) {
					rc = stat( localOutput, &file_status );
					if ( rc < 0 ) {
						output_size = 0;
					} else {
						output_size = file_status.st_size;
					}
				}
				if ( localError ) {
					rc = stat( localError, &file_status );
					if ( rc < 0 ) {
						error_size = 0;
					} else {
						error_size = file_status.st_size;
					}
				}
				sprintf( size_str, "%d %d", output_size, error_size );
				rc = globus_gram_client_job_signal( jobContact,
									GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_STDIO_SIZE,
									size_str, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc == GLOBUS_SUCCESS ) {
					gmState = GM_DONE_SAVE;
				} else if ( rc ==  GLOBUS_GRAM_PROTOCOL_ERROR_STDIO_SIZE ) {
					gmState = GM_STOP_AND_RESTART;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(STDIO_SIZE)", rc );
					gmState = GM_STOP_AND_RESTART;
				}
			}
			break;
		case GM_DONE_SAVE:
			if ( condorState != HELD && condorState != REMOVED ) {
				condorState = COMPLETED;
				rc = addScheddUpdateAction( this, UA_UPDATE_CONDOR_STATE,
											GM_DONE_SAVE );
				if ( rc == PENDING ) {
					break;
				}
			}
			gmState = GM_DONE_COMMIT;
			break;
		case GM_DONE_COMMIT:
			if ( newJM ) {
				rc = globus_gram_client_job_signal( jobContact,
									GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_END,
									NULL, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc != GLOBUS_SUCCESS ) {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_END)", rc );
					gmState = GM_STOP_AND_RESTART;
					break;
				}
			}
			if ( condorState == COMPLETED || condorState == REMOVED ) {
				gmState = GM_DELETE;
			} else {
				gmState = GM_CLEAR_REQUEST;
			}
			break;
		case GM_STOP_AND_RESTART:
			rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_STOP_MANAGER,
								NULL, &status, &error );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(STOP_MANAGER)", rc );
				gmState = GM_CANCEL;
			} else {
				gmState = GM_RESTART;
			}
			break;
		case GM_RESTART:
			if ( jobContact == NULL ) {
				gmState = GM_UNSUBMITTED;
			} else {
				char *job_contact;
				char rsl[4096];
				char buffer[1024];
				struct stat file_status;

				sprintf( rsl, "&(restart=%s)(remote_io_url=%s)", jobContact,
						 gassServerUrl );
				if ( localOutput ) {
					rc = stat( localOutput, &file_status );
					if ( rc < 0 ) {
						file_status.st_size = 0;
					}
					sprintf( buffer, "(stdout=%s%s)(stdout_position=%d)",
							 gassServerUrl, localOutput, file_status.st_size );
					strcat( rsl, buffer );
				}
				if ( localError ) {
					rc = stat( localError, &file_status );
					if ( rc < 0 ) {
						file_status.st_size = 0;
					}
					sprintf( buffer, "(stderr=%s%s)(stderr_position=%d)",
							 gassServerUrl, localError, file_status.st_size );
					strcat( rsl, buffer );
				}
				rc = globus_gram_client_job_request( rmContact, rsl,
										GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
										gramCallbackContact, &job_contact );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONNECTION_FAILED ) {
					connect_failure = true;
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_UNDEFINED_EXE ) {
					newJM = false;
					gmState = GM_CLEAR_REQUEST;
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_WAITING_FOR_COMMIT ) {
					newJM = true;
					jobContact = strdup( job_contact );
					globus_gram_client_job_contact_free( job_contact );
					gmState = GM_SUBMIT_SAVE;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_request()", rc );
					gmState = GM_CLEAR_REQUEST;
				}
			}
			break;
		case GM_RESTART_SAVE:
			rc = addScheddUpdateAction( this, UA_UPDATE_CONTACT_STRING,
										GM_RESTART_SAVE );
			if ( rc == PENDING ) {
				break;
			}
			gmState = GM_RESTART_COMMIT;
			break;
		case GM_RESTART_COMMIT:
			rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_REQUEST,
								NULL, &status, &error );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_REQUEST)", rc );
				gmState = GM_STOP_AND_RESTART;
			}
			gmState = GM_SUBMITTED;
			break;
		case GM_CANCEL:
			if ( globusState != GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE &&
				 globusState != GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
				rc = globus_gram_client_job_cancel( jobContact );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc != GLOBUS_SUCCESS ) {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_cancel()", rc );
					gmState = GM_CLEAR_REQUEST;
					break;
				}
			}
			if ( callbackRegistered ) {
				gmState = GM_CANCEL_WAIT;
			} else {
				// This can happen if we're restarting and fail to
				// reattach to a running jobmanager
				if ( condorState == REMOVED ) {
					gmState = GM_DELETE;
				} else {
					gmState = GM_CLEAR_REQUEST;
				}
			}
			break;
		case GM_CANCEL_WAIT:
			if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
				gmState = GM_DONE_COMMIT;
			} else if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
				gmState = GM_FAILED;
			} else {
				time_t now = time(NULL);
				if ( lastProbeTime < enteredCurrentGmState ) {
					lastProbeTime = enteredCurrentGmState;
				}
				if ( now >= lastProbeTime + JOB_PROBE_INTERVAL ) {
					rc = globus_gram_client_job_status( jobContact, &status,
														&error );
					if ( rc == WOULD_CALL || rc == PENDING ) {
						break;
					}
					if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
						connect_failure = true;
						break;
					}
					if ( rc != GLOBUS_SUCCESS ) {
						// unhandled error
						LOG_GLOBUS_ERROR( "globus_gram_client_job_status()", rc );
						gmState = GM_CLEAR_REQUEST;
					}
					globusState = state;
					globusError = error;
					lastProbeTime = now;
				}
				evaluateStateTid = daemonCore->Register_Timer(
								lastProbeTime + JOB_PROBE_INTERVAL,
								(TimerHandlercpp)&GlobusJob::doEvaluateState,
								"doEvaluateState", (Service*) this );
			}
			break;
		case GM_FAILED:
			if ( globusError == GLOBUS_GRAM_PROTOCOL_ERROR_COMMIT_TIMED_OUT ||
				 globusError == GLOBUS_GRAM_PROTOCOL_ERROR_TTL_EXPIRED ||
				 globusError == GLOBUS_GRAM_PROTOCOL_ERROR_JM_STOPPED ||
				 globusError == GLOBUS_GRAM_PROTOCOL_ERROR_USER_PROXY_EXPIRED ) {
				gmState = GM_RESTART;
			} else {
				if ( newJM ) {
					rc = globus_gram_client_job_signal( jobContact,
									GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_END,
									NULL, &status, &error );
					if ( rc == WOULD_CALL || rc == PENDING ) {
						break;
					}
					if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
						connect_failure = true;
						break;
					}
					if ( rc != GLOBUS_SUCCESS ) {
						// unhandled error
						LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_END)", rc );
						gmState = GM_STOP_AND_RESTART;
						break;
					}
				}
				// TODO: evaluate if the failure is permanent or temporary
				//   if it's temporary, try a restart, but put a limit so
				//   that we don't go into an infinite restart loop
				// Note: Up through globus 2.0 beta, the FAILED state
				//   following a cancel request has an error code of 0
				//   instead of GLOBUS_GRAM_PROTOCOL_ERROR_USER_CANCELLED
				if ( condorState == REMOVED ) {
					gmState = GM_DELETE;
				} else {
					gmState = GM_CLEAR_REQUEST;
				}
			}
			break;
		case GM_DELETE:
			int schedd_actions = UA_DELETE_FROM_SCHEDD;
			if ( condorState == REMOVED ) {
				schedd_actions |= UA_LOG_ABORT_EVENT;
			} else if ( condorState == COMPLETED ) {
				schedd_actions |= UA_LOG_TERMINATE_EVENT;
			}
			rc = addScheddUpdateAction( this, update_actions,
										GM_DELETE );
			if ( rc == PENDING ) {
				break;
			}
			DeleteJob( this );
			break;
		case GM_CLEAR_REQUEST:
			globusState = GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED;
			free( jobContact );
			jobContact = NULL;
			rc = addScheddUpdateAction( this, UA_UPDATE_CONTACT_STRING,
										GM_CLEAR_REQUEST );
			if ( rc == PENDING ) {
				break;
			}
			gmState = GM_UNSUBMITTED;
			break;
		default:
			// Complain loudly!!
		}

		if ( gmState != old_gm_state || globusState != old_globus_state ) {
			reevaluate_state = true;
		}
		if ( globusState != old_globus_state ) {
			enteredCurrentGlobusState = time(NULL);
		}
		if ( gmState != old_gm_state ) {
			enteredCurrentGmState = time(NULL);
		}

	} while ( reevaluate_state );

	if ( connect_failure && !jmUnreachable && !resourceDown ) {
		jmUnreachable = true;
		myResource->RequestPing( this );
	}

	return TRUE;
}

void GlobusJob::NotifyResourceDown()
{
	if ( resourceDown == false ) {
		WriteGlobusResourceDownEventToUserLog( this );
	}
	resourceDown = true;
	jmUnreachable = false;
	// set downtime timestamp?
}

void GlobusJob::NotifyResourceUp()
{
	if ( resourceDown == true ) {
		WriteGlobusResourceUpEventToUserLog( this );
	}
	resourceDown = false;
	if ( jmUnreachable ) {
		gmState = GM_RESTART;
		enteredCurrentGmState = time(NULL);
	}
	jmUnreachable = false;
	SetEvaluateState();
}

void GlobusJob::UpdateCondorState( int new_state )
{
	condorState = new_state;
	SetEvaluateState();
}

void GlobusJob::UpdateGlobusState( int new_state, new_error_code )
{
	if ( globusState != new_state ) {
		// where to put logging of events: here or in EvaluateState?
		int update_actions = UA_UPDATE_GLOBUS_STATE;

		if ( !submitLogged && new_state != UNSUBMITTED ) {
			if ( new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
					// TODO: should SUBMIT_FAILED_EVENT be used only on
					//   certain errors (ones we know are submit-related)?
				update_actions |= UA_LOG_SUBMIT_FAILED_EVENT;
			} else {
				update_actions |= UA_LOG_SUBMIT_EVENT;
			}
		}
		if ( (new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE ||
			  new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ||
			  new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED)
			 && !executeLogged ) {
			update_actions |= UA_LOG_EXECUTE_EVENT;
		}

		globusState = new_state;
		globusStateErrorCode = new_error_code;
		enteredCurrentGlobusState = time(NULL);

		addScheddUpdateAction( this, update_actions, 0 );
	}
	SetEvaluateState(); // should this be inside the if statement?
}

GlobusResrouce *GlobusJob::GetResource()
{
	return myResource;
}

const char *GlobusJob::errorString()
{
	return globus_gram_client_error_string( globusError );
}


char *GlobusJob::buildRSL( ClassAd *classad )
{
	int transfer;
	char rsl[15000];
	char buff[11000];
	char buff2[1024]; // used to build localOutput and localError
	char *iwd;
<commit_msg>* Changed a couple places to call UpdateGlobusState() instead of modifying globusState directly.<commit_after>/***************************Copyright-DO-NOT-REMOVE-THIS-LINE**
 * CONDOR Copyright Notice
 *
 * See LICENSE.TXT for additional notices and disclaimers.
 *
 * Copyright (c)1990-1998 CONDOR Team, Computer Sciences Department, 
 * University of Wisconsin-Madison, Madison, WI.  All Rights Reserved.  
 * No use of the CONDOR Software Program Source Code is authorized 
 * without the express consent of the CONDOR Team.  For more information 
 * contact: CONDOR Team, Attention: Prof<NAME>, 
 * 7367 Computer Sciences, 1210 W. Dayton St., Madison, WI 53706-1685, 
 * (608) 262-0856 or <EMAIL>.
 *
 * U.S. Government Rights Restrictions: Use, duplication, or disclosure 
 * by the U.S. Government is subject to restrictions as set forth in 
 * subparagraph (c)(1)(ii) of The Rights in Technical Data and Computer 
 * Software clause at DFARS 252.227-7013 or subparagraphs (c)(1) and 
 * (2) of Commercial Computer Software-Restricted Rights at 48 CFR 
 * 52.227-19, as applicable, CONDOR Team, Attention: <NAME> 
 * Livny, 7367 Computer Sciences, 1210 W. Dayton St., Madison, 
 * WI 53706-1685, (608) 262-0856 or <EMAIL>.
****************************Copyright-DO-NOT-REMOVE-THIS-LINE**/


#include "condor_common.h"
#include "condor_attributes.h"
#include "condor_debug.h"
#include "environ.h"  // for Environ object

#include "globus_gram_client.h"
#include "globus_gram_error.h"

#include "gridmanager.h"
#include "globusjob.h"


#define JOB_PROBE_INTERVAL		300

// timer id values that indicates the timer is not registered
#define TIMER_UNSET		-1

// GridManager job states
#define GM_INIT					0
#define GM_REGISTER				1
#define GM_STDIO_UPDATE			2
#define GM_UNSUBMITTED			3
#define GM_SUBMIT_SAVE			4
#define GM_SUBMIT_COMMIT		5
#define GM_SUBMITTED			6
#define GM_CHECK_OUTPUT			7
#define GM_DONE_SAVE			8
#define GM_DONE_COMMIT			9
#define GM_STOP_AND_RESTART		10
#define GM_RESTART				11
#define GM_RESTART_SAVE			12
#define GM_RESTART_COMMIT		13
#define GM_CANCEL				14
#define GM_CANCEL_WAIT			15
#define GM_FAILED				16
#define GM_DELETE				17
#define GM_CLEAR_REQUEST		18

char *GMStateNames[] = {
	"GM_INIT",
	"GM_REGISTER",
	"GM_STDIO_UPDATE",
	"GM_UNSUBMITTED",
	"GM_SUBMIT_SAVE",
	"GM_SUBMIT_COMMIT",
	"GM_SUBMITTED",
	"GM_CHECK_OUTPUT",
	"GM_DONE_SAVE",
	"GM_DONE_COMMIT",
	"GM_STOP_AND_RESTART",
	"GM_RESTART",
	"GM_RESTART_SAVE",
	"GM_RESTART_COMMIT",
	"GM_CANCEL",
	"GM_CANCEL_WAIT",
	"GM_FAILED",
	"GM_DELETE",
	"GM_CLEAR_REQUEST"
};

#define LOG_GLOBUS_ERROR(func,error) \
    dprintf(D_ALWAYS, \
			"gmState %s, globusState %d: %s return Globus error %d\n", \
            GMStateNames[gmState],globusState,func,error)

GlobusJob::GlobusJob( GlobusJob& copy )
{
	dprintf(D_ALWAYS, "GlobusJob copy constructor called. This is a No-No!\n");
	ASSERT( 0 );
/*
	removedByUser = copy.removedByUser;
	callbackRegistered = copy.callbackRegistered;
	ignore_callbacks = copy.ignore_callbacks;
	procID = copy.procID;
	exitValue = copy.exitValue;
	jobContact = (copy.jobContact == NULL) ? NULL : strdup( copy.jobContact );
	RSL = (copy.RSL == NULL) ? NULL : strdup( copy.RSL );
	localOutput = (copy.localOutput == NULL) ? NULL : strdup( copy.localOutput );
	localError = (copy.localError == NULL) ? NULL : strdup( copy.localError );
	globusError = copy.globusError;
	jmFailureCode = copy.jmFailureCode;
	userLogFile = (copy.userLogFile == NULL) ? NULL : strdup( copy.userLogFile );
	executeLogged = copy.executeLogged;
	stateChanged = copy.stateChanged;
	newJM = copy.newJM;
	restartingJM = copy.restartingJM;
	restartWhen = copy.restartWhen;
	resourceDown = copy.resourceDown;
	condorState = copy.condorState;
	gmState = copy.gmState;
	globusState = copy.globusState;
	jmUnreachable = false;
	evaluateStateTid = TIMER_UNSET;

	myResource = copy.myResource;
	myResource->RegisterJob( this );
*/
}

GlobusJob::GlobusJob( ClassAd *classad, GlobusResource *resource )
{
	char buf[4096];
	bool full_rsl_given = false;

	removedByUser = false;
	callbackRegistered = false;
	ignore_callbacks = false;
	classad->LookupInteger( ATTR_CLUSTER_ID, procID.cluster );
	classad->LookupInteger( ATTR_PROC_ID, procID.proc );
	jobContact = NULL;
	RSL = NULL;
	localOutput = NULL;
	localError = NULL;
	userLogFile = NULL;
	globusError = 0;
	jmFailureCode = 0;
	exitValue = 0;
	submitLogged = false;
	executeLogged = false;
	stateChanged = false;
	newJM = false;
	restartingJM = false;
	restartWhen = 0;
	gmState = GM_INIT;
	globusState = GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED;
	jmUnreachable = false;
	evaluateStateTid = TIMER_UNSET;
	lastProbeTime = 0;
	enteredCurrentGmState = time(NULL);
	enteredCurrentGlobusState = time(NULL);
	numSubmitAttempts = 0;
	// ad = NULL;

	myResource = resource;
	myResource->RegisterJob( this );
	resourceDown = myResource->IsDown();

	buf[0] = '\0';
	classad->LookupString( ATTR_GLOBUS_RSL, buf );
	if ( buf[0] == '&' ) {
		full_rsl_given = true;
		RSL = strdup( buf );
	}

	buf[0] = '\0';
	classad->LookupString( ATTR_GLOBUS_CONTACT_STRING, buf );
	if ( strcmp( buf, NULL_JOB_CONTACT ) != 0 ) {
		jobContact = strdup( buf );
	}

	classad->LookupInteger( ATTR_GLOBUS_STATUS, globusState );
	classad->LookupInteger( ATTR_JOB_STATUS, condorState );

	globusError = GLOBUS_SUCCESS;

	buf[0] = '\0';
	classad->LookupString( ATTR_ULOG_FILE, buf );
	if ( buf[0] != '\0' ) {
		userLogFile = strdup( buf );
	}

	if ( !full_rsl_given ) {
		RSL = buildRSL( classad );
	}
}

GlobusJob::~GlobusJob()
{
	if ( evaluateStateTid != TIMER_UNSET ) {
		daemonCore->Cancel_Timer( evaluateStateTid );
	}
	if ( myResource != NULL ) {
		myResource->UnregisterJob( this );
	}
	if ( jobContact ) {
		free( jobContact );
	}
	if ( RSL ) {
		free( RSL );
	}
	if ( localOutput ) {
		free( localOutput );
	}
	if ( localError ) {
		free( localError );
	}
	if ( userLogFile ) {
		free( userLogFile );
	}
	if ( myResource ) {
		myResource->UnregisterJob( this );
	}
//	if ( ad ) {
//		delete ad;
//	}
}

void GlobusJob::SetEvaluateState()
{
	if ( evaluateStateTid == TIMER_UNSET ) {
		evaluateStateTid = daemonCore->Register_Timer( 0,
					(TimerHandlercpp)&GlobusJob::doEvaluateState,
					"doEvaluateState", (Service*) this );
	} else {
		daemonCore->Reset_Timer( evaluateStateTid, 0 );
	}
}

int GlobusJob::doEvaluateState()
{
	bool connect_failure = false;
	int old_gm_state;
	int old_globus_state;
	bool reevaluate_state = true;

	int rc;
	int status;
	int error;

	evaluateStateTid = TIMER_UNSET;

	if ( jmUnreachable || resourceDown ) {
		globus_gahp_set_new_call( false );
	} else {
		globus_gahp_set_new_call( true );
	}

	do {
		reevaluate_state = false;
		old_gm_state = gmState;
		old_globus_state = globusState;

		switch ( gmState ) {
		case GM_INIT:
			if ( jobContact == NULL ) {
				gmState = GM_UNSUBMITTED;
			} else {
				if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_PENDING ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
					submitLogged = true;
				}
				if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED ||
					 globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
					executeLogged = true;
				}

				rc = globus_gram_client_job_signal( jobContact,
										(globus_gram_client_job_signal_t)0,
										NULL, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_UNKNOWN_SIGNAL_TYPE ) {
					newJM = true;
					gmState = GM_REGISTER;
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_INVALID_JOB_QUERY ||
							rc == GLOBUS_GRAM_PROTOCOL_ERROR_JOB_QUERY_DENIAL ) {
					newJM = false;
					if ( localOutput || localError ) {
						gmState = GM_CANCEL;
					} else {
						gmState = GM_REGISTER;
					}
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					gmState = GM_RESTART;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(0)", rc );
					gmState = GM_STOP_AND_RESTART
				}
			}
			break;
		case GM_REGISTER:
			rc = globus_gram_client_job_callback_register( jobContact,
										GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
										gramCallbackContact, &status,
										&error );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_callback_register()", rc );
				gmState = GM_STOP_AND_RESTART;
				break;
			}
			callbackRegistered = true;
			UpdateGlobusState( status, error );
			if ( newJM ) {
				gmState = GM_STDIO_UPDATE;
			} else {
				gmState = GM_SUBMITTED;
			}
			break;
		case GM_STDIO_UPDATE:
			char rsl[4096];
			char buffer[1024];
			struct stat file_status;

			sprintf( rsl, "&(remote_io_url=%s)", gassServerUrl );
			if ( localOutput ) {
				rc = stat( localOutput, &file_status );
				if ( rc < 0 ) {
					file_status.st_size = 0;
				}
				sprintf( buffer, "(stdout=%s%s)(stdout_position=%d)",
						 gassServerUrl, localOutput, file_status.st_size );
				strcat( rsl, buffer );
			}
			if ( localError ) {
				rc = stat( localError, &file_status );
				if ( rc < 0 ) {
					file_status.st_size = 0;
				}
				sprintf( buffer, "(stderr=%s%s)(stderr_position=%d)",
						 gassServerUrl, localError, file_status.st_size );
				strcat( rsl, buffer );
			}
			rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_STDIO_UPDATE,
								rsl, &status, &error, new_cal );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(STDIO_UPDATE)", rc );
				gmState = GM_STOP_AND_RESTART;
				break;
			}
			if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED ) {
				gmState = GM_SUBMIT_COMMIT;
			} else {
				gmState = GM_SUBMITTED;
			}
			break;
		case GM_UNSUBMITTED:
			if ( condorState == REMOVED ) {
				gmState = GM_DELETE;
			} else if ( condorState == HELD ) {
				DeleteJob( this );
			} else {
				char *job_contact;
				rc = globus_gram_client_job_request( rmContact, rsl,
										GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
										gramCallbackContact, &job_contact );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONNECTION_FAILED ) {
					connect_failure = true;
					break;
				}
				numSubmitAttempts++;
				if ( rc == GLOBUS_SUCCESS ) {
					newJM = false;
					callbackRegistered = true;
					jobContact = strdup( job_contact );
					globus_gram_client_job_contact_free( job_contact );
					gmState = GM_SUBMIT_SAVE;
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_WAITING_FOR_COMMIT ) {
					newJM = true;
					callbackRegistered = true;
					jobContact = strdup( job_contact );
					globus_gram_client_job_contact_free( job_contact );
					gmState = GM_SUBMIT_SAVE;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_request()", rc );
					globusError = rc;
					WriteGlobusSubmitFailedToUserLog( this );
					gmState = GM_UNSUBMITTED;
					reevaluate_state = true;
				}
			}
			break;
		case GM_SUBMIT_SAVE:
			if ( condorState == REMOVED || condorState == HELD ) {
				gmState = GM_CANCEL;
			} else {
				rc = addScheddUpdateAction( this, UA_UPDATE_CONTACT_STRING,
											GM_SUBMIT_SAVE );
				if ( rc == PENDING ) {
					break;
				}
				if ( newJM ) {
					gmState = GM_SUBMIT_COMMIT;
				} else {
					gmState = GM_SUBMITTED;
				}
			}
			break;
		case GM_SUBMIT_COMMIT:
			if ( condorState == REMOVED || condorState == HELD ) {
				gmState = GM_CANCEL;
			} else {
				rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_REQUEST,
								NULL, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc != GLOBUS_SUCCESS ) {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_REQUEST)", rc );
					WriteGlobusSubmitFailedToUserLog( this );
					gmState = GM_CANCEL;
				} else {
					gmState = GM_SUBMITTED;
				}
			}
			break;
		case GM_SUBMITTED:
			if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
				if ( newJM ) {
					gmState = GM_CHECK_OUTPUT;
				} else {
					gmState = GM_DONE_SAVE;
				}
			} else if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
				gmState = GM_FAILED;
			} else {
				if ( condorState == REMOVED || condorState == HELD ) {
					gmState = GM_CANCEL;
				} else {
					time_t now = time(NULL);
					if ( lastProbeTime < enteredCurrentGmState ) {
						lastProbeTime = enteredCurrentGmState;
					}
					if ( now >= lastProbeTime + JOB_PROBE_INTERVAL ) {
						rc = globus_gram_client_job_status( jobContact,
															&status, &error );
						if ( rc == WOULD_CALL || rc == PENDING ) {
							break;
						}
						if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
							connect_failure = true;
							break;
						}
						if ( rc != GLOBUS_SUCCESS ) {
							// unhandled error
							LOG_GLOBUS_ERROR( "globus_gram_client_job_status()", rc );
							gmState = GM_STOP_AND_RESTART;
							break;
						}
						UpdateGlobusState( state, error );
						lastProbeTime = now;
					}
					evaluateStateTid = daemonCore->Register_Timer(
								(lastProbeTime + JOB_PROBE_INTERVAL) - now,
								(TimerHandlercpp)&GlobusJob::doEvaluateState,
								"doEvaluateState", (Service*) this );
				}
			}
			break;
		case CHECK_OUTPUT:
			if ( condorState == REMOVED || condorState == HELD ) {
				gmState = GM_DONE_COMMIT;
			} else {
				if ( localOutput == NULL && localError == NULL ) {
					gmState = GM_DONE_SAVE;
					break;
				}
				int output_size = -1;
				int error_size = -1;
				struct stat file_status;
				char size_str[50];
				if ( localOutput ) {
					rc = stat( localOutput, &file_status );
					if ( rc < 0 ) {
						output_size = 0;
					} else {
						output_size = file_status.st_size;
					}
				}
				if ( localError ) {
					rc = stat( localError, &file_status );
					if ( rc < 0 ) {
						error_size = 0;
					} else {
						error_size = file_status.st_size;
					}
				}
				sprintf( size_str, "%d %d", output_size, error_size );
				rc = globus_gram_client_job_signal( jobContact,
									GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_STDIO_SIZE,
									size_str, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc == GLOBUS_SUCCESS ) {
					gmState = GM_DONE_SAVE;
				} else if ( rc ==  GLOBUS_GRAM_PROTOCOL_ERROR_STDIO_SIZE ) {
					gmState = GM_STOP_AND_RESTART;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(STDIO_SIZE)", rc );
					gmState = GM_STOP_AND_RESTART;
				}
			}
			break;
		case GM_DONE_SAVE:
			if ( condorState != HELD && condorState != REMOVED ) {
				condorState = COMPLETED;
				rc = addScheddUpdateAction( this, UA_UPDATE_CONDOR_STATE,
											GM_DONE_SAVE );
				if ( rc == PENDING ) {
					break;
				}
			}
			gmState = GM_DONE_COMMIT;
			break;
		case GM_DONE_COMMIT:
			if ( newJM ) {
				rc = globus_gram_client_job_signal( jobContact,
									GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_END,
									NULL, &status, &error );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc != GLOBUS_SUCCESS ) {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_END)", rc );
					gmState = GM_STOP_AND_RESTART;
					break;
				}
			}
			if ( condorState == COMPLETED || condorState == REMOVED ) {
				gmState = GM_DELETE;
			} else {
				gmState = GM_CLEAR_REQUEST;
			}
			break;
		case GM_STOP_AND_RESTART:
			rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_STOP_MANAGER,
								NULL, &status, &error );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(STOP_MANAGER)", rc );
				gmState = GM_CANCEL;
			} else {
				gmState = GM_RESTART;
			}
			break;
		case GM_RESTART:
			if ( jobContact == NULL ) {
				gmState = GM_UNSUBMITTED;
			} else {
				char *job_contact;
				char rsl[4096];
				char buffer[1024];
				struct stat file_status;

				sprintf( rsl, "&(restart=%s)(remote_io_url=%s)", jobContact,
						 gassServerUrl );
				if ( localOutput ) {
					rc = stat( localOutput, &file_status );
					if ( rc < 0 ) {
						file_status.st_size = 0;
					}
					sprintf( buffer, "(stdout=%s%s)(stdout_position=%d)",
							 gassServerUrl, localOutput, file_status.st_size );
					strcat( rsl, buffer );
				}
				if ( localError ) {
					rc = stat( localError, &file_status );
					if ( rc < 0 ) {
						file_status.st_size = 0;
					}
					sprintf( buffer, "(stderr=%s%s)(stderr_position=%d)",
							 gassServerUrl, localError, file_status.st_size );
					strcat( rsl, buffer );
				}
				rc = globus_gram_client_job_request( rmContact, rsl,
										GLOBUS_GRAM_PROTOCOL_JOB_STATE_ALL,
										gramCallbackContact, &job_contact );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONNECTION_FAILED ) {
					connect_failure = true;
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_UNDEFINED_EXE ) {
					newJM = false;
					gmState = GM_CLEAR_REQUEST;
				} else if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_WAITING_FOR_COMMIT ) {
					newJM = true;
					jobContact = strdup( job_contact );
					globus_gram_client_job_contact_free( job_contact );
					gmState = GM_SUBMIT_SAVE;
				} else {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_request()", rc );
					gmState = GM_CLEAR_REQUEST;
				}
			}
			break;
		case GM_RESTART_SAVE:
			rc = addScheddUpdateAction( this, UA_UPDATE_CONTACT_STRING,
										GM_RESTART_SAVE );
			if ( rc == PENDING ) {
				break;
			}
			gmState = GM_RESTART_COMMIT;
			break;
		case GM_RESTART_COMMIT:
			rc = globus_gram_client_job_signal( jobContact,
								GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_REQUEST,
								NULL, &status, &error );
			if ( rc == WOULD_CALL || rc == PENDING ) {
				break;
			}
			if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
				connect_failure = true;
				break;
			}
			if ( rc != GLOBUS_SUCCESS ) {
				// unhandled error
				LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_REQUEST)", rc );
				gmState = GM_STOP_AND_RESTART;
			}
			gmState = GM_SUBMITTED;
			break;
		case GM_CANCEL:
			if ( globusState != GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE &&
				 globusState != GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
				rc = globus_gram_client_job_cancel( jobContact );
				if ( rc == WOULD_CALL || rc == PENDING ) {
					break;
				}
				if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
					connect_failure = true;
					break;
				}
				if ( rc != GLOBUS_SUCCESS ) {
					// unhandled error
					LOG_GLOBUS_ERROR( "globus_gram_client_job_cancel()", rc );
					gmState = GM_CLEAR_REQUEST;
					break;
				}
			}
			if ( callbackRegistered ) {
				gmState = GM_CANCEL_WAIT;
			} else {
				// This can happen if we're restarting and fail to
				// reattach to a running jobmanager
				if ( condorState == REMOVED ) {
					gmState = GM_DELETE;
				} else {
					gmState = GM_CLEAR_REQUEST;
				}
			}
			break;
		case GM_CANCEL_WAIT:
			if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ) {
				gmState = GM_DONE_COMMIT;
			} else if ( globusState == GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
				gmState = GM_FAILED;
			} else {
				time_t now = time(NULL);
				if ( lastProbeTime < enteredCurrentGmState ) {
					lastProbeTime = enteredCurrentGmState;
				}
				if ( now >= lastProbeTime + JOB_PROBE_INTERVAL ) {
					rc = globus_gram_client_job_status( jobContact, &status,
														&error );
					if ( rc == WOULD_CALL || rc == PENDING ) {
						break;
					}
					if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
						connect_failure = true;
						break;
					}
					if ( rc != GLOBUS_SUCCESS ) {
						// unhandled error
						LOG_GLOBUS_ERROR( "globus_gram_client_job_status()", rc );
						gmState = GM_CLEAR_REQUEST;
					}
					UpdateGlobusState( state, error );
					lastProbeTime = now;
				}
				evaluateStateTid = daemonCore->Register_Timer(
								(lastProbeTime + JOB_PROBE_INTERVAL) - now,
								(TimerHandlercpp)&GlobusJob::doEvaluateState,
								"doEvaluateState", (Service*) this );
			}
			break;
		case GM_FAILED:
			if ( globusError == GLOBUS_GRAM_PROTOCOL_ERROR_COMMIT_TIMED_OUT ||
				 globusError == GLOBUS_GRAM_PROTOCOL_ERROR_TTL_EXPIRED ||
				 globusError == GLOBUS_GRAM_PROTOCOL_ERROR_JM_STOPPED ||
				 globusError == GLOBUS_GRAM_PROTOCOL_ERROR_USER_PROXY_EXPIRED ) {
				gmState = GM_RESTART;
			} else {
				if ( newJM ) {
					rc = globus_gram_client_job_signal( jobContact,
									GLOBUS_GRAM_PROTOCOL_JOB_SIGNAL_COMMIT_END,
									NULL, &status, &error );
					if ( rc == WOULD_CALL || rc == PENDING ) {
						break;
					}
					if ( rc == GLOBUS_GRAM_PROTOCOL_ERROR_CONTACTING_JOB_MANAGER ) {
						connect_failure = true;
						break;
					}
					if ( rc != GLOBUS_SUCCESS ) {
						// unhandled error
						LOG_GLOBUS_ERROR( "globus_gram_client_job_signal(COMMIT_END)", rc );
						gmState = GM_STOP_AND_RESTART;
						break;
					}
				}
				// TODO: evaluate if the failure is permanent or temporary
				//   if it's temporary, try a restart, but put a limit so
				//   that we don't go into an infinite restart loop
				// Note: Up through globus 2.0 beta, the FAILED state
				//   following a cancel request has an error code of 0
				//   instead of GLOBUS_GRAM_PROTOCOL_ERROR_USER_CANCELLED
				if ( condorState == REMOVED ) {
					gmState = GM_DELETE;
				} else {
					gmState = GM_CLEAR_REQUEST;
				}
			}
			break;
		case GM_DELETE:
			int schedd_actions = UA_DELETE_FROM_SCHEDD;
			if ( condorState == REMOVED ) {
				schedd_actions |= UA_LOG_ABORT_EVENT;
			} else if ( condorState == COMPLETED ) {
				schedd_actions |= UA_LOG_TERMINATE_EVENT;
			}
			rc = addScheddUpdateAction( this, update_actions,
										GM_DELETE );
			if ( rc == PENDING ) {
				break;
			}
			DeleteJob( this );
			break;
		case GM_CLEAR_REQUEST:
			globusState = GLOBUS_GRAM_PROTOCOL_JOB_STATE_UNSUBMITTED;
			free( jobContact );
			jobContact = NULL;
			rc = addScheddUpdateAction( this, UA_UPDATE_CONTACT_STRING,
										GM_CLEAR_REQUEST );
			if ( rc == PENDING ) {
				break;
			}
			gmState = GM_UNSUBMITTED;
			break;
		default:
			// Complain loudly!!
		}

		if ( gmState != old_gm_state || globusState != old_globus_state ) {
			reevaluate_state = true;
		}
		if ( globusState != old_globus_state ) {
			enteredCurrentGlobusState = time(NULL);
		}
		if ( gmState != old_gm_state ) {
			enteredCurrentGmState = time(NULL);
		}

	} while ( reevaluate_state );

	if ( connect_failure && !jmUnreachable && !resourceDown ) {
		jmUnreachable = true;
		myResource->RequestPing( this );
	}

	return TRUE;
}

void GlobusJob::NotifyResourceDown()
{
	if ( resourceDown == false ) {
		WriteGlobusResourceDownEventToUserLog( this );
	}
	resourceDown = true;
	jmUnreachable = false;
	// set downtime timestamp?
}

void GlobusJob::NotifyResourceUp()
{
	if ( resourceDown == true ) {
		WriteGlobusResourceUpEventToUserLog( this );
	}
	resourceDown = false;
	if ( jmUnreachable ) {
		gmState = GM_RESTART;
		enteredCurrentGmState = time(NULL);
	}
	jmUnreachable = false;
	SetEvaluateState();
}

void GlobusJob::UpdateCondorState( int new_state )
{
	condorState = new_state;
	SetEvaluateState();
}

void GlobusJob::UpdateGlobusState( int new_state, new_error_code )
{
	if ( globusState != new_state ) {
		// where to put logging of events: here or in EvaluateState?
		int update_actions = UA_UPDATE_GLOBUS_STATE;

		if ( !submitLogged && new_state != UNSUBMITTED ) {
			if ( new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_FAILED ) {
					// TODO: should SUBMIT_FAILED_EVENT be used only on
					//   certain errors (ones we know are submit-related)?
				update_actions |= UA_LOG_SUBMIT_FAILED_EVENT;
			} else {
				update_actions |= UA_LOG_SUBMIT_EVENT;
			}
		}
		if ( (new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_ACTIVE ||
			  new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_DONE ||
			  new_state == GLOBUS_GRAM_PROTOCOL_JOB_STATE_SUSPENDED)
			 && !executeLogged ) {
			update_actions |= UA_LOG_EXECUTE_EVENT;
		}

		globusState = new_state;
		globusStateErrorCode = new_error_code;
		enteredCurrentGlobusState = time(NULL);

		addScheddUpdateAction( this, update_actions, 0 );
	}
	SetEvaluateState(); // should this be inside the if statement?
}

GlobusResrouce *GlobusJob::GetResource()
{
	return myResource;
}

const char *GlobusJob::errorString()
{
	return globus_gram_client_error_string( globusError );
}

char *GlobusJob::buildRSL( ClassAd *classad )
{
	int transfer;
	char rsl[15000];
	char buff[11000];
	char buff2[1024]; // used to build localOutput and localError
	char *iwd;

{'fraction_non_alphanumeric': 0.10622859464793638, 'fraction_numerical': 0.007430196784117955, 'mean_word_length': 3.1847773279352225, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3292072', 'n_tokens_mistral': 23953, 'n_tokens_neox': 19166, 'n_words': 3698}