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`

Line
1	/* $Id$ */
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
38	#define LL_DRMAA_MAX_MISSING_TIME (60)
39
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	if( self->state >= DRMAA_PS_DONE )
205	fsd_cond_broadcast( &self->status_cond );
206
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
294	data = ll_get_objs(job, LL_CM, NULL, &obj_count, &err_code);
295	if (data == NULL) {
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;
303	fsd_cond_broadcast( &self->status_cond );
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 */
309	fsd_cond_broadcast( &self->status_cond );
310	}
311	else if (time(NULL) - llself->missing_time > LL_DRMAA_MAX_MISSING_TIME)
312	{
313	fsd_log_error(("Job %s missing for more then %d seconds. Assuming failed", self->job_id, LL_DRMAA_MAX_MISSING_TIME));
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	}
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
348	llself->read_job_info(self, step_state, hold_type);
349
350	fsd_log_info(("Job %s LL State: %d -> DRMAA: %s", self->job_id, step_state, drmaa_job_ps_to_str(self->state)));
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;
388	self->missing_time = 0;
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
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
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"));
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
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: