diff options
57 files changed, 653 insertions, 377 deletions
@@ -5,6 +5,7 @@ ver 0.17 (2011/??/??) * input: - cdio_paranoia: new input plugin to play audio CDs - curl: enable CURLOPT_NETRC + - curl: non-blocking I/O - soup: new input plugin based on libsoup - ffmpeg: support libavformat 0.7 * decoder: 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) { diff --git a/test/dump_playlist.c b/test/dump_playlist.c index bf3fed7c9..079fdeac0 100644 --- a/test/dump_playlist.c +++ b/test/dump_playlist.c @@ -92,11 +92,14 @@ int main(int argc, char **argv) /* open the playlist */ - playlist = playlist_list_open_uri(uri); + GMutex *mutex = g_mutex_new(); + GCond *cond = g_cond_new(); + + playlist = playlist_list_open_uri(uri, mutex, cond); if (playlist == NULL) { /* open the stream and wait until it becomes ready */ - is = input_stream_open(uri, &error); + is = input_stream_open(uri, mutex, cond, &error); if (is == NULL) { if (error != NULL) { g_warning("%s", error->message); @@ -106,19 +109,7 @@ int main(int argc, char **argv) return 2; } - while (!is->ready) { - int ret = input_stream_buffer(is, &error); - if (ret < 0) { - /* error */ - g_warning("%s", error->message); - g_error_free(error); - return 2; - } - - if (ret == 0) - /* nothing was buffered - wait */ - g_usleep(10000); - } + input_stream_lock_wait_ready(is); /* open the playlist */ @@ -157,6 +148,10 @@ int main(int argc, char **argv) playlist_plugin_close(playlist); if (is != NULL) input_stream_close(is); + + g_cond_free(cond); + g_mutex_free(mutex); + playlist_list_global_finish(); input_stream_global_finish(); io_thread_deinit(); diff --git a/test/read_tags.c b/test/read_tags.c index c2e3b2caa..1d742e9ec 100644 --- a/test/read_tags.c +++ b/test/read_tags.c @@ -91,7 +91,7 @@ decoder_read(G_GNUC_UNUSED struct decoder *decoder, struct input_stream *is, void *buffer, size_t length) { - return input_stream_read(is, buffer, length, NULL); + return input_stream_lock_read(is, buffer, length, NULL); } void @@ -193,7 +193,11 @@ int main(int argc, char **argv) tag = decoder_plugin_tag_dup(plugin, path); if (tag == NULL && plugin->stream_tag != NULL) { - struct input_stream *is = input_stream_open(path, &error); + GMutex *mutex = g_mutex_new(); + GCond *cond = g_cond_new(); + + struct input_stream *is = + input_stream_open(path, mutex, cond, &error); if (is == NULL) { g_printerr("Failed to open %s: %s\n", @@ -204,6 +208,9 @@ int main(int argc, char **argv) tag = decoder_plugin_stream_tag(plugin, is); input_stream_close(is); + + g_cond_free(cond); + g_mutex_free(mutex); } decoder_plugin_deinit_all(); diff --git a/test/run_decoder.c b/test/run_decoder.c index efc246f55..455b73ce7 100644 --- a/test/run_decoder.c +++ b/test/run_decoder.c @@ -112,7 +112,7 @@ decoder_read(G_GNUC_UNUSED struct decoder *decoder, struct input_stream *is, void *buffer, size_t length) { - return input_stream_read(is, buffer, length, NULL); + return input_stream_lock_read(is, buffer, length, NULL); } void @@ -209,8 +209,11 @@ int main(int argc, char **argv) decoder_plugin_file_decode(decoder.plugin, &decoder, decoder.uri); } else if (decoder.plugin->stream_decode != NULL) { + GMutex *mutex = g_mutex_new(); + GCond *cond = g_cond_new(); + struct input_stream *is = - input_stream_open(decoder.uri, &error); + input_stream_open(decoder.uri, mutex, cond, &error); if (is == NULL) { if (error != NULL) { g_warning("%s", error->message); @@ -224,6 +227,9 @@ int main(int argc, char **argv) decoder_plugin_stream_decode(decoder.plugin, &decoder, is); input_stream_close(is); + + g_cond_free(cond); + g_mutex_free(mutex); } else { g_printerr("Decoder plugin is not usable\n"); return 1; diff --git a/test/run_input.c b/test/run_input.c index 651d36480..0fe5a01f0 100644 --- a/test/run_input.c +++ b/test/run_input.c @@ -53,20 +53,17 @@ dump_input_stream(struct input_stream *is) size_t num_read; ssize_t num_written; + g_mutex_lock(is->mutex); + /* wait until the stream becomes ready */ - while (!is->ready) { - int ret = input_stream_buffer(is, &error); - if (ret < 0) { - /* error */ - g_warning("%s", error->message); - g_error_free(error); - return 2; - } + input_stream_wait_ready(is); - if (ret == 0) - /* nothing was buffered - wait */ - g_usleep(10000); + if (!input_stream_check(is, &error)) { + g_warning("%s", error->message); + g_error_free(error); + g_mutex_unlock(is->mutex); + return EXIT_FAILURE; } /* print meta data */ @@ -103,9 +100,12 @@ dump_input_stream(struct input_stream *is) if (!input_stream_check(is, &error)) { g_warning("%s", error->message); g_error_free(error); + g_mutex_unlock(is->mutex); return EXIT_FAILURE; } + g_mutex_unlock(is->mutex); + return 0; } @@ -149,7 +149,10 @@ int main(int argc, char **argv) /* open the stream and dump it */ - is = input_stream_open(argv[1], &error); + GMutex *mutex = g_mutex_new(); + GCond *cond = g_cond_new(); + + is = input_stream_open(argv[1], mutex, cond, &error); if (is != NULL) { ret = dump_input_stream(is); input_stream_close(is); @@ -162,6 +165,9 @@ int main(int argc, char **argv) ret = 2; } + g_cond_free(cond); + g_mutex_free(mutex); + /* deinitialize everything */ input_stream_global_finish(); |