[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/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);

	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;
}

/*
 * 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;
	}

	/* 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 a stream in no monitor mode.
 *
 * We need a separate function because this can be called inside a destroy
 * channel path which have the consumer data lock acquired. Also, in no monitor
 * mode, the channel refcount is NOT incremented per stream since the ownership
 * of those streams are INSIDE the channel making the lazy destroy channel not
 * possible for a non monitor stream.
 *
 * Furthermore, there is no need to delete the stream from the global hash
 * table so we avoid useless calls.
 */
static void destroy_no_monitor(struct lttng_consumer_stream *stream)
{
	assert(stream);

	DBG("Consumer stream destroy unmonitored 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);
}

/*
 * Destroy a stream in monitor mode.
 */
static void destroy_monitor(struct lttng_consumer_stream *stream,
		struct lttng_ht *ht)
{
	assert(stream);

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

	/* Remove every reference of the stream in the consumer. */
	consumer_stream_delete(stream, ht);
	/* Destroy tracer buffers of the stream. */
	consumer_stream_destroy_buffers(stream);
	/* Close down everything including the relayd if one. */
	consumer_stream_close(stream);
}

/*
 * 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->chan->monitor) {
		struct lttng_consumer_channel *free_chan = NULL;

		pthread_mutex_lock(&consumer_data.lock);
		pthread_mutex_lock(&stream->lock);

		destroy_monitor(stream, ht);

		/* 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;
		}

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

		pthread_mutex_unlock(&stream->lock);
		pthread_mutex_unlock(&consumer_data.lock);

		if (free_chan) {
			consumer_del_channel(free_chan);
		}
	} else {
		/*
		 * No monitor mode the stream's ownership is in its channel thus we
		 * don't have to handle the channel refcount nor the lazy deletion.
		 */
		destroy_no_monitor(stream);
	}

	/* Free stream within a RCU call. */
	consumer_stream_free(stream);
}
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>
	27	#include <common/relayd/relayd.h>
	28	#include <common/ust-consumer/ust-consumer.h>
	29
	30	#include "consumer-stream.h"
	31
	32	/*
	33	* RCU call to free stream. MUST only be used with call_rcu().
	34	*/
	35	static void free_stream_rcu(struct rcu_head *head)
	36	{
	37	struct lttng_ht_node_u64 *node =
	38	caa_container_of(head, struct lttng_ht_node_u64, head);
	39	struct lttng_consumer_stream *stream =
	40	caa_container_of(node, struct lttng_consumer_stream, node);
	41
	42	pthread_mutex_destroy(&stream->lock);
	43	free(stream);
	44	}
	45
	46	/*
	47	* Close stream on the relayd side. This call can destroy a relayd if the
	48	* conditions are met.
	49	*
	50	* A RCU read side lock MUST be acquired if the relayd object was looked up in
	51	* a hash table before calling this.
	52	*/
	53	void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
	54	struct consumer_relayd_sock_pair *relayd)
	55	{
	56	int ret;
	57
	58	assert(stream);
	59	assert(relayd);
	60
	61	uatomic_dec(&relayd->refcount);
	62	assert(uatomic_read(&relayd->refcount) >= 0);
	63
	64	/* Closing streams requires to lock the control socket. */
	65	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
	66	ret = relayd_send_close_stream(&relayd->control_sock,
	67	stream->relayd_stream_id,
	68	stream->next_net_seq_num - 1);
	69	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
	70	if (ret < 0) {
	71	DBG("Unable to close stream on the relayd. Continuing");
	72	/*
	73	* Continue here. There is nothing we can do for the relayd.
	74	* Chances are that the relayd has closed the socket so we just
	75	* continue cleaning up.
	76	*/
	77	}
	78
	79	/* Both conditions are met, we destroy the relayd. */
	80	if (uatomic_read(&relayd->refcount) == 0 &&
	81	uatomic_read(&relayd->destroy_flag)) {
	82	consumer_destroy_relayd(relayd);
	83	}
10a50311	84	stream->net_seq_idx = (uint64_t) -1ULL;
51230d70 DG	85	}
	86
	87	/*
	88	* Close stream's file descriptors and, if needed, close stream also on the
	89	* relayd side.
	90	*
	91	* The consumer data lock MUST be acquired.
	92	* The stream lock MUST be acquired.
	93	*/
	94	void consumer_stream_close(struct lttng_consumer_stream *stream)
	95	{
	96	int ret;
	97	struct consumer_relayd_sock_pair *relayd;
	98
	99	assert(stream);
	100
	101	switch (consumer_data.type) {
	102	case LTTNG_CONSUMER_KERNEL:
	103	if (stream->mmap_base != NULL) {
	104	ret = munmap(stream->mmap_base, stream->mmap_len);
	105	if (ret != 0) {
	106	PERROR("munmap");
	107	}
	108	}
	109
	110	if (stream->wait_fd >= 0) {
	111	ret = close(stream->wait_fd);
	112	if (ret) {
	113	PERROR("close");
	114	}
10a50311	115	stream->wait_fd = -1;
51230d70 DG	116	}
	117	break;
	118	case LTTNG_CONSUMER32_UST:
	119	case LTTNG_CONSUMER64_UST:
51230d70 DG	120	break;
	121	default:
	122	ERR("Unknown consumer_data type");
	123	assert(0);
	124	}
	125
	126	/* Close output fd. Could be a socket or local file at this point. */
	127	if (stream->out_fd >= 0) {
	128	ret = close(stream->out_fd);
	129	if (ret) {
	130	PERROR("close");
	131	}
10a50311	132	stream->out_fd = -1;
51230d70 DG	133	}
	134
	135	/* Check and cleanup relayd if needed. */
	136	rcu_read_lock();
	137	relayd = consumer_find_relayd(stream->net_seq_idx);
	138	if (relayd != NULL) {
	139	consumer_stream_relayd_close(stream, relayd);
	140	}
	141	rcu_read_unlock();
	142	}
	143
	144	/*
	145	* Delete the stream from all possible hash tables.
	146	*
	147	* The consumer data lock MUST be acquired.
	148	* The stream lock MUST be acquired.
	149	*/
	150	void consumer_stream_delete(struct lttng_consumer_stream *stream,
	151	struct lttng_ht *ht)
	152	{
	153	int ret;
	154	struct lttng_ht_iter iter;
	155
	156	assert(stream);
10a50311 JD	157	/* Should NEVER be called not in monitor mode. */
10a50311 JD	158	assert(stream->chan->monitor);
51230d70 DG	159
	160	rcu_read_lock();
	161
	162	if (ht) {
	163	iter.iter.node = &stream->node.node;
	164	ret = lttng_ht_del(ht, &iter);
	165	assert(!ret);
	166	}
	167
	168	/* Delete from stream per channel ID hash table. */
	169	iter.iter.node = &stream->node_channel_id.node;
	170	/*
	171	* The returned value is of no importance. Even if the node is NOT in the
	172	* hash table, we continue since we may have been called by a code path
	173	* that did not add the stream to a (all) hash table. Same goes for the
	174	* next call ht del call.
	175	*/
	176	(void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
	177
	178	/* Delete from the global stream list. */
	179	iter.iter.node = &stream->node_session_id.node;
	180	/* See the previous ht del on why we ignore the returned value. */
	181	(void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
	182
	183	rcu_read_unlock();
	184
10a50311 JD	185	/* Decrement the stream count of the global consumer data. */
	186	assert(consumer_data.stream_count > 0);
	187	consumer_data.stream_count--;
51230d70 DG	188	}
	189
	190	/*
	191	* Free the given stream within a RCU call.
	192	*/
	193	void consumer_stream_free(struct lttng_consumer_stream *stream)
	194	{
	195	assert(stream);
	196
	197	call_rcu(&stream->node.head, free_stream_rcu);
	198	}
	199
	200	/*
10a50311	201	* Destroy the stream's buffers of the tracer.
51230d70	202	*/
10a50311	203	void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
51230d70	204	{
10a50311 JD	205	assert(stream);
	206
	207	switch (consumer_data.type) {
	208	case LTTNG_CONSUMER_KERNEL:
	209	break;
	210	case LTTNG_CONSUMER32_UST:
	211	case LTTNG_CONSUMER64_UST:
	212	lttng_ustconsumer_del_stream(stream);
	213	break;
	214	default:
	215	ERR("Unknown consumer_data type");
	216	assert(0);
	217	}
	218	}
51230d70	219
10a50311 JD	220	/*
	221	* Destroy a stream in no monitor mode.
	222	*
	223	* We need a separate function because this can be called inside a destroy
	224	* channel path which have the consumer data lock acquired. Also, in no monitor
	225	* mode, the channel refcount is NOT incremented per stream since the ownership
	226	* of those streams are INSIDE the channel making the lazy destroy channel not
	227	* possible for a non monitor stream.
	228	*
	229	* Furthermore, there is no need to delete the stream from the global hash
	230	* table so we avoid useless calls.
	231	*/
	232	static void destroy_no_monitor(struct lttng_consumer_stream *stream)
	233	{
51230d70 DG	234	assert(stream);
51230d70 DG	235
10a50311 JD	236	DBG("Consumer stream destroy unmonitored key: %" PRIu64, stream->key);
	237
	238	/* Destroy tracer buffers of the stream. */
	239	consumer_stream_destroy_buffers(stream);
	240	/* Close down everything including the relayd if one. */
	241	consumer_stream_close(stream);
	242	}
51230d70	243
10a50311 JD	244	/*
	245	* Destroy a stream in monitor mode.
	246	*/
	247	static void destroy_monitor(struct lttng_consumer_stream *stream,
	248	struct lttng_ht *ht)
	249	{
	250	assert(stream);
	251
	252	DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
51230d70 DG	253
	254	/* Remove every reference of the stream in the consumer. */
	255	consumer_stream_delete(stream, ht);
10a50311 JD	256	/* Destroy tracer buffers of the stream. */
10a50311 JD	257	consumer_stream_destroy_buffers(stream);
51230d70 DG	258	/* Close down everything including the relayd if one. */
51230d70 DG	259	consumer_stream_close(stream);
10a50311	260	}
51230d70	261
10a50311 JD	262	/*
	263	* Destroy a stream completely. This will delete, close and free the stream.
	264	* Once return, the stream is NO longer usable. Its channel may get destroyed
	265	* if conditions are met for a monitored stream.
	266	*
	267	* This MUST be called WITHOUT the consumer data and stream lock acquired if
	268	* the stream is in _monitor_ mode else it does not matter.
	269	*/
	270	void consumer_stream_destroy(struct lttng_consumer_stream *stream,
	271	struct lttng_ht *ht)
	272	{
	273	assert(stream);
	274
	275	/* Stream is in monitor mode. */
	276	if (stream->chan->monitor) {
	277	struct lttng_consumer_channel *free_chan = NULL;
51230d70	278
10a50311 JD	279	pthread_mutex_lock(&consumer_data.lock);
10a50311 JD	280	pthread_mutex_lock(&stream->lock);
51230d70	281
10a50311	282	destroy_monitor(stream, ht);
51230d70	283
10a50311 JD	284	/* Update refcount of channel and see if we need to destroy it. */
	285	if (!uatomic_sub_return(&stream->chan->refcount, 1)
	286	&& !uatomic_read(&stream->chan->nb_init_stream_left)) {
	287	free_chan = stream->chan;
	288	}
	289
	290	/* Indicates that the consumer data state MUST be updated after this. */
	291	consumer_data.need_update = 1;
	292
	293	pthread_mutex_unlock(&stream->lock);
	294	pthread_mutex_unlock(&consumer_data.lock);
	295
	296	if (free_chan) {
	297	consumer_del_channel(free_chan);
	298	}
	299	} else {
	300	/*
	301	* No monitor mode the stream's ownership is in its channel thus we
	302	* don't have to handle the channel refcount nor the lazy deletion.
	303	*/
	304	destroy_no_monitor(stream);
51230d70 DG	305	}
	306
	307	/* Free stream within a RCU call. */
	308	consumer_stream_free(stream);
	309	}