data: clean up fd and data management

Do not send the offset and size in the add_mem call, just send the
fd and the flags. The area that we need to map from this to find the
meta, chunk and data are sent in a separate call. This should make
it possible to truncate the memory to a larger size to dynamically
allocate more shared memory for a client.
Remove the Id data type, it's not needed.
Don't automatically map memory in remote.c
Pass the original memory type from server to client.
Handle DmaBuf mem in video-play now that the server passed it on.

spa/include/spa/buffer/buffer.h

@@ -32,33 +32,37 @@ extern "C" {
  *
  * Buffers describe the data and metadata that is exchanged between
  * ports of a node.
- *
  */
 #define SPA_TYPE__Buffer		SPA_TYPE_POINTER_BASE "Buffer"
 #define SPA_TYPE_BUFFER_BASE		SPA_TYPE__Buffer ":"
 
+/** Buffers contain data of a certain type */
 #define SPA_TYPE__Data			SPA_TYPE_ENUM_BASE "DataType"
 #define SPA_TYPE_DATA_BASE		SPA_TYPE__Data ":"
 
+/** The data type for a plain memory pointer. The data member points to
+ * the memory. */
 #define SPA_TYPE_DATA__MemPtr		SPA_TYPE_DATA_BASE "MemPtr"
-#define SPA_TYPE_DATA__MemFd		SPA_TYPE_DATA_BASE "MemFd"
+
-#define SPA_TYPE_DATA__DmaBuf		SPA_TYPE_DATA_BASE "DmaBuf"
+/** base type for fd based memory */
-#define SPA_TYPE_DATA__Id		SPA_TYPE_DATA_BASE "Id"
+#define SPA_TYPE_DATA__Fd		SPA_TYPE_DATA_BASE "Fd"
+#define SPA_TYPE_DATA_FD_BASE		SPA_TYPE_DATA__Fd ":"
+
+#define SPA_TYPE_DATA_FD__MemFd		SPA_TYPE_DATA_FD_BASE "MemFd"
+#define SPA_TYPE_DATA_FD__DmaBuf	SPA_TYPE_DATA_FD_BASE "DmaBuf"
 
 struct spa_type_data {
-	uint32_t MemPtr;
+	uint32_t MemPtr;	/**< system memory */
-	uint32_t MemFd;
+	uint32_t MemFd;		/**< memory accesible with an fd */
-	uint32_t DmaBuf;
+	uint32_t DmaBuf;	/**< dmabuf fd */
-	uint32_t Id;
 };
 
 static inline void spa_type_data_map(struct spa_type_map *map, struct spa_type_data *type)
 {
 	if (type->MemPtr == 0) {
 		type->MemPtr = spa_type_map_get_id(map, SPA_TYPE_DATA__MemPtr);
-		type->MemFd = spa_type_map_get_id(map, SPA_TYPE_DATA__MemFd);
+		type->MemFd = spa_type_map_get_id(map, SPA_TYPE_DATA_FD__MemFd);
-		type->DmaBuf = spa_type_map_get_id(map, SPA_TYPE_DATA__DmaBuf);
+		type->DmaBuf = spa_type_map_get_id(map, SPA_TYPE_DATA_FD__DmaBuf);
-		type->Id = spa_type_map_get_id(map, SPA_TYPE_DATA__Id);
 	}
 }
 

src/examples/video-play.c

@@ -111,7 +111,8 @@ on_stream_new_buffer(void *_data, uint32_t id)
 
 	buf = pw_stream_peek_buffer(stream, id);
 
-	if (buf->datas[0].type == data->type.data.MemFd) {
+	if (buf->datas[0].type == data->type.data.MemFd ||
+	    buf->datas[0].type == data->type.data.DmaBuf) {
 		map = mmap(NULL, buf->datas[0].maxsize + buf->datas[0].mapoffset, PROT_READ,
 			   MAP_PRIVATE, buf->datas[0].fd, 0);
 		sdata = SPA_MEMBER(map, buf->datas[0].mapoffset, uint8_t);

src/extensions/client-node.h

@@ -405,9 +405,7 @@ struct pw_client_node_proxy_events {
 			      uint32_t mem_id,
 			      uint32_t type,
 			      int memfd,
-			      uint32_t flags,
+			      uint32_t flags);
-			      uint32_t offset,
-			      uint32_t size);
 	/**
 	 * Notify the port of buffers
 	 *

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

@@ -544,8 +544,7 @@ spa_proxy_node_port_set_io(struct spa_node *node,
 						     direction, port_id,
 						     memid,
 						     t->data.MemFd,
-						     mem->fd, mem->flags,
+						     mem->fd, mem->flags);
-						     0, mem->offset + mem->size);
 
 		pw_client_node_resource_port_set_io(this->resource,
 						    this->seq,
@@ -634,7 +633,7 @@ spa_proxy_node_port_use_buffers(struct spa_node *node,
 
 		mb[i].buffer = &b->buffer;
 		mb[i].mem_id = n_mem++;
-		mb[i].offset = 0;
+		mb[i].offset = SPA_PTRDIFF(baseptr, m->ptr + m->offset);
 		mb[i].size = data_size;
 
 		pw_client_node_resource_port_add_mem(this->resource,
@@ -642,9 +641,7 @@ spa_proxy_node_port_use_buffers(struct spa_node *node,
 						     port_id,
 						     mb[i].mem_id,
 						     t->data.MemFd,
-						     m->fd, m->flags,
+						     m->fd, m->flags);
-						     SPA_PTRDIFF(baseptr, m->ptr + m->offset),
-						     data_size);
 
 		for (j = 0; j < buffers[i]->n_metas; j++)
 			memcpy(&b->buffer.metas[j], &buffers[i]->metas[j], sizeof(struct spa_meta));
@@ -663,8 +660,7 @@ spa_proxy_node_port_use_buffers(struct spa_node *node,
 								     n_mem,
 								     d->type,
 								     d->fd,
-								     d->flags, d->mapoffset, d->maxsize);
+								     d->flags);
-				b->buffer.datas[j].type = t->data.Id;
 				b->buffer.datas[j].data = SPA_UINT32_TO_PTR(n_mem);
 				n_mem++;
 			} else if (d->type == t->data.MemPtr) {

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

@@ -298,7 +298,7 @@ static bool client_node_demarshal_port_add_mem(void *object, void *data, size_t
 {
 	struct pw_proxy *proxy = object;
 	struct spa_pod_parser prs;
-	uint32_t direction, port_id, mem_id, type, memfd_idx, flags, offset, sz;
+	uint32_t direction, port_id, mem_id, type, memfd_idx, flags;
 	int memfd;
 
 	spa_pod_parser_init(&prs, data, size, 0);
@@ -309,9 +309,7 @@ static bool client_node_demarshal_port_add_mem(void *object, void *data, size_t
 			"i", &mem_id,
 			"I", &type,
 			"i", &memfd_idx,
-			"i", &flags,
+			"i", &flags, NULL) < 0)
-			"i", &offset,
-			"i", &sz, NULL) < 0)
 		return false;
 
 	memfd = pw_protocol_native_get_proxy_fd(proxy, memfd_idx);
@@ -320,7 +318,7 @@ static bool client_node_demarshal_port_add_mem(void *object, void *data, size_t
 								      port_id,
 								      mem_id,
 								      type,
-								      memfd, flags, offset, sz);
+								      memfd, flags);
 	return true;
 }
 
@@ -564,7 +562,7 @@ client_node_marshal_port_add_mem(void *object,
 				 uint32_t port_id,
 				 uint32_t mem_id,
 				 uint32_t type,
-				 int memfd, uint32_t flags, uint32_t offset, uint32_t size)
+				 int memfd, uint32_t flags)
 {
 	struct pw_resource *resource = object;
 	struct spa_pod_builder *b;
@@ -577,9 +575,7 @@ client_node_marshal_port_add_mem(void *object,
 			       "i", mem_id,
 			       "I", type,
 			       "i", pw_protocol_native_add_resource_fd(resource, memfd),
-			       "i", flags,
+			       "i", flags);
-			       "i", offset,
-			       "i", size);
 
 	pw_protocol_native_end_resource(resource, b);
 }

src/pipewire/remote.c

@@ -50,16 +50,15 @@ struct mem_id {
 	uint32_t id;
 	int fd;
 	uint32_t flags;
-	void *ptr;
-	uint32_t offset;
-	uint32_t size;
 };
 
 struct buffer_id {
 	struct spa_list link;
 	uint32_t id;
-	void *buf_ptr;
 	struct spa_buffer *buf;
+	void *ptr;
+	uint32_t offset;
+	uint32_t size;
 };
 
 struct port {
@@ -851,9 +850,6 @@ static struct mem_id *find_mem(struct port *port, uint32_t id)
 
 static void clear_memid(struct mem_id *mid)
 {
-	if (mid->ptr != NULL)
-		munmap(mid->ptr, mid->size + mid->offset);
-	mid->ptr = NULL;
 	close(mid->fd);
 }
 
@@ -873,6 +869,11 @@ static void clear_buffers(struct port *port)
         pw_log_debug("port %p: clear buffers", port);
 
         pw_array_for_each(bid, &port->buffer_ids) {
+		if (bid->ptr != NULL) {
+			if (munmap(bid->ptr, bid->size + bid->offset) < 0)
+				pw_log_warn("failed to unmap: %m");
+		}
+		bid->ptr = NULL;
                 free(bid->buf);
                 bid->buf = NULL;
         }
@@ -891,7 +892,7 @@ static void
 client_node_port_add_mem(void *object,
 			 enum spa_direction direction, uint32_t port_id,
 			 uint32_t mem_id,
-			 uint32_t type, int memfd, uint32_t flags, uint32_t offset, uint32_t size)
+			 uint32_t type, int memfd, uint32_t flags)
 {
 	struct pw_proxy *proxy = object;
 	struct node_data *data = proxy->user_data;
@@ -900,20 +901,17 @@ client_node_port_add_mem(void *object,
 
 	m = find_mem(port, mem_id);
 	if (m) {
-		pw_log_debug("update mem %u, fd %d, flags %d, off %d, size %d",
+		pw_log_debug("update mem %u, fd %d, flags %d",
-			     mem_id, memfd, flags, offset, size);
+			     mem_id, memfd, flags);
 		clear_memid(m);
 	} else {
 		m = pw_array_add(&port->mem_ids, sizeof(struct mem_id));
-		pw_log_debug("add mem %u, fd %d, flags %d, off %d, size %d",
+		pw_log_debug("add mem %u, fd %d, flags %d",
-			     mem_id, memfd, flags, offset, size);
+			     mem_id, memfd, flags);
 	}
 	m->id = mem_id;
 	m->fd = memfd;
 	m->flags = flags;
-	m->ptr = NULL;
-	m->offset = offset;
-	m->size = size;
 }
 
 static void
@@ -928,6 +926,7 @@ client_node_port_use_buffers(void *object,
 	uint32_t i, j, len;
 	struct spa_buffer *b, **bufs;
 	struct port *port;
+	struct pw_type *t = &proxy->remote->core->type;
 	int res, prot;
 
 	port = find_port(data, direction, port_id);
@@ -954,22 +953,23 @@ client_node_port_use_buffers(void *object,
 			continue;
 		}
 
-		if (mid->ptr == NULL) {
-			mid->ptr =
-			    mmap(NULL, mid->size + mid->offset, prot, MAP_SHARED, mid->fd, 0);
-			if (mid->ptr == MAP_FAILED) {
-				mid->ptr = NULL;
-				pw_log_warn("Failed to mmap memory %d %p: %s", mid->size, mid,
-					    strerror(errno));
-				continue;
-			}
-		}
 		len = pw_array_get_len(&port->buffer_ids, struct buffer_id);
 		bid = pw_array_add(&port->buffer_ids, sizeof(struct buffer_id));
 
+		bid->offset = buffers[i].offset;
+		bid->size = buffers[i].size;
+		bid->ptr = mmap(NULL, bid->offset + bid->size, prot, MAP_SHARED, mid->fd, 0);
+		if (bid->ptr == MAP_FAILED) {
+			bid->ptr = NULL;
+			pw_log_warn("Failed to mmap memory %u %p: %s", bid->size, mid,
+				    strerror(errno));
+			continue;
+		}
+		if (mlock(bid->ptr, bid->offset + bid->size) < 0)
+			pw_log_warn("Failed to lock memory %u %s", bid->offset + bid->size, strerror(errno));
+
 		b = buffers[i].buffer;
 
-		bid->buf_ptr = SPA_MEMBER(mid->ptr, mid->offset + buffers[i].offset, void);
 		{
 			size_t size;
 
@@ -992,13 +992,13 @@ client_node_port_use_buffers(void *object,
 		if (bid->id != len) {
 			pw_log_warn("unexpected id %u found, expected %u", bid->id, len);
 		}
-		pw_log_debug("add buffer %d %d %u", mid->id, bid->id, buffers[i].offset);
+		pw_log_debug("add buffer %d %d %u %u", mid->id, bid->id, bid->offset, bid->size);
 
-		offset = 0;
+		offset = bid->offset;
 		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_MEMBER(bid->buf_ptr, offset, void);
+			m->data = SPA_MEMBER(bid->ptr, offset, void);
 			offset += m->size;
 		}
 
@@ -1007,25 +1007,17 @@ client_node_port_use_buffers(void *object,
 
 			memcpy(d, &buffers[i].buffer->datas[j], sizeof(struct spa_data));
 			d->chunk =
-			    SPA_MEMBER(bid->buf_ptr, offset + sizeof(struct spa_chunk) * j,
+			    SPA_MEMBER(bid->ptr, offset + sizeof(struct spa_chunk) * j,
 				       struct spa_chunk);
 
-			if (d->type == proxy->remote->core->type.data.Id) {
+			if (d->type == t->data.MemFd || d->type == t->data.DmaBuf) {
 				struct mem_id *bmid = find_mem(port, SPA_PTR_TO_UINT32(d->data));
-				void *map;
 
-				d->type = proxy->remote->core->type.data.MemFd;
+				d->data = NULL;
 				d->fd = bmid->fd;
-				map = mmap(NULL, d->maxsize + d->mapoffset, prot, MAP_SHARED, d->fd, 0);
+				pw_log_debug(" data %d %u -> fd %d", j, bmid->id, bmid->fd);
-				if (map == MAP_FAILED) {
+			} else if (d->type == t->data.MemPtr) {
-					pw_log_error("data %d failed to mmap memory %m", j);
+				d->data = SPA_MEMBER(bid->ptr, bid->offset + SPA_PTR_TO_INT(d->data), void);
-					res = errno;
-					goto done;
-				}
-				d->data = SPA_MEMBER(map, d->mapoffset, uint8_t);
-				pw_log_debug(" data %d %u -> fd %d mem %p", j, bmid->id, bmid->fd, map);
-			} else if (d->type == proxy->remote->core->type.data.MemPtr) {
-				d->data = SPA_MEMBER(bid->buf_ptr, SPA_PTR_TO_INT(d->data), void);
 				d->fd = -1;
 				pw_log_debug(" data %d %u -> mem %p", j, bid->id, d->data);
 			} else {
@@ -1076,6 +1068,7 @@ client_node_port_set_io(void *object,
 	struct node_data *data = proxy->user_data;
 	struct port *port;
 	struct mem_id *mid;
+	void *ptr;
 
 	port = find_port(data, direction, port_id);
 	if (port == NULL)
@@ -1087,21 +1080,17 @@ client_node_port_set_io(void *object,
 		return;
 	}
 
-	if (mid->ptr == NULL) {
+	ptr = mmap(NULL, offset + size, PROT_READ|PROT_WRITE, MAP_SHARED|MAP_LOCKED, mid->fd, 0);
-		mid->ptr =
+	if (ptr == MAP_FAILED) {
-		    mmap(NULL, mid->size + mid->offset, PROT_READ|PROT_WRITE, MAP_SHARED, mid->fd, 0);
+		pw_log_warn("Failed to mmap memory %d %p: %s", size, mid,
-		if (mid->ptr == MAP_FAILED) {
+			    strerror(errno));
-			mid->ptr = NULL;
+		return;
-			pw_log_warn("Failed to mmap memory %d %p: %s", mid->size, mid,
-				    strerror(errno));
-			return;
-		}
 	}
 
 	spa_node_port_set_io(port->port->node->node,
 			     direction, port_id,
 			     id,
-			     SPA_MEMBER(mid->ptr, offset, void),
+			     SPA_MEMBER(ptr, offset, void),
 			     size);
 
 }

src/pipewire/stream.c

@@ -46,17 +46,16 @@ struct mem_id {
 	uint32_t id;
 	int fd;
 	uint32_t flags;
-	void *ptr;
-	uint32_t offset;
-	uint32_t size;
 };
 
 struct buffer_id {
 	struct spa_list link;
 	uint32_t id;
 	bool used;
-	void *buf_ptr;
 	struct spa_buffer *buf;
+	void *ptr;
+	uint32_t offset;
+	uint32_t size;
 };
 
 struct stream {
@@ -109,9 +108,6 @@ struct stream {
 
 static void clear_memid(struct stream *impl, struct mem_id *mid)
 {
-	if (mid->ptr != NULL)
-		munmap(mid->ptr, mid->size + mid->offset);
-	mid->ptr = NULL;
 	if (mid->fd != -1) {
 		bool has_ref = false;
 		int fd;
@@ -149,6 +145,10 @@ static void clear_buffers(struct pw_stream *stream)
 
 	pw_array_for_each(bid, &impl->buffer_ids) {
 		spa_hook_list_call(&stream->listener_list, struct pw_stream_events, remove_buffer, bid->id);
+		if (bid->ptr != NULL)
+			if (munmap(bid->ptr, bid->size + bid->offset) < 0)
+				pw_log_warn("failed to unmap buffer: %m");
+		bid->ptr = NULL;
 		free(bid->buf);
 		bid->buf = NULL;
 		bid->used = false;
@@ -797,7 +797,7 @@ static void
 client_node_port_add_mem(void *data,
 			 enum spa_direction direction, uint32_t port_id,
 			 uint32_t mem_id,
-			 uint32_t type, int memfd, uint32_t flags, uint32_t offset, uint32_t size)
+			 uint32_t type, int memfd, uint32_t flags)
 {
 	struct stream *impl = data;
 	struct pw_stream *stream = &impl->this;
@@ -805,20 +805,17 @@ client_node_port_add_mem(void *data,
 
 	m = find_mem(stream, mem_id);
 	if (m) {
-		pw_log_debug("update mem %u, fd %d, flags %d, off %d, size %d",
+		pw_log_debug("update mem %u, fd %d, flags %d",
-			     mem_id, memfd, flags, offset, size);
+			     mem_id, memfd, flags);
 		clear_memid(impl, m);
 	} else {
 		m = pw_array_add(&impl->mem_ids, sizeof(struct mem_id));
-		pw_log_debug("add mem %u, fd %d, flags %d, off %d, size %d",
+		pw_log_debug("add mem %u, fd %d, flags %d",
-			     mem_id, memfd, flags, offset, size);
+			     mem_id, memfd, flags);
 	}
 	m->id = mem_id;
 	m->fd = memfd;
 	m->flags = flags;
-	m->ptr = NULL;
-	m->offset = offset;
-	m->size = size;
 }
 
 static void
@@ -829,6 +826,7 @@ client_node_port_use_buffers(void *data,
 {
 	struct stream *impl = data;
 	struct pw_stream *stream = &impl->this;
+	struct pw_type *t = &stream->remote->core->type;
 	struct buffer_id *bid;
 	uint32_t i, j, len;
 	struct spa_buffer *b;
@@ -848,16 +846,6 @@ client_node_port_use_buffers(void *data,
 			continue;
 		}
 
-		if (mid->ptr == NULL) {
-			mid->ptr =
-			    mmap(NULL, mid->size + mid->offset, prot, MAP_SHARED, mid->fd, 0);
-			if (mid->ptr == MAP_FAILED) {
-				mid->ptr = NULL;
-				pw_log_warn("Failed to mmap memory %d %p: %s", mid->size, mid,
-					    strerror(errno));
-				continue;
-			}
-		}
 		len = pw_array_get_len(&impl->buffer_ids, struct buffer_id);
 		bid = pw_array_add(&impl->buffer_ids, sizeof(struct buffer_id));
 		if (impl->direction == SPA_DIRECTION_OUTPUT) {
@@ -869,7 +857,17 @@ client_node_port_use_buffers(void *data,
 
 		b = buffers[i].buffer;
 
-		bid->buf_ptr = SPA_MEMBER(mid->ptr, mid->offset + buffers[i].offset, void);
+		bid->offset = buffers[i].offset;
+		bid->size = buffers[i].size;
+
+		bid->ptr = mmap(NULL, bid->offset + bid->size, prot, MAP_SHARED, mid->fd, 0);
+		if (bid->ptr == MAP_FAILED) {
+			bid->ptr = NULL;
+			pw_log_warn("Failed to mmap memory %d %p: %s", bid->size, mid,
+				    strerror(errno));
+			continue;
+		}
+
 		{
 			size_t size;
 
@@ -893,13 +891,13 @@ client_node_port_use_buffers(void *data,
 			pw_log_warn("unexpected id %u found, expected %u", bid->id, len);
 			impl->in_order = false;
 		}
-		pw_log_debug("add buffer %d %d %u", mid->id, bid->id, buffers[i].offset);
+		pw_log_debug("add buffer %d %d %u %u", mid->id, bid->id, bid->offset, bid->size);
 
-		offset = 0;
+		offset = bid->offset;
 		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_MEMBER(bid->buf_ptr, offset, void);
+			m->data = SPA_MEMBER(bid->ptr, offset, void);
 			offset += m->size;
 		}
 
@@ -908,17 +906,16 @@ client_node_port_use_buffers(void *data,
 
 			memcpy(d, &buffers[i].buffer->datas[j], sizeof(struct spa_data));
 			d->chunk =
-			    SPA_MEMBER(bid->buf_ptr, offset + sizeof(struct spa_chunk) * j,
+			    SPA_MEMBER(bid->ptr, offset + sizeof(struct spa_chunk) * j,
 				       struct spa_chunk);
 
-			if (d->type == stream->remote->core->type.data.Id) {
+			if (d->type == t->data.MemFd || d->type == t->data.DmaBuf) {
 				struct mem_id *bmid = find_mem(stream, SPA_PTR_TO_UINT32(d->data));
-				d->type = stream->remote->core->type.data.MemFd;
 				d->data = NULL;
 				d->fd = bmid->fd;
 				pw_log_debug(" data %d %u -> fd %d", j, bmid->id, bmid->fd);
-			} else if (d->type == stream->remote->core->type.data.MemPtr) {
+			} else if (d->type == t->data.MemPtr) {
-				d->data = SPA_MEMBER(bid->buf_ptr, SPA_PTR_TO_INT(d->data), void);
+				d->data = SPA_MEMBER(bid->ptr, bid->offset + SPA_PTR_TO_INT(d->data), void);
 				d->fd = -1;
 				pw_log_debug(" data %d %u -> mem %p", j, bid->id, d->data);
 			} else {