usr/src/cmd/syslogd/queue.c

   1 /*
   2  * CDDL HEADER START
   3  *
   4  * The contents of this file are subject to the terms of the
   5  * Common Development and Distribution License (the "License").
   6  * You may not use this file except in compliance with the License.
   7  *
   8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
   9  * or http://www.opensolaris.org/os/licensing.
  10  * See the License for the specific language governing permissions
  11  * and limitations under the License.
  12  *
  13  * When distributing Covered Code, include this CDDL HEADER in each
  14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
  15  * If applicable, add the following below this CDDL HEADER, with the
  16  * fields enclosed by brackets "[]" replaced with your own identifying
  17  * information: Portions Copyright [yyyy] [name of copyright owner]
  18  *
  19  * CDDL HEADER END
  20  */
  21 /*
  22  * Copyright 2006 Sun Microsystems, Inc.  All rights reserved.
  23  * Use is subject to license terms.
  24  */
  25
  26 #pragma ident   "%Z%%M% %I%     %E% SMI"
  27
  28 #include <pthread.h>
  29 #include <malloc.h>
  30 #include <memory.h>
  31 #include "dataq.h"
  32 #include <assert.h>
  33
  34 #ifndef NDEBUG
  35 static int
  36 dataq_check(dataq_t *ptr)       /* call while holding lock! */
  37 {
  38         assert(ptr->num_data == ll_check(&ptr->data));
  39         assert(ptr->num_waiters == ll_check(&ptr->waiters));
  40         return (1);
  41 }
  42 #endif
  43
  44 int
  45 dataq_init(dataq_t *ptr)
  46 {
  47         ptr->num_data = 0;
  48         ptr->num_waiters = 0;
  49         ll_init(&ptr->data);
  50         ll_init(&ptr->waiters);
  51         (void) pthread_mutex_init(&ptr->lock, NULL);
  52         assert((pthread_mutex_lock(&ptr->lock) == 0) &&
  53                 (dataq_check(ptr) == 1) &&
  54                 (pthread_mutex_unlock(&ptr->lock) == 0));
  55         return (0);
  56 }
  57
  58 int
  59 dataq_enqueue(dataq_t *dataq, void *in)
  60 {
  61         dataq_data_t *ptr = (dataq_data_t *)malloc(sizeof (*ptr));
  62         dataq_waiter_t *sleeper;
  63
  64         if (ptr == NULL)
  65                 return (-1);
  66         ptr->data = in;
  67         (void) pthread_mutex_lock(&dataq->lock);
  68         assert(dataq_check(dataq));
  69         ll_enqueue(&dataq->data, &ptr->list);
  70         dataq->num_data++;
  71         if (dataq->num_waiters) {
  72                 /*LINTED*/
  73                 sleeper = (dataq_waiter_t *)ll_peek(&dataq->waiters);
  74                 sleeper->wakeup = 1;
  75                 (void) pthread_cond_signal(&sleeper->cv);
  76         }
  77         assert(dataq_check(dataq));
  78         (void) pthread_mutex_unlock(&dataq->lock);
  79         return (0);
  80 }
  81
  82 int
  83 dataq_dequeue(dataq_t *dataq, void **outptr, int try)
  84 {
  85         dataq_data_t *dptr;
  86         dataq_waiter_t *sleeper;
  87
  88         (void) pthread_mutex_lock(&dataq->lock);
  89         if ((dataq->num_waiters > 0) ||
  90             ((dptr = (dataq_data_t *)ll_dequeue(&dataq->data)) == NULL)) {
  91                 dataq_waiter_t wait;
  92                 if (try) {
  93                         (void) pthread_mutex_unlock(&dataq->lock);
  94                         return (1);
  95                 }
  96                 wait.wakeup = 0;
  97                 (void) pthread_cond_init(&wait.cv, NULL);
  98                 dataq->num_waiters++;
  99                 ll_enqueue(&dataq->waiters, &wait.list);
 100                 while (wait.wakeup == 0)
 101                         (void) pthread_cond_wait(&wait.cv, &dataq->lock);
 102                 (void) ll_dequeue(&dataq->waiters);
 103                 dataq->num_waiters--;
 104                 (void) pthread_cond_destroy(&wait.cv);
 105                 dptr = (dataq_data_t *)ll_dequeue(&dataq->data);
 106         }
 107         dataq->num_data--;
 108         if (dataq->num_data && dataq->num_waiters) {
 109                 /*LINTED*/
 110                 sleeper = (dataq_waiter_t *)ll_peek(&dataq->waiters);
 111                 sleeper->wakeup = 1;
 112                 (void) pthread_cond_signal(&sleeper->cv);
 113         }
 114         (void) pthread_mutex_unlock(&dataq->lock);
 115         *outptr = dptr->data;
 116         free(dptr);
 117         return (0);
 118 }
 119
 120 static void
 121 dataq_data_destroy(void * p)
 122 {
 123         dataq_data_t *d = (dataq_data_t *)p;
 124         free(d->data);
 125         free(d);
 126 }
 127
 128 static void
 129 dataq_waiters_destroy(void * p)
 130 {
 131         dataq_waiter_t *d = (dataq_waiter_t *)p;
 132         (void) pthread_cond_destroy(&d->cv);
 133         free(d);
 134 }
 135
 136 int
 137 dataq_destroy(dataq_t *dataq)
 138 {
 139         (void) pthread_mutex_destroy(&dataq->lock);
 140         ll_mapf(&dataq->data, dataq_data_destroy);
 141         ll_mapf(&dataq->waiters, dataq_waiters_destroy);
 142         return (0);
 143 }