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 |
#include <stdlib.h> |
23 |
#include <unistd.h> |
24 |
#include <string.h> |
25 |
#include <glib.h> |
26 |
#include <glib/gprintf.h> |
27 |
#include <poll.h> |
28 |
|
29 |
#define DEFNUMTHREAD 10 |
30 |
#define DEFTIMEOUT 60 |
31 |
|
32 |
/* Structure for the process to be passed to the executing thread */ |
33 |
struct _process_t{ |
34 |
gchar *exec; |
35 |
gchar *jobname; |
36 |
GTimer *timer; |
37 |
GPid pid; |
38 |
gchar *file_stdout; |
39 |
gchar *file_stderr; |
40 |
|
41 |
GError *err; |
42 |
}; |
43 |
|
44 |
typedef struct _process_t process_t; |
45 |
|
46 |
/* Globals for config setup */ |
47 |
static gint numthreads = DEFNUMTHREAD; |
48 |
static gboolean verbose = FALSE; |
49 |
static gboolean quiet = FALSE; |
50 |
static gint timeout = DEFTIMEOUT; |
51 |
static gchar *parseformat = NULL; |
52 |
static gchar *jobparsename = NULL; |
53 |
static gchar *command = NULL; |
54 |
static gchar *outdir = NULL; |
55 |
static gchar *infile = NULL; |
56 |
static gchar *arglist = NULL; |
57 |
|
58 |
/* Command line options */ |
59 |
static GOptionEntry options[] = |
60 |
{ |
61 |
{ "jobs", 'j', 0, G_OPTION_ARG_INT, &numthreads, "Number of jobs to run in parallel [DEFNUMTHREAD]", NULL }, |
62 |
{ "verbose", 'v', 0, G_OPTION_ARG_NONE, &verbose, "Be verbose", NULL }, |
63 |
{ "quiet", 'q', 0, G_OPTION_ARG_NONE, &quiet, "Do not print the output of the commands as they are running", NULL }, |
64 |
{ "timeout", 't', 0, G_OPTION_ARG_INT, &timeout, "Timeout before process is killed [DEFTIMEOUT]", NULL }, |
65 |
{ "format", 'f', 0, G_OPTION_ARG_STRING, &parseformat, "The order of substitiution to be used in the command", NULL }, |
66 |
{ "jobname", 'n', 0, G_OPTION_ARG_STRING, &jobparsename, "If format is used, which variable to use for a output dir", NULL }, |
67 |
{ "command", 'c', 0, G_OPTION_ARG_STRING, &command, "The command to be executed", NULL }, |
68 |
{ "output", 'o', 0, G_OPTION_ARG_STRING, &outdir, "Directory to put all output into", NULL }, |
69 |
{ "stdin", 'i', 0, G_OPTION_ARG_FILENAME, &infile, "Pass contents of filename into stdin of the executing process", NULL }, |
70 |
{ "argsfile", 'a', 0, G_OPTION_ARG_FILENAME, &arglist, "File for list of argumenst if you dont want to use the command line", NULL }, |
71 |
{ NULL } |
72 |
}; |
73 |
|
74 |
/* Linked list of process jobs */ |
75 |
GList *proclist = NULL; |
76 |
|
77 |
/* Take a process_t and execute it, and doing the "right thing" with the output */ |
78 |
void process_child(gpointer data, gpointer user_data){ |
79 |
process_t *proc = (process_t*) data; |
80 |
GIOChannel *soutfile[2], *sout[2], *sinfile, *sin; |
81 |
gchar *execargv[4]; |
82 |
int outpipes[2], inpipes[2]; |
83 |
GError *err = NULL; |
84 |
|
85 |
struct pollfd fds[3]; |
86 |
gint fdssize=2; |
87 |
|
88 |
proc->timer = g_timer_new(); |
89 |
g_timer_start(proc->timer); |
90 |
|
91 |
/* Setup files in output dir if requested to do so */ |
92 |
if(outdir != NULL){ |
93 |
proc->file_stdout = g_strdup_printf("%s/%s-STDOUT", outdir, proc->jobname); |
94 |
soutfile[0] = g_io_channel_new_file(proc->file_stdout, "w", &err); |
95 |
|
96 |
if(soutfile[0] == NULL){ |
97 |
g_printerr("Failed to open %s for writing: %s. Skipping job\n", proc->file_stdout, err->message); |
98 |
return; |
99 |
} |
100 |
|
101 |
proc->file_stderr = g_strdup_printf("%s/%s-STDERR", outdir, proc->jobname); |
102 |
soutfile[1] = g_io_channel_new_file(proc->file_stderr, "w", &err); |
103 |
|
104 |
if(soutfile[1] == NULL){ |
105 |
g_printerr("Failed to open %s for writing: %s. Skipping job\n", proc->file_stderr, err->message); |
106 |
return; |
107 |
} |
108 |
|
109 |
} |
110 |
|
111 |
|
112 |
/* Open stdin file to pass to the process */ |
113 |
if(infile != NULL){ |
114 |
sinfile = g_io_channel_new_file(infile, "r", NULL); |
115 |
pipe(inpipes); |
116 |
} |
117 |
|
118 |
/* Setup argv structure for job */ |
119 |
if (verbose) g_fprintf(stderr, "Starting job '%s'\n", proc->jobname); |
120 |
execargv[0] = "/bin/sh"; |
121 |
execargv[1] = "-c"; |
122 |
execargv[2] = proc->exec; |
123 |
execargv[3] = NULL; |
124 |
|
125 |
/* Exec the job */ |
126 |
if (infile == NULL){ |
127 |
if( ! g_spawn_async_with_pipes(NULL, execargv, NULL, 0, NULL, NULL, &(proc->pid), NULL, &(outpipes[0]), &(outpipes[1]), &err)){ |
128 |
g_printerr("Failed to execute job %s: %s\n", proc->jobname, err->message); |
129 |
return; |
130 |
} |
131 |
}else{ |
132 |
if( ! g_spawn_async_with_pipes(NULL, execargv, NULL, 0, NULL, NULL, &(proc->pid), &(inpipes[1]), &(outpipes[0]), &(outpipes[1]), &err)){ |
133 |
g_printerr("Failed to execute job %s: %s\n", proc->jobname, err->message); |
134 |
return; |
135 |
} |
136 |
close(inpipes[0]); |
137 |
} |
138 |
|
139 |
|
140 |
/* Make a stream out of the pipes for ease of reading from them */ |
141 |
sout[0] = g_io_channel_unix_new(outpipes[0]); |
142 |
sout[1] = g_io_channel_unix_new(outpipes[1]); |
143 |
if(infile != NULL){ |
144 |
sin = g_io_channel_unix_new(inpipes[1]); |
145 |
fds[2].fd = inpipes[1]; |
146 |
fds[2].events = POLLOUT | POLLHUP; |
147 |
fdssize = 3; |
148 |
} |
149 |
|
150 |
|
151 |
/* Setup the poll events */ |
152 |
fds[0].fd = outpipes[0]; |
153 |
fds[1].fd = outpipes[1]; |
154 |
fds[0].events = fds[1].events = POLLIN | POLLPRI | POLLHUP; |
155 |
fds[0].revents = fds[1].revents = fds[2].revents = 0; |
156 |
|
157 |
|
158 |
for(;;){ |
159 |
gint x; |
160 |
gchar *readbuf; |
161 |
gint rdatasize, wdatasize; |
162 |
gboolean readdata = FALSE; |
163 |
fds[0].revents = fds[1].revents = fds[2].revents = 0; |
164 |
|
165 |
poll(fds, fdssize, -1); |
166 |
/* For stdout and stderr see if there is any data, and read it */ |
167 |
for(x=0; x<2; x++){ |
168 |
if((fds[x].revents & POLLIN) == 0){ |
169 |
/* We have data to read */ |
170 |
g_io_channel_read_line(sout[x], &readbuf, &rdatasize, NULL, NULL); |
171 |
if(rdatasize > 0){ |
172 |
/* Print it if unless told not to */ |
173 |
if(!quiet){ |
174 |
g_printf("[%s] [%s] %s", proc->jobname, (x==0) ? "out" : "err", readbuf); |
175 |
} |
176 |
if(outdir != NULL){ |
177 |
g_io_channel_write_chars(soutfile[x], readbuf, rdatasize, &wdatasize, NULL); |
178 |
} |
179 |
readdata = TRUE; |
180 |
free(readbuf); |
181 |
} |
182 |
|
183 |
} |
184 |
} |
185 |
/* See if we need to pump more data down stdin */ |
186 |
if((fds[2].revents & POLLOUT) != 0){ |
187 |
/* We have data we can write */ |
188 |
gchar *nextline; |
189 |
gint nextlinesize; |
190 |
gint nextlinewritesize; |
191 |
GIOStatus s; |
192 |
|
193 |
/* Get the next line, and write it down the stream */ |
194 |
s = g_io_channel_read_line(sinfile, &nextline, &nextlinesize, NULL, NULL); |
195 |
if (nextlinesize > 0){ |
196 |
g_io_channel_write_chars(sin, nextline, nextlinesize, &nextlinewritesize, NULL); |
197 |
} |
198 |
if (s == G_IO_STATUS_EOF){ |
199 |
g_io_channel_shutdown(sin, TRUE, NULL); |
200 |
sin = NULL; |
201 |
fdssize=2; |
202 |
} |
203 |
|
204 |
} |
205 |
/* Even if we did get a hangup - lets make sure there is no more data to read first by looping again */ |
206 |
if (readdata) continue; |
207 |
|
208 |
if(((fds[0].revents & POLLHUP) != 0) && ((fds[1].revents & POLLHUP) != 0)) break; |
209 |
} |
210 |
|
211 |
g_timer_stop(proc->timer); |
212 |
|
213 |
g_io_channel_shutdown(sout[0], TRUE, NULL); |
214 |
g_io_channel_shutdown(sout[1], TRUE, NULL); |
215 |
|
216 |
if((infile != NULL) && (sin != NULL)){ |
217 |
g_io_channel_shutdown(sin, TRUE, NULL); |
218 |
} |
219 |
if (outdir != NULL){ |
220 |
g_io_channel_shutdown(soutfile[0], TRUE, NULL); |
221 |
g_io_channel_shutdown(soutfile[1], TRUE, NULL); |
222 |
} |
223 |
|
224 |
g_spawn_close_pid(proc->pid); |
225 |
|
226 |
if (verbose) g_fprintf(stderr, "Ending job '%s'\n", proc->jobname); |
227 |
|
228 |
} |
229 |
|
230 |
/* Takes a string str, a search string, find, and string to |
231 |
* replace all occurs of find with, replace. Returns a new |
232 |
* leaving original intact. |
233 |
*/ |
234 |
gchar *strrep(gchar *str, gchar *find, gchar *replace){ |
235 |
gchar *ptr, *oldptr; |
236 |
GString *newstr = g_string_new(""); |
237 |
gssize len = strlen(str); |
238 |
gint findlen = strlen(find); |
239 |
|
240 |
ptr = g_strstr_len(str, len, find); |
241 |
oldptr=str; |
242 |
while(ptr != NULL){ |
243 |
/* Copy in data up to this point */ |
244 |
g_string_append_len (newstr, oldptr, (ptr - oldptr)); |
245 |
/* Put in the replacement string */ |
246 |
g_string_append(newstr, replace); |
247 |
|
248 |
oldptr = ptr + findlen; |
249 |
/* BUG - len will now be wrong. But, i only wanted a strstr anyway :) */ |
250 |
ptr = g_strstr_len(oldptr, len, find); |
251 |
} |
252 |
|
253 |
/* Copy remains */ |
254 |
g_string_append_len (newstr, oldptr, (ptr - oldptr)); |
255 |
|
256 |
ptr = g_string_free(newstr, FALSE); |
257 |
|
258 |
return ptr; |
259 |
} |
260 |
|
261 |
|
262 |
/* Takes a cmd before substitution, takes the characters to be substituted |
263 |
* and a line for doign the substitution with. Fills in jobname |
264 |
*/ |
265 |
gchar *genexeccmd(gchar *cmd, gchar *fmt, gchar *line, gchar **jobname){ |
266 |
gchar *newexec, *ptr; |
267 |
int x; |
268 |
gchar *linesep = " "; |
269 |
gchar *fmtsep = " "; |
270 |
|
271 |
gchar **line_array; |
272 |
gchar **fmt_array; |
273 |
|
274 |
if ( fmt == NULL ){ |
275 |
/* No format given - we'll just append the options to the end of the command */ |
276 |
if(jobname != NULL) *jobname = g_strdup(line); |
277 |
return g_strdup_printf("%s %s", cmd, line); |
278 |
} |
279 |
|
280 |
line_array = g_strsplit(line, linesep, 0); |
281 |
fmt_array = g_strsplit(fmt, fmtsep, 0); |
282 |
|
283 |
if(jobparsename != NULL){ |
284 |
if(jobname != NULL){ |
285 |
for(x=0; fmt_array[x] != NULL; x++){ |
286 |
if (line_array[x] == NULL) break; |
287 |
if((strcmp(fmt_array[x], jobparsename) == 0)){ |
288 |
*jobname = g_strdup(line_array[x]); |
289 |
break; |
290 |
} |
291 |
} |
292 |
} |
293 |
}else{ |
294 |
/* Not told us what they want.. We'll just use the first one */ |
295 |
*jobname = g_strdup(line_array[0]); |
296 |
} |
297 |
|
298 |
newexec = g_strdup(cmd); |
299 |
for(x=0; line_array[x] != NULL; x++){ |
300 |
if (fmt_array[x] == NULL) break; |
301 |
ptr = newexec; |
302 |
newexec = strrep(newexec, fmt_array[x], line_array[x]); |
303 |
free(ptr); |
304 |
} |
305 |
|
306 |
|
307 |
return newexec; |
308 |
} |
309 |
|
310 |
int main(int argc, char **argv){ |
311 |
|
312 |
GThreadPool *procpool; |
313 |
GError *pp_err = NULL, *err = NULL; |
314 |
gint x; |
315 |
|
316 |
GOptionContext *optcontext; |
317 |
|
318 |
optcontext = g_option_context_new(" - parallel job executer"); |
319 |
g_option_context_add_main_entries(optcontext, options, NULL); |
320 |
g_option_context_parse (optcontext, &argc, &argv, &err); |
321 |
|
322 |
if(command == NULL){ |
323 |
g_printerr("Command required, see --help for more flags\n"); |
324 |
exit(1); |
325 |
} |
326 |
|
327 |
if(verbose){ |
328 |
g_printerr("Command '%s'\n", command); |
329 |
g_printerr("Timeout '%d'\n", timeout); |
330 |
g_printerr("Jobs '%d'\n", numthreads); |
331 |
} |
332 |
|
333 |
|
334 |
if(argc < 2 && arglist == NULL){ |
335 |
/* We have no arguments */ |
336 |
g_printerr("Missing arguments, see --help for details\n"); |
337 |
exit(1); |
338 |
} |
339 |
|
340 |
if (!g_thread_supported ()){ |
341 |
g_thread_init (NULL); |
342 |
}else{ |
343 |
g_printerr("Threading not supported\n"); |
344 |
} |
345 |
|
346 |
if(verbose) g_printerr("Creating a threadpool %d in size\n", numthreads); |
347 |
procpool = g_thread_pool_new(process_child, NULL, numthreads, FALSE, &pp_err); |
348 |
|
349 |
/* Generate the commands and push the job onto the thread pool */ |
350 |
/* If no substituion is needed */ |
351 |
if (arglist != NULL){ |
352 |
GIOChannel *f; |
353 |
gchar *line; |
354 |
GIOStatus status; |
355 |
|
356 |
f = g_io_channel_new_file(arglist, "r", &err); |
357 |
if (f == NULL){ |
358 |
g_printerr("Failed to open argfile: %s\n", err->message); |
359 |
exit(1); |
360 |
} |
361 |
status = g_io_channel_read_line(f, &line, NULL, NULL, &err); |
362 |
while(status==G_IO_STATUS_NORMAL){ |
363 |
process_t *newproc = g_new(process_t, 1); |
364 |
newproc->err = NULL; |
365 |
|
366 |
line = g_strstrip(line); |
367 |
|
368 |
newproc->exec = genexeccmd(command, parseformat, line, &(newproc->jobname)); |
369 |
proclist = g_list_append(proclist, (gpointer) newproc); |
370 |
|
371 |
if(verbose) g_printerr("Pushing command '%s' into thread pool queue\n", newproc->exec); |
372 |
g_thread_pool_push(procpool, (gpointer) newproc, &(newproc->err)); |
373 |
|
374 |
status = g_io_channel_read_line(f, &line, NULL, NULL, &err); |
375 |
} |
376 |
g_io_channel_close(f); |
377 |
|
378 |
}else{ |
379 |
/* substition is needed */ |
380 |
for(x=1; x<argc; x++){ |
381 |
process_t *newproc = g_new(process_t, 1); |
382 |
newproc->err = NULL; |
383 |
|
384 |
newproc->exec = genexeccmd(command, parseformat, argv[x], &(newproc->jobname)); |
385 |
|
386 |
proclist = g_list_append(proclist, (gpointer) newproc); |
387 |
|
388 |
if(verbose) g_printerr("Pushing command '%s' into thread pool queue\n", newproc->exec); |
389 |
g_thread_pool_push(procpool, (gpointer) newproc, &(newproc->err)); |
390 |
} |
391 |
} |
392 |
|
393 |
|
394 |
/* Wait for the jobs to finish */ |
395 |
/* TODO - Kill jobs that don't finish in time */ |
396 |
while(g_thread_pool_get_num_threads(procpool) > 0){ |
397 |
g_usleep(1000); |
398 |
} |
399 |
|
400 |
return 0; |
401 |
} |