/*

 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>

 * Copyright (C) 2011-2013 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>

 *

 * This program is free software; you can redistribute it and/or modify

 * it under the terms of the GNU General Public License as published by

 * the Free Software Foundation; version 2 of the License only.

 *

 * This program 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 General Public License for more details.

 *

 * You should have received a copy of the GNU General Public License along

 * with this program; if not, write to the Free Software Foundation, Inc.,

 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

 */

#define _GNU_SOURCE

#include <string.h>

#include <lttng/ust-ctl.h>

#include <lttng/ust-abi.h>

#include <lttng/ust-events.h>

#include <sys/mman.h>

#include <byteswap.h>

#include <usterr-signal-safe.h>

#include <ust-comm.h>

#include <helper.h>

#include "../libringbuffer/backend.h"

#include "../libringbuffer/frontend.h"

#include "../liblttng-ust/wait.h"

#include "../liblttng-ust/lttng-rb-clients.h"

#include "../liblttng-ust/clock.h"

/*

 * Number of milliseconds to retry before failing metadata writes on

 * buffer full condition. (10 seconds)

 */

#define LTTNG_METADATA_TIMEOUT_MSEC	10000

/*

 * Channel representation within consumer.

 */

struct ustctl_consumer_channel {

	struct lttng_channel *chan;		/* lttng channel buffers */

	/* initial attributes */

	struct ustctl_consumer_channel_attr attr;

	int wait_fd;				/* monitor close() */

	int wakeup_fd;				/* monitor close() */

};

/*

 * Stream representation within consumer.

 */

struct ustctl_consumer_stream {

	struct lttng_ust_shm_handle *handle;	/* shared-memory handle */

	struct lttng_ust_lib_ring_buffer *buf;

	struct ustctl_consumer_channel *chan;

	int shm_fd, wait_fd, wakeup_fd;

	int cpu;

	uint64_t memory_map_size;

};

extern void lttng_ring_buffer_client_overwrite_init(void);

extern void lttng_ring_buffer_client_overwrite_rt_init(void);

extern void lttng_ring_buffer_client_discard_init(void);

extern void lttng_ring_buffer_client_discard_rt_init(void);

extern void lttng_ring_buffer_metadata_client_init(void);

extern void lttng_ring_buffer_client_overwrite_exit(void);

extern void lttng_ring_buffer_client_overwrite_rt_exit(void);

extern void lttng_ring_buffer_client_discard_exit(void);

extern void lttng_ring_buffer_client_discard_rt_exit(void);

extern void lttng_ring_buffer_metadata_client_exit(void);

volatile enum ust_loglevel ust_loglevel;

int ustctl_release_handle(int sock, int handle)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	if (sock < 0 || handle < 0)

		return 0;

	memset(&lum, 0, sizeof(lum));

	lum.handle = handle;

	lum.cmd = LTTNG_UST_RELEASE;

	return ustcomm_send_app_cmd(sock, &lum, &lur);

}

/*

 * If sock is negative, it means we don't have to notify the other side

 * (e.g. application has already vanished).

 */

int ustctl_release_object(int sock, struct lttng_ust_object_data *data)

{

	int ret;

	if (!data)

		return -EINVAL;

	switch (data->type) {

	case LTTNG_UST_OBJECT_TYPE_CHANNEL:

		if (data->u.channel.wakeup_fd >= 0) {

			ret = close(data->u.channel.wakeup_fd);

			if (ret < 0) {

				ret = -errno;

				return ret;

			}

		}

		free(data->u.channel.data);

		break;

	case LTTNG_UST_OBJECT_TYPE_STREAM:

		if (data->u.stream.shm_fd >= 0) {

			ret = close(data->u.stream.shm_fd);

			if (ret < 0) {

				ret = -errno;

				return ret;

			}

		}

		if (data->u.stream.wakeup_fd >= 0) {

			ret = close(data->u.stream.wakeup_fd);

			if (ret < 0) {

				ret = -errno;

				return ret;

			}

		}

		break;

	case LTTNG_UST_OBJECT_TYPE_EVENT:

	case LTTNG_UST_OBJECT_TYPE_CONTEXT:

		break;

	default:

		assert(0);

	}

	return ustctl_release_handle(sock, data->handle);

}

/*

 * Send registration done packet to the application.

 */

int ustctl_register_done(int sock)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	DBG("Sending register done command to %d", sock);

	memset(&lum, 0, sizeof(lum));

	lum.handle = LTTNG_UST_ROOT_HANDLE;

	lum.cmd = LTTNG_UST_REGISTER_DONE;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	return 0;

}

/*

 * returns session handle.

 */

int ustctl_create_session(int sock)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret, session_handle;

	/* Create session */

	memset(&lum, 0, sizeof(lum));

	lum.handle = LTTNG_UST_ROOT_HANDLE;

	lum.cmd = LTTNG_UST_SESSION;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	session_handle = lur.ret_val;

	DBG("received session handle %u", session_handle);

	return session_handle;

}

int ustctl_create_event(int sock, struct lttng_ust_event *ev,

		struct lttng_ust_object_data *channel_data,

		struct lttng_ust_object_data **_event_data)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	struct lttng_ust_object_data *event_data;

	int ret;

	if (!channel_data || !_event_data)

		return -EINVAL;

	event_data = zmalloc(sizeof(*event_data));

	if (!event_data)

		return -ENOMEM;

	event_data->type = LTTNG_UST_OBJECT_TYPE_EVENT;

	memset(&lum, 0, sizeof(lum));

	lum.handle = channel_data->handle;

	lum.cmd = LTTNG_UST_EVENT;

	strncpy(lum.u.event.name, ev->name,

		LTTNG_UST_SYM_NAME_LEN);

	lum.u.event.instrumentation = ev->instrumentation;

	lum.u.event.loglevel_type = ev->loglevel_type;

	lum.u.event.loglevel = ev->loglevel;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret) {

		free(event_data);

		return ret;

	}

	event_data->handle = lur.ret_val;

	DBG("received event handle %u", event_data->handle);

	*_event_data = event_data;

	return 0;

}

int ustctl_add_context(int sock, struct lttng_ust_context_attr *ctx,

		struct lttng_ust_object_data *obj_data,

		struct lttng_ust_object_data **_context_data)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	struct lttng_ust_object_data *context_data = NULL;

	char *buf = NULL;

	size_t len;

	int ret;

	if (!obj_data || !_context_data) {

		ret = -EINVAL;

		goto end;

	}

	context_data = zmalloc(sizeof(*context_data));

	if (!context_data) {

		ret = -ENOMEM;

		goto end;

	}

	context_data->type = LTTNG_UST_OBJECT_TYPE_CONTEXT;

	memset(&lum, 0, sizeof(lum));

	lum.handle = obj_data->handle;

	lum.cmd = LTTNG_UST_CONTEXT;

	lum.u.context.ctx = ctx->ctx;

	switch (ctx->ctx) {

	case LTTNG_UST_CONTEXT_PERF_THREAD_COUNTER:

		lum.u.context.u.perf_counter = ctx->u.perf_counter;

		break;

	case LTTNG_UST_CONTEXT_APP_CONTEXT:

	{

		size_t provider_name_len = strlen(

				ctx->u.app_ctx.provider_name) + 1;

		size_t ctx_name_len = strlen(ctx->u.app_ctx.ctx_name) + 1;

		lum.u.context.u.app_ctx.provider_name_len = provider_name_len;

		lum.u.context.u.app_ctx.ctx_name_len = ctx_name_len;

		len = provider_name_len + ctx_name_len;

		buf = zmalloc(len);

		if (!buf) {

			ret = -ENOMEM;

			goto end;

		}

		memcpy(buf, ctx->u.app_ctx.provider_name,

				provider_name_len);

		memcpy(buf + provider_name_len, ctx->u.app_ctx.ctx_name,

				ctx_name_len);

		break;

	}

	default:

		break;

	}

	ret = ustcomm_send_app_msg(sock, &lum;;

	if (ret)

		goto end;

	if (buf) {

		/* send var len ctx_name */

		ret = ustcomm_send_unix_sock(sock, buf, len);

		if (ret < 0) {

			goto end;

		}

		if (ret != len) {

			ret = -EINVAL;

			goto end;

		}

	}

	ret = ustcomm_recv_app_reply(sock, &lur, lum.handle, lum.cmd);

	if (ret < 0) {

		goto end;

	}

	context_data->handle = -1;

	DBG("Context created successfully");

	*_context_data = context_data;

	context_data = NULL;

end:

	free(context_data);

	free(buf);

	return ret;

}

int ustctl_set_filter(int sock, struct lttng_ust_filter_bytecode *bytecode,

		struct lttng_ust_object_data *obj_data)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!obj_data)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = obj_data->handle;

	lum.cmd = LTTNG_UST_FILTER;

	lum.u.filter.data_size = bytecode->len;

	lum.u.filter.reloc_offset = bytecode->reloc_offset;

	lum.u.filter.seqnum = bytecode->seqnum;

	ret = ustcomm_send_app_msg(sock, &lum;;

	if (ret)

		return ret;

	/* send var len bytecode */

	ret = ustcomm_send_unix_sock(sock, bytecode->data,

				bytecode->len);

	if (ret < 0) {

		return ret;

	}

	if (ret != bytecode->len)

		return -EINVAL;

	return ustcomm_recv_app_reply(sock, &lur, lum.handle, lum.cmd);

}

int ustctl_set_exclusion(int sock, struct lttng_ust_event_exclusion *exclusion,

		struct lttng_ust_object_data *obj_data)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!obj_data) {

		return -EINVAL;

	}

	memset(&lum, 0, sizeof(lum));

	lum.handle = obj_data->handle;

	lum.cmd = LTTNG_UST_EXCLUSION;

	lum.u.exclusion.count = exclusion->count;

	ret = ustcomm_send_app_msg(sock, &lum;;

	if (ret) {

		return ret;

	}

	/* send var len exclusion names */

	ret = ustcomm_send_unix_sock(sock,

			exclusion->names,

			exclusion->count * LTTNG_UST_SYM_NAME_LEN);

	if (ret < 0) {

		return ret;

	}

	if (ret != exclusion->count * LTTNG_UST_SYM_NAME_LEN) {

		return -EINVAL;

	}

	return ustcomm_recv_app_reply(sock, &lur, lum.handle, lum.cmd);

}

/* Enable event, channel and session ioctl */

int ustctl_enable(int sock, struct lttng_ust_object_data *object)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!object)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = object->handle;

	lum.cmd = LTTNG_UST_ENABLE;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	DBG("enabled handle %u", object->handle);

	return 0;

}

/* Disable event, channel and session ioctl */

int ustctl_disable(int sock, struct lttng_ust_object_data *object)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!object)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = object->handle;

	lum.cmd = LTTNG_UST_DISABLE;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	DBG("disable handle %u", object->handle);

	return 0;

}

int ustctl_start_session(int sock, int handle)

{

	struct lttng_ust_object_data obj;

	obj.handle = handle;

	return ustctl_enable(sock, &obj);

}

int ustctl_stop_session(int sock, int handle)

{

	struct lttng_ust_object_data obj;

	obj.handle = handle;

	return ustctl_disable(sock, &obj);

}

int ustctl_tracepoint_list(int sock)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret, tp_list_handle;

	memset(&lum, 0, sizeof(lum));

	lum.handle = LTTNG_UST_ROOT_HANDLE;

	lum.cmd = LTTNG_UST_TRACEPOINT_LIST;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	tp_list_handle = lur.ret_val;

	DBG("received tracepoint list handle %u", tp_list_handle);

	return tp_list_handle;

}

int ustctl_tracepoint_list_get(int sock, int tp_list_handle,

		struct lttng_ust_tracepoint_iter *iter)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!iter)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = tp_list_handle;

	lum.cmd = LTTNG_UST_TRACEPOINT_LIST_GET;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	DBG("received tracepoint list entry name %s loglevel %d",

		lur.u.tracepoint.name,

		lur.u.tracepoint.loglevel);

	memcpy(iter, &lur.u.tracepoint, sizeof(*iter));

	return 0;

}

int ustctl_tracepoint_field_list(int sock)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret, tp_field_list_handle;

	memset(&lum, 0, sizeof(lum));

	lum.handle = LTTNG_UST_ROOT_HANDLE;

	lum.cmd = LTTNG_UST_TRACEPOINT_FIELD_LIST;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	tp_field_list_handle = lur.ret_val;

	DBG("received tracepoint field list handle %u", tp_field_list_handle);

	return tp_field_list_handle;

}

int ustctl_tracepoint_field_list_get(int sock, int tp_field_list_handle,

		struct lttng_ust_field_iter *iter)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	ssize_t len;

	if (!iter)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = tp_field_list_handle;

	lum.cmd = LTTNG_UST_TRACEPOINT_FIELD_LIST_GET;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	len = ustcomm_recv_unix_sock(sock, iter, sizeof(*iter));

	if (len != sizeof(*iter)) {

		return -EINVAL;

	}

	DBG("received tracepoint field list entry event_name %s event_loglevel %d field_name %s field_type %d",

		iter->event_name,

		iter->loglevel,

		iter->field_name,

		iter->type);

	return 0;

}

int ustctl_tracer_version(int sock, struct lttng_ust_tracer_version *v)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!v)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = LTTNG_UST_ROOT_HANDLE;

	lum.cmd = LTTNG_UST_TRACER_VERSION;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	memcpy(v, &lur.u.version, sizeof(*v));

	DBG("received tracer version");

	return 0;

}

int ustctl_wait_quiescent(int sock)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	memset(&lum, 0, sizeof(lum));

	lum.handle = LTTNG_UST_ROOT_HANDLE;

	lum.cmd = LTTNG_UST_WAIT_QUIESCENT;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	DBG("waited for quiescent state");

	return 0;

}

int ustctl_calibrate(int sock, struct lttng_ust_calibrate *calibrate)

{

	if (!calibrate)

		return -EINVAL;

	return -ENOSYS;

}

int ustctl_sock_flush_buffer(int sock, struct lttng_ust_object_data *object)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!object)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = object->handle;

	lum.cmd = LTTNG_UST_FLUSH_BUFFER;

	ret = ustcomm_send_app_cmd(sock, &lum, &lur);

	if (ret)

		return ret;

	DBG("flushed buffer handle %u", object->handle);

	return 0;

}

static

int ustctl_send_channel(int sock,

		enum lttng_ust_chan_type type,

		void *data,

		uint64_t size,

		int wakeup_fd,

		int send_fd_only)

{

	ssize_t len;

	if (!send_fd_only) {

		/* Send mmap size */

		len = ustcomm_send_unix_sock(sock, &size, sizeof(size));

		if (len != sizeof(size)) {

			if (len < 0)

				return len;

			else

				return -EIO;

		}

		/* Send channel type */

		len = ustcomm_send_unix_sock(sock, &type, sizeof(type));

		if (len != sizeof(type)) {

			if (len < 0)

				return len;

			else

				return -EIO;

		}

	}

	/* Send channel data */

	len = ustcomm_send_unix_sock(sock, data, size);

	if (len != size) {

		if (len < 0)

			return len;

		else

			return -EIO;

	}

	/* Send wakeup fd */

	len = ustcomm_send_fds_unix_sock(sock, &wakeup_fd, 1);

	if (len <= 0) {

		if (len < 0)

			return len;

		else

			return -EIO;

	}

	return 0;

}

static

int ustctl_send_stream(int sock,

		uint32_t stream_nr,

		uint64_t memory_map_size,

		int shm_fd, int wakeup_fd,

		int send_fd_only)

{

	ssize_t len;

	int fds[2];

	if (!send_fd_only) {

		if (shm_fd < 0) {

			/* finish iteration */

			uint64_t v = -1;

			len = ustcomm_send_unix_sock(sock, &v, sizeof(v));

			if (len != sizeof(v)) {

				if (len < 0)

					return len;

				else

					return -EIO;

			}

			return 0;

		}

		/* Send mmap size */

		len = ustcomm_send_unix_sock(sock, &memory_map_size,

			sizeof(memory_map_size));

		if (len != sizeof(memory_map_size)) {

			if (len < 0)

				return len;

			else

				return -EIO;

		}

		/* Send stream nr */

		len = ustcomm_send_unix_sock(sock, &stream_nr,

			sizeof(stream_nr));

		if (len != sizeof(stream_nr)) {

			if (len < 0)

				return len;

			else

				return -EIO;

		}

	}

	/* Send shm fd and wakeup fd */

	fds[0] = shm_fd;

	fds[1] = wakeup_fd;

	len = ustcomm_send_fds_unix_sock(sock, fds, 2);

	if (len <= 0) {

		if (len < 0)

			return len;

		else

			return -EIO;

	}

	return 0;

}

int ustctl_recv_channel_from_consumer(int sock,

		struct lttng_ust_object_data **_channel_data)

{

	struct lttng_ust_object_data *channel_data;

	ssize_t len;

	int wakeup_fd;

	int ret;

	channel_data = zmalloc(sizeof(*channel_data));

	if (!channel_data) {

		ret = -ENOMEM;

		goto error_alloc;

	}

	channel_data->type = LTTNG_UST_OBJECT_TYPE_CHANNEL;

	channel_data->handle = -1;

	/* recv mmap size */

	len = ustcomm_recv_unix_sock(sock, &channel_data->size,

			sizeof(channel_data->size));

	if (len != sizeof(channel_data->size)) {

		if (len < 0)

			ret = len;

		else

			ret = -EINVAL;

		goto error;

	}

	/* recv channel type */

	len = ustcomm_recv_unix_sock(sock, &channel_data->u.channel.type,

			sizeof(channel_data->u.channel.type));

	if (len != sizeof(channel_data->u.channel.type)) {

		if (len < 0)

			ret = len;

		else

			ret = -EINVAL;

		goto error;

	}

	/* recv channel data */

	channel_data->u.channel.data = zmalloc(channel_data->size);

	if (!channel_data->u.channel.data) {

		ret = -ENOMEM;

		goto error;

	}

	len = ustcomm_recv_unix_sock(sock, channel_data->u.channel.data,

			channel_data->size);

	if (len != channel_data->size) {

		if (len < 0)

			ret = len;

		else

			ret = -EINVAL;

		goto error_recv_data;

	}

	/* recv wakeup fd */

	len = ustcomm_recv_fds_unix_sock(sock, &wakeup_fd, 1);

	if (len <= 0) {

		if (len < 0) {

			ret = len;

			goto error_recv_data;

		} else {

			ret = -EIO;

			goto error_recv_data;

		}

	}

	channel_data->u.channel.wakeup_fd = wakeup_fd;

	*_channel_data = channel_data;

	return 0;

error_recv_data:

	free(channel_data->u.channel.data);

error:

	free(channel_data);

error_alloc:

	return ret;

}

int ustctl_recv_stream_from_consumer(int sock,

		struct lttng_ust_object_data **_stream_data)

{

	struct lttng_ust_object_data *stream_data;

	ssize_t len;

	int ret;

	int fds[2];

	stream_data = zmalloc(sizeof(*stream_data));

	if (!stream_data) {

		ret = -ENOMEM;

		goto error_alloc;

	}

	stream_data->type = LTTNG_UST_OBJECT_TYPE_STREAM;

	stream_data->handle = -1;

	/* recv mmap size */

	len = ustcomm_recv_unix_sock(sock, &stream_data->size,

			sizeof(stream_data->size));

	if (len != sizeof(stream_data->size)) {

		if (len < 0)

			ret = len;

		else

			ret = -EINVAL;

		goto error;

	}

	if (stream_data->size == -1) {

		ret = -LTTNG_UST_ERR_NOENT;

		goto error;

	}

	/* recv stream nr */

	len = ustcomm_recv_unix_sock(sock, &stream_data->u.stream.stream_nr,

			sizeof(stream_data->u.stream.stream_nr));

	if (len != sizeof(stream_data->u.stream.stream_nr)) {

		if (len < 0)

			ret = len;

		else

			ret = -EINVAL;

		goto error;

	}

	/* recv shm fd and wakeup fd */

	len = ustcomm_recv_fds_unix_sock(sock, fds, 2);

	if (len <= 0) {

		if (len < 0) {

			ret = len;

			goto error;

		} else {

			ret = -EIO;

			goto error;

		}

	}

	stream_data->u.stream.shm_fd = fds[0];

	stream_data->u.stream.wakeup_fd = fds[1];

	*_stream_data = stream_data;

	return 0;

error:

	free(stream_data);

error_alloc:

	return ret;

}

int ustctl_send_channel_to_ust(int sock, int session_handle,

		struct lttng_ust_object_data *channel_data)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	if (!channel_data)

		return -EINVAL;

	memset(&lum, 0, sizeof(lum));

	lum.handle = session_handle;

	lum.cmd = LTTNG_UST_CHANNEL;

	lum.u.channel.len = channel_data->size;

	lum.u.channel.type = channel_data->u.channel.type;

	ret = ustcomm_send_app_msg(sock, &lum;;

	if (ret)

		return ret;

	ret = ustctl_send_channel(sock,

			channel_data->u.channel.type,

			channel_data->u.channel.data,

			channel_data->size,

			channel_data->u.channel.wakeup_fd,

			1);

	if (ret)

		return ret;

	ret = ustcomm_recv_app_reply(sock, &lur, lum.handle, lum.cmd);

	if (!ret) {

		channel_data->handle = lur.ret_val;

	}

	return ret;

}

int ustctl_send_stream_to_ust(int sock,

		struct lttng_ust_object_data *channel_data,

		struct lttng_ust_object_data *stream_data)

{

	struct ustcomm_ust_msg lum;

	struct ustcomm_ust_reply lur;

	int ret;

	memset(&lum, 0, sizeof(lum));

	lum.handle = channel_data->handle;

	lum.cmd = LTTNG_UST_STREAM;

	lum.u.stream.len = stream_data->size;

	lum.u.stream.stream_nr = stream_data->u.stream.stream_nr;

	ret = ustcomm_send_app_msg(sock, &lum;;

	if (ret)

		return ret;

	assert(stream_data);

	assert(stream_data->type == LTTNG_UST_OBJECT_TYPE_STREAM);

	ret = ustctl_send_stream(sock,

			stream_data->u.stream.stream_nr,

			stream_data->size,

			stream_data->u.stream.shm_fd,

			stream_data->u.stream.wakeup_fd, 1);

	if (ret)

		return ret;

	return ustcomm_recv_app_reply(sock, &lur, lum.handle, lum.cmd);

}