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