aboutsummaryrefslogtreecommitdiffstats
path: root/src
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
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 'src')
-rw-r--r--src/archive/bz2_archive_plugin.c13
-rw-r--r--src/archive/iso9660_archive_plugin.c8
-rw-r--r--src/archive/zzip_archive_plugin.c7
-rw-r--r--src/archive_plugin.c8
-rw-r--r--src/archive_plugin.h6
-rw-r--r--src/decoder/audiofile_decoder_plugin.c4
-rw-r--r--src/decoder/faad_decoder_plugin.c4
-rw-r--r--src/decoder/ffmpeg_decoder_plugin.c5
-rw-r--r--src/decoder/flac_decoder_plugin.c9
-rw-r--r--src/decoder/mad_decoder_plugin.c2
-rw-r--r--src/decoder/modplug_decoder_plugin.c2
-rw-r--r--src/decoder/mp4ff_decoder_plugin.c3
-rw-r--r--src/decoder/mpcdec_decoder_plugin.c2
-rw-r--r--src/decoder/pcm_decoder_plugin.c5
-rw-r--r--src/decoder/sndfile_decoder_plugin.c4
-rw-r--r--src/decoder/vorbis_decoder_plugin.c4
-rw-r--r--src/decoder/wavpack_decoder_plugin.c12
-rw-r--r--src/decoder_api.c19
-rw-r--r--src/decoder_thread.c44
-rw-r--r--src/input/archive_input_plugin.c7
-rw-r--r--src/input/cdio_paranoia_input_plugin.c7
-rw-r--r--src/input/curl_input_plugin.c108
-rw-r--r--src/input/despotify_input_plugin.c7
-rw-r--r--src/input/ffmpeg_input_plugin.c7
-rw-r--r--src/input/file_input_plugin.c7
-rw-r--r--src/input/mms_input_plugin.c7
-rw-r--r--src/input/rewind_input_plugin.c15
-rw-r--r--src/input/soup_input_plugin.c102
-rw-r--r--src/input_internal.c24
-rw-r--r--src/input_internal.h10
-rw-r--r--src/input_plugin.h16
-rw-r--r--src/input_stream.c124
-rw-r--r--src/input_stream.h121
-rw-r--r--src/playlist/asx_playlist_plugin.c3
-rw-r--r--src/playlist/cue_playlist_plugin.c3
-rw-r--r--src/playlist/despotify_playlist_plugin.c3
-rw-r--r--src/playlist/flac_playlist_plugin.c3
-rw-r--r--src/playlist/lastfm_playlist_plugin.c45
-rw-r--r--src/playlist/pls_playlist_plugin.c3
-rw-r--r--src/playlist/rss_playlist_plugin.c3
-rw-r--r--src/playlist/xspf_playlist_plugin.c3
-rw-r--r--src/playlist_any.c15
-rw-r--r--src/playlist_any.h5
-rw-r--r--src/playlist_list.c45
-rw-r--r--src/playlist_list.h7
-rw-r--r--src/playlist_mapper.c27
-rw-r--r--src/playlist_mapper.h5
-rw-r--r--src/playlist_plugin.h10
-rw-r--r--src/playlist_print.c14
-rw-r--r--src/playlist_queue.c14
-rw-r--r--src/song_update.c18
-rw-r--r--src/text_input_stream.c4
52 files changed, 607 insertions, 346 deletions
diff --git a/src/archive/bz2_archive_plugin.c b/src/archive/bz2_archive_plugin.c
index 90418720e..303f22136 100644
--- a/src/archive/bz2_archive_plugin.c
+++ b/src/archive/bz2_archive_plugin.c
@@ -114,7 +114,11 @@ bz2_open(const char *pathname, GError **error_r)
refcount_init(&context->ref);
//open archive
- context->istream = input_stream_open(pathname, error_r);
+ static GStaticMutex mutex = G_STATIC_MUTEX_INIT;
+ context->istream = input_stream_open(pathname,
+ g_static_mutex_get_mutex(&mutex),
+ NULL,
+ error_r);
if (context->istream == NULL) {
g_free(context);
return NULL;
@@ -169,12 +173,15 @@ bz2_close(struct archive_file *file)
/* single archive handling */
static struct input_stream *
-bz2_open_stream(struct archive_file *file, const char *path, GError **error_r)
+bz2_open_stream(struct archive_file *file, const char *path,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
struct bz2_archive_file *context = (struct bz2_archive_file *) file;
struct bz2_input_stream *bis = g_new(struct bz2_input_stream, 1);
- input_stream_init(&bis->base, &bz2_inputplugin, path);
+ input_stream_init(&bis->base, &bz2_inputplugin, path,
+ mutex, cond);
bis->archive = context;
diff --git a/src/archive/iso9660_archive_plugin.c b/src/archive/iso9660_archive_plugin.c
index da55feca2..bb6cb9588 100644
--- a/src/archive/iso9660_archive_plugin.c
+++ b/src/archive/iso9660_archive_plugin.c
@@ -173,15 +173,17 @@ struct iso9660_input_stream {
};
static struct input_stream *
-iso9660_archive_open_stream(struct archive_file *file,
- const char *pathname, GError **error_r)
+iso9660_archive_open_stream(struct archive_file *file, const char *pathname,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
struct iso9660_archive_file *context =
(struct iso9660_archive_file *)file;
struct iso9660_input_stream *iis;
iis = g_new(struct iso9660_input_stream, 1);
- input_stream_init(&iis->base, &iso9660_input_plugin, pathname);
+ input_stream_init(&iis->base, &iso9660_input_plugin, pathname,
+ mutex, cond);
iis->archive = context;
iis->statbuf = iso9660_ifs_stat_translate(context->iso, pathname);
diff --git a/src/archive/zzip_archive_plugin.c b/src/archive/zzip_archive_plugin.c
index 829fb6dd1..ad96b5f89 100644
--- a/src/archive/zzip_archive_plugin.c
+++ b/src/archive/zzip_archive_plugin.c
@@ -135,14 +135,17 @@ struct zzip_input_stream {
static struct input_stream *
zzip_archive_open_stream(struct archive_file *file,
- const char *pathname, GError **error_r)
+ const char *pathname,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
struct zzip_archive *context = (struct zzip_archive *) file;
struct zzip_input_stream *zis;
ZZIP_STAT z_stat;
zis = g_new(struct zzip_input_stream, 1);
- input_stream_init(&zis->base, &zzip_input_plugin, pathname);
+ input_stream_init(&zis->base, &zzip_input_plugin, pathname,
+ mutex, cond);
zis->archive = context;
zis->file = zzip_file_open(context->dir, pathname, 0);
diff --git a/src/archive_plugin.c b/src/archive_plugin.c
index e73035053..cf23e6393 100644
--- a/src/archive_plugin.c
+++ b/src/archive_plugin.c
@@ -81,12 +81,14 @@ archive_file_scan_next(struct archive_file *file)
}
struct input_stream *
-archive_file_open_stream(struct archive_file *file,
- const char *path, GError **error_r)
+archive_file_open_stream(struct archive_file *file, const char *path,
+ GMutex *mutex, GCond *cond,
+ GError **error_r)
{
assert(file != NULL);
assert(file->plugin != NULL);
assert(file->plugin->open_stream != NULL);
- return file->plugin->open_stream(file, path, error_r);
+ return file->plugin->open_stream(file, path, mutex, cond,
+ error_r);
}
diff --git a/src/archive_plugin.h b/src/archive_plugin.h
index 7f038486b..b7b92446d 100644
--- a/src/archive_plugin.h
+++ b/src/archive_plugin.h
@@ -73,6 +73,7 @@ struct archive_plugin {
*/
struct input_stream *(*open_stream)(struct archive_file *af,
const char *path,
+ GMutex *mutex, GCond *cond,
GError **error_r);
/**
@@ -101,7 +102,8 @@ char *
archive_file_scan_next(struct archive_file *file);
struct input_stream *
-archive_file_open_stream(struct archive_file *file,
- const char *path, GError **error_r);
+archive_file_open_stream(struct archive_file *file, const char *path,
+ GMutex *mutex, GCond *cond,
+ GError **error_r);
#endif
diff --git a/src/decoder/audiofile_decoder_plugin.c b/src/decoder/audiofile_decoder_plugin.c
index c862168f8..0cf54bc60 100644
--- a/src/decoder/audiofile_decoder_plugin.c
+++ b/src/decoder/audiofile_decoder_plugin.c
@@ -53,7 +53,7 @@ audiofile_file_read(AFvirtualfile *vfile, void *data, size_t length)
GError *error = NULL;
size_t nbytes;
- nbytes = input_stream_read(is, data, length, &error);
+ nbytes = input_stream_lock_read(is, data, length, &error);
if (nbytes == 0 && error != NULL) {
g_warning("%s", error->message);
g_error_free(error);
@@ -90,7 +90,7 @@ audiofile_file_seek(AFvirtualfile *vfile, long offset, int is_relative)
{
struct input_stream *is = (struct input_stream *) vfile->closure;
int whence = (is_relative ? SEEK_CUR : SEEK_SET);
- if (input_stream_seek(is, offset, whence, NULL)) {
+ if (input_stream_lock_seek(is, offset, whence, NULL)) {
return is->offset;
} else {
return -1;
diff --git a/src/decoder/faad_decoder_plugin.c b/src/decoder/faad_decoder_plugin.c
index 02c72a4a1..91aa5392a 100644
--- a/src/decoder/faad_decoder_plugin.c
+++ b/src/decoder/faad_decoder_plugin.c
@@ -205,7 +205,7 @@ faad_song_duration(struct decoder_buffer *buffer, struct input_stream *is)
/* obtain the duration from the ADTS header */
float song_length = adts_song_duration(buffer);
- input_stream_seek(is, tagsize, SEEK_SET, NULL);
+ input_stream_lock_seek(is, tagsize, SEEK_SET, NULL);
data = decoder_buffer_read(buffer, &length);
if (data != NULL)
@@ -406,7 +406,7 @@ faad_stream_decode(struct decoder *mpd_decoder, struct input_stream *is)
faacDecSetConfiguration(decoder, config);
while (!decoder_buffer_is_full(buffer) &&
- !input_stream_eof(is) &&
+ !input_stream_lock_eof(is) &&
decoder_get_command(mpd_decoder) == DECODE_COMMAND_NONE) {
adts_find_frame(buffer);
decoder_buffer_fill(buffer);
diff --git a/src/decoder/ffmpeg_decoder_plugin.c b/src/decoder/ffmpeg_decoder_plugin.c
index b4f1f0b51..2923c1400 100644
--- a/src/decoder/ffmpeg_decoder_plugin.c
+++ b/src/decoder/ffmpeg_decoder_plugin.c
@@ -105,7 +105,7 @@ mpd_ffmpeg_stream_seek(void *opaque, int64_t pos, int whence)
if (whence == AVSEEK_SIZE)
return stream->input->size;
- if (!input_stream_seek(stream->input, pos, whence, NULL))
+ if (!input_stream_lock_seek(stream->input, pos, whence, NULL))
return -1;
return stream->input->offset;
@@ -320,7 +320,8 @@ ffmpeg_probe(struct decoder *decoder, struct input_stream *is)
unsigned char *buffer = g_malloc(BUFFER_SIZE);
size_t nbytes = decoder_read(decoder, is, buffer, BUFFER_SIZE);
- if (nbytes <= PADDING || !input_stream_seek(is, 0, SEEK_SET, NULL)) {
+ if (nbytes <= PADDING ||
+ !input_stream_lock_seek(is, 0, SEEK_SET, NULL)) {
g_free(buffer);
return NULL;
}
diff --git a/src/decoder/flac_decoder_plugin.c b/src/decoder/flac_decoder_plugin.c
index ca9cd5968..2a826f862 100644
--- a/src/decoder/flac_decoder_plugin.c
+++ b/src/decoder/flac_decoder_plugin.c
@@ -50,7 +50,7 @@ flac_read_cb(G_GNUC_UNUSED const FLAC__StreamDecoder *fd,
if (r == 0) {
if (decoder_get_command(data->decoder) != DECODE_COMMAND_NONE ||
- input_stream_eof(data->input_stream))
+ input_stream_lock_eof(data->input_stream))
return FLAC__STREAM_DECODER_READ_STATUS_END_OF_STREAM;
else
return FLAC__STREAM_DECODER_READ_STATUS_ABORT;
@@ -68,7 +68,8 @@ flac_seek_cb(G_GNUC_UNUSED const FLAC__StreamDecoder *fd,
if (!data->input_stream->seekable)
return FLAC__STREAM_DECODER_SEEK_STATUS_UNSUPPORTED;
- if (!input_stream_seek(data->input_stream, offset, SEEK_SET, NULL))
+ if (!input_stream_lock_seek(data->input_stream, offset, SEEK_SET,
+ NULL))
return FLAC__STREAM_DECODER_SEEK_STATUS_ERROR;
return FLAC__STREAM_DECODER_SEEK_STATUS_OK;
@@ -109,7 +110,7 @@ flac_eof_cb(G_GNUC_UNUSED const FLAC__StreamDecoder *fd, void *fdata)
return (decoder_get_command(data->decoder) != DECODE_COMMAND_NONE &&
decoder_get_command(data->decoder) != DECODE_COMMAND_SEEK) ||
- input_stream_eof(data->input_stream);
+ input_stream_lock_eof(data->input_stream);
}
static void
@@ -449,7 +450,7 @@ oggflac_decode(struct decoder *decoder, struct input_stream *input_stream)
/* rewind the stream, because ogg_stream_type_detect() has
moved it */
- input_stream_seek(input_stream, 0, SEEK_SET, NULL);
+ input_stream_lock_seek(input_stream, 0, SEEK_SET, NULL);
flac_decode_internal(decoder, input_stream, true);
}
diff --git a/src/decoder/mad_decoder_plugin.c b/src/decoder/mad_decoder_plugin.c
index 8f77052f7..8bf3f6546 100644
--- a/src/decoder/mad_decoder_plugin.c
+++ b/src/decoder/mad_decoder_plugin.c
@@ -168,7 +168,7 @@ mp3_data_init(struct mp3_data *data, struct decoder *decoder,
static bool mp3_seek(struct mp3_data *data, long offset)
{
- if (!input_stream_seek(data->input_stream, offset, SEEK_SET, NULL))
+ if (!input_stream_lock_seek(data->input_stream, offset, SEEK_SET, NULL))
return false;
mad_stream_buffer(&data->stream, data->input_buffer, 0);
diff --git a/src/decoder/modplug_decoder_plugin.c b/src/decoder/modplug_decoder_plugin.c
index 341b00927..9345dd240 100644
--- a/src/decoder/modplug_decoder_plugin.c
+++ b/src/decoder/modplug_decoder_plugin.c
@@ -62,7 +62,7 @@ static GByteArray *mod_loadfile(struct decoder *decoder, struct input_stream *is
while (true) {
ret = decoder_read(decoder, is, data, MODPLUG_READ_BLOCK);
if (ret == 0) {
- if (input_stream_eof(is))
+ if (input_stream_lock_eof(is))
/* end of file */
break;
diff --git a/src/decoder/mp4ff_decoder_plugin.c b/src/decoder/mp4ff_decoder_plugin.c
index 38ae5793a..6475211a4 100644
--- a/src/decoder/mp4ff_decoder_plugin.c
+++ b/src/decoder/mp4ff_decoder_plugin.c
@@ -102,7 +102,8 @@ mp4_seek(void *user_data, uint64_t position)
{
struct mp4ff_input_stream *mis = user_data;
- return input_stream_seek(mis->input_stream, position, SEEK_SET, NULL)
+ return input_stream_lock_seek(mis->input_stream, position, SEEK_SET,
+ NULL)
? 0 : -1;
}
diff --git a/src/decoder/mpcdec_decoder_plugin.c b/src/decoder/mpcdec_decoder_plugin.c
index f31dcdb99..7864c0790 100644
--- a/src/decoder/mpcdec_decoder_plugin.c
+++ b/src/decoder/mpcdec_decoder_plugin.c
@@ -61,7 +61,7 @@ mpc_seek_cb(cb_first_arg, mpc_int32_t offset)
{
struct mpc_decoder_data *data = (struct mpc_decoder_data *) cb_data;
- return input_stream_seek(data->is, offset, SEEK_SET, NULL);
+ return input_stream_lock_seek(data->is, offset, SEEK_SET, NULL);
}
static mpc_int32_t
diff --git a/src/decoder/pcm_decoder_plugin.c b/src/decoder/pcm_decoder_plugin.c
index c8340ab67..24ad93cb6 100644
--- a/src/decoder/pcm_decoder_plugin.c
+++ b/src/decoder/pcm_decoder_plugin.c
@@ -52,7 +52,7 @@ pcm_stream_decode(struct decoder *decoder, struct input_stream *is)
size_t nbytes = decoder_read(decoder, is,
buffer, sizeof(buffer));
- if (nbytes == 0 && input_stream_eof(is))
+ if (nbytes == 0 && input_stream_lock_eof(is))
break;
cmd = nbytes > 0
@@ -62,7 +62,8 @@ pcm_stream_decode(struct decoder *decoder, struct input_stream *is)
if (cmd == DECODE_COMMAND_SEEK) {
goffset offset = (goffset)(time_to_size *
decoder_seek_where(decoder));
- if (input_stream_seek(is, offset, SEEK_SET, &error)) {
+ if (input_stream_lock_seek(is, offset, SEEK_SET,
+ &error)) {
decoder_command_finished(decoder);
} else {
g_warning("seeking failed: %s", error->message);
diff --git a/src/decoder/sndfile_decoder_plugin.c b/src/decoder/sndfile_decoder_plugin.c
index dbe9bf067..25952dfd5 100644
--- a/src/decoder/sndfile_decoder_plugin.c
+++ b/src/decoder/sndfile_decoder_plugin.c
@@ -40,7 +40,7 @@ sndfile_vio_seek(sf_count_t offset, int whence, void *user_data)
struct input_stream *is = user_data;
bool success;
- success = input_stream_seek(is, offset, whence, NULL);
+ success = input_stream_lock_seek(is, offset, whence, NULL);
if (!success)
return -1;
@@ -54,7 +54,7 @@ sndfile_vio_read(void *ptr, sf_count_t count, void *user_data)
GError *error = NULL;
size_t nbytes;
- nbytes = input_stream_read(is, ptr, count, &error);
+ nbytes = input_stream_lock_read(is, ptr, count, &error);
if (nbytes == 0 && error != NULL) {
g_warning("%s", error->message);
g_error_free(error);
diff --git a/src/decoder/vorbis_decoder_plugin.c b/src/decoder/vorbis_decoder_plugin.c
index c130005a7..f73a41895 100644
--- a/src/decoder/vorbis_decoder_plugin.c
+++ b/src/decoder/vorbis_decoder_plugin.c
@@ -80,7 +80,7 @@ static int ogg_seek_cb(void *data, ogg_int64_t offset, int whence)
return vis->seekable &&
(!vis->decoder || decoder_get_command(vis->decoder) != DECODE_COMMAND_STOP) &&
- input_stream_seek(vis->input_stream, offset, whence, NULL)
+ input_stream_lock_seek(vis->input_stream, offset, whence, NULL)
? 0 : -1;
}
@@ -290,7 +290,7 @@ vorbis_stream_decode(struct decoder *decoder,
/* rewind the stream, because ogg_stream_type_detect() has
moved it */
- input_stream_seek(input_stream, 0, SEEK_SET, NULL);
+ input_stream_lock_seek(input_stream, 0, SEEK_SET, NULL);
if (!vorbis_is_open(&vis, &vf, decoder, input_stream))
return;
diff --git a/src/decoder/wavpack_decoder_plugin.c b/src/decoder/wavpack_decoder_plugin.c
index 200bf6455..bfb13b2a4 100644
--- a/src/decoder/wavpack_decoder_plugin.c
+++ b/src/decoder/wavpack_decoder_plugin.c
@@ -390,13 +390,15 @@ wavpack_input_get_pos(void *id)
static int
wavpack_input_set_pos_abs(void *id, uint32_t pos)
{
- return input_stream_seek(wpin(id)->is, pos, SEEK_SET, NULL) ? 0 : -1;
+ return input_stream_lock_seek(wpin(id)->is, pos, SEEK_SET, NULL)
+ ? 0 : -1;
}
static int
wavpack_input_set_pos_rel(void *id, int32_t delta, int mode)
{
- return input_stream_seek(wpin(id)->is, delta, mode, NULL) ? 0 : -1;
+ return input_stream_lock_seek(wpin(id)->is, delta, mode, NULL)
+ ? 0 : -1;
}
static int
@@ -447,6 +449,7 @@ wavpack_input_init(struct wavpack_input *isp, struct decoder *decoder,
static struct input_stream *
wavpack_open_wvc(struct decoder *decoder, const char *uri,
+ GMutex *mutex, GCond *cond,
struct wavpack_input *wpi)
{
struct input_stream *is_wvc;
@@ -462,7 +465,7 @@ wavpack_open_wvc(struct decoder *decoder, const char *uri,
return false;
wvc_url = g_strconcat(uri, "c", NULL);
- is_wvc = input_stream_open(wvc_url, NULL);
+ is_wvc = input_stream_open(wvc_url, mutex, cond, NULL);
g_free(wvc_url);
if (is_wvc == NULL)
@@ -499,7 +502,8 @@ wavpack_streamdecode(struct decoder * decoder, struct input_stream *is)
struct wavpack_input isp, isp_wvc;
bool can_seek = is->seekable;
- is_wvc = wavpack_open_wvc(decoder, is->uri, &isp_wvc);
+ is_wvc = wavpack_open_wvc(decoder, is->uri, is->mutex, is->cond,
+ &isp_wvc);
if (is_wvc != NULL) {
open_flags |= OPEN_WVC;
can_seek &= is_wvc->seekable;
diff --git a/src/decoder_api.c b/src/decoder_api.c
index 20e976e69..1f2075638 100644
--- a/src/decoder_api.c
+++ b/src/decoder_api.c
@@ -183,8 +183,19 @@ size_t decoder_read(struct decoder *decoder,
if (length == 0)
return 0;
- if (decoder_check_cancel_read(decoder))
- return 0;
+ input_stream_lock(is);
+
+ while (true) {
+ if (decoder_check_cancel_read(decoder)) {
+ input_stream_unlock(is);
+ return 0;
+ }
+
+ if (input_stream_available(is))
+ break;
+
+ g_cond_wait(is->cond, is->mutex);
+ }
nbytes = input_stream_read(is, buffer, length, &error);
assert(nbytes == 0 || error == NULL);
@@ -195,6 +206,8 @@ size_t decoder_read(struct decoder *decoder,
g_error_free(error);
}
+ input_stream_unlock(is);
+
return nbytes;
}
@@ -241,7 +254,7 @@ update_stream_tag(struct decoder *decoder, struct input_stream *is)
struct tag *tag;
tag = is != NULL
- ? input_stream_tag(is)
+ ? input_stream_lock_tag(is)
: NULL;
if (tag == NULL) {
tag = decoder->song_tag;
diff --git a/src/decoder_thread.c b/src/decoder_thread.c
index 320a04638..dff4ca08f 100644
--- a/src/decoder_thread.c
+++ b/src/decoder_thread.c
@@ -41,18 +41,6 @@
#undef G_LOG_DOMAIN
#define G_LOG_DOMAIN "decoder_thread"
-static enum decoder_command
-decoder_lock_get_command(struct decoder_control *dc)
-{
- enum decoder_command command;
-
- decoder_lock(dc);
- command = dc->command;
- decoder_unlock(dc);
-
- return command;
-}
-
/**
* Marks the current decoder command as "finished" and notifies the
* player thread.
@@ -86,7 +74,7 @@ decoder_input_stream_open(struct decoder_control *dc, const char *uri)
GError *error = NULL;
struct input_stream *is;
- is = input_stream_open(uri, &error);
+ is = input_stream_open(uri, dc->mutex, dc->cond, &error);
if (is == NULL) {
if (error != NULL) {
g_warning("%s", error->message);
@@ -99,19 +87,27 @@ decoder_input_stream_open(struct decoder_control *dc, const char *uri)
/* wait for the input stream to become ready; its metadata
will be available then */
+ decoder_lock(dc);
+
+ input_stream_update(is);
while (!is->ready &&
- decoder_lock_get_command(dc) != DECODE_COMMAND_STOP) {
- int ret;
+ dc->command != DECODE_COMMAND_STOP) {
+ decoder_wait(dc);
- ret = input_stream_buffer(is, &error);
- if (ret < 0) {
- input_stream_close(is);
- g_warning("%s", error->message);
- g_error_free(error);
- return NULL;
- }
+ input_stream_update(is);
}
+ if (!input_stream_check(is, &error)) {
+ decoder_unlock(dc);
+
+ g_warning("%s", error->message);
+ g_error_free(error);
+
+ return NULL;
+ }
+
+ decoder_unlock(dc);
+
return is;
}
@@ -132,11 +128,11 @@ decoder_stream_decode(const struct decoder_plugin *plugin,
if (decoder->dc->command == DECODE_COMMAND_STOP)
return true;
- decoder_unlock(decoder->dc);
-
/* rewind the stream, so each plugin gets a fresh start */
input_stream_seek(input_stream, 0, SEEK_SET, NULL);
+ decoder_unlock(decoder->dc);
+
decoder_plugin_stream_decode(plugin, decoder, input_stream);
decoder_lock(decoder->dc);
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
diff --git a/src/playlist/asx_playlist_plugin.c b/src/playlist/asx_playlist_plugin.c
index b711f83f3..298687859 100644
--- a/src/playlist/asx_playlist_plugin.c
+++ b/src/playlist/asx_playlist_plugin.c
@@ -233,7 +233,8 @@ asx_open_stream(struct input_stream *is)
&parser, asx_parser_destroy);
while (true) {
- nbytes = input_stream_read(is, buffer, sizeof(buffer), &error);
+ nbytes = input_stream_lock_read(is, buffer, sizeof(buffer),
+ &error);
if (nbytes == 0) {
if (error != NULL) {
g_markup_parse_context_free(context);
diff --git a/src/playlist/cue_playlist_plugin.c b/src/playlist/cue_playlist_plugin.c
index e3619a284..3f2d5b34c 100644
--- a/src/playlist/cue_playlist_plugin.c
+++ b/src/playlist/cue_playlist_plugin.c
@@ -41,7 +41,8 @@ struct cue_playlist {
};
static struct playlist_provider *
-cue_playlist_open_uri(const char *uri)
+cue_playlist_open_uri(const char *uri,
+ G_GNUC_UNUSED GMutex *mutex, G_GNUC_UNUSED GCond *cond)
{
struct cue_playlist *playlist;
FILE *file;
diff --git a/src/playlist/despotify_playlist_plugin.c b/src/playlist/despotify_playlist_plugin.c
index 39448e01f..08a32d79d 100644
--- a/src/playlist/despotify_playlist_plugin.c
+++ b/src/playlist/despotify_playlist_plugin.c
@@ -110,7 +110,8 @@ despotify_playlist_finish(void)
static struct playlist_provider *
-despotify_playlist_open_uri(const char *url)
+despotify_playlist_open_uri(const char *url, G_GNUC_UNUSED GMutex *mutex,
+ G_GNUC_UNUSED GCond *cond)
{
struct despotify_playlist *ctx;
struct despotify_session *session;
diff --git a/src/playlist/flac_playlist_plugin.c b/src/playlist/flac_playlist_plugin.c
index 8adf694ed..e20519435 100644
--- a/src/playlist/flac_playlist_plugin.c
+++ b/src/playlist/flac_playlist_plugin.c
@@ -45,7 +45,8 @@ struct flac_playlist {
};
static struct playlist_provider *
-flac_playlist_open_uri(const char *uri)
+flac_playlist_open_uri(const char *uri,
+ G_GNUC_UNUSED GMutex *mutex, G_GNUC_UNUSED GCond *cond)
{
if (!g_path_is_absolute(uri))
/* only local files supported */
diff --git a/src/playlist/lastfm_playlist_plugin.c b/src/playlist/lastfm_playlist_plugin.c
index 5c803cd99..86113643c 100644
--- a/src/playlist/lastfm_playlist_plugin.c
+++ b/src/playlist/lastfm_playlist_plugin.c
@@ -83,15 +83,14 @@ lastfm_finish(void)
* @return data fetched, or NULL on error. Must be freed with g_free.
*/
static char *
-lastfm_get(const char *url)
+lastfm_get(const char *url, GMutex *mutex, GCond *cond)
{
struct input_stream *input_stream;
GError *error = NULL;
- int ret;
char buffer[4096];
size_t length = 0, nbytes;
- input_stream = input_stream_open(url, &error);
+ input_stream = input_stream_open(url, mutex, cond, &error);
if (input_stream == NULL) {
if (error != NULL) {
g_warning("%s", error->message);
@@ -101,15 +100,9 @@ lastfm_get(const char *url)
return NULL;
}
- while (!input_stream->ready) {
- ret = input_stream_buffer(input_stream, &error);
- if (ret < 0) {
- input_stream_close(input_stream);
- g_warning("%s", error->message);
- g_error_free(error);
- return NULL;
- }
- }
+ g_mutex_lock(mutex);
+
+ input_stream_wait_ready(input_stream);
do {
nbytes = input_stream_read(input_stream, buffer + length,
@@ -124,6 +117,7 @@ lastfm_get(const char *url)
break;
/* I/O error */
+ g_mutex_unlock(mutex);
input_stream_close(input_stream);
return NULL;
}
@@ -131,6 +125,8 @@ lastfm_get(const char *url)
length += nbytes;
} while (length < sizeof(buffer));
+ g_mutex_unlock(mutex);
+
input_stream_close(input_stream);
return g_strndup(buffer, length);
}
@@ -162,7 +158,7 @@ lastfm_find(const char *response, const char *name)
}
static struct playlist_provider *
-lastfm_open_uri(const char *uri)
+lastfm_open_uri(const char *uri, GMutex *mutex, GCond *cond)
{
struct lastfm_playlist *playlist;
GError *error = NULL;
@@ -175,7 +171,7 @@ lastfm_open_uri(const char *uri)
"username=", lastfm_config.user, "&"
"passwordmd5=", lastfm_config.md5, "&"
"debug=0&partner=", NULL);
- response = lastfm_get(p);
+ response = lastfm_get(p, mutex, cond);
g_free(p);
if (response == NULL)
return NULL;
@@ -207,7 +203,7 @@ lastfm_open_uri(const char *uri)
NULL);
g_free(escaped_uri);
- response = lastfm_get(p);
+ response = lastfm_get(p, mutex, cond);
g_free(response);
g_free(p);
@@ -229,7 +225,7 @@ lastfm_open_uri(const char *uri)
NULL);
g_free(session);
- playlist->is = input_stream_open(p, &error);
+ playlist->is = input_stream_open(p, mutex, cond, &error);
g_free(p);
if (playlist->is == NULL) {
@@ -243,26 +239,17 @@ lastfm_open_uri(const char *uri)
return NULL;
}
- while (!playlist->is->ready) {
- int ret = input_stream_buffer(playlist->is, &error);
- if (ret < 0) {
- input_stream_close(playlist->is);
- g_free(playlist);
- g_warning("%s", error->message);
- g_error_free(error);
- return NULL;
- }
+ g_mutex_lock(mutex);
- if (ret == 0)
- /* nothing was buffered - wait */
- g_usleep(10000);
- }
+ input_stream_wait_ready(playlist->is);
/* last.fm does not send a MIME type, we have to fake it here
:-( */
g_free(playlist->is->mime);
playlist->is->mime = g_strdup("application/xspf+xml");
+ g_mutex_unlock(mutex);
+
/* parse the XSPF playlist */
playlist->xspf = playlist_list_open_stream(playlist->is, NULL);
diff --git a/src/playlist/pls_playlist_plugin.c b/src/playlist/pls_playlist_plugin.c
index 937f02791..c4e5492af 100644
--- a/src/playlist/pls_playlist_plugin.c
+++ b/src/playlist/pls_playlist_plugin.c
@@ -115,7 +115,8 @@ pls_open_stream(struct input_stream *is)
GString *kf_data = g_string_new("");
do {
- nbytes = input_stream_read(is, buffer, sizeof(buffer), &error);
+ nbytes = input_stream_lock_read(is, buffer, sizeof(buffer),
+ &error);
if (nbytes == 0) {
if (error != NULL) {
g_string_free(kf_data, TRUE);
diff --git a/src/playlist/rss_playlist_plugin.c b/src/playlist/rss_playlist_plugin.c
index 9ce3c6abe..6740cba7e 100644
--- a/src/playlist/rss_playlist_plugin.c
+++ b/src/playlist/rss_playlist_plugin.c
@@ -231,7 +231,8 @@ rss_open_stream(struct input_stream *is)
&parser, rss_parser_destroy);
while (true) {
- nbytes = input_stream_read(is, buffer, sizeof(buffer), &error);
+ nbytes = input_stream_lock_read(is, buffer, sizeof(buffer),
+ &error);
if (nbytes == 0) {
if (error != NULL) {
g_markup_parse_context_free(context);
diff --git a/src/playlist/xspf_playlist_plugin.c b/src/playlist/xspf_playlist_plugin.c
index 89a4a08a4..17d9040e2 100644
--- a/src/playlist/xspf_playlist_plugin.c
+++ b/src/playlist/xspf_playlist_plugin.c
@@ -253,7 +253,8 @@ xspf_open_stream(struct input_stream *is)
&parser, xspf_parser_destroy);
while (true) {
- nbytes = input_stream_read(is, buffer, sizeof(buffer), &error);
+ nbytes = input_stream_lock_read(is, buffer, sizeof(buffer),
+ &error);
if (nbytes == 0) {
if (error != NULL) {
g_markup_parse_context_free(context);
diff --git a/src/playlist_any.c b/src/playlist_any.c
index bed12206a..450ca5932 100644
--- a/src/playlist_any.c
+++ b/src/playlist_any.c
@@ -27,18 +27,20 @@
#include <assert.h>
static struct playlist_provider *
-playlist_open_remote(const char *uri, struct input_stream **is_r)
+playlist_open_remote(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
assert(uri_has_scheme(uri));
- struct playlist_provider *playlist = playlist_list_open_uri(uri);
+ struct playlist_provider *playlist =
+ playlist_list_open_uri(uri, mutex, cond);
if (playlist != NULL) {
*is_r = NULL;
return playlist;
}
GError *error = NULL;
- struct input_stream *is = input_stream_open(uri, &error);
+ struct input_stream *is = input_stream_open(uri, mutex, cond, &error);
if (is == NULL) {
if (error != NULL) {
g_warning("Failed to open %s: %s",
@@ -60,9 +62,10 @@ playlist_open_remote(const char *uri, struct input_stream **is_r)
}
struct playlist_provider *
-playlist_open_any(const char *uri, struct input_stream **is_r)
+playlist_open_any(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
return uri_has_scheme(uri)
- ? playlist_open_remote(uri, is_r)
- : playlist_mapper_open(uri, is_r);
+ ? playlist_open_remote(uri, mutex, cond, is_r)
+ : playlist_mapper_open(uri, mutex, cond, is_r);
}
diff --git a/src/playlist_any.h b/src/playlist_any.h
index 7c13df718..310913de9 100644
--- a/src/playlist_any.h
+++ b/src/playlist_any.h
@@ -20,7 +20,7 @@
#ifndef MPD_PLAYLIST_ANY_H
#define MPD_PLAYLIST_ANY_H
-#include <stdbool.h>
+#include <glib.h>
struct playlist_provider;
struct input_stream;
@@ -35,6 +35,7 @@ struct input_stream;
* freed
*/
struct playlist_provider *
-playlist_open_any(const char *uri, struct input_stream **is_r);
+playlist_open_any(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r);
#endif
diff --git a/src/playlist_list.c b/src/playlist_list.c
index 04c3fa292..1f220eee8 100644
--- a/src/playlist_list.c
+++ b/src/playlist_list.c
@@ -119,7 +119,8 @@ playlist_list_global_finish(void)
}
static struct playlist_provider *
-playlist_list_open_uri_scheme(const char *uri, bool *tried)
+playlist_list_open_uri_scheme(const char *uri, GMutex *mutex, GCond *cond,
+ bool *tried)
{
char *scheme;
struct playlist_provider *playlist = NULL;
@@ -138,7 +139,8 @@ playlist_list_open_uri_scheme(const char *uri, bool *tried)
if (playlist_plugins_enabled[i] && plugin->open_uri != NULL &&
plugin->schemes != NULL &&
string_array_contains(plugin->schemes, scheme)) {
- playlist = playlist_plugin_open_uri(plugin, uri);
+ playlist = playlist_plugin_open_uri(plugin, uri,
+ mutex, cond);
if (playlist != NULL)
break;
@@ -151,7 +153,8 @@ playlist_list_open_uri_scheme(const char *uri, bool *tried)
}
static struct playlist_provider *
-playlist_list_open_uri_suffix(const char *uri, const bool *tried)
+playlist_list_open_uri_suffix(const char *uri, GMutex *mutex, GCond *cond,
+ const bool *tried)
{
const char *suffix;
struct playlist_provider *playlist = NULL;
@@ -168,7 +171,8 @@ playlist_list_open_uri_suffix(const char *uri, const bool *tried)
if (playlist_plugins_enabled[i] && !tried[i] &&
plugin->open_uri != NULL && plugin->suffixes != NULL &&
string_array_contains(plugin->suffixes, suffix)) {
- playlist = playlist_plugin_open_uri(plugin, uri);
+ playlist = playlist_plugin_open_uri(plugin, uri,
+ mutex, cond);
if (playlist != NULL)
break;
}
@@ -178,7 +182,7 @@ playlist_list_open_uri_suffix(const char *uri, const bool *tried)
}
struct playlist_provider *
-playlist_list_open_uri(const char *uri)
+playlist_list_open_uri(const char *uri, GMutex *mutex, GCond *cond)
{
struct playlist_provider *playlist;
/** this array tracks which plugins have already been tried by
@@ -189,9 +193,10 @@ playlist_list_open_uri(const char *uri)
memset(tried, false, sizeof(tried));
- playlist = playlist_list_open_uri_scheme(uri, tried);
+ playlist = playlist_list_open_uri_scheme(uri, mutex, cond, tried);
if (playlist == NULL)
- playlist = playlist_list_open_uri_suffix(uri, tried);
+ playlist = playlist_list_open_uri_suffix(uri, mutex, cond,
+ tried);
return playlist;
}
@@ -278,16 +283,7 @@ playlist_list_open_stream(struct input_stream *is, const char *uri)
const char *suffix;
struct playlist_provider *playlist;
- GError *error = NULL;
- while (!is->ready) {
- int ret = input_stream_buffer(is, &error);
- if (ret < 0) {
- input_stream_close(is);
- g_warning("%s", error->message);
- g_error_free(error);
- return NULL;
- }
- }
+ input_stream_lock_wait_ready(is);
if (is->mime != NULL) {
playlist = playlist_list_open_stream_mime(is);
@@ -322,7 +318,8 @@ playlist_suffix_supported(const char *suffix)
}
struct playlist_provider *
-playlist_list_open_path(const char *path_fs, struct input_stream **is_r)
+playlist_list_open_path(const char *path_fs, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
GError *error = NULL;
const char *suffix;
@@ -335,7 +332,7 @@ playlist_list_open_path(const char *path_fs, struct input_stream **is_r)
if (suffix == NULL || !playlist_suffix_supported(suffix))
return NULL;
- is = input_stream_open(path_fs, &error);
+ is = input_stream_open(path_fs, mutex, cond, &error);
if (is == NULL) {
if (error != NULL) {
g_warning("%s", error->message);
@@ -345,15 +342,7 @@ playlist_list_open_path(const char *path_fs, struct input_stream **is_r)
return NULL;
}
- while (!is->ready) {
- int ret = input_stream_buffer(is, &error);
- if (ret < 0) {
- input_stream_close(is);
- g_warning("%s", error->message);
- g_error_free(error);
- return NULL;
- }
- }
+ input_stream_lock_wait_ready(is);
playlist = playlist_list_open_stream_suffix(is, suffix);
if (playlist != NULL)
diff --git a/src/playlist_list.h b/src/playlist_list.h
index 69eb26b17..4a2485303 100644
--- a/src/playlist_list.h
+++ b/src/playlist_list.h
@@ -20,6 +20,8 @@
#ifndef MPD_PLAYLIST_LIST_H
#define MPD_PLAYLIST_LIST_H
+#include <glib.h>
+
#include <stdbool.h>
struct playlist_provider;
@@ -41,7 +43,7 @@ playlist_list_global_finish(void);
* Opens a playlist by its URI.
*/
struct playlist_provider *
-playlist_list_open_uri(const char *uri);
+playlist_list_open_uri(const char *uri, GMutex *mutex, GCond *cond);
/**
* Opens a playlist from an input stream.
@@ -69,6 +71,7 @@ playlist_suffix_supported(const char *suffix);
* @return a playlist, or NULL on error
*/
struct playlist_provider *
-playlist_list_open_path(const char *path_fs, struct input_stream **is_r);
+playlist_list_open_path(const char *path_fs, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r);
#endif
diff --git a/src/playlist_mapper.c b/src/playlist_mapper.c
index 824b40e0e..13adb80d0 100644
--- a/src/playlist_mapper.c
+++ b/src/playlist_mapper.c
@@ -27,15 +27,16 @@
#include <assert.h>
static struct playlist_provider *
-playlist_open_path(const char *path_fs, struct input_stream **is_r)
+playlist_open_path(const char *path_fs, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
struct playlist_provider *playlist;
- playlist = playlist_list_open_uri(path_fs);
+ playlist = playlist_list_open_uri(path_fs, mutex, cond);
if (playlist != NULL)
*is_r = NULL;
else
- playlist = playlist_list_open_path(path_fs, is_r);
+ playlist = playlist_list_open_path(path_fs, mutex, cond, is_r);
return playlist;
}
@@ -44,7 +45,8 @@ playlist_open_path(const char *path_fs, struct input_stream **is_r)
* Load a playlist from the configured playlist directory.
*/
static struct playlist_provider *
-playlist_open_in_playlist_dir(const char *uri, struct input_stream **is_r)
+playlist_open_in_playlist_dir(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
char *path_fs;
@@ -56,7 +58,8 @@ playlist_open_in_playlist_dir(const char *uri, struct input_stream **is_r)
path_fs = g_build_filename(playlist_directory_fs, uri, NULL);
- struct playlist_provider *playlist = playlist_open_path(path_fs, is_r);
+ struct playlist_provider *playlist =
+ playlist_open_path(path_fs, mutex, cond, is_r);
g_free(path_fs);
return playlist;
@@ -66,7 +69,8 @@ playlist_open_in_playlist_dir(const char *uri, struct input_stream **is_r)
* Load a playlist from the configured music directory.
*/
static struct playlist_provider *
-playlist_open_in_music_dir(const char *uri, struct input_stream **is_r)
+playlist_open_in_music_dir(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
char *path_fs;
@@ -76,25 +80,28 @@ playlist_open_in_music_dir(const char *uri, struct input_stream **is_r)
if (path_fs == NULL)
return NULL;
- struct playlist_provider *playlist = playlist_open_path(path_fs, is_r);
+ struct playlist_provider *playlist =
+ playlist_open_path(path_fs, mutex, cond, is_r);
g_free(path_fs);
return playlist;
}
struct playlist_provider *
-playlist_mapper_open(const char *uri, struct input_stream **is_r)
+playlist_mapper_open(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r)
{
struct playlist_provider *playlist;
if (spl_valid_name(uri)) {
- playlist = playlist_open_in_playlist_dir(uri, is_r);
+ playlist = playlist_open_in_playlist_dir(uri, mutex, cond,
+ is_r);
if (playlist != NULL)
return playlist;
}
if (uri_safe_local(uri)) {
- playlist = playlist_open_in_music_dir(uri, is_r);
+ playlist = playlist_open_in_music_dir(uri, mutex, cond, is_r);
if (playlist != NULL)
return playlist;
}
diff --git a/src/playlist_mapper.h b/src/playlist_mapper.h
index ab8ba982c..9a7187d93 100644
--- a/src/playlist_mapper.h
+++ b/src/playlist_mapper.h
@@ -20,6 +20,8 @@
#ifndef MPD_PLAYLIST_MAPPER_H
#define MPD_PLAYLIST_MAPPER_H
+#include <glib.h>
+
struct input_stream;
/**
@@ -31,6 +33,7 @@ struct input_stream;
* freed
*/
struct playlist_provider *
-playlist_mapper_open(const char *uri, struct input_stream **is_r);
+playlist_mapper_open(const char *uri, GMutex *mutex, GCond *cond,
+ struct input_stream **is_r);
#endif
diff --git a/src/playlist_plugin.h b/src/playlist_plugin.h
index d7d7c7769..a27f651c0 100644
--- a/src/playlist_plugin.h
+++ b/src/playlist_plugin.h
@@ -20,6 +20,8 @@
#ifndef MPD_PLAYLIST_PLUGIN_H
#define MPD_PLAYLIST_PLUGIN_H
+#include <glib.h>
+
#include <stdbool.h>
#include <stddef.h>
@@ -64,7 +66,8 @@ struct playlist_plugin {
* Opens the playlist on the specified URI. This URI has
* either matched one of the schemes or one of the suffixes.
*/
- struct playlist_provider *(*open_uri)(const char *uri);
+ struct playlist_provider *(*open_uri)(const char *uri,
+ GMutex *mutex, GCond *cond);
/**
* Opens the playlist in the specified input stream. It has
@@ -110,9 +113,10 @@ playlist_plugin_finish(const struct playlist_plugin *plugin)
}
static inline struct playlist_provider *
-playlist_plugin_open_uri(const struct playlist_plugin *plugin, const char *uri)
+playlist_plugin_open_uri(const struct playlist_plugin *plugin, const char *uri,
+ GMutex *mutex, GCond *cond)
{
- return plugin->open_uri(uri);
+ return plugin->open_uri(uri, mutex, cond);
}
static inline struct playlist_provider *
diff --git a/src/playlist_print.c b/src/playlist_print.c
index 9962ffc35..a6bf84ccd 100644
--- a/src/playlist_print.c
+++ b/src/playlist_print.c
@@ -171,10 +171,17 @@ playlist_provider_print(struct client *client, const char *uri,
bool
playlist_file_print(struct client *client, const char *uri, bool detail)
{
+ GMutex *mutex = g_mutex_new();
+ GCond *cond = g_cond_new();
+
struct input_stream *is;
- struct playlist_provider *playlist = playlist_open_any(uri, &is);
- if (playlist == NULL)
+ struct playlist_provider *playlist =
+ playlist_open_any(uri, mutex, cond, &is);
+ if (playlist == NULL) {
+ g_cond_free(cond);
+ g_mutex_free(mutex);
return false;
+ }
playlist_provider_print(client, uri, playlist, detail);
playlist_plugin_close(playlist);
@@ -182,5 +189,8 @@ playlist_file_print(struct client *client, const char *uri, bool detail)
if (is != NULL)
input_stream_close(is);
+ g_cond_free(cond);
+ g_mutex_free(mutex);
+
return true;
}
diff --git a/src/playlist_queue.c b/src/playlist_queue.c
index d368fcb23..33885ae21 100644
--- a/src/playlist_queue.c
+++ b/src/playlist_queue.c
@@ -59,10 +59,17 @@ playlist_open_into_queue(const char *uri,
struct playlist *dest, struct player_control *pc,
bool secure)
{
+ GMutex *mutex = g_mutex_new();
+ GCond *cond = g_cond_new();
+
struct input_stream *is;
- struct playlist_provider *playlist = playlist_open_any(uri, &is);
- if (playlist == NULL)
+ struct playlist_provider *playlist =
+ playlist_open_any(uri, mutex, cond, &is);
+ if (playlist == NULL) {
+ g_cond_free(cond);
+ g_mutex_free(mutex);
return PLAYLIST_RESULT_NO_SUCH_LIST;
+ }
enum playlist_result result =
playlist_load_into_queue(uri, playlist, dest, pc, secure);
@@ -71,5 +78,8 @@ playlist_open_into_queue(const char *uri,
if (is != NULL)
input_stream_close(is);
+ g_cond_free(cond);
+ g_mutex_free(mutex);
+
return result;
}
diff --git a/src/song_update.c b/src/song_update.c
index e2a845eef..e7279a6f6 100644
--- a/src/song_update.c
+++ b/src/song_update.c
@@ -131,6 +131,9 @@ song_file_update(struct song *song)
song->mtime = st.st_mtime;
+ GMutex *mutex = NULL;
+ GCond *cond;
+
do {
/* load file tag */
song->tag = decoder_plugin_tag_dup(plugin, path_fs);
@@ -141,8 +144,12 @@ song_file_update(struct song *song)
if (plugin->stream_tag != NULL) {
/* open the input_stream (if not already
open) */
- if (is == NULL)
- is = input_stream_open(path_fs, NULL);
+ if (is == NULL) {
+ mutex = g_mutex_new();
+ cond = g_cond_new();
+ is = input_stream_open(path_fs, mutex, cond,
+ NULL);
+ }
/* now try the stream_tag() method */
if (is != NULL) {
@@ -151,7 +158,7 @@ song_file_update(struct song *song)
if (song->tag != NULL)
break;
- input_stream_seek(is, 0, SEEK_SET, NULL);
+ input_stream_lock_seek(is, 0, SEEK_SET, NULL);
}
}
@@ -161,6 +168,11 @@ song_file_update(struct song *song)
if (is != NULL)
input_stream_close(is);
+ if (mutex != NULL) {
+ g_cond_free(cond);
+ g_mutex_free(mutex);
+ }
+
if (song->tag != NULL && tag_is_empty(song->tag))
song->tag = tag_fallback(path_fs, song->tag);
diff --git a/src/text_input_stream.c b/src/text_input_stream.c
index c71e113c7..5da217eae 100644
--- a/src/text_input_stream.c
+++ b/src/text_input_stream.c
@@ -68,8 +68,8 @@ text_input_stream_read(struct text_input_stream *tis)
do {
dest = fifo_buffer_write(tis->buffer, &length);
if (dest != NULL) {
- nbytes = input_stream_read(tis->is, dest, length,
- &error);
+ nbytes = input_stream_lock_read(tis->is, dest, length,
+ &error);
if (nbytes > 0)
fifo_buffer_append(tis->buffer, nbytes);
else if (error != NULL) {