Context Navigation

source: trunk/ll_drmaa/job.c @ 26

Revision 26, 22.5 KB checked in by mmamonski, 13 years ago (diff)
SupMUC on site fixes: 1. Polling mode 2. Handling missing jobs 3. monitor -> drmaa_monitor 4. force stderr file creation
Property svn:keywords set to `Id Revision`

Rev	Line
[20]	1	/* $Id$ */
[1]	2	/*
	3	* PSNC DRMAA for LL
	4	* Copyright (C) 2010 Poznan Supercomputing and Networking Center
	5	*
	6	* Licensed under the Apache License, Version 2.0 (the "License");
	7	* you may not use this file except in compliance with the License.
	8	* You may obtain a copy of the License at
	9	*
	10	* http://www.apache.org/licenses/LICENSE-2.0
	11	*
	12	* Unless required by applicable law or agreed to in writing, software
	13	* distributed under the License is distributed on an "AS IS" BASIS,
	14	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	15	* See the License for the specific language governing permissions and
	16	* limitations under the License.
	17	*/
	18
	19	#include <string.h>
	20	#include <stdlib.h>
	21	#include <unistd.h>
	22	#include <time.h>
	23
	24	#include <drmaa_utils/common.h>
	25	#include <drmaa_utils/conf.h>
	26	#include <drmaa_utils/datetime.h>
	27	#include <drmaa_utils/drmaa.h>
	28	#include <drmaa_utils/drmaa_util.h>
	29	#include <drmaa_utils/environ.h>
	30	#include <drmaa_utils/template.h>
	31
	32	#include <ll_drmaa/job.h>
	33	#include <ll_drmaa/session.h>
	34	#include <ll_drmaa/util.h>
	35
	36	#include <llapi.h>
	37
[26]	38	#define LL_DRMAA_MAX_MISSING_TIME (60)
	39
[1]	40	static void
	41	lldrmaa_job_control( fsd_job_t *self, int action )
	42	{
	43	int rc = 0;
	44	LL_element *errObj = NULL;
	45	LL_terminate_job_info cancel_info;
	46	char *rest = NULL;
	47	char *token = NULL;
	48	char *ptr = NULL;
	49	char *volatile job_list[2];
	50
	51	job_list[0] = self->job_id;
	52	job_list[1] = NULL;
	53
	54	cancel_info.StepId.from_host = NULL;
	55	ptr = fsd_strdup(self->job_id);
	56	fsd_log_enter(( "({job_id=%s}, action=%s)", self->job_id, drmaa_control_to_str(action) ));
	57
	58	fsd_mutex_lock( &self->session->drm_connection_mutex );
	59	TRY
	60	{
	61	switch( action )
	62	{
	63	case DRMAA_CONTROL_SUSPEND:
	64	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"job::control: could not send %s to job %s. Function not implemented.",drmaa_control_to_str(action), self->job_id);
	65	break;
	66	case DRMAA_CONTROL_HOLD:
	67	rc = ll_control(LL_CONTROL_VERSION, LL_CONTROL_HOLD_USER, NULL, NULL, (char **)job_list, NULL,0);
	68	fsd_log_warning(("This command does not affect a job step that is running unless the job step attempts to enter the Idle state."));
	69	if(rc)
	70	fsd_exc_raise_fmt(lldrmaa_map_control(rc),
	71	"job::control: could not send %s to job %s. Error code: %d,%s",
	72	drmaa_control_to_str(action), self->job_id, rc, lldrmaa_err_control(rc)
	73	);
	74	break;
	75	case DRMAA_CONTROL_RESUME:
	76	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"job::control: could not send %s to job %s. Function not implemented.",drmaa_control_to_str(action), self->job_id);
	77	break;
	78	case DRMAA_CONTROL_RELEASE:
	79	rc = ll_control(LL_CONTROL_VERSION, LL_CONTROL_HOLD_RELEASE, NULL,NULL, (char **)job_list,NULL,0);
	80	if(rc)
	81	fsd_exc_raise_fmt(lldrmaa_map_control(rc),
	82	"job::control: could not send %s to job %s. Error code: %d,%s",
	83	drmaa_control_to_str(action), self->job_id, rc, lldrmaa_err_control(rc)
	84	);
	85	break;
	86	case DRMAA_CONTROL_TERMINATE:
	87	cancel_info.version_num=LL_PROC_VERSION;
	88	cancel_info.msg=NULL;
	89
	90	token = strtok_r(ptr, ".", &rest);
	91	if(token == NULL)
	92	{
	93	fsd_exc_raise_fmt(FSD_DRMAA_ERRNO_INVALID_JOB,"Invalid job_id format");
	94	}
	95	cancel_info.StepId.from_host = fsd_strdup(token);
	96
	97	token = strtok_r(NULL, ".", &rest);
	98	if(token == NULL)
	99	{
	100	fsd_exc_raise_fmt(FSD_DRMAA_ERRNO_INVALID_JOB,"Invalid job_id format");
	101	}
	102
	103	cancel_info.StepId.cluster = fsd_atoi(token);
	104
	105	token = strtok_r(NULL, ".", &rest);
	106	if(token!=NULL)
	107	{
	108	cancel_info.StepId.proc = fsd_atoi(token);
	109	rc = ll_terminate_job(&cancel_info);
	110	}
	111	else /* No StepId specified */
	112	{
	113	fsd_exc_raise_fmt(FSD_DRMAA_ERRNO_INVALID_JOB, "Stepid is NULL");
	114	}
	115	if(rc)
	116	fsd_exc_raise_fmt(
	117	lldrmaa_map_terminate_job(rc),
	118	"job::control: could not send %s to job %s. Error code: %d,%s",
	119	drmaa_control_to_str(action), self->job_id, rc, lldrmaa_err_terminate_job(rc)
	120	);
	121
	122	break;
	123	default:
	124	fsd_exc_raise_fmt(
	125	FSD_ERRNO_INVALID_ARGUMENT,
	126	"job::control: unknown action %d", action );
	127	}
	128
	129	fsd_log_debug(("job::control: successful"));
	130	}
	131	FINALLY
	132	{
	133	fsd_free(ptr);
	134	if(cancel_info.StepId.from_host!=NULL)
	135	fsd_free(cancel_info.StepId.from_host);
	136	if(errObj != NULL)
	137	{
	138	ll_free_objs(errObj);
	139	ll_deallocate(errObj);
	140	}
	141
	142	fsd_mutex_unlock( &self->session->drm_connection_mutex );
	143	}
	144	END_TRY
	145
	146	fsd_log_return(( "" ));
	147	}
	148
	149	void
	150	lldrmaa_job_read_job_info( fsd_job_t *self, enum StepState step_state, enum HoldType hold_type )
	151	{
	152	fsd_log_enter(( "" ));
	153
	154	fsd_log_debug(("step state: %d",step_state));
	155	fsd_log_debug(("hold type: %d",hold_type));
	156	if( hold_type && ( step_state == STATE_HOLD \|\| step_state == STATE_IDLE ))
	157	{
	158	if (hold_type == HOLDTYPE_USER)
	159	self->state = DRMAA_PS_USER_ON_HOLD;
	160	else if (hold_type == HOLDTYPE_SYSTEM)
	161	self->state = DRMAA_PS_SYSTEM_ON_HOLD;
	162	else if (hold_type == HOLDTYPE_USERSYS)
	163	self->state = DRMAA_PS_USER_SYSTEM_ON_HOLD;
	164	else
	165	{
	166	fsd_log_error(("hold_type does not match step_state, %d", hold_type));
	167	fsd_assert(false);
	168	}
	169	}
	170	else if( step_state == STATE_DEFERRED \|\| step_state == STATE_IDLE \|\| step_state == STATE_PENDING )
	171	self->state = DRMAA_PS_QUEUED_ACTIVE;
	172	else if( step_state == STATE_RUNNING )
	173	self->state = DRMAA_PS_RUNNING;
	174	else if( step_state == STATE_PREEMPTED )
	175	self->state = DRMAA_PS_SYSTEM_SUSPENDED;
	176	else if( step_state == STATE_COMPLETED )
	177	self->state = DRMAA_PS_DONE;
	178	else if ( step_state == STATE_VACATED)
	179	{
	180	if( ((lldrmaa_session_t*) self->session)->terminate_job_on_vacated )
	181	{
	182	lldrmaa_job_control(self, DRMAA_CONTROL_TERMINATE);
	183	}
	184	self->state = DRMAA_PS_FAILED;
	185	}
	186	else if( step_state == STATE_CANCELED \|\| step_state == STATE_NOTQUEUED \|\| step_state == STATE_NOTRUN \|\|
	187	step_state == STATE_REJECTED \|\| step_state == STATE_REMOVED \|\| step_state == STATE_TERMINATED \|\|
	188	step_state == STATE_SUBMISSION_ERR)
	189	self->state = DRMAA_PS_FAILED;
	190	else if ( step_state == STATE_COMPLETE_PENDING \|\| step_state == STATE_PREEMPT_PENDING \|\|
	191	step_state == STATE_REJECT_PENDING \|\| step_state == STATE_REMOVE_PENDING \|\|
	192	step_state == STATE_RESUME_PENDING \|\| step_state == STATE_STARTING \|\| step_state == STATE_VACATE_PENDING)
	193	fsd_log_debug(("STATE_X_PENDING"));
	194	else if ( step_state == STATE_UNEXPANDED)
	195	self->state = DRMAA_PS_UNDETERMINED;
	196	else
	197	{
	198	fsd_log_error(("step_state = %d, assert(0)",step_state));
	199	fsd_assert(false);
	200	}
	201
	202	self->last_update_time = time(NULL);
	203
	204	fsd_log_return(( "" ));
	205	}
	206
	207	void
	208	lldrmaa_job_read_job_info_mon( fsd_job_t self, const char state , unsigned status)
	209	{
	210	fsd_log_enter(( "" ));
	211
	212	self->exit_status = status;
	213
	214	if( strcmp(state,"JOB_STARTED") == 0)
	215	{
	216	self->state = DRMAA_PS_RUNNING;
	217	}
	218	else if( strcmp(state,"JOB_COMPLETED") == 0)
	219	{
	220	self->state = DRMAA_PS_DONE;
	221	}
	222	else if( strcmp(state,"JOB_VACATED") == 0 )
	223	{
	224	if( ((lldrmaa_session_t*) self->session)->terminate_job_on_vacated )
	225	{
	226	lldrmaa_job_control(self, DRMAA_CONTROL_TERMINATE);
	227	}
	228	self->state = DRMAA_PS_FAILED;
	229	}
	230	else if( strcmp(state,"JOB_REJECTED") == 0)
	231	{
	232	self->state = DRMAA_PS_FAILED;
	233	}
	234	else if( strcmp(state,"JOB_REMOVED") == 0)
	235	{
	236	self->state = DRMAA_PS_FAILED;
	237	}
	238	else if( strcmp(state,"JOB_NOTRUN") == 0)
	239	{
	240	self->state = DRMAA_PS_FAILED;
	241	}
	242	else
	243	{
	244	fsd_log_error(("state = %s, assert(0)",state));
	245	fsd_assert(false);
	246	}
	247
	248	fsd_log_info(("state: %s -> %s",state, drmaa_job_ps_to_str(self->state)));
	249
	250	if( self->state >= DRMAA_PS_DONE )
	251	fsd_cond_broadcast( &self->status_cond );
	252
	253	fsd_log_return(( "" ));
	254	}
	255
	256	static void
	257	lldrmaa_job_update_status( fsd_job_t *self )
	258	{
	259	enum StepState step_state;
	260	enum HoldType hold_type;
	261	int rc;
	262	int obj_count,err_code;
	263	LL_element * job=NULL,data=NULL,step=NULL;
	264	char **host_list = NULL;
	265	lldrmaa_job_t llself = (lldrmaa_job_t) self;
	266
	267	fsd_log_enter(( "({job_id=%s})", self->job_id ));
	268
	269	fsd_mutex_lock( &self->session->drm_connection_mutex );
	270	TRY
	271	{
	272	fsd_log_debug(( "drm connection locked" ));
	273	job = ll_query(JOBS);
	274	if (!job) {
	275	fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR,
	276	"ll_query() returns NULL. The subroutine was unable to create the appropriate pointer.");
	277	}
	278
	279	fsd_calloc(host_list, 2, char *);
	280	host_list[0] = fsd_strdup(self->job_id);
	281	host_list[1] = NULL;
	282
	283	fsd_log_debug(("host_list[0]: %s",host_list[0]));
	284
	285	rc = ll_set_request(job,QUERY_STEPID,host_list,ALL_DATA);
	286	if (rc) {
	287	fsd_exc_raise_fmt(lldrmaa_map_set_request(rc),
	288	"ll_set_request() return code is non-zero. %s",lldrmaa_err_set_request(rc));
	289	}
	290
[26]	291	data = ll_get_objs(job, LL_CM, NULL, &obj_count, &err_code);
[1]	292	if (data == NULL) {
[26]	293	if (err_code == LL_GET_OBJS_NO_OBJECTS_ERR)
	294	{
	295	if (self->state != DRMAA_PS_UNDETERMINED)
	296	{
	297	fsd_log_info(("Job %s missing. Assuming finished", self->job_id));
	298	self->state = DRMAA_PS_DONE;
	299	self->exit_status = 0;
	300	}
	301	else if (llself->missing_time == 0)
	302	{
	303	llself->missing_time = time(NULL);
	304	fsd_log_debug(("Job %s missing for the first time", self->job_id)); /* Job may not yet be visible in LL */
	305	}
	306	else if (time(NULL) - llself->missing_time > LL_DRMAA_MAX_MISSING_TIME)
	307	{
	308	fsd_log_error(("Job %s missing for more then %d seconds. Assuming failef", self->job_id, LL_DRMAA_MAX_MISSING_TIME));
	309	/* Job may not yet be visible in LL */
	310	self->state = DRMAA_PS_FAILED;
	311	self->exit_status = -1;
	312	}
	313	else
	314	{
	315	fsd_log_debug(("Job %s still missing", self->job_id));
	316	}
	317	}
	318	else
	319	{
	320	fsd_log_error(("Code: %d,%d ll_get_objs() returns NULL. %s", err_code, lldrmaa_map_get_objs(err_code), lldrmaa_err_get_objs(err_code) ));
	321	}
[1]	322	} else {
	323
	324	rc = ll_get_data(data, LL_JobGetFirstStep, &step);
	325	if(rc)
	326	{
	327	fsd_exc_raise_fmt(lldrmaa_map_get_data(rc),
	328	"ll_get_data() return code is non-zero. %s",lldrmaa_err_get_data(rc));
	329	}
	330
	331	rc = ll_get_data(step, LL_StepState,&step_state);
	332	if (rc) {
	333	fsd_exc_raise_fmt(lldrmaa_map_get_data(rc),
	334	"ll_get_data() return code is non-zero. %s",lldrmaa_err_get_data(rc));
	335	}
	336
	337	rc = ll_get_data(step, LL_StepHoldType, &hold_type);
	338	if (rc) {
	339	fsd_exc_raise_fmt(lldrmaa_map_get_data(rc),
	340	"ll_get_data() return code is non-zero. %s",lldrmaa_err_get_data(rc));
	341	}
	342
[26]	343	llself->read_job_info(self, step_state, hold_type);
[1]	344
[26]	345	fsd_log_info(("LL State: %d -> DRMAA: %s", step_state, drmaa_job_ps_to_str(self->state)));
[1]	346	}
	347	}
	348	FINALLY
	349	{
	350	if(job!=NULL)
	351	{
	352	ll_free_objs(job);
	353	ll_deallocate(job);
	354	}
	355	if(data!=NULL)
	356	{
	357	ll_free_objs(data);
	358	ll_deallocate(data);
	359	}
	360	if(step!=NULL)
	361	{
	362	ll_free_objs(step);
	363	ll_deallocate(step);
	364	}
	365	if(host_list!=NULL)
	366	fsd_free_vector (host_list);
	367
	368	fsd_mutex_unlock( &self->session->drm_connection_mutex );
	369	}
	370	END_TRY
	371
	372	fsd_log_return(( "" ));
	373	}
	374
	375	fsd_job_t *
	376	lldrmaa_job_new( char *job_id )
	377	{
	378	lldrmaa_job_t *self = NULL;
	379	self = (lldrmaa_job_t*)fsd_job_new( job_id );
	380	fsd_realloc( self, 1, lldrmaa_job_t );
	381	self->super.control = lldrmaa_job_control;
	382	self->super.update_status = lldrmaa_job_update_status;
[26]	383	self->missing_time = 0;
[1]	384	self->read_job_info_mon = lldrmaa_job_read_job_info_mon;
	385	self->read_job_info = lldrmaa_job_read_job_info;
	386	return (fsd_job_t*)self;
	387	}
	388
	389	char *
	390	lldrmaa_job_create_req(
	391	fsd_drmaa_session_t *session,
	392	const fsd_template_t *jt,
	393	fsd_environ_t **envp,
	394	unsigned bulk,
	395	int start,
	396	int incr
	397	)
	398	{
	399	char *volatile cmd_file = NULL;
	400	fsd_expand_drmaa_ph_t *volatile expand = NULL;
	401	TRY
	402	{
	403	expand = fsd_expand_drmaa_ph_new( NULL, NULL, fsd_strdup("0") );
	404	cmd_file = (char *) lldrmaa_job_create_command_file( session, jt, envp, expand, bulk, start, incr );
	405	}
	406	EXCEPT_DEFAULT
	407	{
	408	remove(cmd_file);
	409	fsd_exc_reraise();
	410	}
	411	FINALLY
	412	{
	413	if( expand )
	414	expand->destroy( expand );
	415	}
	416	END_TRY
	417	return cmd_file;
	418	}
	419
	420
	421	static char *
	422	internal_map_file( fsd_expand_drmaa_ph_t expand, const char path,
	423	bool host_given, const char name )
	424	{
	425	const char *p;
	426	for( p = path; *p != ':'; p++ )
	427	if( *p == '\0' )
	428	fsd_exc_raise_fmt( FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_FORMAT,
	429	"invalid format of drmaa_%s_path: missing colon", name );
	430	if( host_given )
	431	*host_given = ( p != path );
	432	p++;
	433	return expand->expand( expand, fsd_strdup(p),
	434	FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_WD \| FSD_DRMAA_PH_INCR );
	435	}
	436
	437	char *
	438	lldrmaa_job_create_command_file(
	439	fsd_drmaa_session_t *session,
	440	const fsd_template_t *jt,
	441	fsd_environ_t **envp,
	442	fsd_expand_drmaa_ph_t *expand,
	443	unsigned n_jobs,
	444	int start,
	445	int incr
	446	)
	447	{
	448	char template_cmd[] = "/tmp/drmaa_cmd_XXXXXX";
	449	int fdt;
	450	FILE * fd;
	451
	452
	453	if((fdt = mkstemp(template_cmd) ) == -1)
	454	{
	455	fsd_log_error(("Can't create cmd file"));
	456	fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR,"Can't create cmd file");
	457	}
	458	close(fdt);
	459
	460	if((fd = fopen(template_cmd,"w") ) != NULL )
	461	{
	462	const char *input_path_orig = NULL;
	463	const char *output_path_orig = NULL;
	464	const char *error_path_orig = NULL;
	465	char *volatile input_path = NULL;
	466	char *volatile output_path = NULL;
	467	char *volatile error_path = NULL;
	468	bool input_host = false;
	469	bool output_host = false;
	470	bool error_host = false;
	471	bool join_files = false;
	472	const char *value;
	473	const char const vector;
	474	const char *job_category = "default";
	475	fsd_log_debug(("Cmd file: %s opened",template_cmd));
	476	fprintf(fd,"# File created by LoadLeveler DRMAA. If no LL DRMAA applications are running you can delete this file\n");
	477
	478	/* job name */
	479	value = jt->get_attr( jt, DRMAA_JOB_NAME );
	480	if( value )
	481	{
	482	fprintf(fd,"# @ job_name = %s\n",value);
	483	fsd_log_debug(("# @ job_name = %s",value));
	484	}
	485
	486	/* job state at submit */
	487	value = jt->get_attr( jt, DRMAA_JS_STATE );
	488	if( value )
	489	{
	490	if( 0 == strcmp( value, DRMAA_SUBMISSION_STATE_ACTIVE ) )
	491	{}
	492	else if( 0 == strcmp( value, DRMAA_SUBMISSION_STATE_HOLD ) )
	493	{
	494	fprintf(fd,"# @ hold = user\n");
	495	fsd_log_debug(("# @ hold = user"));
	496	}
	497	else
	498	fsd_exc_raise_msg(
	499	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	500	"invalid value of drmaa_js_state attribute" );
	501	}
	502
	503	/* start time */
	504	value = jt->get_attr( jt, DRMAA_START_TIME );
	505	if( value )
	506	{
	507	char buf[30];
	508	const time_t time_ll = fsd_datetime_parse( value );
	509	struct tm temp;
	510	localtime_r(&time_ll,&temp);
	511	strftime(buf, sizeof(buf),"%m/%d/%Y %H:%M:%S", &temp );
	512	fprintf(fd,"# @ startdate = %s\n", buf);
	513	fsd_log_debug(("# @ startdate = %s", buf));
	514	fsd_log_debug(( "\n drmaa_start_time: %s -> %ld", value, (long)time_ll ));
	515	}
	516
	517	/* environment */
	518	vector = jt->get_v_attr( jt, DRMAA_V_ENV );
	519	if( vector )
	520	{
	521	const char const i;
	522
	523	fprintf(fd,"# @ environment =");
	524	fsd_log_debug(("# @ environment ="));
	525	for( i = vector; *i; i++ )
	526	{
	527	fprintf(fd," %s;",*i);
	528	fsd_log_debug((" %s;",*i));
	529	}
	530	fprintf(fd," \n");
	531	}
	532
	533	/* wall clock time soft and hard limit */
	534	{
	535	const char *value2;
	536	value = jt->get_attr( jt, DRMAA_WCT_SLIMIT );
	537	value2 = jt->get_attr( jt, DRMAA_WCT_HLIMIT );
	538	if( value && value2)
	539	{
	540	fprintf(fd,"# @ wall_clock_limit = %s,%s\n",value,value2);
	541	fsd_log_debug(("# @ wall_clock_limit = %s,%s",value,value2));
	542	}
	543	else if (value)
	544	{
	545	fprintf(fd,"# @ wall_clock_limit = %s\n",value);
	546	fsd_log_debug(("# @ wall_clock_limit = %s",value));
	547	}
	548	else if (value2)
	549	{
	550	fprintf(fd,"# @ wall_clock_limit = ,%s\n",value);
	551	fsd_log_debug(("# @ wall_clock_limit = ,%s",value));
	552	}
	553	}
	554
	555	/* native specification */
	556	value = jt->get_attr( jt, DRMAA_NATIVE_SPECIFICATION );
	557	if( value )
	558	{
	559	char * temp = fsd_strdup(value);
	560	temp = fsd_replace(temp,"@","\n# $"); /* infinite loop when try replace @ with # @ */
	561	temp = fsd_replace(temp,"# $","# @ ");
	562
	563	fprintf(fd,"# Native specification:%s\n", temp);
	564	fsd_log_debug(("Native specification:%s", temp));
	565	fsd_free(temp);
	566	}
	567
	568	/* job category */
	569	value = jt->get_attr( jt, DRMAA_JOB_CATEGORY );
	570	if( value )
	571	job_category = value;
	572
	573	{
	574	char * temp;
	575	fsd_conf_option_t *category_value = NULL;
	576	category_value = fsd_conf_dict_get( session->job_categories, job_category );
	577	if( category_value != NULL )
	578	{
	579	if( category_value->type != FSD_CONF_STRING )
	580	fsd_exc_raise_fmt(
	581	FSD_ERRNO_INTERNAL_ERROR,
	582	"configuration error: job category should be string"
	583	);
	584
	585	temp = fsd_strdup(category_value->val.string);
	586	temp = fsd_replace(temp,"@","\n# $"); /* infinite loop when try replace @ with # @ */
	587	temp = fsd_replace(temp,"# $","# @ ");
	588
	589	fprintf(fd,"# Job category:%s\n", temp);
	590	fsd_log_debug(("Job category:\n%s",temp));
	591	fsd_free(temp);
	592	}
	593	else
	594	{
	595	if( value != NULL )
	596	fsd_exc_raise_fmt(
	597	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	598	"invalid job category: %s", job_category
	599	);
	600	}
	601	}
	602
	603
	604	{
	605	unsigned i;
	606	for(i=0;i<n_jobs;i++)
	607	{
	608	expand->set(expand, FSD_DRMAA_PH_INCR,fsd_asprintf("%d", start+incri)); / set current value */
	609
	610	/* job working directory */
	611	value = jt->get_attr( jt, DRMAA_WD );
	612	if( value )
	613	{
	614	char *cwd_expanded = expand->expand( expand, fsd_strdup(value), FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_INCR );
	615
	616	expand->set( expand, FSD_DRMAA_PH_WD, fsd_strdup(cwd_expanded));
	617	fprintf(fd,"# @ initialdir = %s\n",cwd_expanded);
	618	fsd_log_debug(("# @ initialdir = %s",cwd_expanded));
	619
	620	fsd_free(cwd_expanded);
	621	}
	622
	623	TRY
	624	{
	625	/* input path */
	626	input_path_orig = jt->get_attr( jt, DRMAA_INPUT_PATH );
	627	if( input_path_orig )
	628	{
	629	input_path = internal_map_file( expand, input_path_orig, &input_host,
	630	"input" );
	631	fsd_log_debug(( "\n drmaa_input_path: %s -> %s",
	632	input_path_orig, input_path ));
	633	}
	634
	635	/* output path */
	636	output_path_orig = jt->get_attr( jt, DRMAA_OUTPUT_PATH );
	637	if( output_path_orig )
	638	{
	639	output_path = internal_map_file( expand, output_path_orig, &output_host,
	640	"output" );
	641	fsd_log_debug(( "\n drmaa_output_path: %s -> %s",
	642	output_path_orig, output_path ));
	643	}
	644
	645	/* error path */
	646	error_path_orig = jt->get_attr( jt, DRMAA_ERROR_PATH );
	647	if( error_path_orig )
	648	{
	649	error_path = internal_map_file( expand, error_path_orig, &error_host,
	650	"error" );
	651	fsd_log_debug(( "\n drmaa_error_path: %s -> %s",
	652	error_path_orig, error_path ));
	653	}
	654
	655	/* join files */
	656	value = jt->get_attr( jt, DRMAA_JOIN_FILES );
	657	if( value )
	658	{
	659	if( (value[0] == 'y' \|\| value[0] == 'Y') && value[1] == '\0' )
	660	join_files = true;
	661	else if( (value[0] == 'n' \|\| value[0] == 'N') && value[1] == '\0' )
	662	join_files = false;
	663	else
	664	fsd_exc_raise_msg(
	665	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	666	"invalid value of drmaa_join_files attribute" );
	667	}
	668
	669	if( join_files )
	670	{
	671	if( output_path == NULL )
	672	fsd_exc_raise_msg(
	673	FSD_DRMAA_ERRNO_CONFLICTING_ATTRIBUTE_VALUES,
	674	"drmaa_join_files is set and output file is not given" );
	675	if( error_path!=NULL && 0 != strcmp( output_path, error_path ) )
	676	fsd_log_warning(( "Error file was given but will be ignored "
	677	"since drmaa_join_files was set." ));
	678
	679	if (error_path)
	680	fsd_free(error_path);
	681
	682	error_path = fsd_strdup(output_path);
	683	}
	684	else
	685	{
	686	if( error_path == NULL && output_path )
	687	error_path = fsd_strdup( "/dev/null" );
	688	if( output_path == NULL && error_path )
	689	output_path = fsd_strdup( "/dev/null" );
	690	}
	691
	692
	693	/* email addresses to send notifications */
	694	vector = jt->get_v_attr( jt, DRMAA_V_EMAIL );
	695	if( vector && vector[0] )
	696	{
	697	/* only to one email address message may be send */
	698	fprintf(fd,"# @ notify_user = %s\n",vector[0]);
	699	fsd_log_debug(("# @ notify_user = %s\n",vector[0]));
	700	if( vector[1] != NULL )
	701	{
	702	fsd_log_error(( "LL only supports one e-mail "
	703	"notification address" ));
	704	fsd_exc_raise_msg(FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,"LL only supports one e-mail "
	705	"notification address");
	706	}
	707	}
	708
	709	/* block email */
	710	value = jt->get_attr( jt, DRMAA_BLOCK_EMAIL );
	711	if( value )
	712	{
	713	bool block;
	714	if( strcmp(value, "0") == 0 )
	715	{
	716	block = true;
	717	fprintf(fd,"# @ notification = never\n");
	718	fsd_log_debug(("# @ notification = never"));
	719	}
	720	else if( strcmp(value, "1") == 0 )
	721	block = false;
	722	else
	723	fsd_exc_raise_msg(
	724	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	725	"invalid value of drmaa_block_email attribute" );
	726
	727	if( block && output_path == NULL )
	728	{
	729	fsd_log_debug(( "output path not set and we want to block e-mail, "
	730	"set to /dev/null" ));
	731	output_path = fsd_strdup( "/dev/null" );
	732	}
	733	}
	734
	735	if( input_path )
	736	{
	737	fprintf(fd,"# @ input = %s\n", input_path);
	738	fsd_log_debug(("# @ input = %s", input_path));
	739	}
	740
	741	if( output_path )
	742	{
	743	fprintf(fd,"# @ output = %s\n", output_path);
	744	fsd_log_debug(("# @ output = %s", output_path));
	745	}
	746
	747	if( error_path )
	748	{
	749	fprintf(fd,"# @ error = %s\n", error_path);
	750	fsd_log_debug(("# @ error = %s", error_path));
	751	}
	752	}
	753	FINALLY
	754	{
	755	fsd_free( input_path );
	756	fsd_free( output_path );
	757	fsd_free( error_path );
	758	input_path = NULL;
	759	output_path = NULL;
	760	error_path = NULL;
	761	}
	762	END_TRY
	763
	764	fprintf(fd,"# @ queue\n");
	765	fsd_log_debug(("# @ queue"));
[26]	766
	767	fprintf(fd,"\n");
	768	fprintf(fd,"echo >&2\n"); /* this line forces creation of stderr file */
	769
	770	TRY
	771	{
	772	const char *command = NULL;
	773	char *command_expanded = NULL;
	774	const char const i;
	775	int j;
	776
	777	/* remote command */
	778	command = jt->get_attr( jt, DRMAA_REMOTE_COMMAND );
	779	if( command == NULL )
	780	fsd_exc_raise_msg(
	781	FSD_DRMAA_ERRNO_CONFLICTING_ATTRIBUTE_VALUES,
	782	"drmaa_remote_command not set for job template"
	783	);
	784
	785	command_expanded = expand->expand( expand, fsd_strdup(command), FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_WD );
	786
	787	fprintf(fd,"%s", command_expanded); /* we put the commmand at the end of script (instead of using @executable keyword
	788	in roder to avoid coping of the binary */
	789	fsd_log_debug(("command = %s\n", command_expanded));
	790
	791	fsd_free(command_expanded);
	792
	793	/* arguments list */
	794	vector = jt->get_v_attr( jt, DRMAA_V_ARGV );
	795
	796	if( vector )
	797	{
	798	fsd_log_debug(("arguments = "));
	799
	800	for( i = vector, j = 2; *i; i++, j++ )
	801	{
	802	char arg_expanded = expand->expand( expand, fsd_strdup(i), FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_WD );
	803
	804	fprintf(fd," '%s'", arg_expanded);
	805	fsd_log_debug(("%s", arg_expanded));
	806
	807	fsd_free(arg_expanded);
	808	}
	809	}
	810
	811	fprintf(fd," \n");
	812	}
	813	END_TRY
	814
	815
[1]	816	}
	817	}
	818
	819	fclose(fd);
	820	}
	821	else
	822	{
	823	fsd_log_error(("Can't create cmd file"));
	824	fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR,"Can't create cmd file");
	825	}
	826
	827	return fsd_strdup(template_cmd);
	828	}
	829

Note: See TracBrowser for help on using the repository browser.

Download in other formats: