Context Navigation

source: trunk/pbs_drmaa/session.c @ 1

Revision 1, 25.9 KB checked in by mmamonski, 13 years ago (diff)
Torque/PBS DRMAA initial commit

Line
1	/* $Id: session.c 385 2011-01-04 18:24:05Z mamonski $ */
2	/*
3	* FedStage DRMAA for PBS Pro
4	* Copyright (C) 2006-2009 FedStage Systems
5	*
6	* This program is free software: you can redistribute it and/or modify
7	* it under the terms of the GNU General Public License as published by
8	* the Free Software Foundation, either version 3 of the License, or
9	* (at your option) any later version.
10	*
11	* This program is distributed in the hope that it will be useful,
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	* GNU General Public License for more details.
15	*
16	* You should have received a copy of the GNU General Public License
17	* along with this program. If not, see <http://www.gnu.org/licenses/>.
18	*/
19
20	#ifdef HAVE_CONFIG_H
21	# include <config.h>
22	#endif
23
24	#include <stdlib.h>
25	#include <string.h>
26	#include <unistd.h>
27	#include <sys/select.h>
28	#include <sys/stat.h>
29	#include <sys/types.h>
30	#include <fcntl.h>
31
32	#include <pbs_ifl.h>
33	#include <pbs_error.h>
34
35	#include <drmaa_utils/datetime.h>
36	#include <drmaa_utils/drmaa.h>
37	#include <drmaa_utils/iter.h>
38	#include <drmaa_utils/conf.h>
39	#include <drmaa_utils/session.h>
40	#include <drmaa_utils/datetime.h>
41
42	#include <pbs_drmaa/job.h>
43	#include <pbs_drmaa/session.h>
44	#include <pbs_drmaa/submit.h>
45	#include <pbs_drmaa/util.h>
46
47	#include <errno.h>
48
49	#ifndef lint
50	static char rcsid[]
51	# ifdef __GNUC__
52	__attribute__ ((unused))
53	# endif
54	= "$Id: session.c 385 2011-01-04 18:24:05Z mamonski $";
55	#endif
56
57	static void
58	pbsdrmaa_session_destroy( fsd_drmaa_session_t *self );
59
60	static void
61	pbsdrmaa_session_apply_configuration( fsd_drmaa_session_t *self );
62
63	static fsd_job_t *
64	pbsdrmaa_session_new_job( fsd_drmaa_session_t self, const char job_id );
65
66	static bool
67	pbsdrmaa_session_do_drm_keeps_completed_jobs( pbsdrmaa_session_t *self );
68
69	static void
70	pbsdrmaa_session_update_all_jobs_status( fsd_drmaa_session_t *self );
71
72	static void
73	pbsdrmaa_session_wait_thread( fsd_drmaa_session_t self );
74
75	static char *
76	pbsdrmaa_session_run_impl(
77	fsd_drmaa_session_t *self,
78	const fsd_template_t *jt,
79	int bulk_idx
80	);
81
82	static struct attrl *
83	pbsdrmaa_create_status_attrl(void);
84
85
86	fsd_drmaa_session_t *
87	pbsdrmaa_session_new( const char *contact )
88	{
89	pbsdrmaa_session_t *volatile self = NULL;
90
91	if( contact == NULL )
92	contact = "";
93	TRY
94	{
95	self = (pbsdrmaa_session_t*)fsd_drmaa_session_new(contact);
96	fsd_realloc( self, 1, pbsdrmaa_session_t );
97	self->super_wait_thread = NULL;
98
99	self->log_file_initial_size = 0;
100	self->pbs_conn = -1;
101	self->pbs_home = NULL;
102
103	self->wait_thread_log = false;
104	self->status_attrl = NULL;
105
106	self->super_destroy = self->super.destroy;
107	self->super.destroy = pbsdrmaa_session_destroy;
108	self->super.new_job = pbsdrmaa_session_new_job;
109	self->super.update_all_jobs_status
110	= pbsdrmaa_session_update_all_jobs_status;
111	self->super.run_impl = pbsdrmaa_session_run_impl;
112
113	self->super_apply_configuration = self->super.apply_configuration;
114	self->super.apply_configuration = pbsdrmaa_session_apply_configuration;
115
116	self->do_drm_keeps_completed_jobs =
117	pbsdrmaa_session_do_drm_keeps_completed_jobs;
118
119	self->status_attrl = pbsdrmaa_create_status_attrl();
120
121	self->pbs_conn = pbs_connect( self->super.contact );
122	fsd_log_debug(( "pbs_connect(%s) =%d", self->super.contact,
123	self->pbs_conn ));
124	if( self->pbs_conn < 0 )
125	pbsdrmaa_exc_raise_pbs( "pbs_connect" );
126
127	self->super.load_configuration( &self->super, "pbs_drmaa" );
128
129	self->super.missing_jobs = FSD_IGNORE_MISSING_JOBS;
130	if( self->do_drm_keeps_completed_jobs( self ) )
131	self->super.missing_jobs = FSD_IGNORE_QUEUED_MISSING_JOBS;
132	}
133	EXCEPT_DEFAULT
134	{
135	if( self )
136	{
137	self->super.destroy( &self->super );
138	self = NULL;
139	}
140	}
141	END_TRY
142	return (fsd_drmaa_session_t*)self;
143	}
144
145
146	void
147	pbsdrmaa_session_destroy( fsd_drmaa_session_t *self )
148	{
149	pbsdrmaa_session_t pbsself = (pbsdrmaa_session_t)self;
150	self->stop_wait_thread( self );
151	if( pbsself->pbs_conn >= 0 )
152	pbs_disconnect( pbsself->pbs_conn );
153	fsd_free( pbsself->status_attrl );
154	pbsself->super_destroy( self );
155	}
156
157
158	static char *
159	pbsdrmaa_session_run_impl(
160	fsd_drmaa_session_t *self,
161	const fsd_template_t *jt,
162	int bulk_idx
163	)
164	{
165	char *volatile job_id = NULL;
166	fsd_job_t *volatile job = NULL;
167	pbsdrmaa_submit_t *volatile submit = NULL;
168
169	fsd_log_enter(( "(jt=%p, bulk_idx=%d)", (void*)jt, bulk_idx ));
170	TRY
171	{
172	submit = pbsdrmaa_submit_new( self, jt, bulk_idx );
173	submit->eval( submit );
174	job_id = submit->submit( submit );
175	job = self->new_job( self, job_id );
176	job->submit_time = time(NULL);
177	job->flags \|= FSD_JOB_CURRENT_SESSION;
178	self->jobs->add( self->jobs, job );
179	job->release( job ); job = NULL;
180	}
181	EXCEPT_DEFAULT
182	{
183	fsd_free( job_id );
184	fsd_exc_reraise();
185	}
186	FINALLY
187	{
188	if( submit )
189	submit->destroy( submit );
190	if( job )
191	job->release( job );
192	}
193	END_TRY
194	fsd_log_return(( " =%s", job_id ));
195	return job_id;
196	}
197
198
199	static fsd_job_t *
200	pbsdrmaa_session_new_job( fsd_drmaa_session_t self, const char job_id )
201	{
202	fsd_job_t *job;
203	job = pbsdrmaa_job_new( fsd_strdup(job_id) );
204	job->session = self;
205	return job;
206	}
207
208	void
209	pbsdrmaa_session_apply_configuration( fsd_drmaa_session_t *self )
210	{
211	pbsdrmaa_session_t pbsself = (pbsdrmaa_session_t)self;
212	fsd_conf_option_t *pbs_home;
213	pbs_home = fsd_conf_dict_get(self->configuration, "pbs_home" );
214	if( pbs_home )
215	{
216	if( pbs_home->type == FSD_CONF_STRING )
217	{
218	struct stat statbuf;
219	char * volatile log_path;
220	time_t t;
221
222	pbsself->pbs_home = pbs_home->val.string;
223	fsd_log_debug(("pbs_home: %s",pbsself->pbs_home));
224	pbsself->super_wait_thread = pbsself->super.wait_thread;
225	pbsself->super.wait_thread = pbsdrmaa_session_wait_thread;
226	pbsself->wait_thread_log = true;
227
228	time(&t);
229	localtime_r(&t,&pbsself->log_file_initial_time);
230
231	if((log_path = fsd_asprintf("%s/server_logs/%04d%02d%02d",
232	pbsself->pbs_home,
233	pbsself->log_file_initial_time.tm_year + 1900,
234	pbsself->log_file_initial_time.tm_mon + 1,
235	pbsself->log_file_initial_time.tm_mday)) == NULL) {
236	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"WT - Memory allocation wasn't possible");
237	}
238
239	if(stat(log_path,&statbuf) == -1) {
240	char errbuf[256] = "InternalError";
241	(void)strerror_r(errno, errbuf, 256);
242	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"stat error: %s",errbuf);
243	}
244
245	fsd_log_debug(("Log file %s size %d",log_path,(int) statbuf.st_size));
246	pbsself->log_file_initial_size = statbuf.st_size;
247	fsd_free(log_path);
248	}
249	else
250	{
251	pbsself->super.enable_wait_thread = false;
252	pbsself->wait_thread_log = false;
253	fsd_log_debug(("pbs_home not configured. Running standard wait_thread (pooling)."));
254	}
255	}
256
257
258
259	pbsself->super_apply_configuration(self); /* call method from the superclass */
260	}
261
262
263	void
264	pbsdrmaa_session_update_all_jobs_status( fsd_drmaa_session_t *self )
265	{
266	volatile bool conn_lock = false;
267	volatile bool jobs_lock = false;
268	pbsdrmaa_session_t pbsself = (pbsdrmaa_session_t)self;
269	fsd_job_set_t *jobs = self->jobs;
270	struct batch_status *volatile status = NULL;
271
272	fsd_log_enter((""));
273
274	TRY
275	{
276	conn_lock = fsd_mutex_lock( &self->drm_connection_mutex );
277	retry:
278
279	#ifdef PBS_PROFESSIONAL
280	status = pbs_statjob( pbsself->pbs_conn, NULL, NULL, NULL );
281	#else
282	status = pbs_statjob( pbsself->pbs_conn, NULL, pbsself->status_attrl, NULL );
283	#endif
284	fsd_log_debug(( "pbs_statjob( fd=%d, job_id=NULL, attribs={...} ) =%p",
285	pbsself->pbs_conn, (void*)status ));
286	if( status == NULL && pbs_errno != 0 )
287	{
288	if (pbs_errno == PBSE_PROTOCOL \|\| pbs_errno == PBSE_EXPIRED)
289	{
290	pbs_disconnect( pbsself->pbs_conn );
291	sleep(1);
292	pbsself->pbs_conn = pbs_connect( pbsself->super.contact );
293	if( pbsself->pbs_conn < 0 )
294	pbsdrmaa_exc_raise_pbs( "pbs_connect" );
295	else
296	goto retry;
297	}
298	else
299	{
300	pbsdrmaa_exc_raise_pbs( "pbs_statjob" );
301	}
302	}
303	conn_lock = fsd_mutex_unlock( &self->drm_connection_mutex );
304
305	{
306	size_t i;
307	fsd_job_t *job;
308	jobs_lock = fsd_mutex_lock( &jobs->mutex );
309	for( i = 0; i < jobs->tab_size; i++ )
310	for( job = jobs->tab[i]; job != NULL; job = job->next )
311	{
312	fsd_mutex_lock( &job->mutex );
313	job->flags \|= FSD_JOB_MISSING;
314	fsd_mutex_unlock( &job->mutex );
315	}
316	jobs_lock = fsd_mutex_unlock( &jobs->mutex );
317	}
318
319	{
320	struct batch_status *volatile i;
321	for( i = status; i != NULL; i = i->next )
322	{
323	fsd_job_t *job = NULL;
324	fsd_log_debug(( "job_id=%s", i->name ));
325	job = self->get_job( self, i->name );
326	if( job != NULL )
327	{
328	job->flags &= ~FSD_JOB_MISSING;
329	TRY
330	{
331	((pbsdrmaa_job_t*)job)->update( job, i );
332	}
333	FINALLY
334	{
335	job->release( job );
336	}
337	END_TRY
338	}
339	}
340	}
341
342	{
343	size_t volatile i;
344	fsd_job_t *volatile job;
345	jobs_lock = fsd_mutex_lock( &jobs->mutex );
346	for( i = 0; i < jobs->tab_size; i++ )
347	for( job = jobs->tab[i]; job != NULL; job = job->next )
348	{
349	fsd_mutex_lock( &job->mutex );
350	TRY
351	{
352	if( job->flags & FSD_JOB_MISSING )
353	job->on_missing( job );
354	}
355	FINALLY{ fsd_mutex_unlock( &job->mutex ); }
356	END_TRY
357	}
358	jobs_lock = fsd_mutex_unlock( &jobs->mutex );
359	}
360	}
361	FINALLY
362	{
363	if( status != NULL )
364	pbs_statfree( status );
365	if( conn_lock )
366	conn_lock = fsd_mutex_unlock( &self->drm_connection_mutex );
367	if( jobs_lock )
368	jobs_lock = fsd_mutex_unlock( &jobs->mutex );
369	}
370	END_TRY
371
372	fsd_log_return((""));
373	}
374
375
376
377	struct attrl *
378	pbsdrmaa_create_status_attrl(void)
379	{
380	struct attrl *result = NULL;
381	struct attrl *i;
382	const int max_attribs = 16;
383	int n_attribs;
384	int j = 0;
385
386	fsd_log_enter((""));
387	fsd_calloc( result, max_attribs, struct attrl );
388	result[j++].name="job_state";
389	result[j++].name="exit_status";
390	result[j++].name="resources_used";
391	result[j++].name="ctime";
392	result[j++].name="mtime";
393	result[j++].name="qtime";
394	result[j++].name="etime";
395
396	result[j++].name="queue";
397	result[j++].name="Account_Name";
398	result[j++].name="exec_host";
399	result[j++].name="start_time";
400	result[j++].name="mtime";
401	#if 0
402	result[j].name="resources_used"; result[j].resource="walltime"; j++;
403	result[j].name="resources_used"; result[j].resource="cput"; j++;
404	result[j].name="resources_used"; result[j].resource="mem"; j++;
405	result[j].name="resources_used"; result[j].resource="vmem"; j++;
406	result[j].name="Resource_List"; result[j].resource="walltime"; j++;
407	result[j].name="Resource_List"; result[j].resource="cput"; j++;
408	result[j].name="Resource_List"; result[j].resource="mem"; j++;
409	result[j].name="Resource_List"; result[j].resource="vmem"; j++;
410	#endif
411	n_attribs = j;
412	for( i = result; true; i++ )
413	if( i+1 < result + n_attribs )
414	i->next = i+1;
415	else
416	{
417	i->next = NULL;
418	break;
419	}
420
421	#ifdef DEBUGGING
422	fsd_log_return((":"));
423	pbsdrmaa_dump_attrl( result, NULL );
424	#endif
425	return result;
426	}
427
428
429	bool
430	pbsdrmaa_session_do_drm_keeps_completed_jobs( pbsdrmaa_session_t *self )
431	{
432
433	#ifndef PBS_PROFESSIONAL
434	struct attrl default_queue_query;
435	struct attrl keep_completed_query;
436	struct batch_status *default_queue_result = NULL;
437	struct batch_status *keep_completed_result = NULL;
438	const char *default_queue = NULL;
439	const char *keep_completed = NULL;
440	volatile bool result = false;
441	volatile bool conn_lock = false;
442
443	TRY
444	{
445	default_queue_query.next = NULL;
446	default_queue_query.name = "default_queue";
447	default_queue_query.resource = NULL;
448	default_queue_query.value = NULL;
449	keep_completed_query.next = NULL;
450	keep_completed_query.name = "keep_completed";
451	keep_completed_query.resource = NULL;
452	keep_completed_query.value = NULL;
453
454	conn_lock = fsd_mutex_lock( &self->super.drm_connection_mutex );
455
456	default_queue_result =
457	pbs_statserver( self->pbs_conn, &default_queue_query, NULL );
458	if( default_queue_result == NULL )
459	pbsdrmaa_exc_raise_pbs( "pbs_statserver" );
460	if( default_queue_result->attribs
461	&& !strcmp( default_queue_result->attribs->name,
462	"default_queue" ) )
463	default_queue = default_queue_result->attribs->value;
464
465	fsd_log_debug(( "default_queue: %s", default_queue ));
466
467	if( default_queue )
468	{
469	keep_completed_result = pbs_statque( self->pbs_conn,
470	(char*)default_queue, &keep_completed_query, NULL );
471	if( keep_completed_result == NULL )
472	pbsdrmaa_exc_raise_pbs( "pbs_statque" );
473	if( keep_completed_result->attribs
474	&& !strcmp( keep_completed_result->attribs->name,
475	"keep_completed" ) )
476	keep_completed = keep_completed_result->attribs->value;
477	}
478
479	fsd_log_debug(( "keep_completed: %s", keep_completed ));
480	}
481	EXCEPT_DEFAULT
482	{
483	const fsd_exc_t *e = fsd_exc_get();
484	fsd_log_warning(( "PBS server seems not to keep completed jobs\n"
485	"detail: %s", e->message(e) ));
486	result = false;
487	}
488	ELSE
489	{
490	result = false;
491	if( default_queue == NULL )
492	fsd_log_warning(( "no default queue set on PBS server" ));
493	else if( keep_completed == NULL && self->pbs_home == NULL )
494	fsd_log_warning(( "PBS server is not configured to keep completed jobs\n"
495	"in Torque: set keep_completed parameter of default queue\n"
496	" $ qmgr -c 'set queue batch keep_completed = 60'\n"
497	" or configure DRMAA to utilize log files"
498	));
499	else
500	result = true;
501	}
502	FINALLY
503	{
504	if( default_queue_result )
505	pbs_statfree( default_queue_result );
506	if( keep_completed_result )
507	pbs_statfree( keep_completed_result );
508	if( conn_lock )
509	conn_lock = fsd_mutex_unlock( &self->super.drm_connection_mutex );
510
511	}
512	END_TRY
513	#endif
514	return false;
515	}
516
517
518	enum field
519	{
520	FLD_DATE = 0,
521	FLD_EVENT = 1,
522	FLD_OBJ = 2,
523	FLD_TYPE = 3,
524	FLD_ID = 4,
525	FLD_MSG = 5
526	};
527
528	enum field_msg
529	{
530	FLD_MSG_EXIT_STATUS = 0,
531	FLD_MSG_CPUT = 1,
532	FLD_MSG_MEM = 2,
533	FLD_MSG_VMEM = 3,
534	FLD_MSG_WALLTIME = 4
535	};
536
537	#define FLD_MSG_STATUS "0010"
538	#define FLD_MSG_STATE "0008"
539	#define FLD_MSG_LOG "0002"
540
541	ssize_t fsd_getline(char * line,ssize_t size, int fd)
542	{
543	char buf;
544	char * ptr = NULL;
545	ssize_t n = 0, rc;
546	ptr = line;
547	for(n = 1; n< size; n++)
548	{
549	if( (rc = read(fd,&buf,1 )) == 1) {
550	*ptr++ = buf;
551	if(buf == '\n')
552	{
553	break;
554	}
555	}
556	else if (rc == 0) {
557	if (n == 1)
558	return 0;
559	else
560	break;
561	}
562	else
563	return -1;
564	}
565
566	return n;
567	}
568
569	void *
570	pbsdrmaa_session_wait_thread( fsd_drmaa_session_t *self )
571	{
572	pbsdrmaa_session_t pbsself = (pbsdrmaa_session_t) self;
573	fsd_job_t *volatile job = NULL;
574	pbsdrmaa_job_t *volatile pbsjob = NULL;
575	char job_id[256] = "";
576	char event[256] = "";
577	time_t t;
578	struct tm tm;
579
580	tm = pbsself->log_file_initial_time;
581
582	fsd_log_enter(( "" ));
583	fsd_mutex_lock( &self->mutex );
584	TRY
585	{
586	char * volatile log_path = NULL;
587	char buffer[4096] = "";
588	bool volatile date_changed = true;
589	int volatile fd = -1;
590	bool first_open = true;
591
592	fsd_log_debug(("WT - reading log files"));
593
594	while( self->wait_thread_run_flag )
595	TRY
596	{
597	if(date_changed)
598	{
599	int num_tries = 0;
600
601	time(&t);
602	localtime_r(&t,&tm);
603
604	#define DRMAA_WAIT_THREAD_MAX_TRIES (12)
605	/* generate new date, close file and open new */
606	if((log_path = fsd_asprintf("%s/server_logs/%04d%02d%02d",
607	pbsself->pbs_home,
608	tm.tm_year + 1900,
609	tm.tm_mon + 1,
610	tm.tm_mday)) == NULL) {
611	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"WT - Memory allocation wasn't possible");
612	}
613
614	if(fd != -1)
615	close(fd);
616
617	fsd_log_debug(("Log file: %s",log_path));
618
619	retry:
620	if((fd = open(log_path,O_RDONLY) ) == -1 && num_tries > DRMAA_WAIT_THREAD_MAX_TRIES )
621	{
622	fsd_log_error(("Can't open log file. Verify pbs_home. Running standard wait_thread."));
623	fsd_log_error(("Remember that without keep_completed set standard wait_thread won't run correctly"));
624	/pbsself->super.enable_wait_thread = false;/ /* run not wait_thread */
625	pbsself->wait_thread_log = false;
626	pbsself->super.wait_thread = pbsself->super_wait_thread;
627	pbsself->super.wait_thread(self);
628	} else if ( fd == -1 ) {
629	fsd_log_warning(("Can't open log file: %s. Retries count: %d", log_path, num_tries));
630	num_tries++;
631	sleep(5);
632	goto retry;
633	}
634
635	fsd_free(log_path);
636
637	fsd_log_debug(("Log file opened"));
638
639	if(first_open) {
640	fsd_log_debug(("Log file lseek"));
641	if(lseek(fd,pbsself->log_file_initial_size,SEEK_SET) == (off_t) -1) {
642	char errbuf[256] = "InternalError";
643	(void)strerror_r(errno, errbuf, 256);
644	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"lseek error: %s",errbuf);
645	}
646	first_open = false;
647	}
648
649	date_changed = false;
650	}
651
652	while ((fsd_getline(buffer,sizeof(buffer),fd)) > 0)
653	{
654	const char *volatile ptr = buffer;
655	char field[256] = "";
656	int volatile field_n = 0;
657	int n;
658
659	bool volatile job_id_match = false;
660	bool volatile event_match = false;
661	bool volatile log_event = false;
662	bool volatile log_match = false;
663	char * temp_date = NULL;
664
665
666	struct batch_status status;
667	status.next = NULL;
668
669	if( strlcpy(job_id,"",sizeof(job_id)) > sizeof(job_id) ) {
670	fsd_log_error(("WT - strlcpy error"));
671	}
672	if( strlcpy(event,"",sizeof(event)) > sizeof(event) ) {
673	fsd_log_error(("WT - strlcpy error"));
674	}
675	while ( sscanf(ptr, "%255[^;]%n", field, &n) == 1 ) /* divide current line into fields */
676	{
677	if(field_n == FLD_DATE)
678	{
679	temp_date = fsd_strdup(field);
680	}
681	else if(field_n == FLD_EVENT && (strcmp(field,FLD_MSG_STATUS) == 0 \|\|
682	strcmp(field,FLD_MSG_STATE) == 0 ))
683	{
684	/* event described by log line*/
685	if(strlcpy(event, field,sizeof(event)) > sizeof(event)) {
686	fsd_log_error(("WT - strlcpy error"));
687	}
688	event_match = true;
689	}
690	else if(event_match && field_n == FLD_ID)
691	{
692	TRY
693	{
694	job = self->get_job( self, field );
695	pbsjob = (pbsdrmaa_job_t*) job;
696
697	if( job )
698	{
699	if(strlcpy(job_id, field,sizeof(job_id)) > sizeof(job_id)) {
700	fsd_log_error(("WT - strlcpy error"));
701	}
702	fsd_log_debug(("WT - job_id: %s",job_id));
703	status.name = fsd_strdup(job_id);
704	job_id_match = true; /* job_id is in drmaa */
705	}
706	else
707	{
708	fsd_log_debug(("WT - Unknown job: %s", field));
709	}
710	}
711	END_TRY
712	}
713	else if(job_id_match && field_n == FLD_MSG)
714	{
715	/* parse msg - depends on FLD_EVENT*/
716	struct attrl struct_resource_cput,struct_resource_mem,struct_resource_vmem,
717	struct_resource_walltime, struct_status, struct_state, struct_start_time,struct_mtime, struct_queue, struct_account_name;
718
719	bool state_running = false;
720
721	struct_status.name = NULL;
722	struct_status.value = NULL;
723	struct_status.next = NULL;
724	struct_status.resource = NULL;
725
726	struct_state.name = NULL;
727	struct_state.value = NULL;
728	struct_state.next = NULL;
729	struct_state.resource = NULL;
730
731	struct_resource_cput.name = NULL;
732	struct_resource_cput.value = NULL;
733	struct_resource_cput.next = NULL;
734	struct_resource_cput.resource = NULL;
735
736	struct_resource_mem.name = NULL;
737	struct_resource_mem.value = NULL;
738	struct_resource_mem.next = NULL;
739	struct_resource_mem.resource = NULL;
740
741	struct_resource_vmem.name = NULL;
742	struct_resource_vmem.value = NULL;
743	struct_resource_vmem.next = NULL;
744	struct_resource_vmem.resource = NULL;
745
746	struct_resource_walltime.name = NULL;
747	struct_resource_walltime.value = NULL;
748	struct_resource_walltime.next = NULL;
749	struct_resource_walltime.resource = NULL;
750
751	struct_start_time.name = NULL;
752	struct_start_time.value = NULL;
753	struct_start_time.next = NULL;
754	struct_start_time.resource = NULL;
755
756	struct_mtime.name = NULL;
757	struct_mtime.value = NULL;
758	struct_mtime.next = NULL;
759	struct_mtime.resource = NULL;
760
761	struct_queue.name = NULL;
762	struct_queue.value = NULL;
763	struct_queue.next = NULL;
764	struct_queue.resource = NULL;
765
766	struct_account_name.name = NULL;
767	struct_account_name.value = NULL;
768	struct_account_name.next = NULL;
769	struct_account_name.resource = NULL;
770
771
772	if (strcmp(event,FLD_MSG_STATE) == 0)
773	{
774	/* job run, modified, queued etc */
775	int n = 0;
776	status.attribs = &struct_state;
777	struct_state.next = NULL;
778	struct_state.name = "job_state";
779	if(field[0] == 'J') /* Job Queued, Job Modified, Job Run*/
780	{
781	n = 4;
782	}
783	if(field[4] == 'M') {
784	struct tm temp_time_tm;
785	memset(&temp_time_tm, 0, sizeof(temp_time_tm));
786	temp_time_tm.tm_isdst = -1;
787
788	if (strptime(temp_date, "%m/%d/%Y %H:%M:%S", &temp_time_tm) == NULL)
789	{
790	fsd_log_error(("failed to parse mtime: %s", temp_date));
791	}
792	else
793	{
794	time_t temp_time = mktime(&temp_time_tm);
795	status.attribs = &struct_mtime;
796	struct_mtime.name = "mtime";
797	struct_mtime.next = NULL;
798	struct_mtime.value = fsd_asprintf("%lu",temp_time);
799	}
800	}
801	/* != Job deleted and Job to be deleted*/
802	#ifdef PBS_PROFESSIONAL
803	else if (field[4] != 't' && field[10] != 'd') {
804	#else
805	else if(field[4] != 'd') {
806	#endif
807
808	if ((struct_state.value = fsd_asprintf("%c",field[n]) ) == NULL ) { /* 4 first letter of state */
809	fsd_exc_raise_fmt(FSD_ERRNO_INTERNAL_ERROR,"WT - Memory allocation wasn't possible");
810	}
811	if(struct_state.value[0] == 'R'){
812	state_running = true;
813	}
814	}
815	else { /* job terminated - pbs drmaa detects failed as completed with exit_status !=0, aborted with status -1*/
816	struct_status.name = "exit_status";
817	struct_status.value = fsd_strdup("-1");
818	struct_status.next = NULL;
819	struct_state.next = &struct_status;
820	struct_state.value = fsd_strdup("C");
821	}
822	}
823	else /if (strcmp(event,FLD_MSG_STATUS) == 0 )/
824	{
825	/* exit status and rusage */
826	const char *ptr2 = field;
827	char msg[ 256 ] = "";
828	int n2;
829	int msg_field_n = 0;
830
831	struct_resource_cput.name = "resources_used";
832	struct_resource_mem.name = "resources_used";
833	struct_resource_vmem.name = "resources_used";
834	struct_resource_walltime.name = "resources_used";
835	struct_status.name = "exit_status";
836	struct_state.name = "job_state";
837
838	status.attribs = &struct_resource_cput;
839	struct_resource_cput.next = &struct_resource_mem;
840	struct_resource_mem.next = &struct_resource_vmem;
841	struct_resource_vmem.next = &struct_resource_walltime;
842	struct_resource_walltime.next = &struct_status;
843	struct_status.next = &struct_state;
844	struct_state.next = NULL;
845
846	while ( sscanf(ptr2, "%255[^ ]%n", msg, &n2) == 1 )
847	{
848	switch(msg_field_n)
849	{
850	case FLD_MSG_EXIT_STATUS:
851	struct_status.value = fsd_strdup(strchr(msg,'=')+1);
852	break;
853
854	case FLD_MSG_CPUT:
855	struct_resource_cput.resource = "cput";
856	struct_resource_cput.value = fsd_strdup(strchr(msg,'=')+1);
857	break;
858
859	case FLD_MSG_MEM:
860	struct_resource_mem.resource = "mem";
861	struct_resource_mem.value = fsd_strdup(strchr(msg,'=')+1);
862	break;
863
864	case FLD_MSG_VMEM:
865	struct_resource_vmem.resource = "vmem";
866	struct_resource_vmem.value = fsd_strdup(strchr(msg,'=')+1);
867	break;
868
869	case FLD_MSG_WALLTIME:
870	struct_resource_walltime.resource = "walltime";
871	struct_resource_walltime.value = fsd_strdup(strchr(msg,'=')+1);
872	break;
873	}
874
875	ptr2 += n2;
876	msg_field_n++;
877	if ( *ptr2 != ' ' )
878	{
879	break;
880	}
881	++ptr2;
882	}
883	struct_state.value = fsd_strdup("C"); /* we got exit_status so we say that it has completed */
884	}
885
886	if ( state_running )
887	{
888	fsd_log_debug(("WT - forcing update of job: %s", job->job_id ));
889	job->update_status( job );
890	}
891	else
892	{
893	fsd_log_debug(("WT - updating job: %s", job->job_id ));
894	pbsjob->update( job, &status );
895	}
896
897
898	fsd_cond_broadcast( &job->status_cond);
899	fsd_cond_broadcast( &self->wait_condition );
900
901	if ( job )
902	job->release( job );
903
904	fsd_free(struct_resource_cput.value);
905	fsd_free(struct_resource_mem.value);
906	fsd_free(struct_resource_vmem.value);
907	fsd_free(struct_resource_walltime.value);
908	fsd_free(struct_status.value);
909	fsd_free(struct_state.value);
910	fsd_free(struct_start_time.value);
911	fsd_free(struct_mtime.value);
912	fsd_free(struct_queue.value);
913	fsd_free(struct_account_name.value);
914
915	if ( status.name!=NULL )
916	fsd_free(status.name);
917	}
918	else if(field_n == FLD_EVENT && strcmp(field,FLD_MSG_LOG) == 0)
919	{
920	log_event = true;
921	}
922	else if (log_event && field_n == FLD_ID && strcmp(field,"Log") == 0 )
923	{
924	log_match = true;
925	log_event = false;
926	}
927	else if( log_match && field_n == FLD_MSG &&
928	field[0] == 'L' &&
929	field[1] == 'o' &&
930	field[2] == 'g' &&
931	field[3] == ' ' &&
932	field[4] == 'c' &&
933	field[5] == 'l' &&
934	field[6] == 'o' &&
935	field[7] == 's' &&
936	field[8] == 'e' &&
937	field[9] == 'd' ) /* last field in the file - strange bahaviour*/
938	{
939	fsd_log_debug(("WT - Date changed. Closing log file"));
940	date_changed = true;
941	log_match = false;
942	}
943
944	ptr += n;
945	if ( *ptr != ';' )
946	{
947	break; /* end of line */
948	}
949	field_n++;
950	++ptr;
951	}
952
953	if( strlcpy(buffer,"",sizeof(buffer)) > sizeof(buffer) ) {
954	fsd_log_error(("WT - strlcpy error"));
955	}
956
957	fsd_free(temp_date);
958	}
959
960	fsd_mutex_unlock( &self->mutex );
961	usleep(1000000);
962	fsd_mutex_lock( &self->mutex );
963	}
964	EXCEPT_DEFAULT
965	{
966	const fsd_exc_t *e = fsd_exc_get();
967
968	fsd_log_error(( "wait thread: <%d:%s>", e->code(e), e->message(e) ));
969	fsd_exc_reraise();
970	}
971	END_TRY
972
973	if(fd != -1)
974	close(fd);
975	fsd_log_debug(("Log file closed"));
976	}
977	FINALLY
978	{
979	fsd_log_debug(("WT - Terminated."));
980	fsd_mutex_unlock( &self->mutex );
981	}
982	END_TRY
983
984	fsd_log_return(( " =NULL" ));
985	return NULL;
986	}

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

Download in other formats: