From c25834a692ac6d5ccadbd9bce55d6272a38cbd80 Mon Sep 17 00:00:00 2001 From: Wim Taymans Date: Tue, 22 Aug 2017 18:30:10 +0200 Subject: [PATCH] Work on cleanup Add signal handlers for daemon to shut down Add destroy listeners for modules and do cleanup Fix some leaks Simplify port registration in new nodes Hide some more data structures Let the node implementation take care of the reuse_buffer signal because then we can get to the graph nodes to find the destination port. Destroy modules in core cleanup. Modules should undo everything they have done. Activate the link after we negotiated format and buffers. --- spa/plugins/support/loop.c | 6 +- src/daemon/main.c | 16 +- src/modules/module-autolink.c | 40 +++- src/modules/module-client-node.c | 30 ++- src/modules/module-client-node/client-node.c | 2 +- src/modules/module-flatpak.c | 29 +++ src/modules/module-jack.c | 41 ++-- src/modules/module-mixer.c | 31 ++- src/modules/module-protocol-native.c | 229 ++++++++++--------- src/modules/module-suspend-on-idle.c | 37 ++- src/modules/spa/module-monitor.c | 33 ++- src/modules/spa/module-node-factory.c | 2 +- src/modules/spa/module-node.c | 6 +- src/modules/spa/spa-monitor.c | 21 +- src/modules/spa/spa-monitor.h | 6 +- src/modules/spa/spa-node.c | 127 ++++------ src/modules/spa/spa-node.h | 6 +- src/pipewire/core.c | 4 + src/pipewire/global.h | 3 +- src/pipewire/link.c | 26 ++- src/pipewire/link.h | 1 + src/pipewire/module.c | 1 + src/pipewire/node-factory.c | 18 +- src/pipewire/node-factory.h | 2 + src/pipewire/node.c | 38 ++- src/pipewire/node.h | 8 + src/pipewire/port.c | 9 + src/pipewire/private.h | 20 +- src/pipewire/protocol.c | 41 +++- src/pipewire/protocol.h | 85 +++---- src/pipewire/remote.c | 12 +- 31 files changed, 586 insertions(+), 344 deletions(-) diff --git a/spa/plugins/support/loop.c b/spa/plugins/support/loop.c index 0f20194c6..5a203b69f 100644 --- a/spa/plugins/support/loop.c +++ b/spa/plugins/support/loop.c @@ -563,8 +563,10 @@ static void source_signal_func(struct spa_source *source) { struct source_impl *impl = SPA_CONTAINER_OF(source, struct source_impl, source); struct signalfd_siginfo signal_info; + int len; - if (read(source->fd, &signal_info, sizeof(signal_info)) != sizeof(signal_info)) + len = read(source->fd, &signal_info, sizeof signal_info); + if (!(len == -1 && errno == EAGAIN) && len != sizeof signal_info) spa_log_warn(impl->impl->log, NAME " %p: failed to read signal fd %d: %s", source, source->fd, strerror(errno)); @@ -586,10 +588,12 @@ static struct spa_source *loop_add_signal(struct spa_loop_utils *utils, source->source.loop = &impl->loop; source->source.func = source_signal_func; source->source.data = data; + sigemptyset(&mask); sigaddset(&mask, signal_number); source->source.fd = signalfd(-1, &mask, SFD_CLOEXEC | SFD_NONBLOCK); sigprocmask(SIG_BLOCK, &mask, NULL); + source->source.mask = SPA_IO_IN; source->impl = impl; source->close = true; diff --git a/src/daemon/main.c b/src/daemon/main.c index 3566f9c39..88416b153 100644 --- a/src/daemon/main.c +++ b/src/daemon/main.c @@ -17,12 +17,20 @@ * Boston, MA 02110-1301, USA. */ +#include + #include #include #include #include "daemon-config.h" +static void do_quit(void *data, int signal_number) +{ + struct pw_main_loop *loop = data; + pw_main_loop_quit(loop); +} + int main(int argc, char *argv[]) { struct pw_core *core; @@ -45,16 +53,20 @@ int main(int argc, char *argv[]) "pipewire.daemon", "1", NULL); loop = pw_main_loop_new(props); + pw_loop_add_signal(pw_main_loop_get_loop(loop), SIGINT, do_quit, loop); + pw_loop_add_signal(pw_main_loop_get_loop(loop), SIGTERM, do_quit, loop); core = pw_core_new(pw_main_loop_get_loop(loop), props); pw_daemon_config_run_commands(config, core); + pw_log_info("start main loop"); pw_main_loop_run(loop); + pw_log_info("leave main loop"); - pw_main_loop_destroy(loop); - + pw_daemon_config_free(config); pw_core_destroy(core); + pw_main_loop_destroy(loop); return 0; } diff --git a/src/modules/module-autolink.c b/src/modules/module-autolink.c index 839da8689..60ecf102d 100644 --- a/src/modules/module-autolink.c +++ b/src/modules/module-autolink.c @@ -36,6 +36,7 @@ struct impl { struct pw_properties *properties; struct spa_hook core_listener; + struct spa_hook module_listener; struct spa_list node_list; }; @@ -71,8 +72,11 @@ static struct node_info *find_node_info(struct impl *impl, struct pw_node *node) static void link_data_remove(struct link_data *data) { - spa_list_remove(&data->l); - spa_hook_remove(&data->link_listener); + if (data->node_info) { + spa_list_remove(&data->l); + spa_hook_remove(&data->link_listener); + data->node_info = NULL; + } } static void node_info_free(struct node_info *info) @@ -290,6 +294,27 @@ core_global_removed(void *data, struct pw_global *global) } } +static void module_destroy(void *data) +{ + struct impl *impl = data; + struct node_info *info, *t; + + spa_list_for_each_safe(info, t, &impl->node_list, l) + node_info_free(info); + + spa_hook_remove(&impl->core_listener); + spa_hook_remove(&impl->module_listener); + + if (impl->properties) + pw_properties_free(impl->properties); + + free(impl); +} + +const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; const struct pw_core_events core_events = { PW_VERSION_CORE_EVENTS, @@ -322,20 +347,11 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti spa_list_init(&impl->node_list); pw_core_add_listener(core, &impl->core_listener, &core_events, impl); + pw_module_add_listener(module, &impl->module_listener, &module_events, impl); return impl; } -#if 0 -static void module_destroy(struct impl *impl) -{ - pw_log_debug("module %p: destroy", impl); - - spa_hook_remove(&impl->core_listener); - free(impl); -} -#endif - bool pipewire__module_init(struct pw_module *module, const char *args) { return module_init(module, NULL); diff --git a/src/modules/module-client-node.c b/src/modules/module-client-node.c index 05c437fac..0306d406b 100644 --- a/src/modules/module-client-node.c +++ b/src/modules/module-client-node.c @@ -36,6 +36,8 @@ struct pw_protocol *pw_protocol_native_ext_client_node_init(struct pw_core *core struct factory_data { struct pw_node_factory *this; struct pw_properties *properties; + + struct spa_hook module_listener; }; static struct pw_node *create_node(void *_data, @@ -64,6 +66,23 @@ static const struct pw_node_factory_implementation impl_factory = { .create_node = create_node, }; +static void module_destroy(void *data) +{ + struct factory_data *d = data; + + spa_hook_remove(&d->module_listener); + + if (d->properties) + pw_properties_free(d->properties); + + pw_node_factory_destroy(d->this); +} + +const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; + static bool module_init(struct pw_module *module, struct pw_properties *properties) { struct pw_core *core = pw_module_get_core(module); @@ -88,18 +107,11 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti pw_node_factory_export(factory, NULL, pw_module_get_global(module)); + pw_module_add_listener(module, &data->module_listener, &module_events, data); + return true; } -#if 0 -static void module_destroy(struct impl *impl) -{ - pw_log_debug("module %p: destroy", impl); - - free(impl); -} -#endif - bool pipewire__module_init(struct pw_module *module, const char *args) { return module_init(module, NULL); diff --git a/src/modules/module-client-node/client-node.c b/src/modules/module-client-node/client-node.c index 83fbbc2d3..671f4ceb1 100644 --- a/src/modules/module-client-node/client-node.c +++ b/src/modules/module-client-node/client-node.c @@ -1165,7 +1165,7 @@ struct pw_client_node *pw_client_node_new(struct pw_resource *resource, true, &impl->proxy.node, NULL, - properties); + properties, 0); if (this->node == NULL) goto error_no_node; diff --git a/src/modules/module-flatpak.c b/src/modules/module-flatpak.c index 0777cd437..65415641b 100644 --- a/src/modules/module-flatpak.c +++ b/src/modules/module-flatpak.c @@ -44,6 +44,7 @@ struct impl { DBusConnection *bus; struct spa_hook core_listener; + struct spa_hook module_listener; struct spa_list client_list; @@ -711,6 +712,33 @@ static void wakeup_main(void *userdata) pw_loop_enable_idle(pw_core_get_main_loop(impl->core), impl->dispatch_event, true); } +static void module_destroy(void *data) +{ + struct impl *impl = data; + struct client_info *info, *t; + + spa_hook_remove(&impl->core_listener); + spa_hook_remove(&impl->module_listener); + + dbus_connection_close(impl->bus); + dbus_connection_unref(impl->bus); + + spa_list_for_each_safe(info, t, &impl->client_list, link) + client_info_free(info); + + pw_loop_destroy_source(pw_core_get_main_loop(impl->core), impl->dispatch_event); + + if (impl->properties) + pw_properties_free(impl->properties); + + free(impl); +} + +const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; + static bool module_init(struct pw_module *module, struct pw_properties *properties) { struct pw_core *core = pw_module_get_core(module); @@ -743,6 +771,7 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti spa_list_init(&impl->client_list); pw_core_add_listener(core, &impl->core_listener, &core_events, impl); + pw_module_add_listener(module, &impl->module_listener, &module_events, impl); pw_core_set_permission_callback(core, do_permission, impl); diff --git a/src/modules/module-jack.c b/src/modules/module-jack.c index 6b8af786f..cd178e441 100644 --- a/src/modules/module-jack.c +++ b/src/modules/module-jack.c @@ -73,7 +73,7 @@ struct impl { struct pw_core *core; struct pw_type *t; struct pw_module *module; - struct spa_list link; + struct spa_hook module_listener; struct spa_source *timer; @@ -1434,9 +1434,6 @@ static int init_server(struct impl *impl, const char *name, bool promiscuous) for (i = 0; i < CLIENT_NUM; i++) server->synchro_table[i] = JACK_SYNCHRO_INIT; - if (!init_nodes(impl)) - return -1; - s = create_socket(); if (!init_socket_name(&s->addr, name, promiscuous, 0)) @@ -1445,6 +1442,9 @@ static int init_server(struct impl *impl, const char *name, bool promiscuous) if (!add_socket(impl, s)) goto error; + if (!init_nodes(impl)) + goto error; + return 0; error: @@ -1452,6 +1452,26 @@ static int init_server(struct impl *impl, const char *name, bool promiscuous) return -1; } +static void module_destroy(void *data) +{ + struct impl *impl = data; + struct link *ld, *t; + + spa_hook_remove(&impl->module_listener); + + spa_list_for_each_safe(ld, t, &impl->link_list, link_link) + pw_link_destroy(ld->link); + + if (impl->properties) + pw_properties_free(impl->properties); + + free(impl); +} + +static const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; static bool module_init(struct pw_module *module, struct pw_properties *properties) { @@ -1492,6 +1512,8 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti if (init_server(impl, name, promiscuous) < 0) goto error; + pw_module_add_listener(module, &impl->module_listener, &module_events, impl); + return true; error: @@ -1499,17 +1521,6 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti return false; } -#if 0 -static void module_destroy(struct impl *impl) -{ - struct impl *object, *tmp; - - pw_log_debug("module %p: destroy", impl); - - free(impl); -} -#endif - bool pipewire__module_init(struct pw_module *module, const char *args) { return module_init(module, NULL); diff --git a/src/modules/module-mixer.c b/src/modules/module-mixer.c index e15b24087..6b6aec02a 100644 --- a/src/modules/module-mixer.c +++ b/src/modules/module-mixer.c @@ -37,6 +37,7 @@ struct impl { struct pw_core *core; struct pw_type *t; struct pw_module *module; + struct spa_hook module_listener; struct pw_properties *properties; void *hnd; @@ -119,7 +120,7 @@ static struct pw_node *make_node(struct impl *impl) spa_clock = iface; node = pw_spa_node_new(impl->core, NULL, pw_module_get_global(impl->module), - "audiomixer", false, spa_node, spa_clock, NULL); + "audiomixer", false, spa_node, spa_clock, NULL, 0); return node; @@ -166,6 +167,23 @@ static bool on_global(void *data, struct pw_global *global) return true; } +static void module_destroy(void *data) +{ + struct impl *impl = data; + + spa_hook_remove(&impl->module_listener); + + if (impl->properties) + pw_properties_free(impl->properties); + + free(impl); +} + +static const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; + static bool module_init(struct pw_module *module, struct pw_properties *properties) { struct pw_core *core = pw_module_get_core(module); @@ -183,18 +201,11 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti pw_core_for_each_global(core, on_global, impl); + pw_module_add_listener(module, &impl->module_listener, &module_events, impl); + return true; } -#if 0 -static void module_destroy(struct impl *impl) -{ - pw_log_debug("module %p: destroy", impl); - - free(impl); -} -#endif - bool pipewire__module_init(struct pw_module *module, const char *args) { return module_init(module, NULL); diff --git a/src/modules/module-protocol-native.c b/src/modules/module-protocol-native.c index cd2ab5ddd..22d541d37 100644 --- a/src/modules/module-protocol-native.c +++ b/src/modules/module-protocol-native.c @@ -57,10 +57,13 @@ void pw_protocol_native_init(struct pw_protocol *protocol); struct protocol_data { struct pw_module *module; + struct spa_hook module_listener; + struct pw_protocol *protocol; + struct pw_properties *properties; }; -struct connection { - struct pw_protocol_connection this; +struct client { + struct pw_protocol_client this; int fd; @@ -74,8 +77,8 @@ struct connection { struct spa_source *flush_event; }; -struct listener { - struct pw_protocol_listener this; +struct server { + struct pw_protocol_server this; int fd; int fd_lock; @@ -223,11 +226,11 @@ static const struct pw_client_events client_events = { .busy_changed = client_busy_changed, }; -static struct pw_client *client_new(struct listener *l, int fd) +static struct pw_client *client_new(struct server *s, int fd) { struct client_data *this; struct pw_client *client; - struct pw_protocol *protocol = l->this.protocol; + struct pw_protocol *protocol = s->this.protocol; struct protocol_data *pd = protocol->user_data; socklen_t len; struct ucred ucred, *ucredp; @@ -263,7 +266,7 @@ static struct pw_client *client_new(struct listener *l, int fd) goto no_connection; client->protocol = protocol; - spa_list_insert(l->this.client_list.prev, &client->protocol_link); + spa_list_append(&s->this.client_list, &client->protocol_link); pw_client_add_listener(client, &this->client_listener, &client_events, this); @@ -279,7 +282,7 @@ static struct pw_client *client_new(struct listener *l, int fd) return NULL; } -static bool init_socket_name(struct listener *l, const char *name) +static bool init_socket_name(struct server *s, const char *name) { int name_size; const char *runtime_dir; @@ -289,62 +292,62 @@ static bool init_socket_name(struct listener *l, const char *name) return false; } - l->addr.sun_family = AF_LOCAL; - name_size = snprintf(l->addr.sun_path, sizeof(l->addr.sun_path), + s->addr.sun_family = AF_LOCAL; + name_size = snprintf(s->addr.sun_path, sizeof(s->addr.sun_path), "%s/%s", runtime_dir, name) + 1; - if (name_size > (int) sizeof(l->addr.sun_path)) { + if (name_size > (int) sizeof(s->addr.sun_path)) { pw_log_error("socket path \"%s/%s\" plus null terminator exceeds 108 bytes", runtime_dir, name); - *l->addr.sun_path = 0; + *s->addr.sun_path = 0; return false; } return true; } -static bool lock_socket(struct listener *l) +static bool lock_socket(struct server *s) { struct stat socket_stat; - snprintf(l->lock_addr, sizeof(l->lock_addr), "%s%s", l->addr.sun_path, LOCK_SUFFIX); + snprintf(s->lock_addr, sizeof(s->lock_addr), "%s%s", s->addr.sun_path, LOCK_SUFFIX); - l->fd_lock = open(l->lock_addr, O_CREAT | O_CLOEXEC, + s->fd_lock = open(s->lock_addr, O_CREAT | O_CLOEXEC, (S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP)); - if (l->fd_lock < 0) { - pw_log_error("unable to open lockfile %s check permissions", l->lock_addr); + if (s->fd_lock < 0) { + pw_log_error("unable to open lockfile %s check permissions", s->lock_addr); goto err; } - if (flock(l->fd_lock, LOCK_EX | LOCK_NB) < 0) { + if (flock(s->fd_lock, LOCK_EX | LOCK_NB) < 0) { pw_log_error("unable to lock lockfile %s, maybe another daemon is running", - l->lock_addr); + s->lock_addr); goto err_fd; } - if (stat(l->addr.sun_path, &socket_stat) < 0) { + if (stat(s->addr.sun_path, &socket_stat) < 0) { if (errno != ENOENT) { - pw_log_error("did not manage to stat file %s\n", l->addr.sun_path); + pw_log_error("did not manage to stat file %s\n", s->addr.sun_path); goto err_fd; } } else if (socket_stat.st_mode & S_IWUSR || socket_stat.st_mode & S_IWGRP) { - unlink(l->addr.sun_path); + unlink(s->addr.sun_path); } return true; err_fd: - close(l->fd_lock); - l->fd_lock = -1; + close(s->fd_lock); + s->fd_lock = -1; err: - *l->lock_addr = 0; - *l->addr.sun_path = 0; + *s->lock_addr = 0; + *s->addr.sun_path = 0; return false; } static void socket_data(void *data, int fd, enum spa_io mask) { - struct listener *l = data; + struct server *s = data; struct pw_client *client; struct client_data *c; struct sockaddr_un name; @@ -358,7 +361,7 @@ socket_data(void *data, int fd, enum spa_io mask) return; } - client = client_new(l, client_fd); + client = client_new(s, client_fd); if (client == NULL) { pw_log_error("failed to create client"); close(client_fd); @@ -370,27 +373,27 @@ socket_data(void *data, int fd, enum spa_io mask) c->source, SPA_IO_IN | SPA_IO_ERR | SPA_IO_HUP); } -static bool add_socket(struct pw_protocol *protocol, struct listener *l) +static bool add_socket(struct pw_protocol *protocol, struct server *s) { socklen_t size; - if ((l->fd = socket(PF_LOCAL, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) + if ((s->fd = socket(PF_LOCAL, SOCK_STREAM | SOCK_CLOEXEC | SOCK_NONBLOCK, 0)) < 0) return false; - size = offsetof(struct sockaddr_un, sun_path) + strlen(l->addr.sun_path); - if (bind(l->fd, (struct sockaddr *) &l->addr, size) < 0) { + size = offsetof(struct sockaddr_un, sun_path) + strlen(s->addr.sun_path); + if (bind(s->fd, (struct sockaddr *) &s->addr, size) < 0) { pw_log_error("bind() failed with error: %m"); return false; } - if (listen(l->fd, 128) < 0) { + if (listen(s->fd, 128) < 0) { pw_log_error("listen() failed with error: %m"); return false; } - l->loop = pw_core_get_main_loop(protocol->core); - l->source = pw_loop_add_io(l->loop, l->fd, SPA_IO_IN, false, socket_data, l); - if (l->source == NULL) + s->loop = pw_core_get_main_loop(protocol->core); + s->source = pw_loop_add_io(s->loop, s->fd, SPA_IO_IN, false, socket_data, s); + if (s->source == NULL) return false; return true; @@ -410,7 +413,7 @@ get_name(const struct pw_properties *properties) return name; } -static int impl_connect(struct pw_protocol_connection *conn) +static int impl_connect(struct pw_protocol_client *conn) { struct sockaddr_un addr; socklen_t size; @@ -454,7 +457,7 @@ static int impl_connect(struct pw_protocol_connection *conn) static void on_remote_data(void *data, int fd, enum spa_io mask) { - struct connection *impl = data; + struct client *impl = data; struct pw_remote *this = impl->this.remote; struct pw_protocol_native_connection *conn = impl->connection; struct pw_core *core = pw_remote_get_core(this); @@ -522,7 +525,7 @@ on_remote_data(void *data, int fd, enum spa_io mask) static void do_flush_event(void *data, uint64_t count) { - struct connection *impl = data; + struct client *impl = data; impl->flush_signaled = false; if (impl->connection) if (!pw_protocol_native_connection_flush(impl->connection)) @@ -531,7 +534,7 @@ static void do_flush_event(void *data, uint64_t count) static void on_need_flush(void *data) { - struct connection *impl = data; + struct client *impl = data; struct pw_remote *remote = impl->this.remote; if (!impl->flush_signaled) { @@ -545,9 +548,9 @@ static const struct pw_protocol_native_connection_events conn_events = { .need_flush = on_need_flush, }; -static int impl_connect_fd(struct pw_protocol_connection *conn, int fd) +static int impl_connect_fd(struct pw_protocol_client *conn, int fd) { - struct connection *impl = SPA_CONTAINER_OF(conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(conn, struct client, this); struct pw_remote *remote = conn->remote; impl->connection = pw_protocol_native_connection_new(fd); @@ -572,9 +575,9 @@ static int impl_connect_fd(struct pw_protocol_connection *conn, int fd) return -1; } -static void impl_disconnect(struct pw_protocol_connection *conn) +static void impl_disconnect(struct pw_protocol_client *conn) { - struct connection *impl = SPA_CONTAINER_OF(conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(conn, struct client, this); struct pw_remote *remote = conn->remote; impl->disconnecting = true; @@ -592,9 +595,9 @@ static void impl_disconnect(struct pw_protocol_connection *conn) impl->fd = -1; } -static void impl_destroy(struct pw_protocol_connection *conn) +static void impl_destroy(struct pw_protocol_client *conn) { - struct connection *impl = SPA_CONTAINER_OF(conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(conn, struct client, this); struct pw_remote *remote = conn->remote; pw_loop_destroy_source(remote->core->main_loop, impl->flush_event); @@ -603,15 +606,15 @@ static void impl_destroy(struct pw_protocol_connection *conn) free(impl); } -static struct pw_protocol_connection * -impl_new_connection(struct pw_protocol *protocol, - struct pw_remote *remote, - struct pw_properties *properties) +static struct pw_protocol_client * +impl_new_client(struct pw_protocol *protocol, + struct pw_remote *remote, + struct pw_properties *properties) { - struct connection *impl; - struct pw_protocol_connection *this; + struct client *impl; + struct pw_protocol_client *this; - if ((impl = calloc(1, sizeof(struct connection))) == NULL) + if ((impl = calloc(1, sizeof(struct client))) == NULL) return NULL; this = &impl->this; @@ -625,32 +628,34 @@ impl_new_connection(struct pw_protocol *protocol, impl->flush_event = pw_loop_add_event(remote->core->main_loop, do_flush_event, impl); - spa_list_insert(protocol->connection_list.prev, &this->link); + spa_list_append(&protocol->client_list, &this->link); return this; } -static void destroy_listener(struct pw_protocol_listener *listener) +static void destroy_server(struct pw_protocol_server *server) { - struct listener *l = SPA_CONTAINER_OF(listener, struct listener, this); + struct server *s = SPA_CONTAINER_OF(server, struct server, this); - if (l->source) - pw_loop_destroy_source(l->loop, l->source); - if (l->addr.sun_path[0]) - unlink(l->addr.sun_path); - if (l->fd >= 0) - close(l->fd); - if (l->lock_addr[0]) - unlink(l->lock_addr); - if (l->fd_lock >= 0) - close(l->fd_lock); - free(l); + spa_list_remove(&server->link); + + if (s->source) + pw_loop_destroy_source(s->loop, s->source); + if (s->addr.sun_path[0]) + unlink(s->addr.sun_path); + if (s->fd >= 0) + close(s->fd); + if (s->lock_addr[0]) + unlink(s->lock_addr); + if (s->fd_lock >= 0) + close(s->fd_lock); + free(s); } static void on_before_hook(void *_data) { - struct listener *listener = _data; - struct pw_protocol_listener *this = &listener->this; + struct server *server = _data; + struct pw_protocol_server *this = &server->this; struct pw_client *client, *tmp; struct client_data *data; @@ -665,79 +670,79 @@ static const struct spa_loop_control_hooks impl_hooks = { .before = on_before_hook, }; -static struct pw_protocol_listener * -impl_add_listener(struct pw_protocol *protocol, - struct pw_core *core, - struct pw_properties *properties) +static struct pw_protocol_server * +impl_add_server(struct pw_protocol *protocol, + struct pw_core *core, + struct pw_properties *properties) { - struct pw_protocol_listener *this; - struct listener *l; + struct pw_protocol_server *this; + struct server *s; const char *name; - if ((l = calloc(1, sizeof(struct listener))) == NULL) + if ((s = calloc(1, sizeof(struct server))) == NULL) return NULL; - l->fd = -1; - l->fd_lock = -1; + s->fd = -1; + s->fd_lock = -1; - this = &l->this; + this = &s->this; this->protocol = protocol; spa_list_init(&this->client_list); - this->destroy = destroy_listener; + this->destroy = destroy_server; name = get_name(pw_core_get_properties(core)); - if (!init_socket_name(l, name)) + if (!init_socket_name(s, name)) goto error; - if (!lock_socket(l)) + if (!lock_socket(s)) goto error; - if (!add_socket(protocol, l)) + if (!add_socket(protocol, s)) goto error; - spa_list_insert(protocol->listener_list.prev, &this->link); + spa_list_append(&protocol->server_list, &this->link); - pw_loop_add_hook(pw_core_get_main_loop(core), &l->hook, &impl_hooks, l); + pw_loop_add_hook(pw_core_get_main_loop(core), &s->hook, &impl_hooks, s); - pw_log_info("protocol-native %p: Added listener %p", protocol, this); + pw_log_info("protocol-native %p: Added server %p", protocol, this); return this; error: - destroy_listener(this); + destroy_server(this); return NULL; } const static struct pw_protocol_implementaton protocol_impl = { PW_VERSION_PROTOCOL_IMPLEMENTATION, - impl_new_connection, - impl_add_listener, + .new_client = impl_new_client, + .add_server = impl_add_server, }; static struct spa_pod_builder * impl_ext_begin_proxy(struct pw_proxy *proxy, uint8_t opcode) { - struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this); return pw_protocol_native_connection_begin_proxy(impl->connection, proxy, opcode); } static uint32_t impl_ext_add_proxy_fd(struct pw_proxy *proxy, int fd) { - struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this); return pw_protocol_native_connection_add_fd(impl->connection, fd); } static int impl_ext_get_proxy_fd(struct pw_proxy *proxy, uint32_t index) { - struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this); return pw_protocol_native_connection_get_fd(impl->connection, index); } static void impl_ext_end_proxy(struct pw_proxy *proxy, struct spa_pod_builder *builder) { - struct connection *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct connection, this); + struct client *impl = SPA_CONTAINER_OF(proxy->remote->conn, struct client, this); pw_protocol_native_connection_end(impl->connection, builder); } @@ -778,6 +783,23 @@ const static struct pw_protocol_native_ext protocol_ext_impl = { impl_ext_end_resource, }; +static void module_destroy(void *data) +{ + struct protocol_data *d = data; + + spa_hook_remove(&d->module_listener); + + if (d->properties) + pw_properties_free(d->properties); + + pw_protocol_destroy(d->protocol); +} + +static const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; + static bool module_init(struct pw_module *module, struct pw_properties *properties) { struct pw_core *core = pw_module_get_core(module); @@ -800,29 +822,20 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti pw_log_debug("protocol-native %p: new", this); d = pw_protocol_get_user_data(this); + d->protocol = this; d->module = module; + d->properties = properties; if ((val = pw_properties_get(pw_core_get_properties(core), "pipewire.daemon"))) { if (atoi(val) == 1) - impl_add_listener(this, core, properties); + impl_add_server(this, core, properties); } + + pw_module_add_listener(module, &d->module_listener, &module_events, d); + return true; } -#if 0 -static void pw_protocol_native_destroy(struct impl *impl) -{ - struct impl *object, *tmp; - - pw_log_debug("protocol-native %p: destroy", impl); - - spa_list_for_each_safe(object, tmp, &impl->object_list, link) - object_destroy(object); - - free(impl); -} -#endif - bool pipewire__module_init(struct pw_module *module, const char *args) { return module_init(module, NULL); diff --git a/src/modules/module-suspend-on-idle.c b/src/modules/module-suspend-on-idle.c index 11e08ea46..6817a4a5e 100644 --- a/src/modules/module-suspend-on-idle.c +++ b/src/modules/module-suspend-on-idle.c @@ -33,6 +33,7 @@ struct impl { struct pw_type *t; struct pw_properties *properties; + struct spa_hook module_listener; struct spa_hook core_listener; struct spa_list node_list; @@ -127,7 +128,7 @@ core_global_added(void *data, struct pw_global *global) info = calloc(1, sizeof(struct node_info)); info->impl = impl; info->node = node; - spa_list_insert(impl->node_list.prev, &info->link); + spa_list_append(&impl->node_list, &info->link); pw_node_add_listener(node, &info->node_listener, &node_events, info); @@ -151,6 +152,28 @@ core_global_removed(void *data, struct pw_global *global) } } +static void module_destroy(void *data) +{ + struct impl *impl = data; + struct node_info *info, *t; + + spa_list_for_each_safe(info, t, &impl->node_list, link) + node_info_free(info); + + spa_hook_remove(&impl->core_listener); + spa_hook_remove(&impl->module_listener); + + if (impl->properties) + pw_properties_free(impl->properties); + + free(impl); +} + +const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; + const struct pw_core_events core_events = { PW_VERSION_CORE_EVENTS, .global_added = core_global_added, @@ -179,22 +202,12 @@ static bool module_init(struct pw_module *module, struct pw_properties *properti spa_list_init(&impl->node_list); + pw_module_add_listener(module, &impl->module_listener, &module_events, impl); pw_core_add_listener(impl->core, &impl->core_listener, &core_events, impl); return impl; } -#if 0 -static void module_destroy(struct impl *impl) -{ - pw_log_debug("module %p: destroy", impl); - - pw_global_destroy(impl->global); - - free(impl); -} -#endif - bool pipewire__module_init(struct pw_module *module, const char *args) { return module_init(module, NULL); diff --git a/src/modules/spa/module-monitor.c b/src/modules/spa/module-monitor.c index 403a0e888..720fd2dd0 100644 --- a/src/modules/spa/module-monitor.c +++ b/src/modules/spa/module-monitor.c @@ -34,11 +34,32 @@ #include "spa-monitor.h" +struct data { + struct pw_spa_monitor *monitor; + struct spa_hook module_listener; +}; + +static void module_destroy(void *data) +{ + struct data *d = data; + + spa_hook_remove(&d->module_listener); + + pw_spa_monitor_destroy(d->monitor); +} + +const struct pw_module_events module_events = { + PW_VERSION_MODULE_EVENTS, + .destroy = module_destroy, +}; + bool pipewire__module_init(struct pw_module *module, const char *args) { const char *dir; char **argv; int n_tokens; + struct pw_spa_monitor *monitor; + struct data *data; if (args == NULL) goto wrong_arguments; @@ -50,12 +71,18 @@ bool pipewire__module_init(struct pw_module *module, const char *args) if ((dir = getenv("SPA_PLUGIN_DIR")) == NULL) dir = PLUGINDIR; - pw_spa_monitor_load(pw_module_get_core(module), - pw_module_get_global(module), - dir, argv[0], argv[1], argv[2]); + monitor = pw_spa_monitor_load(pw_module_get_core(module), + pw_module_get_global(module), + dir, argv[0], argv[1], argv[2], + sizeof(struct data)); + + data = monitor->user_data; + data->monitor = monitor; pw_free_strv(argv); + pw_module_add_listener(module, &data->module_listener, &module_events, data); + return true; not_enough_arguments: diff --git a/src/modules/spa/module-node-factory.c b/src/modules/spa/module-node-factory.c index 9b9d2863c..e5037e71e 100644 --- a/src/modules/spa/module-node-factory.c +++ b/src/modules/spa/module-node-factory.c @@ -61,7 +61,7 @@ static struct pw_node *create_node(void *_data, lib, factory_name, name, - properties); + properties, 0); if (node == NULL) goto no_mem; diff --git a/src/modules/spa/module-node.c b/src/modules/spa/module-node.c index 653e87baa..cd3631ee4 100644 --- a/src/modules/spa/module-node.c +++ b/src/modules/spa/module-node.c @@ -61,7 +61,11 @@ bool pipewire__module_init(struct pw_module *module, const char *args) pw_free_strv(prop); } - pw_spa_node_load(pw_module_get_core(module), NULL, pw_module_get_global(module), argv[0], argv[1], argv[2], props); + pw_spa_node_load(pw_module_get_core(module), + NULL, + pw_module_get_global(module), + argv[0], argv[1], argv[2], + props, 0); pw_free_strv(argv); diff --git a/src/modules/spa/spa-monitor.c b/src/modules/spa/spa-monitor.c index 66b726e10..1c41f1f2b 100644 --- a/src/modules/spa/spa-monitor.c +++ b/src/modules/spa/spa-monitor.c @@ -42,6 +42,7 @@ struct monitor_item { char *id; struct spa_list link; struct pw_node *node; + struct spa_handle *handle; }; struct impl { @@ -120,10 +121,11 @@ static void add_item(struct pw_spa_monitor *this, struct spa_monitor_item *item) mitem = calloc(1, sizeof(struct monitor_item)); mitem->id = strdup(id); + mitem->handle = handle; mitem->node = pw_spa_node_new(impl->core, NULL, impl->parent, name, - false, node_iface, clock_iface, props); + false, node_iface, clock_iface, props, 0); - spa_list_insert(impl->item_list.prev, &mitem->link); + spa_list_append(&impl->item_list, &mitem->link); } static struct monitor_item *find_item(struct pw_spa_monitor *this, const char *id) @@ -143,6 +145,8 @@ void destroy_item(struct monitor_item *mitem) { pw_node_destroy(mitem->node); spa_list_remove(&mitem->link); + spa_handle_clear(mitem->handle); + free(mitem->handle); free(mitem->id); free(mitem); } @@ -222,7 +226,9 @@ struct pw_spa_monitor *pw_spa_monitor_load(struct pw_core *core, struct pw_global *parent, const char *dir, const char *lib, - const char *factory_name, const char *system_name) + const char *factory_name, + const char *system_name, + size_t user_data_size) { struct impl *impl; struct pw_spa_monitor *this; @@ -271,7 +277,7 @@ struct pw_spa_monitor *pw_spa_monitor_load(struct pw_core *core, goto interface_failed; } - impl = calloc(1, sizeof(struct impl)); + impl = calloc(1, sizeof(struct impl) + user_data_size); impl->core = core; impl->t = t; impl->parent = parent; @@ -284,6 +290,9 @@ struct pw_spa_monitor *pw_spa_monitor_load(struct pw_core *core, this->system_name = strdup(system_name); this->handle = handle; + if (user_data_size > 0) + this->user_data = SPA_MEMBER(impl, sizeof(struct impl), void); + update_monitor(core, this->system_name); spa_list_init(&impl->item_list); @@ -321,10 +330,10 @@ void pw_spa_monitor_destroy(struct pw_spa_monitor *monitor) struct impl *impl = SPA_CONTAINER_OF(monitor, struct impl, this); struct monitor_item *mitem, *tmp; - pw_log_debug("spa-monitor %p: dispose", impl); + pw_log_debug("spa-monitor %p: destroy", impl); spa_list_for_each_safe(mitem, tmp, &impl->item_list, link) - destroy_item(mitem); + destroy_item(mitem); spa_handle_clear(monitor->handle); free(monitor->handle); diff --git a/src/modules/spa/spa-monitor.h b/src/modules/spa/spa-monitor.h index a79941838..ad50cee16 100644 --- a/src/modules/spa/spa-monitor.h +++ b/src/modules/spa/spa-monitor.h @@ -35,6 +35,8 @@ struct pw_spa_monitor { char *factory_name; char *system_name; struct spa_handle *handle; + + void *user_data; }; struct pw_spa_monitor * @@ -42,7 +44,9 @@ pw_spa_monitor_load(struct pw_core *core, struct pw_global *parent, const char *dir, const char *lib, - const char *factory_name, const char *system_name); + const char *factory_name, + const char *system_name, + size_t user_data_size); void pw_spa_monitor_destroy(struct pw_spa_monitor *monitor); diff --git a/src/modules/spa/spa-node.c b/src/modules/spa/spa-node.c index 0c8c21add..416b34393 100644 --- a/src/modules/spa/spa-node.c +++ b/src/modules/spa/spa-node.c @@ -26,8 +26,12 @@ #include #include +#include #include "spa-node.h" +#include "pipewire/node.h" +#include "pipewire/port.h" +#include "pipewire/log.h" #include "pipewire/private.h" struct impl { @@ -170,13 +174,11 @@ static void update_port_ids(struct impl *impl) uint32_t *input_port_ids, *output_port_ids; uint32_t n_input_ports, n_output_ports, max_input_ports, max_output_ports; uint32_t i; - struct spa_list *ports; spa_node_get_n_ports(impl->node, &n_input_ports, &max_input_ports, &n_output_ports, &max_output_ports); - this->info.max_input_ports = max_input_ports; - this->info.max_output_ports = max_output_ports; + pw_node_set_max_ports(this, max_input_ports, max_output_ports); input_port_ids = alloca(sizeof(uint32_t) * n_input_ports); output_port_ids = alloca(sizeof(uint32_t) * n_output_ports); @@ -187,60 +189,14 @@ static void update_port_ids(struct impl *impl) pw_log_debug("node %p: update_port ids %u/%u, %u/%u", this, n_input_ports, max_input_ports, n_output_ports, max_output_ports); - i = 0; - ports = &this->input_ports; - while (true) { - struct pw_port *p = (ports == &this->input_ports) ? NULL : - SPA_CONTAINER_OF(ports, struct pw_port, link); - - if (p && i < n_input_ports && p->port_id == input_port_ids[i]) { - pw_log_debug("node %p: exiting input port %d", this, input_port_ids[i]); - i++; - ports = ports->next; - } else if ((p && i < n_input_ports && input_port_ids[i] < p->port_id) - || i < n_input_ports) { - struct pw_port *np; - pw_log_debug("node %p: input port added %d", this, input_port_ids[i]); - np = make_port(impl, PW_DIRECTION_INPUT, input_port_ids[i]); - - ports = np->link.next; - i++; - } else if (p) { - ports = ports->next; - pw_log_debug("node %p: input port removed %d", this, p->port_id); - pw_port_destroy(p); - } else { - pw_log_debug("node %p: no more input ports", this); - break; - } - } - - i = 0; - ports = &this->output_ports; - while (true) { - struct pw_port *p = (ports == &this->output_ports) ? NULL : - SPA_CONTAINER_OF(ports, struct pw_port, link); - - if (p && i < n_output_ports && p->port_id == output_port_ids[i]) { - pw_log_debug("node %p: exiting output port %d", this, output_port_ids[i]); - i++; - ports = ports->next; - } else if ((p && i < n_output_ports && output_port_ids[i] < p->port_id) - || i < n_output_ports) { - struct pw_port *np; - pw_log_debug("node %p: output port added %d", this, output_port_ids[i]); - np = make_port(impl, PW_DIRECTION_OUTPUT, output_port_ids[i]); - ports = np->link.next; - i++; - } else if (p) { - ports = ports->next; - pw_log_debug("node %p: output port removed %d", this, p->port_id); - pw_port_destroy(p); - } else { - pw_log_debug("node %p: no more output ports", this); - break; - } - } + for (i = 0; i < n_input_ports; i++) { + pw_log_debug("node %p: input port added %d", this, input_port_ids[i]); + make_port(impl, PW_DIRECTION_INPUT, input_port_ids[i]); + } + for (i = 0; i < n_output_ports; i++) { + pw_log_debug("node %p: output port added %d", this, output_port_ids[i]); + make_port(impl, PW_DIRECTION_OUTPUT, output_port_ids[i]); + } } @@ -335,7 +291,7 @@ static void on_node_done(void *data, int seq, int res) } pw_log_debug("spa-node %p: async complete event %d %d", this, seq, res); - spa_hook_list_call(&this->listener_list, struct pw_node_events, async_complete, seq, res); + spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, async_complete, seq, res); } static void on_node_event(void *data, struct spa_event *event) @@ -343,21 +299,21 @@ static void on_node_event(void *data, struct spa_event *event) struct impl *impl = data; struct pw_node *this = impl->this; - spa_hook_list_call(&this->listener_list, struct pw_node_events, event, event); + spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, event, event); } static void on_node_need_input(void *data) { struct impl *impl = data; struct pw_node *this = impl->this; - spa_hook_list_call(&this->listener_list, struct pw_node_events, need_input); + spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, need_input); } static void on_node_have_output(void *data) { struct impl *impl = data; struct pw_node *this = impl->this; - spa_hook_list_call(&this->listener_list, struct pw_node_events, have_output); + spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, have_output); } static void @@ -365,18 +321,8 @@ on_node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id) { struct impl *impl = data; struct pw_node *this = impl->this; - struct spa_graph_port *p, *pp; - - spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_INPUT], link) { - if (p->port_id != port_id) - continue; - - pp = p->peer; - if (pp && pp->callbacks->reuse_buffer) - pp->callbacks->reuse_buffer(pp->callbacks_data, buffer_id); - - break; - } + spa_hook_list_call(pw_node_get_listeners(this), struct pw_node_events, + reuse_buffer, port_id, buffer_id); } static const struct spa_node_callbacks spa_node_callbacks = { @@ -401,7 +347,8 @@ pw_spa_node_new(struct pw_core *core, bool async, struct spa_node *node, struct spa_clock *clock, - struct pw_properties *properties) + struct pw_properties *properties, + size_t user_data_size) { struct pw_node *this; struct impl *impl; @@ -411,8 +358,7 @@ pw_spa_node_new(struct pw_core *core, if (properties == NULL) properties = pw_properties_new(NULL, NULL); - - if (properties) + if (properties == NULL) return NULL; for (i = 0; i < node->info->n_items; i++) @@ -421,7 +367,7 @@ pw_spa_node_new(struct pw_core *core, node->info->items[i].value); } - this = pw_node_new(core, owner, parent, name, properties, sizeof(struct impl)); + this = pw_node_new(core, owner, parent, name, properties, sizeof(struct impl) + user_data_size); if (this == NULL) return NULL; @@ -439,9 +385,8 @@ pw_spa_node_new(struct pw_core *core, if (spa_node_set_callbacks(impl->node, &spa_node_callbacks, impl) < 0) pw_log_warn("spa-node %p: error setting callback", this); - if (!async) { + if (!async) complete_init(impl); - } return this; } @@ -453,6 +398,7 @@ setup_props(struct pw_core *core, struct spa_node *spa_node, struct pw_propertie struct spa_props *props; void *state = NULL; const char *key; + struct pw_type *t = pw_core_get_type(core); if ((res = spa_node_get_props(spa_node, &props)) != SPA_RESULT_OK) { pw_log_debug("spa_node_get_props failed: %d", res); @@ -466,7 +412,7 @@ setup_props(struct pw_core *core, struct spa_node *spa_node, struct pw_propertie if (!spa_type_is_a(key, SPA_TYPE_PROPS_BASE)) continue; - id = spa_type_map_get_id(core->type.map, key); + id = spa_type_map_get_id(t->map, key); if (id == SPA_ID_INVALID) continue; @@ -478,7 +424,7 @@ setup_props(struct pw_core *core, struct spa_node *spa_node, struct pw_propertie switch(prop->body.value.type) { case SPA_POD_TYPE_ID: SPA_POD_VALUE(struct spa_pod_id, &prop->body.value) = - spa_type_map_get_id(core->type.map, value); + spa_type_map_get_id(t->map, value); break; case SPA_POD_TYPE_INT: SPA_POD_VALUE(struct spa_pod_int, &prop->body.value) = atoi(value); @@ -514,7 +460,8 @@ struct pw_node *pw_spa_node_load(struct pw_core *core, const char *lib, const char *factory_name, const char *name, - struct pw_properties *properties) + struct pw_properties *properties, + size_t user_data_size) { struct pw_node *this; struct impl *impl; @@ -530,6 +477,9 @@ struct pw_node *pw_spa_node_load(struct pw_core *core, char *filename; const char *dir; bool async; + const struct spa_support *support; + uint32_t n_support; + struct pw_type *t = pw_core_get_type(core); if ((dir = getenv("SPA_PLUGIN_DIR")) == NULL) dir = PLUGINDIR; @@ -555,21 +505,23 @@ struct pw_node *pw_spa_node_load(struct pw_core *core, break; } + support = pw_core_get_support(core, &n_support); + handle = calloc(1, factory->size); if ((res = spa_handle_factory_init(factory, - handle, NULL, core->support, core->n_support)) < 0) { + handle, NULL, support, n_support)) < 0) { pw_log_error("can't make factory instance: %d", res); goto init_failed; } async = SPA_RESULT_IS_ASYNC(res); - if ((res = spa_handle_get_interface(handle, core->type.spa_node, &iface)) < 0) { + if ((res = spa_handle_get_interface(handle, t->spa_node, &iface)) < 0) { pw_log_error("can't get node interface %d", res); goto interface_failed; } spa_node = iface; - if ((res = spa_handle_get_interface(handle, core->type.spa_clock, &iface)) < 0) { + if ((res = spa_handle_get_interface(handle, t->spa_clock, &iface)) < 0) { iface = NULL; } spa_clock = iface; @@ -580,7 +532,10 @@ struct pw_node *pw_spa_node_load(struct pw_core *core, } } - this = pw_spa_node_new(core, owner, parent, name, async, spa_node, spa_clock, properties); + this = pw_spa_node_new(core, owner, parent, name, async, + spa_node, spa_clock, properties, user_data_size); + + impl = this->user_data; impl->hnd = hnd; impl->handle = handle; impl->lib = filename; diff --git a/src/modules/spa/spa-node.h b/src/modules/spa/spa-node.h index c21c487fa..787edaca1 100644 --- a/src/modules/spa/spa-node.h +++ b/src/modules/spa/spa-node.h @@ -38,7 +38,8 @@ pw_spa_node_new(struct pw_core *core, bool async, struct spa_node *node, struct spa_clock *clock, - struct pw_properties *properties); + struct pw_properties *properties, + size_t user_data_size); struct pw_node * pw_spa_node_load(struct pw_core *core, @@ -47,7 +48,8 @@ pw_spa_node_load(struct pw_core *core, const char *lib, const char *factory_name, const char *name, - struct pw_properties *properties); + struct pw_properties *properties, + size_t user_data_size); #ifdef __cplusplus } diff --git a/src/pipewire/core.c b/src/pipewire/core.c index 6d1aefe66..cbd5cda06 100644 --- a/src/pipewire/core.c +++ b/src/pipewire/core.c @@ -403,10 +403,14 @@ struct pw_core *pw_core_new(struct pw_loop *main_loop, struct pw_properties *pro void pw_core_destroy(struct pw_core *core) { struct pw_global *global, *t; + struct pw_module *module, *tm; pw_log_debug("core %p: destroy", core); spa_hook_list_call(&core->listener_list, struct pw_core_events, destroy); + spa_list_for_each_safe(module, tm, &core->module_list, link) + pw_module_destroy(module); + spa_list_for_each_safe(global, t, &core->global_list, link) pw_global_destroy(global); diff --git a/src/pipewire/global.h b/src/pipewire/global.h index 71057d100..847c7aa8e 100644 --- a/src/pipewire/global.h +++ b/src/pipewire/global.h @@ -89,8 +89,7 @@ pw_global_bind(struct pw_global *global, uint32_t version, uint32_t id); -void -pw_global_destroy(struct pw_global *global); +void pw_global_destroy(struct pw_global *global); #ifdef __cplusplus } diff --git a/src/pipewire/link.c b/src/pipewire/link.c index 85eb5e126..60127a1a4 100644 --- a/src/pipewire/link.c +++ b/src/pipewire/link.c @@ -705,6 +705,15 @@ static int do_start(struct pw_link *this, uint32_t in_state, uint32_t out_state) return res; } +static int +do_activate_link(struct spa_loop *loop, + bool async, uint32_t seq, size_t size, const void *data, void *user_data) +{ + struct pw_link *this = user_data; + spa_graph_port_link(&this->rt.out_port, &this->rt.in_port); + return SPA_RESULT_OK; +} + static int check_states(struct pw_link *this, void *user_data, int res) { struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this); @@ -726,6 +735,8 @@ static int check_states(struct pw_link *this, void *user_data, int res) pw_log_debug("link %p: input state %d, output state %d", this, in_state, out_state); if (in_state == PW_PORT_STATE_STREAMING && out_state == PW_PORT_STATE_STREAMING) { + pw_loop_invoke(this->output->node->data_loop, + do_activate_link, SPA_ID_INVALID, 0, NULL, false, this); pw_link_update_state(this, PW_LINK_STATE_RUNNING, NULL); return SPA_RESULT_OK; } @@ -865,15 +876,6 @@ static void output_port_destroy(void *data) on_port_destroy(&impl->this, impl->this.output); } -static int -do_activate_link(struct spa_loop *loop, - bool async, uint32_t seq, size_t size, const void *data, void *user_data) -{ - struct pw_link *this = user_data; - spa_graph_port_link(&this->rt.out_port, &this->rt.in_port); - return SPA_RESULT_OK; -} - bool pw_link_activate(struct pw_link *this) { struct impl *impl = SPA_CONTAINER_OF(this, struct impl, this); @@ -884,9 +886,6 @@ bool pw_link_activate(struct pw_link *this) impl->active = true; pw_log_debug("link %p: activate", this); - pw_loop_invoke(this->output->node->data_loop, - do_activate_link, SPA_ID_INVALID, 0, NULL, false, this); - this->output->node->n_used_output_links++; this->input->node->n_used_input_links++; @@ -1170,6 +1169,9 @@ void pw_link_destroy(struct pw_link *link) pw_work_queue_destroy(impl->work); + if (link->properties) + pw_properties_free(link->properties); + if (link->info.format) free(link->info.format); diff --git a/src/pipewire/link.h b/src/pipewire/link.h index 5b4edd901..9aede85bc 100644 --- a/src/pipewire/link.h +++ b/src/pipewire/link.h @@ -103,6 +103,7 @@ struct pw_port *pw_link_get_input(struct pw_link *link); /** Find the link between 2 ports \memberof pw_link */ struct pw_link *pw_link_find(struct pw_port *output, struct pw_port *input); +/** Indicate that linked nodes can be idle even when linked with this link */ void pw_link_inc_idle(struct pw_link *link); /** Activate a link \memberof pw_link diff --git a/src/pipewire/module.c b/src/pipewire/module.c index e16f86285..6e0c35932 100644 --- a/src/pipewire/module.c +++ b/src/pipewire/module.c @@ -257,6 +257,7 @@ void pw_module_destroy(struct pw_module *module) struct impl *impl = SPA_CONTAINER_OF(module, struct impl, this); struct pw_resource *resource, *tmp; + pw_log_debug("module %p: destroy", module); spa_hook_list_call(&module->listener_list, struct pw_module_events, destroy); spa_list_for_each_safe(resource, tmp, &module->resource_list, link) diff --git a/src/pipewire/node-factory.c b/src/pipewire/node-factory.c index 9b69d3747..0169ea287 100644 --- a/src/pipewire/node-factory.c +++ b/src/pipewire/node-factory.c @@ -34,15 +34,31 @@ struct pw_node_factory *pw_node_factory_new(struct pw_core *core, if (user_data_size > 0) this->user_data = SPA_MEMBER(this, sizeof(*this), void); + pw_log_debug("node-factory %p: new", this); + return this; } +void pw_node_factory_destroy(struct pw_node_factory *factory) +{ + pw_log_debug("node-factory %p: destroy", factory); + + if (factory->global) { + spa_list_remove(&factory->link); + pw_global_destroy(factory->global); + } + if (factory->name) + free((char *)factory->name); + + free(factory); +} + void pw_node_factory_export(struct pw_node_factory *factory, struct pw_client *owner, struct pw_global *parent) { struct pw_core *core = factory->core; - spa_list_insert(core->node_factory_list.prev, &factory->link); + spa_list_append(&core->node_factory_list, &factory->link); factory->global = pw_core_add_global(core, owner, parent, core->type.node_factory, 0, NULL, factory); } diff --git a/src/pipewire/node-factory.h b/src/pipewire/node-factory.h index a01e10339..0dc702043 100644 --- a/src/pipewire/node-factory.h +++ b/src/pipewire/node-factory.h @@ -60,6 +60,8 @@ void pw_node_factory_export(struct pw_node_factory *factory, struct pw_client *owner, struct pw_global *parent); +void pw_node_factory_destroy(struct pw_node_factory *factory); + void *pw_node_factory_get_user_data(struct pw_node_factory *factory); void pw_node_factory_set_implementation(struct pw_node_factory *factory, diff --git a/src/pipewire/node.c b/src/pipewire/node.c index 5b229142c..ffff1d896 100644 --- a/src/pipewire/node.c +++ b/src/pipewire/node.c @@ -180,6 +180,23 @@ static void node_have_output(void *data) spa_graph_have_output(this->rt.graph, &this->rt.node); } +static void node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id) +{ + struct impl *impl = data; + struct pw_node *this = &impl->this; + struct spa_graph_port *p, *pp; + + spa_list_for_each(p, &this->rt.node.ports[SPA_DIRECTION_INPUT], link) { + if (p->port_id != port_id) + continue; + + pp = p->peer; + if (pp && pp->callbacks->reuse_buffer) + pp->callbacks->reuse_buffer(pp->callbacks_data, buffer_id); + break; + } +} + static void node_unbind_func(void *data) { struct pw_resource *resource = data; @@ -190,7 +207,6 @@ static void update_info(struct pw_node *this) { this->info.input_formats = NULL; - if (!spa_list_is_empty(&this->input_ports)) { struct pw_port *port = spa_list_first(&this->input_ports, struct pw_port, link); @@ -356,6 +372,7 @@ static const struct pw_node_events node_events = { .event = node_event, .need_input = node_need_input, .have_output = node_have_output, + .reuse_buffer = node_reuse_buffer, }; struct pw_node *pw_node_new(struct pw_core *core, @@ -485,6 +502,11 @@ void pw_node_add_listener(struct pw_node *node, spa_hook_list_append(&node->listener_list, listener, events, data); } +struct spa_hook_list *pw_node_get_listeners(struct pw_node *node) +{ + return &node->listener_list; +} + static int do_node_remove(struct spa_loop *loop, bool async, uint32_t seq, size_t size, const void *data, void *user_data) @@ -553,6 +575,20 @@ void pw_node_destroy(struct pw_node *node) free(impl); } +void pw_node_set_max_ports(struct pw_node *node, + uint32_t max_input_ports, + uint32_t max_output_ports) +{ + if (node->info.max_input_ports != max_input_ports) { + node->info.max_input_ports = max_input_ports; + node->info.change_mask |= PW_NODE_CHANGE_MASK_INPUT_PORTS; + } + if (node->info.max_output_ports != max_output_ports) { + node->info.max_output_ports = max_output_ports; + node->info.change_mask |= PW_NODE_CHANGE_MASK_OUTPUT_PORTS; + } +} + bool pw_node_for_each_port(struct pw_node *node, enum pw_direction direction, bool (*callback) (void *data, struct pw_port *port), diff --git a/src/pipewire/node.h b/src/pipewire/node.h index a0741776e..9c2d2b4d5 100644 --- a/src/pipewire/node.h +++ b/src/pipewire/node.h @@ -102,6 +102,8 @@ struct pw_node_events { void (*need_input) (void *data); /** the node has output */ void (*have_output) (void *data); + /** the node has a buffer to reuse */ + void (*reuse_buffer) (void *data, uint32_t port_id, uint32_t buffer_id); }; /** Create a new node \memberof pw_node */ @@ -119,6 +121,10 @@ void pw_node_register(struct pw_node *node); /** Destroy a node */ void pw_node_destroy(struct pw_node *node); +void pw_node_set_max_ports(struct pw_node *node, + uint32_t max_input_ports, + uint32_t max_output_ports); + const struct pw_node_info *pw_node_get_info(struct pw_node *node); void * pw_node_get_user_data(struct pw_node *node); @@ -142,6 +148,8 @@ void pw_node_add_listener(struct pw_node *node, const struct pw_node_events *events, void *data); +struct spa_hook_list *pw_node_get_listeners(struct pw_node *node); + /** iterate the ports in the given direction */ bool pw_node_for_each_port(struct pw_node *node, enum pw_direction direction, diff --git a/src/pipewire/port.c b/src/pipewire/port.c index 9f433a531..8c5ee8540 100644 --- a/src/pipewire/port.c +++ b/src/pipewire/port.c @@ -340,6 +340,15 @@ void pw_port_destroy(struct pw_port *port) pw_log_debug("port %p: free", port); spa_hook_list_call(&port->listener_list, struct pw_port_events, free); + + if (port->buffers) + free(port->buffers); + if (port->allocated) + pw_memblock_free(&port->buffer_mem); + + if (port->properties) + pw_properties_free(port->properties); + free(port); } diff --git a/src/pipewire/private.h b/src/pipewire/private.h index 355d3455b..53990ca2b 100644 --- a/src/pipewire/private.h +++ b/src/pipewire/private.h @@ -37,6 +37,24 @@ struct pw_command { const char *name; /**< command name */ }; +struct pw_protocol { + struct spa_list link; /**< link in core protocol_list */ + struct pw_core *core; /**< core for this protocol */ + + char *name; /**< type name of the protocol */ + + struct spa_list marshal_list; /**< list of marshallers for supported interfaces */ + struct spa_list client_list; /**< list of current clients */ + struct spa_list server_list; /**< list of current servers */ + struct spa_hook_list listener_list; /**< event listeners */ + + const struct pw_protocol_implementaton *implementation; /**< implementation of the protocol */ + + const void *extension; /**< extension API */ + + void *user_data; /**< user data for the implementation */ +}; + struct pw_client { struct pw_core *core; /**< core object */ struct spa_list link; /**< link in core object client list */ @@ -316,7 +334,7 @@ struct pw_remote { struct spa_list stream_list; /**< list of \ref pw_stream objects */ struct spa_list remote_node_list; /**< list of \ref pw_remote_node objects */ - struct pw_protocol_connection *conn; /**< the protocol connection */ + struct pw_protocol_client *conn; /**< the protocol client connection */ enum pw_remote_state state; char *error; diff --git a/src/pipewire/protocol.c b/src/pipewire/protocol.c index 146ce2538..0a98a6bf5 100644 --- a/src/pipewire/protocol.c +++ b/src/pipewire/protocol.c @@ -41,8 +41,9 @@ struct pw_protocol *pw_protocol_new(struct pw_core *core, protocol->name = strdup(name); spa_list_init(&protocol->marshal_list); - spa_list_init(&protocol->connection_list); - spa_list_init(&protocol->listener_list); + spa_list_init(&protocol->server_list); + spa_list_init(&protocol->client_list); + spa_hook_list_init(&protocol->listener_list); if (user_data_size > 0) protocol->user_data = SPA_MEMBER(protocol, sizeof(struct impl), void); @@ -59,34 +60,52 @@ void *pw_protocol_get_user_data(struct pw_protocol *protocol) return protocol->user_data; } +const struct pw_protocol_implementaton * +pw_protocol_get_implementation(struct pw_protocol *protocol) +{ + return protocol->implementation; +} + +const void * +pw_protocol_get_extension(struct pw_protocol *protocol) +{ + return protocol->extension; +} + void pw_protocol_destroy(struct pw_protocol *protocol) { struct impl *impl = SPA_CONTAINER_OF(protocol, struct impl, this); struct marshal *marshal, *t1; - struct pw_protocol_listener *listener, *t2; - struct pw_protocol_connection *connection, *t3; + struct pw_protocol_server *server, *t2; + struct pw_protocol_client *client, *t3; pw_log_info("protocol %p: destroy", protocol); + spa_hook_list_call(&protocol->listener_list, struct pw_protocol_events, destroy); spa_list_remove(&protocol->link); spa_list_for_each_safe(marshal, t1, &protocol->marshal_list, link) free(marshal); - spa_list_for_each_safe(listener, t2, &protocol->listener_list, link) - pw_protocol_listener_destroy(listener); + spa_list_for_each_safe(server, t2, &protocol->server_list, link) + pw_protocol_server_destroy(server); - spa_list_for_each_safe(connection, t3, &protocol->connection_list, link) - pw_protocol_connection_destroy(connection); + spa_list_for_each_safe(client, t3, &protocol->client_list, link) + pw_protocol_client_destroy(client); free(protocol->name); - if (protocol->destroy) - protocol->destroy(protocol); - free(impl); } +void pw_protocol_add_listener(struct pw_protocol *protocol, + struct spa_hook *listener, + const struct pw_protocol_events *events, + void *data) +{ + spa_hook_list_append(&protocol->listener_list, listener, events, data); +} + void pw_protocol_add_marshal(struct pw_protocol *protocol, const struct pw_protocol_marshal *marshal) diff --git a/src/pipewire/protocol.h b/src/pipewire/protocol.h index d61725eff..268b80a27 100644 --- a/src/pipewire/protocol.h +++ b/src/pipewire/protocol.h @@ -26,6 +26,8 @@ extern "C" { #include +struct pw_protocol; + #include #include #include @@ -33,33 +35,33 @@ extern "C" { #define PW_TYPE__Protocol "PipeWire:Protocol" #define PW_TYPE_PROTOCOL_BASE PW_TYPE__Protocol ":" -struct pw_protocol_connection { - struct spa_list link; /**< link in protocol connection_list */ +struct pw_protocol_client { + struct spa_list link; /**< link in protocol client_list */ struct pw_protocol *protocol; /**< the owner protocol */ struct pw_remote *remote; /**< the associated remote */ - int (*connect) (struct pw_protocol_connection *conn); - int (*connect_fd) (struct pw_protocol_connection *conn, int fd); - void (*disconnect) (struct pw_protocol_connection *conn); - void (*destroy) (struct pw_protocol_connection *conn); + int (*connect) (struct pw_protocol_client *client); + int (*connect_fd) (struct pw_protocol_client *client, int fd); + void (*disconnect) (struct pw_protocol_client *client); + void (*destroy) (struct pw_protocol_client *client); }; -#define pw_protocol_connection_connect(c) ((c)->connect(c)) -#define pw_protocol_connection_connect_fd(c,fd) ((c)->connect_fd(c,fd)) -#define pw_protocol_connection_disconnect(c) ((c)->disconnect(c)) -#define pw_protocol_connection_destroy(c) ((c)->destroy(c)) +#define pw_protocol_client_connect(c) ((c)->connect(c)) +#define pw_protocol_client_connect_fd(c,fd) ((c)->connect_fd(c,fd)) +#define pw_protocol_client_disconnect(c) ((c)->disconnect(c)) +#define pw_protocol_client_destroy(c) ((c)->destroy(c)) -struct pw_protocol_listener { - struct spa_list link; /**< link in protocol listener_list */ +struct pw_protocol_server { + struct spa_list link; /**< link in protocol server_list */ struct pw_protocol *protocol; /**< the owner protocol */ - struct spa_list client_list; /**< list of client of this protocol */ + struct spa_list client_list; /**< list of clients of this protocol */ - void (*destroy) (struct pw_protocol_listener *listen); + void (*destroy) (struct pw_protocol_server *listen); }; -#define pw_protocol_listener_destroy(l) ((l)->destroy(l)) +#define pw_protocol_server_destroy(l) ((l)->destroy(l)) struct pw_protocol_marshal { const char *type; /**< interface type */ @@ -75,41 +77,44 @@ struct pw_protocol_marshal { struct pw_protocol_implementaton { #define PW_VERSION_PROTOCOL_IMPLEMENTATION 0 uint32_t version; - struct pw_protocol_connection * (*new_connection) (struct pw_protocol *protocol, - struct pw_remote *remote, - struct pw_properties *properties); - struct pw_protocol_listener * (*add_listener) (struct pw_protocol *protocol, - struct pw_core *core, - struct pw_properties *properties); + + struct pw_protocol_client * (*new_client) (struct pw_protocol *protocol, + struct pw_remote *remote, + struct pw_properties *properties); + struct pw_protocol_server * (*add_server) (struct pw_protocol *protocol, + struct pw_core *core, + struct pw_properties *properties); }; -struct pw_protocol { - struct spa_list link; /**< link in core protocol_list */ - struct pw_core *core; /**< core for this protocol */ +struct pw_protocol_events { +#define PW_VERSION_PROTOCOL_EVENTS 0 + uint32_t version; - char *name; /**< type name of the protocol */ - - struct spa_list marshal_list; /**< list of marshallers for supported interfaces */ - struct spa_list connection_list; /**< list of current connections */ - struct spa_list listener_list; /**< list of current listeners */ - - const struct pw_protocol_implementaton *implementation; /**< implementation of the protocol */ - - const void *extension; /**< extension API */ - - void *user_data; /**< user data for the implementation */ - pw_destroy_t destroy; /**< function to clean up the object */ + void (*destroy) (void *data); }; -#define pw_protocol_new_connection(p,...) ((p)->implementation->new_connection(p,__VA_ARGS__)) -#define pw_protocol_add_listener(p,...) ((p)->implementation->add_listener(p,__VA_ARGS__)) - -#define pw_protocol_ext(p,type,method,...) (((type*)(p)->extension)->method( __VA_ARGS__)) +#define pw_protocol_new_client(p,...) (pw_protocol_get_implementation(p)->new_client(p,__VA_ARGS__)) +#define pw_protocol_add_server(p,...) (pw_protocol_get_implementation(p)->add_server(p,__VA_ARGS__)) +#define pw_protocol_ext(p,type,method,...) (((type*)pw_protocol_get_extension(p))->method( __VA_ARGS__)) struct pw_protocol *pw_protocol_new(struct pw_core *core, const char *name, size_t user_data_size); +void pw_protocol_destroy(struct pw_protocol *protocol); + void *pw_protocol_get_user_data(struct pw_protocol *protocol); +const struct pw_protocol_implementaton * +pw_protocol_get_implementation(struct pw_protocol *protocol); + +const void * +pw_protocol_get_extension(struct pw_protocol *protocol); + + +void pw_protocol_add_listener(struct pw_protocol *protocol, + struct spa_hook *listener, + const struct pw_protocol_events *events, + void *data); + /** \class pw_protocol * * \brief Manages protocols and their implementation diff --git a/src/pipewire/remote.c b/src/pipewire/remote.c index d7f556476..34f933f3d 100644 --- a/src/pipewire/remote.c +++ b/src/pipewire/remote.c @@ -241,7 +241,7 @@ struct pw_remote *pw_remote_new(struct pw_core *core, if (protocol == NULL) goto no_protocol; - this->conn = pw_protocol_new_connection(protocol, this, properties); + this->conn = pw_protocol_new_client(protocol, this, properties); if (this->conn == NULL) goto no_connection; @@ -273,7 +273,7 @@ void pw_remote_destroy(struct pw_remote *remote) spa_list_for_each_safe(stream, s2, &remote->stream_list, link) pw_stream_destroy(stream); - pw_protocol_connection_destroy (remote->conn); + pw_protocol_client_destroy (remote->conn); spa_list_remove(&remote->link); @@ -322,7 +322,7 @@ static int do_connect(struct pw_remote *remote) return 0; no_proxy: - pw_protocol_connection_disconnect (remote->conn); + pw_protocol_client_disconnect (remote->conn); pw_remote_update_state(remote, PW_REMOTE_STATE_ERROR, "can't connect: no memory"); return -1; } @@ -348,7 +348,7 @@ int pw_remote_connect(struct pw_remote *remote) pw_remote_update_state(remote, PW_REMOTE_STATE_CONNECTING, NULL); - if ((res = pw_protocol_connection_connect (remote->conn)) < 0) { + if ((res = pw_protocol_client_connect (remote->conn)) < 0) { pw_remote_update_state(remote, PW_REMOTE_STATE_ERROR, "connect failed"); return res; } @@ -362,7 +362,7 @@ int pw_remote_connect_fd(struct pw_remote *remote, int fd) pw_remote_update_state(remote, PW_REMOTE_STATE_CONNECTING, NULL); - if ((res = pw_protocol_connection_connect_fd (remote->conn, fd)) < 0) { + if ((res = pw_protocol_client_connect_fd (remote->conn, fd)) < 0) { pw_remote_update_state(remote, PW_REMOTE_STATE_ERROR, "connect_fd failed"); return res; } @@ -379,7 +379,7 @@ void pw_remote_disconnect(struct pw_remote *remote) spa_list_for_each_safe(stream, s2, &remote->stream_list, link) pw_stream_disconnect(stream); - pw_protocol_connection_disconnect (remote->conn); + pw_protocol_client_disconnect (remote->conn); spa_list_for_each_safe(proxy, t2, &remote->proxy_list, link) pw_proxy_destroy(proxy);