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
dataq_check(dataq_t * ptr)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
dataq_init(dataq_t * ptr)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
dataq_enqueue(dataq_t * dataq,void * in)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
dataq_dequeue(dataq_t * dataq,void ** outptr,int try)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
dataq_data_destroy(void * p)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
dataq_waiters_destroy(void * p)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
dataq_destroy(dataq_t * dataq)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 }
144