138 |
p->commands_max = 4; |
p->commands_max = 4; |
139 |
p->commands = xmalloc (p->commands_max * sizeof *p->commands); |
p->commands = xmalloc (p->commands_max * sizeof *p->commands); |
140 |
p->pids = NULL; |
p->pids = NULL; |
141 |
|
p->writeto = p->readfrom = 0; |
142 |
|
p->writefd = p->readfd = -1; |
143 |
return p; |
return p; |
144 |
} |
} |
145 |
|
|
210 |
void pipeline_start (pipeline *p) |
void pipeline_start (pipeline *p) |
211 |
{ |
{ |
212 |
int i; |
int i; |
213 |
int last_input = 0; |
int last_input = -1; |
214 |
|
int infd[2]; |
215 |
|
|
216 |
assert (!p->pids); /* pipeline not started already */ |
assert (!p->pids); /* pipeline not started already */ |
|
|
|
217 |
p->pids = xmalloc (p->ncommands * sizeof *p->pids); |
p->pids = xmalloc (p->ncommands * sizeof *p->pids); |
218 |
|
|
219 |
|
if (p->writeto) { |
220 |
|
if (pipe (infd) < 0) |
221 |
|
error (FATAL, errno, _("pipe failed")); |
222 |
|
last_input = infd[0]; |
223 |
|
p->writefd = infd[1]; |
224 |
|
} |
225 |
|
|
226 |
for (i = 0; i < p->ncommands; i++) { |
for (i = 0; i < p->ncommands; i++) { |
227 |
int pdes[2]; |
int pdes[2]; |
228 |
pid_t pid; |
pid_t pid; |
229 |
|
|
230 |
if (i != p->ncommands - 1) { |
if (i != p->ncommands - 1 || p->readfrom) { |
231 |
if (pipe (pdes) < 0) |
if (pipe (pdes) < 0) |
232 |
error (FATAL, errno, _("pipe failed")); |
error (FATAL, errno, _("pipe failed")); |
233 |
|
if (i == p->ncommands - 1) |
234 |
|
p->readfd = pdes[0]; |
235 |
} |
} |
236 |
|
|
237 |
pid = fork (); |
pid = fork (); |
238 |
if (pid < 0) |
if (pid < 0) |
239 |
error (FATAL, errno, _("fork failed")); |
error (FATAL, errno, _("fork failed")); |
240 |
if (pid == 0) { |
if (pid == 0) { |
241 |
/* child */ |
/* child */ |
242 |
if (last_input != 0) { |
if (last_input != -1) { |
243 |
if (close (0) < 0) |
if (close (0) < 0) |
244 |
error (FATAL, errno, |
error (FATAL, errno, |
245 |
_("close failed")); |
_("close failed")); |
249 |
error (FATAL, errno, |
error (FATAL, errno, |
250 |
_("close failed")); |
_("close failed")); |
251 |
} |
} |
252 |
if (i != p->ncommands - 1) { |
|
253 |
|
if (i != p->ncommands - 1 || p->readfrom) { |
254 |
if (close (1) < 0) |
if (close (1) < 0) |
255 |
error (FATAL, errno, |
error (FATAL, errno, |
256 |
_("close failed")); |
_("close failed")); |
263 |
error (FATAL, errno, |
error (FATAL, errno, |
264 |
_("close failed")); |
_("close failed")); |
265 |
} |
} |
266 |
|
|
267 |
|
if (p->writefd != -1) |
268 |
|
close (p->writefd); |
269 |
|
|
270 |
execvp (p->commands[i]->name, p->commands[i]->argv); |
execvp (p->commands[i]->name, p->commands[i]->argv); |
271 |
error (EXEC_FAILED_EXIT_STATUS, errno, |
error (EXEC_FAILED_EXIT_STATUS, errno, |
272 |
_("couldn't exec %s"), p->commands[i]->name); |
_("couldn't exec %s"), p->commands[i]->name); |
273 |
} |
} |
274 |
|
|
275 |
/* in the parent */ |
/* in the parent */ |
276 |
if (last_input != 0) { |
if (last_input != -1) { |
277 |
if (close (last_input) < 0) |
if (close (last_input) < 0) |
278 |
error (FATAL, errno, _("close failed")); |
error (FATAL, errno, _("close failed")); |
279 |
} |
} |
280 |
if (i != p->ncommands - 1) { |
if (i != p->ncommands - 1 || p->readfrom) { |
281 |
if (close (pdes[1]) < 0) |
if (close (pdes[1]) < 0) |
282 |
error (FATAL, errno, _("close failed")); |
error (FATAL, errno, _("close failed")); |
283 |
last_input = pdes[0]; |
last_input = pdes[0]; |
289 |
/* TODO: Perhaps it would be useful to wait on multiple pipelines |
/* TODO: Perhaps it would be useful to wait on multiple pipelines |
290 |
* simultaneously? Then you could do: |
* simultaneously? Then you could do: |
291 |
* |
* |
292 |
|
* p1->readfrom = 1; |
293 |
|
* p2->writeto = 1; |
294 |
|
* p3->writeto = 1; |
295 |
* pipeline_start (p1); |
* pipeline_start (p1); |
296 |
* pipeline_start (p2); |
* pipeline_start (p2); |
297 |
* pipeline_start (p3); |
* pipeline_start (p3); |
301 |
* |
* |
302 |
* ... and have processes exit as their input is closed by other processes |
* ... and have processes exit as their input is closed by other processes |
303 |
* exiting, etc. |
* exiting, etc. |
304 |
|
* |
305 |
|
* This function is kind of broken if multiple pipelines exist, anyway, or |
306 |
|
* even if there are other child processes not created by the pipeline |
307 |
|
* library. Unfortunately, spinning and polling non-blocking waitpid() is |
308 |
|
* going to suck. Perhaps ignoring the second case is OK. |
309 |
*/ |
*/ |
310 |
int pipeline_wait (pipeline *p) |
int pipeline_wait (pipeline *p) |
311 |
{ |
{ |
314 |
|
|
315 |
assert (p->pids); /* pipeline started */ |
assert (p->pids); /* pipeline started */ |
316 |
|
|
317 |
|
if (p->writefd != -1) { |
318 |
|
close (p->writefd); |
319 |
|
p->writefd = -1; |
320 |
|
} |
321 |
|
|
322 |
while (proc_count > 0) { |
while (proc_count > 0) { |
323 |
int i; |
int i; |
324 |
int status; |
int status; |
360 |
} |
} |
361 |
} |
} |
362 |
|
|
363 |
|
if (p->readfd != -1) { |
364 |
|
close (p->readfd); |
365 |
|
p->readfd = -1; |
366 |
|
} |
367 |
|
|
368 |
free (p->pids); |
free (p->pids); |
369 |
p->pids = NULL; |
p->pids = NULL; |
370 |
|
|