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