[lttng-tools.git] / src / lib / lttng-ctl / channel.cpp

/*
 * Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
 *
 * SPDX-License-Identifier: LGPL-2.1-only
 *
 */

#include <lttng/notification/notification-internal.h>
#include <lttng/notification/channel-internal.h>
#include <lttng/condition/condition-internal.h>
#include <lttng/endpoint.h>
#include <common/defaults.h>
#include <common/error.h>
#include <common/dynamic-buffer.h>
#include <common/utils.h>
#include <common/defaults.h>
#include <common/payload.h>
#include <common/payload-view.h>
#include <common/unix.h>
#include "lttng-ctl-helper.h"
#include <common/compat/poll.h>

static
int handshake(struct lttng_notification_channel *channel);

/*
 * Populates the reception buffer with the next complete message.
 * The caller must acquire the channel's lock.
 */
static
int receive_message(struct lttng_notification_channel *channel)
{
	ssize_t ret;
	struct lttng_notification_channel_message msg;

	lttng_payload_clear(&channel->reception_payload);

	ret = lttcomm_recv_unix_sock(channel->socket, &msg, sizeof(msg));
	if (ret <= 0) {
		ret = -1;
		goto error;
	}

	if (msg.size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
		ret = -1;
		goto error;
	}

	/* Add message header at buffer's start. */
	ret = lttng_dynamic_buffer_append(&channel->reception_payload.buffer, &msg,
			sizeof(msg));
	if (ret) {
		goto error;
	}

	/* Reserve space for the payload. */
	ret = lttng_dynamic_buffer_set_size(&channel->reception_payload.buffer,
			channel->reception_payload.buffer.size + msg.size);
	if (ret) {
		goto error;
	}

	/* Receive message payload. */
	ret = lttcomm_recv_unix_sock(channel->socket,
			channel->reception_payload.buffer.data + sizeof(msg), msg.size);
	if (ret < (ssize_t) msg.size) {
		ret = -1;
		goto error;
	}

	/* Receive message fds. */
	if (msg.fds != 0) {
		ret = lttcomm_recv_payload_fds_unix_sock(channel->socket,
				msg.fds, &channel->reception_payload);
		if (ret < sizeof(int) * msg.fds) {
			ret = -1;
			goto error;
		}
	}
	ret = 0;
end:
	return ret;
error:
	lttng_payload_clear(&channel->reception_payload);
	goto end;
}

static
enum lttng_notification_channel_message_type get_current_message_type(
		struct lttng_notification_channel *channel)
{
	struct lttng_notification_channel_message *msg;

	LTTNG_ASSERT(channel->reception_payload.buffer.size >= sizeof(*msg));

	msg = (struct lttng_notification_channel_message *)
			channel->reception_payload.buffer.data;
	return (enum lttng_notification_channel_message_type) msg->type;
}

static
struct lttng_notification *create_notification_from_current_message(
		struct lttng_notification_channel *channel)
{
	ssize_t ret;
	struct lttng_notification *notification = NULL;

	if (channel->reception_payload.buffer.size <=
			sizeof(struct lttng_notification_channel_message)) {
		goto end;
	}

	{
		struct lttng_payload_view view = lttng_payload_view_from_payload(
				&channel->reception_payload,
				sizeof(struct lttng_notification_channel_message),
				-1);

		ret = lttng_notification_create_from_payload(
				&view, &notification);
	}

	if (ret != channel->reception_payload.buffer.size -
			sizeof(struct lttng_notification_channel_message)) {
		lttng_notification_destroy(notification);
		notification = NULL;
		goto end;
	}
end:
	return notification;
}

struct lttng_notification_channel *lttng_notification_channel_create(
		struct lttng_endpoint *endpoint)
{
	int fd, ret;
	bool is_in_tracing_group = false, is_root = false;
	char *sock_path = NULL;
	struct lttng_notification_channel *channel = NULL;

	if (!endpoint ||
			endpoint != lttng_session_daemon_notification_endpoint) {
		goto end;
	}

	sock_path = (char *) zmalloc(LTTNG_PATH_MAX);
	if (!sock_path) {
		goto end;
	}

	channel = (lttng_notification_channel *) zmalloc(sizeof(struct lttng_notification_channel));
	if (!channel) {
		goto end;
	}
	channel->socket = -1;
	pthread_mutex_init(&channel->lock, NULL);
	lttng_payload_init(&channel->reception_payload);
	CDS_INIT_LIST_HEAD(&channel->pending_notifications.list);

	is_root = (getuid() == 0);
	if (!is_root) {
		is_in_tracing_group = lttng_check_tracing_group();
	}

	if (is_root || is_in_tracing_group) {
		ret = lttng_strncpy(sock_path,
				DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK,
				LTTNG_PATH_MAX);
		if (ret) {
			ret = -LTTNG_ERR_INVALID;
			goto error;
		}

		ret = lttcomm_connect_unix_sock(sock_path);
		if (ret >= 0) {
			fd = ret;
			goto set_fd;
		}
	}

	/* Fallback to local session daemon. */
	ret = snprintf(sock_path, LTTNG_PATH_MAX,
			DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
			utils_get_home_dir());
	if (ret < 0 || ret >= LTTNG_PATH_MAX) {
		goto error;
	}

	ret = lttcomm_connect_unix_sock(sock_path);
	if (ret < 0) {
		goto error;
	}
	fd = ret;

set_fd:
	channel->socket = fd;

	ret = handshake(channel);
	if (ret) {
		goto error;
	}
end:
	free(sock_path);
	return channel;
error:
	lttng_notification_channel_destroy(channel);
	channel = NULL;
	goto end;
}

enum lttng_notification_channel_status
lttng_notification_channel_get_next_notification(
		struct lttng_notification_channel *channel,
		struct lttng_notification **_notification)
{
	int ret;
	struct lttng_notification *notification = NULL;
	enum lttng_notification_channel_status status =
			LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	struct lttng_poll_event events;

	if (!channel || !_notification) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
		goto end;
	}

	pthread_mutex_lock(&channel->lock);

	if (channel->pending_notifications.count) {
		struct pending_notification *pending_notification;

		LTTNG_ASSERT(!cds_list_empty(&channel->pending_notifications.list));

		/* Deliver one of the pending notifications. */
		pending_notification = cds_list_first_entry(
				&channel->pending_notifications.list,
				struct pending_notification,
				node);
		notification = pending_notification->notification;
		if (!notification) {
			status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
		}
		cds_list_del(&pending_notification->node);
		channel->pending_notifications.count--;
		free(pending_notification);
		goto end_unlock;
	}

	/*
	 * Block on interruptible epoll/poll() instead of the message reception
	 * itself as the recvmsg() wrappers always restart on EINTR. We choose
	 * to wait using interruptible epoll/poll() in order to:
	 *   1) Return if a signal occurs,
	 *   2) Not deal with partially received messages.
	 *
	 * The drawback to this approach is that we assume that messages
	 * are complete/well formed. If a message is shorter than its
	 * announced length, receive_message() will block on recvmsg()
	 * and never return (even if a signal is received).
	 */
	ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
	if (ret < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_unlock;
	}
	ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
	if (ret < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}
	ret = lttng_poll_wait_interruptible(&events, -1);
	if (ret <= 0) {
		status = (ret == -1 && errno == EINTR) ?
			LTTNG_NOTIFICATION_CHANNEL_STATUS_INTERRUPTED :
			LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}

	ret = receive_message(channel);
	if (ret) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}

	switch (get_current_message_type(channel)) {
	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
		notification = create_notification_from_current_message(
				channel);
		if (!notification) {
			status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
			goto end_clean_poll;
		}
		break;
	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
		/* No payload to consume. */
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
		break;
	default:
		/* Protocol error. */
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}

end_clean_poll:
	lttng_poll_clean(&events);
end_unlock:
	pthread_mutex_unlock(&channel->lock);
	*_notification = notification;
end:
	return status;
}

static
int enqueue_dropped_notification(
		struct lttng_notification_channel *channel)
{
	int ret = 0;
	struct pending_notification *pending_notification;
	struct cds_list_head *last_element =
			channel->pending_notifications.list.prev;

	pending_notification = caa_container_of(last_element,
			struct pending_notification, node);
	if (!pending_notification->notification) {
		/*
		 * The last enqueued notification indicates dropped
		 * notifications; there is nothing to do as we group
		 * dropped notifications together.
		 */
		goto end;
	}

	if (channel->pending_notifications.count >=
			DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT &&
			pending_notification->notification) {
		/*
		 * Discard the last enqueued notification to indicate
		 * that notifications were dropped at this point.
		 */
		lttng_notification_destroy(
				pending_notification->notification);
		pending_notification->notification = NULL;
		goto end;
	}

	pending_notification = (struct pending_notification *) zmalloc(sizeof(*pending_notification));
	if (!pending_notification) {
		ret = -1;
		goto end;
	}
	CDS_INIT_LIST_HEAD(&pending_notification->node);
	cds_list_add(&pending_notification->node,
			&channel->pending_notifications.list);
	channel->pending_notifications.count++;
end:
	return ret;
}

static
int enqueue_notification_from_current_message(
		struct lttng_notification_channel *channel)
{
	int ret = 0;
	struct lttng_notification *notification;
	struct pending_notification *pending_notification;

	if (channel->pending_notifications.count >=
			DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT) {
		/* Drop the notification. */
		ret = enqueue_dropped_notification(channel);
		goto end;
	}

	pending_notification = (struct pending_notification *) zmalloc(sizeof(*pending_notification));
	if (!pending_notification) {
		ret = -1;
		goto error;
	}
	CDS_INIT_LIST_HEAD(&pending_notification->node);

	notification = create_notification_from_current_message(channel);
	if (!notification) {
		ret = -1;
		goto error;
	}

	pending_notification->notification = notification;
	cds_list_add(&pending_notification->node,
			&channel->pending_notifications.list);
	channel->pending_notifications.count++;
end:
	return ret;
error:
	free(pending_notification);
	goto end;
}

enum lttng_notification_channel_status
lttng_notification_channel_has_pending_notification(
		struct lttng_notification_channel *channel,
		bool *_notification_pending)
{
	int ret;
	enum lttng_notification_channel_status status =
			LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	struct lttng_poll_event events;

	if (!channel || !_notification_pending) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
		goto end;
	}

	pthread_mutex_lock(&channel->lock);

	if (channel->pending_notifications.count) {
		*_notification_pending = true;
		goto end_unlock;
	}

	if (channel->socket < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED;
		goto end_unlock;
	}

	/*
	 * Check, without blocking, if data is available on the channel's
	 * socket. If there is data available, it is safe to read (blocking)
	 * on the socket for a message from the session daemon.
	 *
	 * Since all commands wait for the session daemon's reply before
	 * releasing the channel's lock, the protocol only allows for
	 * notifications and "notification dropped" messages to come
	 * through. If we receive a different message type, it is
	 * considered a protocol error.
	 *
	 * Note that this function is not guaranteed not to block. This
	 * will block until our peer (the session daemon) has sent a complete
	 * message if we see data available on the socket. If the peer does
	 * not respect the protocol, this may block indefinitely.
	 */
	ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
	if (ret < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_unlock;
	}
	ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
	if (ret < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}
	/* timeout = 0: return immediately. */
	ret = lttng_poll_wait_interruptible(&events, 0);
	if (ret == 0) {
		/* No data available. */
		*_notification_pending = false;
		goto end_clean_poll;
	} else if (ret < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}

	/* Data available on socket. */
	ret = receive_message(channel);
	if (ret) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}

	switch (get_current_message_type(channel)) {
	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
		ret = enqueue_notification_from_current_message(channel);
		if (ret) {
			goto end_clean_poll;
		}
		*_notification_pending = true;
		break;
	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
		ret = enqueue_dropped_notification(channel);
		if (ret) {
			goto end_clean_poll;
		}
		*_notification_pending = true;
		break;
	default:
		/* Protocol error. */
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_clean_poll;
	}

end_clean_poll:
	lttng_poll_clean(&events);
end_unlock:
	pthread_mutex_unlock(&channel->lock);
end:
	return status;
}

static
int receive_command_reply(struct lttng_notification_channel *channel,
		enum lttng_notification_channel_status *status)
{
	int ret;
	struct lttng_notification_channel_command_reply *reply;

	while (true) {
		enum lttng_notification_channel_message_type msg_type;

		ret = receive_message(channel);
		if (ret) {
			goto end;
		}

		msg_type = get_current_message_type(channel);
		switch (msg_type) {
		case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY:
			goto exit_loop;
		case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
			ret = enqueue_notification_from_current_message(
					channel);
			if (ret) {
				goto end;
			}
			break;
		case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
			ret = enqueue_dropped_notification(channel);
			if (ret) {
				goto end;
			}
			break;
		case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
		{
			struct lttng_notification_channel_command_handshake *handshake;

			handshake = (struct lttng_notification_channel_command_handshake *)
					(channel->reception_payload.buffer.data +
					sizeof(struct lttng_notification_channel_message));
			channel->version.major = handshake->major;
			channel->version.minor = handshake->minor;
			channel->version.set = true;
			break;
		}
		default:
			ret = -1;
			goto end;
		}
	}

exit_loop:
	if (channel->reception_payload.buffer.size <
			(sizeof(struct lttng_notification_channel_message) +
			sizeof(*reply))) {
		/* Invalid message received. */
		ret = -1;
		goto end;
	}

	reply = (struct lttng_notification_channel_command_reply *)
			(channel->reception_payload.buffer.data +
			sizeof(struct lttng_notification_channel_message));
	*status = (enum lttng_notification_channel_status) reply->status;
end:
	return ret;
}

static
int handshake(struct lttng_notification_channel *channel)
{
	ssize_t ret;
	enum lttng_notification_channel_status status =
			LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	struct lttng_notification_channel_command_handshake handshake = {
		.major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
		.minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
	};
	struct lttng_notification_channel_message msg_header = {
		.type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
		.size = sizeof(handshake),
		.fds = 0,
	};
	char send_buffer[sizeof(msg_header) + sizeof(handshake)];

	memcpy(send_buffer, &msg_header, sizeof(msg_header));
	memcpy(send_buffer + sizeof(msg_header), &handshake, sizeof(handshake));

	pthread_mutex_lock(&channel->lock);

	ret = lttcomm_send_creds_unix_sock(channel->socket, send_buffer,
			sizeof(send_buffer));
	if (ret < 0) {
		goto end_unlock;
	}

	/* Receive handshake info from the sessiond. */
	ret = receive_command_reply(channel, &status);
	if (ret < 0) {
		goto end_unlock;
	}

	if (!channel->version.set) {
		ret = -1;
		goto end_unlock;
	}

	if (channel->version.major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
		ret = -1;
		goto end_unlock;
	}

end_unlock:
	pthread_mutex_unlock(&channel->lock);
	return ret;
}

static
enum lttng_notification_channel_status send_condition_command(
		struct lttng_notification_channel *channel,
		enum lttng_notification_channel_message_type type,
		const struct lttng_condition *condition)
{
	int socket;
	ssize_t ret;
	enum lttng_notification_channel_status status =
			LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	struct lttng_payload payload;
	struct lttng_notification_channel_message cmd_header = {
		.type = (int8_t) type,
		.size =0,
		.fds = 0,
	};

	lttng_payload_init(&payload);

	if (!channel) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
		goto end;
	}

	LTTNG_ASSERT(type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE ||
		type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE);

	pthread_mutex_lock(&channel->lock);
	socket = channel->socket;

	if (!lttng_condition_validate(condition)) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
		goto end_unlock;
	}

	ret = lttng_dynamic_buffer_append(&payload.buffer, &cmd_header,
			sizeof(cmd_header));
	if (ret) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_unlock;
	}

	ret = lttng_condition_serialize(condition, &payload);
	if (ret) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
		goto end_unlock;
	}

	/* Update payload length. */
	((struct lttng_notification_channel_message *) payload.buffer.data)->size =
			(uint32_t) (payload.buffer.size - sizeof(cmd_header));

	{
		struct lttng_payload_view pv =
				lttng_payload_view_from_payload(
						&payload, 0, -1);
		const int fd_count =
				lttng_payload_view_get_fd_handle_count(&pv);

		/* Update fd count. */
		((struct lttng_notification_channel_message *) payload.buffer.data)->fds =
			(uint32_t) fd_count;

		ret = lttcomm_send_unix_sock(
			socket, pv.buffer.data, pv.buffer.size);
		if (ret < 0) {
			status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
			goto end_unlock;
		}

		/* Pass fds if present. */
		if (fd_count > 0) {
			ret = lttcomm_send_payload_view_fds_unix_sock(socket,
					&pv);
			if (ret < 0) {
				status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
				goto end_unlock;
			}
		}
	}

	ret = receive_command_reply(channel, &status);
	if (ret < 0) {
		status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
		goto end_unlock;
	}
end_unlock:
	pthread_mutex_unlock(&channel->lock);
end:
	lttng_payload_reset(&payload);
	return status;
}

enum lttng_notification_channel_status lttng_notification_channel_subscribe(
		struct lttng_notification_channel *channel,
		const struct lttng_condition *condition)
{
	return send_condition_command(channel,
			LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE,
			condition);
}

enum lttng_notification_channel_status lttng_notification_channel_unsubscribe(
		struct lttng_notification_channel *channel,
		const struct lttng_condition *condition)
{
	return send_condition_command(channel,
			LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE,
			condition);
}

void lttng_notification_channel_destroy(
		struct lttng_notification_channel *channel)
{
	if (!channel) {
		return;
	}

	if (channel->socket >= 0) {
		(void) lttcomm_close_unix_sock(channel->socket);
	}
	pthread_mutex_destroy(&channel->lock);
	lttng_payload_reset(&channel->reception_payload);
	free(channel);
}
Commit	Line	Data
	1	/*
	2	* Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
	3	*
	4	* SPDX-License-Identifier: LGPL-2.1-only
	5	*
	6	*/
	7
	8	#include <lttng/notification/notification-internal.h>
	9	#include <lttng/notification/channel-internal.h>
	10	#include <lttng/condition/condition-internal.h>
	11	#include <lttng/endpoint.h>
	12	#include <common/defaults.h>
	13	#include <common/error.h>
	14	#include <common/dynamic-buffer.h>
	15	#include <common/utils.h>
	16	#include <common/defaults.h>
	17	#include <common/payload.h>
	18	#include <common/payload-view.h>
	19	#include <common/unix.h>
	20	#include "lttng-ctl-helper.h"
	21	#include <common/compat/poll.h>
	22
	23	static
	24	int handshake(struct lttng_notification_channel *channel);
	25
	26	/*
	27	* Populates the reception buffer with the next complete message.
	28	* The caller must acquire the channel's lock.
	29	*/
	30	static
	31	int receive_message(struct lttng_notification_channel *channel)
	32	{
	33	ssize_t ret;
	34	struct lttng_notification_channel_message msg;
	35
	36	lttng_payload_clear(&channel->reception_payload);
	37
	38	ret = lttcomm_recv_unix_sock(channel->socket, &msg, sizeof(msg));
	39	if (ret <= 0) {
	40	ret = -1;
	41	goto error;
	42	}
	43
	44	if (msg.size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
	45	ret = -1;
	46	goto error;
	47	}
	48
	49	/* Add message header at buffer's start. */
	50	ret = lttng_dynamic_buffer_append(&channel->reception_payload.buffer, &msg,
	51	sizeof(msg));
	52	if (ret) {
	53	goto error;
	54	}
	55
	56	/* Reserve space for the payload. */
	57	ret = lttng_dynamic_buffer_set_size(&channel->reception_payload.buffer,
	58	channel->reception_payload.buffer.size + msg.size);
	59	if (ret) {
	60	goto error;
	61	}
	62
	63	/* Receive message payload. */
	64	ret = lttcomm_recv_unix_sock(channel->socket,
	65	channel->reception_payload.buffer.data + sizeof(msg), msg.size);
	66	if (ret < (ssize_t) msg.size) {
	67	ret = -1;
	68	goto error;
	69	}
	70
	71	/* Receive message fds. */
	72	if (msg.fds != 0) {
	73	ret = lttcomm_recv_payload_fds_unix_sock(channel->socket,
	74	msg.fds, &channel->reception_payload);
	75	if (ret < sizeof(int) * msg.fds) {
	76	ret = -1;
	77	goto error;
	78	}
	79	}
	80	ret = 0;
	81	end:
	82	return ret;
	83	error:
	84	lttng_payload_clear(&channel->reception_payload);
	85	goto end;
	86	}
	87
	88	static
	89	enum lttng_notification_channel_message_type get_current_message_type(
	90	struct lttng_notification_channel *channel)
	91	{
	92	struct lttng_notification_channel_message *msg;
	93
	94	LTTNG_ASSERT(channel->reception_payload.buffer.size >= sizeof(*msg));
	95
	96	msg = (struct lttng_notification_channel_message *)
	97	channel->reception_payload.buffer.data;
	98	return (enum lttng_notification_channel_message_type) msg->type;
	99	}
	100
	101	static
	102	struct lttng_notification *create_notification_from_current_message(
	103	struct lttng_notification_channel *channel)
	104	{
	105	ssize_t ret;
	106	struct lttng_notification *notification = NULL;
	107
	108	if (channel->reception_payload.buffer.size <=
	109	sizeof(struct lttng_notification_channel_message)) {
	110	goto end;
	111	}
	112
	113	{
	114	struct lttng_payload_view view = lttng_payload_view_from_payload(
	115	&channel->reception_payload,
	116	sizeof(struct lttng_notification_channel_message),
	117	-1);
	118
	119	ret = lttng_notification_create_from_payload(
	120	&view, &notification);
	121	}
	122
	123	if (ret != channel->reception_payload.buffer.size -
	124	sizeof(struct lttng_notification_channel_message)) {
	125	lttng_notification_destroy(notification);
	126	notification = NULL;
	127	goto end;
	128	}
	129	end:
	130	return notification;
	131	}
	132
	133	struct lttng_notification_channel *lttng_notification_channel_create(
	134	struct lttng_endpoint *endpoint)
	135	{
	136	int fd, ret;
	137	bool is_in_tracing_group = false, is_root = false;
	138	char *sock_path = NULL;
	139	struct lttng_notification_channel *channel = NULL;
	140
	141	if (!endpoint \|\|
	142	endpoint != lttng_session_daemon_notification_endpoint) {
	143	goto end;
	144	}
	145
	146	sock_path = (char *) zmalloc(LTTNG_PATH_MAX);
	147	if (!sock_path) {
	148	goto end;
	149	}
	150
	151	channel = (lttng_notification_channel *) zmalloc(sizeof(struct lttng_notification_channel));
	152	if (!channel) {
	153	goto end;
	154	}
	155	channel->socket = -1;
	156	pthread_mutex_init(&channel->lock, NULL);
	157	lttng_payload_init(&channel->reception_payload);
	158	CDS_INIT_LIST_HEAD(&channel->pending_notifications.list);
	159
	160	is_root = (getuid() == 0);
	161	if (!is_root) {
	162	is_in_tracing_group = lttng_check_tracing_group();
	163	}
	164
	165	if (is_root \|\| is_in_tracing_group) {
	166	ret = lttng_strncpy(sock_path,
	167	DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK,
	168	LTTNG_PATH_MAX);
	169	if (ret) {
	170	ret = -LTTNG_ERR_INVALID;
	171	goto error;
	172	}
	173
	174	ret = lttcomm_connect_unix_sock(sock_path);
	175	if (ret >= 0) {
	176	fd = ret;
	177	goto set_fd;
	178	}
	179	}
	180
	181	/* Fallback to local session daemon. */
	182	ret = snprintf(sock_path, LTTNG_PATH_MAX,
	183	DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
	184	utils_get_home_dir());
	185	if (ret < 0 \|\| ret >= LTTNG_PATH_MAX) {
	186	goto error;
	187	}
	188
	189	ret = lttcomm_connect_unix_sock(sock_path);
	190	if (ret < 0) {
	191	goto error;
	192	}
	193	fd = ret;
	194
	195	set_fd:
	196	channel->socket = fd;
	197
	198	ret = handshake(channel);
	199	if (ret) {
	200	goto error;
	201	}
	202	end:
	203	free(sock_path);
	204	return channel;
	205	error:
	206	lttng_notification_channel_destroy(channel);
	207	channel = NULL;
	208	goto end;
	209	}
	210
	211	enum lttng_notification_channel_status
	212	lttng_notification_channel_get_next_notification(
	213	struct lttng_notification_channel *channel,
	214	struct lttng_notification **_notification)
	215	{
	216	int ret;
	217	struct lttng_notification *notification = NULL;
	218	enum lttng_notification_channel_status status =
	219	LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	220	struct lttng_poll_event events;
	221
	222	if (!channel \|\| !_notification) {
	223	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
	224	goto end;
	225	}
	226
	227	pthread_mutex_lock(&channel->lock);
	228
	229	if (channel->pending_notifications.count) {
	230	struct pending_notification *pending_notification;
	231
	232	LTTNG_ASSERT(!cds_list_empty(&channel->pending_notifications.list));
	233
	234	/* Deliver one of the pending notifications. */
	235	pending_notification = cds_list_first_entry(
	236	&channel->pending_notifications.list,
	237	struct pending_notification,
	238	node);
	239	notification = pending_notification->notification;
	240	if (!notification) {
	241	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
	242	}
	243	cds_list_del(&pending_notification->node);
	244	channel->pending_notifications.count--;
	245	free(pending_notification);
	246	goto end_unlock;
	247	}
	248
	249	/*
	250	* Block on interruptible epoll/poll() instead of the message reception
	251	* itself as the recvmsg() wrappers always restart on EINTR. We choose
	252	* to wait using interruptible epoll/poll() in order to:
	253	* 1) Return if a signal occurs,
	254	* 2) Not deal with partially received messages.
	255	*
	256	* The drawback to this approach is that we assume that messages
	257	* are complete/well formed. If a message is shorter than its
	258	* announced length, receive_message() will block on recvmsg()
	259	* and never return (even if a signal is received).
	260	*/
	261	ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
	262	if (ret < 0) {
	263	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	264	goto end_unlock;
	265	}
	266	ret = lttng_poll_add(&events, channel->socket, LPOLLIN \| LPOLLERR);
	267	if (ret < 0) {
	268	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	269	goto end_clean_poll;
	270	}
	271	ret = lttng_poll_wait_interruptible(&events, -1);
	272	if (ret <= 0) {
	273	status = (ret == -1 && errno == EINTR) ?
	274	LTTNG_NOTIFICATION_CHANNEL_STATUS_INTERRUPTED :
	275	LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	276	goto end_clean_poll;
	277	}
	278
	279	ret = receive_message(channel);
	280	if (ret) {
	281	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	282	goto end_clean_poll;
	283	}
	284
	285	switch (get_current_message_type(channel)) {
	286	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
	287	notification = create_notification_from_current_message(
	288	channel);
	289	if (!notification) {
	290	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	291	goto end_clean_poll;
	292	}
	293	break;
	294	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
	295	/* No payload to consume. */
	296	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
	297	break;
	298	default:
	299	/* Protocol error. */
	300	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	301	goto end_clean_poll;
	302	}
	303
	304	end_clean_poll:
	305	lttng_poll_clean(&events);
	306	end_unlock:
	307	pthread_mutex_unlock(&channel->lock);
	308	*_notification = notification;
	309	end:
	310	return status;
	311	}
	312
	313	static
	314	int enqueue_dropped_notification(
	315	struct lttng_notification_channel *channel)
	316	{
	317	int ret = 0;
	318	struct pending_notification *pending_notification;
	319	struct cds_list_head *last_element =
	320	channel->pending_notifications.list.prev;
	321
	322	pending_notification = caa_container_of(last_element,
	323	struct pending_notification, node);
	324	if (!pending_notification->notification) {
	325	/*
	326	* The last enqueued notification indicates dropped
	327	* notifications; there is nothing to do as we group
	328	* dropped notifications together.
	329	*/
	330	goto end;
	331	}
	332
	333	if (channel->pending_notifications.count >=
	334	DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT &&
	335	pending_notification->notification) {
	336	/*
	337	* Discard the last enqueued notification to indicate
	338	* that notifications were dropped at this point.
	339	*/
	340	lttng_notification_destroy(
	341	pending_notification->notification);
	342	pending_notification->notification = NULL;
	343	goto end;
	344	}
	345
	346	pending_notification = (struct pending_notification ) zmalloc(sizeof(pending_notification));
	347	if (!pending_notification) {
	348	ret = -1;
	349	goto end;
	350	}
	351	CDS_INIT_LIST_HEAD(&pending_notification->node);
	352	cds_list_add(&pending_notification->node,
	353	&channel->pending_notifications.list);
	354	channel->pending_notifications.count++;
	355	end:
	356	return ret;
	357	}
	358
	359	static
	360	int enqueue_notification_from_current_message(
	361	struct lttng_notification_channel *channel)
	362	{
	363	int ret = 0;
	364	struct lttng_notification *notification;
	365	struct pending_notification *pending_notification;
	366
	367	if (channel->pending_notifications.count >=
	368	DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT) {
	369	/* Drop the notification. */
	370	ret = enqueue_dropped_notification(channel);
	371	goto end;
	372	}
	373
	374	pending_notification = (struct pending_notification ) zmalloc(sizeof(pending_notification));
	375	if (!pending_notification) {
	376	ret = -1;
	377	goto error;
	378	}
	379	CDS_INIT_LIST_HEAD(&pending_notification->node);
	380
	381	notification = create_notification_from_current_message(channel);
	382	if (!notification) {
	383	ret = -1;
	384	goto error;
	385	}
	386
	387	pending_notification->notification = notification;
	388	cds_list_add(&pending_notification->node,
	389	&channel->pending_notifications.list);
	390	channel->pending_notifications.count++;
	391	end:
	392	return ret;
	393	error:
	394	free(pending_notification);
	395	goto end;
	396	}
	397
	398	enum lttng_notification_channel_status
	399	lttng_notification_channel_has_pending_notification(
	400	struct lttng_notification_channel *channel,
	401	bool *_notification_pending)
	402	{
	403	int ret;
	404	enum lttng_notification_channel_status status =
	405	LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	406	struct lttng_poll_event events;
	407
	408	if (!channel \|\| !_notification_pending) {
	409	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
	410	goto end;
	411	}
	412
	413	pthread_mutex_lock(&channel->lock);
	414
	415	if (channel->pending_notifications.count) {
	416	*_notification_pending = true;
	417	goto end_unlock;
	418	}
	419
	420	if (channel->socket < 0) {
	421	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED;
	422	goto end_unlock;
	423	}
	424
	425	/*
	426	* Check, without blocking, if data is available on the channel's
	427	* socket. If there is data available, it is safe to read (blocking)
	428	* on the socket for a message from the session daemon.
	429	*
	430	* Since all commands wait for the session daemon's reply before
	431	* releasing the channel's lock, the protocol only allows for
	432	* notifications and "notification dropped" messages to come
	433	* through. If we receive a different message type, it is
	434	* considered a protocol error.
	435	*
	436	* Note that this function is not guaranteed not to block. This
	437	* will block until our peer (the session daemon) has sent a complete
	438	* message if we see data available on the socket. If the peer does
	439	* not respect the protocol, this may block indefinitely.
	440	*/
	441	ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
	442	if (ret < 0) {
	443	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	444	goto end_unlock;
	445	}
	446	ret = lttng_poll_add(&events, channel->socket, LPOLLIN \| LPOLLERR);
	447	if (ret < 0) {
	448	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	449	goto end_clean_poll;
	450	}
	451	/* timeout = 0: return immediately. */
	452	ret = lttng_poll_wait_interruptible(&events, 0);
	453	if (ret == 0) {
	454	/* No data available. */
	455	*_notification_pending = false;
	456	goto end_clean_poll;
	457	} else if (ret < 0) {
	458	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	459	goto end_clean_poll;
	460	}
	461
	462	/* Data available on socket. */
	463	ret = receive_message(channel);
	464	if (ret) {
	465	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	466	goto end_clean_poll;
	467	}
	468
	469	switch (get_current_message_type(channel)) {
	470	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
	471	ret = enqueue_notification_from_current_message(channel);
	472	if (ret) {
	473	goto end_clean_poll;
	474	}
	475	*_notification_pending = true;
	476	break;
	477	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
	478	ret = enqueue_dropped_notification(channel);
	479	if (ret) {
	480	goto end_clean_poll;
	481	}
	482	*_notification_pending = true;
	483	break;
	484	default:
	485	/* Protocol error. */
	486	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	487	goto end_clean_poll;
	488	}
	489
	490	end_clean_poll:
	491	lttng_poll_clean(&events);
	492	end_unlock:
	493	pthread_mutex_unlock(&channel->lock);
	494	end:
	495	return status;
	496	}
	497
	498	static
	499	int receive_command_reply(struct lttng_notification_channel *channel,
	500	enum lttng_notification_channel_status *status)
	501	{
	502	int ret;
	503	struct lttng_notification_channel_command_reply *reply;
	504
	505	while (true) {
	506	enum lttng_notification_channel_message_type msg_type;
	507
	508	ret = receive_message(channel);
	509	if (ret) {
	510	goto end;
	511	}
	512
	513	msg_type = get_current_message_type(channel);
	514	switch (msg_type) {
	515	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY:
	516	goto exit_loop;
	517	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
	518	ret = enqueue_notification_from_current_message(
	519	channel);
	520	if (ret) {
	521	goto end;
	522	}
	523	break;
	524	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
	525	ret = enqueue_dropped_notification(channel);
	526	if (ret) {
	527	goto end;
	528	}
	529	break;
	530	case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
	531	{
	532	struct lttng_notification_channel_command_handshake *handshake;
	533
	534	handshake = (struct lttng_notification_channel_command_handshake *)
	535	(channel->reception_payload.buffer.data +
	536	sizeof(struct lttng_notification_channel_message));
	537	channel->version.major = handshake->major;
	538	channel->version.minor = handshake->minor;
	539	channel->version.set = true;
	540	break;
	541	}
	542	default:
	543	ret = -1;
	544	goto end;
	545	}
	546	}
	547
	548	exit_loop:
	549	if (channel->reception_payload.buffer.size <
	550	(sizeof(struct lttng_notification_channel_message) +
	551	sizeof(*reply))) {
	552	/* Invalid message received. */
	553	ret = -1;
	554	goto end;
	555	}
	556
	557	reply = (struct lttng_notification_channel_command_reply *)
	558	(channel->reception_payload.buffer.data +
	559	sizeof(struct lttng_notification_channel_message));
	560	*status = (enum lttng_notification_channel_status) reply->status;
	561	end:
	562	return ret;
	563	}
	564
	565	static
	566	int handshake(struct lttng_notification_channel *channel)
	567	{
	568	ssize_t ret;
	569	enum lttng_notification_channel_status status =
	570	LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	571	struct lttng_notification_channel_command_handshake handshake = {
	572	.major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
	573	.minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
	574	};
	575	struct lttng_notification_channel_message msg_header = {
	576	.type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
	577	.size = sizeof(handshake),
	578	.fds = 0,
	579	};
	580	char send_buffer[sizeof(msg_header) + sizeof(handshake)];
	581
	582	memcpy(send_buffer, &msg_header, sizeof(msg_header));
	583	memcpy(send_buffer + sizeof(msg_header), &handshake, sizeof(handshake));
	584
	585	pthread_mutex_lock(&channel->lock);
	586
	587	ret = lttcomm_send_creds_unix_sock(channel->socket, send_buffer,
	588	sizeof(send_buffer));
	589	if (ret < 0) {
	590	goto end_unlock;
	591	}
	592
	593	/* Receive handshake info from the sessiond. */
	594	ret = receive_command_reply(channel, &status);
	595	if (ret < 0) {
	596	goto end_unlock;
	597	}
	598
	599	if (!channel->version.set) {
	600	ret = -1;
	601	goto end_unlock;
	602	}
	603
	604	if (channel->version.major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
	605	ret = -1;
	606	goto end_unlock;
	607	}
	608
	609	end_unlock:
	610	pthread_mutex_unlock(&channel->lock);
	611	return ret;
	612	}
	613
	614	static
	615	enum lttng_notification_channel_status send_condition_command(
	616	struct lttng_notification_channel *channel,
	617	enum lttng_notification_channel_message_type type,
	618	const struct lttng_condition *condition)
	619	{
	620	int socket;
	621	ssize_t ret;
	622	enum lttng_notification_channel_status status =
	623	LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
	624	struct lttng_payload payload;
	625	struct lttng_notification_channel_message cmd_header = {
	626	.type = (int8_t) type,
	627	.size =0,
	628	.fds = 0,
	629	};
	630
	631	lttng_payload_init(&payload);
	632
	633	if (!channel) {
	634	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
	635	goto end;
	636	}
	637
	638	LTTNG_ASSERT(type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE \|\|
	639	type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE);
	640
	641	pthread_mutex_lock(&channel->lock);
	642	socket = channel->socket;
	643
	644	if (!lttng_condition_validate(condition)) {
	645	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
	646	goto end_unlock;
	647	}
	648
	649	ret = lttng_dynamic_buffer_append(&payload.buffer, &cmd_header,
	650	sizeof(cmd_header));
	651	if (ret) {
	652	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	653	goto end_unlock;
	654	}
	655
	656	ret = lttng_condition_serialize(condition, &payload);
	657	if (ret) {
	658	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
	659	goto end_unlock;
	660	}
	661
	662	/* Update payload length. */
	663	((struct lttng_notification_channel_message *) payload.buffer.data)->size =
	664	(uint32_t) (payload.buffer.size - sizeof(cmd_header));
	665
	666	{
	667	struct lttng_payload_view pv =
	668	lttng_payload_view_from_payload(
	669	&payload, 0, -1);
	670	const int fd_count =
	671	lttng_payload_view_get_fd_handle_count(&pv);
	672
	673	/* Update fd count. */
	674	((struct lttng_notification_channel_message *) payload.buffer.data)->fds =
	675	(uint32_t) fd_count;
	676
	677	ret = lttcomm_send_unix_sock(
	678	socket, pv.buffer.data, pv.buffer.size);
	679	if (ret < 0) {
	680	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	681	goto end_unlock;
	682	}
	683
	684	/* Pass fds if present. */
	685	if (fd_count > 0) {
	686	ret = lttcomm_send_payload_view_fds_unix_sock(socket,
	687	&pv);
	688	if (ret < 0) {
	689	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	690	goto end_unlock;
	691	}
	692	}
	693	}
	694
	695	ret = receive_command_reply(channel, &status);
	696	if (ret < 0) {
	697	status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
	698	goto end_unlock;
	699	}
	700	end_unlock:
	701	pthread_mutex_unlock(&channel->lock);
	702	end:
	703	lttng_payload_reset(&payload);
	704	return status;
	705	}
	706
	707	enum lttng_notification_channel_status lttng_notification_channel_subscribe(
	708	struct lttng_notification_channel *channel,
	709	const struct lttng_condition *condition)
	710	{
	711	return send_condition_command(channel,
	712	LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE,
	713	condition);
	714	}
	715
	716	enum lttng_notification_channel_status lttng_notification_channel_unsubscribe(
	717	struct lttng_notification_channel *channel,
	718	const struct lttng_condition *condition)
	719	{
	720	return send_condition_command(channel,
	721	LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE,
	722	condition);
	723	}
	724
	725	void lttng_notification_channel_destroy(
	726	struct lttng_notification_channel *channel)
	727	{
	728	if (!channel) {
	729	return;
	730	}
	731
	732	if (channel->socket >= 0) {
	733	(void) lttcomm_close_unix_sock(channel->socket);
	734	}
	735	pthread_mutex_destroy(&channel->lock);
	736	lttng_payload_reset(&channel->reception_payload);
	737	free(channel);
	738	}