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         signal_handler.fd = signalfd(-1, &sigmask, SFD_CLOEXEC);
 344         if (job_request_handler.fd < 0) {
 345                 pr_err("Failed to create signal_fd: %m\n");
 346                 return -1;
 347         }
 348
 349         register_event_handler(&signal_handler);
 350
 351         if (max_jobs_requested > 0) {
 352                 max_jobs = max_jobs_requested;
 353                 goto no_count_cpus;
 354         }
 355         max_jobs++;
 356
 357         file = fopen("/proc/cpuinfo", "ro");
 358         if (!file) {
 359                 pr_err("Failed to open /proc/cpuinfo: %m\n");
 360                 goto open_fail;
 361         }
 362
 363         /*
 364          * The CPU count algorithm simply reads the first 8 bytes from
 365          * the /proc/cpuinfo and then expects that line to be there as
 366          * many times as there are CPUs.
 367          */
 368         ret = fread(match, 1, sizeof(match), file);
 369         if (ret < sizeof(match)) {
 370                 pr_err("read %d bytes when expecting %zd %m\n",
 371                         ret, sizeof(match));
 372                 goto read_fail;
 373         }
 374
 375         while(fgets(buf, sizeof(buf), file)) {
 376                 if (!strncmp(buf, match, sizeof(match)))
 377                         max_jobs++;
 378         }
 379
 380 open_fail:
 381 read_fail:
 382         fclose(file);
 383
 384 no_count_cpus:
 385         pr_info("Set maximum number of parallel jobs to %d\n", max_jobs);
 386
 387         max_jobs_pending = max_jobs * 10 + 25;
 388         pr_info("Set maximum number of pending jobs to %d\n", max_jobs_pending);
 389
 390         /* Create worker threads */
 391         thread = calloc(sizeof(*thread), max_jobs);
 392         for (i = 0; i < max_jobs; i++)
 393                 pthread_create(&thread[i], NULL, worker_thread, (void *)i);
 394
 395         return 0;
 396 }
 397
 398 int poll_job_requests(int timeout)
 399 {
 400         struct epoll_event event;
 401         struct event_handler *job_handler;
 402         int ret;
 403
 404         /* Convert positive seconds to milliseconds */
 405         timeout = timeout > 0 ? 1000 * timeout : timeout;
 406
 407         ret = epoll_wait(epoll_fd, &event, 1, timeout);
 408
 409         if (ret == -1) {
 410                 if (errno != EINTR) {
 411                         pr_err("epoll_wait: %m\n");
 412                         return -1;
 413                 }
 414
 415                 /*
 416                  * If epoll_wait() was interrupted, better start
 417                  * everything again from the beginning
 418                  */
 419                 return 0;
 420         }
 421
 422         if (ret == 0) {
 423                 pr_info("Timed out\n");
 424                 goto out;
 425         }
 426
 427         job_handler = event.data.ptr;
 428
 429         if (!job_handler || !job_handler->handle_event) {
 430                 pr_err("Corrupted event handler for fd %d\n",
 431                         event.data.fd);
 432                 goto out;
 433         }
 434
 435         pr_debug("Running handler %s to handle events from fd %d\n",
 436                 job_handler->name, job_handler->fd);
 437         job_handler->handle_event(job_handler);
 438
 439 out:
 440         pr_info("Jobs active: %u, pending: %u\n", job_count, jobs_pending);
 441         return ret;
 442 }
 443
 444 /*
 445  * Per process flag indicating whether this child has requested fork
 446  * limiting. If it has, it must also tell the master parent when it
 447  * has died so that the parent can give next pending job permission to
 448  * go.
 449  */
 450 static int is_limited_fork;
 451
 452 int do_fork(void)
 453 {
 454         int child;
 455         child = fork();
 456         if (child < 0) {
 457                 pr_err("fork() failed: %m\n");
 458                 return -1;
 459         }
 460
 461         if (child) {
 462                 child_count++;
 463                 pr_debug("Fork %d, child %d\n", child_count, child);
 464                 return child;
 465         }
 466
 467         /*
 468          * Also do not notify the master parent the death of this
 469          * child. Only childs that have been created with
 470          * do_fork_limited() can have this flag set.
 471          */
 472         is_limited_fork = 0;
 473
 474         /*
 475          * Close unused ends of the job control pipes. Only the parent
 476          * which controls the jobs may have the write end open of the
 477          * job_get_permission_fd and the read end of the
 478          * job_request_fd. Failing to close the pipe ends properly
 479          * will cause the childs to wait forever for the run
 480          * permission in case parent dies prematurely.
 481          *
 482          * Note! The file descriptor must be closed once and only
 483          * once. They are marked to -1 to make it impossible for
 484          * subseqent do_fork() calls from closing them again (in which
 485          * case some other file descriptor might already be reserved
 486          * for the same number) and prevent accidentally closing some
 487          * innocent file descriptors that are still in use.
 488          */
 489         if (job_get_permission_fd[1] >= 0) {
 490                 close(job_get_permission_fd[1]);
 491                 job_get_permission_fd[1] = -1;
 492         }
 493         if (job_request_fd[0] >= 0) {
 494                 close(job_request_fd[0]);
 495                 job_request_fd[0] = -1;
 496         }
 497
 498         /* reset child's child count */
 499         child_count = 0;
 500         parent_count++;
 501         return 0;
 502 }
 503
 504 static int request_fork(int request)
 505 {
 506         int pid = getpid();
 507
 508         pid = request > 0 ? pid : -pid;
 509
 510         return write(job_request_fd[1], &pid, sizeof(pid));
 511 }
 512
 513 static void limited_fork_exit_handler(void)
 514 {
 515         if (is_limited_fork)
 516                 request_fork(-1);
 517 }
 518
 519 /*
 520  * Like do_fork(), but allow the child continue only after the global
 521  * job count is low enough.
 522  *
 523  * We allow the parent to continue other more important activities but
 524  * child respects the limit of global active processes.
 525  */
 526 int do_fork_limited(void)
 527 {
 528         int child, ret;
 529         char byte;
 530
 531         child = do_fork();
 532         if (child)
 533                 return child;
 534
 535         /* Remember to notify the parent when we are done */
 536         atexit(limited_fork_exit_handler);
 537         is_limited_fork = 1;
 538
 539         pr_debug("Requesting permission to go\n");
 540
 541         /* Signal the parent that we are here, waiting to go */
 542         request_fork(1);
 543
 544         /*
 545          * The parent will tell us when we can continue. If there were
 546          * multiple children waiting for their turn to run only one
 547          * will be reading the content byte from the pipe and getting
 548          * the permission to run.
 549          */
 550         ret = read(job_get_permission_fd[0], &byte, sizeof(byte));
 551         if (ret == 0)
 552                 pr_err("Error requesting run, did the parent die?\n");
 553
 554         if (ret < 0)
 555                 pr_err("Job control request failure: %m\n");
 556
 557         if (byte < 0) {
 558                 pr_info("Did not get permission to execute. Terminating\n");
 559
 560                 /*
 561                  * Avoid running exit handler, that would tell the
 562                  * parent we died normally and decrement the job
 563                  * counters.
 564                  */
 565                 raise(SIGKILL);
 566         }
 567
 568         pr_debug("Continuing\n");
 569         return child;
 570 }
 571
 572 int harvest_zombies(int pid)
 573 {
 574         int status;
 575         struct rusage rusage;
 576         char *status_str = NULL;
 577         int code = 0;
 578
 579         if (child_count == 0)
 580                 return 0;
 581
 582         if (pid)
 583                 pr_debug("Waiting on pid %d, children left: %d\n", pid,
 584                         child_count);
 585
 586         do {
 587                 pid = wait4(pid, &status, 0, &rusage);
 588                 if (pid < 0) {
 589                         pr_err("Error on waitid(): %m\n");
 590                         return 0;
 591                 }
 592                 /* Wait until the child has become a zombie */
 593         } while (!WIFEXITED(status) && !WIFSIGNALED(status));
 594
 595         child_count--;
 596         if (WIFEXITED(status)) {
 597                 status_str = "exited with status";
 598                 code = WEXITSTATUS(status);
 599         } else if (WIFSIGNALED(status)) {
 600                 status_str = "killed by signal";
 601                 code = WTERMSIG(status);
 602         }
 603         pr_debug("pid %d: %s %d. Children left: %d\n", pid,
 604                 status_str, code, child_count);
 605         pr_debug("pid %d: User time: %ld.%03lds, System %ld.%03lds\n", pid,
 606                 (long)rusage.ru_utime.tv_sec, rusage.ru_utime.tv_usec / 1000,
 607                 (long)rusage.ru_stime.tv_sec, rusage.ru_stime.tv_usec / 1000);
 608
 609         return 1;
 610 }
 611
 612 /*
 613  * Runs a command cmd with params argv, connects stdin and stdout to
 614  * readfd and writefd
 615  *
 616  * Returns the pid of the executed process
 617  */
 618 int run_piped(const char *cmd, char *const argv[],
 619               int *stdinfd, int *stdoutfd, int *stderrfd)
 620 {
 621         int ifd[2], ofd[2], efd[2], pid;
 622
 623         pr_info("Running command %s\n", cmd);
 624
 625         if (stdinfd && pipe(ifd)) {
 626                 pr_err("pipe() failed: %m\n");
 627                 return -1;
 628         }
 629
 630         if (stdoutfd && pipe(ofd)) {
 631                 pr_err("pipe() failed: %m\n");
 632                 return -1;
 633         }
 634
 635         if (stderrfd && pipe(efd)) {
 636                 pr_err("pipe() failed: %m\n");
 637                 return -1;
 638         }
 639
 640         pid = do_fork();
 641         if (pid) { /* Parent side */
 642                 if (stdinfd) {
 643                         close(ifd[0]);
 644                         *stdinfd = ifd[0];
 645                 }
 646
 647                 if (stdoutfd) {
 648                         close(ofd[1]);
 649                         *stdoutfd = ofd[0];
 650                 }
 651
 652                 if (stderrfd) {
 653                         close(efd[1]);
 654                         *stderrfd = efd[0];
 655                 }
 656
 657                 return pid;
 658         }
 659
 660         if (stdinfd) {
 661                 close(ifd[1]);
 662                 dup2(ifd[0], STDIN_FILENO);
 663         }
 664
 665         if (stdoutfd) {
 666                 close(ofd[0]);
 667                 dup2(ofd[1], STDOUT_FILENO);
 668         }
 669
 670         if (stderrfd) {
 671                 close(efd[0]);
 672                 dup2(efd[1], STDERR_FILENO);
 673         }
 674
 675         /* Now we have redirected standard streams to parent process */
 676         execvp(cmd, argv);
 677         pr_err("Failed to execv command %s: %m\n", cmd);
 678         exit(1);
 679
 680         return 0;
 681 }
 682
 683 /*
 684  * Runs a command cmd with params argv, connects stdin and stdout to
 685  * readfd and writefd
 686  *
 687  * Returns the pid of the executed process
 688  */
 689 int run_piped_stream(const char *cmd, char *const argv[],
 690                      FILE **stdinf, FILE **stdoutf, FILE **stderrf)
 691 {
 692         int ifd, ofd, efd, pid;
 693         int *i, *o, *e;
 694
 695         if (stdinf)
 696                 i = &ifd;
 697         else
 698                 i = 0;
 699         if (stdoutf)
 700                 o = &ofd;
 701         else
 702                 o = 0;
 703         if (stderrf)
 704                 e = &efd;
 705         else
 706                 e = 0;
 707
 708         pid = run_piped(cmd, argv, i, o, e);
 709
 710         if (stdinf) {
 711                 *stdinf = fdopen(ifd, "r");
 712                 if (*stdinf == NULL) {
 713                         pr_err("Error opening file stream for fd %d: %m\n",
 714                                ifd);
 715                         return -1;
 716                 }
 717         }
 718
 719         if (stdoutf) {
 720                 *stdoutf = fdopen(ofd, "r");
 721                 if (*stdoutf == NULL) {
 722                         pr_err("Error opening file stream for fd %d: %m\n",
 723                                ofd);
 724                         return -1;
 725                 }
 726         }
 727
 728         if (stderrf) {
 729                 *stderrf = fdopen(efd, "r");
 730                 if (*stderrf == NULL) {
 731                         pr_err("Error opening file stream for fd %d: %m\n",
 732                                efd);
 733                         return -1;
 734                 }
 735         }
 736
 737         return pid;
 738 }
 739
 740 /*
 741  * Forks a child and executes a command to run on parallel
 742  */
 743
 744 #define max(a,b) (a) < (b) ? (b) : (a)
 745 #define BUF_SIZE (128*1024)
 746 int run(const char *cmd, char *const argv[])
 747 {
 748         int child, error;
 749         int ofd, efd;
 750         fd_set rfds;
 751         int maxfd;
 752         int eof = 0;
 753
 754         if ((child = do_fork()))
 755             return child;
 756
 757         child = run_piped(cmd, argv, NULL, &ofd, &efd);
 758
 759         FD_ZERO(&rfds);
 760         FD_SET(ofd, &rfds);
 761         FD_SET(efd, &rfds);
 762
 763         while (!eof) {
 764                 char *sptr , *eptr;
 765                 char rbuf[BUF_SIZE];
 766                 int bytes;
 767                 int is_stderr = 0;
 768
 769                 maxfd = max(ofd, efd);
 770                 error = select(maxfd, &rfds, NULL, NULL, NULL);
 771
 772                 if (error < 0) {
 773                         pr_err("Error with select: %m\n");
 774                         break;
 775                 }
 776
 777                 if (FD_ISSET(ofd, &rfds)) {
 778                         bytes = read(ofd, rbuf, BUF_SIZE);
 779                         goto print;
 780                 }
 781
 782                 if (FD_ISSET(efd, &rfds)) {
 783                         is_stderr = 1;
 784                         bytes = read(efd, rbuf, BUF_SIZE);
 785                         goto print;
 786                 }
 787
 788                 pr_err("select() returned unknown fd\n");
 789                 break;
 790
 791 print:
 792                 if (bytes < 0) {
 793                         pr_err("read() failed: %m\n");
 794                         break;
 795                 }
 796
 797                 /*
 798                  * Workaround: When a process had die and it has only
 799                  * written to stderr, select() doesn't indicate that
 800                  * there might be something to read in stderr fd. To
 801                  * work around this issue, we try to read stderr just
 802                  * in case in order to ensure everything gets read.
 803                  */
 804                 if (bytes == 0) {
 805                         bytes = read(efd, rbuf, BUF_SIZE);
 806                         is_stderr = 1;
 807                         eof = 1;
 808                 }
 809
 810                 sptr = eptr = rbuf;
 811                 while (bytes--) {
 812                         if (*eptr == '\n') {
 813                                 *eptr = 0;
 814                                 if (is_stderr)
 815                                         pr_err("%s: stderr: %s\n",
 816                                                 cmd, sptr);
 817                                 else
 818                                         pr_info("%s: stdout: %s\n",
 819                                                 cmd, sptr);
 820                                 sptr = eptr;
 821                         }
 822                         eptr++;
 823                 }
 824         }
 825
 826         close(ofd);
 827         close(efd);
 828
 829         harvest_zombies(child);
 830
 831         exit(1);
 832         return 0;
 833 }
 834
 835 int register_event_handler(struct event_handler *handler)
 836 {
 837         struct epoll_event ev;
 838         int ret;
 839
 840         if (handler->fd <= 0) {
 841                 pr_err("Invalid file descriptor of %d\n", handler->fd);
 842                 return -1;
 843         }
 844
 845         if (!handler->handle_event) {
 846                 pr_err("Handler callback missing\n");
 847                 return -1;
 848         }
 849
 850         pr_info("Registering handler for %s, fd %d\n",
 851                 handler->name, handler->fd);
 852
 853         ev.data.fd = handler->fd;
 854         ev.data.ptr = handler;
 855         ev.events = handler->events;
 856         ret = epoll_ctl(epoll_fd, EPOLL_CTL_ADD, handler->fd, &ev);
 857         if (ret) {
 858                 pr_err("Failed to add epoll_fd: %m\n");
 859                 return -1;
 860         }
 861
 862         return 0;
 863 }
 864
 865 void _mutex_lock_acquired(struct mutex *lock, char *file, int line)
 866 {
 867         lock->line = line;
 868         lock->file = file;
 869 }
 870
 871 int _mutex_lock(struct mutex *lock, char *file, int line)
 872 {
 873         int ret = 0;
 874
 875         if (!pthread_mutex_trylock(&lock->lock))
 876                 goto out_lock;
 877
 878         pr_info("Lock contention on lock %s on %s:%d\n",
 879                 lock->name, lock->file, lock->line);
 880
 881         ret = pthread_mutex_lock(&lock->lock);
 882         if (ret)
 883                 pr_err("Acquirin lock %s failed: %m, acquired %s:%d\n",
 884                         lock->name, lock->file, lock->line);
 885
 886 out_lock:
 887         _mutex_lock_acquired(lock, file, line);
 888         return ret;
 889 }
 890
 891 int _mutex_unlock(struct mutex *lock)
 892 {
 893         lock->line = 0;
 894         lock->file = NULL;
 895         pthread_mutex_unlock(&lock->lock);
 896
 897         return 0;
 898 }