4 #include <sys/select.h>
8 #include <sys/signalfd.h>
9 #include <sys/resource.h>
14 static int child_count;
15 static int parent_count;
16 static int job_request_fd[2];
17 static int job_get_permission_fd[2];
20 static unsigned int max_jobs;
21 static unsigned int job_count;
22 static unsigned int jobs_pending;
24 int get_child_count(void)
29 int get_parent_count(void)
34 static int handle_signals(void)
36 struct signalfd_siginfo siginfo;
39 ret = read(signal_fd, &siginfo, sizeof(siginfo));
40 if (ret < sizeof(siginfo)) {
41 pr_err("Expected %zd from read, got %d: %m\n",
42 sizeof(siginfo), ret);
46 if (siginfo.ssi_signo != SIGCHLD) {
47 pr_err("Unexpected signal %d, ignoring\n", siginfo.ssi_signo);
51 harvest_zombies(siginfo.ssi_pid);
57 * Initialize the jobcontrol.
59 * Create the pipes that are used to grant children execution
60 * permissions. If max_jobs is zero, count the number of CPUs from
61 * /proc/cpuinfo and use that.
63 int init_jobcontrol(int max_jobs_requested)
65 struct epoll_event ev;
72 if (pipe2(job_request_fd, O_NONBLOCK | O_CLOEXEC)) {
73 pr_err("Failed to create pipe: %m\n");
77 if (pipe2(job_get_permission_fd, O_CLOEXEC)) {
78 pr_err("Failed to create pipe: %m\n");
82 epoll_fd = epoll_create(1);
84 pr_err("Failed to epoll_create(): %m\n");
89 ev.data.fd = job_request_fd[0];
90 ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, job_request_fd[0], &ev);
92 pr_err("Failed to add epoll_fd: %m\n");
96 sigemptyset(&sigmask);
97 sigaddset(&sigmask, SIGCHLD);
99 /* Block SIGCHLD so that it becomes readable via signalfd */
100 ret = sigprocmask(SIG_BLOCK, &sigmask, NULL);
102 pr_err("Failed to sigprocmask: %m\n");
105 signal_fd = signalfd(-1, &sigmask, SFD_CLOEXEC);
107 pr_err("Failed to create signal_fd: %m\n");
111 ev.data.fd = signal_fd;
112 ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, signal_fd, &ev);
114 pr_err("Failed to add epoll_fd: %m\n");
119 if (max_jobs_requested > 0) {
120 max_jobs = max_jobs_requested;
125 file = fopen("/proc/cpuinfo", "ro");
127 pr_err("Failed to open /proc/cpuinfo: %m\n");
132 * The CPU count algorithm simply reads the first 8 bytes from
133 * the /proc/cpuinfo and then expects that line to be there as
134 * many times as there are CPUs.
136 ret = fread(match, 1, sizeof(match), file);
137 if (ret < sizeof(match)) {
138 pr_err("read %d bytes when expecting %zd %m\n",
143 while(fgets(buf, sizeof(buf), file)) {
144 if (!strncmp(buf, match, sizeof(match)))
153 pr_info("Set maximum number of parallel jobs to %d\n", max_jobs);
158 static int grant_new_job(void)
164 pr_info("Granting new job. %d jobs currently and %d pending\n",
165 job_count, jobs_pending);
167 ret = write(job_get_permission_fd[1], &byte, 1);
169 pr_err("Failed to write 1 byte: %m\n");
176 static int handle_job_request(void)
180 ret = read(job_request_fd[0], &pid, sizeof(pid));
182 pr_err("Failed to read: %m\n");
187 pr_info("Read zero bytes\n");
192 if (job_count >= max_jobs) {
195 ret = grant_new_job();
198 } else if (pid < 0) {
199 if (job_count > max_jobs)
200 pr_err("BUG: Job %u jobs exceeded limit %u\n",
201 job_count, max_jobs);
203 pr_info("Job %d finished\n", -pid);
207 ret = grant_new_job();
215 int poll_job_requests(int timeout)
217 struct epoll_event event;
220 /* Convert positive seconds to milliseconds */
221 timeout = timeout > 0 ? 1000 * timeout : timeout;
223 ret = epoll_wait(epoll_fd, &event, 1, timeout);
226 if (errno != EINTR) {
227 pr_err("epoll_wait: %m\n");
232 * If epoll_wait() was interrupted, better start
233 * everything again from the beginning
239 pr_info("Timed out\n");
243 if (event.data.fd == job_request_fd[0])
244 handle_job_request();
245 else if (event.data.fd == signal_fd)
248 pr_err("Unknown fd: %d\n", event.data.fd);
251 pr_info("Jobs active: %u, pending: %u\n", job_count, jobs_pending);
256 * Per process flag indicating whether this child has requested fork
257 * limiting. If it has, it must also tell the master parent when it
258 * has died so that the parent can give next pending job permission to
261 static int is_limited_fork;
268 pr_err("fork() failed: %m\n");
274 pr_info("Fork %d, child %d\n", child_count, child);
279 * Also do not notify the master parent the death of this
280 * child. Only childs that have been created with
281 * do_fork_limited() can have this flag set.
286 * Close unused ends of the job control pipes. Only the parent
287 * which controls the jobs may have the write end open of the
288 * job_get_permission_fd and the read end of the
289 * job_request_fd. Failing to close the pipe ends properly
290 * will cause the childs to wait forever for the run
291 * permission in case parent dies prematurely.
293 * Note! The file descriptor must be closed once and only
294 * once. They are marked to -1 to make it impossible for
295 * subseqent do_fork() calls from closing them again (in which
296 * case some other file descriptor might already be reserved
297 * for the same number) and prevent accidentally closing some
298 * innocent file descriptors that are still in use.
300 if (job_get_permission_fd[1] >= 0) {
301 close(job_get_permission_fd[1]);
302 job_get_permission_fd[1] = -1;
304 if (job_request_fd[0] >= 0) {
305 close(job_request_fd[0]);
306 job_request_fd[0] = -1;
309 /* reset child's child count */
315 static int request_fork(int request)
319 pid = request > 0 ? pid : -pid;
321 return write(job_request_fd[1], &pid, sizeof(pid));
324 static void limited_fork_exit_handler(void)
331 * Like do_fork(), but allow the child continue only after the global
332 * job count is low enough.
334 * We allow the parent to continue other more important activities but
335 * child respects the limit of global active processes.
337 int do_fork_limited(void)
346 /* Remember to notify the parent when we are done */
347 atexit(limited_fork_exit_handler);
350 pr_info("Requesting permission to go\n");
352 /* Signal the parent that we are here, waiting to go */
356 * The parent will tell us when we can continue. If there were
357 * multiple children waiting for their turn to run only one
358 * will be reading the content byte from the pipe and getting
359 * the permission to run.
361 ret = read(job_get_permission_fd[0], &byte, sizeof(byte));
363 pr_err("Error requesting run, did the parent die?\n");
366 pr_err("Job control request failure: %m\n");
368 pr_info("Continuing\n");
372 int harvest_zombies(int pid)
375 struct rusage rusage;
376 char *status_str = NULL;
379 if (child_count == 0)
383 pr_info("Waiting on pid %d, children left: %d\n", pid,
387 pid = wait4(pid, &status, 0, &rusage);
389 pr_err("Error on waitid(): %m\n");
392 /* Wait until the child has become a zombie */
393 } while (!WIFEXITED(status) && !WIFSIGNALED(status));
396 if (WIFEXITED(status)) {
397 status_str = "exited with status";
398 code = WEXITSTATUS(status);
399 } else if (WIFSIGNALED(status)) {
400 status_str = "killed by signal";
401 code = WTERMSIG(status);
403 pr_info("pid %d: %s %d. Children left: %d\n", pid,
404 status_str, code, child_count);
405 pr_info("pid %d: User time: %ld.%03lds, System %ld.%03lds\n", pid,
406 (long)rusage.ru_utime.tv_sec, rusage.ru_utime.tv_usec / 1000,
407 (long)rusage.ru_stime.tv_sec, rusage.ru_stime.tv_usec / 1000);
413 * Runs a command cmd with params argv, connects stdin and stdout to
416 * Returns the pid of the executed process
418 int run_piped(const char *cmd, char *const argv[],
419 int *stdinfd, int *stdoutfd, int *stderrfd)
421 int ifd[2], ofd[2], efd[2], pid;
423 pr_info("Running command %s\n", cmd);
425 if (stdinfd && pipe(ifd)) {
426 pr_err("pipe() failed: %m\n");
430 if (stdoutfd && pipe(ofd)) {
431 pr_err("pipe() failed: %m\n");
435 if (stderrfd && pipe(efd)) {
436 pr_err("pipe() failed: %m\n");
441 if (pid) { /* Parent side */
462 dup2(ifd[0], STDIN_FILENO);
467 dup2(ofd[1], STDOUT_FILENO);
472 dup2(efd[1], STDERR_FILENO);
475 /* Now we have redirected standard streams to parent process */
477 pr_err("Failed to execv command %s: %m\n", cmd);
484 * Runs a command cmd with params argv, connects stdin and stdout to
487 * Returns the pid of the executed process
489 int run_piped_stream(const char *cmd, char *const argv[],
490 FILE **stdinf, FILE **stdoutf, FILE **stderrf)
492 int ifd, ofd, efd, pid;
508 pid = run_piped(cmd, argv, i, o, e);
511 *stdinf = fdopen(ifd, "r");
512 if (*stdinf == NULL) {
513 pr_err("Error opening file stream for fd %d: %m\n",
520 *stdoutf = fdopen(ofd, "r");
521 if (*stdoutf == NULL) {
522 pr_err("Error opening file stream for fd %d: %m\n",
529 *stderrf = fdopen(efd, "r");
530 if (*stderrf == NULL) {
531 pr_err("Error opening file stream for fd %d: %m\n",
541 * Forks a child and executes a command to run on parallel
544 #define max(a,b) (a) < (b) ? (b) : (a)
545 #define BUF_SIZE (128*1024)
546 int run(const char *cmd, char *const argv[])
554 if ((child = do_fork()))
557 child = run_piped(cmd, argv, NULL, &ofd, &efd);
569 maxfd = max(ofd, efd);
570 error = select(maxfd, &rfds, NULL, NULL, NULL);
573 pr_err("Error with select: %m\n");
577 if (FD_ISSET(ofd, &rfds)) {
578 bytes = read(ofd, rbuf, BUF_SIZE);
582 if (FD_ISSET(efd, &rfds)) {
584 bytes = read(efd, rbuf, BUF_SIZE);
588 pr_err("select() returned unknown fd\n");
593 pr_err("read() failed: %m\n");
598 * Workaround: When a process had die and it has only
599 * written to stderr, select() doesn't indicate that
600 * there might be something to read in stderr fd. To
601 * work around this issue, we try to read stderr just
602 * in case in order to ensure everything gets read.
605 bytes = read(efd, rbuf, BUF_SIZE);
615 pr_err("%s: stderr: %s\n",
618 pr_info("%s: stdout: %s\n",
629 harvest_zombies(child);