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