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