src/common/consumer-stream.c

   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>
  22 #include <sys/mman.h>
  23 #include <unistd.h>
  24
  25 #include <common/common.h>
  26 #include <common/relayd/relayd.h>
  27 #include <common/ust-consumer/ust-consumer.h>
  28
  29 #include "consumer-stream.h"
  30
  31 /*
  32  * RCU call to free stream. MUST only be used with call_rcu().
  33  */
  34 static void free_stream_rcu(struct rcu_head *head)
  35 {
  36         struct lttng_ht_node_u64 *node =
  37                 caa_container_of(head, struct lttng_ht_node_u64, head);
  38         struct lttng_consumer_stream *stream =
  39                 caa_container_of(node, struct lttng_consumer_stream, node);
  40
  41         pthread_mutex_destroy(&stream->lock);
  42         free(stream);
  43 }
  44
  45 /*
  46  * Close stream on the relayd side. This call can destroy a relayd if the
  47  * conditions are met.
  48  *
  49  * A RCU read side lock MUST be acquired if the relayd object was looked up in
  50  * a hash table before calling this.
  51  */
  52 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
  53                 struct consumer_relayd_sock_pair *relayd)
  54 {
  55         int ret;
  56
  57         assert(stream);
  58         assert(relayd);
  59
  60         uatomic_dec(&relayd->refcount);
  61         assert(uatomic_read(&relayd->refcount) >= 0);
  62
  63         /* Closing streams requires to lock the control socket. */
  64         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
  65         ret = relayd_send_close_stream(&relayd->control_sock,
  66                         stream->relayd_stream_id,
  67                         stream->next_net_seq_num - 1);
  68         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
  69         if (ret < 0) {
  70                 DBG("Unable to close stream on the relayd. Continuing");
  71                 /*
  72                  * Continue here. There is nothing we can do for the relayd.
  73                  * Chances are that the relayd has closed the socket so we just
  74                  * continue cleaning up.
  75                  */
  76         }
  77
  78         /* Both conditions are met, we destroy the relayd. */
  79         if (uatomic_read(&relayd->refcount) == 0 &&
  80                         uatomic_read(&relayd->destroy_flag)) {
  81                 consumer_destroy_relayd(relayd);
  82         }
  83 }
  84
  85 /*
  86  * Close stream's file descriptors and, if needed, close stream also on the
  87  * relayd side.
  88  *
  89  * The consumer data lock MUST be acquired.
  90  * The stream lock MUST be acquired.
  91  */
  92 void consumer_stream_close(struct lttng_consumer_stream *stream)
  93 {
  94         int ret;
  95         struct consumer_relayd_sock_pair *relayd;
  96
  97         assert(stream);
  98
  99         switch (consumer_data.type) {
 100         case LTTNG_CONSUMER_KERNEL:
 101                 if (stream->mmap_base != NULL) {
 102                         ret = munmap(stream->mmap_base, stream->mmap_len);
 103                         if (ret != 0) {
 104                                 PERROR("munmap");
 105                         }
 106                 }
 107
 108                 if (stream->wait_fd >= 0) {
 109                         ret = close(stream->wait_fd);
 110                         if (ret) {
 111                                 PERROR("close");
 112                         }
 113                 }
 114                 break;
 115         case LTTNG_CONSUMER32_UST:
 116         case LTTNG_CONSUMER64_UST:
 117                 lttng_ustconsumer_del_stream(stream);
 118                 break;
 119         default:
 120                 ERR("Unknown consumer_data type");
 121                 assert(0);
 122         }
 123
 124         /* Close output fd. Could be a socket or local file at this point. */
 125         if (stream->out_fd >= 0) {
 126                 ret = close(stream->out_fd);
 127                 if (ret) {
 128                         PERROR("close");
 129                 }
 130         }
 131
 132         /* Check and cleanup relayd if needed. */
 133         rcu_read_lock();
 134         relayd = consumer_find_relayd(stream->net_seq_idx);
 135         if (relayd != NULL) {
 136                 consumer_stream_relayd_close(stream, relayd);
 137         }
 138         rcu_read_unlock();
 139 }
 140
 141 /*
 142  * Delete the stream from all possible hash tables.
 143  *
 144  * The consumer data lock MUST be acquired.
 145  * The stream lock MUST be acquired.
 146  */
 147 void consumer_stream_delete(struct lttng_consumer_stream *stream,
 148                 struct lttng_ht *ht)
 149 {
 150         int ret;
 151         struct lttng_ht_iter iter;
 152
 153         assert(stream);
 154
 155         rcu_read_lock();
 156
 157         if (ht) {
 158                 iter.iter.node = &stream->node.node;
 159                 ret = lttng_ht_del(ht, &iter);
 160                 assert(!ret);
 161         }
 162
 163         /* Delete from stream per channel ID hash table. */
 164         iter.iter.node = &stream->node_channel_id.node;
 165         /*
 166          * The returned value is of no importance. Even if the node is NOT in the
 167          * hash table, we continue since we may have been called by a code path
 168          * that did not add the stream to a (all) hash table. Same goes for the
 169          * next call ht del call.
 170          */
 171         (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
 172
 173         /* Delete from the global stream list. */
 174         iter.iter.node = &stream->node_session_id.node;
 175         /* See the previous ht del on why we ignore the returned value. */
 176         (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
 177
 178         rcu_read_unlock();
 179
 180         /*
 181          * For a *non* monitored stream, we MUST NOT decrement or else the data
 182          * thread will use the wrong value or stream for its local stream set.
 183          */
 184         if (stream->chan->monitor) {
 185                 /* Decrement the stream count of the global consumer data. */
 186                 assert(consumer_data.stream_count > 0);
 187                 consumer_data.stream_count--;
 188         }
 189 }
 190
 191 /*
 192  * Free the given stream within a RCU call.
 193  */
 194 void consumer_stream_free(struct lttng_consumer_stream *stream)
 195 {
 196         assert(stream);
 197
 198         call_rcu(&stream->node.head, free_stream_rcu);
 199 }
 200
 201 /*
 202  * Destroy a stream completely. This will delete, close and free the stream.
 203  * Once return, the stream is NO longer usable. Its channel may get destroyed
 204  * if conditions are met.
 205  *
 206  * This MUST be called WITHOUT the consumer data and stream lock acquired.
 207  */
 208 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 209                 struct lttng_ht *ht)
 210 {
 211         struct lttng_consumer_channel *free_chan = NULL;
 212
 213         assert(stream);
 214
 215         DBG("Consumer stream destroy - wait_fd: %d", stream->wait_fd);
 216
 217         pthread_mutex_lock(&consumer_data.lock);
 218         pthread_mutex_lock(&stream->lock);
 219
 220         /* Remove every reference of the stream in the consumer. */
 221         consumer_stream_delete(stream, ht);
 222
 223         /* Close down everything including the relayd if one. */
 224         consumer_stream_close(stream);
 225
 226         /* Update refcount of channel and see if we need to destroy it. */
 227         if (!uatomic_sub_return(&stream->chan->refcount, 1)
 228                         && !uatomic_read(&stream->chan->nb_init_stream_left)) {
 229                 free_chan = stream->chan;
 230         }
 231
 232         /* Indicates that the consumer data state MUST be updated after this. */
 233         consumer_data.need_update = 1;
 234
 235         pthread_mutex_unlock(&stream->lock);
 236         pthread_mutex_unlock(&consumer_data.lock);
 237
 238         if (free_chan) {
 239                 consumer_del_channel(free_chan);
 240         }
 241
 242         /* Free stream within a RCU call. */
 243         consumer_stream_free(stream);
 244 }