[lttng-tools.git] / src / common / consumer-stream.c

/*
 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
 *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 * Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License, version 2 only, as
 * published by the Free Software Foundation.
 *
 * 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 <assert.h>
#include <inttypes.h>
#include <sys/mman.h>
#include <unistd.h>

#include <common/common.h>
#include <common/index/index.h>
#include <common/relayd/relayd.h>
#include <common/ust-consumer/ust-consumer.h>

#include "consumer-stream.h"

/*
 * RCU call to free stream. MUST only be used with call_rcu().
 */
static void free_stream_rcu(struct rcu_head *head)
{
	struct lttng_ht_node_u64 *node =
		caa_container_of(head, struct lttng_ht_node_u64, head);
	struct lttng_consumer_stream *stream =
		caa_container_of(node, struct lttng_consumer_stream, node);

	pthread_mutex_destroy(&stream->lock);
	free(stream);
}

/*
 * Close stream on the relayd side. This call can destroy a relayd if the
 * conditions are met.
 *
 * A RCU read side lock MUST be acquired if the relayd object was looked up in
 * a hash table before calling this.
 */
void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
		struct consumer_relayd_sock_pair *relayd)
{
	int ret;

	assert(stream);
	assert(relayd);

	if (stream->sent_to_relayd) {
		uatomic_dec(&relayd->refcount);
		assert(uatomic_read(&relayd->refcount) >= 0);
	}

	/* Closing streams requires to lock the control socket. */
	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
	ret = relayd_send_close_stream(&relayd->control_sock,
			stream->relayd_stream_id,
			stream->next_net_seq_num - 1);
	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	if (ret < 0) {
		DBG("Unable to close stream on the relayd. Continuing");
		/*
		 * Continue here. There is nothing we can do for the relayd.
		 * Chances are that the relayd has closed the socket so we just
		 * continue cleaning up.
		 */
	}

	/* Both conditions are met, we destroy the relayd. */
	if (uatomic_read(&relayd->refcount) == 0 &&
			uatomic_read(&relayd->destroy_flag)) {
		consumer_destroy_relayd(relayd);
	}
	stream->net_seq_idx = (uint64_t) -1ULL;
	stream->sent_to_relayd = 0;
}

/*
 * Close stream's file descriptors and, if needed, close stream also on the
 * relayd side.
 *
 * The consumer data lock MUST be acquired.
 * The stream lock MUST be acquired.
 */
void consumer_stream_close(struct lttng_consumer_stream *stream)
{
	int ret;
	struct consumer_relayd_sock_pair *relayd;

	assert(stream);

	switch (consumer_data.type) {
	case LTTNG_CONSUMER_KERNEL:
		if (stream->mmap_base != NULL) {
			ret = munmap(stream->mmap_base, stream->mmap_len);
			if (ret != 0) {
				PERROR("munmap");
			}
		}

		if (stream->wait_fd >= 0) {
			ret = close(stream->wait_fd);
			if (ret) {
				PERROR("close");
			}
			stream->wait_fd = -1;
		}
		break;
	case LTTNG_CONSUMER32_UST:
	case LTTNG_CONSUMER64_UST:
		break;
	default:
		ERR("Unknown consumer_data type");
		assert(0);
	}

	/* Close output fd. Could be a socket or local file at this point. */
	if (stream->out_fd >= 0) {
		ret = close(stream->out_fd);
		if (ret) {
			PERROR("close");
		}
		stream->out_fd = -1;
	}

	if (stream->index_fd >= 0) {
		ret = close(stream->index_fd);
		if (ret) {
			PERROR("close stream index_fd");
		}
		stream->index_fd = -1;
	}

	/* Check and cleanup relayd if needed. */
	rcu_read_lock();
	relayd = consumer_find_relayd(stream->net_seq_idx);
	if (relayd != NULL) {
		consumer_stream_relayd_close(stream, relayd);
	}
	rcu_read_unlock();
}

/*
 * Delete the stream from all possible hash tables.
 *
 * The consumer data lock MUST be acquired.
 * The stream lock MUST be acquired.
 */
void consumer_stream_delete(struct lttng_consumer_stream *stream,
		struct lttng_ht *ht)
{
	int ret;
	struct lttng_ht_iter iter;

	assert(stream);
	/* Should NEVER be called not in monitor mode. */
	assert(stream->chan->monitor);

	rcu_read_lock();

	if (ht) {
		iter.iter.node = &stream->node.node;
		ret = lttng_ht_del(ht, &iter);
		assert(!ret);
	}

	/* Delete from stream per channel ID hash table. */
	iter.iter.node = &stream->node_channel_id.node;
	/*
	 * The returned value is of no importance. Even if the node is NOT in the
	 * hash table, we continue since we may have been called by a code path
	 * that did not add the stream to a (all) hash table. Same goes for the
	 * next call ht del call.
	 */
	(void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);

	/* Delete from the global stream list. */
	iter.iter.node = &stream->node_session_id.node;
	/* See the previous ht del on why we ignore the returned value. */
	(void) lttng_ht_del(consumer_data.stream_list_ht, &iter);

	rcu_read_unlock();

	/* Decrement the stream count of the global consumer data. */
	assert(consumer_data.stream_count > 0);
	consumer_data.stream_count--;
}

/*
 * Free the given stream within a RCU call.
 */
void consumer_stream_free(struct lttng_consumer_stream *stream)
{
	assert(stream);

	call_rcu(&stream->node.head, free_stream_rcu);
}

/*
 * Destroy the stream's buffers of the tracer.
 */
void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
{
	assert(stream);

	switch (consumer_data.type) {
	case LTTNG_CONSUMER_KERNEL:
		break;
	case LTTNG_CONSUMER32_UST:
	case LTTNG_CONSUMER64_UST:
		lttng_ustconsumer_del_stream(stream);
		break;
	default:
		ERR("Unknown consumer_data type");
		assert(0);
	}
}

/*
 * Destroy and close a already created stream.
 */
static void destroy_close_stream(struct lttng_consumer_stream *stream)
{
	assert(stream);

	DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);

	/* Destroy tracer buffers of the stream. */
	consumer_stream_destroy_buffers(stream);
	/* Close down everything including the relayd if one. */
	consumer_stream_close(stream);
}

/*
 * Decrement the stream's channel refcount and if down to 0, return the channel
 * pointer so it can be destroyed by the caller or NULL if not.
 */
static struct lttng_consumer_channel *unref_channel(
		struct lttng_consumer_stream *stream)
{
	struct lttng_consumer_channel *free_chan = NULL;

	assert(stream);
	assert(stream->chan);

	/* Update refcount of channel and see if we need to destroy it. */
	if (!uatomic_sub_return(&stream->chan->refcount, 1)
			&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
		free_chan = stream->chan;
	}

	return free_chan;
}

/*
 * Destroy a stream completely. This will delete, close and free the stream.
 * Once return, the stream is NO longer usable. Its channel may get destroyed
 * if conditions are met for a monitored stream.
 *
 * This MUST be called WITHOUT the consumer data and stream lock acquired if
 * the stream is in _monitor_ mode else it does not matter.
 */
void consumer_stream_destroy(struct lttng_consumer_stream *stream,
		struct lttng_ht *ht)
{
	assert(stream);

	/* Stream is in monitor mode. */
	if (stream->monitor) {
		struct lttng_consumer_channel *free_chan = NULL;

		/*
		 * This means that the stream was successfully removed from the streams
		 * list of the channel and sent to the right thread managing this
		 * stream thus being globally visible.
		 */
		if (stream->globally_visible) {
			pthread_mutex_lock(&consumer_data.lock);
			pthread_mutex_lock(&stream->chan->lock);
			pthread_mutex_lock(&stream->lock);
			/* Remove every reference of the stream in the consumer. */
			consumer_stream_delete(stream, ht);

			destroy_close_stream(stream);

			/* Update channel's refcount of the stream. */
			free_chan = unref_channel(stream);

			/* Indicates that the consumer data state MUST be updated after this. */
			consumer_data.need_update = 1;

			pthread_mutex_unlock(&stream->lock);
			pthread_mutex_unlock(&stream->chan->lock);
			pthread_mutex_unlock(&consumer_data.lock);
		} else {
			/*
			 * If the stream is not visible globally, this needs to be done
			 * outside of the consumer data lock section.
			 */
			free_chan = unref_channel(stream);
		}

		if (free_chan) {
			consumer_del_channel(free_chan);
		}
	} else {
		destroy_close_stream(stream);
	}

	/* Free stream within a RCU call. */
	consumer_stream_free(stream);
}

/*
 * Write index of a specific stream either on the relayd or local disk.
 *
 * Return 0 on success or else a negative value.
 */
int consumer_stream_write_index(struct lttng_consumer_stream *stream,
		struct lttng_packet_index *index)
{
	int ret;
	struct consumer_relayd_sock_pair *relayd;

	assert(stream);
	assert(index);

	rcu_read_lock();
	relayd = consumer_find_relayd(stream->net_seq_idx);
	if (relayd) {
		ret = relayd_send_index(&relayd->control_sock, index,
				stream->relayd_stream_id, stream->next_net_seq_num - 1);
	} else {
		ret = index_write(stream->index_fd, index,
				sizeof(struct lttng_packet_index));
	}
	if (ret < 0) {
		goto error;
	}

error:
	rcu_read_unlock();
	return ret;
}
Commit	Line	Data
51230d70 DG	1	/*
	2	* Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
	3	* Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	4	* Copyright (C) 2013 - David Goulet <dgoulet@efficios.com>
	5	*
	6	* This program is free software; you can redistribute it and/or modify it
	7	* under the terms of the GNU General Public License, version 2 only, as
	8	* published by the Free Software Foundation.
	9	*
	10	* This program is distributed in the hope that it will be useful, but WITHOUT
	11	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	12	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
	13	* more details.
	14	*
	15	* You should have received a copy of the GNU General Public License along with
	16	* this program; if not, write to the Free Software Foundation, Inc., 51
	17	* Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
	18	*/
	19
	20	#define _GNU_SOURCE
	21	#include <assert.h>
10a50311	22	#include <inttypes.h>
51230d70 DG	23	#include <sys/mman.h>
	24	#include <unistd.h>
	25
	26	#include <common/common.h>
1c20f0e2	27	#include <common/index/index.h>
51230d70 DG	28	#include <common/relayd/relayd.h>
	29	#include <common/ust-consumer/ust-consumer.h>
	30
	31	#include "consumer-stream.h"
	32
	33	/*
	34	* RCU call to free stream. MUST only be used with call_rcu().
	35	*/
	36	static void free_stream_rcu(struct rcu_head *head)
	37	{
	38	struct lttng_ht_node_u64 *node =
	39	caa_container_of(head, struct lttng_ht_node_u64, head);
	40	struct lttng_consumer_stream *stream =
	41	caa_container_of(node, struct lttng_consumer_stream, node);
	42
	43	pthread_mutex_destroy(&stream->lock);
	44	free(stream);
	45	}
	46
	47	/*
	48	* Close stream on the relayd side. This call can destroy a relayd if the
	49	* conditions are met.
	50	*
	51	* A RCU read side lock MUST be acquired if the relayd object was looked up in
	52	* a hash table before calling this.
	53	*/
	54	void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
	55	struct consumer_relayd_sock_pair *relayd)
	56	{
	57	int ret;
	58
	59	assert(stream);
	60	assert(relayd);
	61
d01178b6 DG	62	if (stream->sent_to_relayd) {
	63	uatomic_dec(&relayd->refcount);
	64	assert(uatomic_read(&relayd->refcount) >= 0);
	65	}
51230d70 DG	66
	67	/* Closing streams requires to lock the control socket. */
	68	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
	69	ret = relayd_send_close_stream(&relayd->control_sock,
	70	stream->relayd_stream_id,
	71	stream->next_net_seq_num - 1);
	72	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	73	if (ret < 0) {
	74	DBG("Unable to close stream on the relayd. Continuing");
	75	/*
	76	* Continue here. There is nothing we can do for the relayd.
	77	* Chances are that the relayd has closed the socket so we just
	78	* continue cleaning up.
	79	*/
	80	}
	81
	82	/* Both conditions are met, we destroy the relayd. */
	83	if (uatomic_read(&relayd->refcount) == 0 &&
	84	uatomic_read(&relayd->destroy_flag)) {
	85	consumer_destroy_relayd(relayd);
	86	}
10a50311	87	stream->net_seq_idx = (uint64_t) -1ULL;
d01178b6	88	stream->sent_to_relayd = 0;
51230d70 DG	89	}
	90
	91	/*
	92	* Close stream's file descriptors and, if needed, close stream also on the
	93	* relayd side.
	94	*
	95	* The consumer data lock MUST be acquired.
	96	* The stream lock MUST be acquired.
	97	*/
	98	void consumer_stream_close(struct lttng_consumer_stream *stream)
	99	{
	100	int ret;
	101	struct consumer_relayd_sock_pair *relayd;
	102
	103	assert(stream);
	104
	105	switch (consumer_data.type) {
	106	case LTTNG_CONSUMER_KERNEL:
	107	if (stream->mmap_base != NULL) {
	108	ret = munmap(stream->mmap_base, stream->mmap_len);
	109	if (ret != 0) {
	110	PERROR("munmap");
	111	}
	112	}
	113
	114	if (stream->wait_fd >= 0) {
	115	ret = close(stream->wait_fd);
	116	if (ret) {
	117	PERROR("close");
	118	}
10a50311	119	stream->wait_fd = -1;
51230d70 DG	120	}
	121	break;
	122	case LTTNG_CONSUMER32_UST:
	123	case LTTNG_CONSUMER64_UST:
51230d70 DG	124	break;
	125	default:
	126	ERR("Unknown consumer_data type");
	127	assert(0);
	128	}
	129
	130	/* Close output fd. Could be a socket or local file at this point. */
	131	if (stream->out_fd >= 0) {
	132	ret = close(stream->out_fd);
	133	if (ret) {
	134	PERROR("close");
	135	}
10a50311	136	stream->out_fd = -1;
51230d70 DG	137	}
51230d70 DG	138
309167d2 JD	139	if (stream->index_fd >= 0) {
	140	ret = close(stream->index_fd);
	141	if (ret) {
	142	PERROR("close stream index_fd");
	143	}
	144	stream->index_fd = -1;
	145	}
	146
51230d70 DG	147	/* Check and cleanup relayd if needed. */
	148	rcu_read_lock();
	149	relayd = consumer_find_relayd(stream->net_seq_idx);
	150	if (relayd != NULL) {
	151	consumer_stream_relayd_close(stream, relayd);
	152	}
	153	rcu_read_unlock();
	154	}
	155
	156	/*
	157	* Delete the stream from all possible hash tables.
	158	*
	159	* The consumer data lock MUST be acquired.
	160	* The stream lock MUST be acquired.
	161	*/
	162	void consumer_stream_delete(struct lttng_consumer_stream *stream,
	163	struct lttng_ht *ht)
	164	{
	165	int ret;
	166	struct lttng_ht_iter iter;
	167
	168	assert(stream);
10a50311 JD	169	/* Should NEVER be called not in monitor mode. */
10a50311 JD	170	assert(stream->chan->monitor);
51230d70 DG	171
	172	rcu_read_lock();
	173
	174	if (ht) {
	175	iter.iter.node = &stream->node.node;
	176	ret = lttng_ht_del(ht, &iter);
	177	assert(!ret);
	178	}
	179
	180	/* Delete from stream per channel ID hash table. */
	181	iter.iter.node = &stream->node_channel_id.node;
	182	/*
	183	* The returned value is of no importance. Even if the node is NOT in the
	184	* hash table, we continue since we may have been called by a code path
	185	* that did not add the stream to a (all) hash table. Same goes for the
	186	* next call ht del call.
	187	*/
	188	(void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
	189
	190	/* Delete from the global stream list. */
	191	iter.iter.node = &stream->node_session_id.node;
	192	/* See the previous ht del on why we ignore the returned value. */
	193	(void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
	194
	195	rcu_read_unlock();
	196
10a50311 JD	197	/* Decrement the stream count of the global consumer data. */
	198	assert(consumer_data.stream_count > 0);
	199	consumer_data.stream_count--;
51230d70 DG	200	}
	201
	202	/*
	203	* Free the given stream within a RCU call.
	204	*/
	205	void consumer_stream_free(struct lttng_consumer_stream *stream)
	206	{
	207	assert(stream);
	208
	209	call_rcu(&stream->node.head, free_stream_rcu);
	210	}
	211
	212	/*
10a50311	213	* Destroy the stream's buffers of the tracer.
51230d70	214	*/
10a50311	215	void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
51230d70	216	{
10a50311 JD	217	assert(stream);
	218
	219	switch (consumer_data.type) {
	220	case LTTNG_CONSUMER_KERNEL:
	221	break;
	222	case LTTNG_CONSUMER32_UST:
	223	case LTTNG_CONSUMER64_UST:
	224	lttng_ustconsumer_del_stream(stream);
	225	break;
	226	default:
	227	ERR("Unknown consumer_data type");
	228	assert(0);
	229	}
	230	}
51230d70	231
10a50311	232	/*
4891ece8	233	* Destroy and close a already created stream.
10a50311	234	*/
4891ece8	235	static void destroy_close_stream(struct lttng_consumer_stream *stream)
10a50311	236	{
51230d70 DG	237	assert(stream);
51230d70 DG	238
4891ece8	239	DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
10a50311 JD	240
	241	/* Destroy tracer buffers of the stream. */
	242	consumer_stream_destroy_buffers(stream);
	243	/* Close down everything including the relayd if one. */
	244	consumer_stream_close(stream);
	245	}
51230d70	246
10a50311	247	/*
4891ece8 DG	248	* Decrement the stream's channel refcount and if down to 0, return the channel
4891ece8 DG	249	* pointer so it can be destroyed by the caller or NULL if not.
10a50311	250	*/
4891ece8 DG	251	static struct lttng_consumer_channel *unref_channel(
4891ece8 DG	252	struct lttng_consumer_stream *stream)
10a50311	253	{
4891ece8 DG	254	struct lttng_consumer_channel *free_chan = NULL;
4891ece8 DG	255
10a50311	256	assert(stream);
4891ece8	257	assert(stream->chan);
10a50311	258
4891ece8 DG	259	/* Update refcount of channel and see if we need to destroy it. */
	260	if (!uatomic_sub_return(&stream->chan->refcount, 1)
	261	&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
	262	free_chan = stream->chan;
	263	}
51230d70	264
4891ece8	265	return free_chan;
10a50311	266	}
51230d70	267
10a50311 JD	268	/*
	269	* Destroy a stream completely. This will delete, close and free the stream.
	270	* Once return, the stream is NO longer usable. Its channel may get destroyed
	271	* if conditions are met for a monitored stream.
	272	*
	273	* This MUST be called WITHOUT the consumer data and stream lock acquired if
	274	* the stream is in _monitor_ mode else it does not matter.
	275	*/
	276	void consumer_stream_destroy(struct lttng_consumer_stream *stream,
	277	struct lttng_ht *ht)
	278	{
	279	assert(stream);
	280
	281	/* Stream is in monitor mode. */
4891ece8	282	if (stream->monitor) {
10a50311	283	struct lttng_consumer_channel *free_chan = NULL;
51230d70	284
4891ece8 DG	285	/*
	286	* This means that the stream was successfully removed from the streams
	287	* list of the channel and sent to the right thread managing this
	288	* stream thus being globally visible.
	289	*/
	290	if (stream->globally_visible) {
	291	pthread_mutex_lock(&consumer_data.lock);
a9838785	292	pthread_mutex_lock(&stream->chan->lock);
4891ece8 DG	293	pthread_mutex_lock(&stream->lock);
	294	/* Remove every reference of the stream in the consumer. */
	295	consumer_stream_delete(stream, ht);
	296
	297	destroy_close_stream(stream);
	298
	299	/* Update channel's refcount of the stream. */
	300	free_chan = unref_channel(stream);
	301
	302	/* Indicates that the consumer data state MUST be updated after this. */
	303	consumer_data.need_update = 1;
	304
	305	pthread_mutex_unlock(&stream->lock);
a9838785	306	pthread_mutex_unlock(&stream->chan->lock);
4891ece8 DG	307	pthread_mutex_unlock(&consumer_data.lock);
	308	} else {
	309	/*
	310	* If the stream is not visible globally, this needs to be done
	311	* outside of the consumer data lock section.
	312	*/
	313	free_chan = unref_channel(stream);
10a50311 JD	314	}
10a50311 JD	315
10a50311 JD	316	if (free_chan) {
	317	consumer_del_channel(free_chan);
	318	}
	319	} else {
4891ece8	320	destroy_close_stream(stream);
51230d70 DG	321	}
	322
	323	/* Free stream within a RCU call. */
	324	consumer_stream_free(stream);
	325	}
1c20f0e2 JD	326
	327	/*
	328	* Write index of a specific stream either on the relayd or local disk.
	329	*
	330	* Return 0 on success or else a negative value.
	331	*/
	332	int consumer_stream_write_index(struct lttng_consumer_stream *stream,
	333	struct lttng_packet_index *index)
	334	{
	335	int ret;
	336	struct consumer_relayd_sock_pair *relayd;
	337
	338	assert(stream);
	339	assert(index);
	340
	341	rcu_read_lock();
	342	relayd = consumer_find_relayd(stream->net_seq_idx);
	343	if (relayd) {
	344	ret = relayd_send_index(&relayd->control_sock, index,
	345	stream->relayd_stream_id, stream->next_net_seq_num - 1);
	346	} else {
	347	ret = index_write(stream->index_fd, index,
	348	sizeof(struct lttng_packet_index));
	349	}
	350	if (ret < 0) {
	351	goto error;
	352	}
	353
	354	error:
	355	rcu_read_unlock();
	356	return ret;
	357	}