git.itanic.dy.fi Git - rrdd/blob - process.c

   1 #include <unistd.h>
   2 #include <fcntl.h>
   3 #include <sys/select.h>
   4 #include <sys/epoll.h>
   5 #include <stdio.h>
   6 #include <sys/wait.h>
   7 #include <sys/signalfd.h>
   8 #include <sys/resource.h>
   9
  10 #include "process.h"
  11 #include "debug.h"
  12
  13 static int child_count;
  14 static int parent_count;
  15 static int job_request_fd[2];
  16 static int job_get_permission_fd[2];
  17 static int epoll_fd;
  18 static unsigned int max_jobs;
  19 static unsigned int job_count;
  20 static unsigned int jobs_pending;
  21 static unsigned int max_jobs_pending;
  22
  23 struct work_struct {
  24         const char *name;
  25         int (*work_fn)(void *);
  26         void *arg;
  27         struct work_struct *next;
  28 };
  29
  30 struct work_queue {
  31         struct work_struct *work;
  32         int length;
  33         char *name;
  34         struct mutex lock;
  35 };
  36
  37 struct work_queue work_queues[WORK_PRIORITIES_NUM] = {
  38         {
  39                 .name = "high priority",
  40                 .lock = {
  41                         .name = "high_prio_queue",
  42                         .lock = PTHREAD_MUTEX_INITIALIZER,
  43                 },
  44         },
  45         {
  46                 .name = "low priority",
  47                 .lock = {
  48                         .name = "low_prio_queue",
  49                         .lock = PTHREAD_MUTEX_INITIALIZER,
  50                 },
  51         },
  52 };
  53
  54 struct mutex work_pending_mutex = {
  55         .name = "work_pending",
  56         .lock = PTHREAD_MUTEX_INITIALIZER,
  57 };
  58 pthread_cond_t work_pending_cond = PTHREAD_COND_INITIALIZER;
  59
  60 static int run_work_on_queue(struct work_queue *queue)
  61 {
  62         struct work_struct *work;
  63
  64         mutex_lock(&queue->lock);
  65
  66         if (!queue->work) {
  67                 pr_info("No  work to run on queue %s\n", queue->name);
  68                 mutex_unlock(&queue->lock);
  69                 return 0;
  70         }
  71
  72         /* Take next work */
  73         work = queue->work;
  74         queue->work = work->next;
  75         queue->length--;
  76
  77         /*
  78          * If queue is not empty, try waking up more workers. It is
  79          * possible that when work were queued, the first worker did
  80          * not wake up soon enough and
  81          */
  82         if (queue->length > 0)
  83                 pthread_cond_signal(&work_pending_cond);
  84
  85         mutex_unlock(&queue->lock);
  86
  87         pr_info("Executing work %s from queue %s, %d still pending\n",
  88                 work->name, queue->name, queue->length);
  89
  90         work->work_fn(work->arg);
  91         pr_info("Work %s done\n", work->name);
  92         free(work);
  93
  94         return 1;
  95 }
  96
  97 static void *worker_thread(void *arg)
  98 {
  99         int ret;
 100
 101         char name[16];
 102
 103         snprintf(name, sizeof(name), "worker%ld", (long)arg);
 104         pthread_setname_np(pthread_self(), name);
 105
 106         while (1) {
 107                 while (1) {
 108                         int prio, work_done = 0;
 109
 110                         /*
 111                          * Execute as many works from the queues as
 112                          * there are, starting from highest priority
 113                          * queue
 114                          */
 115                         for (prio = 0; prio < WORK_PRIORITIES_NUM; prio++) {
 116                                 work_done =
 117                                         run_work_on_queue(&work_queues[prio]);
 118                                 if (work_done)
 119                                         break;
 120                         }
 121
 122                         if (!work_done)
 123                                 break;
 124                 }
 125
 126                 pr_info("Worker going to sleep\n");
 127                 ret = pthread_cond_wait(&work_pending_cond,
 128                                         &work_pending_mutex.lock);
 129                 if (ret < 0)
 130                         pr_err("Error: %m\n");
 131
 132                 mutex_lock_acquired(&work_pending_mutex);
 133
 134                 mutex_unlock(&work_pending_mutex);
 135
 136         }
 137
 138         return NULL;
 139 }
 140
 141 int queue_work(unsigned int priority, char *name,
 142         int (work_fn)(void *arg), void *arg)
 143 {
 144         struct work_queue *queue;
 145         struct work_struct *work, *last_work;
 146
 147         if (priority >= WORK_PRIORITIES_NUM) {
 148                 pr_err("Invalid priority: %d\n", priority);
 149                 return -EINVAL;
 150         }
 151
 152         work = calloc(sizeof(*work), 1);
 153
 154         work->name = name;
 155         work->work_fn = work_fn;
 156         work->arg = arg;
 157
 158         queue = &work_queues[priority];
 159
 160         /* Insert new work at the end of the work queue */
 161         mutex_lock(&queue->lock);
 162
 163         last_work = queue->work;
 164         while (last_work && last_work->next)
 165                 last_work = last_work->next;
 166
 167         if (!last_work)
 168                 queue->work = work;
 169         else
 170                 last_work->next = work;
 171
 172         pr_info("Inserted work %s in queue %s, with %d pending items\n",
 173                 work->name, queue->name, queue->length);
 174         queue->length++;
 175         mutex_unlock(&queue->lock);
 176
 177         pthread_cond_signal(&work_pending_cond);
 178
 179         return 0;
 180 }
 181
 182 int get_child_count(void)
 183 {
 184         return child_count;
 185 }
 186
 187 int get_parent_count(void)
 188 {
 189         return parent_count;
 190 }
 191
 192 static int handle_signals(struct event_handler *h)
 193 {
 194         struct signalfd_siginfo siginfo;
 195         int ret;
 196
 197         ret = read(h->fd, &siginfo, sizeof(siginfo));
 198         if (ret < sizeof(siginfo)) {
 199                 pr_err("Expected %zd from read, got %d: %m\n",
 200                         sizeof(siginfo), ret);
 201                 return -1;
 202         }
 203
 204         if (siginfo.ssi_signo != SIGCHLD) {
 205                 pr_err("Unexpected signal %d, ignoring\n", siginfo.ssi_signo);
 206                 return -1;
 207         }
 208
 209         harvest_zombies(siginfo.ssi_pid);
 210
 211         return 0;
 212 }
 213
 214 static int grant_new_job(void)
 215 {
 216         int ret;
 217         char byte = 0;
 218
 219         job_count++;
 220         pr_info("Granting new job. %d jobs currently and %d pending\n",
 221                 job_count, jobs_pending);
 222
 223         ret = write(job_get_permission_fd[1], &byte, 1);
 224         if (ret != 1) {
 225                 pr_err("Failed to write 1 byte: %m\n");
 226                 return -1;
 227         }
 228
 229         return 0;
 230 }
 231
 232 static int deny_job(void)
 233 {
 234         int ret;
 235         char byte = -1;
 236
 237         pr_info("Denying new job. %d jobs currently and %d pending, "
 238                 "limit of pending jobs is %d\n",
 239                 job_count, jobs_pending, max_jobs_pending);
 240
 241         ret = write(job_get_permission_fd[1], &byte, 1);
 242         if (ret != 1) {
 243                 pr_err("Failed to write 1 byte: %m\n");
 244                 return -1;
 245         }
 246
 247         return 0;
 248 }
 249
 250 static int handle_job_request(struct event_handler *h)
 251 {
 252         int ret, pid;
 253
 254         ret = read(job_request_fd[0], &pid, sizeof(pid));
 255         if (ret < 0) {
 256                 pr_err("Failed to read: %m\n");
 257                 return -1;
 258         }
 259
 260         if (ret == 0) {
 261                 pr_info("Read zero bytes\n");
 262                 return 0;
 263         }
 264
 265         if (pid > 0) {
 266                 if (job_count >= max_jobs) {
 267                         if (jobs_pending < max_jobs_pending)
 268                                 jobs_pending++;
 269                         else
 270                                 deny_job();
 271                 } else {
 272                         ret = grant_new_job();
 273                         return 0;
 274                 }
 275         } else if (pid < 0) {
 276                 if (job_count > max_jobs)
 277                         pr_err("BUG: Job %u jobs exceeded limit %u\n",
 278                                 job_count, max_jobs);
 279
 280                 pr_info("Job %d finished\n", -pid);
 281                 job_count--;
 282                 if (jobs_pending) {
 283                         jobs_pending--;
 284                         ret = grant_new_job();
 285                         return 0;
 286                 }
 287         }
 288
 289         return 0;
 290 }
 291
 292 struct event_handler signal_handler = {
 293         .handle_event = handle_signals,
 294         .events = EPOLLIN,
 295         .name = "signal",
 296 };
 297
 298 struct event_handler job_request_handler = {
 299         .handle_event = handle_job_request,
 300         .events = EPOLLIN,
 301         .name = "job_request",
 302 };
 303
 304 /*
 305  * Initialize the jobcontrol.
 306  *
 307  * Create the pipes that are used to grant children execution
 308  * permissions. If max_jobs is zero, count the number of CPUs from
 309  * /proc/cpuinfo and use that.
 310  */
 311 int init_jobcontrol(int max_jobs_requested)
 312 {
 313         FILE *file;
 314         int ret;
 315         sigset_t sigmask;
 316         char buf[256];
 317         char match[8];
 318         pthread_t *thread;
 319         int i;
 320
 321         if (pipe2(job_request_fd, O_NONBLOCK | O_CLOEXEC)) {
 322                 pr_err("Failed to create pipe: %m\n");
 323                 return -1;
 324         }
 325
 326         if (pipe2(job_get_permission_fd, O_CLOEXEC)) {
 327                 pr_err("Failed to create pipe: %m\n");
 328                 return -1;
 329         }
 330
 331         epoll_fd = epoll_create(1);
 332         if (epoll_fd == -1) {
 333                 pr_err("Failed to epoll_create(): %m\n");
 334                 return -1;
 335         }
 336
 337         job_request_handler.fd = job_request_fd[0];
 338         register_event_handler(&job_request_handler);
 339
 340         sigemptyset(&sigmask);
 341         sigaddset(&sigmask, SIGCHLD);
 342
 343         /* Block SIGCHLD so that it becomes readable via signalfd */
 344         ret = sigprocmask(SIG_BLOCK, &sigmask, NULL);
 345         if (ret < 0) {
 346                 pr_err("Failed to sigprocmask: %m\n");
 347         }
 348
 349         signal_handler.fd = signalfd(-1, &sigmask, SFD_CLOEXEC);
 350         if (job_request_handler.fd < 0) {
 351                 pr_err("Failed to create signal_fd: %m\n");
 352                 return -1;
 353         }
 354
 355         register_event_handler(&signal_handler);
 356
 357         if (max_jobs_requested > 0) {
 358                 max_jobs = max_jobs_requested;
 359                 goto no_count_cpus;
 360         }
 361         max_jobs++;
 362
 363         file = fopen("/proc/cpuinfo", "ro");
 364         if (!file) {
 365                 pr_err("Failed to open /proc/cpuinfo: %m\n");
 366                 goto open_fail;
 367         }
 368
 369         /*
 370          * The CPU count algorithm simply reads the first 8 bytes from
 371          * the /proc/cpuinfo and then expects that line to be there as
 372          * many times as there are CPUs.
 373          */
 374         ret = fread(match, 1, sizeof(match), file);
 375         if (ret < sizeof(match)) {
 376                 pr_err("read %d bytes when expecting %zd %m\n",
 377                         ret, sizeof(match));
 378                 goto read_fail;
 379         }
 380
 381         while(fgets(buf, sizeof(buf), file)) {
 382                 if (!strncmp(buf, match, sizeof(match)))
 383                         max_jobs++;
 384         }
 385
 386 open_fail:
 387 read_fail:
 388         fclose(file);
 389
 390 no_count_cpus:
 391         pr_info("Set maximum number of parallel jobs to %d\n", max_jobs);
 392
 393         max_jobs_pending = max_jobs * 10 + 25;
 394         pr_info("Set maximum number of pending jobs to %d\n", max_jobs_pending);
 395
 396         /* Create worker threads */
 397         thread = calloc(sizeof(*thread), max_jobs);
 398         for (i = 0; i < max_jobs; i++)
 399                 pthread_create(&thread[i], NULL, worker_thread, (void *)i);
 400
 401         /*
 402          * Magic sleep. There are too many fork() calls at the moment
 403          * so we must ensure our threads don't print anything out
 404          * while a fork() is executed. Otherwise the child will
 405          * inherit glibc internal locks while they are locked, and
 406          * function calls such as printf will deadlock.
 407          */
 408         sleep(1);
 409
 410         return 0;
 411 }
 412
 413 int poll_job_requests(int timeout)
 414 {
 415         struct epoll_event event;
 416         struct event_handler *job_handler;
 417         int ret;
 418
 419         /* Convert positive seconds to milliseconds */
 420         timeout = timeout > 0 ? 1000 * timeout : timeout;
 421
 422         ret = epoll_wait(epoll_fd, &event, 1, timeout);
 423
 424         if (ret == -1) {
 425                 if (errno != EINTR) {
 426                         pr_err("epoll_wait: %m\n");
 427                         return -1;
 428                 }
 429
 430                 /*
 431                  * If epoll_wait() was interrupted, better start
 432                  * everything again from the beginning
 433                  */
 434                 return 0;
 435         }
 436
 437         if (ret == 0) {
 438                 pr_info("Timed out\n");
 439                 goto out;
 440         }
 441
 442         job_handler = event.data.ptr;
 443
 444         if (!job_handler || !job_handler->handle_event) {
 445                 pr_err("Corrupted event handler for fd %d\n",
 446                         event.data.fd);
 447                 goto out;
 448         }
 449
 450         pr_debug("Running handler %s to handle events from fd %d\n",
 451                 job_handler->name, job_handler->fd);
 452         job_handler->handle_event(job_handler);
 453
 454 out:
 455         pr_info("Jobs active: %u, pending: %u\n", job_count, jobs_pending);
 456         return ret;
 457 }
 458
 459 /*
 460  * Per process flag indicating whether this child has requested fork
 461  * limiting. If it has, it must also tell the master parent when it
 462  * has died so that the parent can give next pending job permission to
 463  * go.
 464  */
 465 static int is_limited_fork;
 466
 467 int do_fork(void)
 468 {
 469         int child;
 470         child = fork();
 471         if (child < 0) {
 472                 pr_err("fork() failed: %m\n");
 473                 return -1;
 474         }
 475
 476         if (child) {
 477                 child_count++;
 478                 pr_debug("Fork %d, child %d\n", child_count, child);
 479                 return child;
 480         }
 481
 482         /*
 483          * Also do not notify the master parent the death of this
 484          * child. Only childs that have been created with
 485          * do_fork_limited() can have this flag set.
 486          */
 487         is_limited_fork = 0;
 488
 489         /*
 490          * Close unused ends of the job control pipes. Only the parent
 491          * which controls the jobs may have the write end open of the
 492          * job_get_permission_fd and the read end of the
 493          * job_request_fd. Failing to close the pipe ends properly
 494          * will cause the childs to wait forever for the run
 495          * permission in case parent dies prematurely.
 496          *
 497          * Note! The file descriptor must be closed once and only
 498          * once. They are marked to -1 to make it impossible for
 499          * subseqent do_fork() calls from closing them again (in which
 500          * case some other file descriptor might already be reserved
 501          * for the same number) and prevent accidentally closing some
 502          * innocent file descriptors that are still in use.
 503          */
 504         if (job_get_permission_fd[1] >= 0) {
 505                 close(job_get_permission_fd[1]);
 506                 job_get_permission_fd[1] = -1;
 507         }
 508         if (job_request_fd[0] >= 0) {
 509                 close(job_request_fd[0]);
 510                 job_request_fd[0] = -1;
 511         }
 512
 513         /* reset child's child count */
 514         child_count = 0;
 515         parent_count++;
 516         return 0;
 517 }
 518
 519 static int request_fork(int request)
 520 {
 521         int pid = getpid();
 522
 523         pid = request > 0 ? pid : -pid;
 524
 525         return write(job_request_fd[1], &pid, sizeof(pid));
 526 }
 527
 528 static void limited_fork_exit_handler(void)
 529 {
 530         if (is_limited_fork)
 531                 request_fork(-1);
 532 }
 533
 534 /*
 535  * Like do_fork(), but allow the child continue only after the global
 536  * job count is low enough.
 537  *
 538  * We allow the parent to continue other more important activities but
 539  * child respects the limit of global active processes.
 540  */
 541 int do_fork_limited(void)
 542 {
 543         int child, ret;
 544         char byte;
 545
 546         child = do_fork();
 547         if (child)
 548                 return child;
 549
 550         /* Remember to notify the parent when we are done */
 551         atexit(limited_fork_exit_handler);
 552         is_limited_fork = 1;
 553
 554         pr_debug("Requesting permission to go\n");
 555
 556         /* Signal the parent that we are here, waiting to go */
 557         request_fork(1);
 558
 559         /*
 560          * The parent will tell us when we can continue. If there were
 561          * multiple children waiting for their turn to run only one
 562          * will be reading the content byte from the pipe and getting
 563          * the permission to run.
 564          */
 565         ret = read(job_get_permission_fd[0], &byte, sizeof(byte));
 566         if (ret == 0)
 567                 pr_err("Error requesting run, did the parent die?\n");
 568
 569         if (ret < 0)
 570                 pr_err("Job control request failure: %m\n");
 571
 572         if (byte < 0) {
 573                 pr_info("Did not get permission to execute. Terminating\n");
 574
 575                 /*
 576                  * Avoid running exit handler, that would tell the
 577                  * parent we died normally and decrement the job
 578                  * counters.
 579                  */
 580                 raise(SIGKILL);
 581         }
 582
 583         pr_debug("Continuing\n");
 584         return child;
 585 }
 586
 587 int harvest_zombies(int pid)
 588 {
 589         int status;
 590         struct rusage rusage;
 591         char *status_str = NULL;
 592         int code = 0;
 593
 594         if (child_count == 0)
 595                 return 0;
 596
 597         if (pid)
 598                 pr_debug("Waiting on pid %d, children left: %d\n", pid,
 599                         child_count);
 600
 601         do {
 602                 pid = wait4(pid, &status, 0, &rusage);
 603                 if (pid < 0) {
 604                         pr_err("Error on waitid(): %m\n");
 605                         return 0;
 606                 }
 607                 /* Wait until the child has become a zombie */
 608         } while (!WIFEXITED(status) && !WIFSIGNALED(status));
 609
 610         child_count--;
 611         if (WIFEXITED(status)) {
 612                 status_str = "exited with status";
 613                 code = WEXITSTATUS(status);
 614         } else if (WIFSIGNALED(status)) {
 615                 status_str = "killed by signal";
 616                 code = WTERMSIG(status);
 617         }
 618         pr_debug("pid %d: %s %d. Children left: %d\n", pid,
 619                 status_str, code, child_count);
 620         pr_debug("pid %d: User time: %ld.%03lds, System %ld.%03lds\n", pid,
 621                 (long)rusage.ru_utime.tv_sec, rusage.ru_utime.tv_usec / 1000,
 622                 (long)rusage.ru_stime.tv_sec, rusage.ru_stime.tv_usec / 1000);
 623
 624         return 1;
 625 }
 626
 627 /*
 628  * Runs a command cmd with params argv, connects stdin and stdout to
 629  * readfd and writefd
 630  *
 631  * Returns the pid of the executed process
 632  */
 633 int run_piped(const char *cmd, char *const argv[],
 634               int *stdinfd, int *stdoutfd, int *stderrfd)
 635 {
 636         int ifd[2], ofd[2], efd[2], pid;
 637
 638         pr_info("Running command %s\n", cmd);
 639
 640         if (stdinfd && pipe(ifd)) {
 641                 pr_err("pipe() failed: %m\n");
 642                 return -1;
 643         }
 644
 645         if (stdoutfd && pipe(ofd)) {
 646                 pr_err("pipe() failed: %m\n");
 647                 return -1;
 648         }
 649
 650         if (stderrfd && pipe(efd)) {
 651                 pr_err("pipe() failed: %m\n");
 652                 return -1;
 653         }
 654
 655         pid = do_fork();
 656         if (pid) { /* Parent side */
 657                 if (stdinfd) {
 658                         close(ifd[0]);
 659                         *stdinfd = ifd[0];
 660                 }
 661
 662                 if (stdoutfd) {
 663                         close(ofd[1]);
 664                         *stdoutfd = ofd[0];
 665                 }
 666
 667                 if (stderrfd) {
 668                         close(efd[1]);
 669                         *stderrfd = efd[0];
 670                 }
 671
 672                 return pid;
 673         }
 674
 675         if (stdinfd) {
 676                 close(ifd[1]);
 677                 dup2(ifd[0], STDIN_FILENO);
 678         }
 679
 680         if (stdoutfd) {
 681                 close(ofd[0]);
 682                 dup2(ofd[1], STDOUT_FILENO);
 683         }
 684
 685         if (stderrfd) {
 686                 close(efd[0]);
 687                 dup2(efd[1], STDERR_FILENO);
 688         }
 689
 690         /* Now we have redirected standard streams to parent process */
 691         execvp(cmd, argv);
 692         pr_err("Failed to execv command %s: %m\n", cmd);
 693         exit(1);
 694
 695         return 0;
 696 }
 697
 698 /*
 699  * Runs a command cmd with params argv, connects stdin and stdout to
 700  * readfd and writefd
 701  *
 702  * Returns the pid of the executed process
 703  */
 704 int run_piped_stream(const char *cmd, char *const argv[],
 705                      FILE **stdinf, FILE **stdoutf, FILE **stderrf)
 706 {
 707         int ifd, ofd, efd, pid;
 708         int *i, *o, *e;
 709
 710         if (stdinf)
 711                 i = &ifd;
 712         else
 713                 i = 0;
 714         if (stdoutf)
 715                 o = &ofd;
 716         else
 717                 o = 0;
 718         if (stderrf)
 719                 e = &efd;
 720         else
 721                 e = 0;
 722
 723         pid = run_piped(cmd, argv, i, o, e);
 724
 725         if (stdinf) {
 726                 *stdinf = fdopen(ifd, "r");
 727                 if (*stdinf == NULL) {
 728                         pr_err("Error opening file stream for fd %d: %m\n",
 729                                ifd);
 730                         return -1;
 731                 }
 732         }
 733
 734         if (stdoutf) {
 735                 *stdoutf = fdopen(ofd, "r");
 736                 if (*stdoutf == NULL) {
 737                         pr_err("Error opening file stream for fd %d: %m\n",
 738                                ofd);
 739                         return -1;
 740                 }
 741         }
 742
 743         if (stderrf) {
 744                 *stderrf = fdopen(efd, "r");
 745                 if (*stderrf == NULL) {
 746                         pr_err("Error opening file stream for fd %d: %m\n",
 747                                efd);
 748                         return -1;
 749                 }
 750         }
 751
 752         return pid;
 753 }
 754
 755 /*
 756  * Forks a child and executes a command to run on parallel
 757  */
 758
 759 #define max(a,b) (a) < (b) ? (b) : (a)
 760 #define BUF_SIZE (128*1024)
 761 int run(const char *cmd, char *const argv[])
 762 {
 763         int child, error;
 764         int ofd, efd;
 765         fd_set rfds;
 766         int maxfd;
 767         int eof = 0;
 768
 769         if ((child = do_fork()))
 770             return child;
 771
 772         child = run_piped(cmd, argv, NULL, &ofd, &efd);
 773
 774         FD_ZERO(&rfds);
 775         FD_SET(ofd, &rfds);
 776         FD_SET(efd, &rfds);
 777
 778         while (!eof) {
 779                 char *sptr , *eptr;
 780                 char rbuf[BUF_SIZE];
 781                 int bytes;
 782                 int is_stderr = 0;
 783
 784                 maxfd = max(ofd, efd);
 785                 error = select(maxfd, &rfds, NULL, NULL, NULL);
 786
 787                 if (error < 0) {
 788                         pr_err("Error with select: %m\n");
 789                         break;
 790                 }
 791
 792                 if (FD_ISSET(ofd, &rfds)) {
 793                         bytes = read(ofd, rbuf, BUF_SIZE);
 794                         goto print;
 795                 }
 796
 797                 if (FD_ISSET(efd, &rfds)) {
 798                         is_stderr = 1;
 799                         bytes = read(efd, rbuf, BUF_SIZE);
 800                         goto print;
 801                 }
 802
 803                 pr_err("select() returned unknown fd\n");
 804                 break;
 805
 806 print:
 807                 if (bytes < 0) {
 808                         pr_err("read() failed: %m\n");
 809                         break;
 810                 }
 811
 812                 /*
 813                  * Workaround: When a process had die and it has only
 814                  * written to stderr, select() doesn't indicate that
 815                  * there might be something to read in stderr fd. To
 816                  * work around this issue, we try to read stderr just
 817                  * in case in order to ensure everything gets read.
 818                  */
 819                 if (bytes == 0) {
 820                         bytes = read(efd, rbuf, BUF_SIZE);
 821                         is_stderr = 1;
 822                         eof = 1;
 823                 }
 824
 825                 sptr = eptr = rbuf;
 826                 while (bytes--) {
 827                         if (*eptr == '\n') {
 828                                 *eptr = 0;
 829                                 if (is_stderr)
 830                                         pr_err("%s: stderr: %s\n",
 831                                                 cmd, sptr);
 832                                 else
 833                                         pr_info("%s: stdout: %s\n",
 834                                                 cmd, sptr);
 835                                 sptr = eptr;
 836                         }
 837                         eptr++;
 838                 }
 839         }
 840
 841         close(ofd);
 842         close(efd);
 843
 844         harvest_zombies(child);
 845
 846         exit(1);
 847         return 0;
 848 }
 849
 850 int register_event_handler(struct event_handler *handler)
 851 {
 852         struct epoll_event ev;
 853         int ret;
 854
 855         if (handler->fd <= 0) {
 856                 pr_err("Invalid file descriptor of %d\n", handler->fd);
 857                 return -1;
 858         }
 859
 860         if (!handler->handle_event) {
 861                 pr_err("Handler callback missing\n");
 862                 return -1;
 863         }
 864
 865         pr_info("Registering handler for %s, fd %d\n",
 866                 handler->name, handler->fd);
 867
 868         ev.data.fd = handler->fd;
 869         ev.data.ptr = handler;
 870         ev.events = handler->events;
 871         ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, handler->fd, &ev);
 872         if (ret) {
 873                 pr_err("Failed to add epoll_fd: %m\n");
 874                 return -1;
 875         }
 876
 877         return 0;
 878 }
 879
 880 void _mutex_lock_acquired(struct mutex *lock, char *file, int line)
 881 {
 882         lock->line = line;
 883         lock->file = file;
 884 }
 885
 886 int _mutex_lock(struct mutex *lock, char *file, int line)
 887 {
 888         int ret = 0;
 889
 890         if (!pthread_mutex_trylock(&lock->lock))
 891                 goto out_lock;
 892
 893         pr_info("Lock contention on lock %s on %s:%d\n",
 894                 lock->name, lock->file, lock->line);
 895
 896         ret = pthread_mutex_lock(&lock->lock);
 897         if (ret)
 898                 pr_err("Acquirin lock %s failed: %m, acquired %s:%d\n",
 899                         lock->name, lock->file, lock->line);
 900
 901 out_lock:
 902         _mutex_lock_acquired(lock, file, line);
 903         return ret;
 904 }
 905
 906 int _mutex_unlock(struct mutex *lock)
 907 {
 908         lock->line = 0;
 909         lock->file = NULL;
 910         pthread_mutex_unlock(&lock->lock);
 911
 912         return 0;
 913 }