aboutsummaryrefslogtreecommitdiffstats
path: root/src/input/curl_input_plugin.c
diff options
context:
space:
mode:
authorMax Kellermann <max@duempel.org>2011-09-14 21:46:41 +0200
committerMax Kellermann <max@duempel.org>2011-09-16 21:22:13 +0200
commit754f26a97c816781e80500d98f2515ae97836145 (patch)
treecaa7dbaa879b29d018a4559524390670ad33a605 /src/input/curl_input_plugin.c
parent29241c4f835797f635816a9f37528aa981f722b5 (diff)
downloadmpd-754f26a97c816781e80500d98f2515ae97836145.tar.gz
mpd-754f26a97c816781e80500d98f2515ae97836145.tar.xz
mpd-754f26a97c816781e80500d98f2515ae97836145.zip
input_stream: non-blocking I/O
Add GMutex, GCond attributes which will be used by callers to conditionally wait on the stream. Remove the (now-useless) plugin method buffer(), wait on GCond instead. Lock the input_stream before each method call. Do the same with the playlist plugins.
Diffstat (limited to '')
-rw-r--r--src/input/curl_input_plugin.c108
1 files changed, 36 insertions, 72 deletions
diff --git a/src/input/curl_input_plugin.c b/src/input/curl_input_plugin.c
index 4f3e9c6e4..88a5556d2 100644
--- a/src/input/curl_input_plugin.c
+++ b/src/input/curl_input_plugin.c
@@ -81,9 +81,6 @@ struct input_curl {
/** the curl handles */
CURL *easy;
- GMutex *mutex;
- GCond *cond;
-
/** the GMainLoop source used to poll all CURL file
descriptors */
GSource *source;
@@ -433,11 +430,11 @@ input_curl_abort_all_requests(GError *error)
input_curl_easy_free(c);
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
c->postponed_error = g_error_copy(error);
c->base.ready = true;
- g_cond_broadcast(c->cond);
- g_mutex_unlock(c->mutex);
+ g_cond_broadcast(c->base.cond);
+ g_mutex_unlock(c->base.mutex);
}
g_error_free(error);
@@ -457,7 +454,7 @@ input_curl_request_done(struct input_curl *c, CURLcode result, long status)
assert(c->easy == NULL);
assert(c->postponed_error == NULL);
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
if (result != CURLE_OK) {
c->postponed_error = g_error_new(curl_quark(), result,
@@ -470,8 +467,8 @@ input_curl_request_done(struct input_curl *c, CURLcode result, long status)
}
c->base.ready = true;
- g_cond_broadcast(c->cond);
- g_mutex_unlock(c->mutex);
+ g_cond_broadcast(c->base.cond);
+ g_mutex_unlock(c->base.mutex);
}
static void
@@ -763,9 +760,6 @@ input_curl_free(struct input_curl *c)
g_queue_free(c->buffers);
- g_mutex_free(c->mutex);
- g_cond_free(c->cond);
-
if (c->postponed_error != NULL)
g_error_free(c->postponed_error);
@@ -779,15 +773,12 @@ input_curl_check(struct input_stream *is, GError **error_r)
{
struct input_curl *c = (struct input_curl *)is;
- g_mutex_lock(c->mutex);
-
bool success = c->postponed_error == NULL;
if (!success) {
g_propagate_error(error_r, c->postponed_error);
c->postponed_error = NULL;
}
- g_mutex_unlock(c->mutex);
return success;
}
@@ -805,7 +796,7 @@ static bool
fill_buffer(struct input_curl *c, GError **error_r)
{
while (c->easy != NULL && g_queue_is_empty(c->buffers))
- g_cond_wait(c->cond, c->mutex);
+ g_cond_wait(c->base.cond, c->base.mutex);
if (c->postponed_error != NULL) {
g_propagate_error(error_r, c->postponed_error);
@@ -906,6 +897,15 @@ copy_icy_tag(struct input_curl *c)
c->tag = tag;
}
+static bool
+input_curl_available(struct input_stream *is)
+{
+ struct input_curl *c = (struct input_curl *)is;
+
+ return c->postponed_error != NULL || c->easy == NULL ||
+ !g_queue_is_empty(c->buffers);
+}
+
static size_t
input_curl_read(struct input_stream *is, void *ptr, size_t size,
GError **error_r)
@@ -915,16 +915,12 @@ input_curl_read(struct input_stream *is, void *ptr, size_t size,
size_t nbytes = 0;
char *dest = ptr;
- g_mutex_lock(c->mutex);
-
do {
/* fill the buffer */
success = fill_buffer(c, error_r);
- if (!success) {
- g_mutex_unlock(c->mutex);
+ if (!success)
return 0;
- }
/* send buffer contents */
@@ -944,14 +940,12 @@ input_curl_read(struct input_stream *is, void *ptr, size_t size,
#if LIBCURL_VERSION_NUM >= 0x071200
if (c->paused && curl_total_buffer_size(c) < CURL_RESUME_AT) {
- g_mutex_unlock(c->mutex);
+ g_mutex_unlock(c->base.mutex);
io_thread_call(input_curl_resume, c);
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
}
#endif
- g_mutex_unlock(c->mutex);
-
return nbytes;
}
@@ -968,33 +962,7 @@ input_curl_eof(G_GNUC_UNUSED struct input_stream *is)
{
struct input_curl *c = (struct input_curl *)is;
- g_mutex_lock(c->mutex);
- bool eof = c->easy == NULL && g_queue_is_empty(c->buffers);
- g_mutex_unlock(c->mutex);
-
- return eof;
-}
-
-static int
-input_curl_buffer(struct input_stream *is, GError **error_r)
-{
- struct input_curl *c = (struct input_curl *)is;
-
- g_mutex_lock(c->mutex);
-
- int result;
- if (c->postponed_error != NULL) {
- g_propagate_error(error_r, c->postponed_error);
- c->postponed_error = NULL;
- result = -1;
- } else if (g_queue_is_empty(c->buffers))
- result = 0;
- else
- result = 1;
-
- g_mutex_unlock(c->mutex);
-
- return result;
+ return c->easy == NULL && g_queue_is_empty(c->buffers);
}
/** called by curl when new data is available */
@@ -1092,12 +1060,12 @@ input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
if (size == 0)
return 0;
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
#if LIBCURL_VERSION_NUM >= 0x071200
if (curl_total_buffer_size(c) + size >= CURL_MAX_BUFFERED) {
c->paused = true;
- g_mutex_unlock(c->mutex);
+ g_mutex_unlock(c->base.mutex);
return CURL_WRITEFUNC_PAUSE;
}
#endif
@@ -1108,11 +1076,10 @@ input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
memcpy(buffer->data, ptr, size);
g_queue_push_tail(c->buffers, buffer);
-
c->base.ready = true;
- g_cond_broadcast(c->cond);
- g_mutex_unlock(c->mutex);
+ g_cond_broadcast(c->base.cond);
+ g_mutex_unlock(c->base.mutex);
return size;
}
@@ -1219,8 +1186,6 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
/* check if we can fast-forward the buffer */
- g_mutex_lock(c->mutex);
-
while (offset > is->offset && !g_queue_is_empty(c->buffers)) {
struct buffer *buffer;
size_t length;
@@ -1238,13 +1203,13 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
is->offset += length;
}
- g_mutex_unlock(c->mutex);
-
if (offset == is->offset)
return true;
/* close the old connection and open a new one */
+ g_mutex_unlock(c->base.mutex);
+
input_curl_easy_free_indirect(c);
input_curl_flush_buffers(c);
@@ -1272,36 +1237,35 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
if (!input_curl_easy_add_indirect(c, error_r))
return false;
- g_mutex_lock(c->mutex);
+ g_mutex_lock(c->base.mutex);
while (!c->base.ready)
- g_cond_wait(c->cond, c->mutex);
+ g_cond_wait(c->base.cond, c->base.mutex);
if (c->postponed_error != NULL) {
g_propagate_error(error_r, c->postponed_error);
c->postponed_error = NULL;
- g_mutex_unlock(c->mutex);
return false;
}
- g_mutex_unlock(c->mutex);
-
return true;
}
static struct input_stream *
-input_curl_open(const char *url, GError **error_r)
+input_curl_open(const char *url, GMutex *mutex, GCond *cond,
+ GError **error_r)
{
+ assert(mutex != NULL);
+ assert(cond != NULL);
+
struct input_curl *c;
if (strncmp(url, "http://", 7) != 0)
return NULL;
c = g_new0(struct input_curl, 1);
- input_stream_init(&c->base, &input_plugin_curl, url);
-
- c->mutex = g_mutex_new();
- c->cond = g_cond_new();
+ input_stream_init(&c->base, &input_plugin_curl, url,
+ mutex, cond);
c->url = g_strdup(url);
c->buffers = g_queue_new();
@@ -1337,7 +1301,7 @@ const struct input_plugin input_plugin_curl = {
.close = input_curl_close,
.check = input_curl_check,
.tag = input_curl_tag,
- .buffer = input_curl_buffer,
+ .available = input_curl_available,
.read = input_curl_read,
.eof = input_curl_eof,
.seek = input_curl_seek,