RepoMirrors
/
mpv
mirror of https://github.com/mpv-player/mpv
1
0
Fork 0
Code Issues Releases Wiki Activity
mpv/filters/filter.c

791 lines
23 KiB
C
Raw Blame History

#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]);
}
Reference in New Issue View Git Blame Copy Permalink