mirrors/ffmpeg
0
0 Fork 0
mirror of https://git.ffmpeg.org/ffmpeg.git synced 2026-10-28 18:15:24 +08:00
Code Activity
Files
master
ffmpeg/libavformat/libcurl.c
T
Kacper Michajłow 6aff325497 avformat/libcurl: keep requesting when the file size is unknown 
Keep requesting from where each reply stopped when the file size is
unknown ("Content-Range: bytes S-E/*"). Treat a 416 on such a follow-up
request as EOF and take the size from its "bytes */N". A 416 that states
a size past the requested offset is an error. A compressed reply is not
continued, as its range counts compressed bytes.

Signed-off-by: Kacper Michajłow <kasper93@gmail.com>
2026-10-12 22:42:24 +02:00

1779 lines
61 KiB
C
Raw Permalink Blame History

/*
* libcurl based HTTP(S) protocol
* Copyright (C) 2026 Kacper Michajłow
*
* This file is part of FFmpeg.
*
* FFmpeg is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* FFmpeg is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with FFmpeg; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
*/
#include "config_components.h"
#include <curl/curl.h>
#include <errno.h>
#include <inttypes.h>
#include <limits.h>
#include <stdlib.h>
#include <string.h>
#include "libavutil/avstring.h"
#include "libavutil/bprint.h"
#include "libavutil/dict.h"
#include "libavutil/error.h"
#include "libavutil/fifo.h"
#include "libavutil/log.h"
#include "libavutil/macros.h"
#include "libavutil/mem.h"
#include "libavutil/opt.h"
#include "libavutil/thread.h"
#include "libavutil/time.h"
#include "avformat.h"
#include "http.h"
#include "internal.h"
#include "network.h"
#include "url.h"
#include "version.h"
#define DEFAULT_USER_AGENT "Lavf/" AV_STRINGIFY(LIBAVFORMAT_VERSION)
#define CURL_DEFAULT_BUFFER_SIZE (4 << 20)
/* Blocking waits wake up this often so url_read()/open can poll the interrupt
* callback. */
#define CURL_WAIT_US 100000
/* Cap on the accumulated "Icy-*" reply headers exported to the caller. */
#define ICY_MAX_HEADERS 65536
/* Largest in-band metadata block: the length byte counts 16 byte units. */
#define ICY_MAX_BLOCK (255 * 16)
/* How many microseconds to wait before retrying a failed request; grows
* exponentially (2^) with each consecutive failure up to MAX_US. Note that the
* first retry is always immediate. */
#define CURL_RETRY_BASE_US 1000000
#define CURL_RETRY_MAX_US 60000000
typedef struct CurlContext CurlContext;
enum cmd_kind {
CMD_ADD, /* add the easy handle to the multi and start the transfer */
CMD_REMOVE, /* remove the easy handle from the multi */
CMD_UNPAUSE, /* resume a transfer paused because the FIFO was full */
CMD_SEEK, /* restart the transfer at a new byte offset */
};
typedef struct CurlCmd {
enum cmd_kind kind;
CurlContext *ctx;
int64_t pos; /* CMD_SEEK target offset */
int sync; /* caller waits for completion */
int done;
struct CurlCmd *next;
} CurlCmd;
typedef struct CurlLoop {
AVFormatContext *avfc; /* owning context (if any) */
pthread_t thread;
CURLM *multi;
CURLSH *share; /* shared cookies/HSTS */
int has_cookies; /* libcurl was built with cookie support */
pthread_mutex_t mutex; /* guards the command queue, exit, share and cmd->done */
pthread_cond_t cond; /* signaled when a sync command completes */
CurlCmd *cmd_head, *cmd_tail;
int exit;
/* Connection statistics (updated by loop thread) */
int64_t total_bytes;
int64_t total_time_us;
int num_connections;
int num_redirects;
int num_requests;
int num_errors;
} CurlLoop;
struct CurlContext {
const AVClass *class;
URLContext *h;
CurlLoop *loop;
int private_loop; /* loop is owned by this context (not shared) */
CURL *easy;
struct curl_slist *header_list;
struct curl_slist *alias_list;
/* AVOptions. */
char *user_agent;
char *referer;
char *headers;
char *http_proxy;
char *cookies;
char *ca_file;
char *cert_file;
char *key_file;
char *location; /* effective URL after redirects (output) */
int64_t off; /* initial byte offset */
int64_t end_off; /* exclusive upper byte bound (0 = none) */
int tls_verify;
int seekable_opt;
int connect_timeout;
int max_redirects;
int multiple_requests;
int http_version;
int64_t buffer_size;
int64_t request_size;
int64_t initial_request_size;
int64_t short_seek_size;
int max_retries;
int icy;
int icy_status;
char *icy_metadata_headers; /* "Icy-*" reply headers (output) */
char *icy_metadata_packet; /* last in-band block (output) */
AVDictionary *metadata; /* ICY metadata (output) */
/* URL thread bookkeeping, not touched by loop thread */
int64_t logical_pos; /* next byte url_read() will return, caller side */
int retry_count; /* consecutive recoverable failures */
int64_t retry_time; /* timestamp of next retry */
int64_t icy_data_read; /* payload bytes since the last block, caller side */
int icy_block_len; /* -1 while the length byte is pending */
int icy_block_filled;
uint8_t icy_block[ICY_MAX_BLOCK + 1];
/* Producer bookkeeping, touched only by the loop thread. */
int active; /* currently added to the multi */
int64_t request_start; /* absolute offset the current request began at */
int64_t request_received;/* bytes delivered in the current request */
int64_t request_end; /* expected end of request, or -1 if unknown */
int is_initial; /* using reduced request size */
int seek_queued; /* soft seeking; drain remaining bytes until done */
/* Per-response-block header scratch, loop thread only. */
int hdr_accept_ranges;
int hdr_compressed;
int64_t hdr_content_start; /* inclusive start, or -1 */
int64_t hdr_content_end; /* inclusive end, or -1 */
int64_t hdr_content_total; /* if known, or -1 */
int64_t hdr_content_length; /* if known, or -1 */
AVDictionary *hdr_icy; /* "Icy-*" headers of this block */
int64_t hdr_icy_metaint; /* in-band metadata interval, or -1 */
/* Probe result. Set by the loop thread, read by url_open() once probed. */
int probed;
int probe_ok; /* the first reply can be read */
int stream_ok;
int seekable;
int64_t content_size;
/* Shared transfer state, guarded by mutex. */
pthread_mutex_t mutex;
pthread_cond_t cond;
AVFifo *fifo;
int paused; /* write callback paused, FIFO was full */
int aborted; /* transfer should stop (open was interrupted) */
int status; /* current stream status (AVERROR code) */
CURLcode curl_status; /* corresponding libcurl status code */
int64_t icy_metaint; /* in-band metadata interval, 0 if none */
};
/* Guards lazy creation of a format context's shared loop. */
static AVMutex curl_loop_lock = AV_MUTEX_INITIALIZER;
static int curlcode_to_averror(CURLcode code)
{
switch (code) {
case CURLE_OK: return 0;
case CURLE_URL_MALFORMAT:
case CURLE_UNSUPPORTED_PROTOCOL:
case CURLE_BAD_FUNCTION_ARGUMENT: return AVERROR(EINVAL);
case CURLE_UNKNOWN_OPTION:
case CURLE_NOT_BUILT_IN: return AVERROR(ENOSYS);
case CURLE_COULDNT_RESOLVE_PROXY:
case CURLE_COULDNT_RESOLVE_HOST: return AVERROR(EHOSTUNREACH);
case CURLE_COULDNT_CONNECT: return AVERROR(ECONNREFUSED);
case CURLE_OPERATION_TIMEDOUT: return AVERROR(ETIMEDOUT);
case CURLE_LOGIN_DENIED:
case CURLE_REMOTE_ACCESS_DENIED: return AVERROR(EACCES);
case CURLE_OUT_OF_MEMORY: return AVERROR(ENOMEM);
case CURLE_PEER_FAILED_VERIFICATION:
case CURLE_SSL_CACERT_BADFILE: return AVERROR_INVALIDDATA;
default: return AVERROR(EIO);
}
}
static int curl_setopt_checked(CurlContext *c, const char *name, CURLcode code)
{
if (code == CURLE_OK)
return 0;
av_log(c->h, AV_LOG_ERROR, "Could not set %s: %s\n",
name, curl_easy_strerror(code));
return curlcode_to_averror(code);
}
static int curlmcode_to_averror(CURLMcode code)
{
switch (code) {
case CURLM_OK: return 0;
case CURLM_UNKNOWN_OPTION: return AVERROR(EINVAL);
case CURLM_OUT_OF_MEMORY: return AVERROR(ENOMEM);
default: return AVERROR(EIO);
}
}
static int curlmcode_to_curlcode(CURLMcode code)
{
switch (code) {
case CURLM_OK: return CURLE_OK;
case CURLM_UNKNOWN_OPTION: return CURLE_UNKNOWN_OPTION;
case CURLM_OUT_OF_MEMORY: return CURLE_OUT_OF_MEMORY;
case CURLM_ABORTED_BY_CALLBACK: return CURLE_ABORTED_BY_CALLBACK;
case CURLM_UNRECOVERABLE_POLL: return CURLE_UNRECOVERABLE_POLL;
default: return CURLE_FAILED_INIT;
}
}
static int is_recoverable(CURLcode code)
{
switch (code) {
case CURLE_RECV_ERROR:
case CURLE_SEND_ERROR:
case CURLE_PARTIAL_FILE:
case CURLE_OPERATION_TIMEDOUT:
case CURLE_GOT_NOTHING:
case CURLE_COULDNT_CONNECT:
case CURLE_COULDNT_RESOLVE_HOST:
case CURLE_HTTP2:
case CURLE_HTTP2_STREAM:
return 1;
default:
return 0;
}
}
/* ------------------------------------------------------------------------- */
/* curl callbacks (run on the loop thread) */
/* ------------------------------------------------------------------------- */
static void update_status_locked(CurlContext *c, int status, CURLcode code)
{
if (c->status)
return;
c->status = status;
c->curl_status = code;
}
static size_t write_callback(char *ptr, size_t size, size_t nmemb, void *userdata)
{
CurlContext *c = userdata;
size_t bytes = size * nmemb;
size_t space;
pthread_mutex_lock(&c->mutex);
if (c->aborted || !c->stream_ok) {
pthread_mutex_unlock(&c->mutex);
return CURL_WRITEFUNC_ERROR;
}
if (c->seek_queued) {
pthread_mutex_unlock(&c->mutex);
return bytes; /* discard */
}
/* Prevent overflow / non-addressable byte ranges */
if (bytes > INT64_MAX - c->request_start - c->request_received) {
av_log(c->h, AV_LOG_ERROR, "Server sent back more data than addressable "
"at offset %"PRId64"\n", c->request_start);
c->loop->num_errors++;
c->stream_ok = 0;
if (!c->status)
c->status = AVERROR(ERANGE);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return CURL_WRITEFUNC_ERROR;
}
space = av_fifo_can_write(c->fifo);
if (space < bytes) {
/* pause the transfer and wait for the consumer to drain. */
c->paused = 1;
pthread_mutex_unlock(&c->mutex);
return CURL_WRITEFUNC_PAUSE;
}
av_fifo_write(c->fifo, ptr, bytes);
c->paused = 0;
c->request_received += bytes;
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return bytes;
}
/* Return 1 if an offset was successfully parsed, 0 otherwise */
static int parse_offset(const char *str, int64_t *out, const char **ptr)
{
if (!av_isdigit(str[0]))
return 0;
errno = 0;
char *end;
int64_t val = strtoll(str, &end, 10);
if (errno == ERANGE)
return 0;
*out = val;
*ptr = end;
return 1;
}
/* "bytes $from-$to/$document_size" */
static void parse_content_range(CurlContext *c, const char *v)
{
int64_t start = -1, end = -1, total = -1;
while (av_isspace(*v))
v++;
if (!av_stristart(v, "bytes ", &v))
return;
if (!av_strstart(v, "*", &v)) {
if (!parse_offset(v, &start, &v) ||
!av_strstart(v, "-", &v) ||
!parse_offset(v, &end, &v))
return;
}
if (!av_strstart(v, "/", &v))
return;
if (!av_strstart(v, "*", &v) && !parse_offset(v, &total, &v))
return;
while (av_isspace(*v))
v++;
if (v[0] || (total < 0 && start < 0))
return; // reject trailing bytes or "*/*"
/* only set these fields if the header was recognized; ignore otherwise */
c->hdr_content_start = start;
c->hdr_content_end = end;
c->hdr_content_total = total;
}
/* Parse a decimal header value, bounded by len since curl does not promise a
* NUL terminated header buffer. Returns -1 if absent, malformed or too large. */
static int64_t parse_metaint(const char *p, size_t len)
{
int64_t v = 0;
size_t i = 0;
while (i < len && av_isspace(p[i]))
i++;
if (i == len || !av_isdigit(p[i]))
return -1;
for (; i < len && av_isdigit(p[i]); i++) {
if (v > (INT_MAX - (p[i] - '0')) / 10)
return -1;
v = v * 10 + (p[i] - '0');
}
// Reject 4junk
while (i < len && av_isspace(p[i]))
i++;
if (i != len)
return -1;
return v;
}
/* Store one "Tag: value" reply header, tolerating a missing space after the
* colon and the CRLF curl leaves on the line. */
static void store_icy_header(AVDictionary **dict, const char *ptr, size_t len)
{
const char *colon = memchr(ptr, ':', len);
const char *val, *end = ptr + len;
char *key, *value;
if (!colon)
return;
val = colon + 1;
while (val < end && av_isspace(*val))
val++;
while (end > val && (end[-1] == '\r' || end[-1] == '\n'))
end--;
key = av_strndup(ptr, colon - ptr);
value = av_strndup(val, end - val);
/* Multikey keeps repeated headers in arrival order, as http.c reports them. */
if (key && value)
av_dict_set(dict, key, value, AV_DICT_DONT_STRDUP_KEY |
AV_DICT_DONT_STRDUP_VAL | AV_DICT_MULTIKEY);
else {
av_free(key);
av_free(value);
}
}
/* Export the reply's "Icy-*" headers. Called once, before probed is set, so
* that from then on only the caller thread touches the exported metadata. */
static void commit_icy_headers(CurlContext *c)
{
const AVDictionaryEntry *e = NULL;
AVBPrint bp;
if (!c->hdr_icy)
return;
av_bprint_init(&bp, 0, ICY_MAX_HEADERS);
while ((e = av_dict_iterate(c->hdr_icy, e)))
av_bprintf(&bp, "%s: %s\n", e->key, e->value);
av_freep(&c->icy_metadata_headers);
if (av_bprint_finalize(&bp, &c->icy_metadata_headers) < 0)
c->icy_metadata_headers = NULL;
av_dict_copy(&c->metadata, c->hdr_icy, 0);
}
static int verify_content_range(CurlContext *c, int64_t start, int64_t end,
int64_t total)
{
if (start < 0 || start != c->request_start) {
av_log(c->h, AV_LOG_ERROR, "Server sent back unexpected reply "
"with offset %"PRId64" (expected %"PRId64")\n",
start, c->request_start);
return 0;
}
if (end >= 0 && end < start) {
av_log(c->h, AV_LOG_ERROR, "Server sent back backwards content range "
"%"PRId64"-%"PRId64"\n", start, end);
return 0;
}
if (total >= 0 && (start >= total || end >= total)) {
av_log(c->h, AV_LOG_ERROR, "Server sent back content range "
"%"PRId64"-%"PRId64" that exceeds the total size "
"%"PRId64"\n", start, end, total);
return 0;
}
if (c->hdr_content_length >= 0 && end >= 0 &&
(c->hdr_content_length - 1 > INT64_MAX - start ||
start + (c->hdr_content_length - 1) != end))
{
av_log(c->h, AV_LOG_ERROR, "Server sent back content range "
"%"PRId64"-%"PRId64" that doesn't match the content length "
"%"PRId64"\n", start, end, c->hdr_content_length);
return 0;
}
return 1;
}
static size_t header_callback(char *ptr, size_t size, size_t nitems, void *userdata)
{
CurlContext *c = userdata;
size_t len = size * nitems;
size_t n = len;
long status = 0;
/* Only "ICY 200" is aliased, so these two prefixes cover every status line
* curl lets through. */
if (av_strncasecmp(ptr, "HTTP/", 5) == 0 ||
av_strncasecmp(ptr, "ICY ", 4) == 0) {
c->hdr_accept_ranges = 0;
c->hdr_compressed = 0;
c->hdr_content_start = -1;
c->hdr_content_end = -1;
c->hdr_content_total = -1;
c->hdr_content_length = -1;
c->hdr_icy_metaint = -1;
av_dict_free(&c->hdr_icy);
return len;
}
if (av_strncasecmp(ptr, "Accept-Ranges:", 14) == 0) {
c->hdr_accept_ranges = !!av_stristr(ptr + 14, "bytes");
return len;
}
if (av_strncasecmp(ptr, "Content-Encoding:", 17) == 0) {
c->hdr_compressed = !av_stristr(ptr + 17, "identity");
return len;
}
if (av_strncasecmp(ptr, "Content-Range:", 14) == 0) {
parse_content_range(c, ptr + 14);
return len;
}
if (av_strncasecmp(ptr, "icy-metaint:", 12) == 0) {
c->hdr_icy_metaint = parse_metaint(ptr + 12, len - 12);
return len;
}
/* Collected per block so that headers from a redirect hop, which the
* interim early return below discards, do not reach the caller. */
if (av_strncasecmp(ptr, "icy-", 4) == 0) {
store_icy_header(&c->hdr_icy, ptr, len);
return len;
}
/* Otherwise act only on the blank line that terminates the header block. */
while (n && (ptr[n - 1] == '\r' || ptr[n - 1] == '\n'))
n--;
if (n)
return len;
curl_easy_getinfo(c->easy, CURLINFO_RESPONSE_CODE, &status);
/* Interim (1xx) and redirect (3xx) responses produce an intermediate header
* block, wait for the final one. */
if (status < 200 || (status >= 300 && status < 400))
return len;
curl_off_t cl = -1;
if (curl_easy_getinfo(c->easy, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &cl) == CURLE_OK)
c->hdr_content_length = cl >= 0 ? cl : -1;
pthread_mutex_lock(&c->mutex);
if (status >= 200 && status < 300) {
int64_t content_start = status == 206 ? c->hdr_content_start : 0;
int64_t content_end = status == 206 ? c->hdr_content_end : -1;
int64_t content_total = status == 206 ? c->hdr_content_total : c->hdr_content_length;
if (content_end < 0 && content_total > 0)
content_end = content_total - 1;
if (!verify_content_range(c, content_start, content_end, content_total)) {
c->loop->num_errors++;
c->stream_ok = 0;
update_status_locked(c, AVERROR(EIO), CURLE_OK);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return len;
}
/* Don't unlearn a known size when a reply omits it. */
if (!c->hdr_compressed && content_total >= 0)
c->content_size = content_total;
if (!c->hdr_compressed) {
int64_t total = content_total;
if (total < 0 && status != 206)
total = c->hdr_content_length;
/* Don't unlearn a known size when a reply omits it. */
if (total >= 0)
c->content_size = total;
if (content_end < 0)
content_end = c->content_size > 0 ? c->content_size - 1 : -1;
}
c->stream_ok = 1;
c->request_end = c->hdr_compressed ? -1 : content_end;
/* Capture the post-redirect URL, this is exposed as "location" AVOption
* for compatibility with http.c. */
if (!c->probed) {
const char *eff = NULL;
if (curl_easy_getinfo(c->easy, CURLINFO_EFFECTIVE_URL, &eff) == CURLE_OK
&& eff) {
char *dup = av_strdup(eff);
if (dup) {
av_free(c->location);
c->location = dup;
}
}
commit_icy_headers(c);
}
/* A compressed body is addressed in encoded form, so byte offsets are
* meaningless: not seekable. Note that we prefer compression over
* seekability, servers don't offer media in compressed form, so it
* gives us free compression for other payloads like text playlist. */
c->seekable = !c->hdr_compressed &&
(status == 206 || c->hdr_accept_ranges);
/* Apply the user override on every reply so re-evaluation of a
* follow-up reply doesn't clobber it. */
if (c->seekable_opt >= 0)
c->seekable = c->seekable_opt;
c->icy_metaint = 0;
if (c->hdr_icy_metaint > 0) {
c->icy_metaint = c->hdr_icy_metaint;
if (c->seekable_opt > 0)
av_log(c->h, AV_LOG_WARNING, "Ignoring seekable=1, the "
"reply carries in-band ICY metadata\n");
/* In-band metadata makes byte offsets meaningless. Staying
* non-seekable also keeps on_done() from restarting a request
* without draining the FIFO, which would desync the
* interleave. */
c->seekable = 0;
c->content_size = -1;
}
} else if (status == 416 && c->probed && c->content_size < 0 &&
(c->hdr_content_total < 0 ||
c->hdr_content_total <= c->request_start)) {
/* A 416 on a follow-up request of a file of unknown size is EOF, unless
* it states a size past the requested offset. */
c->content_size = c->hdr_content_total;
c->stream_ok = 0;
update_status_locked(c, AVERROR_EOF, CURLE_OK);
} else {
c->loop->num_errors++;
c->stream_ok = 0;
update_status_locked(c, ff_http_averror(status, AVERROR(EIO)), CURLE_OK);
}
if (!c->probed) {
c->probe_ok = c->stream_ok;
c->probed = 1;
}
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return len;
}
static int xferinfo_callback(void *userdata, curl_off_t dltotal, curl_off_t dlnow,
curl_off_t ultotal, curl_off_t ulnow)
{
CurlContext *c = userdata;
int aborted;
pthread_mutex_lock(&c->mutex);
aborted = c->aborted;
pthread_mutex_unlock(&c->mutex);
return aborted; /* non-zero aborts the transfer */
}
/* (Re)issue the request for the current offset and add it to the multi. Loop
* thread only. */
static void start_request(CurlContext *c)
{
char range[48];
const char *range_value = NULL;
CURLcode cc;
int ret;
/* At or past the end of the file, report EOF without a request. */
if (c->probed && c->seekable &&
((c->content_size >= 0 && c->request_start >= c->content_size) ||
(c->end_off > 0 && c->request_start >= c->end_off))) {
pthread_mutex_lock(&c->mutex);
update_status_locked(c, AVERROR_EOF, CURLE_OK);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return;
}
if (!c->probed || c->seekable) {
int64_t start = c->request_start;
int64_t request_size = c->request_size;
if (c->is_initial && c->initial_request_size > 0)
request_size = c->initial_request_size;
if (request_size > 0 || c->end_off > 0) {
int64_t end = INT64_MAX;
if (request_size > 0 && start <= INT64_MAX - request_size)
end = start + request_size - 1;
if (c->content_size > 0)
end = FFMIN(end, c->content_size - 1);
if (c->end_off > 0)
end = FFMIN(end, c->end_off - 1);
snprintf(range, sizeof(range), "%"PRId64"-%"PRId64, start, end);
} else {
snprintf(range, sizeof(range), "%"PRId64"-", start);
}
range_value = range;
}
cc = curl_easy_setopt(c->easy, CURLOPT_RANGE, range_value);
if ((ret = curl_setopt_checked(c, "CURLOPT_RANGE", cc)) < 0) {
c->loop->num_errors++;
pthread_mutex_lock(&c->mutex);
update_status_locked(c, ret, cc);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return;
}
c->loop->num_requests++;
c->request_received = 0;
c->request_end = -1;
c->active = 1;
CURLMcode res = curl_multi_add_handle(c->loop->multi, c->easy);
if (res != CURLM_OK) {
av_log(c->h, AV_LOG_ERROR, "curl_multi_add_handle: %s\n",
curl_multi_strerror(res));
c->active = 0;
pthread_mutex_lock(&c->mutex);
update_status_locked(c, curlmcode_to_averror(res), curlmcode_to_curlcode(res));
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
}
}
static void update_statistics(CurlContext *c)
{
CurlLoop *loop = c->loop;
CURL *e = c->easy;
curl_off_t recv = 0, time = 0;
curl_easy_getinfo(e, CURLINFO_SIZE_DOWNLOAD_T, &recv);
curl_easy_getinfo(e, CURLINFO_TOTAL_TIME_T, &time);
if (recv) {
av_log(c->h, AV_LOG_DEBUG, "%"PRId64" bytes received in %"PRId64" ms\n",
(int64_t) recv, (int64_t) time / 1000);
loop->total_bytes += recv;
loop->total_time_us += time;
}
long num_conns = 0, num_redirs = 0;
curl_easy_getinfo(e, CURLINFO_NUM_CONNECTS, &num_conns);
curl_easy_getinfo(e, CURLINFO_REDIRECT_COUNT, &num_redirs);
loop->num_connections += (int) num_conns;
loop->num_redirects += (int) num_redirs;
}
/* Transfer finished (or failed) */
static void on_done(CurlContext *c, CURLcode code)
{
int64_t received;
int aborted;
pthread_mutex_lock(&c->mutex);
aborted = c->aborted;
received = c->request_received;
/* Advance past delivered bytes so a retry or seek resumes at the right offset. */
if (received > INT64_MAX - c->request_start) {
received = 0;
aborted = 1;
update_status_locked(c, AVERROR(EIO), code);
pthread_cond_broadcast(&c->cond);
}
c->request_start += received;
c->request_received = 0;
pthread_mutex_unlock(&c->mutex);
update_statistics(c);
if (!c->probed) {
/* Connection died before any usable header arrived. */
pthread_mutex_lock(&c->mutex);
c->probed = 1;
c->stream_ok = 0;
c->loop->num_errors++;
update_status_locked(c, curlcode_to_averror(code), code);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return;
}
if (aborted)
return;
if (c->seek_queued) {
/* previous soft seek drain finished; can start new request now */
c->seek_queued = 0;
start_request(c);
return;
}
if (code == CURLE_OK && c->stream_ok) {
int64_t file_end = c->content_size >= 0 ? c->content_size - 1 : INT64_MAX;
if (c->end_off > 0)
file_end = FFMIN(file_end, c->end_off - 1);
if (c->seekable && c->request_end >= 0 && c->request_end < file_end) {
if (!received) {
av_log(c->h, AV_LOG_ERROR, "Server sent back no data at offset "
"%"PRId64"\n", c->request_start);
c->loop->num_errors++;
pthread_mutex_lock(&c->mutex);
update_status_locked(c, AVERROR(EIO), CURLE_OK);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return;
}
c->is_initial = 0;
start_request(c);
return;
}
pthread_mutex_lock(&c->mutex);
c->status = AVERROR_EOF;
c->curl_status = CURLE_OK;
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
return;
}
if (c->stream_ok) {
av_log(c->h, AV_LOG_WARNING, "%s\n", curl_easy_strerror(code));
c->loop->num_errors++;
}
/* Unhandled generic curl error */
pthread_mutex_lock(&c->mutex);
update_status_locked(c, curlcode_to_averror(code), code);
pthread_cond_broadcast(&c->cond);
pthread_mutex_unlock(&c->mutex);
}
/* ------------------------------------------------------------------------- */
/* event loop thread + command queue */
/* ------------------------------------------------------------------------- */
static int test_short_seek(CurlContext *c)
{
if (c->seek_queued)
return 1; /* short seek already queued */
if (c->short_seek_size <= 0 || /* short seek disabled */
c->request_end < 0) /* content size not known */
return 0;
const int64_t last = c->request_end - c->request_start;
return last - c->request_received < c->short_seek_size;
}
static void execute_command(CurlLoop *loop, CurlCmd *cmd)
{
CurlContext *c = cmd->ctx;
switch (cmd->kind) {
case CMD_ADD:
start_request(c);
break;
case CMD_REMOVE:
if (c->active) {
curl_multi_remove_handle(loop->multi, c->easy);
update_statistics(c);
c->active = 0;
}
break;
case CMD_UNPAUSE:
pthread_mutex_lock(&c->mutex);
c->paused = 0;
pthread_mutex_unlock(&c->mutex);
curl_easy_pause(c->easy, CURLPAUSE_CONT);
break;
case CMD_SEEK:
if (c->active && test_short_seek(c)) {
c->seek_queued = 1;
} else if (c->active) {
curl_multi_remove_handle(loop->multi, c->easy);
c->active = 0;
}
pthread_mutex_lock(&c->mutex);
av_fifo_reset2(c->fifo);
const int was_paused = c->paused;
c->aborted = 0;
c->paused = 0;
c->status = 0;
c->curl_status = 0;
pthread_mutex_unlock(&c->mutex);
c->request_start = cmd->pos;
c->request_received = 0;
if (!c->seek_queued)
start_request(c);
else if (was_paused)
curl_easy_pause(c->easy, CURLPAUSE_CONT);
break;
}
}
static void *curl_worker(void *arg)
{
CurlLoop *loop = arg;
ff_thread_setname("curl");
while (1) {
CurlCmd *cmd;
CURLMsg *msg;
int running = 0, left = 0, do_exit;
pthread_mutex_lock(&loop->mutex);
cmd = loop->cmd_head;
if (cmd) {
loop->cmd_head = cmd->next;
if (!loop->cmd_head)
loop->cmd_tail = NULL;
}
do_exit = loop->exit;
pthread_mutex_unlock(&loop->mutex);
if (cmd) {
execute_command(loop, cmd);
if (cmd->sync) {
pthread_mutex_lock(&loop->mutex);
cmd->done = 1;
pthread_cond_broadcast(&loop->cond);
pthread_mutex_unlock(&loop->mutex);
} else {
av_free(cmd);
}
continue; /* drain the whole queue before pumping curl */
}
if (do_exit)
break;
curl_multi_perform(loop->multi, &running);
while ((msg = curl_multi_info_read(loop->multi, &left))) {
CurlContext *c = NULL;
if (msg->msg != CURLMSG_DONE)
continue;
curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &c);
curl_multi_remove_handle(loop->multi, msg->easy_handle);
if (c) {
c->active = 0;
on_done(c, msg->data.result);
}
}
curl_multi_poll(loop->multi, NULL, 0, 1000, NULL);
}
return NULL;
}
/* Dispatch a command to the loop. For sync commands the caller blocks until the
* loop thread has executed it. Returns 0 or a negative AVERROR. */
static int curl_dispatch(CurlLoop *loop, enum cmd_kind kind, CurlContext *c,
int64_t pos, int sync)
{
CurlCmd stackcmd = {0};
CurlCmd *cmd = sync ? &stackcmd : av_mallocz(sizeof(*cmd));
if (!cmd)
return AVERROR(ENOMEM);
cmd->kind = kind;
cmd->ctx = c;
cmd->pos = pos;
cmd->sync = sync;
pthread_mutex_lock(&loop->mutex);
if (loop->cmd_tail)
loop->cmd_tail->next = cmd;
else
loop->cmd_head = cmd;
loop->cmd_tail = cmd;
curl_multi_wakeup(loop->multi);
if (sync) {
while (!cmd->done)
pthread_cond_wait(&loop->cond, &loop->mutex);
}
pthread_mutex_unlock(&loop->mutex);
return 0;
}
static void share_lock_callback(CURL *handle, curl_lock_data data,
curl_lock_access access, void *userdata)
{
CurlLoop *loop = userdata;
pthread_mutex_lock(&loop->mutex);
}
static void share_unlock_callback(CURL *handle, curl_lock_data data,
void *userdata)
{
CurlLoop *loop = userdata;
pthread_mutex_unlock(&loop->mutex);
}
static CurlLoop *curl_loop_create(AVFormatContext *avfc)
{
CurlLoop *loop = av_mallocz(sizeof(*loop));
CURLSHcode res;
if (!loop)
return NULL;
loop->avfc = avfc;
if (pthread_mutex_init(&loop->mutex, NULL))
goto fail;
if (pthread_cond_init(&loop->cond, NULL)) {
pthread_mutex_destroy(&loop->mutex);
goto fail;
}
if (curl_global_init(CURL_GLOBAL_DEFAULT) != CURLE_OK)
goto fail2;
loop->multi = curl_multi_init();
if (!loop->multi)
goto fail3;
if (curl_multi_setopt(loop->multi, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX) != CURLM_OK)
goto fail3;
loop->share = curl_share_init();
if (!loop->share)
goto fail3;
if (curl_share_setopt(loop->share, CURLSHOPT_USERDATA, loop) != CURLSHE_OK ||
curl_share_setopt(loop->share, CURLSHOPT_LOCKFUNC, share_lock_callback) != CURLSHE_OK ||
curl_share_setopt(loop->share, CURLSHOPT_UNLOCKFUNC, share_unlock_callback) != CURLSHE_OK)
goto fail3;
res = curl_share_setopt(loop->share, CURLSHOPT_SHARE, CURL_LOCK_DATA_COOKIE);
if (res != CURLSHE_OK && res != CURLSHE_NOT_BUILT_IN)
goto fail3;
loop->has_cookies = res == CURLSHE_OK;
res = curl_share_setopt(loop->share, CURLSHOPT_SHARE, CURL_LOCK_DATA_HSTS);
if (res != CURLSHE_OK && res != CURLSHE_NOT_BUILT_IN)
goto fail3;
if (pthread_create(&loop->thread, NULL, curl_worker, loop))
goto fail3;
return loop;
fail3:
curl_multi_cleanup(loop->multi);
curl_share_cleanup(loop->share);
curl_global_cleanup();
fail2:
pthread_cond_destroy(&loop->cond);
pthread_mutex_destroy(&loop->mutex);
fail:
av_free(loop);
return NULL;
}
static void print_statistics(CurlLoop *loop)
{
AVFormatContext *avfc = loop->avfc;
if (loop->total_bytes) {
double time = loop->total_time_us / 1000000.0;
double avg = time ? loop->total_bytes / time : 0;
av_log(avfc, AV_LOG_VERBOSE,
"libcurl: Overall %"PRId64" bytes received in %.0f ms = %.0f kB/s\n",
loop->total_bytes, time * 1e3, avg / 1e3);
}
if (loop->num_connections || loop->num_errors) {
av_log(avfc, AV_LOG_VERBOSE,
"libcurl: %d connections, %d redirects, %d requests, %d errors\n",
loop->num_connections, loop->num_redirects, loop->num_requests,
loop->num_errors);
}
}
static void curl_loop_destroy(CurlLoop *loop)
{
pthread_mutex_lock(&loop->mutex);
loop->exit = 1;
curl_multi_wakeup(loop->multi);
pthread_mutex_unlock(&loop->mutex);
pthread_join(loop->thread, NULL);
print_statistics(loop);
curl_multi_cleanup(loop->multi);
curl_share_cleanup(loop->share);
pthread_cond_destroy(&loop->cond);
pthread_mutex_destroy(&loop->mutex);
av_free(loop);
/* Released after the thread is joined and the multi handle is gone. */
curl_global_cleanup();
}
/* Attach a context to its event loop. With an owning AVFormatContext the loop is
* created lazily, cached on it, and shared across the demuxer's transfers so curl
* reuses connections; it is freed at format teardown. Without one the context
* gets a private loop freed on close. */
static int curl_loop_attach(CurlContext *c, AVFormatContext *avfc)
{
if (!avfc) {
c->loop = curl_loop_create(NULL);
c->private_loop = 1;
return c->loop ? 0 : AVERROR(ENOMEM);
}
pthread_mutex_lock(&curl_loop_lock);
c->loop = ffformatcontext(avfc)->curl_loop;
if (!c->loop) {
c->loop = curl_loop_create(avfc);
ffformatcontext(avfc)->curl_loop = c->loop;
}
pthread_mutex_unlock(&curl_loop_lock);
return c->loop ? 0 : AVERROR(ENOMEM);
}
void ff_curl_loop_free(struct CurlLoop **loop)
{
if (loop && *loop) {
curl_loop_destroy(*loop);
*loop = NULL;
}
}
/* ------------------------------------------------------------------------- */
/* URLProtocol callbacks */
/* ------------------------------------------------------------------------- */
static int libcurl_close(URLContext *h);
static int debug_callback(CURL *easy, curl_infotype type, char *data,
size_t size, void *userdata)
{
CurlContext *c = userdata;
const char *prefix, *p = data, *end = data + size;
switch (type) {
case CURLINFO_TEXT: prefix = "* "; break;
case CURLINFO_HEADER_IN: prefix = "< "; break;
case CURLINFO_HEADER_OUT: prefix = "> "; break;
default: return 0;
}
/* Split multiline payload into each log. */
while (p < end) {
const char *nl = memchr(p, '\n', end - p);
size_t len = (nl ? nl : end) - p;
while (len && p[len - 1] == '\r')
len--;
av_log(c->h, AV_LOG_DEBUG, "%s%.*s\n", prefix, (int)len, p);
if (!nl)
break;
p = nl + 1;
}
return 0;
}
static int slist_append(struct curl_slist **list, const char *str)
{
struct curl_slist *tmp = curl_slist_append(*list, str);
if (!tmp)
return AVERROR(ENOMEM);
*list = tmp;
return 0;
}
/* Build the custom request header list from the referer and headers options. */
static int build_headers(CurlContext *c)
{
int user_set_icy = 0, ret = 0;
if (c->referer && c->referer[0]) {
char *h = av_asprintf("Referer: %s", c->referer);
if (!h)
return AVERROR(ENOMEM);
ret = slist_append(&c->header_list, h);
av_free(h);
if (ret < 0)
return ret;
}
if (c->headers && c->headers[0]) {
char *copy = av_strdup(c->headers);
char *line, *saveptr = NULL;
if (!copy)
return AVERROR(ENOMEM);
for (line = av_strtok(copy, "\r\n", &saveptr); line && ret >= 0;
line = av_strtok(NULL, "\r\n", &saveptr)) {
if (!av_strncasecmp(line, "Icy-MetaData:", 13))
user_set_icy = 1;
ret = slist_append(&c->header_list, line);
}
av_free(copy);
if (ret < 0)
return ret;
}
/* libcurl does not deduplicate the list, so only add ours if the user
* did not already ask for one. */
if (c->icy && !user_set_icy)
return slist_append(&c->header_list, "Icy-MetaData: 1");
return 0;
}
static int setup_protocols(CurlContext *c)
{
const char *wl = c->h->protocol_whitelist;
const char *bl = c->h->protocol_blacklist;
int ret;
if (!wl && !bl)
return 0;
AVBPrint bp;
av_bprint_init(&bp, 0, AV_BPRINT_SIZE_AUTOMATIC);
curl_version_info_data *info = curl_version_info(CURLVERSION_NOW);
for (const char *const *p = info->protocols; *p; p++) {
const char *proto = *p;
if (av_strcasecmp(proto, "http") && av_strcasecmp(proto, "https"))
continue; /* only http(s) are supported by libcurl.c at the moment */
if (wl && av_match_list(proto, wl, ',') <= 0)
continue;
if (bl && av_match_list(proto, bl, ',') > 0)
continue;
if (bp.len)
av_bprint_chars(&bp, ',', 1);
av_bprintf(&bp, "%s", proto);
}
if (!av_bprint_is_complete(&bp)) {
av_bprint_finalize(&bp, NULL);
return AVERROR(ENOMEM);
}
if (!bp.len) {
av_log(c->h, AV_LOG_ERROR, "Set of allowed protocols is empty.\n");
av_bprint_finalize(&bp, NULL);
return AVERROR(EINVAL);
}
ret = curl_setopt_checked(c, "CURLOPT_PROTOCOLS_STR",
curl_easy_setopt(c->easy, CURLOPT_PROTOCOLS_STR, bp.str));
if (!ret)
ret = curl_setopt_checked(c, "CURLOPT_REDIR_PROTOCOLS_STR",
curl_easy_setopt(c->easy, CURLOPT_REDIR_PROTOCOLS_STR, bp.str));
av_bprint_finalize(&bp, NULL);
return ret;
}
#define CURL_SETOPT(opt, val) \
do { \
int ret_ = curl_setopt_checked(c, #opt, \
curl_easy_setopt(e, opt, val)); \
if (ret_ < 0) \
return ret_; \
} while (0)
static int setup_curl(CurlContext *c)
{
const curl_version_info_data *info = curl_version_info(CURLVERSION_NOW);
CURL *e = c->easy;
CURLcode cc;
int ret;
const char *url = c->h->filename;
/* Drop an optional "libcurl:" prefix that forces this protocol. */
av_strstart(url, "libcurl:", &url);
CURL_SETOPT(CURLOPT_URL, url);
CURL_SETOPT(CURLOPT_PRIVATE, c);
CURL_SETOPT(CURLOPT_NOSIGNAL, 1L);
CURL_SETOPT(CURLOPT_SHARE, c->loop->share);
CURL_SETOPT(CURLOPT_WRITEFUNCTION, write_callback);
CURL_SETOPT(CURLOPT_WRITEDATA, c);
CURL_SETOPT(CURLOPT_HEADERFUNCTION, header_callback);
CURL_SETOPT(CURLOPT_HEADERDATA, c);
CURL_SETOPT(CURLOPT_NOPROGRESS, 0L);
CURL_SETOPT(CURLOPT_XFERINFOFUNCTION, xferinfo_callback);
CURL_SETOPT(CURLOPT_XFERINFODATA, c);
if (av_log_get_level() >= AV_LOG_DEBUG) {
CURL_SETOPT(CURLOPT_VERBOSE, 1L);
CURL_SETOPT(CURLOPT_DEBUGFUNCTION, debug_callback);
CURL_SETOPT(CURLOPT_DEBUGDATA, c);
}
CURL_SETOPT(CURLOPT_FOLLOWLOCATION, 1L);
CURL_SETOPT(CURLOPT_MAXREDIRS, (long)c->max_redirects);
CURL_SETOPT(CURLOPT_HTTP_VERSION, (long)c->http_version);
CURL_SETOPT(CURLOPT_TCP_KEEPALIVE, c->multiple_requests ? 1L : 0L);
CURL_SETOPT(CURLOPT_FORBID_REUSE, c->multiple_requests ? 0L : 1L);
if (info->features & CURL_VERSION_HSTS)
CURL_SETOPT(CURLOPT_HSTS_CTRL, (long)CURLHSTS_ENABLE);
CURL_SETOPT(CURLOPT_ACCEPT_ENCODING,
c->off > 0 || c->end_off > 0 ? "identity" : "");
if (c->connect_timeout > 0)
CURL_SETOPT(CURLOPT_CONNECTTIMEOUT_MS,
(long)c->connect_timeout * 1000);
if (c->user_agent && c->user_agent[0])
CURL_SETOPT(CURLOPT_USERAGENT, c->user_agent);
if (c->http_proxy && c->http_proxy[0])
CURL_SETOPT(CURLOPT_PROXY, c->http_proxy);
if (info->features & CURL_VERSION_SSL)
CURL_SETOPT(CURLOPT_SSL_OPTIONS, (long)CURLSSLOPT_NATIVE_CA);
CURL_SETOPT(CURLOPT_SSL_VERIFYPEER, c->tls_verify ? 1L : 0L);
CURL_SETOPT(CURLOPT_SSL_VERIFYHOST, c->tls_verify ? 2L : 0L);
if (c->ca_file)
CURL_SETOPT(CURLOPT_CAINFO, c->ca_file);
if (c->cert_file)
CURL_SETOPT(CURLOPT_SSLCERT, c->cert_file);
if (c->key_file)
CURL_SETOPT(CURLOPT_SSLKEY, c->key_file);
if (c->loop->has_cookies) {
CURL_SETOPT(CURLOPT_COOKIEFILE, "");
} else if (c->cookies && c->cookies[0]) {
av_log(c->h, AV_LOG_ERROR, "libcurl was built without cookie support\n");
return AVERROR(ENOSYS);
}
if (c->cookies && c->cookies[0]) {
char *copy = av_strdup(c->cookies);
char *line, *saveptr = NULL;
if (!copy)
return AVERROR(ENOMEM);
for (line = av_strtok(copy, "\r\n", &saveptr); line;
line = av_strtok(NULL, "\r\n", &saveptr)) {
char *sc = av_asprintf("Set-Cookie: %s", line);
if (!sc) {
av_free(copy);
return AVERROR(ENOMEM);
}
cc = curl_easy_setopt(e, CURLOPT_COOKIELIST, sc);
av_free(sc);
if ((ret = curl_setopt_checked(c, "CURLOPT_COOKIELIST", cc)) < 0) {
av_free(copy);
return ret;
}
}
av_free(copy);
}
if ((ret = build_headers(c)) < 0)
return ret;
if (c->header_list)
CURL_SETOPT(CURLOPT_HTTPHEADER, c->header_list);
/* Shoutcast v1 answers "ICY 200 OK", which curl would otherwise reject as
* HTTP/0.9 before any header reaches header_callback(). */
if (c->icy_status) {
if ((ret = slist_append(&c->alias_list, "ICY 200")) < 0)
return ret;
CURL_SETOPT(CURLOPT_HTTP200ALIASES, c->alias_list);
}
return 0;
}
#undef CURL_SETOPT
static void curl_cond_wait(CurlContext *c)
{
int64_t t = av_gettime() + CURL_WAIT_US;
struct timespec ts = { .tv_sec = t / 1000000,
.tv_nsec = (t % 1000000) * 1000 };
pthread_cond_timedwait(&c->cond, &c->mutex, &ts);
}
/* Block until the transfer has been probed, the stream errored, or the open was
* interrupted. Returns 0, or a negative AVERROR. */
static int wait_for_probe(CurlContext *c)
{
URLContext *h = c->h;
int ret = 0;
pthread_mutex_lock(&c->mutex);
while (!c->probed && !c->status) {
if (ff_check_interrupt(&h->interrupt_callback)) {
c->aborted = 1;
ret = AVERROR_EXIT;
break;
}
curl_cond_wait(c);
}
if (!ret) {
if (!c->probe_ok)
ret = c->status ? c->status : AVERROR(EIO);
}
pthread_mutex_unlock(&c->mutex);
return ret;
}
/* Scales by the recurrence relationship x := 2x + 1, i.e. 2^n - 1 */
static int64_t retry_delay(CurlContext *c)
{
int64_t factor = INT64_MAX >> FFMAX(63 - c->retry_count, 0);
if (factor >= CURL_RETRY_MAX_US / CURL_RETRY_BASE_US)
return CURL_RETRY_MAX_US;
return factor * CURL_RETRY_BASE_US;
}
static int retry_request_locked(URLContext *h, int nonblock)
{
CurlContext *c = h->priv_data;
const int status = c->status;
const CURLcode code = c->curl_status;
pthread_mutex_unlock(&c->mutex);
if (!c->retry_time) {
if (c->retry_count >= c->max_retries) {
av_log(h, AV_LOG_ERROR, "Maximum number of retries (%d) reached\n",
c->max_retries);
return status;
}
const int64_t delay = retry_delay(c);
c->retry_time = av_gettime_relative() + delay;
c->retry_count++;
av_log(h, AV_LOG_WARNING, "Retrying (#%d) from %"PRId64" in %.3fs "
"after: %s (%s)\n", c->retry_count, c->logical_pos,
delay * 1e-6, av_err2str(status),
curl_easy_strerror(code));
}
const int64_t sleep_us = c->retry_time - av_gettime_relative();
if (sleep_us > 0 && !nonblock) {
av_usleep(FFMIN(sleep_us, CURL_WAIT_US));
} else if (sleep_us <= 0) {
/* Use a synchronous request to ensure that the seek is registered, and
* the reset of c->state is observable, before the next libcurl_read()
* call, otherwise this might hit the exact same retry path a second
* time. */
int ret = curl_dispatch(c->loop, CMD_SEEK, c, c->logical_pos, 1);
if (ret < 0)
return ret;
c->retry_time = 0;
}
return AVERROR(EAGAIN); /* allow caller to handle interrupts and retry */
}
static int libcurl_open(URLContext *h, const char *url, int flags,
AVDictionary **options)
{
/* Guard against non-thread-safe libcurl builds. This should never happen,
* since libcurl is used only on platforms with thread support, and thread
* safety is enabled unconditionally in libcurl when the platform supports
* threads or atomics. */
curl_version_info_data *info = curl_version_info(CURLVERSION_NOW);
if (!(info->features & CURL_VERSION_THREADSAFE))
return AVERROR(ENOSYS);
CurlContext *c = h->priv_data;
const char *eff_url = h->filename;
int ret;
c->h = h;
c->content_size = -1;
c->request_start = c->off;
c->request_end = -1;
c->logical_pos = c->off;
c->is_initial = 1;
c->icy_block_len = -1;
/* Report the request URL until header_callback replaces it post-redirect. */
av_strstart(eff_url, "libcurl:", &eff_url);
av_freep(&c->location);
c->location = av_strdup(eff_url);
ret = pthread_mutex_init(&c->mutex, NULL);
if (ret)
return AVERROR(ret);
ret = pthread_cond_init(&c->cond, NULL);
if (ret) {
pthread_mutex_destroy(&c->mutex);
return AVERROR(ret);
}
c->fifo = av_fifo_alloc2(c->buffer_size, 1, 0);
if (!c->fifo) {
ret = AVERROR(ENOMEM);
goto fail;
}
ret = curl_loop_attach(c, h->avfc);
if (ret < 0)
goto fail;
c->easy = curl_easy_init();
if (!c->easy) {
ret = AVERROR(ENOMEM);
goto fail;
}
ret = setup_protocols(c);
if (ret < 0)
goto fail;
ret = setup_curl(c);
if (ret < 0)
goto fail;
ret = curl_dispatch(c->loop, CMD_ADD, c, 0, 0);
if (ret < 0)
goto fail;
ret = wait_for_probe(c);
if (ret < 0)
goto fail;
pthread_mutex_lock(&c->mutex);
h->is_streamed = !c->seekable;
pthread_mutex_unlock(&c->mutex);
return 0;
fail:
libcurl_close(h);
return ret;
}
/* Export the metadata block, which the packet parser splits in place. */
static int update_icy_metadata(CurlContext *c)
{
av_freep(&c->icy_metadata_packet);
c->icy_metadata_packet = av_strdup((char *)c->icy_block);
if (!c->icy_metadata_packet)
return AVERROR(ENOMEM);
ff_http_parse_icy_packet(c->h, &c->metadata, (char *)c->icy_block);
return 0;
}
/* Consume the metadata block at the current interleave boundary, with the
* mutex held. Returns 1 once a block is complete and ready to export. */
static int drain_icy_block(CurlContext *c)
{
size_t avail;
while ((avail = av_fifo_can_read(c->fifo))) {
int n;
if (c->icy_block_len < 0) {
uint8_t units;
av_fifo_read(c->fifo, &units, 1);
c->icy_block_len = units * 16;
c->icy_block_filled = 0;
/* A zero length byte means the metadata did not change. */
if (!c->icy_block_len) {
c->icy_block_len = -1;
c->icy_data_read = 0;
return 0;
}
continue;
}
n = FFMIN(avail, (size_t)(c->icy_block_len - c->icy_block_filled));
av_fifo_read(c->fifo, c->icy_block + c->icy_block_filled, n);
c->icy_block_filled += n;
if (c->icy_block_filled < c->icy_block_len)
return 0;
c->icy_block[c->icy_block_len] = 0;
c->icy_block_len = -1;
c->icy_data_read = 0;
return 1;
}
return 0;
}
static int libcurl_read(URLContext *h, unsigned char *buf, int size)
{
CurlContext *c = h->priv_data;
int nonblock = h->flags & AVIO_FLAG_NONBLOCK;
int unpause, ret;
pthread_mutex_lock(&c->mutex);
while (1) {
size_t avail;
/* A failed export loses one metadata update but leaves the interleave
* in sync, since the block is fully consumed either way. */
if (c->icy_metaint > 0 && c->icy_data_read == c->icy_metaint &&
drain_icy_block(c) && (ret = update_icy_metadata(c)) < 0)
break;
avail = av_fifo_can_read(c->fifo);
if (c->icy_metaint > 0)
avail = FFMIN(avail, (size_t)(c->icy_metaint - c->icy_data_read));
if (avail) {
ret = FFMIN(avail, (size_t)size);
av_fifo_read(c->fifo, buf, ret);
c->icy_data_read += ret;
c->logical_pos += ret;
c->retry_count = 0;
c->retry_time = 0;
break;
}
if (c->status) {
if (c->seekable && is_recoverable(c->curl_status))
return retry_request_locked(h, nonblock);
if (c->status == AVERROR_EOF && c->icy_block_len >= 0)
av_log(h, AV_LOG_WARNING,
"Stream ended inside an ICY metadata block\n");
ret = c->status;
break;
}
if (nonblock) {
ret = AVERROR(EAGAIN);
break;
}
curl_cond_wait(c);
/* Return to the avio layer so it can poll the interrupt callback. */
nonblock = 1;
}
/* Resume a paused transfer once the FIFO is at least half empty, on every
* exit path because ICY framing can be drained without returning media. */
unpause = c->paused && av_fifo_can_write(c->fifo) * 2 >= c->buffer_size;
pthread_mutex_unlock(&c->mutex);
if (unpause)
curl_dispatch(c->loop, CMD_UNPAUSE, c, 0, 0);
return ret;
}
static int64_t libcurl_seek(URLContext *h, int64_t pos, int whence)
{
CurlContext *c = h->priv_data;
int64_t newpos;
pthread_mutex_lock(&c->mutex);
const int64_t content_size = c->content_size;
const int seekable = c->seekable;
pthread_mutex_unlock(&c->mutex);
if (whence == AVSEEK_SIZE)
return content_size >= 0 ? content_size : AVERROR(ENOSYS);
if (!seekable)
return AVERROR(ENOSYS);
switch (whence) {
case SEEK_SET:
newpos = pos;
break;
case SEEK_CUR:
if (pos > INT64_MAX - c->logical_pos)
return AVERROR(ERANGE);
newpos = c->logical_pos + pos;
break;
case SEEK_END:
if (content_size < 0)
return AVERROR(ENOSYS);
if (pos > INT64_MAX - content_size)
return AVERROR(ERANGE);
newpos = content_size + pos;
break;
default:
return AVERROR(EINVAL);
}
if (newpos < 0)
return AVERROR(EINVAL);
if (newpos == c->logical_pos)
return newpos;
/* Restart the transfer at the new offset. Any failure of the new request
* surfaces on the following url_read(). */
curl_dispatch(c->loop, CMD_SEEK, c, newpos, 1);
c->logical_pos = newpos;
c->retry_count = 0;
c->retry_time = 0;
return newpos;
}
static int libcurl_close(URLContext *h)
{
CurlContext *c = h->priv_data;
if (c->loop) {
if (c->easy) {
/* Ensure the handle is out of the multi before we free it. */
curl_dispatch(c->loop, CMD_REMOVE, c, 0, 1);
curl_easy_cleanup(c->easy);
c->easy = NULL;
}
/* A shared loop outlives the transfer for connection reuse. */
if (c->private_loop)
curl_loop_destroy(c->loop);
c->loop = NULL;
}
if (c->header_list)
curl_slist_free_all(c->header_list);
if (c->alias_list)
curl_slist_free_all(c->alias_list);
av_dict_free(&c->hdr_icy);
av_fifo_freep2(&c->fifo);
pthread_cond_destroy(&c->cond);
pthread_mutex_destroy(&c->mutex);
return 0;
}
static int libcurl_get_short_seek(URLContext *h)
{
CurlContext *c = h->priv_data;
if (c->short_seek_size >= 1)
return FFMIN(c->short_seek_size, INT_MAX);
return AVERROR(ENOSYS);
}
#define OFFSET(x) offsetof(CurlContext, x)
#define D AV_OPT_FLAG_DECODING_PARAM
#define E AV_OPT_FLAG_ENCODING_PARAM
static const AVOption options[] = {
{ "user_agent", "override User-Agent header", OFFSET(user_agent), AV_OPT_TYPE_STRING, { .str = DEFAULT_USER_AGENT }, 0, 0, D },
{ "referer", "override Referer header", OFFSET(referer), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D },
{ "headers", "set custom HTTP headers, can override built in default headers", OFFSET(headers), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D | E },
{ "http_proxy", "set HTTP proxy to tunnel through", OFFSET(http_proxy), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D | E },
{ "cookies", "set cookies to be sent in applicable future requests, use newline delimited Set-Cookie HTTP field value syntax", OFFSET(cookies), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D },
{ "location", "the actual location of the data received", OFFSET(location), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D | E },
{ "offset", "initial byte offset", OFFSET(off), AV_OPT_TYPE_INT64, { .i64 = 0 }, 0, INT64_MAX, D },
{ "end_offset", "try to limit the request to bytes preceding this offset", OFFSET(end_off), AV_OPT_TYPE_INT64, { .i64 = 0 }, 0, INT64_MAX, D },
{ "seekable", "control seekability of connection", OFFSET(seekable_opt), AV_OPT_TYPE_BOOL, { .i64 = -1 }, -1, 1, D },
{ "tls_verify", "verify the peer certificate and hostname", OFFSET(tls_verify), AV_OPT_TYPE_BOOL, { .i64 = 1 }, 0, 1, D | E },
{ "ca_file", "certificate authority bundle file", OFFSET(ca_file), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D | E },
{ "cert_file", "client certificate file", OFFSET(cert_file), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D | E },
{ "key_file", "client private key file", OFFSET(key_file), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, D | E },
{ "connect_timeout", "connection timeout in seconds (0 = libcurl default)", OFFSET(connect_timeout), AV_OPT_TYPE_INT, { .i64 = 0 }, 0, INT_MAX / 1000, D | E },
{ "max_redirects", "maximum number of redirects to follow", OFFSET(max_redirects), AV_OPT_TYPE_INT, { .i64 = 16 }, 0, INT_MAX, D },
{ "multiple_requests", "reuse the connection across requests (HTTP keep-alive)", OFFSET(multiple_requests), AV_OPT_TYPE_BOOL, { .i64 = 1 }, 0, 1, D | E },
{ "max_retries", "maximum number of retries after a recoverable error", OFFSET(max_retries), AV_OPT_TYPE_INT, { .i64 = 5 }, 0, INT_MAX, D },
{ "icy", "request ICY metadata", OFFSET(icy), AV_OPT_TYPE_BOOL, { .i64 = 1 }, 0, 1, D },
{ "icy_status", "accept legacy Shoutcast \"ICY 200 OK\" status lines", OFFSET(icy_status), AV_OPT_TYPE_BOOL, { .i64 = 1 }, 0, 1, D },
{ "icy_metadata_headers", "return ICY metadata headers", OFFSET(icy_metadata_headers), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, AV_OPT_FLAG_EXPORT },
{ "icy_metadata_packet", "return current ICY metadata packet", OFFSET(icy_metadata_packet), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, AV_OPT_FLAG_EXPORT },
{ "metadata", "metadata read from the bitstream", OFFSET(metadata), AV_OPT_TYPE_DICT, {0}, 0, 0, AV_OPT_FLAG_EXPORT },
{ "buffer_size", "receive buffer size in bytes", OFFSET(buffer_size), AV_OPT_TYPE_INT64, { .i64 = CURL_DEFAULT_BUFFER_SIZE }, CURL_MAX_WRITE_SIZE, INT_MAX, D },
{ "request_size", "split a transfer into ranged requests of at most this many bytes (0 = unlimited)", OFFSET(request_size), AV_OPT_TYPE_INT64, { .i64 = 0 }, 0, INT64_MAX, D },
{ "initial_request_size", "size (in bytes) of initial requests made during probing / header parsing", OFFSET(initial_request_size), AV_OPT_TYPE_INT64, { .i64 = 0 }, 0, INT64_MAX, D },
{ "http_version", "HTTP version to use", OFFSET(http_version), AV_OPT_TYPE_INT, { .i64 = CURL_HTTP_VERSION_NONE }, 0, INT_MAX, D, .unit = "http_version" },
{ "auto", "negotiate the best supported version", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_NONE }, 0, 0, D, .unit = "http_version" },
{ "1.0", "HTTP/1.0", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_1_0 }, 0, 0, D, .unit = "http_version" },
{ "1.1", "HTTP/1.1", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_1_1 }, 0, 0, D, .unit = "http_version" },
{ "2", "HTTP/2", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_2 }, 0, 0, D, .unit = "http_version" },
{ "2tls", "HTTP/2 over TLS only", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_2TLS }, 0, 0, D, .unit = "http_version" },
{ "2-prior-knowledge", "HTTP/2 without an upgrade handshake", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_2_PRIOR_KNOWLEDGE }, 0, 0, D, .unit = "http_version" },
{ "3", "HTTP/3, fall back to earlier versions", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_3 }, 0, 0, D, .unit = "http_version" },
{ "3only", "HTTP/3 only", 0, AV_OPT_TYPE_CONST, { .i64 = CURL_HTTP_VERSION_3ONLY }, 0, 0, D, .unit = "http_version" },
{ "short_seek_size", "threshold to favor readahead over seek", OFFSET(short_seek_size), AV_OPT_TYPE_INT64, { .i64 = 0 }, 0, INT64_MAX, D },
{ NULL }
};
static const AVClass libcurl_context_class = {
.class_name = "libcurl",
.item_name = av_default_item_name,
.option = options,
.version = LIBAVUTIL_VERSION_INT,
};
const URLProtocol ff_libcurl_protocol = {
.name = "libcurl",
.url_open2 = libcurl_open,
.url_read = libcurl_read,
.url_seek = libcurl_seek,
.url_close = libcurl_close,
.url_get_short_seek = libcurl_get_short_seek,
.priv_data_size = sizeof(CurlContext),
.priv_data_class = &libcurl_context_class,
.flags = URL_PROTOCOL_FLAG_NETWORK,
.default_whitelist = "http,https,libcurl",
};
View Git Blame Copy Permalink