[PATCH BlueZ 3/5] gatt: Switch from pipe2 to sockepair for Acquire*

[Luiz Augusto von Dentz <luiz.dentz@gmail.com>]

From: Luiz Augusto von Dentz <luiz.von.dentz@intel.com>

This enables to use sendmsg with MSG_NOSIGNAL.
---
 src/gatt-client.c   | 99 ++++++++++++++++++++++++++-------------------
 src/gatt-database.c | 26 +++++++-----
 2 files changed, 72 insertions(+), 53 deletions(-)

diff --git a/src/gatt-client.c b/src/gatt-client.c
index 55aa5e423..45ed3b170 100644
--- a/src/gatt-client.c
+++ b/src/gatt-client.c
@@ -90,7 +90,7 @@ struct async_dbus_op {
 	async_dbus_op_complete_t complete;
 };
 
-struct pipe_io {
+struct sock_io {
 	DBusMessage *msg;
 	struct io *io;
 	void (*destroy)(void *data);
@@ -109,8 +109,8 @@ struct characteristic {
 	char *path;
 
 	unsigned int ready_id;
-	struct pipe_io *write_io;
-	struct pipe_io *notify_io;
+	struct sock_io *write_io;
+	struct sock_io *notify_io;
 
 	struct async_dbus_op *read_op;
 	struct async_dbus_op *write_op;
@@ -1054,19 +1054,30 @@ fail:
 	return btd_error_not_supported(msg);
 }
 
-static bool chrc_pipe_read(struct io *io, void *user_data)
+static bool sock_read(struct io *io, void *user_data)
 {
 	struct characteristic *chrc = user_data;
 	struct bt_gatt_client *gatt = chrc->service->client->gatt;
+	struct msghdr msg;
 	uint8_t buf[512];
+	struct iovec iov;
 	int fd = io_get_fd(io);
 	ssize_t bytes_read;
 
-	bytes_read = read(fd, buf, sizeof(buf));
-	if (bytes_read < 0)
+	iov.iov_base = buf;
+	iov.iov_len = sizeof(buf);
+
+	memset(&msg, 0, sizeof(msg));
+	msg.msg_iov = &iov;
+	msg.msg_iovlen = 1;
+
+	bytes_read = recvmsg(fd, &msg, MSG_DONTWAIT);
+	if (bytes_read < 0) {
+		error("recvmsg: %s", strerror(errno));
 		return false;
+	}
 
-	if (!gatt)
+	if (!gatt || bytes_read == 0)
 		return false;
 
 	bt_gatt_client_write_without_response(gatt, chrc->value_handle,
@@ -1076,7 +1087,7 @@ static bool chrc_pipe_read(struct io *io, void *user_data)
 	return true;
 }
 
-static void pipe_io_destroy(struct pipe_io *io)
+static void sock_io_destroy(struct sock_io *io)
 {
 	if (io->destroy)
 		io->destroy(io->data);
@@ -1088,20 +1099,19 @@ static void pipe_io_destroy(struct pipe_io *io)
 	free(io);
 }
 
-static void characteristic_destroy_pipe(struct characteristic *chrc,
-							struct io *io)
+static void destroy_sock(struct characteristic *chrc, struct io *io)
 {
 	queue_remove(chrc->service->client->ios, io);
 
 	if (chrc->write_io && io == chrc->write_io->io) {
-		pipe_io_destroy(chrc->write_io);
+		sock_io_destroy(chrc->write_io);
 		chrc->write_io = NULL;
 		g_dbus_emit_property_changed(btd_get_dbus_connection(),
 						chrc->path,
 						GATT_CHARACTERISTIC_IFACE,
 						"WriteAcquired");
 	} else if (chrc->notify_io) {
-		pipe_io_destroy(chrc->notify_io);
+		sock_io_destroy(chrc->notify_io);
 		chrc->notify_io = NULL;
 		g_dbus_emit_property_changed(btd_get_dbus_connection(),
 						chrc->path,
@@ -1110,22 +1120,21 @@ static void characteristic_destroy_pipe(struct characteristic *chrc,
 	}
 }
 
-static bool characteristic_pipe_hup(struct io *io, void *user_data)
+static bool sock_hup(struct io *io, void *user_data)
 {
 	struct characteristic *chrc = user_data;
 
 	DBG("%s: io %p", chrc->path, io);
 
-	characteristic_destroy_pipe(chrc, io);
+	destroy_sock(chrc, io);
 
 	return false;
 }
 
-static DBusMessage *characteristic_create_pipe(struct characteristic *chrc,
-						DBusMessage *msg)
+static DBusMessage *create_sock(struct characteristic *chrc, DBusMessage *msg)
 {
 	struct bt_gatt_client *gatt = chrc->service->client->gatt;
-	int pipefd[2];
+	int fds[2];
 	struct io *io;
 	bool dir;
 	uint16_t mtu;
@@ -1134,33 +1143,34 @@ static DBusMessage *characteristic_create_pipe(struct characteristic *chrc,
 	if (!gatt || !bt_gatt_client_is_ready(gatt))
 		return btd_error_failed(msg, "Not connected");
 
-	if (pipe2(pipefd, O_DIRECT | O_NONBLOCK | O_CLOEXEC) < 0)
+	if (socketpair(AF_LOCAL, SOCK_SEQPACKET | SOCK_NONBLOCK | SOCK_CLOEXEC,
+								0, fds) < 0)
 		return btd_error_failed(msg, strerror(errno));
 
 	dir = dbus_message_has_member(msg, "AcquireWrite");
 
-	io = io_new(pipefd[!dir]);
+	io = io_new(fds[!dir]);
 	if (!io) {
-		close(pipefd[0]);
-		close(pipefd[1]);
+		close(fds[0]);
+		close(fds[1]);
 		return btd_error_failed(msg, strerror(EIO));
 	}
 
 	io_set_close_on_destroy(io, true);
 
-	if (!io_set_read_handler(io, chrc_pipe_read, chrc, NULL))
+	if (!io_set_read_handler(io, sock_read, chrc, NULL))
 		goto fail;
 
-	if (!io_set_disconnect_handler(io, characteristic_pipe_hup, chrc, NULL))
+	if (!io_set_disconnect_handler(io, sock_hup, chrc, NULL))
 		goto fail;
 
 	mtu = bt_gatt_client_get_mtu(gatt);
 
-	reply = g_dbus_create_reply(msg, DBUS_TYPE_UNIX_FD, &pipefd[dir],
+	reply = g_dbus_create_reply(msg, DBUS_TYPE_UNIX_FD, &fds[dir],
 					DBUS_TYPE_UINT16, &mtu,
 					DBUS_TYPE_INVALID);
 
-	close(pipefd[dir]);
+	close(fds[dir]);
 
 	if (dir) {
 		chrc->write_io->io = io;
@@ -1185,7 +1195,7 @@ static DBusMessage *characteristic_create_pipe(struct characteristic *chrc,
 
 fail:
 	io_destroy(io);
-	close(pipefd[dir]);
+	close(fds[dir]);
 	return btd_error_failed(msg, strerror(EIO));
 }
 
@@ -1197,7 +1207,7 @@ static void characteristic_ready(bool success, uint8_t ecode, void *user_data)
 	chrc->ready_id = 0;
 
 	if (chrc->write_io && chrc->write_io->msg) {
-		reply = characteristic_create_pipe(chrc, chrc->write_io->msg);
+		reply = create_sock(chrc, chrc->write_io->msg);
 
 		g_dbus_send_message(btd_get_dbus_connection(), reply);
 
@@ -1206,7 +1216,7 @@ static void characteristic_ready(bool success, uint8_t ecode, void *user_data)
 	}
 
 	if (chrc->notify_io && chrc->notify_io->msg) {
-		reply = characteristic_create_pipe(chrc, chrc->notify_io->msg);
+		reply = create_sock(chrc, chrc->notify_io->msg);
 
 		g_dbus_send_message(btd_get_dbus_connection(), reply);
 
@@ -1230,7 +1240,7 @@ static DBusMessage *characteristic_acquire_write(DBusConnection *conn,
 	if (!(chrc->props & BT_GATT_CHRC_PROP_WRITE_WITHOUT_RESP))
 		return btd_error_not_supported(msg);
 
-	chrc->write_io = new0(struct pipe_io, 1);
+	chrc->write_io = new0(struct sock_io, 1);
 
 	if (!bt_gatt_client_is_ready(gatt)) {
 		/* GATT not ready, wait until it becomes ready */
@@ -1242,7 +1252,7 @@ static DBusMessage *characteristic_acquire_write(DBusConnection *conn,
 		return NULL;
 	}
 
-	return characteristic_create_pipe(chrc, msg);
+	return create_sock(chrc, msg);
 }
 
 struct notify_client {
@@ -1414,21 +1424,26 @@ static void register_notify_cb(uint16_t att_ecode, void *user_data)
 static void notify_io_cb(uint16_t value_handle, const uint8_t *value,
 					uint16_t length, void *user_data)
 {
+	struct msghdr msg;
 	struct iovec iov;
 	struct notify_client *client = user_data;
 	struct characteristic *chrc = client->chrc;
 	int err;
 
-	/* Drop notification if the pipe is not ready */
-	if (!chrc->notify_io->io)
+	/* Drop notification if the sock is not ready */
+	if (!chrc->notify_io || !chrc->notify_io->io)
 		return;
 
 	iov.iov_base = (void *) value;
 	iov.iov_len = length;
 
-	err = io_send(chrc->notify_io->io, &iov, 1);
+	memset(&msg, 0, sizeof(msg));
+	msg.msg_iov = &iov;
+	msg.msg_iovlen = 1;
+
+	err = sendmsg(io_get_fd(chrc->notify_io->io), &msg, MSG_NOSIGNAL);
 	if (err < 0)
-		error("io_send: %s", strerror(-err));
+		error("sendmsg: %s", strerror(errno));
 }
 
 static void register_notify_io_cb(uint16_t att_ecode, void *user_data)
@@ -1499,7 +1514,7 @@ static DBusMessage *characteristic_acquire_notify(DBusConnection *conn,
 
 	queue_push_tail(chrc->notify_clients, client);
 
-	chrc->notify_io = new0(struct pipe_io, 1);
+	chrc->notify_io = new0(struct sock_io, 1);
 	chrc->notify_io->data = client;
 	chrc->notify_io->msg = dbus_message_ref(msg);
 	chrc->notify_io->destroy = notify_io_destroy;
@@ -1586,16 +1601,16 @@ static DBusMessage *characteristic_stop_notify(DBusConnection *conn,
 	const char *sender = dbus_message_get_sender(msg);
 	struct notify_client *client;
 
+	if (chrc->notify_io) {
+		destroy_sock(chrc, chrc->notify_io->io);
+		return dbus_message_new_method_return(msg);
+	}
+
 	client = queue_remove_if(chrc->notify_clients, match_notify_sender,
 							(void *) sender);
 	if (!client)
 		return btd_error_failed(msg, "No notify session started");
 
-	if (chrc->notify_io) {
-		characteristic_destroy_pipe(chrc, chrc->notify_io->io);
-		return dbus_message_new_method_return(msg);
-	}
-
 	queue_remove(chrc->service->client->all_notify_clients, client);
 	bt_gatt_client_unregister_notify(gatt, client->notify_id);
 	update_notifying(chrc);
@@ -1664,12 +1679,12 @@ static void characteristic_free(void *data)
 
 	if (chrc->write_io) {
 		queue_remove(chrc->service->client->ios, chrc->write_io->io);
-		pipe_io_destroy(chrc->write_io);
+		sock_io_destroy(chrc->write_io);
 	}
 
 	if (chrc->notify_io) {
 		queue_remove(chrc->service->client->ios, chrc->notify_io->io);
-		pipe_io_destroy(chrc->notify_io);
+		sock_io_destroy(chrc->notify_io);
 	}
 
 	queue_destroy(chrc->notify_clients, remove_client);
diff --git a/src/gatt-database.c b/src/gatt-database.c
index ec584fc3c..c31afa0e9 100644
--- a/src/gatt-database.c
+++ b/src/gatt-database.c
@@ -2139,7 +2139,7 @@ static struct pending_op *send_write(struct btd_device *device,
 	return NULL;
 }
 
-static bool pipe_hup(struct io *io, void *user_data)
+static bool sock_hup(struct io *io, void *user_data)
 {
 	struct external_chrc *chrc = user_data;
 
@@ -2155,7 +2155,7 @@ static bool pipe_hup(struct io *io, void *user_data)
 	return false;
 }
 
-static bool pipe_io_read(struct io *io, void *user_data)
+static bool sock_io_read(struct io *io, void *user_data)
 {
 	struct external_chrc *chrc = user_data;
 	uint8_t buf[512];
@@ -2176,7 +2176,7 @@ static bool pipe_io_read(struct io *io, void *user_data)
 	return true;
 }
 
-static struct io *pipe_io_new(int fd, void *user_data)
+static struct io *sock_io_new(int fd, void *user_data)
 {
 	struct io *io;
 
@@ -2184,21 +2184,25 @@ static struct io *pipe_io_new(int fd, void *user_data)
 
 	io_set_close_on_destroy(io, true);
 
-	io_set_read_handler(io, pipe_io_read, user_data, NULL);
+	io_set_read_handler(io, sock_io_read, user_data, NULL);
 
-	io_set_disconnect_handler(io, pipe_hup, user_data, NULL);
+	io_set_disconnect_handler(io, sock_hup, user_data, NULL);
 
 	return io;
 }
 
-static int pipe_io_send(struct io *io, const void *data, size_t len)
+static int sock_io_send(struct io *io, const void *data, size_t len)
 {
+	struct msghdr msg;
 	struct iovec iov;
 
 	iov.iov_base = (void *) data;
 	iov.iov_len = len;
 
-	return io_send(io, &iov, 1);
+	memset(&msg, 0, sizeof(msg));
+	msg.msg_iov = &iov;
+
+	return sendmsg(io_get_fd(io), &msg, MSG_NOSIGNAL);
 }
 
 static void acquire_write_reply(DBusMessage *message, void *user_data)
@@ -2227,9 +2231,9 @@ static void acquire_write_reply(DBusMessage *message, void *user_data)
 
 	DBG("AcquireWrite success: fd %d MTU %u\n", fd, mtu);
 
-	chrc->write_io = pipe_io_new(fd, chrc);
+	chrc->write_io = sock_io_new(fd, chrc);
 
-	if (pipe_io_send(chrc->write_io, op->data.iov_base,
+	if (sock_io_send(chrc->write_io, op->data.iov_base,
 				op->data.iov_len) < 0)
 		goto retry;
 
@@ -2308,7 +2312,7 @@ static void acquire_notify_reply(DBusMessage *message, void *user_data)
 
 	DBG("AcquireNotify success: fd %d MTU %u\n", fd, mtu);
 
-	chrc->notify_io = pipe_io_new(fd, chrc);
+	chrc->notify_io = sock_io_new(fd, chrc);
 
 	__sync_fetch_and_add(&chrc->ntfy_cnt, 1);
 
@@ -2685,7 +2689,7 @@ static void chrc_write_cb(struct gatt_db_attribute *attrib,
 		chrc->prep_authorized = false;
 
 	if (chrc->write_io) {
-		if (pipe_io_send(chrc->write_io, value, len) < 0) {
+		if (sock_io_send(chrc->write_io, value, len) < 0) {
 			error("Unable to write: %s", strerror(errno));
 			goto fail;
 		}
-- 
2.17.2