Move transport to client-node extension

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

@@ -35,11 +35,11 @@
 #include "pipewire/pipewire.h"
 #include "pipewire/private.h"
 #include "pipewire/interfaces.h"
-#include "pipewire/transport.h"
 
 #include "pipewire/core.h"
 #include "modules/spa/spa-node.h"
 #include "client-node.h"
+#include "transport.h"
 
 /** \cond */
 
@@ -93,6 +93,8 @@ struct proxy {
 	struct spa_log *log;
 	struct spa_loop *data_loop;
 
+	struct pw_type_event_client_node type_event_client_node;
+
 	const struct spa_node_callbacks *callbacks;
 	void *callbacks_data;
 
@@ -120,7 +122,7 @@ struct impl {
 
 	struct proxy proxy;
 
-	struct pw_transport *transport;
+	struct pw_client_node_transport *transport;
 
 	struct pw_listener node_listener;
 	struct pw_listener resource_listener;
@@ -719,9 +721,9 @@ spa_proxy_node_port_reuse_buffer(struct spa_node *node, uint32_t port_id, uint32
 
 	spa_log_trace(this->log, "reuse buffer %d", buffer_id);
 	{
-		struct pw_event_transport_reuse_buffer rb = PW_EVENT_TRANSPORT_REUSE_BUFFER_INIT
-		    (impl->t->event_transport.ReuseBuffer, port_id, buffer_id);
-		pw_transport_add_event(impl->transport, (struct spa_event *) &rb);
+		struct pw_event_client_node_reuse_buffer rb = PW_EVENT_CLIENT_NODE_REUSE_BUFFER_INIT
+		    (this->type_event_client_node.ReuseBuffer, port_id, buffer_id);
+		pw_client_node_transport_add_event(impl->transport, (struct spa_event *) &rb);
 	}
 
 	return SPA_RESULT_OK;
@@ -766,8 +768,8 @@ static int spa_proxy_node_process_input(struct spa_node *node)
 		impl->transport->inputs[i] = *io;
 		io->status = SPA_RESULT_NEED_BUFFER;
 	}
-	pw_transport_add_event(impl->transport,
-			       &SPA_EVENT_INIT(impl->t->event_transport.ProcessInput));
+	pw_client_node_transport_add_event(impl->transport,
+			       &SPA_EVENT_INIT(this->type_event_client_node.ProcessInput));
 	do_flush(this);
 
 	if (this->callbacks->need_input)
@@ -800,8 +802,8 @@ static int spa_proxy_node_process_output(struct spa_node *node)
 		*io = tmp;
 		pw_log_trace("%d %d  %d", io->status, io->buffer_id, io->status);
 	}
-	pw_transport_add_event(impl->transport,
-			       &SPA_EVENT_INIT(impl->t->event_transport.ProcessOutput));
+	pw_client_node_transport_add_event(impl->transport,
+			       &SPA_EVENT_INIT(this->type_event_client_node.ProcessOutput));
 	do_flush(this);
 
 	return res;
@@ -812,7 +814,7 @@ static int handle_node_event(struct proxy *this, struct spa_event *event)
 	struct impl *impl = SPA_CONTAINER_OF(this, struct impl, proxy);
 	int i;
 
-	if (SPA_EVENT_TYPE(event) == impl->t->event_transport.HaveOutput) {
+	if (SPA_EVENT_TYPE(event) == this->type_event_client_node.HaveOutput) {
 		for (i = 0; i < MAX_OUTPUTS; i++) {
 			struct spa_port_io *io = this->out_ports[i].io;
 
@@ -823,11 +825,11 @@ static int handle_node_event(struct proxy *this, struct spa_event *event)
 			pw_log_trace("%d %d", io->status, io->buffer_id);
 		}
 		this->callbacks->have_output(this->callbacks_data);
-	} else if (SPA_EVENT_TYPE(event) == impl->t->event_transport.NeedInput) {
+	} else if (SPA_EVENT_TYPE(event) == this->type_event_client_node.NeedInput) {
 		this->callbacks->need_input(this->callbacks_data);
-	} else if (SPA_EVENT_TYPE(event) == impl->t->event_transport.ReuseBuffer) {
-		struct pw_event_transport_reuse_buffer *p =
-		    (struct pw_event_transport_reuse_buffer *) event;
+	} else if (SPA_EVENT_TYPE(event) == this->type_event_client_node.ReuseBuffer) {
+		struct pw_event_client_node_reuse_buffer *p =
+		    (struct pw_event_client_node_reuse_buffer *) event;
 		this->callbacks->reuse_buffer(this->callbacks_data, p->body.port_id.value,
 					     p->body.buffer_id.value);
 	}
@@ -935,9 +937,10 @@ static void proxy_on_data_fd_events(struct spa_source *source)
 			spa_log_warn(this->log, "proxy %p: error reading event: %s",
 					this, strerror(errno));
 
-		while (pw_transport_next_event(impl->transport, &event) == SPA_RESULT_OK) {
+		while (pw_client_node_transport_next_event(impl->transport, &event) == SPA_RESULT_OK) {
 			struct spa_event *ev = alloca(SPA_POD_SIZE(&event));
-			pw_transport_parse_event(impl->transport, ev);
+			pw_client_node_transport_parse_event(impl->transport, ev);
+                        pw_pod_remap(&ev->pod, &this->resource->client->types);;
 			handle_node_event(this, ev);
 		}
 	}
@@ -982,13 +985,22 @@ proxy_init(struct proxy *this,
 			this->log = support[i].data;
 		else if (strcmp(support[i].type, SPA_TYPE_LOOP__DataLoop) == 0)
 			this->data_loop = support[i].data;
+		else if (strcmp(support[i].type, SPA_TYPE__TypeMap) == 0)
+			this->map = support[i].data;
 	}
 	if (this->data_loop == NULL) {
 		spa_log_error(this->log, "a data-loop is needed");
+		return SPA_RESULT_ERROR;
+	}
+	if (this->map == NULL) {
+		spa_log_error(this->log, "a type map is needed");
+		return SPA_RESULT_ERROR;
 	}
 
 	this->node = proxy_node;
 
+	pw_type_event_client_node_map(this->map, &this->type_event_client_node);
+
 	this->data_source.func = proxy_on_data_fd_events;
 	this->data_source.data = this;
 	this->data_source.fd = -1;
@@ -1035,22 +1047,20 @@ static void node_initialized(void *data)
 	struct impl *impl = data;
 	struct pw_client_node *this = &impl->this;
 	struct pw_node *node = this->node;
-	struct pw_transport_info info;
 	int readfd, writefd;
 	const struct pw_node_info *i = pw_node_get_info(node);
 
 	if (this->resource == NULL)
 		return;
 
-	impl->transport = pw_transport_new(i->max_input_ports, i->max_output_ports, 0);
+	impl->transport = pw_client_node_transport_new(i->max_input_ports, i->max_output_ports);
 	impl->transport->area->n_input_ports = i->n_input_ports;
 	impl->transport->area->n_output_ports = i->n_output_ports;
 
 	client_node_get_fds(this, &readfd, &writefd);
-	pw_transport_get_info(impl->transport, &info);
 
 	pw_client_node_resource_transport(this->resource, pw_global_get_id(pw_node_get_global(node)),
-					  readfd, writefd, info.memfd, info.offset, info.size);
+					  readfd, writefd, impl->transport);
 }
 
 static int proxy_clear(struct proxy *this)
@@ -1093,7 +1103,7 @@ static void node_free(void *data)
 	proxy_clear(&impl->proxy);
 
 	if (impl->transport)
-		pw_transport_destroy(impl->transport);
+		pw_client_node_transport_destroy(impl->transport);
 
 	pw_listener_remove(&impl->node_listener);
 

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

@@ -30,6 +30,8 @@
 #include "extensions/protocol-native.h"
 #include "extensions/client-node.h"
 
+#include "transport.h"
+
 static void
 client_node_marshal_done(void *object, int seq, int res)
 {
@@ -371,8 +373,10 @@ static bool client_node_demarshal_transport(void *object, void *data, size_t siz
 {
 	struct pw_proxy *proxy = object;
 	struct spa_pod_iter it;
-	uint32_t node_id, ridx, widx, memfd_idx, offset, sz;
-	int readfd, writefd, memfd;
+	uint32_t node_id, ridx, widx, memfd_idx;
+	int readfd, writefd;
+	struct pw_client_node_transport_info info;
+	struct pw_client_node_transport *transport;
 
 	if (!spa_pod_iter_struct(&it, data, size) ||
 	    !spa_pod_iter_get(&it,
@@ -380,19 +384,21 @@ static bool client_node_demarshal_transport(void *object, void *data, size_t siz
 			      SPA_POD_TYPE_INT, &ridx,
 			      SPA_POD_TYPE_INT, &widx,
 			      SPA_POD_TYPE_INT, &memfd_idx,
-			      SPA_POD_TYPE_INT, &offset,
-			      SPA_POD_TYPE_INT, &sz, 0))
+			      SPA_POD_TYPE_INT, &info.offset,
+			      SPA_POD_TYPE_INT, &info.size, 0))
 		return false;
 
 	readfd = pw_protocol_native_get_proxy_fd(proxy, ridx);
 	writefd = pw_protocol_native_get_proxy_fd(proxy, widx);
-	memfd = pw_protocol_native_get_proxy_fd(proxy, memfd_idx);
-	if (readfd == -1 || writefd == -1 || memfd_idx == -1)
+	info.memfd = pw_protocol_native_get_proxy_fd(proxy, memfd_idx);
+
+	if (readfd == -1 || writefd == -1 || info.memfd == -1)
 		return false;
 
+	transport = pw_client_node_transport_new_from_info(&info);
+
 	pw_proxy_notify(proxy, struct pw_client_node_proxy_events, transport, node_id,
-									readfd, writefd,
-									memfd, offset, sz);
+								   readfd, writefd, transport);
 	return true;
 }
 
@@ -616,11 +622,14 @@ client_node_marshal_port_command(void *object,
 }
 
 static void client_node_marshal_transport(void *object, uint32_t node_id, int readfd, int writefd,
-					  int memfd, uint32_t offset, uint32_t size)
+					  struct pw_client_node_transport *transport)
 {
 	struct pw_resource *resource = object;
 	struct spa_pod_builder *b;
 	struct spa_pod_frame f;
+	struct pw_client_node_transport_info info;
+
+	pw_client_node_transport_get_info(transport, &info);
 
 	b = pw_protocol_native_begin_resource(resource, PW_CLIENT_NODE_PROXY_EVENT_TRANSPORT);
 
@@ -628,8 +637,9 @@ static void client_node_marshal_transport(void *object, uint32_t node_id, int re
 			       SPA_POD_TYPE_INT, node_id,
 			       SPA_POD_TYPE_INT, pw_protocol_native_add_resource_fd(resource, readfd),
 			       SPA_POD_TYPE_INT, pw_protocol_native_add_resource_fd(resource, writefd),
-			       SPA_POD_TYPE_INT, pw_protocol_native_add_resource_fd(resource, memfd),
-			       SPA_POD_TYPE_INT, offset, SPA_POD_TYPE_INT, size);
+			       SPA_POD_TYPE_INT, pw_protocol_native_add_resource_fd(resource, info.memfd),
+			       SPA_POD_TYPE_INT, info.offset,
+			       SPA_POD_TYPE_INT, info.size);
 
 	pw_protocol_native_end_resource(resource, b);
 }

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

@@ -0,0 +1,284 @@
+/* PipeWire
+ * Copyright (C) 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <unistd.h>
+#include <errno.h>
+#include <sys/mman.h>
+
+#include <pipewire/log.h>
+#include <extensions/client-node.h>
+
+#include "transport.h"
+
+/** \cond */
+
+#define INPUT_BUFFER_SIZE       (1<<12)
+#define OUTPUT_BUFFER_SIZE      (1<<12)
+
+struct transport {
+	struct pw_client_node_transport trans;
+
+	struct pw_memblock mem;
+	size_t offset;
+
+	struct spa_event current;
+	uint32_t current_index;
+};
+/** \endcond */
+
+static size_t area_get_size(struct pw_client_node_area *area)
+{
+	size_t size;
+	size = sizeof(struct pw_client_node_area);
+	size += area->max_input_ports * sizeof(struct spa_port_io);
+	size += area->max_output_ports * sizeof(struct spa_port_io);
+	size += sizeof(struct spa_ringbuffer);
+	size += INPUT_BUFFER_SIZE;
+	size += sizeof(struct spa_ringbuffer);
+	size += OUTPUT_BUFFER_SIZE;
+	return size;
+}
+
+static void transport_setup_area(void *p, struct pw_client_node_transport *trans)
+{
+	struct pw_client_node_area *a;
+
+	trans->area = a = p;
+	p = SPA_MEMBER(p, sizeof(struct pw_client_node_area), struct spa_port_io);
+
+	trans->inputs = p;
+	p = SPA_MEMBER(p, a->max_input_ports * sizeof(struct spa_port_io), void);
+
+	trans->outputs = p;
+	p = SPA_MEMBER(p, a->max_output_ports * sizeof(struct spa_port_io), void);
+
+	trans->input_buffer = p;
+	p = SPA_MEMBER(p, sizeof(struct spa_ringbuffer), void);
+
+	trans->input_data = p;
+	p = SPA_MEMBER(p, INPUT_BUFFER_SIZE, void);
+
+	trans->output_buffer = p;
+	p = SPA_MEMBER(p, sizeof(struct spa_ringbuffer), void);
+
+	trans->output_data = p;
+	p = SPA_MEMBER(p, OUTPUT_BUFFER_SIZE, void);
+}
+
+static void transport_reset_area(struct pw_client_node_transport *trans)
+{
+	int i;
+	struct pw_client_node_area *a = trans->area;
+
+	for (i = 0; i < a->max_input_ports; i++) {
+		trans->inputs[i].status = SPA_RESULT_OK;
+		trans->inputs[i].buffer_id = SPA_ID_INVALID;
+	}
+	for (i = 0; i < a->max_output_ports; i++) {
+		trans->outputs[i].status = SPA_RESULT_OK;
+		trans->outputs[i].buffer_id = SPA_ID_INVALID;
+	}
+	spa_ringbuffer_init(trans->input_buffer, INPUT_BUFFER_SIZE);
+	spa_ringbuffer_init(trans->output_buffer, OUTPUT_BUFFER_SIZE);
+}
+
+static int add_event(struct pw_client_node_transport *trans, struct spa_event *event)
+{
+	struct transport *impl = (struct transport *) trans;
+	int32_t filled, avail;
+	uint32_t size, index;
+
+	if (impl == NULL || event == NULL)
+		return SPA_RESULT_INVALID_ARGUMENTS;
+
+	filled = spa_ringbuffer_get_write_index(trans->output_buffer, &index);
+	avail = trans->output_buffer->size - filled;
+	size = SPA_POD_SIZE(event);
+	if (avail < size)
+		return SPA_RESULT_ERROR;
+
+	spa_ringbuffer_write_data(trans->output_buffer,
+				  trans->output_data,
+				  index & trans->output_buffer->mask, event, size);
+	spa_ringbuffer_write_update(trans->output_buffer, index + size);
+
+	return SPA_RESULT_OK;
+}
+
+static int next_event(struct pw_client_node_transport *trans, struct spa_event *event)
+{
+	struct transport *impl = (struct transport *) trans;
+	int32_t avail;
+
+	if (impl == NULL || event == NULL)
+		return SPA_RESULT_INVALID_ARGUMENTS;
+
+	avail = spa_ringbuffer_get_read_index(trans->input_buffer, &impl->current_index);
+	if (avail < sizeof(struct spa_event))
+		return SPA_RESULT_ENUM_END;
+
+	spa_ringbuffer_read_data(trans->input_buffer,
+				 trans->input_data,
+				 impl->current_index & trans->input_buffer->mask,
+				 &impl->current, sizeof(struct spa_event));
+
+	*event = impl->current;
+
+	return SPA_RESULT_OK;
+}
+
+static int parse_event(struct pw_client_node_transport *trans, void *event)
+{
+	struct transport *impl = (struct transport *) trans;
+	uint32_t size;
+
+	if (impl == NULL || event == NULL)
+		return SPA_RESULT_INVALID_ARGUMENTS;
+
+	size = SPA_POD_SIZE(&impl->current);
+
+	spa_ringbuffer_read_data(trans->input_buffer,
+				 trans->input_data,
+				 impl->current_index & trans->input_buffer->mask, event, size);
+	spa_ringbuffer_read_update(trans->input_buffer, impl->current_index + size);
+
+	return SPA_RESULT_OK;
+}
+
+/** Create a new transport
+ * \param max_input_ports maximum number of input_ports
+ * \param max_output_ports maximum number of output_ports
+ * \return a newly allocated \ref pw_client_node_transport
+ * \memberof pw_client_node_transport
+ */
+struct pw_client_node_transport *
+pw_client_node_transport_new(uint32_t max_input_ports, uint32_t max_output_ports)
+{
+	struct transport *impl;
+	struct pw_client_node_transport *trans;
+	struct pw_client_node_area area;
+
+	area.max_input_ports = max_input_ports;
+	area.n_input_ports = 0;
+	area.max_output_ports = max_output_ports;
+	area.n_output_ports = 0;
+
+	impl = calloc(1, sizeof(struct transport));
+	if (impl == NULL)
+		return NULL;
+
+	trans = &impl->trans;
+	impl->offset = 0;
+
+	pw_memblock_alloc(PW_MEMBLOCK_FLAG_WITH_FD |
+			  PW_MEMBLOCK_FLAG_MAP_READWRITE |
+			  PW_MEMBLOCK_FLAG_SEAL, area_get_size(&area), &impl->mem);
+
+	memcpy(impl->mem.ptr, &area, sizeof(struct pw_client_node_area));
+	transport_setup_area(impl->mem.ptr, trans);
+	transport_reset_area(trans);
+
+	trans->add_event = add_event;
+	trans->next_event = next_event;
+	trans->parse_event = parse_event;
+
+	return trans;
+}
+
+struct pw_client_node_transport *
+pw_client_node_transport_new_from_info(struct pw_client_node_transport_info *info)
+{
+	struct transport *impl;
+	struct pw_client_node_transport *trans;
+	void *tmp;
+
+	impl = calloc(1, sizeof(struct transport));
+	if (impl == NULL)
+		return NULL;
+
+	trans = &impl->trans;
+
+	impl->mem.flags = PW_MEMBLOCK_FLAG_MAP_READWRITE | PW_MEMBLOCK_FLAG_WITH_FD;
+	impl->mem.fd = info->memfd;
+	impl->mem.offset = info->offset;
+	impl->mem.size = info->size;
+	if (pw_memblock_map(&impl->mem) != SPA_RESULT_OK) {
+		pw_log_warn("transport %p: failed to map fd %d: %s", impl, info->memfd,
+			    strerror(errno));
+		goto mmap_failed;
+	}
+
+	impl->offset = info->offset;
+
+	transport_setup_area(impl->mem.ptr, trans);
+
+	tmp = trans->output_buffer;
+	trans->output_buffer = trans->input_buffer;
+	trans->input_buffer = tmp;
+
+	tmp = trans->output_data;
+	trans->output_data = trans->input_data;
+	trans->input_data = tmp;
+
+	trans->add_event = add_event;
+	trans->next_event = next_event;
+	trans->parse_event = parse_event;
+
+	return trans;
+
+      mmap_failed:
+	free(impl);
+	return NULL;
+}
+
+/** Destroy a transport
+ * \param trans a transport to destroy
+ * \memberof pw_client_node_transport
+ */
+void pw_client_node_transport_destroy(struct pw_client_node_transport *trans)
+{
+	struct transport *impl = (struct transport *) trans;
+
+	pw_log_debug("transport %p: destroy", trans);
+
+	pw_memblock_free(&impl->mem);
+	free(impl);
+}
+
+/** Get transport info
+ * \param trans the transport to get info of
+ * \param[out] info transport info
+ * \return 0 on success
+ *
+ * Fill \a info with the transport info of \a trans. This information can be
+ * passed to the client to set up the shared transport.
+ *
+ * \memberof pw_client_node_transport
+ */
+int pw_client_node_transport_get_info(struct pw_client_node_transport *trans,
+				      struct pw_client_node_transport_info *info)
+{
+	struct transport *impl = (struct transport *) trans;
+
+	info->memfd = impl->mem.fd;
+	info->offset = impl->offset;
+	info->size = impl->mem.size;
+
+	return SPA_RESULT_OK;
+}

src/modules/module-client-node/transport.h

@@ -0,0 +1,58 @@
+/* PipeWire
+ * Copyright (C) 2016 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __PIPEWIRE_CLIENT_NODE_TRANSPORT_H__
+#define __PIPEWIRE_CLIENT_NODE_TRANSPORT_H__
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <string.h>
+
+#include <spa/defs.h>
+#include <spa/node.h>
+
+#include <pipewire/mem.h>
+
+/** information about the transport region \memberof pw_client_node */
+struct pw_client_node_transport_info {
+	int memfd;		/**< the memfd of the transport area */
+	uint32_t offset;	/**< offset to map \a memfd at */
+	uint32_t size;		/**< size of memfd mapping */
+};
+
+struct pw_client_node_transport *
+pw_client_node_transport_new(uint32_t max_input_ports, uint32_t max_output_ports);
+
+struct pw_client_node_transport *
+pw_client_node_transport_new_from_info(struct pw_client_node_transport_info *info);
+
+void
+pw_client_node_transport_destroy(struct pw_client_node_transport *trans);
+
+int
+pw_client_node_transport_get_info(struct pw_client_node_transport *trans,
+				  struct pw_client_node_transport_info *info);
+
+#ifdef __cplusplus
+} /* extern "C" */
+#endif
+
+#endif /* __PIPEWIRE_CLIENT_NODE_TRANSPORT_H__ */