Merge branch 'v0.16.x'

5 files changed, 127 insertions, 6 deletions

diff --git a/NEWS b/NEWS
index 0698f5d1c..8d9c335a6 100644
--- a/NEWS
+++ b/NEWS
@@ -23,6 +23,8 @@ ver 0.16.4 (2011/??/??)
 * fix memory leaks
 * don't resume playback when seeking to another song while paused
 * apply follow_inside_symlinks to absolute symlinks
+* input:
+  - curl: limit the receive buffer size
 * decoder:
   - ffmpeg: workaround for semantic API change in recent ffmpeg versions
   - flac: validate the sample rate when scanning the tag
@@ -31,6 +33,9 @@ ver 0.16.4 (2011/??/??)
   - vorbis: don't send end-of-stream on flush
 * output:
   - alsa: fix SIGFPE when alsa announces a period size of 0
+  - httpd: don't warn on client disconnect
+  - pulse: fix deadlock when resuming the stream
+  - pulse: fix deadlock when the stream was suspended
 
 
 ver 0.16.3 (2011/06/04)
diff --git a/src/decoder/mpcdec_decoder_plugin.c b/src/decoder/mpcdec_decoder_plugin.c
index d28c22305..f31dcdb99 100644
--- a/src/decoder/mpcdec_decoder_plugin.c
+++ b/src/decoder/mpcdec_decoder_plugin.c
@@ -153,7 +153,6 @@ mpcdec_decode(struct decoder *mpd_decoder, struct input_stream *is)
 	mpc_uint32_t ret;
 	int32_t chunk[G_N_ELEMENTS(sample_buffer)];
 	long bit_rate = 0;
-	mpc_uint32_t vbr_update_acc;
 	mpc_uint32_t vbr_update_bits;
 	enum decoder_command cmd = DECODE_COMMAND_NONE;
 
@@ -243,10 +242,11 @@ mpcdec_decode(struct decoder *mpd_decoder, struct input_stream *is)
 				decoder_seek_error(mpd_decoder);
 		}
 
-		vbr_update_acc = 0;
 		vbr_update_bits = 0;
 
 #ifdef MPC_IS_OLD_API
+		mpc_uint32_t vbr_update_acc = 0;
+
 		ret = mpc_decoder_decode(&decoder, sample_buffer,
 					 &vbr_update_acc, &vbr_update_bits);
 		if (ret == 0 || ret == (mpc_uint32_t)-1)
diff --git a/src/input/curl_input_plugin.c b/src/input/curl_input_plugin.c
index 09010c220..20e8dea07 100644
--- a/src/input/curl_input_plugin.c
+++ b/src/input/curl_input_plugin.c
@@ -43,6 +43,13 @@
 #define G_LOG_DOMAIN "input_curl"
 
 /**
+ * Do not buffer more than this number of bytes.  It should be a
+ * reasonable limit that doesn't make low-end machines suffer too
+ * much, but doesn't cause stuttering on high-latency lines.
+ */
+static const size_t CURL_MAX_BUFFERED = 512 * 1024;
+
+/**
  * Buffers created by input_curl_writefunction().
  */
 struct buffer {
@@ -144,6 +151,25 @@ input_curl_finish(void)
 	curl_global_cleanup();
 }
 
+/**
+ * Determine the total sizes of all buffers, including portions that
+ * have already been consumed.
+ */
+G_GNUC_PURE
+static size_t
+curl_total_buffer_size(const struct input_curl *c)
+{
+	size_t total = 0;
+
+	for (GList *i = g_queue_peek_head_link(c->buffers);
+	     i != NULL; i = g_list_next(i)) {
+		struct buffer *buffer = i->data;
+		total += buffer->size;
+	}
+
+	return total;
+}
+
 static void
 buffer_free_callback(gpointer data, G_GNUC_UNUSED gpointer user_data)
 {
@@ -473,6 +499,10 @@ static int
 input_curl_buffer(struct input_stream *is, GError **error_r)
 {
 	struct input_curl *c = (struct input_curl *)is;
+
+	if (curl_total_buffer_size(c) >= CURL_MAX_BUFFERED)
+		return 0;
+
 	CURLMcode mcode;
 	int running_handles;
 	bool ret;
diff --git a/src/output/httpd_client.c b/src/output/httpd_client.c
index 32f355a43..e2c49c6c8 100644
--- a/src/output/httpd_client.c
+++ b/src/output/httpd_client.c
@@ -143,6 +143,8 @@ httpd_client_unref_page(gpointer data, G_GNUC_UNUSED gpointer user_data)
 void
 httpd_client_free(struct httpd_client *client)
 {
+	assert(client != NULL);
+
 	if (client->state == RESPONSE) {
 		if (client->write_source_id != 0)
 			g_source_remove(client->write_source_id);
@@ -169,6 +171,8 @@ httpd_client_free(struct httpd_client *client)
 static void
 httpd_client_close(struct httpd_client *client)
 {
+	assert(client != NULL);
+
 	httpd_output_remove_client(client->httpd, client);
 	httpd_client_free(client);
 }
@@ -179,6 +183,9 @@ httpd_client_close(struct httpd_client *client)
 static void
 httpd_client_begin_response(struct httpd_client *client)
 {
+	assert(client != NULL);
+	assert(client->state != RESPONSE);
+
 	client->state = RESPONSE;
 	client->write_source_id = 0;
 	client->pages = g_queue_new();
@@ -239,6 +246,9 @@ httpd_client_handle_line(struct httpd_client *client, const char *line)
 static char *
 httpd_client_read_line(struct httpd_client *client)
 {
+	assert(client != NULL);
+	assert(client->state != RESPONSE);
+
 	const char *p, *newline;
 	size_t length;
 	char *line;
@@ -271,6 +281,7 @@ httpd_client_send_response(struct httpd_client *client)
 	GIOStatus status;
 	gsize bytes_written;
 
+	assert(client != NULL);
 	assert(client->state == RESPONSE);
 
 	if (!client->metadata_requested) {
@@ -334,14 +345,19 @@ httpd_client_send_response(struct httpd_client *client)
 static bool
 httpd_client_received(struct httpd_client *client)
 {
+	assert(client != NULL);
+	assert(client->state != RESPONSE);
+
 	char *line;
 	bool success;
 
 	while ((line = httpd_client_read_line(client)) != NULL) {
 		success = httpd_client_handle_line(client, line);
 		g_free(line);
-		if (!success)
+		if (!success) {
+			assert(client->state != RESPONSE);
 			return false;
+		}
 
 		if (client->state == RESPONSE) {
 			if (!fifo_buffer_is_empty(client->input)) {
@@ -370,7 +386,14 @@ httpd_client_read(struct httpd_client *client)
 	if (client->state == RESPONSE) {
 		/* the client has already sent the request, and he
 		   must not send more */
-		g_warning("unexpected input from client");
+		char buffer[1];
+
+		status = g_io_channel_read_chars(client->channel, buffer,
+						 sizeof(buffer), &bytes_read,
+						 NULL);
+		if (status == G_IO_STATUS_NORMAL)
+			g_warning("unexpected input from client");
+
 		return false;
 	}
 
diff --git a/src/output/pulse_output_plugin.c b/src/output/pulse_output_plugin.c
index 70cfb9784..9fb0b3ade 100644
--- a/src/output/pulse_output_plugin.c
+++ b/src/output/pulse_output_plugin.c
@@ -207,6 +207,9 @@ pulse_output_subscribe_cb(pa_context *context,
 static bool
 pulse_output_connect(struct pulse_output *po, GError **error_r)
 {
+	assert(po != NULL);
+	assert(po->context != NULL);
+
 	int error;
 
 	error = pa_context_connect(po->context, po->server,
@@ -229,6 +232,9 @@ pulse_output_connect(struct pulse_output *po, GError **error_r)
 static bool
 pulse_output_setup_context(struct pulse_output *po, GError **error_r)
 {
+	assert(po != NULL);
+	assert(po->mainloop != NULL);
+
 	po->context = pa_context_new(pa_threaded_mainloop_get_api(po->mainloop),
 				     MPD_PULSE_NAME);
 	if (po->context == NULL) {
@@ -257,6 +263,9 @@ pulse_output_setup_context(struct pulse_output *po, GError **error_r)
 static void
 pulse_output_delete_context(struct pulse_output *po)
 {
+	assert(po != NULL);
+	assert(po->context != NULL);
+
 	pa_context_disconnect(po->context);
 	pa_context_unref(po->context);
 	po->context = NULL;
@@ -347,6 +356,8 @@ pulse_output_disable(void *data)
 {
 	struct pulse_output *po = data;
 
+	assert(po->mainloop != NULL);
+
 	pa_threaded_mainloop_stop(po->mainloop);
 	if (po->context != NULL)
 		pulse_output_delete_context(po);
@@ -363,6 +374,8 @@ pulse_output_disable(void *data)
 static bool
 pulse_output_wait_connection(struct pulse_output *po, GError **error_r)
 {
+	assert(po->mainloop != NULL);
+
 	pa_context_state_t state;
 
 	pa_threaded_mainloop_lock(po->mainloop);
@@ -399,11 +412,32 @@ pulse_output_wait_connection(struct pulse_output *po, GError **error_r)
 	}
 }
 
+#if PA_CHECK_VERSION(0,9,8)
+
+static void
+pulse_output_stream_suspended_cb(G_GNUC_UNUSED pa_stream *stream, void *userdata)
+{
+	struct pulse_output *po = userdata;
+
+	assert(stream == po->stream || po->stream == NULL);
+	assert(po->mainloop != NULL);
+
+	/* wake up the main loop to break out of the loop in
+	   pulse_output_play() */
+	pa_threaded_mainloop_signal(po->mainloop, 0);
+}
+
+#endif
+
 static void
 pulse_output_stream_state_cb(pa_stream *stream, void *userdata)
 {
 	struct pulse_output *po = userdata;
 
+	assert(stream == po->stream || po->stream == NULL);
+	assert(po->mainloop != NULL);
+	assert(po->context != NULL);
+
 	switch (pa_stream_get_state(stream)) {
 	case PA_STREAM_READY:
 		if (po->mixer != NULL)
@@ -432,6 +466,8 @@ pulse_output_stream_write_cb(G_GNUC_UNUSED pa_stream *stream, size_t nbytes,
 {
 	struct pulse_output *po = userdata;
 
+	assert(po->mainloop != NULL);
+
 	po->writable = nbytes;
 	pa_threaded_mainloop_signal(po->mainloop, 0);
 }
@@ -444,6 +480,8 @@ pulse_output_open(void *data, struct audio_format *audio_format,
 	pa_sample_spec ss;
 	int error;
 
+	assert(po->mainloop != NULL);
+
 	if (po->context != NULL) {
 		switch (pa_context_get_state(po->context)) {
 		case PA_CONTEXT_UNCONNECTED:
@@ -487,6 +525,11 @@ pulse_output_open(void *data, struct audio_format *audio_format,
 		return false;
 	}
 
+#if PA_CHECK_VERSION(0,9,8)
+	pa_stream_set_suspended_callback(po->stream,
+					 pulse_output_stream_suspended_cb, po);
+#endif
+
 	pa_stream_set_state_callback(po->stream,
 				     pulse_output_stream_state_cb, po);
 	pa_stream_set_write_callback(po->stream,
@@ -522,6 +565,8 @@ pulse_output_close(void *data)
 	struct pulse_output *po = data;
 	pa_operation *o;
 
+	assert(po->mainloop != NULL);
+
 	pa_threaded_mainloop_lock(po->mainloop);
 
 	if (pa_stream_get_state(po->stream) == PA_STREAM_READY) {
@@ -556,6 +601,8 @@ pulse_output_check_stream(struct pulse_output *po)
 {
 	pa_stream_state_t state = pa_stream_get_state(po->stream);
 
+	assert(po->mainloop != NULL);
+
 	switch (state) {
 	case PA_STREAM_READY:
 	case PA_STREAM_FAILED:
@@ -637,6 +684,8 @@ pulse_output_stream_pause(struct pulse_output *po, bool pause,
 {
 	pa_operation *o;
 
+	assert(po->mainloop != NULL);
+	assert(po->context != NULL);
 	assert(po->stream != NULL);
 
 	o = pa_stream_cork(po->stream, pause,
@@ -667,6 +716,7 @@ pulse_output_play(void *data, const void *chunk, size_t size, GError **error_r)
 	struct pulse_output *po = data;
 	int error;
 
+	assert(po->mainloop != NULL);
 	assert(po->stream != NULL);
 
 	pa_threaded_mainloop_lock(po->mainloop);
@@ -683,19 +733,30 @@ pulse_output_play(void *data, const void *chunk, size_t size, GError **error_r)
 	/* unpause if previously paused */
 
 	if (pulse_output_stream_is_paused(po) &&
-	    !pulse_output_stream_pause(po, false, error_r))
+	    !pulse_output_stream_pause(po, false, error_r)) {
+		pa_threaded_mainloop_unlock(po->mainloop);
 		return 0;
+	}
 
 	/* wait until the server allows us to write */
 
 	while (po->writable == 0) {
+#if PA_CHECK_VERSION(0,9,8)
+		if (pa_stream_is_suspended(po->stream)) {
+			pa_threaded_mainloop_unlock(po->mainloop);
+			g_set_error(error_r, pulse_output_quark(), 0,
+				    "suspended");
+			return 0;
+		}
+#endif
+
 		pa_threaded_mainloop_wait(po->mainloop);
 
 		if (pa_stream_get_state(po->stream) != PA_STREAM_READY) {
 			pa_threaded_mainloop_unlock(po->mainloop);
 			g_set_error(error_r, pulse_output_quark(), 0,
 				    "disconnected");
-			return false;
+			return 0;
 		}
 	}
 
@@ -725,6 +786,7 @@ pulse_output_cancel(void *data)
 	struct pulse_output *po = data;
 	pa_operation *o;
 
+	assert(po->mainloop != NULL);
 	assert(po->stream != NULL);
 
 	pa_threaded_mainloop_lock(po->mainloop);
@@ -756,6 +818,7 @@ pulse_output_pause(void *data)
 	struct pulse_output *po = data;
 	GError *error = NULL;
 
+	assert(po->mainloop != NULL);
 	assert(po->stream != NULL);
 
 	pa_threaded_mainloop_lock(po->mainloop);