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 static sigset_t rpmsqCaught;
 152
 153 static struct rpmsig_s {
 154     int signum;
 155     rpmsqAction_t handler;
 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 int rpmsqIsCaught(int signum)
 175 {
 176     return sigismember(&rpmsqCaught, signum);
 177 }
 178
 179 #ifdef SA_SIGINFO
 180 void rpmsqAction(int signum,
 181                 void * info, void * context)
 182 #else
 183 void rpmsqAction(int signum)
 184 #endif
 185 {
 186     int save = errno;
 187     rpmsig tbl;
 188
 189     for (tbl = rpmsigTbl; tbl->signum >= 0; tbl++) {
 190         if (tbl->signum != signum)
 191             continue;
 192
 193         (void) sigaddset(&rpmsqCaught, signum);
 194
 195         switch (signum) {
 196         case SIGCHLD:
 197             while (1) {
 198                 rpmsq sq;
 199                 int status = 0;
 200                 pid_t reaped = waitpid(0, &status, WNOHANG);
 201
 202                 /* XXX errno set to ECHILD/EINVAL/EINTR. */
 203                 if (reaped <= 0)
 204                     break;
 205
 206                 /* XXX insque(3)/remque(3) are dequeue, not ring. */
 207                 for (sq = rpmsqQueue->q_forw;
 208                      sq != NULL && sq != rpmsqQueue;
 209                      sq = sq->q_forw)
 210                 {
 211                     int ret;
 212
 213                     if (sq->child != reaped)
 214                         continue;
 215                     sq->reaped = reaped;
 216                     sq->status = status;
 217
 218                     /* Unlock the mutex.  The waiter will then be able to
 219                      * aquire the lock.
 220                      *
 221                      * XXX: jbj, wtd, if this fails?
 222                      */
 223                     ret = pthread_mutex_unlock(&sq->mutex);
 224
 225                     break;
 226                 }
 227             }
 228             break;
 229         default:
 230             break;
 231         }
 232         break;
 233     }
 234     errno = save;
 235 }
 236
 237 int rpmsqEnable(int signum, rpmsqAction_t handler)
 238 {
 239     int tblsignum = (signum >= 0 ? signum : -signum);
 240     struct sigaction sa;
 241     rpmsig tbl;
 242     int ret = -1;
 243
 244     (void) DO_LOCK ();
 245     if (rpmsqQueue->id == NULL)
 246         rpmsqQueue->id = ME();
 247     for (tbl = rpmsigTbl; tbl->signum >= 0; tbl++) {
 248         if (tblsignum != tbl->signum)
 249             continue;
 250
 251         if (signum >= 0) {                      /* Enable. */
 252             if (ADD_REF(tbl) <= 0) {
 253                 (void) sigdelset(&rpmsqCaught, tbl->signum);
 254
 255                 /* XXX Don't set a signal handler if already SIG_IGN */
 256                 (void) sigaction(tbl->signum, NULL, &tbl->oact);
 257                 if (tbl->oact.sa_handler == SIG_IGN)
 258                     continue;
 259
 260                 (void) sigemptyset (&sa.sa_mask);
 261 #ifdef SA_SIGINFO
 262                 sa.sa_flags = SA_SIGINFO;
 263 #else
 264                 sa.sa_flags = 0;
 265 #endif
 266                 sa.sa_sigaction = (void*)(handler != NULL ? handler : tbl->handler);
 267                 if (sigaction(tbl->signum, &sa, &tbl->oact) < 0) {
 268                     SUB_REF(tbl);
 269                     break;
 270                 }
 271                 tbl->active = 1;                /* XXX just in case */
 272                 if (handler != NULL)
 273                     tbl->handler = handler;
 274             }
 275         } else {                                /* Disable. */
 276             if (SUB_REF(tbl) <= 0) {
 277                 if (sigaction(tbl->signum, &tbl->oact, NULL) < 0)
 278                     break;
 279                 tbl->active = 0;                /* XXX just in case */
 280                 tbl->handler = (handler != NULL ? handler : rpmsqAction);
 281             }
 282         }
 283         ret = tbl->active;
 284         break;
 285     }
 286     (void) DO_UNLOCK ();
 287     return ret;
 288 }
 289
 290 pid_t rpmsqFork(rpmsq sq)
 291 {
 292     pid_t pid;
 293     int xx;
 294     int nothreads = 0;   /* XXX: Shouldn't this be a global? */
 295
 296     if (sq->reaper) {
 297         xx = rpmsqInsert(sq, NULL);
 298 #ifdef _RPMSQ_DEBUG
 299 if (_rpmsq_debug)
 300 fprintf(stderr, "    Enable(%p): %p\n", ME(), sq);
 301 #endif
 302         xx = rpmsqEnable(SIGCHLD, NULL);
 303     }
 304
 305     xx = pipe(sq->pipes);
 306
 307     xx = sighold(SIGCHLD);
 308
 309     /*
 310      * Initialize the cond var mutex.   We have to aquire the lock we
 311      * use for the condition before we fork.  Otherwise it is possible for
 312      * the child to exit, we get sigchild and the sig handler to send
 313      * the condition signal before we are waiting on the condition.
 314      */
 315     if (!nothreads) {
 316         if(pthread_mutex_lock(&sq->mutex)) {
 317             /* Yack we did not get the lock, lets just give up */
 318             xx = close(sq->pipes[0]);
 319             xx = close(sq->pipes[1]);
 320             sq->pipes[0] = sq->pipes[1] = -1;
 321             goto out;
 322         }
 323     }
 324
 325     pid = fork();
 326     if (pid < (pid_t) 0) {              /* fork failed.  */
 327         sq->child = (pid_t)-1;
 328         xx = close(sq->pipes[0]);
 329         xx = close(sq->pipes[1]);
 330         sq->pipes[0] = sq->pipes[1] = -1;
 331         goto out;
 332     } else if (pid == (pid_t) 0) {      /* Child. */
 333         int yy;
 334
 335         /* Block to permit parent time to wait. */
 336         xx = close(sq->pipes[1]);
 337         xx = read(sq->pipes[0], &yy, sizeof(yy));
 338         xx = close(sq->pipes[0]);
 339         sq->pipes[0] = sq->pipes[1] = -1;
 340
 341 #ifdef _RPMSQ_DEBUG
 342 if (_rpmsq_debug)
 343 fprintf(stderr, "     Child(%p): %p child %d\n", ME(), sq, getpid());
 344 #endif
 345
 346     } else {                            /* Parent. */
 347
 348         sq->child = pid;
 349
 350 #ifdef _RPMSQ_DEBUG
 351 if (_rpmsq_debug)
 352 fprintf(stderr, "    Parent(%p): %p child %d\n", ME(), sq, sq->child);
 353 #endif
 354
 355     }
 356
 357 out:
 358     xx = sigrelse(SIGCHLD);
 359     return sq->child;
 360 }
 361
 362 /**
 363  * Wait for child process to be reaped, and unregister SIGCHLD handler.
 364  * @todo Rewrite to use waitpid on helper thread.
 365  * @param sq            scriptlet queue element
 366  * @return              0 on success
 367  */
 368 static int rpmsqWaitUnregister(rpmsq sq)
 369 {
 370     int nothreads = 0;
 371     int ret = 0;
 372     int xx;
 373
 374     /* Protect sq->reaped from handler changes. */
 375     ret = sighold(SIGCHLD);
 376
 377     /* Start the child, linux often runs child before parent. */
 378     if (sq->pipes[0] >= 0)
 379         xx = close(sq->pipes[0]);
 380     if (sq->pipes[1] >= 0)
 381         xx = close(sq->pipes[1]);
 382     sq->pipes[0] = sq->pipes[1] = -1;
 383
 384     /* Put a stopwatch on the time spent waiting to measure performance gain. */
 385     (void) rpmswEnter(&sq->op, -1);
 386
 387     /* Wait for handler to receive SIGCHLD. */
 388     while (ret == 0 && sq->reaped != sq->child) {
 389         if (nothreads)
 390             /* Note that sigpause re-enables SIGCHLD. */
 391             ret = sigpause(SIGCHLD);
 392         else {
 393             xx = sigrelse(SIGCHLD);
 394
 395             /*
 396              * We start before the fork with this mutex locked;
 397              * The only one that unlocks this the signal handler.
 398              * So if we get the lock the child has been reaped.
 399              */
 400             ret = pthread_mutex_lock(&sq->mutex);
 401             xx = sighold(SIGCHLD);
 402         }
 403     }
 404
 405     /* Accumulate stopwatch time spent waiting, potential performance gain. */
 406     sq->ms_scriptlets += rpmswExit(&sq->op, -1)/1000;
 407
 408     xx = sigrelse(SIGCHLD);
 409
 410 #ifdef _RPMSQ_DEBUG
 411 if (_rpmsq_debug)
 412 fprintf(stderr, "      Wake(%p): %p child %d reaper %d ret %d\n", ME(), sq, sq->child, sq->reaper, ret);
 413 #endif
 414
 415     /* Remove processed SIGCHLD item from queue. */
 416     xx = rpmsqRemove(sq);
 417
 418     /* Disable SIGCHLD handler on refcount == 0. */
 419     xx = rpmsqEnable(-SIGCHLD, NULL);
 420 #ifdef _RPMSQ_DEBUG
 421 if (_rpmsq_debug)
 422 fprintf(stderr, "   Disable(%p): %p\n", ME(), sq);
 423 #endif
 424
 425     return ret;
 426 }
 427
 428 pid_t rpmsqWait(rpmsq sq)
 429 {
 430
 431 #ifdef _RPMSQ_DEBUG
 432 if (_rpmsq_debug)
 433 fprintf(stderr, "      Wait(%p): %p child %d reaper %d\n", ME(), sq, sq->child, sq->reaper);
 434 #endif
 435
 436     if (sq->reaper) {
 437         (void) rpmsqWaitUnregister(sq);
 438     } else {
 439         pid_t reaped;
 440         int status;
 441         do {
 442             reaped = waitpid(sq->child, &status, 0);
 443         } while (reaped >= 0 && reaped != sq->child);
 444         sq->reaped = reaped;
 445         sq->status = status;
 446 #ifdef _RPMSQ_DEBUG
 447 if (_rpmsq_debug)
 448 fprintf(stderr, "   Waitpid(%p): %p child %d reaped %d\n", ME(), sq, sq->child, sq->reaped);
 449 #endif
 450     }
 451
 452 #ifdef _RPMSQ_DEBUG
 453 if (_rpmsq_debug)
 454 fprintf(stderr, "      Fini(%p): %p child %d status 0x%x\n", ME(), sq, sq->child, sq->status);
 455 #endif
 456
 457     return sq->reaped;
 458 }
 459
 460 void * rpmsqThread(void * (*start) (void * arg), void * arg)
 461 {
 462     pthread_t pth;
 463     int ret;
 464
 465     ret = pthread_create(&pth, NULL, start, arg);
 466     return (ret == 0 ? (void *)pth : NULL);
 467 }
 468
 469 int rpmsqJoin(void * thread)
 470 {
 471     pthread_t pth = (pthread_t) thread;
 472     if (thread == NULL)
 473         return EINVAL;
 474     return pthread_join(pth, NULL);
 475 }
 476
 477 int rpmsqThreadEqual(void * thread)
 478 {
 479     pthread_t t1 = (pthread_t) thread;
 480     pthread_t t2 = pthread_self();
 481     return pthread_equal(t1, t2);
 482 }
 483
 484 /**
 485  * SIGCHLD cancellation handler.
 486  */
 487 static void
 488 sigchld_cancel (void *arg)
 489 {
 490     pid_t child = *(pid_t *) arg;
 491     pid_t result;
 492
 493     (void) kill(child, SIGKILL);
 494
 495     do {
 496         result = waitpid(child, NULL, 0);
 497     } while (result == (pid_t)-1 && errno == EINTR);
 498
 499     (void) DO_LOCK ();
 500     if (SUB_REF (rpmsigTbl_sigchld) == 0) {
 501         (void) rpmsqEnable(-SIGQUIT, NULL);
 502         (void) rpmsqEnable(-SIGINT, NULL);
 503     }
 504     (void) DO_UNLOCK ();
 505 }
 506
 507 /**
 508  * Execute a command, returning its status.
 509  */
 510 int
 511 rpmsqExecve (const char ** argv)
 512 {
 513     int oldtype;
 514     int status = -1;
 515     pid_t pid = 0;
 516     pid_t result;
 517     sigset_t newMask, oldMask;
 518
 519 #ifndef PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP
 520         INIT_LOCK ();
 521 #endif
 522
 523     (void) DO_LOCK ();
 524     if (ADD_REF (rpmsigTbl_sigchld) == 0) {
 525         if (rpmsqEnable(SIGINT, NULL) < 0) {
 526             SUB_REF (rpmsigTbl_sigchld);
 527             goto out;
 528         }
 529         if (rpmsqEnable(SIGQUIT, NULL) < 0) {
 530             SUB_REF (rpmsigTbl_sigchld);
 531             goto out_restore_sigint;
 532         }
 533     }
 534     (void) DO_UNLOCK ();
 535
 536     (void) sigemptyset (&newMask);
 537     (void) sigaddset (&newMask, SIGCHLD);
 538     if (sigprocmask (SIG_BLOCK, &newMask, &oldMask) < 0) {
 539         (void) DO_LOCK ();
 540         if (SUB_REF (rpmsigTbl_sigchld) == 0)
 541             goto out_restore_sigquit_and_sigint;
 542         goto out;
 543     }
 544
 545     CLEANUP_HANDLER(sigchld_cancel, &pid, &oldtype);
 546
 547     pid = fork ();
 548     if (pid < (pid_t) 0) {              /* fork failed.  */
 549         goto out;
 550     } else if (pid == (pid_t) 0) {      /* Child. */
 551
 552         /* Restore the signals.  */
 553         (void) sigaction (SIGINT, &rpmsigTbl_sigint->oact, NULL);
 554         (void) sigaction (SIGQUIT, &rpmsigTbl_sigquit->oact, NULL);
 555         (void) sigprocmask (SIG_SETMASK, &oldMask, NULL);
 556
 557         /* Reset rpmsigTbl lock and refcnt. */
 558         INIT_LOCK ();
 559
 560         (void) execve (argv[0], (char *const *) argv, environ);
 561         _exit (127);
 562     } else {                            /* Parent. */
 563         do {
 564             result = waitpid(pid, &status, 0);
 565         } while (result == (pid_t)-1 && errno == EINTR);
 566         if (result != pid)
 567             status = -1;
 568     }
 569
 570     CLEANUP_RESET(0, oldtype);
 571
 572     (void) DO_LOCK ();
 573     if ((SUB_REF (rpmsigTbl_sigchld) == 0 &&
 574         (rpmsqEnable(-SIGINT, NULL) < 0 || rpmsqEnable (-SIGQUIT, NULL) < 0))
 575       || sigprocmask (SIG_SETMASK, &oldMask, NULL) != 0)
 576     {
 577         status = -1;
 578     }
 579     goto out;
 580
 581 out_restore_sigquit_and_sigint:
 582     (void) rpmsqEnable(-SIGQUIT, NULL);
 583 out_restore_sigint:
 584     (void) rpmsqEnable(-SIGINT, NULL);
 585 out:
 586     (void) DO_UNLOCK ();
 587     return status;
 588 }