aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/input/curl_input_plugin.c98
1 files changed, 69 insertions, 29 deletions
diff --git a/src/input/curl_input_plugin.c b/src/input/curl_input_plugin.c
index 1969ceb7e..c377b6426 100644
--- a/src/input/curl_input_plugin.c
+++ b/src/input/curl_input_plugin.c
@@ -178,11 +178,13 @@ curl_quark(void)
/**
* Find a request by its CURL "easy" handle.
*
- * The caller must lock the mutex.
+ * Runs in the I/O thread. No lock needed.
*/
static struct input_curl *
input_curl_find_request(CURL *easy)
{
+ assert(io_thread_inside());
+
for (GSList *i = curl.requests; i != NULL; i = g_slist_next(i)) {
struct input_curl *c = i->data;
if (c->easy == easy)
@@ -197,6 +199,8 @@ input_curl_find_request(CURL *easy)
static gpointer
input_curl_resume(gpointer data)
{
+ assert(io_thread_inside());
+
struct input_curl *c = data;
if (c->paused) {
@@ -240,7 +244,7 @@ input_curl_fd_events(int fd, fd_set *rfds, fd_set *wfds, fd_set *efds)
* Updates all registered GPollFD objects, unregisters old ones,
* registers new ones.
*
- * The caller must lock the mutex. Runs in the I/O thread.
+ * Runs in the I/O thread. No lock needed.
*/
static void
curl_update_fds(void)
@@ -308,14 +312,11 @@ static gboolean
input_curl_dirty_callback(G_GNUC_UNUSED gpointer data)
{
assert(io_thread_inside());
- g_static_mutex_lock(&curl.mutex);
-
assert(curl.dirty_source_id != 0 || curl.requests == NULL);
curl.dirty_source_id = 0;
curl_update_fds();
- g_static_mutex_unlock(&curl.mutex);
return false;
}
@@ -323,7 +324,7 @@ input_curl_dirty_callback(G_GNUC_UNUSED gpointer data)
* Schedule a refresh of curl.fds. Does nothing if that is already
* scheduled.
*
- * The caller must lock the mutex.
+ * No lock needed.
*/
static void
input_curl_schedule_update(void)
@@ -336,9 +337,13 @@ input_curl_schedule_update(void)
io_thread_idle_add(input_curl_dirty_callback, NULL);
}
+/**
+ * Runs in the I/O thread. No lock needed.
+ */
static bool
input_curl_easy_add(struct input_curl *c, GError **error_r)
{
+ assert(io_thread_inside());
assert(c != NULL);
assert(c->easy != NULL);
assert(input_curl_find_request(c->easy) == NULL);
@@ -358,15 +363,50 @@ input_curl_easy_add(struct input_curl *c, GError **error_r)
return true;
}
+struct easy_add_params {
+ struct input_curl *c;
+ GError **error_r;
+};
+
+static gpointer
+input_curl_easy_add_callback(gpointer data)
+{
+ const struct easy_add_params *params = data;
+
+ bool success = input_curl_easy_add(params->c, params->error_r);
+ return GUINT_TO_POINTER(success);
+}
+
+/**
+ * Call input_curl_easy_add() in the I/O thread. May be called from
+ * any thread. Caller must not hold a mutex.
+ */
+static bool
+input_curl_easy_add_indirect(struct input_curl *c, GError **error_r)
+{
+ assert(c != NULL);
+ assert(c->easy != NULL);
+
+ struct easy_add_params params = {
+ .c = c,
+ .error_r = error_r,
+ };
+
+ gpointer result =
+ io_thread_call(input_curl_easy_add_callback, &params);
+ return GPOINTER_TO_UINT(result);
+}
+
/**
* Frees the current "libcurl easy" handle, and everything associated
* with it.
*
- * The caller must lock the mutex.
+ * Runs in the I/O thread.
*/
static void
input_curl_easy_free(struct input_curl *c)
{
+ assert(io_thread_inside());
assert(c != NULL);
if (c->easy == NULL)
@@ -390,13 +430,9 @@ input_curl_easy_free_callback(gpointer data)
{
struct input_curl *c = data;
- g_static_mutex_lock(&curl.mutex);
-
input_curl_easy_free(c);
curl_update_fds();
- g_static_mutex_unlock(&curl.mutex);
-
return NULL;
}
@@ -416,7 +452,7 @@ input_curl_easy_free_indirect(struct input_curl *c)
/**
* Abort and free all HTTP requests.
*
- * The caller must lock the mutex. Runs in the I/O thread.
+ * Runs in the I/O thread. The caller must not hold locks.
*/
static void
input_curl_abort_all_requests(GError *error)
@@ -424,6 +460,8 @@ input_curl_abort_all_requests(GError *error)
assert(io_thread_inside());
assert(error != NULL);
+ g_static_mutex_lock(&curl.mutex);
+
while (curl.requests != NULL) {
struct input_curl *c = curl.requests->data;
assert(c->postponed_error == NULL);
@@ -436,12 +474,13 @@ input_curl_abort_all_requests(GError *error)
g_error_free(error);
g_cond_broadcast(curl.cond);
+ g_static_mutex_unlock(&curl.mutex);
}
/**
* A HTTP request is finished.
*
- * The caller must lock the mutex. Runs in the I/O thread.
+ * Runs in the I/O thread. The caller must not hold locks.
*/
static void
input_curl_request_done(struct input_curl *c, CURLcode result, long status)
@@ -451,6 +490,8 @@ input_curl_request_done(struct input_curl *c, CURLcode result, long status)
assert(c->easy == NULL);
assert(c->postponed_error == NULL);
+ g_static_mutex_lock(&curl.mutex);
+
if (result != CURLE_OK) {
c->postponed_error = g_error_new(curl_quark(), result,
"curl failed: %s",
@@ -463,6 +504,7 @@ input_curl_request_done(struct input_curl *c, CURLcode result, long status)
c->base.ready = true;
g_cond_broadcast(curl.cond);
+ g_static_mutex_unlock(&curl.mutex);
}
static void
@@ -481,7 +523,7 @@ input_curl_handle_done(CURL *easy_handle, CURLcode result)
/**
* Check for finished HTTP responses.
*
- * The caller must lock the mutex. Runs in the I/O thread.
+ * Runs in the I/O thread. The caller must not hold locks.
*/
static void
input_curl_info_read(void)
@@ -501,7 +543,7 @@ input_curl_info_read(void)
/**
* Give control to CURL.
*
- * The caller must lock the mutex. Runs in the I/O thread.
+ * Runs in the I/O thread. The caller must not hold locks.
*/
static bool
input_curl_perform(void)
@@ -611,13 +653,9 @@ input_curl_source_dispatch(G_GNUC_UNUSED GSource *source,
G_GNUC_UNUSED GSourceFunc callback,
G_GNUC_UNUSED gpointer user_data)
{
- g_static_mutex_lock(&curl.mutex);
-
if (input_curl_perform())
input_curl_info_read();
- g_static_mutex_unlock(&curl.mutex);
-
return true;
}
@@ -1074,9 +1112,12 @@ input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
if (size == 0)
return 0;
+ g_static_mutex_lock(&curl.mutex);
+
#if LIBCURL_VERSION_NUM >= 0x071200
if (curl_total_buffer_size(c) + size >= CURL_MAX_BUFFERED) {
c->paused = true;
+ g_static_mutex_unlock(&curl.mutex);
return CURL_WRITEFUNC_PAUSE;
}
#endif
@@ -1091,6 +1132,7 @@ input_curl_writefunction(void *ptr, size_t size, size_t nmemb, void *stream)
c->base.ready = true;
g_cond_broadcast(curl.cond);
+ g_static_mutex_unlock(&curl.mutex);
return size;
}
@@ -1197,6 +1239,8 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
/* check if we can fast-forward the buffer */
+ g_static_mutex_lock(&curl.mutex);
+
while (offset > is->offset && !g_queue_is_empty(c->buffers)) {
struct buffer *buffer;
size_t length;
@@ -1214,6 +1258,8 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
is->offset += length;
}
+ g_static_mutex_unlock(&curl.mutex);
+
if (offset == is->offset)
return true;
@@ -1241,14 +1287,12 @@ input_curl_seek(struct input_stream *is, goffset offset, int whence,
curl_easy_setopt(c->easy, CURLOPT_RANGE, c->range);
}
- g_static_mutex_lock(&curl.mutex);
-
c->base.ready = false;
- if (!input_curl_easy_add(c, error_r)) {
- g_static_mutex_unlock(&curl.mutex);
+ if (!input_curl_easy_add_indirect(c, error_r))
return false;
- }
+
+ g_static_mutex_lock(&curl.mutex);
while (!c->base.ready)
g_cond_wait(curl.cond, g_static_mutex_get_mutex(&curl.mutex));
@@ -1291,15 +1335,11 @@ input_curl_open(const char *url, GError **error_r)
return NULL;
}
- g_static_mutex_lock(&curl.mutex);
- if (!input_curl_easy_add(c, error_r)) {
- g_static_mutex_unlock(&curl.mutex);
+ if (!input_curl_easy_add_indirect(c, error_r)) {
input_curl_free(c);
return NULL;
}
- g_static_mutex_unlock(&curl.mutex);
-
return &c->base;
}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-16 05:43:37 +0000