projects/pjob/pjob.c

/*
 * Peter Saunders pjob.c
 * Copyright (C) 2000-2005 Peter Saunders
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <glib.h>
#include <glib/gprintf.h>
#include <poll.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/resource.h>

#define DEFNUMTHREAD 10
#define DEFTIMEOUT 60

/* Structure for the process to be passed to the executing thread */
struct _process_t{
        gchar *exec;
        gchar *jobname;
        GTimer *timer;
        GPid pid;
        gint stat_loc;
        gchar *file_stdout;
        gchar *file_stderr;

        GError *err;
};

typedef struct _process_t process_t;

/* Globals for config setup */
static gint numthreads = DEFNUMTHREAD;
static gboolean verbose = FALSE;
static gboolean quiet = FALSE;
static gint timeout = DEFTIMEOUT;
static gchar *parseformat = NULL;
static gchar *jobparsename = NULL;
static gchar *command = NULL;
static gchar *outdir = NULL;
static gchar *infile = NULL;
static gchar *arglist = NULL;

/* Command line options */
static GOptionEntry options[] =
{
        { "jobs", 'j', 0, G_OPTION_ARG_INT, &numthreads, "Number of jobs to run in parallel [DEFNUMTHREAD]", NULL },
        { "verbose", 'v', 0, G_OPTION_ARG_NONE, &verbose, "Be verbose", NULL },
        { "quiet", 'q', 0, G_OPTION_ARG_NONE, &quiet, "Do not print the output of the commands as they are running", NULL },
        { "timeout", 't', 0, G_OPTION_ARG_INT, &timeout, "Timeout before process is killed [DEFTIMEOUT]", NULL },
        { "format", 'f', 0, G_OPTION_ARG_STRING, &parseformat, "The order of substitiution to be used in the command", NULL },
        { "jobname", 'n', 0, G_OPTION_ARG_STRING, &jobparsename, "If format is used, which variable to use for a output dir", NULL },
        { "command", 'c', 0, G_OPTION_ARG_STRING, &command, "The command to be executed", NULL },
        { "output", 'o', 0, G_OPTION_ARG_STRING, &outdir, "Directory to put all output into", NULL },
        { "stdin", 'i', 0, G_OPTION_ARG_FILENAME, &infile, "Pass contents of filename into stdin of the executing process", NULL },
        { "argsfile", 'a', 0, G_OPTION_ARG_FILENAME, &arglist, "File for list of argumenst if you dont want to use the command line", NULL },
        { NULL }
};

/* Linked list of process jobs */
GList *proclist = NULL;

/* Take a process_t and execute it, and doing the "right thing" with the output */
void process_child(gpointer data, gpointer user_data){
        process_t *proc = (process_t*) data;
        GIOChannel *soutfile[2], *sout[2], *sinfile, *sin;
        gchar *execargv[4];
        int outpipes[2], inpipes[2];
        GError *err = NULL;

        struct pollfd fds[3];
        gint fdssize=2;

        proc->timer = g_timer_new();
        g_timer_start(proc->timer);

        /* Setup files in output dir if requested to do so */
        if(outdir != NULL){
                proc->file_stdout = g_strdup_printf("%s/%s-STDOUT", outdir, proc->jobname);
                soutfile[0] = g_io_channel_new_file(proc->file_stdout, "w", &err);

                if(soutfile[0] == NULL){
                        g_printerr("Failed to open %s for writing: %s. Skipping job\n", proc->file_stdout, err->message);
                        return;
                }
                
                proc->file_stderr = g_strdup_printf("%s/%s-STDERR", outdir, proc->jobname);
                soutfile[1] = g_io_channel_new_file(proc->file_stderr, "w", &err);

                if(soutfile[1] == NULL){
                        g_printerr("Failed to open %s for writing: %s. Skipping job\n", proc->file_stderr, err->message);
                        return;
                }       

        }


        /* Open stdin file to pass to the process */
        if(infile != NULL){
                sinfile = g_io_channel_new_file(infile, "r", NULL);
                pipe(inpipes);
        }

        /* Setup argv structure for job */
        if (verbose) g_fprintf(stderr, "Starting job '%s'\n", proc->jobname);
        execargv[0] = "/bin/sh";
        execargv[1] = "-c";
        execargv[2] = proc->exec;
        execargv[3] = NULL;

        /* Exec the job */
        if (infile == NULL){
                if( ! g_spawn_async_with_pipes(NULL, execargv, NULL, G_SPAWN_DO_NOT_REAP_CHILD, NULL, NULL, &(proc->pid), NULL, &(outpipes[0]), &(outpipes[1]), &err)){
                        g_printerr("Failed to execute job %s: %s\n", proc->jobname, err->message);
                        return;
                }
        }else{
                if( ! g_spawn_async_with_pipes(NULL, execargv, NULL, G_SPAWN_DO_NOT_REAP_CHILD, NULL, NULL, &(proc->pid), &(inpipes[1]), &(outpipes[0]), &(outpipes[1]), &err)){
                        g_printerr("Failed to execute job %s: %s\n", proc->jobname, err->message);
                        return;
                }
                close(inpipes[0]);
        }


        /* Make a stream out of the pipes for ease of reading from them */
        sout[0] = g_io_channel_unix_new(outpipes[0]);
        sout[1] = g_io_channel_unix_new(outpipes[1]);
        if(infile != NULL){
                sin = g_io_channel_unix_new(inpipes[1]);
                fds[2].fd = inpipes[1];
                fds[2].events = POLLOUT | POLLHUP;
                fdssize = 3;
        }


        /* Setup the poll events */
        fds[0].fd = outpipes[0];
        fds[1].fd = outpipes[1];
        fds[0].events = fds[1].events = POLLIN | POLLPRI | POLLHUP;
        fds[0].revents = fds[1].revents = fds[2].revents = 0;


        for(;;){
                gint x;
                gchar *readbuf;
                gint rdatasize, wdatasize;
                gboolean readdata = FALSE;
                fds[0].revents = fds[1].revents = fds[2].revents = 0;

                poll(fds, fdssize, -1);
                /* For stdout and stderr see if there is any data, and read it */
                for(x=0; x<2; x++){
                        if((fds[x].revents & POLLIN) != 0){
                                /* We have data to read */
                                g_io_channel_read_line(sout[x], &readbuf, &rdatasize, NULL, NULL);
                                if(rdatasize > 0){
                                        /* Print it if unless told not to */
                                        if(!quiet){
                                                g_printf("[%s] [%s] %s", proc->jobname, (x==0) ? "out" : "err", readbuf);
                                        }
                                        if(outdir != NULL){
                                                g_io_channel_write_chars(soutfile[x], readbuf, rdatasize, &wdatasize, NULL);
                                        }
                                        readdata = TRUE;
                                        free(readbuf);
                                }

                        }
                }
                /* See if we need to pump more data down stdin */
                if((fds[2].revents & POLLOUT) != 0){
                        /* We have data we can write */
                        gchar *nextline;
                        gint nextlinesize;
                        gint nextlinewritesize;
                        GIOStatus s;

                        /* Get the next line, and write it down the stream */
                        s = g_io_channel_read_line(sinfile, &nextline, &nextlinesize, NULL, NULL);
                        if (nextlinesize > 0){
                                g_io_channel_write_chars(sin, nextline, nextlinesize, &nextlinewritesize, NULL);
                        }
                        if (s == G_IO_STATUS_EOF){
                                g_io_channel_shutdown(sin, TRUE, NULL);
                                sin = NULL;
                                fdssize=2;
                        }

                }
                /* Even if we did get a hangup - lets make sure there is no more data to read first by looping again */
                if (readdata) continue;

                if(((fds[0].revents & POLLHUP) != 0) && ((fds[1].revents & POLLHUP) != 0)) break;
        }       

        while((waitpid(proc->pid, &(proc->stat_loc), 0)) != proc->pid);

        g_timer_stop(proc->timer);

        /* If process exited cleanly */
        if (WIFEXITED(proc->stat_loc)){
                /* Get the exit code */
                if (verbose) g_fprintf(stderr, "Job '%s' exited with code %d. Exec time: %.2f\n", proc->jobname, WEXITSTATUS(proc->stat_loc), g_timer_elapsed(proc->timer, 0));
        }else{
                /* Otherwise - find out what it died with */
                /* TODO - this doesn't work quite right.. Mainly because its looking at the shell process, so the
                 * child of the /bin/sh which does get a signal, this isn't passed up. Although, it handly tells
                 * us if the /bin/sh gets a SEGV etc ;)
                 */
                g_fprintf(stderr, "Job %s exited with signal %d. Exec time: %.2f\n", proc->jobname, (WTERMSIG(proc->stat_loc)), g_timer_elapsed(proc->timer, 0));
        }


        g_io_channel_shutdown(sout[0], TRUE, NULL);
        g_io_channel_shutdown(sout[1], TRUE, NULL);

        if((infile != NULL) && (sin != NULL)){
                g_io_channel_shutdown(sin, TRUE, NULL);
        }
        if (outdir != NULL){
                g_io_channel_shutdown(soutfile[0], TRUE, NULL);
                g_io_channel_shutdown(soutfile[1], TRUE, NULL);
        }

        g_spawn_close_pid(proc->pid);


}

/* Takes a string str, a search string, find, and string to
 * replace all occurs of find with, replace. Returns a new
 * leaving original intact.
 */
gchar *strrep(gchar *str, gchar *find, gchar *replace){
        gchar *ptr, *oldptr;
        GString *newstr = g_string_new("");
        gssize len = strlen(str);
        gint findlen = strlen(find);

        ptr = g_strstr_len(str, len, find);
        oldptr=str;
        while(ptr != NULL){
                /* Copy in data up to this point */
                g_string_append_len (newstr, oldptr, (ptr - oldptr));
                /* Put in the replacement string */
                g_string_append(newstr, replace);

                oldptr = ptr + findlen;
                /* BUG - len will now be wrong. But, i only wanted a strstr anyway :) */
                ptr = g_strstr_len(oldptr, len, find);
        }

        /* Copy remains */
        g_string_append_len (newstr, oldptr, (ptr - oldptr));

        ptr = g_string_free(newstr, FALSE);

        return ptr;
}


/* Takes a cmd before substitution, takes the characters to be substituted
 * and a line for doign the substitution with. Fills in jobname
 */
gchar *genexeccmd(gchar *cmd, gchar *fmt, gchar *line, gchar **jobname){
        gchar *newexec, *ptr;
        int x;
        gchar *linesep = " ";
        gchar *fmtsep = " ";

        gchar **line_array;
        gchar **fmt_array;

        if ( fmt == NULL ){
                /* No format given - we'll just append the options to the end of the command */
                if(jobname != NULL) *jobname = g_strdup(line);
                return g_strdup_printf("%s %s", cmd, line);
        }

        line_array = g_strsplit(line, linesep, 0);
        fmt_array = g_strsplit(fmt, fmtsep, 0);

        if(jobparsename != NULL){
                if(jobname != NULL){
                        for(x=0; fmt_array[x] != NULL; x++){
                                if (line_array[x] == NULL) break;
                                if((strcmp(fmt_array[x], jobparsename) == 0)){
                                        *jobname = g_strdup(line_array[x]);
                                        break;
                                }
                        }
                }
        }else{
                /* Not told us what they want.. We'll just use the first one */
                *jobname = g_strdup(line_array[0]);
        }

        newexec = g_strdup(cmd);
        for(x=0; line_array[x] != NULL; x++){
                if (fmt_array[x] == NULL) break;
                ptr = newexec;
                newexec = strrep(newexec, fmt_array[x], line_array[x]);
                free(ptr);
        }


        return newexec;
}

int main(int argc, char **argv){

        GThreadPool *procpool;
        GError *pp_err = NULL, *err = NULL;
        gint x;

        struct rlimit rlp;

        GOptionContext *optcontext;

        optcontext = g_option_context_new(" - parallel job executer");
        g_option_context_add_main_entries(optcontext, options, NULL);
        g_option_context_parse (optcontext, &argc, &argv, &err);

        if(command == NULL){
                g_printerr("Command required, see --help for more flags\n");
                exit(1);
        }

        if(verbose){
                g_printerr("Command '%s'\n", command);
                g_printerr("Timeout '%d'\n", timeout);
                g_printerr("Jobs '%d'\n", numthreads);
        }
        
        
        if(argc < 2 && arglist == NULL){
                /* We have no arguments */
                g_printerr("Missing arguments, see --help for details\n");
                exit(1);
        }

        if (!g_thread_supported ()){ 
                g_thread_init (NULL);
        }else{
                g_printerr("Threading not supported\n");
        }

        /* Up the number of FD's to the "hard" limit.
         * This is mainly to get around the very small default
         * solaris has, or 256
         */
        getrlimit(RLIMIT_NOFILE, &rlp);
        rlp.rlim_cur = rlp.rlim_max;
        setrlimit(RLIMIT_NOFILE, &rlp);
        
        if(verbose) g_printerr("Creating a threadpool %d in size\n", numthreads);
        procpool = g_thread_pool_new(process_child, NULL, numthreads, FALSE, &pp_err);

        /* Generate the commands and push the job onto the thread pool */
        /* If no substituion is needed */
        if (arglist != NULL){
                GIOChannel *f;
                gchar *line;
                GIOStatus status;

                f = g_io_channel_new_file(arglist, "r", &err);
                if (f == NULL){
                        g_printerr("Failed to open argfile: %s\n", err->message);
                        exit(1);
                }
                status = g_io_channel_read_line(f, &line, NULL, NULL, &err);
                while(status==G_IO_STATUS_NORMAL){
                        process_t *newproc = g_new(process_t, 1);
                        newproc->err = NULL;
                        
                        line = g_strstrip(line);
                
                        newproc->exec = genexeccmd(command, parseformat, line, &(newproc->jobname));
                        proclist = g_list_append(proclist, (gpointer) newproc);
                
                        if(verbose) g_printerr("Pushing command '%s' into thread pool queue\n", newproc->exec);
                        g_thread_pool_push(procpool, (gpointer) newproc, &(newproc->err));
                        
                        status = g_io_channel_read_line(f, &line, NULL, NULL, &err);
                }
                g_io_channel_close(f);
                
        }else{
                /* substition is needed */
                for(x=1; x<argc; x++){
                        process_t *newproc = g_new(process_t, 1);
                        newproc->err = NULL;

                        newproc->exec = genexeccmd(command, parseformat, argv[x],  &(newproc->jobname));

                        proclist = g_list_append(proclist, (gpointer) newproc);

                        if(verbose) g_printerr("Pushing command '%s' into thread pool queue\n", newproc->exec);
                        g_thread_pool_push(procpool, (gpointer) newproc, &(newproc->err));
                }
        }
                

        /* Wait for the jobs to finish */
        /* TODO - Kill jobs that don't finish in time */
        while(g_thread_pool_get_num_threads(procpool) > 0){
                g_usleep(1000);
        }

        return 0;
}
Revision:	1.6
Committed:	Mon Jan 30 21:46:59 2006 UTC (18 years, 2 months ago) by pajs
Content type:	text/plain
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.5:	+11 -0 lines
Log Message:	Implemented the use of hard/soft limits. Need to start autoconfing around stuff like that
#	User	Rev	Content
1	pajs	1.1	/*
2			* Peter Saunders pjob.c
3			* Copyright (C) 2000-2005 Peter Saunders
4			*
5			* This program is free software; you can redistribute it and/or
6			* modify it under the terms of the GNU General Public License
7			* as published by the Free Software Foundation; either version 2
8			* of the License, or (at your option) any later version.
9			*
10			* This program is distributed in the hope that it will be useful,
11			* but WITHOUT ANY WARRANTY; without even the implied warranty of
12			* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			* GNU General Public License for more details.
14			*
15			* You should have received a copy of the GNU General Public License
16			* along with this program; if not, write to the Free Software
17			* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
18			*
19			*/
20
21			#include <stdio.h>
22	pajs	1.2	#include <stdlib.h>
23			#include <unistd.h>
24			#include <string.h>
25	pajs	1.1	#include <glib.h>
26	pajs	1.2	#include <glib/gprintf.h>
27	pajs	1.1	#include <poll.h>
28	pajs	1.4	#include <sys/types.h>
29			#include <sys/wait.h>
30	pajs	1.6	#include <sys/resource.h>
31	pajs	1.1
32			#define DEFNUMTHREAD 10
33			#define DEFTIMEOUT 60
34
35			/* Structure for the process to be passed to the executing thread */
36			struct _process_t{
37			gchar *exec;
38			gchar *jobname;
39			GTimer *timer;
40			GPid pid;
41	pajs	1.4	gint stat_loc;
42	pajs	1.1	gchar *file_stdout;
43			gchar *file_stderr;
44
45			GError *err;
46			};
47
48			typedef struct _process_t process_t;
49
50			/* Globals for config setup */
51			static gint numthreads = DEFNUMTHREAD;
52			static gboolean verbose = FALSE;
53			static gboolean quiet = FALSE;
54			static gint timeout = DEFTIMEOUT;
55			static gchar *parseformat = NULL;
56			static gchar *jobparsename = NULL;
57			static gchar *command = NULL;
58			static gchar *outdir = NULL;
59			static gchar *infile = NULL;
60			static gchar *arglist = NULL;
61
62			/* Command line options */
63			static GOptionEntry options[] =
64			{
65			{ "jobs", 'j', 0, G_OPTION_ARG_INT, &numthreads, "Number of jobs to run in parallel [DEFNUMTHREAD]", NULL },
66			{ "verbose", 'v', 0, G_OPTION_ARG_NONE, &verbose, "Be verbose", NULL },
67			{ "quiet", 'q', 0, G_OPTION_ARG_NONE, &quiet, "Do not print the output of the commands as they are running", NULL },
68			{ "timeout", 't', 0, G_OPTION_ARG_INT, &timeout, "Timeout before process is killed [DEFTIMEOUT]", NULL },
69			{ "format", 'f', 0, G_OPTION_ARG_STRING, &parseformat, "The order of substitiution to be used in the command", NULL },
70			{ "jobname", 'n', 0, G_OPTION_ARG_STRING, &jobparsename, "If format is used, which variable to use for a output dir", NULL },
71			{ "command", 'c', 0, G_OPTION_ARG_STRING, &command, "The command to be executed", NULL },
72			{ "output", 'o', 0, G_OPTION_ARG_STRING, &outdir, "Directory to put all output into", NULL },
73			{ "stdin", 'i', 0, G_OPTION_ARG_FILENAME, &infile, "Pass contents of filename into stdin of the executing process", NULL },
74			{ "argsfile", 'a', 0, G_OPTION_ARG_FILENAME, &arglist, "File for list of argumenst if you dont want to use the command line", NULL },
75			{ NULL }
76			};
77
78			/* Linked list of process jobs */
79			GList *proclist = NULL;
80
81			/* Take a process_t and execute it, and doing the "right thing" with the output */
82			void process_child(gpointer data, gpointer user_data){
83			process_t proc = (process_t) data;
84			GIOChannel soutfile[2], sout[2], sinfile, sin;
85			gchar *execargv[4];
86			int outpipes[2], inpipes[2];
87			GError *err = NULL;
88
89			struct pollfd fds[3];
90			gint fdssize=2;
91
92	pajs	1.3	proc->timer = g_timer_new();
93			g_timer_start(proc->timer);
94
95	pajs	1.1	/* Setup files in output dir if requested to do so */
96			if(outdir != NULL){
97			proc->file_stdout = g_strdup_printf("%s/%s-STDOUT", outdir, proc->jobname);
98			soutfile[0] = g_io_channel_new_file(proc->file_stdout, "w", &err);
99
100			if(soutfile[0] == NULL){
101			g_printerr("Failed to open %s for writing: %s. Skipping job\n", proc->file_stdout, err->message);
102			return;
103			}
104
105			proc->file_stderr = g_strdup_printf("%s/%s-STDERR", outdir, proc->jobname);
106			soutfile[1] = g_io_channel_new_file(proc->file_stderr, "w", &err);
107
108			if(soutfile[1] == NULL){
109			g_printerr("Failed to open %s for writing: %s. Skipping job\n", proc->file_stderr, err->message);
110			return;
111			}
112
113			}
114
115
116			/* Open stdin file to pass to the process */
117			if(infile != NULL){
118			sinfile = g_io_channel_new_file(infile, "r", NULL);
119			pipe(inpipes);
120			}
121
122			/* Setup argv structure for job */
123			if (verbose) g_fprintf(stderr, "Starting job '%s'\n", proc->jobname);
124			execargv[0] = "/bin/sh";
125			execargv[1] = "-c";
126			execargv[2] = proc->exec;
127			execargv[3] = NULL;
128
129			/* Exec the job */
130			if (infile == NULL){
131	pajs	1.4	if( ! g_spawn_async_with_pipes(NULL, execargv, NULL, G_SPAWN_DO_NOT_REAP_CHILD, NULL, NULL, &(proc->pid), NULL, &(outpipes[0]), &(outpipes[1]), &err)){
132	pajs	1.1	g_printerr("Failed to execute job %s: %s\n", proc->jobname, err->message);
133			return;
134			}
135			}else{
136	pajs	1.4	if( ! g_spawn_async_with_pipes(NULL, execargv, NULL, G_SPAWN_DO_NOT_REAP_CHILD, NULL, NULL, &(proc->pid), &(inpipes[1]), &(outpipes[0]), &(outpipes[1]), &err)){
137	pajs	1.1	g_printerr("Failed to execute job %s: %s\n", proc->jobname, err->message);
138			return;
139			}
140			close(inpipes[0]);
141			}
142
143
144			/* Make a stream out of the pipes for ease of reading from them */
145			sout[0] = g_io_channel_unix_new(outpipes[0]);
146			sout[1] = g_io_channel_unix_new(outpipes[1]);
147			if(infile != NULL){
148			sin = g_io_channel_unix_new(inpipes[1]);
149			fds[2].fd = inpipes[1];
150			fds[2].events = POLLOUT \| POLLHUP;
151			fdssize = 3;
152			}
153
154
155			/* Setup the poll events */
156			fds[0].fd = outpipes[0];
157			fds[1].fd = outpipes[1];
158			fds[0].events = fds[1].events = POLLIN \| POLLPRI \| POLLHUP;
159			fds[0].revents = fds[1].revents = fds[2].revents = 0;
160
161
162			for(;;){
163			gint x;
164			gchar *readbuf;
165			gint rdatasize, wdatasize;
166			gboolean readdata = FALSE;
167			fds[0].revents = fds[1].revents = fds[2].revents = 0;
168
169			poll(fds, fdssize, -1);
170			/* For stdout and stderr see if there is any data, and read it */
171			for(x=0; x<2; x++){
172	pajs	1.5	if((fds[x].revents & POLLIN) != 0){
173	pajs	1.1	/* We have data to read */
174			g_io_channel_read_line(sout[x], &readbuf, &rdatasize, NULL, NULL);
175			if(rdatasize > 0){
176			/* Print it if unless told not to */
177			if(!quiet){
178			g_printf("[%s] [%s] %s", proc->jobname, (x==0) ? "out" : "err", readbuf);
179			}
180			if(outdir != NULL){
181			g_io_channel_write_chars(soutfile[x], readbuf, rdatasize, &wdatasize, NULL);
182			}
183			readdata = TRUE;
184			free(readbuf);
185			}
186
187			}
188			}
189			/* See if we need to pump more data down stdin */
190	pajs	1.3	if((fds[2].revents & POLLOUT) != 0){
191	pajs	1.1	/* We have data we can write */
192			gchar *nextline;
193			gint nextlinesize;
194			gint nextlinewritesize;
195			GIOStatus s;
196
197			/* Get the next line, and write it down the stream */
198			s = g_io_channel_read_line(sinfile, &nextline, &nextlinesize, NULL, NULL);
199			if (nextlinesize > 0){
200			g_io_channel_write_chars(sin, nextline, nextlinesize, &nextlinewritesize, NULL);
201			}
202			if (s == G_IO_STATUS_EOF){
203			g_io_channel_shutdown(sin, TRUE, NULL);
204	pajs	1.2	sin = NULL;
205	pajs	1.1	fdssize=2;
206			}
207
208			}
209			/* Even if we did get a hangup - lets make sure there is no more data to read first by looping again */
210			if (readdata) continue;
211
212	pajs	1.3	if(((fds[0].revents & POLLHUP) != 0) && ((fds[1].revents & POLLHUP) != 0)) break;
213	pajs	1.1	}
214	pajs	1.3
215	pajs	1.4	while((waitpid(proc->pid, &(proc->stat_loc), 0)) != proc->pid);
216
217	pajs	1.3	g_timer_stop(proc->timer);
218	pajs	1.1
219	pajs	1.4	/* If process exited cleanly */
220			if (WIFEXITED(proc->stat_loc)){
221			/* Get the exit code */
222			if (verbose) g_fprintf(stderr, "Job '%s' exited with code %d. Exec time: %.2f\n", proc->jobname, WEXITSTATUS(proc->stat_loc), g_timer_elapsed(proc->timer, 0));
223			}else{
224			/* Otherwise - find out what it died with */
225			/* TODO - this doesn't work quite right.. Mainly because its looking at the shell process, so the
226			* child of the /bin/sh which does get a signal, this isn't passed up. Although, it handly tells
227			* us if the /bin/sh gets a SEGV etc ;)
228			*/
229			g_fprintf(stderr, "Job %s exited with signal %d. Exec time: %.2f\n", proc->jobname, (WTERMSIG(proc->stat_loc)), g_timer_elapsed(proc->timer, 0));
230			}
231
232
233	pajs	1.1	g_io_channel_shutdown(sout[0], TRUE, NULL);
234			g_io_channel_shutdown(sout[1], TRUE, NULL);
235
236			if((infile != NULL) && (sin != NULL)){
237			g_io_channel_shutdown(sin, TRUE, NULL);
238			}
239			if (outdir != NULL){
240			g_io_channel_shutdown(soutfile[0], TRUE, NULL);
241			g_io_channel_shutdown(soutfile[1], TRUE, NULL);
242			}
243
244			g_spawn_close_pid(proc->pid);
245
246
247			}
248
249			/* Takes a string str, a search string, find, and string to
250			* replace all occurs of find with, replace. Returns a new
251			* leaving original intact.
252			*/
253			gchar strrep(gchar str, gchar find, gchar replace){
254			gchar ptr, oldptr;
255			GString *newstr = g_string_new("");
256			gssize len = strlen(str);
257			gint findlen = strlen(find);
258
259			ptr = g_strstr_len(str, len, find);
260			oldptr=str;
261			while(ptr != NULL){
262			/* Copy in data up to this point */
263			g_string_append_len (newstr, oldptr, (ptr - oldptr));
264			/* Put in the replacement string */
265			g_string_append(newstr, replace);
266
267			oldptr = ptr + findlen;
268			/* BUG - len will now be wrong. But, i only wanted a strstr anyway :) */
269			ptr = g_strstr_len(oldptr, len, find);
270			}
271
272			/* Copy remains */
273			g_string_append_len (newstr, oldptr, (ptr - oldptr));
274
275			ptr = g_string_free(newstr, FALSE);
276
277			return ptr;
278			}
279
280
281			/* Takes a cmd before substitution, takes the characters to be substituted
282			* and a line for doign the substitution with. Fills in jobname
283			*/
284			gchar genexeccmd(gchar cmd, gchar fmt, gchar line, gchar **jobname){
285			gchar newexec, ptr;
286			int x;
287			gchar *linesep = " ";
288			gchar *fmtsep = " ";
289
290			gchar **line_array;
291			gchar **fmt_array;
292
293			if ( fmt == NULL ){
294			/* No format given - we'll just append the options to the end of the command */
295			if(jobname != NULL) *jobname = g_strdup(line);
296			return g_strdup_printf("%s %s", cmd, line);
297			}
298
299			line_array = g_strsplit(line, linesep, 0);
300			fmt_array = g_strsplit(fmt, fmtsep, 0);
301
302			if(jobparsename != NULL){
303			if(jobname != NULL){
304			for(x=0; fmt_array[x] != NULL; x++){
305			if (line_array[x] == NULL) break;
306			if((strcmp(fmt_array[x], jobparsename) == 0)){
307			*jobname = g_strdup(line_array[x]);
308			break;
309			}
310			}
311			}
312			}else{
313			/* Not told us what they want.. We'll just use the first one */
314			*jobname = g_strdup(line_array[0]);
315			}
316
317			newexec = g_strdup(cmd);
318			for(x=0; line_array[x] != NULL; x++){
319			if (fmt_array[x] == NULL) break;
320			ptr = newexec;
321			newexec = strrep(newexec, fmt_array[x], line_array[x]);
322			free(ptr);
323			}
324
325
326			return newexec;
327			}
328
329			int main(int argc, char **argv){
330
331			GThreadPool *procpool;
332			GError pp_err = NULL, err = NULL;
333			gint x;
334
335	pajs	1.6	struct rlimit rlp;
336
337	pajs	1.1	GOptionContext *optcontext;
338
339			optcontext = g_option_context_new(" - parallel job executer");
340			g_option_context_add_main_entries(optcontext, options, NULL);
341			g_option_context_parse (optcontext, &argc, &argv, &err);
342
343			if(command == NULL){
344			g_printerr("Command required, see --help for more flags\n");
345			exit(1);
346			}
347
348			if(verbose){
349			g_printerr("Command '%s'\n", command);
350			g_printerr("Timeout '%d'\n", timeout);
351			g_printerr("Jobs '%d'\n", numthreads);
352			}
353
354
355			if(argc < 2 && arglist == NULL){
356			/* We have no arguments */
357			g_printerr("Missing arguments, see --help for details\n");
358			exit(1);
359			}
360
361			if (!g_thread_supported ()){
362			g_thread_init (NULL);
363			}else{
364			g_printerr("Threading not supported\n");
365			}
366	pajs	1.6
367			/* Up the number of FD's to the "hard" limit.
368			* This is mainly to get around the very small default
369			* solaris has, or 256
370			*/
371			getrlimit(RLIMIT_NOFILE, &rlp);
372			rlp.rlim_cur = rlp.rlim_max;
373			setrlimit(RLIMIT_NOFILE, &rlp);
374	pajs	1.1
375			if(verbose) g_printerr("Creating a threadpool %d in size\n", numthreads);
376			procpool = g_thread_pool_new(process_child, NULL, numthreads, FALSE, &pp_err);
377
378			/* Generate the commands and push the job onto the thread pool */
379			/* If no substituion is needed */
380			if (arglist != NULL){
381			GIOChannel *f;
382			gchar *line;
383			GIOStatus status;
384
385			f = g_io_channel_new_file(arglist, "r", &err);
386			if (f == NULL){
387			g_printerr("Failed to open argfile: %s\n", err->message);
388			exit(1);
389			}
390			status = g_io_channel_read_line(f, &line, NULL, NULL, &err);
391			while(status==G_IO_STATUS_NORMAL){
392			process_t *newproc = g_new(process_t, 1);
393			newproc->err = NULL;
394
395			line = g_strstrip(line);
396
397			newproc->exec = genexeccmd(command, parseformat, line, &(newproc->jobname));
398			proclist = g_list_append(proclist, (gpointer) newproc);
399
400			if(verbose) g_printerr("Pushing command '%s' into thread pool queue\n", newproc->exec);
401			g_thread_pool_push(procpool, (gpointer) newproc, &(newproc->err));
402
403			status = g_io_channel_read_line(f, &line, NULL, NULL, &err);
404			}
405			g_io_channel_close(f);
406
407			}else{
408			/* substition is needed */
409			for(x=1; x<argc; x++){
410			process_t *newproc = g_new(process_t, 1);
411			newproc->err = NULL;
412
413			newproc->exec = genexeccmd(command, parseformat, argv[x], &(newproc->jobname));
414
415			proclist = g_list_append(proclist, (gpointer) newproc);
416
417			if(verbose) g_printerr("Pushing command '%s' into thread pool queue\n", newproc->exec);
418			g_thread_pool_push(procpool, (gpointer) newproc, &(newproc->err));
419			}
420			}
421
422
423			/* Wait for the jobs to finish */
424			/* TODO - Kill jobs that don't finish in time */
425			while(g_thread_pool_get_num_threads(procpool) > 0){
426	pajs	1.2	g_usleep(1000);
427	pajs	1.1	}
428
429			return 0;
430			}