mirror of
https://github.com/MariaDB/server.git
synced 2025-02-02 20:11:42 +01:00
122 lines
3.9 KiB
C++
122 lines
3.9 KiB
C++
/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
|
|
// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
|
|
#ident "$Id$"
|
|
#ident "Copyright (c) 2007-2012 Tokutek Inc. All rights reserved."
|
|
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
|
|
|
|
typedef struct workqueue *WORKQUEUE;
|
|
struct workqueue {
|
|
WORKITEM head, tail; // head and tail of the linked list of work items
|
|
pthread_cond_t wait_read; // wait for read
|
|
int want_read; // number of threads waiting to read
|
|
pthread_cond_t wait_write; // wait for write
|
|
int want_write; // number of threads waiting to write
|
|
int ninq; // number of work items in the queue
|
|
char closed; // kicks waiting threads off of the write queue
|
|
};
|
|
|
|
// initialize a workqueue
|
|
// expects: the workqueue is not initialized
|
|
// effects: the workqueue is set to empty and the condition variable is initialized
|
|
|
|
static void workqueue_init(WORKQUEUE wq) {
|
|
wq->head = wq->tail = 0;
|
|
int r;
|
|
r = pthread_cond_init(&wq->wait_read, 0); assert(r == 0);
|
|
wq->want_read = 0;
|
|
r = pthread_cond_init(&wq->wait_write, 0); assert(r == 0);
|
|
wq->want_write = 0;
|
|
wq->ninq = 0;
|
|
wq->closed = 0;
|
|
}
|
|
|
|
// destroy a workqueue
|
|
// expects: the workqueue must be initialized and empty
|
|
|
|
static void workqueue_destroy(WORKQUEUE wq) {
|
|
assert(wq->head == 0 && wq->tail == 0);
|
|
int r;
|
|
r = pthread_cond_destroy(&wq->wait_read); assert(r == 0);
|
|
r = pthread_cond_destroy(&wq->wait_write); assert(r == 0);
|
|
}
|
|
|
|
// close the workqueue
|
|
// effects: signal any threads blocked in the workqueue
|
|
|
|
static void workqueue_set_closed(WORKQUEUE wq) {
|
|
wq->closed = 1;
|
|
int r;
|
|
r = pthread_cond_broadcast(&wq->wait_read); assert(r == 0);
|
|
r = pthread_cond_broadcast(&wq->wait_write); assert(r == 0);
|
|
}
|
|
|
|
// determine whether or not the write queue is empty
|
|
// return: 1 if the write queue is empty, otherwise 0
|
|
|
|
static int workqueue_empty(WORKQUEUE wq) {
|
|
return wq->head == 0;
|
|
}
|
|
|
|
// put a work item at the tail of the write queue
|
|
// expects: the mutex is locked
|
|
// effects: append the workitem to the end of the write queue and signal
|
|
// any readers
|
|
|
|
static void workqueue_enq(WORKQUEUE wq, WORKITEM workitem) {
|
|
workitem->next_wq = 0;
|
|
if (wq->tail)
|
|
wq->tail->next_wq = workitem;
|
|
else
|
|
wq->head = workitem;
|
|
wq->tail = workitem;
|
|
wq->ninq++;
|
|
if (wq->want_read) {
|
|
int r = pthread_cond_signal(&wq->wait_read); assert(r == 0);
|
|
}
|
|
}
|
|
|
|
// get a workitem from the head of the write queue
|
|
// expects: the mutex is locked
|
|
// effects: wait until the workqueue is not empty, remove the first workitem from the
|
|
// write queue and return it
|
|
// returns: 0 if success, otherwise an error
|
|
|
|
static int workqueue_deq(WORKQUEUE wq, pthread_mutex_t *mutex, WORKITEM *workitemptr) {
|
|
while (workqueue_empty(wq)) {
|
|
if (wq->closed)
|
|
return EINVAL;
|
|
wq->want_read++;
|
|
int r = pthread_cond_wait(&wq->wait_read, mutex); assert(r == 0);
|
|
wq->want_read--;
|
|
}
|
|
WORKITEM workitem = wq->head;
|
|
wq->head = workitem->next_wq;
|
|
if (wq->head == 0)
|
|
wq->tail = 0;
|
|
wq->ninq--;
|
|
workitem->next_wq = 0;
|
|
*workitemptr = workitem;
|
|
return 0;
|
|
}
|
|
|
|
#if 0
|
|
|
|
// suspend the writer thread
|
|
// expects: the mutex is locked
|
|
|
|
static void workqueue_wait_write(WORKQUEUE wq, pthread_mutex_t *mutex) {
|
|
wq->want_write++;
|
|
int r = pthread_cond_wait(&wq->wait_write, mutex); assert(r == 0);
|
|
wq->want_write--;
|
|
}
|
|
|
|
// wakeup the writer threads
|
|
// expects: the mutex is locked
|
|
|
|
static void workqueue_wakeup_write(WORKQUEUE wq) {
|
|
if (wq->want_write) {
|
|
int r = pthread_cond_broadcast(&wq->wait_write); assert(r == 0);
|
|
}
|
|
}
|
|
|
|
#endif
|