mem: Add tag to memmap

Add a tag field when creating a memmap so that we can do lookup on it.
This makes it easier to implement the tracking of mappings for io areas.
Remove custom io memory tracking and use the tags.

Add flags to spa_chunk to make data corrupted. The flags on the buffer
stay constant for the life of the buffer. Add flags to mark memory
readable and writable. Mark memory readonly in v4l2-source.

Pass the daemon activation area to the client in the transport event.
This never changes and need to be handled differently from the other
activation areas.

Use the right flags when importing memory.

Add the (desired) memory type to mempool_alloc.

improve some debug.

src/modules/module-client-node/client-node.c

@@ -120,6 +120,7 @@ struct node {
 	struct spa_callbacks callbacks;
 
 	struct pw_resource *resource;
+	struct pw_client *client;
 
 	struct spa_source data_source;
 	int writefd;
@@ -164,8 +165,6 @@ struct impl {
 #define pw_client_node_resource(r,m,v,...)	\
 	pw_resource_call_res(r,struct pw_client_node_proxy_events,m,v,__VA_ARGS__)
 
-#define pw_client_node_resource_add_mem(r,...)	\
-	pw_client_node_resource(r,add_mem,0,__VA_ARGS__)
 #define pw_client_node_resource_transport(r,...)	\
 	pw_client_node_resource(r,transport,0,__VA_ARGS__)
 #define pw_client_node_resource_set_param(r,...)	\
@@ -249,7 +248,7 @@ static int clear_buffers(struct node *this, struct mix *mix)
 				uint32_t id;
 
 				id = SPA_PTR_TO_UINT32(b->buffer.datas[j].data);
-				m = pw_mempool_find_id(this->resource->client->pool, id);
+				m = pw_mempool_find_id(this->client->pool, id);
 				if (m) {
 					pw_log_debug(NAME " %p: mem %d", impl, m->id);
 					pw_memblock_unref(m);
@@ -352,7 +351,7 @@ static int impl_node_set_io(void *object, uint32_t id, void *data, size_t size)
 		if (mem_offset + size > mem->map->size)
 			return -EINVAL;
 
-		m = pw_mempool_import_block(this->resource->client->pool, mem);
+		m = pw_mempool_import_block(this->client->pool, mem);
 		if (m == NULL)
 			return -errno;
 
@@ -658,7 +657,7 @@ static int do_port_set_io(struct impl *impl,
 		if (mem_offset + size > mem->map->size)
 			return -EINVAL;
 
-		m = pw_mempool_import_block(this->resource->client->pool, mem);
+		m = pw_mempool_import_block(this->client->pool, mem);
 		if (m == NULL)
 			return -errno;
 
@@ -762,7 +761,7 @@ do_port_use_buffers(struct impl *impl,
 				data_size += d->maxsize;
 		}
 
-		m = pw_mempool_import_block(this->resource->client->pool, mem);
+		m = pw_mempool_import_block(this->client->pool, mem);
 		if (m == NULL)
 			return -errno;
 
@@ -786,10 +785,18 @@ do_port_use_buffers(struct impl *impl,
 
 			if (d->type == SPA_DATA_DmaBuf ||
 			    d->type == SPA_DATA_MemFd) {
-				m = pw_mempool_import(this->resource->client->pool,
-					d->type, d->fd, d->flags);
+				uint32_t flags = PW_MEMBLOCK_FLAG_DONT_CLOSE;
+
+				if (d->flags & SPA_DATA_FLAG_READABLE)
+					flags |= PW_MEMBLOCK_FLAG_READABLE;
+				if (d->flags & SPA_DATA_FLAG_WRITABLE)
+					flags |= PW_MEMBLOCK_FLAG_WRITABLE;
+
+				m = pw_mempool_import(this->client->pool,
+					flags, d->type, d->fd);
 				if (m == NULL)
 					return -errno;
+
 				b->buffer.datas[j].type = SPA_DATA_MemId;
 				b->buffer.datas[j].data = SPA_UINT32_TO_PTR(m->id);
 			} else if (d->type == SPA_DATA_MemPtr) {
@@ -896,7 +903,7 @@ client_node_get_node(void *data,
 
 	impl->bind_node_version = version;
 	impl->bind_node_id = new_id;
-	pw_map_insert_at(&this->resource->client->objects, new_id, NULL);
+	pw_map_insert_at(&this->client->objects, new_id, NULL);
 
 	return NULL;
 }
@@ -1160,19 +1167,15 @@ void pw_client_node_registered(struct pw_client_node *this, struct pw_global *gl
 	struct pw_memblock *m;
 
 	pw_log_debug(NAME " %p: %d", this, node_id);
-	pw_client_node_resource_transport(this->resource,
-					  node_id,
-					  impl->other_fds[0],
-					  impl->other_fds[1]);
-
 	m = pw_mempool_import_block(this->resource->client->pool, node->activation);
 	if (m == NULL) {
 		pw_log_debug(NAME " %p: can't import block: %m", this);
 		return;
 	}
 
-	pw_client_node_resource_set_activation(this->resource,
+	pw_client_node_resource_transport(this->resource,
 					  node_id,
+					  impl->other_fds[0],
 					  impl->other_fds[1],
 					  m->id,
 					  0,
@@ -1208,11 +1211,12 @@ static void node_initialized(void *data)
 
 	size = sizeof(struct spa_io_buffers) * MAX_AREAS;
 
-	if (pw_mempool_alloc(impl->core->pool,
-				PW_MEMBLOCK_FLAG_READWRITE |
-				PW_MEMBLOCK_FLAG_MAP |
-				PW_MEMBLOCK_FLAG_SEAL,
-				size, &impl->io_areas) < 0)
+	impl->io_areas = pw_mempool_alloc(impl->core->pool,
+			PW_MEMBLOCK_FLAG_READWRITE |
+			PW_MEMBLOCK_FLAG_MAP |
+			PW_MEMBLOCK_FLAG_SEAL,
+			SPA_DATA_MemFd, size);
+	if (impl->io_areas == NULL)
                 return;
 
 	pw_log_debug(NAME " %p: io areas %p", node, impl->io_areas->map->ptr);
@@ -1475,7 +1479,7 @@ static void node_peer_added(void *data, struct pw_node *peer)
 	if (this->resource == NULL)
 		return;
 
-	m = pw_mempool_import_block(this->resource->client->pool, peer->activation);
+	m = pw_mempool_import_block(this->client->pool, peer->activation);
 	if (m == NULL) {
 		pw_log_debug(NAME " %p: can't ensure mem: %m", this);
 		return;
@@ -1500,7 +1504,7 @@ static void node_peer_removed(void *data, struct pw_node *peer)
 	if (this->resource == NULL)
 		return;
 
-	m = pw_mempool_find_fd(this->resource->client->pool,
+	m = pw_mempool_find_fd(this->client->pool,
 			peer->activation->fd);
 	if (m == NULL) {
 		pw_log_warn(NAME " %p: unknown peer %p fd:%d", &impl->this, peer,
@@ -1596,6 +1600,7 @@ struct pw_client_node *pw_client_node_new(struct pw_resource *resource,
 	node_init(&impl->node, NULL, support, n_support);
 	impl->node.impl = impl;
 	impl->node.resource = resource;
+	impl->node.client = client;
 	this->flags = do_register ? 0 : 1;
 
 	pw_map_init(&impl->io_map, 64, 64);
@@ -1606,7 +1611,7 @@ struct pw_client_node *pw_client_node_new(struct pw_resource *resource,
 	this->resource = resource;
 	this->parent = parent;
 	this->node = pw_spa_node_new(core,
-				     pw_resource_get_client(this->resource),
+				     client,
 				     parent,
 				     name,
 				     PW_SPA_NODE_FLAG_ASYNC |

src/modules/module-client-node/protocol-native.c

@@ -266,14 +266,17 @@ static int client_node_demarshal_transport(void *object, const struct pw_protoco
 {
 	struct pw_proxy *proxy = object;
 	struct spa_pod_parser prs;
-	uint32_t node_id, ridx, widx;
+	uint32_t node_id, ridx, widx, mem_id, offset, sz;
 	int readfd, writefd;
 
 	spa_pod_parser_init(&prs, msg->data, msg->size);
 	if (spa_pod_parser_get_struct(&prs,
 			SPA_POD_Int(&node_id),
 			SPA_POD_Int(&ridx),
-			SPA_POD_Int(&widx)) < 0)
+			SPA_POD_Int(&widx),
+			SPA_POD_Int(&mem_id),
+			SPA_POD_Int(&offset),
+			SPA_POD_Int(&sz)) < 0)
 		return -EINVAL;
 
 	readfd = pw_protocol_native_get_proxy_fd(proxy, ridx);
@@ -283,7 +286,8 @@ static int client_node_demarshal_transport(void *object, const struct pw_protoco
 		return -EINVAL;
 
 	pw_proxy_notify(proxy, struct pw_client_node_proxy_events, transport, 0, node_id,
-								   readfd, writefd);
+								   readfd, writefd, mem_id,
+								   offset, sz);
 	return 0;
 }
 
@@ -540,7 +544,8 @@ static int client_node_demarshal_set_io(void *object, const struct pw_protocol_n
 	return 0;
 }
 
-static int client_node_marshal_transport(void *object, uint32_t node_id, int readfd, int writefd)
+static int client_node_marshal_transport(void *object, uint32_t node_id, int readfd, int writefd,
+		uint32_t mem_id, uint32_t offset, uint32_t size)
 {
 	struct pw_protocol_native_message *msg;
 	struct pw_resource *resource = object;
@@ -551,7 +556,10 @@ static int client_node_marshal_transport(void *object, uint32_t node_id, int rea
 	spa_pod_builder_add_struct(b,
 			       SPA_POD_Int(node_id),
 			       SPA_POD_Int(pw_protocol_native_add_resource_fd(resource, readfd)),
-			       SPA_POD_Int(pw_protocol_native_add_resource_fd(resource, writefd)));
+			       SPA_POD_Int(pw_protocol_native_add_resource_fd(resource, writefd)),
+			       SPA_POD_Int(mem_id),
+			       SPA_POD_Int(offset),
+			       SPA_POD_Int(size));
 
 	return pw_protocol_native_end_resource(resource, b);
 }

src/modules/module-client-node/remote-node.c

@@ -51,11 +51,6 @@ struct buffer {
 	struct pw_memmap *mem;
 };
 
-struct io {
-	uint32_t id;
-	struct pw_memmap *mem;
-};
-
 struct mix {
 	struct spa_list link;
 	struct pw_port *port;
@@ -63,7 +58,6 @@ struct mix {
 	struct pw_port_mix mix;
 	struct pw_array buffers;
 	bool active;
-	struct io ios[MAX_IO];
 };
 
 struct link {
@@ -79,6 +73,7 @@ struct node_data {
 
 	uint32_t remote_id;
 	int rtwritefd;
+	struct pw_memmap *activation;
 
 	struct mix mix_pool[MAX_MIX];
 	struct spa_list mix[2];
@@ -94,7 +89,6 @@ struct node_data {
 	struct spa_hook proxy_listener;
         struct pw_proxy *proxy;
 
-	struct io ios[MAX_IO];
 	struct spa_io_position *position;
 
 	struct pw_array links;
@@ -102,50 +96,6 @@ struct node_data {
 
 /** \endcond */
 
-static void init_ios(struct io *ios)
-{
-	int i;
-	for (i = 0; i < MAX_IO; i++)
-		ios[i].id = SPA_ID_INVALID;
-}
-
-static void clear_io(struct io *io)
-{
-	pw_log_debug("%p clear id:%u mem:%p", io, io->id, io->mem);
-	pw_memmap_free(io->mem);
-	io->mem = NULL;
-	io->id = SPA_ID_INVALID;
-}
-
-static struct io *update_io(struct node_data *data, struct io *ios,
-		uint32_t id, struct pw_memmap *mem)
-{
-	int i;
-	struct io *io, *f = NULL;
-
-	pw_log_debug("node %p: update id:%u mem:%p", data, id, mem);
-
-	for (i = 0; i < MAX_IO; i++) {
-		io = &ios[i];
-		if (io->id == SPA_ID_INVALID && f == NULL)
-			f = io;
-		else if (io->id == id) {
-			if (io->mem && io->mem != mem)
-				clear_io(io);
-			f = io;
-			break;
-		}
-	}
-	if (f == NULL)
-		return NULL;
-
-	io = f;
-	io->id = id;
-	io->mem = mem;
-
-	return io;
-}
-
 static struct link *find_activation(struct pw_array *links, uint32_t node_id)
 {
 	struct link *l;
@@ -179,6 +129,7 @@ static void clean_transport(struct node_data *data)
 	}
 	pw_array_clear(&data->links);
 
+	pw_memmap_free(data->activation);
 	close(data->rtwritefd);
 	data->remote_id = SPA_ID_INVALID;
 	data->have_transport = false;
@@ -192,7 +143,6 @@ static void mix_init(struct mix *mix, struct pw_port *port, uint32_t mix_id)
 	mix->active = false;
 	pw_array_init(&mix->buffers, 32);
 	pw_array_ensure_size(&mix->buffers, sizeof(struct buffer) * 64);
-	init_ios(mix->ios);
 }
 
 static int
@@ -278,7 +228,7 @@ static struct mix *ensure_mix(struct node_data *data,
 
 
 static int client_node_transport(void *object, uint32_t node_id,
-			int readfd, int writefd)
+			int readfd, int writefd, uint32_t mem_id, uint32_t offset, uint32_t size)
 {
 	struct pw_proxy *proxy = object;
 	struct node_data *data = proxy->user_data;
@@ -286,16 +236,25 @@ static int client_node_transport(void *object, uint32_t node_id,
 
 	clean_transport(data);
 
-	data->have_transport = true;
-	data->remote_id = node_id;
+	data->activation = pw_mempool_map_id(proxy->remote->pool, mem_id,
+				PW_MEMMAP_FLAG_READWRITE, offset, size, NULL);
+	if (data->activation == NULL) {
+		pw_log_debug("remote-node %p: can't map activation: %m", proxy);
+		return -errno;
+	}
 
-	pw_log_debug("remote-node %p: create transport with fds %d %d for node %u",
-		proxy, readfd, writefd, node_id);
+	data->remote_id = node_id;
+	data->node->rt.activation = data->activation->ptr;
+
+	pw_log_debug("remote-node %p: fds:%d %d node:%u activation:%p",
+		proxy, readfd, writefd, node_id, data->activation->ptr);
 
         data->rtwritefd = writefd;
 	close(data->node->source.fd);
 	data->node->source.fd = readfd;
 
+	data->have_transport = true;
+
 	if (data->node->active)
 		pw_client_node_proxy_set_active(data->node_proxy, true);
 
@@ -441,14 +400,17 @@ client_node_set_io(void *object,
 	struct node_data *data = proxy->user_data;
 	struct pw_memmap *mm;
 	void *ptr;
+	uint32_t tag[5] = { data->remote_id, id, };
 
 	if (memid == SPA_ID_INVALID) {
-		size = 0;
+		if ((mm = pw_mempool_find_tag(proxy->remote->pool, tag)) != NULL)
+			pw_memmap_free(mm);
 		mm = ptr = NULL;
+		size = 0;
 	}
 	else {
 		mm = pw_mempool_map_id(proxy->remote->pool, memid,
-				PROT_READ|PROT_WRITE, offset, size);
+				PW_MEMMAP_FLAG_READWRITE, offset, size, tag);
 		if (mm == NULL) {
 			pw_log_warn("can't map memory id %u: %m", memid);
 			return -errno;
@@ -459,8 +421,6 @@ client_node_set_io(void *object,
 	pw_log_debug("node %p: set io %s %p", proxy,
 			spa_debug_type_find_name(spa_type_io, id), ptr);
 
-	update_io(data, data->ios, id, mm);
-
 	switch (id) {
 	case SPA_IO_Position:
 		data->position = ptr;
@@ -535,7 +495,7 @@ static int clear_buffers(struct node_data *data, struct mix *mix)
         struct buffer *b;
 	int res;
 
-        pw_log_debug("port %p: clear buffers %d", port, mix->mix_id);
+        pw_log_debug("port %p: clear buffers mix:%d", port, mix->mix_id);
 	if ((res = pw_port_use_buffers(port, mix->mix_id, NULL, 0)) < 0) {
 		pw_log_error("port %p: error clear buffers %s", port, spa_strerror(res));
 		return res;
@@ -609,7 +569,7 @@ client_node_port_use_buffers(void *object,
 		goto error_exit;
 	}
 
-	prot = PROT_READ | (direction == SPA_DIRECTION_OUTPUT ? PROT_WRITE : 0);
+	prot = PW_MEMMAP_FLAG_READ | (direction == SPA_DIRECTION_OUTPUT ? PW_MEMMAP_FLAG_WRITE : 0);
 
 	/* clear previous buffers */
 	clear_buffers(data, mix);
@@ -622,7 +582,7 @@ client_node_port_use_buffers(void *object,
 		struct pw_memmap *mm;
 
 		mm = pw_mempool_map_id(proxy->remote->pool, buffers[i].mem_id,
-				prot, buffers[i].offset, buffers[i].size);
+				prot, buffers[i].offset, buffers[i].size, NULL);
 		if (mm == NULL) {
 			res = -errno;
 			goto error_exit_cleanup;
@@ -735,6 +695,7 @@ client_node_port_set_io(void *object,
 	struct pw_memmap *mm;
 	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) {
@@ -743,12 +704,15 @@ client_node_port_set_io(void *object,
 	}
 
 	if (memid == SPA_ID_INVALID) {
+		if ((mm = pw_mempool_find_tag(proxy->remote->pool, tag)) != NULL)
+			pw_memmap_free(mm);
+
 		mm = ptr = NULL;
 		size = 0;
 	}
 	else {
 		mm = pw_mempool_map_id(proxy->remote->pool, memid,
-				PROT_READ|PROT_WRITE, offset, size);
+				PW_MEMMAP_FLAG_READWRITE, offset, size, tag);
 		if (mm == NULL) {
 			res = -errno;
 			goto error_exit;
@@ -759,8 +723,6 @@ client_node_port_set_io(void *object,
 	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);
 
-	update_io(data, mix->ios, id, mm);
-
 	if (id == SPA_IO_Buffers) {
 		if (ptr == NULL && mix->mix.io)
 			deactivate_mix(data, mix);
@@ -799,6 +761,7 @@ static int link_signal_func(void *user_data)
 
 	if (write(link->signalfd, &cmd, sizeof(cmd)) != sizeof(cmd))
 		pw_log_warn("link %p: write failed %m", link);
+
 	return 0;
 }
 
@@ -825,7 +788,7 @@ client_node_set_activation(void *object,
 	}
 	else {
 		mm = pw_mempool_map_id(proxy->remote->pool, memid,
-				PROT_READ|PROT_WRITE, offset, size);
+				PW_MEMMAP_FLAG_READWRITE, offset, size, NULL);
 		if (mm == NULL) {
 			res = -errno;
 			goto error_exit;
@@ -837,6 +800,8 @@ client_node_set_activation(void *object,
 	if (data->remote_id == node_id) {
 		pw_log_debug("node %p: our activation %u: %u %u %u %p", node, node_id,
 				memid, offset, size, ptr);
+		if (mm)
+			pw_memmap_free(mm);
 		close(signalfd);
 		return 0;
 	}
@@ -1098,7 +1063,6 @@ static struct pw_proxy *node_export(struct pw_remote *remote, void *object, bool
 	data->core = pw_node_get_core(node);
 	data->node_proxy = (struct pw_client_node_proxy *)proxy;
 	data->remote_id = SPA_ID_INVALID;
-	init_ios(data->ios);
 
 	node->exported = true;