/*
* CDDL HEADER START
*
* The contents of this file are subject to the terms of the
* Common Development and Distribution License (the "License").
* You may not use this file except in compliance with the License.
*
* You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
* or http://www.opensolaris.org/os/licensing.
* See the License for the specific language governing permissions
* and limitations under the License.
*
* When distributing Covered Code, include this CDDL HEADER in each
* file and include the License file at usr/src/OPENSOLARIS.LICENSE.
* If applicable, add the following below this CDDL HEADER, with the
* fields enclosed by brackets "[]" replaced with your own identifying
* information: Portions Copyright [yyyy] [name of copyright owner]
*
* CDDL HEADER END
*/
/*
* Copyright 2006 Sun Microsystems, Inc. All rights reserved.
* Use is subject to license terms.
*/
#pragma ident "%Z%%M% %I% %E% SMI"
#include <pthread.h>
#include <malloc.h>
#include <memory.h>
#include "dataq.h"
#include <assert.h>
#ifndef NDEBUG
static int
dataq_check(dataq_t *ptr) /* call while holding lock! */
{
assert(ptr->num_data == ll_check(&ptr->data));
assert(ptr->num_waiters == ll_check(&ptr->waiters));
return (1);
}
#endif
int
dataq_init(dataq_t *ptr)
{
ptr->num_data = 0;
ptr->num_waiters = 0;
ll_init(&ptr->data);
ll_init(&ptr->waiters);
(void) pthread_mutex_init(&ptr->lock, NULL);
assert((pthread_mutex_lock(&ptr->lock) == 0) &&
(dataq_check(ptr) == 1) &&
(pthread_mutex_unlock(&ptr->lock) == 0));
return (0);
}
int
dataq_enqueue(dataq_t *dataq, void *in)
{
dataq_data_t *ptr = (dataq_data_t *)malloc(sizeof (*ptr));
dataq_waiter_t *sleeper;
if (ptr == NULL)
return (-1);
ptr->data = in;
(void) pthread_mutex_lock(&dataq->lock);
assert(dataq_check(dataq));
ll_enqueue(&dataq->data, &ptr->list);
dataq->num_data++;
if (dataq->num_waiters) {
/*LINTED*/
sleeper = (dataq_waiter_t *)ll_peek(&dataq->waiters);
sleeper->wakeup = 1;
(void) pthread_cond_signal(&sleeper->cv);
}
assert(dataq_check(dataq));
(void) pthread_mutex_unlock(&dataq->lock);
return (0);
}
int
dataq_dequeue(dataq_t *dataq, void **outptr, int try)
{
dataq_data_t *dptr;
dataq_waiter_t *sleeper;
(void) pthread_mutex_lock(&dataq->lock);
if ((dataq->num_waiters > 0) ||
((dptr = (dataq_data_t *)ll_dequeue(&dataq->data)) == NULL)) {
dataq_waiter_t wait;
if (try) {
(void) pthread_mutex_unlock(&dataq->lock);
return (1);
}
wait.wakeup = 0;
(void) pthread_cond_init(&wait.cv, NULL);
dataq->num_waiters++;
ll_enqueue(&dataq->waiters, &wait.list);
while (wait.wakeup == 0)
(void) pthread_cond_wait(&wait.cv, &dataq->lock);
(void) ll_dequeue(&dataq->waiters);
dataq->num_waiters--;
(void) pthread_cond_destroy(&wait.cv);
dptr = (dataq_data_t *)ll_dequeue(&dataq->data);
}
dataq->num_data--;
if (dataq->num_data && dataq->num_waiters) {
/*LINTED*/
sleeper = (dataq_waiter_t *)ll_peek(&dataq->waiters);
sleeper->wakeup = 1;
(void) pthread_cond_signal(&sleeper->cv);
}
(void) pthread_mutex_unlock(&dataq->lock);
*outptr = dptr->data;
free(dptr);
return (0);
}
static void
dataq_data_destroy(void * p)
{
dataq_data_t *d = (dataq_data_t *)p;
free(d->data);
free(d);
}
static void
dataq_waiters_destroy(void * p)
{
dataq_waiter_t *d = (dataq_waiter_t *)p;
(void) pthread_cond_destroy(&d->cv);
free(d);
}
int
dataq_destroy(dataq_t *dataq)
{
(void) pthread_mutex_destroy(&dataq->lock);
ll_mapf(&dataq->data, dataq_data_destroy);
ll_mapf(&dataq->waiters, dataq_waiters_destroy);
return (0);
}