Context Navigation

job.c @ 27

Revision 27, 22.7 KB checked in by mmamonski, 12 years ago (diff)
use job category first
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
[27]	204	if( self->state >= DRMAA_PS_DONE )
	205	fsd_cond_broadcast( &self->status_cond );
	206
[1]	207	fsd_log_return(( "" ));
	208	}
	209
	210	void
	211	lldrmaa_job_read_job_info_mon( fsd_job_t self, const char state , unsigned status)
	212	{
	213	fsd_log_enter(( "" ));
	214
	215	self->exit_status = status;
	216
	217	if( strcmp(state,"JOB_STARTED") == 0)
	218	{
	219	self->state = DRMAA_PS_RUNNING;
	220	}
	221	else if( strcmp(state,"JOB_COMPLETED") == 0)
	222	{
	223	self->state = DRMAA_PS_DONE;
	224	}
	225	else if( strcmp(state,"JOB_VACATED") == 0 )
	226	{
	227	if( ((lldrmaa_session_t*) self->session)->terminate_job_on_vacated )
	228	{
	229	lldrmaa_job_control(self, DRMAA_CONTROL_TERMINATE);
	230	}
	231	self->state = DRMAA_PS_FAILED;
	232	}
	233	else if( strcmp(state,"JOB_REJECTED") == 0)
	234	{
	235	self->state = DRMAA_PS_FAILED;
	236	}
	237	else if( strcmp(state,"JOB_REMOVED") == 0)
	238	{
	239	self->state = DRMAA_PS_FAILED;
	240	}
	241	else if( strcmp(state,"JOB_NOTRUN") == 0)
	242	{
	243	self->state = DRMAA_PS_FAILED;
	244	}
	245	else
	246	{
	247	fsd_log_error(("state = %s, assert(0)",state));
	248	fsd_assert(false);
	249	}
	250
	251	fsd_log_info(("state: %s -> %s",state, drmaa_job_ps_to_str(self->state)));
	252
	253	if( self->state >= DRMAA_PS_DONE )
	254	fsd_cond_broadcast( &self->status_cond );
	255
	256	fsd_log_return(( "" ));
	257	}
	258
	259	static void
	260	lldrmaa_job_update_status( fsd_job_t *self )
	261	{
	262	enum StepState step_state;
	263	enum HoldType hold_type;
	264	int rc;
	265	int obj_count,err_code;
	266	LL_element * job=NULL,data=NULL,step=NULL;
	267	char **host_list = NULL;
	268	lldrmaa_job_t llself = (lldrmaa_job_t) self;
	269
	270	fsd_log_enter(( "({job_id=%s})", self->job_id ));
	271
	272	fsd_mutex_lock( &self->session->drm_connection_mutex );
	273	TRY
	274	{
	275	fsd_log_debug(( "drm connection locked" ));
	276	job = ll_query(JOBS);
	277	if (!job) {
	278	fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR,
	279	"ll_query() returns NULL. The subroutine was unable to create the appropriate pointer.");
	280	}
	281
	282	fsd_calloc(host_list, 2, char *);
	283	host_list[0] = fsd_strdup(self->job_id);
	284	host_list[1] = NULL;
	285
	286	fsd_log_debug(("host_list[0]: %s",host_list[0]));
	287
	288	rc = ll_set_request(job,QUERY_STEPID,host_list,ALL_DATA);
	289	if (rc) {
	290	fsd_exc_raise_fmt(lldrmaa_map_set_request(rc),
	291	"ll_set_request() return code is non-zero. %s",lldrmaa_err_set_request(rc));
	292	}
	293
[26]	294	data = ll_get_objs(job, LL_CM, NULL, &obj_count, &err_code);
[1]	295	if (data == NULL) {
[26]	296	if (err_code == LL_GET_OBJS_NO_OBJECTS_ERR)
	297	{
	298	if (self->state != DRMAA_PS_UNDETERMINED)
	299	{
	300	fsd_log_info(("Job %s missing. Assuming finished", self->job_id));
	301	self->state = DRMAA_PS_DONE;
	302	self->exit_status = 0;
[27]	303	fsd_cond_broadcast( &self->status_cond );
[26]	304	}
	305	else if (llself->missing_time == 0)
	306	{
	307	llself->missing_time = time(NULL);
	308	fsd_log_debug(("Job %s missing for the first time", self->job_id)); /* Job may not yet be visible in LL */
[27]	309	fsd_cond_broadcast( &self->status_cond );
[26]	310	}
	311	else if (time(NULL) - llself->missing_time > LL_DRMAA_MAX_MISSING_TIME)
	312	{
[27]	313	fsd_log_error(("Job %s missing for more then %d seconds. Assuming failed", self->job_id, LL_DRMAA_MAX_MISSING_TIME));
[26]	314	/* Job may not yet be visible in LL */
	315	self->state = DRMAA_PS_FAILED;
	316	self->exit_status = -1;
	317	}
	318	else
	319	{
	320	fsd_log_debug(("Job %s still missing", self->job_id));
	321	}
	322	}
	323	else
	324	{
	325	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) ));
	326	}
[1]	327	} else {
	328
	329	rc = ll_get_data(data, LL_JobGetFirstStep, &step);
	330	if(rc)
	331	{
	332	fsd_exc_raise_fmt(lldrmaa_map_get_data(rc),
	333	"ll_get_data() return code is non-zero. %s",lldrmaa_err_get_data(rc));
	334	}
	335
	336	rc = ll_get_data(step, LL_StepState,&step_state);
	337	if (rc) {
	338	fsd_exc_raise_fmt(lldrmaa_map_get_data(rc),
	339	"ll_get_data() return code is non-zero. %s",lldrmaa_err_get_data(rc));
	340	}
	341
	342	rc = ll_get_data(step, LL_StepHoldType, &hold_type);
	343	if (rc) {
	344	fsd_exc_raise_fmt(lldrmaa_map_get_data(rc),
	345	"ll_get_data() return code is non-zero. %s",lldrmaa_err_get_data(rc));
	346	}
	347
[26]	348	llself->read_job_info(self, step_state, hold_type);
[1]	349
[27]	350	fsd_log_info(("Job %s LL State: %d -> DRMAA: %s", self->job_id, step_state, drmaa_job_ps_to_str(self->state)));
[1]	351	}
	352	}
	353	FINALLY
	354	{
	355	if(job!=NULL)
	356	{
	357	ll_free_objs(job);
	358	ll_deallocate(job);
	359	}
	360	if(data!=NULL)
	361	{
	362	ll_free_objs(data);
	363	ll_deallocate(data);
	364	}
	365	if(step!=NULL)
	366	{
	367	ll_free_objs(step);
	368	ll_deallocate(step);
	369	}
	370	if(host_list!=NULL)
	371	fsd_free_vector (host_list);
	372
	373	fsd_mutex_unlock( &self->session->drm_connection_mutex );
	374	}
	375	END_TRY
	376
	377	fsd_log_return(( "" ));
	378	}
	379
	380	fsd_job_t *
	381	lldrmaa_job_new( char *job_id )
	382	{
	383	lldrmaa_job_t *self = NULL;
	384	self = (lldrmaa_job_t*)fsd_job_new( job_id );
	385	fsd_realloc( self, 1, lldrmaa_job_t );
	386	self->super.control = lldrmaa_job_control;
	387	self->super.update_status = lldrmaa_job_update_status;
[26]	388	self->missing_time = 0;
[1]	389	self->read_job_info_mon = lldrmaa_job_read_job_info_mon;
	390	self->read_job_info = lldrmaa_job_read_job_info;
	391	return (fsd_job_t*)self;
	392	}
	393
	394	char *
	395	lldrmaa_job_create_req(
	396	fsd_drmaa_session_t *session,
	397	const fsd_template_t *jt,
	398	fsd_environ_t **envp,
	399	unsigned bulk,
	400	int start,
	401	int incr
	402	)
	403	{
	404	char *volatile cmd_file = NULL;
	405	fsd_expand_drmaa_ph_t *volatile expand = NULL;
	406	TRY
	407	{
	408	expand = fsd_expand_drmaa_ph_new( NULL, NULL, fsd_strdup("0") );
	409	cmd_file = (char *) lldrmaa_job_create_command_file( session, jt, envp, expand, bulk, start, incr );
	410	}
	411	EXCEPT_DEFAULT
	412	{
	413	remove(cmd_file);
	414	fsd_exc_reraise();
	415	}
	416	FINALLY
	417	{
	418	if( expand )
	419	expand->destroy( expand );
	420	}
	421	END_TRY
	422	return cmd_file;
	423	}
	424
	425
	426	static char *
	427	internal_map_file( fsd_expand_drmaa_ph_t expand, const char path,
	428	bool host_given, const char name )
	429	{
	430	const char *p;
	431	for( p = path; *p != ':'; p++ )
	432	if( *p == '\0' )
	433	fsd_exc_raise_fmt( FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_FORMAT,
	434	"invalid format of drmaa_%s_path: missing colon", name );
	435	if( host_given )
	436	*host_given = ( p != path );
	437	p++;
	438	return expand->expand( expand, fsd_strdup(p),
	439	FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_WD \| FSD_DRMAA_PH_INCR );
	440	}
	441
	442	char *
	443	lldrmaa_job_create_command_file(
	444	fsd_drmaa_session_t *session,
	445	const fsd_template_t *jt,
	446	fsd_environ_t **envp,
	447	fsd_expand_drmaa_ph_t *expand,
	448	unsigned n_jobs,
	449	int start,
	450	int incr
	451	)
	452	{
	453	char template_cmd[] = "/tmp/drmaa_cmd_XXXXXX";
	454	int fdt;
	455	FILE * fd;
	456
	457
	458	if((fdt = mkstemp(template_cmd) ) == -1)
	459	{
	460	fsd_log_error(("Can't create cmd file"));
	461	fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR,"Can't create cmd file");
	462	}
	463	close(fdt);
	464
	465	if((fd = fopen(template_cmd,"w") ) != NULL )
	466	{
	467	const char *input_path_orig = NULL;
	468	const char *output_path_orig = NULL;
	469	const char *error_path_orig = NULL;
	470	char *volatile input_path = NULL;
	471	char *volatile output_path = NULL;
	472	char *volatile error_path = NULL;
	473	bool input_host = false;
	474	bool output_host = false;
	475	bool error_host = false;
	476	bool join_files = false;
	477	const char *value;
	478	const char const vector;
	479	const char *job_category = "default";
	480	fsd_log_debug(("Cmd file: %s opened",template_cmd));
	481	fprintf(fd,"# File created by LoadLeveler DRMAA. If no LL DRMAA applications are running you can delete this file\n");
	482
[27]	483	/* job category */
	484	value = jt->get_attr( jt, DRMAA_JOB_CATEGORY );
	485	if( value )
	486	job_category = value;
	487
	488	{
	489	char * temp;
	490	fsd_conf_option_t *category_value = NULL;
	491	category_value = fsd_conf_dict_get( session->job_categories, job_category );
	492	if( category_value != NULL )
	493	{
	494	if( category_value->type != FSD_CONF_STRING )
	495	fsd_exc_raise_fmt(
	496	FSD_ERRNO_INTERNAL_ERROR,
	497	"configuration error: job category should be string"
	498	);
	499
	500	temp = fsd_strdup(category_value->val.string);
	501	temp = fsd_replace(temp,"@","\n# $"); /* infinite loop when try replace @ with # @ */
	502	temp = fsd_replace(temp,"# $","# @ ");
	503
	504	fprintf(fd,"# Job category:%s\n", temp);
	505	fsd_log_debug(("Job category:\n%s",temp));
	506	fsd_free(temp);
	507	}
	508	else
	509	{
	510	if( value != NULL )
	511	fsd_exc_raise_fmt(
	512	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	513	"invalid job category: %s", job_category
	514	);
	515	}
	516	}
	517
[1]	518	/* job name */
	519	value = jt->get_attr( jt, DRMAA_JOB_NAME );
	520	if( value )
	521	{
	522	fprintf(fd,"# @ job_name = %s\n",value);
	523	fsd_log_debug(("# @ job_name = %s",value));
	524	}
	525
	526	/* job state at submit */
	527	value = jt->get_attr( jt, DRMAA_JS_STATE );
	528	if( value )
	529	{
	530	if( 0 == strcmp( value, DRMAA_SUBMISSION_STATE_ACTIVE ) )
	531	{}
	532	else if( 0 == strcmp( value, DRMAA_SUBMISSION_STATE_HOLD ) )
	533	{
	534	fprintf(fd,"# @ hold = user\n");
	535	fsd_log_debug(("# @ hold = user"));
	536	}
	537	else
	538	fsd_exc_raise_msg(
	539	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	540	"invalid value of drmaa_js_state attribute" );
	541	}
	542
	543	/* start time */
	544	value = jt->get_attr( jt, DRMAA_START_TIME );
	545	if( value )
	546	{
	547	char buf[30];
	548	const time_t time_ll = fsd_datetime_parse( value );
	549	struct tm temp;
	550	localtime_r(&time_ll,&temp);
	551	strftime(buf, sizeof(buf),"%m/%d/%Y %H:%M:%S", &temp );
	552	fprintf(fd,"# @ startdate = %s\n", buf);
	553	fsd_log_debug(("# @ startdate = %s", buf));
	554	fsd_log_debug(( "\n drmaa_start_time: %s -> %ld", value, (long)time_ll ));
	555	}
	556
	557	/* environment */
	558	vector = jt->get_v_attr( jt, DRMAA_V_ENV );
	559	if( vector )
	560	{
	561	const char const i;
	562
	563	fprintf(fd,"# @ environment =");
	564	fsd_log_debug(("# @ environment ="));
	565	for( i = vector; *i; i++ )
	566	{
	567	fprintf(fd," %s;",*i);
	568	fsd_log_debug((" %s;",*i));
	569	}
	570	fprintf(fd," \n");
	571	}
	572
	573	/* wall clock time soft and hard limit */
	574	{
	575	const char *value2;
	576	value = jt->get_attr( jt, DRMAA_WCT_SLIMIT );
	577	value2 = jt->get_attr( jt, DRMAA_WCT_HLIMIT );
	578	if( value && value2)
	579	{
	580	fprintf(fd,"# @ wall_clock_limit = %s,%s\n",value,value2);
	581	fsd_log_debug(("# @ wall_clock_limit = %s,%s",value,value2));
	582	}
	583	else if (value)
	584	{
	585	fprintf(fd,"# @ wall_clock_limit = %s\n",value);
	586	fsd_log_debug(("# @ wall_clock_limit = %s",value));
	587	}
	588	else if (value2)
	589	{
	590	fprintf(fd,"# @ wall_clock_limit = ,%s\n",value);
	591	fsd_log_debug(("# @ wall_clock_limit = ,%s",value));
	592	}
	593	}
	594
	595	/* native specification */
	596	value = jt->get_attr( jt, DRMAA_NATIVE_SPECIFICATION );
	597	if( value )
	598	{
	599	char * temp = fsd_strdup(value);
	600	temp = fsd_replace(temp,"@","\n# $"); /* infinite loop when try replace @ with # @ */
	601	temp = fsd_replace(temp,"# $","# @ ");
	602
	603	fprintf(fd,"# Native specification:%s\n", temp);
	604	fsd_log_debug(("Native specification:%s", temp));
	605	fsd_free(temp);
	606	}
	607
	608
	609
	610	{
	611	unsigned i;
	612	for(i=0;i<n_jobs;i++)
	613	{
	614	expand->set(expand, FSD_DRMAA_PH_INCR,fsd_asprintf("%d", start+incri)); / set current value */
	615
	616	/* job working directory */
	617	value = jt->get_attr( jt, DRMAA_WD );
	618	if( value )
	619	{
	620	char *cwd_expanded = expand->expand( expand, fsd_strdup(value), FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_INCR );
	621
	622	expand->set( expand, FSD_DRMAA_PH_WD, fsd_strdup(cwd_expanded));
	623	fprintf(fd,"# @ initialdir = %s\n",cwd_expanded);
	624	fsd_log_debug(("# @ initialdir = %s",cwd_expanded));
	625
	626	fsd_free(cwd_expanded);
	627	}
	628
	629	TRY
	630	{
	631	/* input path */
	632	input_path_orig = jt->get_attr( jt, DRMAA_INPUT_PATH );
	633	if( input_path_orig )
	634	{
	635	input_path = internal_map_file( expand, input_path_orig, &input_host,
	636	"input" );
	637	fsd_log_debug(( "\n drmaa_input_path: %s -> %s",
	638	input_path_orig, input_path ));
	639	}
	640
	641	/* output path */
	642	output_path_orig = jt->get_attr( jt, DRMAA_OUTPUT_PATH );
	643	if( output_path_orig )
	644	{
	645	output_path = internal_map_file( expand, output_path_orig, &output_host,
	646	"output" );
	647	fsd_log_debug(( "\n drmaa_output_path: %s -> %s",
	648	output_path_orig, output_path ));
	649	}
	650
	651	/* error path */
	652	error_path_orig = jt->get_attr( jt, DRMAA_ERROR_PATH );
	653	if( error_path_orig )
	654	{
	655	error_path = internal_map_file( expand, error_path_orig, &error_host,
	656	"error" );
	657	fsd_log_debug(( "\n drmaa_error_path: %s -> %s",
	658	error_path_orig, error_path ));
	659	}
	660
	661	/* join files */
	662	value = jt->get_attr( jt, DRMAA_JOIN_FILES );
	663	if( value )
	664	{
	665	if( (value[0] == 'y' \|\| value[0] == 'Y') && value[1] == '\0' )
	666	join_files = true;
	667	else if( (value[0] == 'n' \|\| value[0] == 'N') && value[1] == '\0' )
	668	join_files = false;
	669	else
	670	fsd_exc_raise_msg(
	671	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	672	"invalid value of drmaa_join_files attribute" );
	673	}
	674
	675	if( join_files )
	676	{
	677	if( output_path == NULL )
	678	fsd_exc_raise_msg(
	679	FSD_DRMAA_ERRNO_CONFLICTING_ATTRIBUTE_VALUES,
	680	"drmaa_join_files is set and output file is not given" );
	681	if( error_path!=NULL && 0 != strcmp( output_path, error_path ) )
	682	fsd_log_warning(( "Error file was given but will be ignored "
	683	"since drmaa_join_files was set." ));
	684
	685	if (error_path)
	686	fsd_free(error_path);
	687
	688	error_path = fsd_strdup(output_path);
	689	}
	690	else
	691	{
	692	if( error_path == NULL && output_path )
	693	error_path = fsd_strdup( "/dev/null" );
	694	if( output_path == NULL && error_path )
	695	output_path = fsd_strdup( "/dev/null" );
	696	}
	697
	698
	699	/* email addresses to send notifications */
	700	vector = jt->get_v_attr( jt, DRMAA_V_EMAIL );
	701	if( vector && vector[0] )
	702	{
	703	/* only to one email address message may be send */
	704	fprintf(fd,"# @ notify_user = %s\n",vector[0]);
	705	fsd_log_debug(("# @ notify_user = %s\n",vector[0]));
	706	if( vector[1] != NULL )
	707	{
	708	fsd_log_error(( "LL only supports one e-mail "
	709	"notification address" ));
	710	fsd_exc_raise_msg(FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,"LL only supports one e-mail "
	711	"notification address");
	712	}
	713	}
	714
	715	/* block email */
	716	value = jt->get_attr( jt, DRMAA_BLOCK_EMAIL );
	717	if( value )
	718	{
	719	bool block;
	720	if( strcmp(value, "0") == 0 )
	721	{
	722	block = true;
	723	fprintf(fd,"# @ notification = never\n");
	724	fsd_log_debug(("# @ notification = never"));
	725	}
	726	else if( strcmp(value, "1") == 0 )
	727	block = false;
	728	else
	729	fsd_exc_raise_msg(
	730	FSD_DRMAA_ERRNO_INVALID_ATTRIBUTE_VALUE,
	731	"invalid value of drmaa_block_email attribute" );
	732
	733	if( block && output_path == NULL )
	734	{
	735	fsd_log_debug(( "output path not set and we want to block e-mail, "
	736	"set to /dev/null" ));
	737	output_path = fsd_strdup( "/dev/null" );
	738	}
	739	}
	740
	741	if( input_path )
	742	{
	743	fprintf(fd,"# @ input = %s\n", input_path);
	744	fsd_log_debug(("# @ input = %s", input_path));
	745	}
	746
	747	if( output_path )
	748	{
	749	fprintf(fd,"# @ output = %s\n", output_path);
	750	fsd_log_debug(("# @ output = %s", output_path));
	751	}
	752
	753	if( error_path )
	754	{
	755	fprintf(fd,"# @ error = %s\n", error_path);
	756	fsd_log_debug(("# @ error = %s", error_path));
	757	}
	758	}
	759	FINALLY
	760	{
	761	fsd_free( input_path );
	762	fsd_free( output_path );
	763	fsd_free( error_path );
	764	input_path = NULL;
	765	output_path = NULL;
	766	error_path = NULL;
	767	}
	768	END_TRY
	769
	770	fprintf(fd,"# @ queue\n");
	771	fsd_log_debug(("# @ queue"));
[26]	772
	773	fprintf(fd,"\n");
	774	fprintf(fd,"echo >&2\n"); /* this line forces creation of stderr file */
	775
	776	TRY
	777	{
	778	const char *command = NULL;
	779	char *command_expanded = NULL;
	780	const char const i;
	781	int j;
	782
	783	/* remote command */
	784	command = jt->get_attr( jt, DRMAA_REMOTE_COMMAND );
	785	if( command == NULL )
	786	fsd_exc_raise_msg(
	787	FSD_DRMAA_ERRNO_CONFLICTING_ATTRIBUTE_VALUES,
	788	"drmaa_remote_command not set for job template"
	789	);
	790
	791	command_expanded = expand->expand( expand, fsd_strdup(command), FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_WD );
	792
	793	fprintf(fd,"%s", command_expanded); /* we put the commmand at the end of script (instead of using @executable keyword
	794	in roder to avoid coping of the binary */
	795	fsd_log_debug(("command = %s\n", command_expanded));
	796
	797	fsd_free(command_expanded);
	798
	799	/* arguments list */
	800	vector = jt->get_v_attr( jt, DRMAA_V_ARGV );
	801
	802	if( vector )
	803	{
	804	fsd_log_debug(("arguments = "));
	805
	806	for( i = vector, j = 2; *i; i++, j++ )
	807	{
	808	char arg_expanded = expand->expand( expand, fsd_strdup(i), FSD_DRMAA_PH_HD \| FSD_DRMAA_PH_WD );
	809
	810	fprintf(fd," '%s'", arg_expanded);
	811	fsd_log_debug(("%s", arg_expanded));
	812
	813	fsd_free(arg_expanded);
	814	}
	815	}
	816
	817	fprintf(fd," \n");
	818	}
	819	END_TRY
	820
	821
[1]	822	}
	823	}
	824
	825	fclose(fd);
	826	}
	827	else
	828	{
	829	fsd_log_error(("Can't create cmd file"));
	830	fsd_exc_raise_msg(FSD_ERRNO_INTERNAL_ERROR,"Can't create cmd file");
	831	}
	832
	833	return fsd_strdup(template_cmd);
	834	}
	835

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

Context Navigation

source: trunk/ll_drmaa/job.c @ 27

Download in other formats: