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.

14 files changed, 377 insertions, 185 deletions

diff --git a/src/input/archive_input_plugin.c b/src/input/archive_input_plugin.c
index 8d78f4c89..4a038b9e2 100644
--- a/src/input/archive_input_plugin.c
+++ b/src/input/archive_input_plugin.c
@@ -34,7 +34,9 @@
  * plugin and gzip fetches file from disk
  */
 static struct input_stream *
-input_archive_open(const char *pathname, GError **error_r)
+input_archive_open(const char *pathname,
+		   GMutex *mutex, GCond *cond,
+		   GError **error_r)
 {
 	const struct archive_plugin *arplug;
 	struct archive_file *file;
@@ -65,7 +67,8 @@ input_archive_open(const char *pathname, GError **error_r)
 		return NULL;
 
 	//setup fileops
-	is = archive_file_open_stream(file, filename, error_r);
+	is = archive_file_open_stream(file, filename, mutex, cond,
+				      error_r);
 	archive_file_close(file);
 	g_free(pname);
 
diff --git a/src/input/cdio_paranoia_input_plugin.c b/src/input/cdio_paranoia_input_plugin.c
index da2ed1096..1a1c8d2c5 100644
--- a/src/input/cdio_paranoia_input_plugin.c
+++ b/src/input/cdio_paranoia_input_plugin.c
@@ -149,7 +149,9 @@ cdio_detect_device(void)
 }
 
 static struct input_stream *
-input_cdio_open(const char *uri, GError **error_r)
+input_cdio_open(const char *uri,
+		GMutex *mutex, GCond *cond,
+		GError **error_r)
 {
 	struct input_cdio_paranoia *i;
 
@@ -158,7 +160,8 @@ input_cdio_open(const char *uri, GError **error_r)
 		return NULL;
 
 	i = g_new(struct input_cdio_paranoia, 1);
-	input_stream_init(&i->base, &input_plugin_cdio_paranoia, uri);
+	input_stream_init(&i->base, &input_plugin_cdio_paranoia, uri,
+			  mutex, cond);
 
 	/* initialize everything (should be already) */
 	i->drv = NULL;
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,
diff --git a/src/input/despotify_input_plugin.c b/src/input/despotify_input_plugin.c
index ef78fb1e0..200a0afd6 100644
--- a/src/input/despotify_input_plugin.c
+++ b/src/input/despotify_input_plugin.c
@@ -97,7 +97,9 @@ static void callback(G_GNUC_UNUSED struct despotify_session* ds,
 
 
 static struct input_stream *
-input_despotify_open(const char *url, G_GNUC_UNUSED GError **error_r)
+input_despotify_open(const char *url,
+		     GMutex *mutex, GCond *cond,
+		     G_GNUC_UNUSED GError **error_r)
 {
 	struct input_despotify *ctx;
 	struct despotify_session *session;
@@ -131,7 +133,8 @@ input_despotify_open(const char *url, G_GNUC_UNUSED GError **error_r)
 		return NULL;
 	}
 
-	input_stream_init(&ctx->base, &input_plugin_despotify, url);
+	input_stream_init(&ctx->base, &input_plugin_despotify, url,
+			  mutex, cond);
 	ctx->session = session;
 	ctx->track = track;
 	ctx->tag = mpd_despotify_tag_from_track(track);
diff --git a/src/input/ffmpeg_input_plugin.c b/src/input/ffmpeg_input_plugin.c
index 6caa7ea04..1c64b52c1 100644
--- a/src/input/ffmpeg_input_plugin.c
+++ b/src/input/ffmpeg_input_plugin.c
@@ -74,7 +74,9 @@ input_ffmpeg_init(G_GNUC_UNUSED const struct config_param *param,
 }
 
 static struct input_stream *
-input_ffmpeg_open(const char *uri, GError **error_r)
+input_ffmpeg_open(const char *uri,
+		  GMutex *mutex, GCond *cond,
+		  GError **error_r)
 {
 	struct input_ffmpeg *i;
 
@@ -87,7 +89,8 @@ input_ffmpeg_open(const char *uri, GError **error_r)
 		return NULL;
 
 	i = g_new(struct input_ffmpeg, 1);
-	input_stream_init(&i->base, &input_plugin_ffmpeg, uri);
+	input_stream_init(&i->base, &input_plugin_ffmpeg, uri,
+			  mutex, cond);
 
 #if LIBAVFORMAT_VERSION_INT >= AV_VERSION_INT(53,1,0)
 	int ret = avio_open(&i->h, uri, AVIO_FLAG_READ);
diff --git a/src/input/file_input_plugin.c b/src/input/file_input_plugin.c
index 10b753d24..5ee3f200b 100644
--- a/src/input/file_input_plugin.c
+++ b/src/input/file_input_plugin.c
@@ -46,7 +46,9 @@ file_quark(void)
 }
 
 static struct input_stream *
-input_file_open(const char *filename, GError **error_r)
+input_file_open(const char *filename,
+		GMutex *mutex, GCond *cond,
+		GError **error_r)
 {
 	int fd, ret;
 	struct stat st;
@@ -85,7 +87,8 @@ input_file_open(const char *filename, GError **error_r)
 #endif
 
 	fis = g_new(struct file_input_stream, 1);
-	input_stream_init(&fis->base, &input_plugin_file, filename);
+	input_stream_init(&fis->base, &input_plugin_file, filename,
+			  mutex, cond);
 
 	fis->base.size = st.st_size;
 	fis->base.seekable = true;
diff --git a/src/input/mms_input_plugin.c b/src/input/mms_input_plugin.c
index eb3e5d26c..cff15125b 100644
--- a/src/input/mms_input_plugin.c
+++ b/src/input/mms_input_plugin.c
@@ -46,7 +46,9 @@ mms_quark(void)
 }
 
 static struct input_stream *
-input_mms_open(const char *url, GError **error_r)
+input_mms_open(const char *url,
+	       GMutex *mutex, GCond *cond,
+	       GError **error_r)
 {
 	struct input_mms *m;
 
@@ -57,7 +59,8 @@ input_mms_open(const char *url, GError **error_r)
 		return NULL;
 
 	m = g_new(struct input_mms, 1);
-	input_stream_init(&m->base, &input_plugin_mms, url);
+	input_stream_init(&m->base, &input_plugin_mms, url,
+			  mutex, cond);
 
 	m->mms = mmsx_connect(NULL, NULL, url, 128 * 1024);
 	if (m->mms == NULL) {
diff --git a/src/input/rewind_input_plugin.c b/src/input/rewind_input_plugin.c
index 2a3eecf82..cf06fc57b 100644
--- a/src/input/rewind_input_plugin.c
+++ b/src/input/rewind_input_plugin.c
@@ -132,16 +132,12 @@ input_rewind_tag(struct input_stream *is)
 	return input_stream_tag(r->input);
 }
 
-static int
-input_rewind_buffer(struct input_stream *is, GError **error_r)
+static bool
+input_rewind_available(struct input_stream *is)
 {
 	struct input_rewind *r = (struct input_rewind *)is;
 
-	int ret = input_stream_buffer(r->input, error_r);
-	if (ret < 0 || !reading_from_buffer(r))
-		copy_attributes(r);
-
-	return ret;
+	return input_stream_available(r->input);
 }
 
 static size_t
@@ -232,7 +228,7 @@ static const struct input_plugin rewind_input_plugin = {
 	.check = input_rewind_check,
 	.update = input_rewind_update,
 	.tag = input_rewind_tag,
-	.buffer = input_rewind_buffer,
+	.available = input_rewind_available,
 	.read = input_rewind_read,
 	.eof = input_rewind_eof,
 	.seek = input_rewind_seek,
@@ -251,7 +247,8 @@ input_rewind_open(struct input_stream *is)
 		return is;
 
 	c = g_new(struct input_rewind, 1);
-	input_stream_init(&c->base, &rewind_input_plugin, is->uri);
+	input_stream_init(&c->base, &rewind_input_plugin, is->uri,
+			  is->mutex, is->cond);
 	c->tail = 0;
 	c->input = is;
 
diff --git a/src/input/soup_input_plugin.c b/src/input/soup_input_plugin.c
index 23665c1a2..dc005a58c 100644
--- a/src/input/soup_input_plugin.c
+++ b/src/input/soup_input_plugin.c
@@ -46,9 +46,6 @@ static SoupSession *soup_session;
 struct input_soup {
 	struct input_stream base;
 
-	GMutex *mutex;
-	GCond *cond;
-
 	SoupMessage *msg;
 
 	GQueue *buffers;
@@ -124,14 +121,14 @@ input_soup_session_callback(G_GNUC_UNUSED SoupSession *session,
 	assert(msg == s->msg);
 	assert(!s->completed);
 
-	g_mutex_lock(s->mutex);
+	g_mutex_lock(s->base.mutex);
 
 	s->base.ready = true;
 	s->alive = false;
 	s->completed = true;
 
-	g_cond_broadcast(s->cond);
-	g_mutex_unlock(s->mutex);
+	g_cond_broadcast(s->base.cond);
+	g_mutex_unlock(s->base.mutex);
 }
 
 static void
@@ -140,7 +137,7 @@ input_soup_got_headers(SoupMessage *msg, gpointer user_data)
 	struct input_soup *s = user_data;
 
 	if (!SOUP_STATUS_IS_SUCCESSFUL(msg->status_code)) {
-		g_mutex_lock(s->mutex);
+		g_mutex_lock(s->base.mutex);
 
 		if (s->postponed_error == NULL)
 			s->postponed_error =
@@ -148,7 +145,7 @@ input_soup_got_headers(SoupMessage *msg, gpointer user_data)
 					    "got HTTP status %d",
 					    msg->status_code);
 
-		g_mutex_unlock(s->mutex);
+		g_mutex_unlock(s->base.mutex);
 
 		soup_session_cancel_message(soup_session, msg,
 					    SOUP_STATUS_CANCELLED);
@@ -157,10 +154,10 @@ input_soup_got_headers(SoupMessage *msg, gpointer user_data)
 
 	soup_message_body_set_accumulate(msg->response_body, false);
 
-	g_mutex_lock(s->mutex);
+	g_mutex_lock(s->base.mutex);
 	s->base.ready = true;
-	g_cond_broadcast(s->cond);
-	g_mutex_unlock(s->mutex);
+	g_cond_broadcast(s->base.cond);
+	g_mutex_unlock(s->base.mutex);
 }
 
 static void
@@ -170,7 +167,7 @@ input_soup_got_chunk(SoupMessage *msg, SoupBuffer *chunk, gpointer user_data)
 
 	assert(msg == s->msg);
 
-	g_mutex_lock(s->mutex);
+	g_mutex_lock(s->base.mutex);
 
 	g_queue_push_tail(s->buffers, soup_buffer_copy(chunk));
 	s->total_buffered += chunk->length;
@@ -180,8 +177,8 @@ input_soup_got_chunk(SoupMessage *msg, SoupBuffer *chunk, gpointer user_data)
 		soup_session_pause_message(soup_session, msg);
 	}
 
-	g_cond_broadcast(s->cond);
-	g_mutex_unlock(s->mutex);
+	g_cond_broadcast(s->base.cond);
+	g_mutex_unlock(s->base.mutex);
 }
 
 static void
@@ -191,14 +188,14 @@ input_soup_got_body(G_GNUC_UNUSED SoupMessage *msg, gpointer user_data)
 
 	assert(msg == s->msg);
 
-	g_mutex_lock(s->mutex);
+	g_mutex_lock(s->base.mutex);
 
 	s->base.ready = true;
 	s->eof = true;
 	s->alive = false;
 
-	g_cond_broadcast(s->cond);
-	g_mutex_unlock(s->mutex);
+	g_cond_broadcast(s->base.cond);
+	g_mutex_unlock(s->base.mutex);
 }
 
 static bool
@@ -216,7 +213,7 @@ input_soup_wait_data(struct input_soup *s)
 
 		assert(s->current_consumed == 0);
 
-		g_cond_wait(s->cond, s->mutex);
+		g_cond_wait(s->base.cond, s->base.mutex);
 	}
 }
 
@@ -232,16 +229,16 @@ input_soup_queue(gpointer data)
 }
 
 static struct input_stream *
-input_soup_open(const char *uri, G_GNUC_UNUSED GError **error_r)
+input_soup_open(const char *uri,
+		GMutex *mutex, GCond *cond,
+		G_GNUC_UNUSED GError **error_r)
 {
 	if (strncmp(uri, "http://", 7) != 0)
 		return NULL;
 
 	struct input_soup *s = g_new(struct input_soup, 1);
-	input_stream_init(&s->base, &input_plugin_soup, uri);
-
-	s->mutex = g_mutex_new();
-	s->cond = g_cond_new();
+	input_stream_init(&s->base, &input_plugin_soup, uri,
+			  mutex, cond);
 
 	s->buffers = g_queue_new();
 	s->current_consumed = 0;
@@ -288,25 +285,22 @@ input_soup_close(struct input_stream *is)
 {
 	struct input_soup *s = (struct input_soup *)is;
 
-	g_mutex_lock(s->mutex);
+	g_mutex_lock(s->base.mutex);
 
 	if (!s->completed) {
 		/* the messages's session callback hasn't been invoked
 		   yet; cancel it and wait for completion */
 
-		g_mutex_unlock(s->mutex);
+		g_mutex_unlock(s->base.mutex);
 
 		io_thread_call(input_soup_cancel, s);
 
-		g_mutex_lock(s->mutex);
+		g_mutex_lock(s->base.mutex);
 		while (!s->completed)
-			g_cond_wait(s->cond, s->mutex);
+			g_cond_wait(s->base.cond, s->base.mutex);
 	}
 
-	g_mutex_unlock(s->mutex);
-
-	g_mutex_free(s->mutex);
-	g_cond_free(s->cond);
+	g_mutex_unlock(s->base.mutex);
 
 	SoupBuffer *buffer;
 	while ((buffer = g_queue_pop_head(s->buffers)) != NULL)
@@ -325,54 +319,21 @@ input_soup_check(struct input_stream *is, GError **error_r)
 {
 	struct input_soup *s = (struct input_soup *)is;
 
-	g_mutex_lock(s->mutex);
-
 	bool success = s->postponed_error == NULL;
 	if (!success) {
 		g_propagate_error(error_r, s->postponed_error);
 		s->postponed_error = NULL;
 	}
 
-	g_mutex_unlock(s->mutex);
 	return success;
 }
 
-static int
-input_soup_buffer(struct input_stream *is, GError **error_r)
+static bool
+input_soup_available(struct input_stream *is)
 {
 	struct input_soup *s = (struct input_soup *)is;
 
-	g_mutex_lock(s->mutex);
-
-	if (s->pause) {
-		if (s->total_buffered >= SOUP_MAX_BUFFERED) {
-			g_mutex_unlock(s->mutex);
-			return 1;
-		}
-
-		s->pause = false;
-		soup_session_unpause_message(soup_session, s->msg);
-	}
-
-
-	bool success = input_soup_wait_data(s);
-
-	if (!success) {
-		if (s->postponed_error != NULL) {
-			g_propagate_error(error_r, s->postponed_error);
-			s->postponed_error = NULL;
-		} else
-			g_set_error_literal(error_r, soup_quark(), 0,
-					    "HTTP failure");
-	}
-
-	g_mutex_unlock(s->mutex);
-
-	if (!success) {
-		return -1;
-	}
-
-	return 1;
+	return s->eof || !s->alive || !g_queue_is_empty(s->buffers);
 }
 
 static size_t
@@ -381,8 +342,6 @@ input_soup_read(struct input_stream *is, void *ptr, size_t size,
 {
 	struct input_soup *s = (struct input_soup *)is;
 
-	g_mutex_lock(s->mutex);
-
 	if (!input_soup_wait_data(s)) {
 		assert(!s->alive);
 
@@ -392,8 +351,6 @@ input_soup_read(struct input_stream *is, void *ptr, size_t size,
 		} else
 			g_set_error_literal(error_r, soup_quark(), 0,
 					    "HTTP failure");
-
-		g_mutex_unlock(s->mutex);
 		return 0;
 	}
 
@@ -442,7 +399,6 @@ input_soup_read(struct input_stream *is, void *ptr, size_t size,
 	size_t nbytes = p - p0;
 	s->base.offset += nbytes;
 
-	g_mutex_unlock(s->mutex);
 	return nbytes;
 }
 
@@ -462,7 +418,7 @@ const struct input_plugin input_plugin_soup = {
 	.open = input_soup_open,
 	.close = input_soup_close,
 	.check = input_soup_check,
-	.buffer = input_soup_buffer,
+	.available = input_soup_available,
 	.read = input_soup_read,
 	.eof = input_soup_eof,
 };
diff --git a/src/input_internal.c b/src/input_internal.c
index 4d675fc97..92a71856e 100644
--- a/src/input_internal.c
+++ b/src/input_internal.c
@@ -25,7 +25,7 @@
 
 void
 input_stream_init(struct input_stream *is, const struct input_plugin *plugin,
-		  const char *uri)
+		  const char *uri, GMutex *mutex, GCond *cond)
 {
 	assert(is != NULL);
 	assert(plugin != NULL);
@@ -33,6 +33,8 @@ input_stream_init(struct input_stream *is, const struct input_plugin *plugin,
 
 	is->plugin = plugin;
 	is->uri = g_strdup(uri);
+	is->mutex = mutex;
+	is->cond = cond;
 	is->ready = false;
 	is->seekable = false;
 	is->size = -1;
@@ -49,3 +51,23 @@ input_stream_deinit(struct input_stream *is)
 	g_free(is->uri);
 	g_free(is->mime);
 }
+
+void
+input_stream_signal_client(struct input_stream *is)
+{
+	if (is->cond != NULL)
+		g_cond_broadcast(is->cond);
+}
+
+void
+input_stream_set_ready(struct input_stream *is)
+{
+	g_mutex_lock(is->mutex);
+
+	if (!is->ready) {
+		is->ready = true;
+		input_stream_signal_client(is);
+	}
+
+	g_mutex_unlock(is->mutex);
+}
diff --git a/src/input_internal.h b/src/input_internal.h
index 260ea12a6..d95142e46 100644
--- a/src/input_internal.h
+++ b/src/input_internal.h
@@ -22,14 +22,22 @@
 
 #include "check.h"
 
+#include <glib.h>
+
 struct input_stream;
 struct input_plugin;
 
 void
 input_stream_init(struct input_stream *is, const struct input_plugin *plugin,
-		  const char *uri);
+		  const char *uri, GMutex *mutex, GCond *cond);
 
 void
 input_stream_deinit(struct input_stream *is);
 
+void
+input_stream_signal_client(struct input_stream *is);
+
+void
+input_stream_set_ready(struct input_stream *is);
+
 #endif
diff --git a/src/input_plugin.h b/src/input_plugin.h
index 4e0993d12..6b0c77c85 100644
--- a/src/input_plugin.h
+++ b/src/input_plugin.h
@@ -48,7 +48,9 @@ struct input_plugin {
 	 */
 	void (*finish)(void);
 
-	struct input_stream *(*open)(const char *uri, GError **error_r);
+	struct input_stream *(*open)(const char *uri,
+				     GMutex *mutex, GCond *cond,
+				     GError **error_r);
 	void (*close)(struct input_stream *is);
 
 	/**
@@ -66,7 +68,17 @@ struct input_plugin {
 	void (*update)(struct input_stream *is);
 
 	struct tag *(*tag)(struct input_stream *is);
-	int (*buffer)(struct input_stream *is, GError **error_r);
+
+	/**
+	 * Returns true if the next read operation will not block:
+	 * either data is available, or end-of-stream has been
+	 * reached, or an error has occurred.
+	 *
+	 * If this method is unimplemented, then it is assumed that
+	 * reading will never block.
+	 */
+	bool (*available)(struct input_stream *is);
+
 	size_t (*read)(struct input_stream *is, void *ptr, size_t size,
 		       GError **error_r);
 	bool (*eof)(struct input_stream *is);
diff --git a/src/input_stream.c b/src/input_stream.c
index 164df9860..60a1559ba 100644
--- a/src/input_stream.c
+++ b/src/input_stream.c
@@ -33,10 +33,13 @@ input_quark(void)
 }
 
 struct input_stream *
-input_stream_open(const char *url, GError **error_r)
+input_stream_open(const char *url,
+		  GMutex *mutex, GCond *cond,
+		  GError **error_r)
 {
 	GError *error = NULL;
 
+	assert(mutex != NULL);
 	assert(error_r == NULL || *error_r == NULL);
 
 	for (unsigned i = 0; input_plugins[i] != NULL; ++i) {
@@ -46,7 +49,7 @@ input_stream_open(const char *url, GError **error_r)
 		if (!input_plugins_enabled[i])
 			continue;
 
-		is = plugin->open(url, &error);
+		is = plugin->open(url, mutex, cond, &error);
 		if (is != NULL) {
 			assert(is->plugin != NULL);
 			assert(is->plugin->close != NULL);
@@ -87,26 +90,108 @@ input_stream_update(struct input_stream *is)
 		is->plugin->update(is);
 }
 
+void
+input_stream_wait_ready(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->mutex != NULL);
+	assert(is->cond != NULL);
+
+	while (true) {
+		input_stream_update(is);
+		if (is->ready)
+			break;
+
+		g_cond_wait(is->cond, is->mutex);
+	}
+}
+
+void
+input_stream_lock_wait_ready(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->mutex != NULL);
+	assert(is->cond != NULL);
+
+	g_mutex_lock(is->mutex);
+	input_stream_wait_ready(is);
+	g_mutex_unlock(is->mutex);
+}
+
 bool
 input_stream_seek(struct input_stream *is, goffset offset, int whence,
 		  GError **error_r)
 {
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
 	if (is->plugin->seek == NULL)
 		return false;
 
 	return is->plugin->seek(is, offset, whence, error_r);
 }
 
+bool
+input_stream_lock_seek(struct input_stream *is, goffset offset, int whence,
+		       GError **error_r)
+{
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	if (is->plugin->seek == NULL)
+		return false;
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_seek(is, offset, whence, error_r);
+
+	g_mutex_lock(is->mutex);
+	bool success = input_stream_seek(is, offset, whence, error_r);
+	g_mutex_unlock(is->mutex);
+	return success;
+}
+
 struct tag *
 input_stream_tag(struct input_stream *is)
 {
 	assert(is != NULL);
+	assert(is->plugin != NULL);
 
 	return is->plugin->tag != NULL
 		? is->plugin->tag(is)
 		: NULL;
 }
 
+struct tag *
+input_stream_lock_tag(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	if (is->plugin->tag == NULL)
+		return false;
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_tag(is);
+
+	g_mutex_lock(is->mutex);
+	struct tag *tag = input_stream_tag(is);
+	g_mutex_unlock(is->mutex);
+	return tag;
+}
+
+bool
+input_stream_available(struct input_stream *is)
+{
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	return is->plugin->available != NULL
+		? is->plugin->available(is)
+		: true;
+}
+
 size_t
 input_stream_read(struct input_stream *is, void *ptr, size_t size,
 		  GError **error_r)
@@ -117,6 +202,23 @@ input_stream_read(struct input_stream *is, void *ptr, size_t size,
 	return is->plugin->read(is, ptr, size, error_r);
 }
 
+size_t
+input_stream_lock_read(struct input_stream *is, void *ptr, size_t size,
+		       GError **error_r)
+{
+	assert(ptr != NULL);
+	assert(size > 0);
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_read(is, ptr, size, error_r);
+
+	g_mutex_lock(is->mutex);
+	size_t nbytes = input_stream_read(is, ptr, size, error_r);
+	g_mutex_unlock(is->mutex);
+	return nbytes;
+}
+
 void input_stream_close(struct input_stream *is)
 {
 	is->plugin->close(is);
@@ -127,11 +229,19 @@ bool input_stream_eof(struct input_stream *is)
 	return is->plugin->eof(is);
 }
 
-int
-input_stream_buffer(struct input_stream *is, GError **error_r)
+bool
+input_stream_lock_eof(struct input_stream *is)
 {
-	if (is->plugin->buffer == NULL)
-		return 0;
+	assert(is != NULL);
+	assert(is->plugin != NULL);
+
+	if (is->mutex == NULL)
+		/* no locking */
+		return input_stream_eof(is);
 
-	return is->plugin->buffer(is, error_r);
+	g_mutex_lock(is->mutex);
+	bool eof = input_stream_eof(is);
+	g_mutex_unlock(is->mutex);
+	return eof;
 }
+
diff --git a/src/input_stream.h b/src/input_stream.h
index 7866562ae..6a10831d2 100644
--- a/src/input_stream.h
+++ b/src/input_stream.h
@@ -46,6 +46,26 @@ struct input_stream {
 	char *uri;
 
 	/**
+	 * A mutex that protects the mutable attributes of this object
+	 * and its implementation.  It must be locked before calling
+	 * any of the public methods.
+	 *
+	 * This object is allocated by the client, and the client is
+	 * responsible for freeing it.
+	 */
+	GMutex *mutex;
+
+	/**
+	 * A cond that gets signalled when the state of this object
+	 * changes from the I/O thread.  The client of this object may
+	 * wait on it.  Optional, may be NULL.
+	 *
+	 * This object is allocated by the client, and the client is
+	 * responsible for freeing it.
+	 */
+	GCond *cond;
+
+	/**
 	 * indicates whether the stream is ready for reading and
 	 * whether the other attributes in this struct are valid
 	 */
@@ -76,20 +96,43 @@ struct input_stream {
  * Opens a new input stream.  You may not access it until the "ready"
  * flag is set.
  *
+ * @param mutex a mutex that is used to protect this object; must be
+ * locked before calling any of the public methods
+ * @param cond a cond that gets signalled when the state of
+ * this object changes; may be NULL if the caller doesn't want to get
+ * notifications
  * @return an #input_stream object on success, NULL on error
  */
-gcc_nonnull(1)
+gcc_nonnull(1, 2)
 G_GNUC_MALLOC
 struct input_stream *
-input_stream_open(const char *uri, GError **error_r);
+input_stream_open(const char *uri,
+		  GMutex *mutex, GCond *cond,
+		  GError **error_r);
 
 /**
  * Close the input stream and free resources.
+ *
+ * The caller must not lock the mutex.
  */
 gcc_nonnull(1)
 void
 input_stream_close(struct input_stream *is);
 
+gcc_nonnull(1)
+static inline void
+input_stream_lock(struct input_stream *is)
+{
+	g_mutex_lock(is->mutex);
+}
+
+gcc_nonnull(1)
+static inline void
+input_stream_unlock(struct input_stream *is)
+{
+	g_mutex_unlock(is->mutex);
+}
+
 /**
  * Check for errors that may have occurred in the I/O thread.
  *
@@ -108,9 +151,28 @@ void
 input_stream_update(struct input_stream *is);
 
 /**
+ * Wait until the stream becomes ready.
+ *
+ * The caller must lock the mutex.
+ */
+gcc_nonnull(1)
+void
+input_stream_wait_ready(struct input_stream *is);
+
+/**
+ * Wrapper for input_stream_wait_locked() which locks and unlocks the
+ * mutex; the caller must not be holding it already.
+ */
+gcc_nonnull(1)
+void
+input_stream_lock_wait_ready(struct input_stream *is);
+
+/**
  * Seeks to the specified position in the stream.  This will most
  * likely fail if the "seekable" flag is false.
  *
+ * The caller must lock the mutex.
+ *
  * @param is the input_stream object
  * @param offset the relative offset
  * @param whence the base of the seek, one of SEEK_SET, SEEK_CUR, SEEK_END
@@ -121,15 +183,37 @@ input_stream_seek(struct input_stream *is, goffset offset, int whence,
 		  GError **error_r);
 
 /**
+ * Wrapper for input_stream_seek() which locks and unlocks the
+ * mutex; the caller must not be holding it already.
+ */
+gcc_nonnull(1)
+bool
+input_stream_lock_seek(struct input_stream *is, goffset offset, int whence,
+		       GError **error_r);
+
+/**
  * Returns true if the stream has reached end-of-file.
+ *
+ * The caller must lock the mutex.
  */
 gcc_nonnull(1)
 G_GNUC_PURE
 bool input_stream_eof(struct input_stream *is);
 
 /**
+ * Wrapper for input_stream_eof() which locks and unlocks the mutex;
+ * the caller must not be holding it already.
+ */
+gcc_nonnull(1)
+G_GNUC_PURE
+bool
+input_stream_lock_eof(struct input_stream *is);
+
+/**
  * Reads the tag from the stream.
  *
+ * The caller must lock the mutex.
+ *
  * @return a tag object which must be freed with tag_free(), or NULL
  * if the tag has not changed since the last call
  */
@@ -139,20 +223,32 @@ struct tag *
 input_stream_tag(struct input_stream *is);
 
 /**
- * Reads some of the stream into its buffer.  The following return
- * codes are defined: -1 = error, 1 = something was buffered, 0 =
- * nothing was buffered.
+ * Wrapper for input_stream_tag() which locks and unlocks the
+ * mutex; the caller must not be holding it already.
+ */
+gcc_nonnull(1)
+G_GNUC_MALLOC
+struct tag *
+input_stream_lock_tag(struct input_stream *is);
+
+/**
+ * Returns true if the next read operation will not block: either data
+ * is available, or end-of-stream has been reached, or an error has
+ * occurred.
  *
- * The semantics of this function are not well-defined, and it will
- * eventually be removed.
+ * The caller must lock the mutex.
  */
 gcc_nonnull(1)
-int input_stream_buffer(struct input_stream *is, GError **error_r);
+G_GNUC_PURE
+bool
+input_stream_available(struct input_stream *is);
 
 /**
  * Reads data from the stream into the caller-supplied buffer.
  * Returns 0 on error or eof (check with input_stream_eof()).
  *
+ * The caller must lock the mutex.
+ *
  * @param is the input_stream object
  * @param ptr the buffer to read into
  * @param size the maximum number of bytes to read
@@ -163,4 +259,13 @@ size_t
 input_stream_read(struct input_stream *is, void *ptr, size_t size,
 		  GError **error_r);
 
+/**
+ * Wrapper for input_stream_tag() which locks and unlocks the
+ * mutex; the caller must not be holding it already.
+ */
+gcc_nonnull(1, 2)
+size_t
+input_stream_lock_read(struct input_stream *is, void *ptr, size_t size,
+		       GError **error_r);
+
 #endif