rpmio/rpmsq.c

   1 /** \ingroup rpmio
   2  * \file rpmio/rpmsq.c
   3  */
   4
   5 #include "system.h"
   6
   7 #include <signal.h>
   8 #include <sys/signal.h>
   9 #include <sys/wait.h>
  10 #include <search.h>
  11
  12 #if defined(HAVE_PTHREAD_H)
  13
  14 #include <pthread.h>
  15
  16 /* XXX suggested in bugzilla #159024 */
  17 #if PTHREAD_MUTEX_DEFAULT != PTHREAD_MUTEX_NORMAL
  18   #error RPM expects PTHREAD_MUTEX_DEFAULT == PTHREAD_MUTEX_NORMAL
  19 #endif
  20
  21 #ifndef PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP
  22 static pthread_mutex_t rpmsigTbl_lock = PTHREAD_MUTEX_INITIALIZER;
  23 #else
  24 static pthread_mutex_t rpmsigTbl_lock = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
  25 #endif
  26
  27 #define DO_LOCK()       pthread_mutex_lock(&rpmsigTbl_lock);
  28 #define DO_UNLOCK()     pthread_mutex_unlock(&rpmsigTbl_lock);
  29 #define INIT_LOCK()     \
  30     {   pthread_mutexattr_t attr; \
  31         (void) pthread_mutexattr_init(&attr); \
  32         (void) pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); \
  33         (void) pthread_mutex_init (&rpmsigTbl_lock, &attr); \
  34         (void) pthread_mutexattr_destroy(&attr); \
  35         rpmsigTbl_sigchld->active = 0; \
  36     }
  37 #define ADD_REF(__tbl)  (__tbl)->active++
  38 #define SUB_REF(__tbl)  --(__tbl)->active
  39 #define CLEANUP_HANDLER(__handler, __arg, __oldtypeptr) \
  40     (void) pthread_setcanceltype (PTHREAD_CANCEL_ASYNCHRONOUS, (__oldtypeptr));\
  41         pthread_cleanup_push((__handler), (__arg));
  42 #define CLEANUP_RESET(__execute, __oldtype) \
  43     pthread_cleanup_pop(__execute); \
  44     (void) pthread_setcanceltype ((__oldtype), &(__oldtype));
  45
  46 #define SAME_THREAD(_a, _b)     pthread_equal(((pthread_t)_a), ((pthread_t)_b))
  47
  48 #define ME()    ((void *)pthread_self())
  49
  50 #else
  51
  52 #define DO_LOCK()
  53 #define DO_UNLOCK()
  54 #define INIT_LOCK()
  55 #define ADD_REF(__tbl)  (0)
  56 #define SUB_REF(__tbl)  (0)
  57 #define CLEANUP_HANDLER(__handler, __arg, __oldtypeptr)
  58 #define CLEANUP_RESET(__execute, __oldtype)
  59
  60 #define SAME_THREAD(_a, _b)     (42)
  61
  62 #define ME()    (((void *)getpid()))
  63
  64 #endif  /* HAVE_PTHREAD_H */
  65
  66 #define _RPMSQ_INTERNAL
  67 #include <rpm/rpmsq.h>
  68
  69 #include "debug.h"
  70
  71 #define _RPMSQ_DEBUG    0
  72 int _rpmsq_debug = _RPMSQ_DEBUG;
  73
  74 static struct rpmsqElem rpmsqRock;
  75
  76 static rpmsq rpmsqQueue = &rpmsqRock;
  77
  78 /** \ingroup rpmsq
  79  * Insert node into from queue.
  80  * @param elem          node to link
  81  * @param prev          previous node from queue
  82  * @return              0 on success
  83  */
  84 static int rpmsqInsert(void * elem, void * prev)
  85 {
  86     rpmsq sq = (rpmsq) elem;
  87     int ret = -1;
  88
  89     if (sq != NULL) {
  90 #ifdef _RPMSQ_DEBUG
  91 if (_rpmsq_debug)
  92 fprintf(stderr, "    Insert(%p): %p\n", ME(), sq);
  93 #endif
  94         ret = sighold(SIGCHLD);
  95         if (ret == 0) {
  96             sq->child = 0;
  97             sq->reaped = 0;
  98             sq->status = 0;
  99             sq->reaper = 1;
 100             sq->pipes[0] = sq->pipes[1] = -1;
 101
 102             sq->id = ME();
 103             ret = pthread_mutex_init(&sq->mutex, NULL);
 104             insque(elem, (prev != NULL ? prev : rpmsqQueue));
 105             ret = sigrelse(SIGCHLD);
 106         }
 107     }
 108     return ret;
 109 }
 110
 111 /** \ingroup rpmsq
 112  * Remove node from queue.
 113  * @param elem          node to link
 114  * @return              0 on success
 115  */
 116 static int rpmsqRemove(void * elem)
 117 {
 118     rpmsq sq = (rpmsq) elem;
 119     int ret = -1;
 120
 121     if (elem != NULL) {
 122
 123 #ifdef _RPMSQ_DEBUG
 124 if (_rpmsq_debug)
 125 fprintf(stderr, "    Remove(%p): %p\n", ME(), sq);
 126 #endif
 127         ret = sighold (SIGCHLD);
 128         if (ret == 0) {
 129             remque(elem);
 130
 131             /* Unlock the mutex and then destroy it */
 132             if((ret = pthread_mutex_unlock(&sq->mutex)) == 0)
 133                 ret = pthread_mutex_destroy(&sq->mutex);
 134
 135             sq->id = NULL;
 136             if (sq->pipes[1])   ret = close(sq->pipes[1]);
 137             if (sq->pipes[0])   ret = close(sq->pipes[0]);
 138             sq->pipes[0] = sq->pipes[1] = -1;
 139 #ifdef  NOTYET  /* rpmpsmWait debugging message needs */
 140             sq->reaper = 1;
 141             sq->status = 0;
 142             sq->reaped = 0;
 143             sq->child = 0;
 144 #endif
 145             ret = sigrelse(SIGCHLD);
 146         }
 147     }
 148     return ret;
 149 }
 150
 151 sigset_t rpmsqCaught;
 152
 153 static struct rpmsig_s {
 154     int signum;
 155     void (*handler) (int signum, void * info, void * context);
 156     int active;
 157     struct sigaction oact;
 158 } rpmsigTbl[] = {
 159     { SIGINT,   rpmsqAction },
 160 #define rpmsigTbl_sigint        (&rpmsigTbl[0])
 161     { SIGQUIT,  rpmsqAction },
 162 #define rpmsigTbl_sigquit       (&rpmsigTbl[1])
 163     { SIGCHLD,  rpmsqAction },
 164 #define rpmsigTbl_sigchld       (&rpmsigTbl[2])
 165     { SIGHUP,   rpmsqAction },
 166 #define rpmsigTbl_sighup        (&rpmsigTbl[3])
 167     { SIGTERM,  rpmsqAction },
 168 #define rpmsigTbl_sigterm       (&rpmsigTbl[4])
 169     { SIGPIPE,  rpmsqAction },
 170 #define rpmsigTbl_sigpipe       (&rpmsigTbl[5])
 171     { -1,       NULL },
 172 };
 173
 174 void rpmsqAction(int signum,
 175                 void * info, void * context)
 176 {
 177     int save = errno;
 178     rpmsig tbl;
 179
 180     for (tbl = rpmsigTbl; tbl->signum >= 0; tbl++) {
 181         if (tbl->signum != signum)
 182             continue;
 183
 184         (void) sigaddset(&rpmsqCaught, signum);
 185
 186         switch (signum) {
 187         case SIGCHLD:
 188             while (1) {
 189                 rpmsq sq;
 190                 int status = 0;
 191                 pid_t reaped = waitpid(0, &status, WNOHANG);
 192
 193                 /* XXX errno set to ECHILD/EINVAL/EINTR. */
 194                 if (reaped <= 0)
 195                     break;
 196
 197                 /* XXX insque(3)/remque(3) are dequeue, not ring. */
 198                 for (sq = rpmsqQueue->q_forw;
 199                      sq != NULL && sq != rpmsqQueue;
 200                      sq = sq->q_forw)
 201                 {
 202                     int ret;
 203
 204                     if (sq->child != reaped)
 205                         continue;
 206                     sq->reaped = reaped;
 207                     sq->status = status;
 208
 209                     /* Unlock the mutex.  The waiter will then be able to
 210                      * aquire the lock.
 211                      *
 212                      * XXX: jbj, wtd, if this fails?
 213                      */
 214                     ret = pthread_mutex_unlock(&sq->mutex);
 215
 216                     break;
 217                 }
 218             }
 219             break;
 220         default:
 221             break;
 222         }
 223         break;
 224     }
 225     errno = save;
 226 }
 227
 228 int rpmsqEnable(int signum, rpmsqAction_t handler)
 229 {
 230     int tblsignum = (signum >= 0 ? signum : -signum);
 231     struct sigaction sa;
 232     rpmsig tbl;
 233     int ret = -1;
 234
 235     (void) DO_LOCK ();
 236     if (rpmsqQueue->id == NULL)
 237         rpmsqQueue->id = ME();
 238     for (tbl = rpmsigTbl; tbl->signum >= 0; tbl++) {
 239         if (tblsignum != tbl->signum)
 240             continue;
 241
 242         if (signum >= 0) {                      /* Enable. */
 243             if (ADD_REF(tbl) <= 0) {
 244                 (void) sigdelset(&rpmsqCaught, tbl->signum);
 245
 246                 /* XXX Don't set a signal handler if already SIG_IGN */
 247                 (void) sigaction(tbl->signum, NULL, &tbl->oact);
 248                 if (tbl->oact.sa_handler == SIG_IGN)
 249                     continue;
 250
 251                 (void) sigemptyset (&sa.sa_mask);
 252                 sa.sa_flags = SA_SIGINFO;
 253                 sa.sa_sigaction = (void*)(handler != NULL ? handler : tbl->handler);
 254                 if (sigaction(tbl->signum, &sa, &tbl->oact) < 0) {
 255                     SUB_REF(tbl);
 256                     break;
 257                 }
 258                 tbl->active = 1;                /* XXX just in case */
 259                 if (handler != NULL)
 260                     tbl->handler = handler;
 261             }
 262         } else {                                /* Disable. */
 263             if (SUB_REF(tbl) <= 0) {
 264                 if (sigaction(tbl->signum, &tbl->oact, NULL) < 0)
 265                     break;
 266                 tbl->active = 0;                /* XXX just in case */
 267                 tbl->handler = (handler != NULL ? handler : rpmsqAction);
 268             }
 269         }
 270         ret = tbl->active;
 271         break;
 272     }
 273     (void) DO_UNLOCK ();
 274     return ret;
 275 }
 276
 277 pid_t rpmsqFork(rpmsq sq)
 278 {
 279     pid_t pid;
 280     int xx;
 281     int nothreads = 0;   /* XXX: Shouldn't this be a global? */
 282
 283     if (sq->reaper) {
 284         xx = rpmsqInsert(sq, NULL);
 285 #ifdef _RPMSQ_DEBUG
 286 if (_rpmsq_debug)
 287 fprintf(stderr, "    Enable(%p): %p\n", ME(), sq);
 288 #endif
 289         xx = rpmsqEnable(SIGCHLD, NULL);
 290     }
 291
 292     xx = pipe(sq->pipes);
 293
 294     xx = sighold(SIGCHLD);
 295
 296     /*
 297      * Initialize the cond var mutex.   We have to aquire the lock we
 298      * use for the condition before we fork.  Otherwise it is possible for
 299      * the child to exit, we get sigchild and the sig handler to send
 300      * the condition signal before we are waiting on the condition.
 301      */
 302     if (!nothreads) {
 303         if(pthread_mutex_lock(&sq->mutex)) {
 304             /* Yack we did not get the lock, lets just give up */
 305             xx = close(sq->pipes[0]);
 306             xx = close(sq->pipes[1]);
 307             sq->pipes[0] = sq->pipes[1] = -1;
 308             goto out;
 309         }
 310     }
 311
 312     pid = fork();
 313     if (pid < (pid_t) 0) {              /* fork failed.  */
 314         sq->child = (pid_t)-1;
 315         xx = close(sq->pipes[0]);
 316         xx = close(sq->pipes[1]);
 317         sq->pipes[0] = sq->pipes[1] = -1;
 318         goto out;
 319     } else if (pid == (pid_t) 0) {      /* Child. */
 320         int yy;
 321
 322         /* Block to permit parent time to wait. */
 323         xx = close(sq->pipes[1]);
 324         xx = read(sq->pipes[0], &yy, sizeof(yy));
 325         xx = close(sq->pipes[0]);
 326         sq->pipes[0] = sq->pipes[1] = -1;
 327
 328 #ifdef _RPMSQ_DEBUG
 329 if (_rpmsq_debug)
 330 fprintf(stderr, "     Child(%p): %p child %d\n", ME(), sq, getpid());
 331 #endif
 332
 333     } else {                            /* Parent. */
 334
 335         sq->child = pid;
 336
 337 #ifdef _RPMSQ_DEBUG
 338 if (_rpmsq_debug)
 339 fprintf(stderr, "    Parent(%p): %p child %d\n", ME(), sq, sq->child);
 340 #endif
 341
 342     }
 343
 344 out:
 345     xx = sigrelse(SIGCHLD);
 346     return sq->child;
 347 }
 348
 349 /**
 350  * Wait for child process to be reaped, and unregister SIGCHLD handler.
 351  * @todo Rewrite to use waitpid on helper thread.
 352  * @param sq            scriptlet queue element
 353  * @return              0 on success
 354  */
 355 static int rpmsqWaitUnregister(rpmsq sq)
 356 {
 357     int nothreads = 0;
 358     int ret = 0;
 359     int xx;
 360
 361     /* Protect sq->reaped from handler changes. */
 362     ret = sighold(SIGCHLD);
 363
 364     /* Start the child, linux often runs child before parent. */
 365     if (sq->pipes[0] >= 0)
 366         xx = close(sq->pipes[0]);
 367     if (sq->pipes[1] >= 0)
 368         xx = close(sq->pipes[1]);
 369     sq->pipes[0] = sq->pipes[1] = -1;
 370
 371     /* Put a stopwatch on the time spent waiting to measure performance gain. */
 372     (void) rpmswEnter(&sq->op, -1);
 373
 374     /* Wait for handler to receive SIGCHLD. */
 375     while (ret == 0 && sq->reaped != sq->child) {
 376         if (nothreads)
 377             /* Note that sigpause re-enables SIGCHLD. */
 378             ret = sigpause(SIGCHLD);
 379         else {
 380             xx = sigrelse(SIGCHLD);
 381
 382             /*
 383              * We start before the fork with this mutex locked;
 384              * The only one that unlocks this the signal handler.
 385              * So if we get the lock the child has been reaped.
 386              */
 387             ret = pthread_mutex_lock(&sq->mutex);
 388             xx = sighold(SIGCHLD);
 389         }
 390     }
 391
 392     /* Accumulate stopwatch time spent waiting, potential performance gain. */
 393     sq->ms_scriptlets += rpmswExit(&sq->op, -1)/1000;
 394
 395     xx = sigrelse(SIGCHLD);
 396
 397 #ifdef _RPMSQ_DEBUG
 398 if (_rpmsq_debug)
 399 fprintf(stderr, "      Wake(%p): %p child %d reaper %d ret %d\n", ME(), sq, sq->child, sq->reaper, ret);
 400 #endif
 401
 402     /* Remove processed SIGCHLD item from queue. */
 403     xx = rpmsqRemove(sq);
 404
 405     /* Disable SIGCHLD handler on refcount == 0. */
 406     xx = rpmsqEnable(-SIGCHLD, NULL);
 407 #ifdef _RPMSQ_DEBUG
 408 if (_rpmsq_debug)
 409 fprintf(stderr, "   Disable(%p): %p\n", ME(), sq);
 410 #endif
 411
 412     return ret;
 413 }
 414
 415 pid_t rpmsqWait(rpmsq sq)
 416 {
 417
 418 #ifdef _RPMSQ_DEBUG
 419 if (_rpmsq_debug)
 420 fprintf(stderr, "      Wait(%p): %p child %d reaper %d\n", ME(), sq, sq->child, sq->reaper);
 421 #endif
 422
 423     if (sq->reaper) {
 424         (void) rpmsqWaitUnregister(sq);
 425     } else {
 426         pid_t reaped;
 427         int status;
 428         do {
 429             reaped = waitpid(sq->child, &status, 0);
 430         } while (reaped >= 0 && reaped != sq->child);
 431         sq->reaped = reaped;
 432         sq->status = status;
 433 #ifdef _RPMSQ_DEBUG
 434 if (_rpmsq_debug)
 435 fprintf(stderr, "   Waitpid(%p): %p child %d reaped %d\n", ME(), sq, sq->child, sq->reaped);
 436 #endif
 437     }
 438
 439 #ifdef _RPMSQ_DEBUG
 440 if (_rpmsq_debug)
 441 fprintf(stderr, "      Fini(%p): %p child %d status 0x%x\n", ME(), sq, sq->child, sq->status);
 442 #endif
 443
 444     return sq->reaped;
 445 }
 446
 447 void * rpmsqThread(void * (*start) (void * arg), void * arg)
 448 {
 449     pthread_t pth;
 450     int ret;
 451
 452     ret = pthread_create(&pth, NULL, start, arg);
 453     return (ret == 0 ? (void *)pth : NULL);
 454 }
 455
 456 int rpmsqJoin(void * thread)
 457 {
 458     pthread_t pth = (pthread_t) thread;
 459     if (thread == NULL)
 460         return EINVAL;
 461     return pthread_join(pth, NULL);
 462 }
 463
 464 int rpmsqThreadEqual(void * thread)
 465 {
 466     pthread_t t1 = (pthread_t) thread;
 467     pthread_t t2 = pthread_self();
 468     return pthread_equal(t1, t2);
 469 }
 470
 471 /**
 472  * SIGCHLD cancellation handler.
 473  */
 474 static void
 475 sigchld_cancel (void *arg)
 476 {
 477     pid_t child = *(pid_t *) arg;
 478     pid_t result;
 479
 480     (void) kill(child, SIGKILL);
 481
 482     do {
 483         result = waitpid(child, NULL, 0);
 484     } while (result == (pid_t)-1 && errno == EINTR);
 485
 486     (void) DO_LOCK ();
 487     if (SUB_REF (rpmsigTbl_sigchld) == 0) {
 488         (void) rpmsqEnable(-SIGQUIT, NULL);
 489         (void) rpmsqEnable(-SIGINT, NULL);
 490     }
 491     (void) DO_UNLOCK ();
 492 }
 493
 494 /**
 495  * Execute a command, returning its status.
 496  */
 497 int
 498 rpmsqExecve (const char ** argv)
 499 {
 500     int oldtype;
 501     int status = -1;
 502     pid_t pid = 0;
 503     pid_t result;
 504     sigset_t newMask, oldMask;
 505
 506 #ifndef PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP
 507         INIT_LOCK ();
 508 #endif
 509
 510     (void) DO_LOCK ();
 511     if (ADD_REF (rpmsigTbl_sigchld) == 0) {
 512         if (rpmsqEnable(SIGINT, NULL) < 0) {
 513             SUB_REF (rpmsigTbl_sigchld);
 514             goto out;
 515         }
 516         if (rpmsqEnable(SIGQUIT, NULL) < 0) {
 517             SUB_REF (rpmsigTbl_sigchld);
 518             goto out_restore_sigint;
 519         }
 520     }
 521     (void) DO_UNLOCK ();
 522
 523     (void) sigemptyset (&newMask);
 524     (void) sigaddset (&newMask, SIGCHLD);
 525     if (sigprocmask (SIG_BLOCK, &newMask, &oldMask) < 0) {
 526         (void) DO_LOCK ();
 527         if (SUB_REF (rpmsigTbl_sigchld) == 0)
 528             goto out_restore_sigquit_and_sigint;
 529         goto out;
 530     }
 531
 532     CLEANUP_HANDLER(sigchld_cancel, &pid, &oldtype);
 533
 534     pid = fork ();
 535     if (pid < (pid_t) 0) {              /* fork failed.  */
 536         goto out;
 537     } else if (pid == (pid_t) 0) {      /* Child. */
 538
 539         /* Restore the signals.  */
 540         (void) sigaction (SIGINT, &rpmsigTbl_sigint->oact, NULL);
 541         (void) sigaction (SIGQUIT, &rpmsigTbl_sigquit->oact, NULL);
 542         (void) sigprocmask (SIG_SETMASK, &oldMask, NULL);
 543
 544         /* Reset rpmsigTbl lock and refcnt. */
 545         INIT_LOCK ();
 546
 547         (void) execve (argv[0], (char *const *) argv, environ);
 548         _exit (127);
 549     } else {                            /* Parent. */
 550         do {
 551             result = waitpid(pid, &status, 0);
 552         } while (result == (pid_t)-1 && errno == EINTR);
 553         if (result != pid)
 554             status = -1;
 555     }
 556
 557     CLEANUP_RESET(0, oldtype);
 558
 559     (void) DO_LOCK ();
 560     if ((SUB_REF (rpmsigTbl_sigchld) == 0 &&
 561         (rpmsqEnable(-SIGINT, NULL) < 0 || rpmsqEnable (-SIGQUIT, NULL) < 0))
 562       || sigprocmask (SIG_SETMASK, &oldMask, NULL) != 0)
 563     {
 564         status = -1;
 565     }
 566     goto out;
 567
 568 out_restore_sigquit_and_sigint:
 569     (void) rpmsqEnable(-SIGQUIT, NULL);
 570 out_restore_sigint:
 571     (void) rpmsqEnable(-SIGINT, NULL);
 572 out:
 573     (void) DO_UNLOCK ();
 574     return status;
 575 }