introduce struct condition as a more correct version of Notify

src/Makefile.am

@@ -36,6 +36,7 @@ mpd_headers = \
 	buffer2array.h \
 	charConv.h \
 	command.h \
+	condition.h \
 	conf.h \
 	dbUtils.h \
 	decode.h \
@@ -97,6 +98,7 @@ mpd_SOURCES = \
 	buffer2array.c \
 	charConv.c \
 	command.c \
+	condition.c \
 	conf.c \
 	dbUtils.c \
 	decode.c \

src/condition.c

+/* the Music Player Daemon (MPD)
+ * Copyright (C) 2003-2007 by Warren Dukes (warren.dukes@gmail.com)
+ * Copyright (C) 2008 Max Kellermann <max@duempel.org>
+ * This project's homepage is: http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+
+#include "condition.h"
+#include "utils.h"
+#include "log.h"
+
+void cond_init(struct condition *cond)
+{
+	xpthread_mutex_init(&cond->mutex, NULL);
+	xpthread_cond_init(&cond->cond, NULL);
+}
+
+void cond_enter(struct condition *cond)
+{
+	pthread_mutex_lock(&cond->mutex);
+}
+
+void cond_leave(struct condition *cond)
+{
+	pthread_mutex_unlock(&cond->mutex);
+}
+
+void cond_wait(struct condition *cond)
+{
+	pthread_cond_wait(&cond->cond, &cond->mutex);
+}
+
+static struct timespec * ts_timeout(struct timespec *ts, const long sec)
+{
+	struct timeval tv;
+	gettimeofday(&tv, NULL);
+	ts->tv_sec = tv.tv_sec + sec;
+	ts->tv_nsec = tv.tv_usec * 1000;
+	return ts;
+}
+
+int cond_timedwait(struct condition *cond, const long sec)
+{
+	struct timespec ts;
+	int ret = pthread_cond_timedwait(&cond->cond, &cond->mutex,
+	                                 ts_timeout(&ts, sec));
+	if (!ret || ret == ETIMEDOUT)
+		return ret;
+	FATAL("cond_timedwait: %s\n", strerror(ret));
+	return ret;
+}
+
+int cond_signal_async(struct condition *cond)
+{
+	if (!pthread_mutex_trylock(&cond->mutex)) {
+		pthread_cond_signal(&cond->cond);
+		pthread_mutex_unlock(&cond->mutex);
+		return 0;
+	}
+	return EBUSY;
+}
+
+void cond_signal_sync(struct condition *cond)
+{
+	pthread_cond_signal(&cond->cond);
+}
+
+void cond_destroy(struct condition *cond)
+{
+	xpthread_cond_destroy(&cond->cond);
+	xpthread_mutex_destroy(&cond->mutex);
+}

src/condition.h

+/* the Music Player Daemon (MPD)
+ * Copyright (C) 2003-2007 by Warren Dukes (warren.dukes@gmail.com)
+ * Copyright (C) 2008 Max Kellermann <max@duempel.org>
+ * This project's homepage is: http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+
+#ifndef CONDITION_H
+#define CONDITION_H
+
+#include "os_compat.h"
+
+struct condition {
+	pthread_mutex_t mutex;
+	pthread_cond_t cond;
+};
+
+void cond_init(struct condition *cond);
+
+/**
+ * The thread which shall be notified by this object must call this
+ * function before any cond_wait() invocation.  It locks the mutex.
+ */
+void cond_enter(struct condition *cond);
+
+/**
+ * Neutralize cond_leave().
+ */
+void cond_leave(struct condition *cond);
+
+/**
+ * Wait for a conditio.  Return immediately if we have already
+ * been notified since we last returned from cond_wait().
+ */
+void cond_wait(struct condition *cond);
+
+/**
+ * Wait for a condition with timeout
+ *
+ * @param sec number of seconds to wait for (subject to change)
+ *
+ * @return ETIMEDOUT if timed out, 0 if notification was received
+ */
+int cond_timedwait(struct condition *cond, const long sec);
+
+/**
+ * Notify the thread there is a waiter.  This function never blocks.
+ *
+ * @return EBUSY if it was unable to lock the mutex, 0 on success
+ */
+int cond_signal_async(struct condition *cond);
+
+/**
+ * Notify the thread synchronously, i.e. wait until it can deliver
+ * the notification.
+ */
+void cond_signal_sync(struct condition *cond);
+
+/**
+ * cond_destroy - destroy the cond and internal structures
+ */
+void cond_destroy(struct condition *cond);
+
+#endif /* CONDITION_H */

src/inputStream_http.c

@@ -24,6 +24,7 @@
 #include "conf.h"
 #include "os_compat.h"
 #include "ringbuf.h"
+#include "condition.h"
 
 enum conn_state { /* only written by io thread, read by both */
 	CONN_STATE_NEW,             /* just (re)initialized */
@@ -71,14 +72,9 @@ struct http_data {
 		pthread_t io_thread;
 		struct ringbuf *rb;
 
-		/* TODO: fix Notify so it doesn't use ugly "pending" flag */
-		pthread_mutex_t full_lock;
-		pthread_cond_t full_cond;
-		pthread_mutex_t empty_lock;
-		pthread_cond_t empty_cond;
-
-		pthread_mutex_t action_lock;
-		pthread_cond_t action_cond;
+		struct condition full_cond;
+		struct condition empty_cond;
+		struct condition action_cond;
 	/* } */
 
 	int nr_redirect;
@@ -112,12 +108,9 @@ static void init_http_data(struct http_data *data)
 	data->icy_offset = 0;
 	data->rb = ringbuf_create(buffer_size);
 
-	pthread_cond_init(&data->action_cond, NULL);
-	pthread_mutex_init(&data->action_lock, NULL);
-	pthread_cond_init(&data->full_cond, NULL);
-	pthread_mutex_init(&data->full_lock, NULL);
-	pthread_cond_init(&data->empty_cond, NULL);
-	pthread_mutex_init(&data->empty_lock, NULL);
+	cond_init(&data->action_cond);
+	cond_init(&data->full_cond);
+	cond_init(&data->empty_cond);
 }
 
 static struct http_data *new_http_data(void)
@@ -135,12 +128,9 @@ static void free_http_data(struct http_data * data)
 	if (data->proxy_auth) free(data->proxy_auth);
 	if (data->http_auth) free(data->http_auth);
 
-	xpthread_cond_destroy(&data->action_cond);
-	xpthread_mutex_destroy(&data->action_lock);
-	xpthread_cond_destroy(&data->full_cond);
-	xpthread_mutex_destroy(&data->full_lock);
-	xpthread_cond_destroy(&data->empty_cond);
-	xpthread_mutex_destroy(&data->empty_lock);
+	cond_destroy(&data->action_cond);
+	cond_destroy(&data->full_cond);
+	cond_destroy(&data->empty_cond);
 
 	xclose(data->pipe_fds[0]);
 	xclose(data->pipe_fds[1]);
@@ -242,15 +232,6 @@ static int parse_url(struct http_data * data, char *url)
 	return 0;
 }
 
-static struct timespec * ts_timeout(struct timespec *ts, const long sec)
-{
-	struct timeval tv;
-	gettimeofday(&tv, NULL);
-	ts->tv_sec = tv.tv_sec + sec;
-	ts->tv_nsec = tv.tv_usec * 1000;
-	return ts;
-}
-
 /* triggers an action and waits for completion */
 static int trigger_action(struct http_data *data,
                           enum conn_action action,
@@ -259,7 +240,7 @@ static int trigger_action(struct http_data *data,
 	int ret = -1;
 
 	assert(!pthread_equal(data->io_thread, pthread_self()));
-	pthread_mutex_lock(&data->action_lock);
+	cond_enter(&data->action_cond);
 	if (data->action != CONN_ACTION_NONE)
 		goto out;
 	data->action = action;
@@ -270,17 +251,13 @@ static int trigger_action(struct http_data *data,
 		data->action = CONN_ACTION_NONE;
 		goto out;
 	}
-	if (nonblocking) {
-		struct timespec ts;
-		pthread_cond_timedwait(&data->action_cond,
-		                       &data->action_lock,
-		                       ts_timeout(&ts, 1));
-	} else {
-		pthread_cond_wait(&data->action_cond, &data->action_lock);
-	}
+	if (nonblocking)
+		cond_timedwait(&data->action_cond, 1);
+	else
+		cond_wait(&data->action_cond);
 	ret = 0;
 out:
-	pthread_mutex_unlock(&data->action_lock);
+	cond_leave(&data->action_cond);
 	return ret;
 }
 
@@ -288,10 +265,10 @@ static int take_action(struct http_data *data)
 {
 	assert(pthread_equal(data->io_thread, pthread_self()));
 
-	pthread_mutex_lock(&data->action_lock);
+	cond_enter(&data->action_cond);
 	switch (data->action) {
 	case CONN_ACTION_NONE:
-		pthread_mutex_unlock(&data->action_lock);
+		cond_leave(&data->action_cond);
 		return 0;
 	case CONN_ACTION_DOSEEK:
 		data->state = CONN_STATE_NEW;
@@ -302,8 +279,8 @@ static int take_action(struct http_data *data)
 	xclose(data->fd);
 	data->fd = -1;
 	data->action = CONN_ACTION_NONE;
-	pthread_cond_signal(&data->action_cond);
-	pthread_mutex_unlock(&data->action_lock);
+	cond_signal_sync(&data->action_cond);
+	cond_leave(&data->action_cond);
 	return 1;
 }
 
@@ -450,7 +427,7 @@ static void await_buffer_space(struct http_data *data)
 {
 	assert(pthread_equal(data->io_thread, pthread_self()));
 	assert_state(CONN_STATE_BUFFER_FULL);
-	pthread_cond_wait(&data->full_cond, &data->full_lock);
+	cond_wait(&data->full_cond);
 	if (ringbuf_write_space(data->rb) > 0)
 		data->state = CONN_STATE_BUFFER;
 	/* else spurious wakeup or action triggered ... */
@@ -459,32 +436,20 @@ static void await_buffer_space(struct http_data *data)
 static void feed_starved(struct http_data *data)
 {
 	assert(pthread_equal(data->io_thread, pthread_self()));
-
-	if (!pthread_mutex_trylock(&data->empty_lock)) {
-		pthread_cond_signal(&data->empty_cond);
-		pthread_mutex_unlock(&data->empty_lock);
-	}
+	cond_signal_async(&data->empty_cond);
 }
 
 static int starved_wait(struct http_data *data, const long sec)
 {
-	struct timespec ts;
-
 	assert(!pthread_equal(data->io_thread, pthread_self()));
-	return pthread_cond_timedwait(&data->empty_cond,
-	                              &data->empty_lock,
-	                              ts_timeout(&ts, sec));
+	return cond_timedwait(&data->empty_cond, sec);
 }
 
 static int awaken_buffer_task(struct http_data *data)
 {
 	assert(!pthread_equal(data->io_thread, pthread_self()));
-	if (!pthread_mutex_trylock(&data->full_lock)) {
-		pthread_cond_signal(&data->full_cond);
-		pthread_mutex_unlock(&data->full_lock);
-		return 1;
-	}
-	return 0;
+
+	return ! cond_signal_async(&data->full_cond);
 }
 
 static ssize_t buffer_data(InputStream *is)
@@ -741,7 +706,7 @@ static void * http_io_task(void *arg)
 	InputStream *is = (InputStream *) arg;
 	struct http_data *data = (struct http_data *) is->data;
 
-	pthread_mutex_lock(&data->full_lock);
+	cond_enter(&data->full_cond);
 	while (1) {
 		take_action(data);
 		switch (data->state) {
@@ -782,7 +747,7 @@ err:
 	err_close(data);
 closed:
 	assert_state(CONN_STATE_CLOSED);
-	pthread_mutex_unlock(&data->full_lock);
+	cond_leave(&data->full_cond);
 	return NULL;
 }
 
@@ -813,7 +778,7 @@ int inputStream_httpOpen(InputStream * is, char *url)
 	if (pthread_create(&data->io_thread, &attr, http_io_task, is))
 		FATAL("failed to spawn http_io_task: %s", strerror(errno));
 
-	pthread_mutex_lock(&data->empty_lock);
+	cond_enter(&data->empty_cond); /* httpClose will leave this */
 	return 0;
 }
 
@@ -982,7 +947,7 @@ int inputStream_httpClose(InputStream * is)
 	while (data->state != CONN_STATE_CLOSED)
 		trigger_action(data, CONN_ACTION_CLOSE, 1);
 	pthread_join(data->io_thread, NULL);
-	pthread_mutex_unlock(&data->empty_lock);
+	cond_leave(&data->empty_cond);
 	free_http_data(data);
 	return 0;
 }