music_pipe: added music_pipe_push()

Added music_pipe_allocate(), music_pipe_push() and
music_pipe_cancel().  Those functions allow the caller (decoder thread
in this case) to do its own chunk management.  The functions
music_pipe_flush() and music_pipe_tag() can now be removed.

Makefile.am

@@ -164,6 +164,7 @@ src_mpd_SOURCES = \
 	src/decoder_thread.c \
 	src/decoder_control.c \
 	src/decoder_api.c \
+	src/decoder_internal.c \
 	src/directory.c \
 	src/directory_save.c \
 	src/directory_print.c \

src/decoder_api.c

@@ -26,6 +26,7 @@
 
 #include "normalize.h"
 #include "pipe.h"
+#include "chunk.h"
 
 #include <glib.h>
 
@@ -85,9 +86,16 @@ void decoder_command_finished(G_GNUC_UNUSED struct decoder * decoder)
 	assert(dc.command != DECODE_COMMAND_SEEK ||
 	       dc.seek_error || decoder->seeking);
 
-	if (dc.command == DECODE_COMMAND_SEEK)
+	if (dc.command == DECODE_COMMAND_SEEK) {
 		/* delete frames from the old song position */
+
+		if (decoder->chunk != NULL) {
+			music_pipe_cancel(decoder->chunk);
+			decoder->chunk = NULL;
+		}
+
 		music_pipe_clear();
+	}
 
 	dc.command = DECODE_COMMAND_NONE;
 	notify_signal(&pc.notify);
@@ -147,35 +155,28 @@ size_t decoder_read(struct decoder *decoder,
 }
 
 /**
- * All chunks are full of decoded data; wait for the player to free
+ * Sends a #tag as-is to the music pipe.  Flushes the current chunk
- * one.
+ * (decoder.chunk) if there is one.
  */
 static enum decoder_command
-need_chunks(struct input_stream *is, bool do_wait)
+do_send_tag(struct decoder *decoder, struct input_stream *is,
+	    const struct tag *tag)
 {
-	if (dc.command == DECODE_COMMAND_STOP ||
+	struct music_chunk *chunk;
-	    dc.command == DECODE_COMMAND_SEEK)
-		return dc.command;
 
-	if ((is == NULL || input_stream_buffer(is) <= 0) && do_wait) {
+	if (decoder->chunk != NULL) {
-		notify_wait(&dc.notify);
+		/* there is a partial chunk - flush it, we want the
-		notify_signal(&pc.notify);
+		   tag in a new chunk */
-
+		enum decoder_command cmd =
-		return dc.command;
+			decoder_flush_chunk(decoder, is);
-	}
-
-	return DECODE_COMMAND_NONE;
-}
-
-static enum decoder_command
-do_send_tag(struct input_stream *is, const struct tag *tag)
-{
-	while (!music_pipe_tag(tag)) {
-		enum decoder_command cmd = need_chunks(is, true);
 		if (cmd != DECODE_COMMAND_NONE)
 			return cmd;
 	}
 
+	assert(decoder->chunk == NULL);
+
+	chunk = decoder_get_chunk(decoder);
+	chunk->tag = tag_dup(tag);
 	return DECODE_COMMAND_NONE;
 }
 
@@ -226,11 +227,11 @@ decoder_data(struct decoder *decoder,
 
 			tag = tag_merge(decoder->stream_tag,
 					decoder->decoder_tag);
-			cmd = do_send_tag(is, tag);
+			cmd = do_send_tag(decoder, is, tag);
 			tag_free(tag);
 		} else
 			/* send only the stream tag */
-			cmd = do_send_tag(is, decoder->stream_tag);
+			cmd = do_send_tag(decoder, is, decoder->stream_tag);
 
 		if (cmd != DECODE_COMMAND_NONE)
 			return cmd;
@@ -250,14 +251,18 @@ decoder_data(struct decoder *decoder,
 	}
 
 	while (length > 0) {
+		struct music_chunk *chunk;
+		char *dest;
 		size_t nbytes;
-		char *dest = music_pipe_write(&dc.out_audio_format,
+		bool full;
-					      data_time, bitRate,
+
-					      &nbytes);
+		chunk = decoder_get_chunk(decoder);
+		dest = music_chunk_write(chunk, &dc.out_audio_format,
+					 data_time, bitRate, &nbytes);
 		if (dest == NULL) {
-			/* the music pipe is full: wait for more
+			/* the chunk is full, flush it */
-			   room */
+			enum decoder_command cmd =
-			enum decoder_command cmd = need_chunks(is, true);
+				decoder_flush_chunk(decoder, is);
 			if (cmd != DECODE_COMMAND_NONE)
 				return cmd;
 			continue;
@@ -283,7 +288,14 @@ decoder_data(struct decoder *decoder,
 
 		/* expand the music pipe chunk */
 
-		music_pipe_expand(&dc.out_audio_format, nbytes);
+		full = music_chunk_expand(chunk, &dc.out_audio_format, nbytes);
+		if (full) {
+			/* the chunk is full, flush it */
+			enum decoder_command cmd =
+				decoder_flush_chunk(decoder, is);
+			if (cmd != DECODE_COMMAND_NONE)
+				return cmd;
+		}
 
 		data += nbytes;
 		length -= nbytes;
@@ -318,11 +330,11 @@ decoder_tag(G_GNUC_UNUSED struct decoder *decoder, struct input_stream *is,
 		struct tag *merged;
 
 		merged = tag_merge(decoder->stream_tag, decoder->decoder_tag);
-		cmd = do_send_tag(is, merged);
+		cmd = do_send_tag(decoder, is, merged);
 		tag_free(merged);
 	} else
 		/* send only the decoder tag */
-		cmd = do_send_tag(is, tag);
+		cmd = do_send_tag(decoder, is, tag);
 
 	return cmd;
 }

src/decoder_internal.c

@@ -0,0 +1,80 @@
+/*
+ * Copyright (C) 2003-2009 The Music Player Daemon Project
+ * 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 "decoder_internal.h"
+#include "decoder_control.h"
+#include "player_control.h"
+#include "pipe.h"
+#include "input_stream.h"
+
+#include <assert.h>
+
+/**
+ * All chunks are full of decoded data; wait for the player to free
+ * one.
+ */
+static enum decoder_command
+need_chunks(struct input_stream *is, bool do_wait)
+{
+	if (dc.command == DECODE_COMMAND_STOP ||
+	    dc.command == DECODE_COMMAND_SEEK)
+		return dc.command;
+
+	if ((is == NULL || input_stream_buffer(is) <= 0) && do_wait) {
+		notify_wait(&dc.notify);
+		notify_signal(&pc.notify);
+
+		return dc.command;
+	}
+
+	return DECODE_COMMAND_NONE;
+}
+
+struct music_chunk *
+decoder_get_chunk(struct decoder *decoder)
+{
+	assert(decoder != NULL);
+
+	if (decoder->chunk != NULL)
+		return decoder->chunk;
+
+	decoder->chunk = music_pipe_allocate();
+	return decoder->chunk;
+}
+
+enum decoder_command
+decoder_flush_chunk(struct decoder *decoder, struct input_stream *is)
+{
+	bool success;
+	enum decoder_command cmd;
+
+	assert(decoder != NULL);
+	assert(decoder->chunk != NULL);
+
+	while (true) {
+		success = music_pipe_push(decoder->chunk);
+		if (success) {
+			decoder->chunk = NULL;
+			return DECODE_COMMAND_NONE;
+		}
+
+		cmd = need_chunks(is, true);
+		if (cmd != DECODE_COMMAND_NONE)
+			return cmd;
+	}
+}

src/decoder_internal.h

@@ -19,8 +19,11 @@
 #ifndef MPD_DECODER_INTERNAL_H
 #define MPD_DECODER_INTERNAL_H
 
+#include "decoder_command.h"
 #include "pcm_convert.h"
 
+struct input_stream;
+
 struct decoder {
 	struct pcm_convert_state conv_state;
 
@@ -31,6 +34,25 @@ struct decoder {
 
 	/** the last tag received from the decoder plugin */
 	struct tag *decoder_tag;
+
+	/** the chunk currently being written to */
+	struct music_chunk *chunk;
 };
 
+/**
+ * Returns the current chunk the decoder writes to, or allocates a new
+ * chunk if there is none.
+ */
+struct music_chunk *
+decoder_get_chunk(struct decoder *decoder);
+
+/**
+ * Flushes a chunk.  Waits for room in the music pipe if required.
+ *
+ * @return DECODE_COMMAND_NONE on success, any other command if we
+ * have received a decoder command while waiting
+ */
+enum decoder_command
+decoder_flush_chunk(struct decoder *decoder, struct input_stream *is);
+
 #endif

src/decoder_thread.c

@@ -98,6 +98,7 @@ static void decoder_run_song(const struct song *song, const char *uri)
 	decoder.seeking = false;
 	decoder.stream_tag = NULL;
 	decoder.decoder_tag = NULL;
+	decoder.chunk = NULL;
 
 	dc.state = DECODE_STATE_START;
 	dc.command = DECODE_COMMAND_NONE;
@@ -194,7 +195,10 @@ static void decoder_run_song(const struct song *song, const char *uri)
 
 	pcm_convert_deinit(&decoder.conv_state);
 
-	music_pipe_flush();
+	/* flush the last chunk */
+	if (decoder.chunk != NULL &&
+	    decoder_flush_chunk(&decoder, NULL) != DECODE_COMMAND_NONE)
+		music_pipe_cancel(decoder.chunk);
 
 	if (close_instream)
 		input_stream_close(&input_stream);

src/pipe.c

@@ -94,22 +94,6 @@ static void output_buffer_expand(unsigned i)
 		notify_signal(music_pipe.notify);
 }
 
-void music_pipe_flush(void)
-{
-	struct music_chunk *chunk = music_pipe_get_chunk(music_pipe.end);
-
-	if (chunk->length > 0) {
-		unsigned int next = successor(music_pipe.end);
-		if (next == music_pipe.begin)
-			/* all buffers are full; we have to wait for
-			   the player to free one, so don't flush
-			   right now */
-			return;
-
-		output_buffer_expand(next);
-	}
-}
-
 void music_pipe_set_lazy(bool lazy)
 {
 	music_pipe.lazy = lazy;
@@ -163,92 +147,40 @@ music_pipe_get_chunk(const unsigned i)
 	return &music_pipe.chunks[i];
 }
 
-/**
+struct music_chunk *
- * Return the tail chunk which has room for additional data.
+music_pipe_allocate(void)
- *
- * @return the chunk which has room for more data; NULL if there is no
- * room.
- */
-static struct music_chunk *
-tail_chunk(size_t frame_size)
 {
-	unsigned int next;
 	struct music_chunk *chunk;
 
+	/* the music_pipe.end chunk is always kept initialized */
 	chunk = music_pipe_get_chunk(music_pipe.end);
-	assert(chunk->length <= sizeof(chunk->data));
+	assert(chunk->length == 0);
-	assert((chunk->length % frame_size) == 0);
-
-	if (chunk->length + frame_size > sizeof(chunk->data)) {
-		/* this chunk is full; allocate a new chunk */
-		next = successor(music_pipe.end);
-		if (music_pipe.begin == next)
-			/* no chunks available */
-			return NULL;
-
-		output_buffer_expand(next);
-		chunk = music_pipe_get_chunk(next);
-		assert(chunk->length == 0);
-	}
 
 	return chunk;
 }
 
-void *
+bool
-music_pipe_write(const struct audio_format *audio_format,
+music_pipe_push(struct music_chunk *chunk)
-		 float data_time, uint16_t bit_rate,
-		 size_t *max_length_r)
 {
-	const size_t frame_size = audio_format_frame_size(audio_format);
+	unsigned int next;
-	struct music_chunk *chunk;
 
-	chunk = tail_chunk(frame_size);
+	assert(chunk == music_pipe_get_chunk(music_pipe.end));
-	if (chunk == NULL)
-		return NULL;
 
-	return music_chunk_write(chunk, audio_format, data_time, bit_rate,
+	next = successor(music_pipe.end);
-				 max_length_r);
+	if (music_pipe.begin == next)
+		/* no room */
+		return false;
+
+	output_buffer_expand(next);
+	return true;
 }
 
 void
-music_pipe_expand(const struct audio_format *audio_format, size_t length)
+music_pipe_cancel(struct music_chunk *chunk)
 {
-	const size_t frame_size = audio_format_frame_size(audio_format);
+	assert(chunk == music_pipe_get_chunk(music_pipe.end));
-	struct music_chunk *chunk;
-	bool full;
 
-	/* no partial frames allowed */
+	music_chunk_free(chunk);
-	assert(length % frame_size == 0);
-
-	chunk = tail_chunk(frame_size);
-	assert(chunk != NULL);
-
-	full = music_chunk_expand(chunk, audio_format, length);
-	if (full)
-		music_pipe_flush();
-}
-
-bool music_pipe_tag(const struct tag *tag)
-{
-	struct music_chunk *chunk;
-
-	chunk = music_pipe_get_chunk(music_pipe.end);
-	if (chunk->length > 0 || chunk->tag != NULL) {
-		/* this chunk is not empty; allocate a new chunk,
-		   because chunk.tag refers to the beginning of the
-		   chunk data */
-		unsigned next = successor(music_pipe.end);
-		if (music_pipe.begin == next)
-			/* no chunks available */
-			return false;
-
-		output_buffer_expand(next);
-		chunk = music_pipe_get_chunk(next);
-		assert(chunk->length == 0 && chunk->tag == NULL);
-	}
-
-	chunk->tag = tag_dup(tag);
-	return true;
 }
 
 void music_pipe_skip(unsigned num)

src/pipe.h

@@ -57,8 +57,6 @@ void music_pipe_free(void);
 
 void music_pipe_clear(void);
 
-void music_pipe_flush(void);
-
 /**
  * When a chunk is decoded, we wake up the player thread to tell him
  * about it.  In "lazy" mode, we only wake him up when the buffer was
@@ -120,6 +118,32 @@ music_pipe_peek(void)
 	return music_pipe_get_chunk(music_pipe.begin);
 }
 
+/**
+ * Allocates a chunk for writing.  When you are finished, append it
+ * with music_pipe_push().
+ *
+ * @return an empty chunk
+ */
+struct music_chunk *
+music_pipe_allocate(void);
+
+/**
+ * Appends a chunk at the end of the music pipe.
+ *
+ * @param chunk a chunk allocated with music_pipe_allocate()
+ * @return true on success, false if there is no room
+ */
+bool
+music_pipe_push(struct music_chunk *chunk);
+
+/**
+ * Cancels a chunk that has been allocated with music_pipe_allocate().
+ *
+ * @param chunk a chunk allocated with music_pipe_allocate()
+ */
+void
+music_pipe_cancel(struct music_chunk *chunk);
+
 /**
  * Prepares appending to the music pipe.  Returns a buffer where you
  * may write into.  After you are finished, call music_pipe_expand().
@@ -138,12 +162,6 @@ music_pipe_write(const struct audio_format *audio_format,
 void
 music_pipe_expand(const struct audio_format *audio_format, size_t length);
 
-/**
- * Send a tag.  This is usually called when a new song within a stream
- * begins.
- */
-bool music_pipe_tag(const struct tag *tag);
-
 void music_pipe_skip(unsigned num);
 
 /**