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 #include <errno.h>
  12 #include <stdio.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 ADD_REF(__tbl)  (__tbl)->active++
  30 #define SUB_REF(__tbl)  --(__tbl)->active
  31
  32 #define ME()    ((void *)pthread_self())
  33
  34 #define _RPMSQ_INTERNAL
  35 #include <rpm/rpmsq.h>
  36
  37 #include "debug.h"
  38
  39 static struct rpmsqElem rpmsqRock;
  40
  41 static rpmsq rpmsqQueue = &rpmsqRock;
  42
  43 /** \ingroup rpmsq
  44  * Insert node into from queue.
  45  * @param elem          node to link
  46  * @param prev          previous node from queue
  47  * @return              0 on success
  48  */
  49 static int rpmsqInsert(void * elem, void * prev)
  50 {
  51     rpmsq sq = (rpmsq) elem;
  52     int ret = -1;
  53
  54     if (sq != NULL) {
  55         ret = sighold(SIGCHLD);
  56         if (ret == 0) {
  57             sq->child = 0;
  58             sq->reaped = 0;
  59             sq->status = 0;
  60             sq->reaper = 1;
  61             sq->pipes[0] = sq->pipes[1] = -1;
  62
  63             sq->id = ME();
  64             ret = pthread_mutex_init(&sq->mutex, NULL);
  65             insque(elem, (prev != NULL ? prev : rpmsqQueue));
  66             ret = sigrelse(SIGCHLD);
  67         }
  68     }
  69     return ret;
  70 }
  71
  72 /** \ingroup rpmsq
  73  * Remove node from queue.
  74  * @param elem          node to link
  75  * @return              0 on success
  76  */
  77 static int rpmsqRemove(void * elem)
  78 {
  79     rpmsq sq = (rpmsq) elem;
  80     int ret = -1;
  81
  82     if (elem != NULL) {
  83         ret = sighold (SIGCHLD);
  84         if (ret == 0) {
  85             remque(elem);
  86
  87             /* Unlock the mutex and then destroy it */
  88             if((ret = pthread_mutex_unlock(&sq->mutex)) == 0)
  89                 ret = pthread_mutex_destroy(&sq->mutex);
  90
  91             sq->id = NULL;
  92             if (sq->pipes[1])   ret = close(sq->pipes[1]);
  93             if (sq->pipes[0])   ret = close(sq->pipes[0]);
  94             sq->pipes[0] = sq->pipes[1] = -1;
  95             ret = sigrelse(SIGCHLD);
  96         }
  97     }
  98     return ret;
  99 }
 100
 101 static sigset_t rpmsqCaught;
 102
 103 static struct rpmsig_s {
 104     int signum;
 105     rpmsqAction_t handler;
 106     int active;
 107     struct sigaction oact;
 108 } rpmsigTbl[] = {
 109     { SIGINT,   rpmsqAction },
 110 #define rpmsigTbl_sigint        (&rpmsigTbl[0])
 111     { SIGQUIT,  rpmsqAction },
 112 #define rpmsigTbl_sigquit       (&rpmsigTbl[1])
 113     { SIGCHLD,  rpmsqAction },
 114 #define rpmsigTbl_sigchld       (&rpmsigTbl[2])
 115     { SIGHUP,   rpmsqAction },
 116 #define rpmsigTbl_sighup        (&rpmsigTbl[3])
 117     { SIGTERM,  rpmsqAction },
 118 #define rpmsigTbl_sigterm       (&rpmsigTbl[4])
 119     { SIGPIPE,  rpmsqAction },
 120 #define rpmsigTbl_sigpipe       (&rpmsigTbl[5])
 121     { -1,       NULL },
 122 };
 123
 124 int rpmsqIsCaught(int signum)
 125 {
 126     return sigismember(&rpmsqCaught, signum);
 127 }
 128
 129 #ifdef SA_SIGINFO
 130 void rpmsqAction(int signum, siginfo_t * info, void * context)
 131 #else
 132 void rpmsqAction(int signum)
 133 #endif
 134 {
 135     int save = errno;
 136     rpmsig tbl;
 137
 138     for (tbl = rpmsigTbl; tbl->signum >= 0; tbl++) {
 139         if (tbl->signum != signum)
 140             continue;
 141
 142         (void) sigaddset(&rpmsqCaught, signum);
 143
 144         switch (signum) {
 145         case SIGCHLD:
 146             while (1) {
 147                 rpmsq sq;
 148                 int status = 0;
 149                 pid_t reaped = waitpid(0, &status, WNOHANG);
 150
 151                 /* XXX errno set to ECHILD/EINVAL/EINTR. */
 152                 if (reaped <= 0)
 153                     break;
 154
 155                 /* XXX insque(3)/remque(3) are dequeue, not ring. */
 156                 for (sq = rpmsqQueue->q_forw;
 157                      sq != NULL && sq != rpmsqQueue;
 158                      sq = sq->q_forw)
 159                 {
 160                     int ret;
 161
 162                     if (sq->child != reaped)
 163                         continue;
 164                     sq->reaped = reaped;
 165                     sq->status = status;
 166
 167                     /* Unlock the mutex.  The waiter will then be able to
 168                      * aquire the lock.
 169                      *
 170                      * XXX: jbj, wtd, if this fails?
 171                      */
 172                     ret = pthread_mutex_unlock(&sq->mutex);
 173
 174                     break;
 175                 }
 176             }
 177             break;
 178         default:
 179             break;
 180         }
 181         break;
 182     }
 183     errno = save;
 184 }
 185
 186 int rpmsqEnable(int signum, rpmsqAction_t handler)
 187 {
 188     int tblsignum = (signum >= 0 ? signum : -signum);
 189     struct sigaction sa;
 190     rpmsig tbl;
 191     int ret = -1;
 192
 193     (void) DO_LOCK ();
 194     if (rpmsqQueue->id == NULL)
 195         rpmsqQueue->id = ME();
 196     for (tbl = rpmsigTbl; tbl->signum >= 0; tbl++) {
 197         if (tblsignum != tbl->signum)
 198             continue;
 199
 200         if (signum >= 0) {                      /* Enable. */
 201             if (ADD_REF(tbl) <= 0) {
 202                 (void) sigdelset(&rpmsqCaught, tbl->signum);
 203
 204                 /* XXX Don't set a signal handler if already SIG_IGN */
 205                 (void) sigaction(tbl->signum, NULL, &tbl->oact);
 206                 if (tbl->oact.sa_handler == SIG_IGN)
 207                     continue;
 208
 209                 (void) sigemptyset (&sa.sa_mask);
 210 #ifdef SA_SIGINFO
 211                 sa.sa_flags = SA_SIGINFO;
 212 #else
 213                 sa.sa_flags = 0;
 214 #endif
 215                 sa.sa_sigaction = (handler != NULL ? handler : tbl->handler);
 216                 if (sigaction(tbl->signum, &sa, &tbl->oact) < 0) {
 217                     SUB_REF(tbl);
 218                     break;
 219                 }
 220                 tbl->active = 1;                /* XXX just in case */
 221                 if (handler != NULL)
 222                     tbl->handler = handler;
 223             }
 224         } else {                                /* Disable. */
 225             if (SUB_REF(tbl) <= 0) {
 226                 if (sigaction(tbl->signum, &tbl->oact, NULL) < 0)
 227                     break;
 228                 tbl->active = 0;                /* XXX just in case */
 229                 tbl->handler = (handler != NULL ? handler : rpmsqAction);
 230             }
 231         }
 232         ret = tbl->active;
 233         break;
 234     }
 235     (void) DO_UNLOCK ();
 236     return ret;
 237 }
 238
 239 pid_t rpmsqFork(rpmsq sq)
 240 {
 241     pid_t pid;
 242     int xx;
 243     int nothreads = 0;   /* XXX: Shouldn't this be a global? */
 244
 245     if (sq->reaper) {
 246         xx = rpmsqInsert(sq, NULL);
 247         xx = rpmsqEnable(SIGCHLD, NULL);
 248     }
 249
 250     xx = pipe(sq->pipes);
 251
 252     xx = sighold(SIGCHLD);
 253
 254     /*
 255      * Initialize the cond var mutex.   We have to aquire the lock we
 256      * use for the condition before we fork.  Otherwise it is possible for
 257      * the child to exit, we get sigchild and the sig handler to send
 258      * the condition signal before we are waiting on the condition.
 259      */
 260     if (!nothreads) {
 261         if(pthread_mutex_lock(&sq->mutex)) {
 262             /* Yack we did not get the lock, lets just give up */
 263             xx = close(sq->pipes[0]);
 264             xx = close(sq->pipes[1]);
 265             sq->pipes[0] = sq->pipes[1] = -1;
 266             goto out;
 267         }
 268     }
 269
 270     pid = fork();
 271     if (pid < (pid_t) 0) {              /* fork failed.  */
 272         sq->child = (pid_t)-1;
 273         xx = close(sq->pipes[0]);
 274         xx = close(sq->pipes[1]);
 275         sq->pipes[0] = sq->pipes[1] = -1;
 276         goto out;
 277     } else if (pid == (pid_t) 0) {      /* Child. */
 278         int yy;
 279
 280         /* Block to permit parent time to wait. */
 281         xx = close(sq->pipes[1]);
 282         xx = read(sq->pipes[0], &yy, sizeof(yy));
 283         xx = close(sq->pipes[0]);
 284         sq->pipes[0] = sq->pipes[1] = -1;
 285     } else {                            /* Parent. */
 286         sq->child = pid;
 287     }
 288
 289 out:
 290     xx = sigrelse(SIGCHLD);
 291     return sq->child;
 292 }
 293
 294 /**
 295  * Wait for child process to be reaped, and unregister SIGCHLD handler.
 296  * @todo Rewrite to use waitpid on helper thread.
 297  * @param sq            scriptlet queue element
 298  * @return              0 on success
 299  */
 300 static int rpmsqWaitUnregister(rpmsq sq)
 301 {
 302     int nothreads = 0;
 303     int ret = 0;
 304     int xx;
 305
 306     /* Protect sq->reaped from handler changes. */
 307     ret = sighold(SIGCHLD);
 308
 309     /* Start the child, linux often runs child before parent. */
 310     if (sq->pipes[0] >= 0)
 311         xx = close(sq->pipes[0]);
 312     if (sq->pipes[1] >= 0)
 313         xx = close(sq->pipes[1]);
 314     sq->pipes[0] = sq->pipes[1] = -1;
 315
 316     /* Put a stopwatch on the time spent waiting to measure performance gain. */
 317     (void) rpmswEnter(&sq->op, -1);
 318
 319     /* Wait for handler to receive SIGCHLD. */
 320     while (ret == 0 && sq->reaped != sq->child) {
 321         if (nothreads)
 322             /* Note that sigpause re-enables SIGCHLD. */
 323             ret = sigpause(SIGCHLD);
 324         else {
 325             xx = sigrelse(SIGCHLD);
 326
 327             /*
 328              * We start before the fork with this mutex locked;
 329              * The only one that unlocks this the signal handler.
 330              * So if we get the lock the child has been reaped.
 331              */
 332             ret = pthread_mutex_lock(&sq->mutex);
 333             xx = sighold(SIGCHLD);
 334         }
 335     }
 336
 337     /* Accumulate stopwatch time spent waiting, potential performance gain. */
 338     sq->ms_scriptlets += rpmswExit(&sq->op, -1)/1000;
 339
 340     xx = sigrelse(SIGCHLD);
 341
 342     /* Remove processed SIGCHLD item from queue. */
 343     xx = rpmsqRemove(sq);
 344
 345     /* Disable SIGCHLD handler on refcount == 0. */
 346     xx = rpmsqEnable(-SIGCHLD, NULL);
 347
 348     return ret;
 349 }
 350
 351 pid_t rpmsqWait(rpmsq sq)
 352 {
 353     if (sq->reaper) {
 354         (void) rpmsqWaitUnregister(sq);
 355     } else {
 356         pid_t reaped;
 357         int status;
 358         do {
 359             reaped = waitpid(sq->child, &status, 0);
 360         } while (reaped >= 0 && reaped != sq->child);
 361         sq->reaped = reaped;
 362         sq->status = status;
 363     }
 364
 365     return sq->reaped;
 366 }