Adding upstream version 0.3.65.upstream/0.3.65upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1339 insertions, 0 deletions

diff --git a/src/modules/module-client-node/remote-node.c b/src/modules/module-client-node/remote-node.c
new file mode 100644
index 0000000..051ab07
--- /dev/null
+++ b/src/modules/module-client-node/remote-node.c
@@ -0,0 +1,1339 @@
+/* PipeWire
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <errno.h>
+#include <time.h>
+#include <sys/mman.h>
+
+#include <spa/pod/parser.h>
+#include <spa/pod/dynamic.h>
+#include <spa/node/utils.h>
+#include <spa/utils/result.h>
+#include <spa/debug/types.h>
+
+#include "pipewire/pipewire.h"
+#include "pipewire/private.h"
+
+#include "pipewire/extensions/protocol-native.h"
+#include "pipewire/extensions/client-node.h"
+
+#define MAX_BUFFERS	64
+
+PW_LOG_TOPIC_EXTERN(mod_topic);
+#define PW_LOG_TOPIC_DEFAULT mod_topic
+
+/** \cond */
+static bool mlock_warned = false;
+
+struct buffer {
+	uint32_t id;
+	struct spa_buffer *buf;
+	struct pw_memmap *mem;
+};
+
+struct mix {
+	struct spa_list link;
+	struct pw_impl_port *port;
+	uint32_t mix_id;
+	struct pw_impl_port_mix mix;
+	struct pw_array buffers;
+	bool active;
+};
+
+struct node_data {
+	struct pw_context *context;
+
+	struct pw_mempool *pool;
+
+	uint32_t remote_id;
+	int rtwritefd;
+	struct pw_memmap *activation;
+
+	struct spa_list mix[2];
+	struct spa_list free_mix;
+
+	struct pw_impl_node *node;
+	struct spa_hook node_listener;
+	unsigned int do_free:1;
+	unsigned int have_transport:1;
+	unsigned int allow_mlock:1;
+	unsigned int warn_mlock:1;
+
+	struct pw_client_node *client_node;
+	struct spa_hook client_node_listener;
+	struct spa_hook proxy_client_node_listener;
+
+	struct spa_list links;
+};
+
+struct link {
+	struct spa_list link;
+	struct node_data *data;
+	struct pw_memmap *map;
+	struct pw_node_target target;
+	uint32_t node_id;
+	int signalfd;
+};
+
+/** \endcond */
+
+static struct link *find_activation(struct spa_list *links, uint32_t node_id)
+{
+	struct link *l;
+
+	spa_list_for_each(l, links, link) {
+		if (l->node_id == node_id)
+			return l;
+	}
+	return NULL;
+}
+
+static int
+do_deactivate_link(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct link *link = user_data;
+	pw_log_trace("link %p deactivate", link);
+	spa_list_remove(&link->target.link);
+	return 0;
+}
+
+static void clear_link(struct node_data *data, struct link *link)
+{
+	struct pw_context *context = data->context;
+	pw_log_debug("link %p", link);
+	pw_loop_invoke(context->data_loop,
+		do_deactivate_link, SPA_ID_INVALID, NULL, 0, true, link);
+	pw_memmap_free(link->map);
+	spa_system_close(context->data_system, link->signalfd);
+	spa_list_remove(&link->link);
+	free(link);
+}
+
+static void clean_transport(struct node_data *data)
+{
+	struct link *l;
+	uint32_t tag[5] = { data->remote_id, };
+	struct pw_memmap *mm;
+
+	if (!data->have_transport)
+		return;
+
+	spa_list_consume(l, &data->links, link)
+		clear_link(data, l);
+
+	while ((mm = pw_mempool_find_tag(data->pool, tag, sizeof(uint32_t))) != NULL) {
+		if (mm->tag[1] == SPA_ID_INVALID)
+			spa_node_set_io(data->node->node, mm->tag[2], NULL, 0);
+
+		pw_memmap_free(mm);
+	}
+
+	pw_memmap_free(data->activation);
+	data->node->rt.activation = data->node->activation->map->ptr;
+
+	spa_system_close(data->context->data_system, data->rtwritefd);
+	data->have_transport = false;
+}
+
+static void mix_init(struct mix *mix, struct pw_impl_port *port, uint32_t mix_id)
+{
+	pw_log_debug("port %p: mix init %d.%d", port, port->port_id, mix_id);
+	mix->port = port;
+	mix->mix_id = mix_id;
+	pw_impl_port_init_mix(port, &mix->mix);
+	mix->active = false;
+	pw_array_init(&mix->buffers, 32);
+	pw_array_ensure_size(&mix->buffers, sizeof(struct buffer) * 64);
+}
+
+static int
+do_deactivate_mix(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct mix *mix = user_data;
+	spa_list_remove(&mix->mix.rt_link);
+        return 0;
+}
+
+static int
+deactivate_mix(struct node_data *data, struct mix *mix)
+{
+	if (mix->active) {
+		pw_log_debug("node %p: mix %p deactivate", data, mix);
+		pw_loop_invoke(data->context->data_loop,
+                       do_deactivate_mix, SPA_ID_INVALID, NULL, 0, true, mix);
+		mix->active = false;
+	}
+	return 0;
+}
+
+static int
+do_activate_mix(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct mix *mix = user_data;
+
+	spa_list_append(&mix->port->rt.mix_list, &mix->mix.rt_link);
+        return 0;
+}
+
+static int
+activate_mix(struct node_data *data, struct mix *mix)
+{
+	if (!mix->active) {
+		pw_log_debug("node %p: mix %p activate", data, mix);
+		pw_loop_invoke(data->context->data_loop,
+                       do_activate_mix, SPA_ID_INVALID, NULL, 0, false, mix);
+		mix->active = true;
+	}
+	return 0;
+}
+
+static struct mix *find_mix(struct node_data *data,
+		enum spa_direction direction, uint32_t port_id, uint32_t mix_id)
+{
+	struct mix *mix;
+
+	spa_list_for_each(mix, &data->mix[direction], link) {
+		if (mix->port->port_id == port_id &&
+		    mix->mix_id == mix_id) {
+			pw_log_debug("port %p: found mix %d:%d.%d", mix->port,
+					direction, port_id, mix_id);
+			return mix;
+		}
+	}
+	return NULL;
+}
+
+static struct mix *ensure_mix(struct node_data *data,
+		enum spa_direction direction, uint32_t port_id, uint32_t mix_id)
+{
+	struct mix *mix;
+	struct pw_impl_port *port;
+
+	if ((mix = find_mix(data, direction, port_id, mix_id)))
+		return mix;
+
+	port = pw_impl_node_find_port(data->node, direction, port_id);
+	if (port == NULL)
+		return NULL;
+
+	if (spa_list_is_empty(&data->free_mix)) {
+		if ((mix = calloc(1, sizeof(*mix))) == NULL)
+			return NULL;
+	} else {
+		mix = spa_list_first(&data->free_mix, struct mix, link);
+		spa_list_remove(&mix->link);
+	}
+
+	mix_init(mix, port, mix_id);
+	spa_list_append(&data->mix[direction], &mix->link);
+
+	return mix;
+}
+
+
+static int client_node_transport(void *_data,
+			int readfd, int writefd, uint32_t mem_id, uint32_t offset, uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+
+	clean_transport(data);
+
+	data->activation = pw_mempool_map_id(data->pool, mem_id,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, NULL);
+	if (data->activation == NULL) {
+		pw_log_warn("remote-node %p: can't map activation: %m", proxy);
+		return -errno;
+	}
+
+	data->node->rt.activation = data->activation->ptr;
+
+	pw_log_debug("remote-node %p: fds:%d %d node:%u activation:%p",
+		proxy, readfd, writefd, data->remote_id, data->activation->ptr);
+
+	data->rtwritefd = writefd;
+	spa_system_close(data->context->data_system, data->node->source.fd);
+	data->node->source.fd = readfd;
+
+	data->have_transport = true;
+
+	if (data->node->active)
+		pw_client_node_set_active(data->client_node, true);
+
+	return 0;
+}
+
+static int add_node_update(struct node_data *data, uint32_t change_mask, uint32_t info_mask)
+{
+	struct pw_impl_node *node = data->node;
+	struct spa_node_info ni = SPA_NODE_INFO_INIT();
+	uint32_t n_params = 0;
+	struct spa_pod **params = NULL;
+	int res;
+
+	if (change_mask & PW_CLIENT_NODE_UPDATE_PARAMS) {
+		uint32_t i, idx, id;
+		uint8_t buf[4096];
+		struct spa_pod_dynamic_builder b;
+
+		for (i = 0; i < node->info.n_params; i++) {
+			struct spa_pod *param;
+
+			id = node->info.params[i].id;
+			if (id == SPA_PARAM_Invalid)
+				continue;
+
+			for (idx = 0;;) {
+				spa_pod_dynamic_builder_init(&b, buf, sizeof(buf), 4096);
+
+	                        res = spa_node_enum_params_sync(node->node,
+							id, &idx, NULL, &param, &b.b);
+				if (res == 1) {
+					void *p;
+					p = pw_reallocarray(params, n_params + 1, sizeof(struct spa_pod *));
+					if (p == NULL) {
+						res = -errno;
+						pw_log_error("realloc failed: %m");
+					} else {
+						params = p;
+						params[n_params++] = spa_pod_copy(param);
+					}
+				}
+				spa_pod_dynamic_builder_clean(&b);
+				if (res != 1)
+	                                break;
+			}
+                }
+	}
+	if (change_mask & PW_CLIENT_NODE_UPDATE_INFO) {
+		ni.max_input_ports = node->info.max_input_ports;
+		ni.max_output_ports = node->info.max_output_ports;
+		ni.change_mask = info_mask;
+		ni.flags = node->spa_flags;
+		ni.props = node->info.props;
+		ni.params = node->info.params;
+		ni.n_params = node->info.n_params;
+	}
+
+        res = pw_client_node_update(data->client_node,
+				change_mask,
+				n_params,
+				(const struct spa_pod **)params,
+				&ni);
+
+	if (params) {
+		while (n_params > 0)
+			free(params[--n_params]);
+		free(params);
+	}
+	return res;
+}
+
+static int add_port_update(struct node_data *data, struct pw_impl_port *port, uint32_t change_mask)
+{
+	struct spa_port_info pi = SPA_PORT_INFO_INIT();
+	uint32_t n_params = 0;
+	struct spa_pod **params = NULL;
+	int res;
+
+	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_PARAMS) {
+		uint32_t i, idx, id;
+		uint8_t buf[4096];
+		struct spa_pod_dynamic_builder b;
+
+		for (i = 0; i < port->info.n_params; i++) {
+			struct spa_pod *param;
+
+			id = port->info.params[i].id;
+			if (id == SPA_PARAM_Invalid)
+				continue;
+
+			for (idx = 0;;) {
+				spa_pod_dynamic_builder_init(&b, buf, sizeof(buf), 4096);
+
+	                        res = spa_node_port_enum_params_sync(port->node->node,
+							port->direction, port->port_id,
+							id, &idx, NULL, &param, &b.b);
+				if (res == 1) {
+					void *p;
+					p = pw_reallocarray(params, n_params + 1, sizeof(struct spa_pod*));
+					if (p == NULL) {
+						res = -errno;
+						pw_log_error("realloc failed: %m");
+					} else {
+						params = p;
+						params[n_params++] = spa_pod_copy(param);
+					}
+				}
+				spa_pod_dynamic_builder_clean(&b);
+
+				if (res != 1)
+	                                break;
+
+			}
+                }
+	}
+	if (change_mask & PW_CLIENT_NODE_PORT_UPDATE_INFO) {
+		pi.change_mask = SPA_PORT_CHANGE_MASK_FLAGS |
+			SPA_PORT_CHANGE_MASK_RATE |
+			SPA_PORT_CHANGE_MASK_PROPS |
+			SPA_PORT_CHANGE_MASK_PARAMS;
+		pi.flags = port->spa_flags;
+		pi.rate = SPA_FRACTION(0, 1);
+		pi.props = &port->properties->dict;
+		SPA_FLAG_CLEAR(pi.flags, SPA_PORT_FLAG_DYNAMIC_DATA);
+		pi.n_params = port->info.n_params;
+		pi.params = port->info.params;
+	}
+
+	res = pw_client_node_port_update(data->client_node,
+                                         port->direction,
+                                         port->port_id,
+                                         change_mask,
+                                         n_params,
+                                         (const struct spa_pod **)params,
+					 &pi);
+	if (params) {
+		while (n_params > 0)
+			free(params[--n_params]);
+		free(params);
+	}
+	return res;
+}
+
+static int
+client_node_set_param(void *_data, uint32_t id, uint32_t flags,
+		      const struct spa_pod *param)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	int res;
+
+	pw_log_debug("node %p: set_param %s:", proxy,
+			spa_debug_type_find_name(spa_type_param, id));
+
+	res = spa_node_set_param(data->node->node, id, flags, param);
+
+	if (res < 0) {
+		pw_log_error("node %p: set_param %s (%d) %p: %s", proxy,
+				spa_debug_type_find_name(spa_type_param, id),
+				id, param, spa_strerror(res));
+		pw_proxy_errorf(proxy, res, "node_set_param(%s) failed: %s",
+				spa_debug_type_find_name(spa_type_param, id),
+				spa_strerror(res));
+	}
+	return res;
+}
+
+static int
+client_node_set_io(void *_data,
+		   uint32_t id,
+		   uint32_t memid,
+		   uint32_t offset,
+		   uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct pw_memmap *old, *mm;
+	void *ptr;
+	uint32_t tag[5] = { data->remote_id, SPA_ID_INVALID, id, };
+	int res;
+
+	old = pw_mempool_find_tag(data->pool, tag, sizeof(tag));
+
+	if (memid == SPA_ID_INVALID) {
+		mm = ptr = NULL;
+		size = 0;
+	} else {
+		mm = pw_mempool_map_id(data->pool, memid,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, tag);
+		if (mm == NULL) {
+			pw_log_warn("can't map memory id %u: %m", memid);
+			res = -errno;
+			goto exit;
+		}
+		ptr = mm->ptr;
+	}
+
+	pw_log_debug("node %p: set io %s %p", proxy,
+			spa_debug_type_find_name(spa_type_io, id), ptr);
+
+	res =  spa_node_set_io(data->node->node, id, ptr, size);
+
+	pw_memmap_free(old);
+exit:
+	if (res < 0) {
+		pw_log_error("node %p: set_io: %s", proxy, spa_strerror(res));
+		pw_proxy_errorf(proxy, res, "node_set_io failed: %s", spa_strerror(res));
+	}
+	return res;
+}
+
+static int client_node_event(void *data, const struct spa_event *event)
+{
+	pw_log_warn("unhandled node event %d", SPA_EVENT_TYPE(event));
+	return -ENOTSUP;
+}
+
+static int client_node_command(void *_data, const struct spa_command *command)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	int res;
+
+	switch (SPA_NODE_COMMAND_ID(command)) {
+	case SPA_NODE_COMMAND_Pause:
+		pw_log_debug("node %p: pause", proxy);
+
+		if ((res = pw_impl_node_set_state(data->node, PW_NODE_STATE_IDLE)) < 0) {
+			pw_log_warn("node %p: pause failed", proxy);
+			pw_proxy_error(proxy, res, "pause failed");
+		}
+
+		break;
+	case SPA_NODE_COMMAND_Start:
+		pw_log_debug("node %p: start", proxy);
+
+		if ((res = pw_impl_node_set_state(data->node, PW_NODE_STATE_RUNNING)) < 0) {
+			pw_log_warn("node %p: start failed", proxy);
+			pw_proxy_error(proxy, res, "start failed");
+		}
+		break;
+
+	case SPA_NODE_COMMAND_Suspend:
+		pw_log_debug("node %p: suspend", proxy);
+		if ((res = pw_impl_node_set_state(data->node, PW_NODE_STATE_SUSPENDED)) < 0) {
+			pw_log_warn("node %p: suspend failed", proxy);
+			pw_proxy_error(proxy, res, "suspend failed");
+		}
+		break;
+	case SPA_NODE_COMMAND_RequestProcess:
+		res = pw_impl_node_send_command(data->node, command);
+		break;
+	default:
+		pw_log_warn("unhandled node command %d", SPA_NODE_COMMAND_ID(command));
+		res = -ENOTSUP;
+		pw_proxy_errorf(proxy, res, "command %d not supported", SPA_NODE_COMMAND_ID(command));
+	}
+	return res;
+}
+
+static int
+client_node_add_port(void *_data, enum spa_direction direction, uint32_t port_id,
+		const struct spa_dict *props)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	pw_log_warn("add port not supported");
+	pw_proxy_error(proxy, -ENOTSUP, "add port not supported");
+	return -ENOTSUP;
+}
+
+static int
+client_node_remove_port(void *_data, enum spa_direction direction, uint32_t port_id)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	pw_log_warn("remove port not supported");
+	pw_proxy_error(proxy, -ENOTSUP, "remove port not supported");
+	return -ENOTSUP;
+}
+
+static int clear_buffers(struct node_data *data, struct mix *mix)
+{
+	struct pw_impl_port *port = mix->port;
+        struct buffer *b;
+	int res;
+
+        pw_log_debug("port %p: clear %zd buffers mix:%d", port,
+			pw_array_get_len(&mix->buffers, struct buffer *),
+			mix->mix_id);
+
+	if ((res = pw_impl_port_use_buffers(port, &mix->mix, 0, NULL, 0)) < 0) {
+		pw_log_error("port %p: error clear buffers %s", port, spa_strerror(res));
+		return res;
+	}
+
+        pw_array_for_each(b, &mix->buffers) {
+		pw_log_debug("port %p: clear buffer %d map %p %p",
+			port, b->id, b->mem, b->buf);
+		pw_memmap_free(b->mem);
+		free(b->buf);
+        }
+	mix->buffers.size = 0;
+	return 0;
+}
+
+static int
+client_node_port_set_param(void *_data,
+			   enum spa_direction direction, uint32_t port_id,
+			   uint32_t id, uint32_t flags,
+			   const struct spa_pod *param)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct pw_impl_port *port;
+	int res;
+
+	port = pw_impl_node_find_port(data->node, direction, port_id);
+	if (port == NULL) {
+		res = -EINVAL;
+		goto error_exit;
+	}
+
+	pw_log_debug("port %p: set_param %s %p", port,
+			spa_debug_type_find_name(spa_type_param, id), param);
+
+	res = pw_impl_port_set_param(port, id, flags, param);
+	if (res < 0)
+		goto error_exit;
+
+	if (id == SPA_PARAM_Format) {
+		struct mix *mix;
+		spa_list_for_each(mix, &data->mix[direction], link) {
+			if (mix->port->port_id == port_id)
+				clear_buffers(data, mix);
+		}
+	}
+	return res;
+
+error_exit:
+	pw_log_error("port %p: set_param %d %p: %s", port, id, param, spa_strerror(res));
+	pw_proxy_errorf(proxy, res, "port_set_param(%s) failed: %s",
+				spa_debug_type_find_name(spa_type_param, id),
+				spa_strerror(res));
+	return res;
+}
+
+static int
+client_node_port_use_buffers(void *_data,
+			     enum spa_direction direction, uint32_t port_id, uint32_t mix_id,
+			     uint32_t flags,
+			     uint32_t n_buffers, struct pw_client_node_buffer *buffers)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct buffer *bid;
+	uint32_t i, j;
+	struct spa_buffer *b, **bufs;
+	struct mix *mix;
+	int res, prot;
+
+	mix = ensure_mix(data, direction, port_id, mix_id);
+	if (mix == NULL) {
+		res = -ENOENT;
+		goto error_exit;
+	}
+
+	if (n_buffers > MAX_BUFFERS)
+		return -ENOSPC;
+
+	prot = PW_MEMMAP_FLAG_READWRITE;
+
+	/* clear previous buffers */
+	clear_buffers(data, mix);
+
+	bufs = alloca(n_buffers * sizeof(struct spa_buffer *));
+
+	for (i = 0; i < n_buffers; i++) {
+		size_t size;
+		off_t offset;
+		struct pw_memmap *mm;
+
+		mm = pw_mempool_map_id(data->pool, buffers[i].mem_id,
+				prot, buffers[i].offset, buffers[i].size, NULL);
+		if (mm == NULL) {
+			res = -errno;
+			goto error_exit_cleanup;
+		}
+
+		bid = pw_array_add(&mix->buffers, sizeof(struct buffer));
+		if (bid == NULL) {
+			res = -errno;
+			goto error_exit_cleanup;
+		}
+		bid->id = i;
+		bid->mem = mm;
+
+		if (data->allow_mlock && mlock(mm->ptr, mm->size) < 0)
+			if (errno != ENOMEM || !mlock_warned) {
+				pw_log(data->warn_mlock ? SPA_LOG_LEVEL_WARN : SPA_LOG_LEVEL_DEBUG,
+						"Failed to mlock memory %p %u: %s",
+						mm->ptr, mm->size,
+						errno == ENOMEM ?
+						"This is not a problem but for best performance, "
+						"consider increasing RLIMIT_MEMLOCK" : strerror(errno));
+				mlock_warned |= errno == ENOMEM;
+			}
+
+		size = sizeof(struct spa_buffer);
+		for (j = 0; j < buffers[i].buffer->n_metas; j++)
+			size += sizeof(struct spa_meta);
+		for (j = 0; j < buffers[i].buffer->n_datas; j++)
+			size += sizeof(struct spa_data);
+
+		b = bid->buf = malloc(size);
+		if (b == NULL) {
+			res = -errno;
+			goto error_exit_cleanup;
+		}
+		memcpy(b, buffers[i].buffer, sizeof(struct spa_buffer));
+
+		b->metas = SPA_PTROFF(b, sizeof(struct spa_buffer), struct spa_meta);
+		b->datas = SPA_PTROFF(b->metas, sizeof(struct spa_meta) * b->n_metas,
+				       struct spa_data);
+
+		pw_log_debug("add buffer mem:%d id:%d offset:%u size:%u %p", mm->block->id,
+				bid->id, buffers[i].offset, buffers[i].size, bid->buf);
+
+		offset = 0;
+		for (j = 0; j < b->n_metas; j++) {
+			struct spa_meta *m = &b->metas[j];
+			memcpy(m, &buffers[i].buffer->metas[j], sizeof(struct spa_meta));
+			m->data = SPA_PTROFF(mm->ptr, offset, void);
+			offset += SPA_ROUND_UP_N(m->size, 8);
+		}
+
+		for (j = 0; j < b->n_datas; j++) {
+			struct spa_data *d = &b->datas[j];
+
+			memcpy(d, &buffers[i].buffer->datas[j], sizeof(struct spa_data));
+			d->chunk =
+			    SPA_PTROFF(mm->ptr, offset + sizeof(struct spa_chunk) * j,
+				       struct spa_chunk);
+
+			if (flags & SPA_NODE_BUFFERS_FLAG_ALLOC)
+				continue;
+
+			if (d->type == SPA_DATA_MemId) {
+				uint32_t mem_id = SPA_PTR_TO_UINT32(d->data);
+				struct pw_memblock *bm;
+
+				bm = pw_mempool_find_id(data->pool, mem_id);
+				if (bm == NULL) {
+					pw_log_error("unknown buffer mem %u", mem_id);
+					res = -ENODEV;
+					goto error_exit_cleanup;
+				}
+
+				d->fd = bm->fd;
+				d->type = bm->type;
+				d->data = NULL;
+
+				pw_log_debug(" data %d %u -> fd %d maxsize %d",
+						j, bm->id, bm->fd, d->maxsize);
+			} else if (d->type == SPA_DATA_MemPtr) {
+				int offs = SPA_PTR_TO_INT(d->data);
+				d->data = SPA_PTROFF(mm->ptr, offs, void);
+				d->fd = -1;
+				pw_log_debug(" data %d id:%u -> mem:%p offs:%d maxsize:%d",
+						j, bid->id, d->data, offs, d->maxsize);
+			} else {
+				pw_log_warn("unknown buffer data type %d", d->type);
+			}
+		}
+		bufs[i] = b;
+	}
+
+	if ((res = pw_impl_port_use_buffers(mix->port, &mix->mix, flags, bufs, n_buffers)) < 0)
+		goto error_exit_cleanup;
+
+	if (flags & SPA_NODE_BUFFERS_FLAG_ALLOC) {
+		pw_client_node_port_buffers(data->client_node,
+				direction, port_id, mix_id,
+				n_buffers,
+				bufs);
+	}
+	return res;
+
+error_exit_cleanup:
+	clear_buffers(data, mix);
+error_exit:
+        pw_log_error("port %p: use_buffers: %d %s", mix, res, spa_strerror(res));
+	pw_proxy_errorf(proxy, res, "port_use_buffers error: %s", spa_strerror(res));
+	return res;
+}
+
+static int
+client_node_port_set_io(void *_data,
+                        uint32_t direction,
+                        uint32_t port_id,
+                        uint32_t mix_id,
+                        uint32_t id,
+                        uint32_t memid,
+                        uint32_t offset,
+                        uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct mix *mix;
+	struct pw_memmap *mm, *old;
+	void *ptr;
+	int res = 0;
+	uint32_t tag[5] = { data->remote_id, direction, port_id, mix_id, id };
+
+	mix = ensure_mix(data, direction, port_id, mix_id);
+	if (mix == NULL) {
+		res = -ENOENT;
+		goto exit;
+	}
+
+	old = pw_mempool_find_tag(data->pool, tag, sizeof(tag));
+
+	if (memid == SPA_ID_INVALID) {
+		mm = ptr = NULL;
+		size = 0;
+	}
+	else {
+		mm = pw_mempool_map_id(data->pool, memid,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, tag);
+		if (mm == NULL) {
+			pw_log_warn("can't map memory id %u: %m", memid);
+			res = -errno;
+			goto exit;
+		}
+		ptr = mm->ptr;
+	}
+
+	pw_log_debug("port %p: set io:%s new:%p old:%p", mix->port,
+			spa_debug_type_find_name(spa_type_io, id), ptr, mix->mix.io);
+
+	if (id == SPA_IO_Buffers) {
+		if (ptr == NULL && mix->mix.io)
+			deactivate_mix(data, mix);
+	}
+
+	if ((res = spa_node_port_set_io(mix->port->mix,
+			     direction, mix->mix.port.port_id, id, ptr, size)) < 0) {
+		if (res == -ENOTSUP)
+			res = 0;
+		else
+			goto exit_free;
+	}
+	if (id == SPA_IO_Buffers) {
+		mix->mix.io = ptr;
+		if (ptr)
+			activate_mix(data, mix);
+	}
+exit_free:
+	pw_memmap_free(old);
+exit:
+	if (res < 0) {
+		pw_log_error("port %p: set_io: %s", mix, spa_strerror(res));
+		pw_proxy_errorf(proxy, res, "port_set_io failed: %s", spa_strerror(res));
+	}
+	return res;
+}
+
+static int link_signal_func(void *user_data)
+{
+	struct link *link = user_data;
+	struct spa_system *data_system = link->data->context->data_system;
+
+	pw_log_trace_fp("link %p: signal %p", link, link->target.activation);
+	if (SPA_UNLIKELY(spa_system_eventfd_write(data_system, link->signalfd, 1) < 0))
+		pw_log_warn("link %p: write failed %m", link);
+
+	return 0;
+}
+
+static int
+do_activate_link(struct spa_loop *loop,
+                bool async, uint32_t seq, const void *data, size_t size, void *user_data)
+{
+	struct link *link = user_data;
+	struct node_data *d = link->data;
+	pw_log_trace("link %p activate", link);
+	spa_list_append(&d->node->rt.target_list, &link->target.link);
+	return 0;
+}
+
+static int
+client_node_set_activation(void *_data,
+                        uint32_t node_id,
+                        int signalfd,
+                        uint32_t memid,
+                        uint32_t offset,
+                        uint32_t size)
+{
+	struct node_data *data = _data;
+	struct pw_proxy *proxy = (struct pw_proxy*)data->client_node;
+	struct pw_impl_node *node = data->node;
+	struct pw_memmap *mm;
+	void *ptr;
+	struct link *link;
+	int res = 0;
+
+	if (data->remote_id == node_id) {
+		pw_log_debug("node %p: our activation %u: %u %u %u", node, node_id,
+				memid, offset, size);
+		spa_system_close(data->context->data_system, signalfd);
+		return 0;
+	}
+
+	if (memid == SPA_ID_INVALID) {
+		mm = ptr = NULL;
+		size = 0;
+	} else {
+		mm = pw_mempool_map_id(data->pool, memid,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, NULL);
+		if (mm == NULL) {
+			res = -errno;
+			goto error_exit;
+		}
+		ptr = mm->ptr;
+	}
+	pw_log_debug("node %p: set activation %d %p %u %u", node, node_id, ptr, offset, size);
+
+	if (ptr) {
+		link = calloc(1, sizeof(struct link));
+		if (link == NULL) {
+			res = -errno;
+			goto error_exit;
+		}
+		link->data = data;
+		link->node_id = node_id;
+		link->map = mm;
+		link->target.activation = ptr;
+		link->signalfd = signalfd;
+		link->target.signal_func = link_signal_func;
+		link->target.data = link;
+		link->target.node = NULL;
+		spa_list_append(&data->links, &link->link);
+
+		pw_loop_invoke(data->context->data_loop,
+                       do_activate_link, SPA_ID_INVALID, NULL, 0, false, link);
+
+		pw_log_debug("node %p: link %p: fd:%d id:%u state %p required %d, pending %d",
+				node, link, signalfd,
+				link->target.activation->position.clock.id,
+				&link->target.activation->state[0],
+				link->target.activation->state[0].required,
+				link->target.activation->state[0].pending);
+	} else {
+		link = find_activation(&data->links, node_id);
+		if (link == NULL) {
+			res = -ENOENT;
+			goto error_exit;
+		}
+		clear_link(data, link);
+	}
+	return res;
+
+error_exit:
+	pw_log_error("node %p: set activation %d: %s", node, node_id, spa_strerror(res));
+	pw_proxy_errorf(proxy, res, "set_activation: %s", spa_strerror(res));
+	return res;
+}
+
+static const struct pw_client_node_events client_node_events = {
+	PW_VERSION_CLIENT_NODE_EVENTS,
+	.transport = client_node_transport,
+	.set_param = client_node_set_param,
+	.set_io = client_node_set_io,
+	.event = client_node_event,
+	.command = client_node_command,
+	.add_port = client_node_add_port,
+	.remove_port = client_node_remove_port,
+	.port_set_param = client_node_port_set_param,
+	.port_use_buffers = client_node_port_use_buffers,
+	.port_set_io = client_node_port_set_io,
+	.set_activation = client_node_set_activation,
+};
+
+static void do_node_init(struct node_data *data)
+{
+	struct pw_impl_port *port;
+
+	pw_log_debug("%p: node %p init", data, data->node);
+	add_node_update(data, PW_CLIENT_NODE_UPDATE_PARAMS |
+				PW_CLIENT_NODE_UPDATE_INFO,
+				SPA_NODE_CHANGE_MASK_FLAGS |
+				SPA_NODE_CHANGE_MASK_PROPS |
+				SPA_NODE_CHANGE_MASK_PARAMS);
+
+	spa_list_for_each(port, &data->node->input_ports, link) {
+		add_port_update(data, port,
+				PW_CLIENT_NODE_PORT_UPDATE_PARAMS |
+				PW_CLIENT_NODE_PORT_UPDATE_INFO);
+	}
+	spa_list_for_each(port, &data->node->output_ports, link) {
+		add_port_update(data, port,
+				PW_CLIENT_NODE_PORT_UPDATE_PARAMS |
+				PW_CLIENT_NODE_PORT_UPDATE_INFO);
+	}
+}
+
+static void clear_mix(struct node_data *data, struct mix *mix)
+{
+	pw_log_debug("port %p: mix clear %d.%d", mix->port, mix->port->port_id, mix->mix_id);
+
+	deactivate_mix(data, mix);
+
+	spa_list_remove(&mix->link);
+
+	clear_buffers(data, mix);
+	pw_array_clear(&mix->buffers);
+
+	spa_list_append(&data->free_mix, &mix->link);
+	pw_impl_port_release_mix(mix->port, &mix->mix);
+}
+
+static void clean_node(struct node_data *d)
+{
+	struct mix *mix;
+
+	if (d->have_transport) {
+		spa_list_consume(mix, &d->mix[SPA_DIRECTION_INPUT], link)
+			clear_mix(d, mix);
+		spa_list_consume(mix, &d->mix[SPA_DIRECTION_OUTPUT], link)
+			clear_mix(d, mix);
+	}
+	spa_list_consume(mix, &d->free_mix, link) {
+		spa_list_remove(&mix->link);
+		free(mix);
+	}
+	clean_transport(d);
+}
+
+static void node_destroy(void *data)
+{
+	struct node_data *d = data;
+
+	pw_log_debug("%p: destroy", d);
+
+	clean_node(d);
+}
+
+static void node_free(void *data)
+{
+	struct node_data *d = data;
+	pw_log_debug("%p: free", d);
+	d->node = NULL;
+}
+
+static void node_info_changed(void *data, const struct pw_node_info *info)
+{
+	struct node_data *d = data;
+	uint32_t change_mask, info_mask;
+
+	pw_log_debug("info changed %p", d);
+
+	if (d->client_node == NULL)
+		return;
+
+	change_mask = PW_CLIENT_NODE_UPDATE_INFO;
+	info_mask = SPA_NODE_CHANGE_MASK_FLAGS;
+	if (info->change_mask & PW_NODE_CHANGE_MASK_PROPS) {
+		info_mask |= SPA_NODE_CHANGE_MASK_PROPS;
+	}
+	if (info->change_mask & PW_NODE_CHANGE_MASK_PARAMS) {
+		change_mask |= PW_CLIENT_NODE_UPDATE_PARAMS;
+		info_mask |= SPA_NODE_CHANGE_MASK_PARAMS;
+	}
+	add_node_update(d, change_mask, info_mask);
+}
+
+static void node_port_info_changed(void *data, struct pw_impl_port *port,
+		const struct pw_port_info *info)
+{
+	struct node_data *d = data;
+	uint32_t change_mask = 0;
+
+	pw_log_debug("info changed %p", d);
+
+	if (d->client_node == NULL)
+		return;
+
+	if (info->change_mask & PW_PORT_CHANGE_MASK_PROPS)
+		change_mask |= PW_CLIENT_NODE_PORT_UPDATE_INFO;
+	if (info->change_mask & PW_PORT_CHANGE_MASK_PARAMS) {
+		change_mask |= PW_CLIENT_NODE_PORT_UPDATE_PARAMS;
+		change_mask |= PW_CLIENT_NODE_PORT_UPDATE_INFO;
+	}
+	add_port_update(d, port, change_mask);
+}
+
+static void node_port_removed(void *data, struct pw_impl_port *port)
+{
+	struct node_data *d = data;
+	struct mix *mix, *tmp;
+
+	pw_log_debug("removed %p", d);
+
+	if (d->client_node == NULL)
+		return;
+
+	pw_client_node_port_update(d->client_node,
+			port->direction,
+			port->port_id,
+			0, 0, NULL, NULL);
+
+	spa_list_for_each_safe(mix, tmp, &d->mix[port->direction], link) {
+		if (mix->port == port)
+			clear_mix(d, mix);
+	}
+}
+
+static void node_active_changed(void *data, bool active)
+{
+	struct node_data *d = data;
+	pw_log_debug("active %d", active);
+
+	if (d->client_node == NULL)
+		return;
+
+	pw_client_node_set_active(d->client_node, active);
+}
+
+static void node_event(void *data, const struct spa_event *event)
+{
+	struct node_data *d = data;
+	pw_log_debug("%p", d);
+
+	if (d->client_node == NULL)
+		return;
+	pw_client_node_event(d->client_node, event);
+}
+
+static const struct pw_impl_node_events node_events = {
+	PW_VERSION_IMPL_NODE_EVENTS,
+	.destroy = node_destroy,
+	.free = node_free,
+	.info_changed = node_info_changed,
+	.port_info_changed = node_port_info_changed,
+	.port_removed = node_port_removed,
+	.active_changed = node_active_changed,
+	.event = node_event,
+};
+
+static void client_node_removed(void *_data)
+{
+	struct node_data *data = _data;
+	pw_log_debug("%p: removed", data);
+
+	spa_hook_remove(&data->proxy_client_node_listener);
+	spa_hook_remove(&data->client_node_listener);
+
+	if (data->node) {
+		spa_hook_remove(&data->node_listener);
+		pw_impl_node_set_state(data->node, PW_NODE_STATE_SUSPENDED);
+
+		clean_node(data);
+
+		if (data->do_free)
+			pw_impl_node_destroy(data->node);
+	}
+	data->client_node = NULL;
+}
+
+static void client_node_destroy(void *_data)
+{
+	struct node_data *data = _data;
+
+	pw_log_debug("%p: destroy", data);
+	client_node_removed(_data);
+}
+
+static void client_node_bound(void *_data, uint32_t global_id)
+{
+	struct node_data *data = _data;
+	pw_log_debug("%p: bound %u", data, global_id);
+	data->remote_id = global_id;
+}
+
+static const struct pw_proxy_events proxy_client_node_events = {
+	PW_VERSION_PROXY_EVENTS,
+	.removed = client_node_removed,
+	.destroy = client_node_destroy,
+	.bound = client_node_bound,
+};
+
+static int node_ready(void *d, int status)
+{
+	struct node_data *data = d;
+	struct pw_impl_node *node = data->node;
+	struct pw_node_activation *a = node->rt.activation;
+	struct spa_system *data_system = data->context->data_system;
+	struct timespec ts;
+	struct pw_impl_port *p;
+
+	pw_log_trace_fp("node %p: ready driver:%d exported:%d status:%d", node,
+			node->driver, node->exported, status);
+
+	if (status & SPA_STATUS_HAVE_DATA) {
+		spa_list_for_each(p, &node->rt.output_mix, rt.node_link)
+			spa_node_process(p->mix);
+	}
+
+	spa_system_clock_gettime(data_system, CLOCK_MONOTONIC, &ts);
+	a->status = PW_NODE_ACTIVATION_TRIGGERED;
+	a->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
+
+	if (SPA_UNLIKELY(spa_system_eventfd_write(data_system, data->rtwritefd, 1) < 0))
+		pw_log_warn("node %p: write failed %m", node);
+
+	return 0;
+}
+
+static int node_reuse_buffer(void *data, uint32_t port_id, uint32_t buffer_id)
+{
+	return 0;
+}
+
+static int node_xrun(void *d, uint64_t trigger, uint64_t delay, struct spa_pod *info)
+{
+	struct node_data *data = d;
+	struct pw_impl_node *node = data->node;
+	struct pw_node_activation *a = node->rt.activation;
+
+	a->xrun_count++;
+	a->xrun_time = trigger;
+	a->xrun_delay = delay;
+	a->max_delay = SPA_MAX(a->max_delay, delay);
+
+	pw_log_debug("node %p: XRun! count:%u time:%"PRIu64" delay:%"PRIu64" max:%"PRIu64,
+			node, a->xrun_count, trigger, delay, a->max_delay);
+
+	pw_context_driver_emit_xrun(data->context, node);
+
+	return 0;
+}
+
+static const struct spa_node_callbacks node_callbacks = {
+	SPA_VERSION_NODE_CALLBACKS,
+	.ready = node_ready,
+	.reuse_buffer = node_reuse_buffer,
+	.xrun = node_xrun
+};
+
+static struct pw_proxy *node_export(struct pw_core *core, void *object, bool do_free,
+		size_t user_data_size)
+{
+	struct pw_impl_node *node = object;
+	struct pw_proxy *client_node;
+	struct node_data *data;
+
+	user_data_size = SPA_ROUND_UP_N(user_data_size, __alignof__(struct node_data));
+
+	client_node = pw_core_create_object(core,
+			"client-node",
+			PW_TYPE_INTERFACE_ClientNode,
+			PW_VERSION_CLIENT_NODE,
+			&node->properties->dict,
+			user_data_size + sizeof(struct node_data));
+	if (client_node == NULL)
+		goto error;
+
+	data = pw_proxy_get_user_data(client_node);
+	data = SPA_PTROFF(data, user_data_size, struct node_data);
+	data->pool = pw_core_get_mempool(core);
+	data->node = node;
+	data->do_free = do_free;
+	data->context = pw_impl_node_get_context(node);
+	data->client_node = (struct pw_client_node *)client_node;
+	data->remote_id = SPA_ID_INVALID;
+
+
+	data->allow_mlock = pw_properties_get_bool(node->properties, "mem.allow-mlock",
+						   data->context->settings.mem_allow_mlock);
+
+	data->warn_mlock = pw_properties_get_bool(node->properties, "mem.warn-mlock",
+						  data->context->settings.mem_warn_mlock);
+
+	node->exported = true;
+
+	spa_list_init(&data->free_mix);
+	spa_list_init(&data->mix[0]);
+	spa_list_init(&data->mix[1]);
+
+	spa_list_init(&data->links);
+
+	pw_proxy_add_listener(client_node,
+			&data->proxy_client_node_listener,
+			&proxy_client_node_events, data);
+
+	spa_node_set_callbacks(node->node, &node_callbacks, data);
+	pw_impl_node_add_listener(node, &data->node_listener, &node_events, data);
+
+	pw_client_node_add_listener(data->client_node,
+					  &data->client_node_listener,
+					  &client_node_events,
+					  data);
+	do_node_init(data);
+
+	return client_node;
+error:
+	if (do_free)
+		pw_impl_node_destroy(node);
+	return NULL;
+
+}
+
+struct pw_proxy *pw_core_node_export(struct pw_core *core,
+		const char *type, const struct spa_dict *props, void *object,
+		size_t user_data_size)
+{
+	struct pw_impl_node *node = object;
+
+	if (props)
+		pw_impl_node_update_properties(node, props);
+	return node_export(core, object, false, user_data_size);
+}
+
+struct pw_proxy *pw_core_spa_node_export(struct pw_core *core,
+		const char *type, const struct spa_dict *props, void *object,
+		size_t user_data_size)
+{
+	struct pw_impl_node *node;
+	struct pw_proxy *proxy;
+	const char *str;
+	bool do_register;
+
+	str = props ? spa_dict_lookup(props, PW_KEY_OBJECT_REGISTER) : NULL;
+	do_register = str ? pw_properties_parse_bool(str) : true;
+
+	node = pw_context_create_node(pw_core_get_context(core),
+			props ? pw_properties_new_dict(props) : NULL, 0);
+	if (node == NULL)
+		return NULL;
+
+	pw_impl_node_set_implementation(node, (struct spa_node*)object);
+
+	if (do_register)
+		pw_impl_node_register(node, NULL);
+
+	proxy = node_export(core, node, true, user_data_size);
+	if (proxy)
+		pw_impl_node_set_active(node, true);
+
+	return proxy;
+}