mpv/filters/filter.c

791 lines
23 KiB
C
Raw Normal View History

video: rewrite filtering glue code Get rid of the old vf.c code. Replace it with a generic filtering framework, which can potentially handle more than just --vf. At least reimplementing --af with this code is planned. This changes some --vf semantics (including runtime behavior and the "vf" command). The most important ones are listed in interface-changes. vf_convert.c is renamed to f_swscale.c. It is now an internal filter that can not be inserted by the user manually. f_lavfi.c is a refactor of player/lavfi.c. The latter will be removed once --lavfi-complex is reimplemented on top of f_lavfi.c. (which is conceptually easy, but a big mess due to the data flow changes). The existing filters are all changed heavily. The data flow of the new filter framework is different. Especially EOF handling changes - EOF is now a "frame" rather than a state, and must be passed through exactly once. Another major thing is that all filters must support dynamic format changes. The filter reconfig() function goes away. (This sounds complex, but since all filters need to handle EOF draining anyway, they can use the same code, and it removes the mess with reconfig() having to predict the output format, which completely breaks with libavfilter anyway.) In addition, there is no automatic format negotiation or conversion. libavfilter's primitive and insufficient API simply doesn't allow us to do this in a reasonable way. Instead, filters can use f_autoconvert as sub-filter, and tell it which formats they support. This filter will in turn add actual conversion filters, such as f_swscale, to perform necessary format changes. vf_vapoursynth.c uses the same basic principle of operation as before, but with worryingly different details in data flow. Still appears to work. The hardware deint filters (vf_vavpp.c, vf_d3d11vpp.c, vf_vdpaupp.c) are heavily changed. Fortunately, they all used refqueue.c, which is for sharing the data flow logic (especially for managing future/past surfaces and such). It turns out it can be used to factor out most of the data flow. Some of these filters accepted software input. Instead of having ad-hoc upload code in each filter, surface upload is now delegated to f_autoconvert, which can use f_hwupload to perform this. Exporting VO capabilities is still a big mess (mp_stream_info stuff). The D3D11 code drops the redundant image formats, and all code uses the hw_subfmt (sw_format in FFmpeg) instead. Although that too seems to be a big mess for now. f_async_queue is unused.
2018-01-16 10:53:44 +00:00
#include <pthread.h>
#include "common/common.h"
#include "common/global.h"
#include "common/msg.h"
#include "video/hwdec.h"
#include "filter.h"
#include "filter_internal.h"
// Note about connections:
// They can be confusing, because pins come in pairs, and multiple pins can be
// transitively connected via mp_pin_connect(). To avoid dealing with this,
// mp_pin.conn is used to skip redundant connected pins.
// Consider <1a|1b> a symbol for mp_pin pair #1 and f1 as filter #1. Then:
// f1 <-> <1a|1b> <-> <2a|2b> <-> <3a|3b> <-> f2
// would be a connection from 1a to 3b. 1a could be a private pin of f1 (e.g.
// mp_filter.ppin[0]), and 1b would be the public pin (e.g. mp_filter.pin[0]).
// A user could have called mp_pin_connect(2a, 1b) mp_pin_connect(3a, 2b)
// (assuming 1b has dir==MP_PIN_OUT). The end result are the following values:
// pin user_conn conn manual_connection within_conn (uses mp_pin.data)
// 1a NULL 3b f1 false no
// 1b 2a NULL NULL true no
// 2a 1b NULL NULL true no
// 2b 3a NULL NULL true no
// 3a 2b NULL NULL true no
// 3b NULL 1a f2 false yes
// The minimal case of f1 <-> <1a|1b> <-> f2 (1b dir=out) would be:
// 1a NULL 1b f1 false no
// 1b NULL 1a f2 false yes
// In both cases, only the final output pin uses mp_pin.data/data_requested.
struct mp_pin {
const char *name;
enum mp_pin_dir dir;
struct mp_pin *other; // paired mp_pin representing other end
struct mp_filter *owner;
struct mp_pin *user_conn; // as set by mp_pin_connect()
struct mp_pin *conn; // transitive, actual end of the connection
// Set if the pin is considered connected, but has no user_conn. pin
// state changes are handled by the given filter. (Defaults to the root
// filter if the pin is for the user of a filter graph.)
// As an invariant, conn and manual_connection are both either set or unset.
struct mp_filter *manual_connection;
// Set if the pin is indirect part of a connection chain, but not one of
// the end pins. Basically it's a redundant in-between pin. You never access
// these with the pin data flow functions, because only the end pins matter.
// This flag is for checking and enforcing this.
bool within_conn;
// This is used for the final output mp_pin in connections only.
bool data_requested; // true if out wants new data
struct mp_frame data; // possibly buffered frame (MP_FRAME_NONE if
// empty, usually only temporary)
};
// Root filters create this, all other filters reference it.
struct filter_runner {
struct mpv_global *global;
void (*wakeup_cb)(void *ctx);
void *wakeup_ctx;
struct mp_filter *root_filter;
// If we're currently running the filter graph (for avoiding recursion).
bool filtering;
// Set of filters which need process() to be called. A filter is in this
// array iff mp_filter_internal.pending==true.
struct mp_filter **pending;
int num_pending;
// Any outside pins have changed state.
bool external_pending;
// For async notifications only. We don't bother making this fine grained
// across filters.
pthread_mutex_t async_lock;
// Wakeup is pending. Protected by async_lock.
bool async_wakeup_sent;
// Similar to pending[]. Uses mp_filter_internal.async_pending. Protected
// by async_lock.
struct mp_filter **async_pending;
int num_async_pending;
};
struct mp_filter_internal {
const struct mp_filter_info *info;
struct mp_filter *parent;
struct filter_runner *runner;
struct mp_filter **children;
int num_children;
struct mp_filter *error_handler;
char *name;
bool pending;
bool async_pending;
bool failed;
};
static void add_pending(struct mp_filter *f)
{
struct filter_runner *r = f->in->runner;
if (f->in->pending)
return;
// This should probably really be some sort of priority queue, but for now
// something naive and dumb does the job too.
f->in->pending = true;
MP_TARRAY_APPEND(r, r->pending, r->num_pending, f);
}
// Called when new work needs to be done on a pin belonging to the filter:
// - new data was requested
// - new data has been queued
// - or just an connect/disconnect/async notification happened
// This means the process function for this filter has to be called next.
static void update_filter(struct mp_filter *src, struct mp_filter *f)
{
assert(f);
struct filter_runner *r = f->in->runner;
// Make sure the filter knows it has to make progress.
if (src->in->runner != r) {
// Connected to a different graph. The user has to drive those manually,
// and we simplify tell the user via the mp_filter_run() return value.
r->external_pending = true;
} else if (!f->in->pending) {
add_pending(f);
if (!r->filtering) {
// Likely the "outer" API user used an external manually connected
// pin, so do recursive filtering (as a not strictly necessary
// feature which makes outside I/O with filters easier).
// Also don't lose the pending state, which the user may or may not
// care about.
// Note that we must avoid calling this from within filtering,
// because that would make the process() functions recursive and
// reentrant (and hard to reason about).
r->external_pending |= mp_filter_run(r->root_filter);
}
// Need to tell user that something changed.
if (f == r->root_filter)
r->external_pending = true;
}
}
void mp_filter_internal_mark_progress(struct mp_filter *f)
{
struct filter_runner *r = f->in->runner;
assert(r->filtering); // only call from f's process()
add_pending(f);
}
// Basically copy the async notifications to the sync ones. Done so that the
// sync notifications don't need any locking.
static void flush_async_notifications(struct filter_runner *r, bool queue)
{
pthread_mutex_lock(&r->async_lock);
for (int n = 0; n < r->num_async_pending; n++) {
struct mp_filter *f = r->async_pending[n];
if (queue)
add_pending(f);
f->in->async_pending = false;
}
r->num_async_pending = 0;
r->async_wakeup_sent = false;
pthread_mutex_unlock(&r->async_lock);
}
bool mp_filter_run(struct mp_filter *filter)
{
struct filter_runner *r = filter->in->runner;
r->filtering = true;
flush_async_notifications(r, true);
while (r->num_pending) {
struct mp_filter *next = r->pending[r->num_pending - 1];
r->num_pending -= 1;
next->in->pending = false;
if (next->in->info->process)
next->in->info->process(next);
}
r->filtering = false;
bool externals = r->external_pending;
r->external_pending = false;
return externals;
}
bool mp_pin_can_transfer_data(struct mp_pin *dst, struct mp_pin *src)
{
return mp_pin_in_needs_data(dst) && mp_pin_out_request_data(src);
}
bool mp_pin_transfer_data(struct mp_pin *dst, struct mp_pin *src)
{
if (!mp_pin_can_transfer_data(dst, src))
return false;
mp_pin_in_write(dst, mp_pin_out_read(src));
return true;
}
bool mp_pin_in_needs_data(struct mp_pin *p)
{
assert(p->dir == MP_PIN_IN);
assert(!p->within_conn);
return p->conn && p->conn->manual_connection && p->conn->data_requested;
}
bool mp_pin_in_write(struct mp_pin *p, struct mp_frame frame)
{
if (!mp_pin_in_needs_data(p) || frame.type == MP_FRAME_NONE) {
if (frame.type)
MP_ERR(p->owner, "losing frame on %s\n", p->name);
mp_frame_unref(&frame);
return false;
}
assert(p->conn->data.type == MP_FRAME_NONE);
p->conn->data = frame;
p->conn->data_requested = false;
update_filter(p->owner, p->conn->manual_connection);
return true;
}
bool mp_pin_out_has_data(struct mp_pin *p)
{
assert(p->dir == MP_PIN_OUT);
assert(!p->within_conn);
return p->conn && p->conn->manual_connection && p->data.type != MP_FRAME_NONE;
}
bool mp_pin_out_request_data(struct mp_pin *p)
{
if (mp_pin_out_has_data(p))
return true;
if (p->conn && p->conn->manual_connection && !p->data_requested) {
p->data_requested = true;
update_filter(p->owner, p->conn->manual_connection);
}
return mp_pin_out_has_data(p);
}
struct mp_frame mp_pin_out_read(struct mp_pin *p)
{
if (!mp_pin_out_request_data(p))
return MP_NO_FRAME;
struct mp_frame res = p->data;
p->data = MP_NO_FRAME;
return res;
}
void mp_pin_out_unread(struct mp_pin *p, struct mp_frame frame)
{
assert(p->dir == MP_PIN_OUT);
assert(!p->within_conn);
assert(p->conn && p->conn->manual_connection);
// Unread is allowed strictly only if you didn't do anything else with
// the pin since the time you read it.
assert(!mp_pin_out_has_data(p));
assert(!p->data_requested);
p->data = frame;
}
void mp_pin_out_repeat_eof(struct mp_pin *p)
{
mp_pin_out_unread(p, MP_EOF_FRAME);
}
// Follow mp_pin pairs/connection into the "other" direction of the pin, until
// the last pin is found. (In the simplest case, this is just p->other.) E.g.:
// <1a|1b> <-> <2a|2b> <-> <3a|3b>
// find_connected_end(2b)==1a
// find_connected_end(1b)==1a
// find_connected_end(1a)==3b
static struct mp_pin *find_connected_end(struct mp_pin *p)
{
while (1) {
struct mp_pin *other = p->other;
if (!other->user_conn)
return other;
p = other->user_conn;
}
assert(0);
}
// With p being part of a connection, create the pin_connection and set all
// state flags.
static void init_connection(struct mp_pin *p)
{
if (p->dir == MP_PIN_IN)
p = p->other;
struct mp_pin *in = find_connected_end(p);
struct mp_pin *out = find_connected_end(p->other);
// These are the "outer" pins by definition, they have no user connections.
assert(!in->user_conn);
assert(!out->user_conn);
// Logicaly, the ends are always manual connections. A pin chain without
// manual connections at the ends is still disconnected (or if this
// attempted to extend an existing connection, becomes dangling and gets
// disconnected).
if (!in->manual_connection && !out->manual_connection)
return;
assert(in->dir == MP_PIN_IN);
assert(out->dir == MP_PIN_OUT);
struct mp_pin *cur = in;
while (cur) {
assert(!cur->within_conn && !cur->other->within_conn);
assert(!cur->conn && !cur->other->conn);
assert(!cur->data_requested); // unused for in pins
assert(!cur->data.type); // unused for in pins
assert(!cur->other->data_requested); // unset for unconnected out pins
assert(!cur->other->data.type); // unset for unconnected out pins
cur->within_conn = cur->other->within_conn = true;
cur = cur->other->user_conn;
}
in->conn = out;
in->within_conn = false;
out->conn = in;
out->within_conn = false;
// Scheduling so far will be messed up.
add_pending(in->manual_connection);
add_pending(out->manual_connection);
}
void mp_pin_connect(struct mp_pin *dst, struct mp_pin *src)
{
assert(src->dir == MP_PIN_OUT);
assert(dst->dir == MP_PIN_IN);
if (dst->user_conn == src) {
assert(src->user_conn == dst);
return;
}
mp_pin_disconnect(src);
mp_pin_disconnect(dst);
src->user_conn = dst;
dst->user_conn = src;
init_connection(src);
}
void mp_pin_set_manual_connection(struct mp_pin *p, bool connected)
{
mp_pin_set_manual_connection_for(p, connected ? p->owner->in->parent : NULL);
}
void mp_pin_set_manual_connection_for(struct mp_pin *p, struct mp_filter *f)
{
if (p->manual_connection == f)
return;
if (p->within_conn)
mp_pin_disconnect(p);
p->manual_connection = f;
init_connection(p);
}
struct mp_filter *mp_pin_get_manual_connection(struct mp_pin *p)
{
return p->manual_connection;
}
static void deinit_connection(struct mp_pin *p)
{
if (p->dir == MP_PIN_OUT)
p = p->other;
p = find_connected_end(p);
while (p) {
p->conn = p->other->conn = NULL;
p->within_conn = p->other->within_conn = false;
assert(!p->other->data_requested); // unused for in pins
assert(!p->other->data.type); // unused for in pins
p->data_requested = false;
if (p->data.type)
MP_WARN(p->owner, "dropping frame due to pin disconnect\n");
if (p->data_requested)
MP_WARN(p->owner, "dropping request due to pin disconnect\n");
mp_frame_unref(&p->data);
p = p->other->user_conn;
}
}
void mp_pin_disconnect(struct mp_pin *p)
{
if (!mp_pin_is_connected(p))
return;
p->manual_connection = NULL;
struct mp_pin *conn = p->user_conn;
if (conn) {
p->user_conn = NULL;
conn->user_conn = NULL;
deinit_connection(conn);
}
deinit_connection(p);
}
bool mp_pin_is_connected(struct mp_pin *p)
{
return p->user_conn || p->manual_connection;
}
const char *mp_pin_get_name(struct mp_pin *p)
{
return p->name;
}
enum mp_pin_dir mp_pin_get_dir(struct mp_pin *p)
{
return p->dir;
}
const char *mp_filter_get_name(struct mp_filter *f)
{
return f->in->name;
}
void mp_filter_set_name(struct mp_filter *f, const char *name)
{
talloc_free(f->in->name);
f->in->name = talloc_strdup(f, name);
}
struct mp_pin *mp_filter_get_named_pin(struct mp_filter *f, const char *name)
{
for (int n = 0; n < f->num_pins; n++) {
if (name && strcmp(f->pins[n]->name, name) == 0)
return f->pins[n];
}
return NULL;
}
void mp_filter_set_error_handler(struct mp_filter *f, struct mp_filter *handler)
{
f->in->error_handler = handler;
}
void mp_filter_internal_mark_failed(struct mp_filter *f)
{
while (f) {
f->in->failed = true;
if (f->in->error_handler) {
add_pending(f->in->error_handler);
break;
}
f = f->in->parent;
}
}
bool mp_filter_has_failed(struct mp_filter *filter)
{
bool failed = filter->in->failed;
filter->in->failed = false;
return failed;
}
static void reset_pin(struct mp_pin *p)
{
if (!p->conn || p->dir != MP_PIN_OUT) {
assert(!p->data.type);
assert(!p->data_requested);
}
mp_frame_unref(&p->data);
p->data_requested = false;
}
void mp_filter_reset(struct mp_filter *filter)
{
for (int n = 0; n < filter->in->num_children; n++)
mp_filter_reset(filter->in->children[n]);
for (int n = 0; n < filter->num_pins; n++) {
struct mp_pin *p = filter->ppins[n];
reset_pin(p);
reset_pin(p->other);
}
if (filter->in->info->reset)
filter->in->info->reset(filter);
}
struct mp_pin *mp_filter_add_pin(struct mp_filter *f, enum mp_pin_dir dir,
const char *name)
{
assert(dir == MP_PIN_IN || dir == MP_PIN_OUT);
assert(name && name[0]);
assert(!mp_filter_get_named_pin(f, name));
// "Public" pin
struct mp_pin *p = talloc_ptrtype(NULL, p);
*p = (struct mp_pin){
.name = talloc_strdup(p, name),
.dir = dir,
.owner = f,
.manual_connection = f->in->parent,
};
// "Private" paired pin
p->other = talloc_ptrtype(NULL, p);
*p->other = (struct mp_pin){
.name = p->name,
.dir = p->dir == MP_PIN_IN ? MP_PIN_OUT : MP_PIN_IN,
.owner = f,
.other = p,
.manual_connection = f,
};
MP_TARRAY_GROW(f, f->pins, f->num_pins);
MP_TARRAY_GROW(f, f->ppins, f->num_pins);
f->pins[f->num_pins] = p;
f->ppins[f->num_pins] = p->other;
f->num_pins += 1;
init_connection(p);
return p->other;
}
void mp_filter_remove_pin(struct mp_filter *f, struct mp_pin *p)
{
if (!p)
return;
assert(p->owner == f);
mp_pin_disconnect(p);
mp_pin_disconnect(p->other);
int index = -1;
for (int n = 0; n < f->num_pins; n++) {
if (f->ppins[n] == p) {
index = n;
break;
}
}
assert(index >= 0);
talloc_free(f->pins[index]);
talloc_free(f->ppins[index]);
int count = f->num_pins;
MP_TARRAY_REMOVE_AT(f->pins, count, index);
count = f->num_pins;
MP_TARRAY_REMOVE_AT(f->ppins, count, index);
f->num_pins -= 1;
}
bool mp_filter_command(struct mp_filter *f, struct mp_filter_command *cmd)
{
return f->in->info->command ? f->in->info->command(f, cmd) : false;
}
struct mp_stream_info *mp_filter_find_stream_info(struct mp_filter *f)
{
while (f) {
if (f->stream_info)
return f->stream_info;
f = f->in->parent;
}
return NULL;
}
struct AVBufferRef *mp_filter_load_hwdec_device(struct mp_filter *f, int avtype)
{
struct mp_stream_info *info = mp_filter_find_stream_info(f);
if (!info || !info->hwdec_devs)
return NULL;
hwdec_devices_request_all(info->hwdec_devs);
return hwdec_devices_get_lavc(info->hwdec_devs, avtype);
}
static void filter_wakeup(struct mp_filter *f, bool mark_only)
{
struct filter_runner *r = f->in->runner;
pthread_mutex_lock(&r->async_lock);
if (!f->in->async_pending) {
f->in->async_pending = true;
// (not using a talloc parent for thread safety reasons)
MP_TARRAY_APPEND(NULL, r->async_pending, r->num_async_pending, f);
if (!mark_only && !r->async_wakeup_sent) {
if (r->wakeup_cb)
r->wakeup_cb(r->wakeup_ctx);
r->async_wakeup_sent = true;
}
}
pthread_mutex_unlock(&r->async_lock);
}
void mp_filter_wakeup(struct mp_filter *f)
{
filter_wakeup(f, false);
}
void mp_filter_mark_async_progress(struct mp_filter *f)
{
filter_wakeup(f, true);
}
void mp_filter_free_children(struct mp_filter *f)
{
while(f->in->num_children)
talloc_free(f->in->children[0]);
}
static void filter_destructor(void *p)
{
struct mp_filter *f = p;
struct filter_runner *r = f->in->runner;
if (f->in->info->destroy)
f->in->info->destroy(f);
// For convenience, free child filters.
mp_filter_free_children(f);
while (f->num_pins)
mp_filter_remove_pin(f, f->ppins[0]);
// Just make sure the filter is not still in the async notifications set.
// There will be no more new notifications at this point (due to destroy()).
flush_async_notifications(r, false);
for (int n = 0; n < r->num_pending; n++) {
if (r->pending[n] == f) {
MP_TARRAY_REMOVE_AT(r->pending, r->num_pending, n);
break;
}
}
if (f->in->parent) {
struct mp_filter_internal *p_in = f->in->parent->in;
for (int n = 0; n < p_in->num_children; n++) {
if (p_in->children[n] == f) {
MP_TARRAY_REMOVE_AT(p_in->children, p_in->num_children, n);
break;
}
}
}
if (r->root_filter == f) {
assert(!f->in->parent);
pthread_mutex_destroy(&r->async_lock);
talloc_free(r->async_pending);
talloc_free(r);
}
}
struct mp_filter *mp_filter_create_with_params(struct mp_filter_params *params)
{
struct mp_filter *f = talloc(NULL, struct mp_filter);
talloc_set_destructor(f, filter_destructor);
*f = (struct mp_filter){
.priv = params->info->priv_size ?
talloc_zero_size(f, params->info->priv_size) : NULL,
.global = params->global,
.in = talloc(f, struct mp_filter_internal),
};
*f->in = (struct mp_filter_internal){
.info = params->info,
.parent = params->parent,
.runner = params->parent ? params->parent->in->runner : NULL,
};
if (!f->in->runner) {
assert(params->global);
f->in->runner = talloc(NULL, struct filter_runner);
*f->in->runner = (struct filter_runner){
.global = params->global,
.root_filter = f,
};
pthread_mutex_init(&f->in->runner->async_lock, NULL);
}
if (!f->global)
f->global = f->in->runner->global;
if (f->in->parent) {
struct mp_filter_internal *parent = f->in->parent->in;
MP_TARRAY_APPEND(parent, parent->children, parent->num_children, f);
}
f->log = mp_log_new(f, f->global->log, params->info->name);
if (f->in->info->init) {
if (!f->in->info->init(f, params)) {
talloc_free(f);
return NULL;
}
}
return f;
}
struct mp_filter *mp_filter_create(struct mp_filter *parent,
const struct mp_filter_info *info)
{
assert(parent);
assert(info);
struct mp_filter_params params = {
.info = info,
.parent = parent,
};
return mp_filter_create_with_params(&params);
}
// (the root filter is just a dummy filter - nothing special about it, except
// that it has no parent, and serves as manual connection for "external" pins)
static const struct mp_filter_info filter_root = {
.name = "root",
};
struct mp_filter *mp_filter_create_root(struct mpv_global *global)
{
struct mp_filter_params params = {
.info = &filter_root,
.global = global,
};
return mp_filter_create_with_params(&params);
}
void mp_filter_root_set_wakeup_cb(struct mp_filter *root,
void (*wakeup_cb)(void *ctx), void *ctx)
{
struct filter_runner *r = root->in->runner;
pthread_mutex_lock(&r->async_lock);
r->wakeup_cb = wakeup_cb;
r->wakeup_ctx = ctx;
pthread_mutex_unlock(&r->async_lock);
}
static const char *filt_name(struct mp_filter *f)
{
return f ? f->in->info->name : "-";
}
static void dump_pin_state(struct mp_filter *f, struct mp_pin *pin)
{
MP_WARN(f, " [%p] %s %s c=%s[%p] f=%s[%p] m=%s[%p] %s %s %s\n",
pin, pin->name, pin->dir == MP_PIN_IN ? "->" : "<-",
pin->user_conn ? filt_name(pin->user_conn->owner) : "-", pin->user_conn,
pin->conn ? filt_name(pin->conn->owner) : "-", pin->conn,
filt_name(pin->manual_connection), pin->manual_connection,
pin->within_conn ? "(within)" : "",
pin->data_requested ? "(request)" : "",
mp_frame_type_str(pin->data.type));
}
void mp_filter_dump_states(struct mp_filter *f)
{
MP_WARN(f, "%s[%p] (%s[%p])\n", filt_name(f), f,
filt_name(f->in->parent), f->in->parent);
for (int n = 0; n < f->num_pins; n++) {
dump_pin_state(f, f->pins[n]);
dump_pin_state(f, f->ppins[n]);
}
for (int n = 0; n < f->in->num_children; n++)
mp_filter_dump_states(f->in->children[n]);
}