impl-node: move signalfd into pw_node_target

We don't need an extra signal_func, we can write to the signalfd
directly.

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

@@ -889,12 +889,14 @@ static int impl_node_process(void *object)
 	/* this should not be called, we call the exported node
 	 * directly */
 	spa_log_warn(impl->log, "exported node activation");
-	if (SPA_UNLIKELY(spa_system_clock_gettime(impl->data_system, CLOCK_MONOTONIC, &ts) < 0))
+	spa_system_clock_gettime(impl->data_system, CLOCK_MONOTONIC, &ts);
-		spa_zero(ts);
 	n->rt.activation->status = PW_NODE_ACTIVATION_TRIGGERED;
 	n->rt.activation->signal_time = SPA_TIMESPEC_TO_NSEC(&ts);
 
-	return n->rt.target.signal_func(n->rt.target.data);
+	if (SPA_UNLIKELY(spa_system_eventfd_write(n->rt.target.system, n->rt.target.fd, 1) < 0))
+		pw_log_warn("%p: write failed %m", impl);
+
+	return SPA_STATUS_OK;
 }
 
 static struct pw_node *

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

@@ -82,7 +82,6 @@ struct link {
 	struct pw_memmap *map;
 	struct pw_node_target target;
 	uint32_t node_id;
-	int signalfd;
 };
 
 /** \endcond */
@@ -114,7 +113,7 @@ static void clear_link(struct node_data *data, struct link *link)
 	pw_loop_invoke(data->data_loop,
 		do_deactivate_link, SPA_ID_INVALID, NULL, 0, true, link);
 	pw_memmap_free(link->map);
-	spa_system_close(data->data_system, link->signalfd);
+	spa_system_close(link->target.system, link->target.fd);
 	spa_list_remove(&link->link);
 	free(link);
 }
@@ -852,18 +851,6 @@ exit:
 	return res;
 }
 
-static int link_signal_func(void *user_data)
-{
-	struct link *link = user_data;
-	struct spa_system *data_system = link->data->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)
@@ -922,10 +909,8 @@ client_node_set_activation(void *_data,
 		link->node_id = node_id;
 		link->map = mm;
 		link->target.activation = ptr;
-		link->signalfd = signalfd;
+		link->target.system = data->data_system;
-		link->target.signal_func = link_signal_func;
+		link->target.fd = signalfd;
-		link->target.data = link;
-		link->target.node = NULL;
 		spa_list_append(&data->links, &link->link);
 
 		pw_loop_invoke(data->data_loop,

src/pipewire/impl-link.c

@@ -72,8 +72,8 @@ static struct pw_node_peer *pw_node_peer_ref(struct pw_impl_node *onode, struct
 	peer->active_count = 0;
 	peer->target.node = inode;
 	peer->target.activation = inode->rt.activation;
-	peer->target.signal_func = inode->rt.target.signal_func;
+	peer->target.system = inode->data_system;
-	peer->target.data = inode->rt.target.data;
+	peer->target.fd = inode->source.fd;
 
 	spa_list_append(&onode->peer_list, &peer->link);
 	pw_log_debug("new peer %p from %p to %p", peer, onode, inode);

src/pipewire/impl-node.c

@@ -81,7 +81,8 @@ static void add_node(struct pw_impl_node *this, struct pw_impl_node *driver)
 	/* signal the driver */
 	this->rt.driver_target.activation = driver->rt.activation;
 	this->rt.driver_target.node = driver;
-	this->rt.driver_target.data = driver;
+	this->rt.driver_target.system = driver->data_system;
+	this->rt.driver_target.fd = driver->source.fd;
 	spa_list_append(&this->rt.target_list, &this->rt.driver_target.link);
 
 	spa_list_append(&driver->rt.target_list, &this->rt.target.link);
@@ -112,7 +113,7 @@ static void remove_node(struct pw_impl_node *this)
 		return;
 
 	pw_log_trace("%p: remove from driver %p %p %p",
-			this, this->rt.driver_target.data,
+			this, this->rt.driver_target.node,
 			this->rt.driver_target.activation, this->rt.activation);
 
 	spa_list_remove(&this->rt.target.link);
@@ -146,12 +147,11 @@ do_node_add(struct spa_loop *loop, bool async, uint32_t seq,
 	struct pw_impl_node *driver = this->driver_node;
 
 	if (!this->added) {
-		struct spa_system *data_system = this->data_system;
 		uint64_t dummy;
 		int res;
 
 		/* clear the eventfd in case it was written to while the node was stopped */
-		res = spa_system_eventfd_read(data_system, this->source.fd, &dummy);
+		res = spa_system_eventfd_read(this->data_system, this->source.fd, &dummy);
 		if (SPA_UNLIKELY(res != -EAGAIN && res != 0))
 			pw_log_warn("%p: read failed %m", this);
 
@@ -1097,6 +1097,13 @@ static inline uint64_t get_time_ns(struct spa_system *system)
 	return SPA_TIMESPEC_TO_NSEC(&ts);
 }
 
+static inline void node_signal(struct pw_impl_node *this)
+{
+	pw_log_trace_fp("node %p %s", this, this->name);
+	if (SPA_UNLIKELY(spa_system_eventfd_write(this->data_system, this->source.fd, 1) < 0))
+		pw_log_warn("node %p: write failed %m", this);
+}
+
 static inline int resume_node(struct pw_impl_node *this, int status, uint64_t nsec)
 {
 	struct pw_node_target *t;
@@ -1113,24 +1120,13 @@ static inline int resume_node(struct pw_impl_node *this, int status, uint64_t ns
 		if (pw_node_activation_state_dec(state, 1)) {
 			a->status = PW_NODE_ACTIVATION_TRIGGERED;
 			a->signal_time = nsec;
-			t->signal_func(t->data);
+			if (SPA_UNLIKELY(spa_system_eventfd_write(t->system, t->fd, 1) < 0))
+				pw_log_warn("node %p: write failed %m", this);
 		}
 	}
 	return 0;
 }
 
-static inline int node_signal_func(void *data)
-{
-	struct pw_impl_node *this = data;
-	struct spa_system *data_system = this->data_system;
-
-	pw_log_trace_fp("node %p %s", this, this->name);
-	if (SPA_UNLIKELY(spa_system_eventfd_write(data_system, this->source.fd, 1) < 0))
-		pw_log_warn("node %p: write failed %m", this);
-
-	return SPA_STATUS_OK;
-}
-
 static inline void calculate_stats(struct pw_impl_node *this,  struct pw_node_activation *a)
 {
 	uint64_t signal_time = a->signal_time;
@@ -1207,7 +1203,7 @@ int pw_impl_node_trigger(struct pw_impl_node *node)
 		uint64_t nsec = get_time_ns(node->data_system);
 		a->status = PW_NODE_ACTIVATION_TRIGGERED;
 		a->signal_time = nsec;
-		node_signal_func(node);
+		node_signal(node);
 	}
 	return 0;
 }
@@ -1215,7 +1211,6 @@ int pw_impl_node_trigger(struct pw_impl_node *node)
 static void node_on_fd_events(struct spa_source *source)
 {
 	struct pw_impl_node *this = source->data;
-	struct spa_system *data_system = this->data_system;
 
 	if (SPA_UNLIKELY(source->rmask & (SPA_IO_ERR | SPA_IO_HUP))) {
 		pw_log_warn("%p: got socket error %08x", this, source->rmask);
@@ -1224,7 +1219,7 @@ static void node_on_fd_events(struct spa_source *source)
 	if (SPA_LIKELY(source->rmask & SPA_IO_IN)) {
 		uint64_t cmd;
 
-		if (SPA_UNLIKELY(spa_system_eventfd_read(data_system, this->source.fd, &cmd) < 0))
+		if (SPA_UNLIKELY(spa_system_eventfd_read(this->data_system, this->source.fd, &cmd) < 0))
 			pw_log_warn("%p: read failed %m", this);
 		else if (SPA_UNLIKELY(cmd > 1))
 			pw_log_info("(%s-%u) client missed %"PRIu64" wakeups",
@@ -1351,9 +1346,8 @@ struct pw_impl_node *pw_context_create_node(struct pw_context *context,
 	this->rt.activation = this->activation->map->ptr;
 	this->rt.target.activation = this->rt.activation;
 	this->rt.target.node = this;
-	this->rt.target.signal_func = node_signal_func;
+	this->rt.target.system = this->data_system;
-	this->rt.target.data = this;
+	this->rt.target.fd = this->source.fd;
-	this->rt.driver_target.signal_func = node_signal_func;
 
 	reset_position(this, &this->rt.activation->position);
 	this->rt.activation->sync_timeout = DEFAULT_SYNC_TIMEOUT;
@@ -1718,7 +1712,7 @@ static int node_ready(void *data, int status)
 						state->pending, state->required);
 				dump_states(node);
 			}
-			node_signal_func(node);
+			node_signal(node);
 		} else {
 			uint64_t signal_time = a->signal_time;
 			/* old nodes set the TRIGGERED status on node_ready, patch this

src/pipewire/private.h

@@ -586,8 +586,8 @@ struct pw_node_target {
 	struct spa_list link;
 	struct pw_impl_node *node;
 	struct pw_node_activation *activation;
-	int (*signal_func) (void *data);
+	struct spa_system *system;
-	void *data;
+	int fd;
 	unsigned int active:1;
 };