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 <inttypes.h>
  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         }
  84         stream->net_seq_idx = (uint64_t) -1ULL;
  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                         }
 115                         stream->wait_fd = -1;
 116                 }
 117                 break;
 118         case LTTNG_CONSUMER32_UST:
 119         case LTTNG_CONSUMER64_UST:
 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                 }
 132                 stream->out_fd = -1;
 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);
 157         /* Should NEVER be called not in monitor mode. */
 158         assert(stream->chan->monitor);
 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
 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  * 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 /*
 201  * Destroy the stream's buffers of the tracer.
 202  */
 203 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 204 {
 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 }
 219
 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 {
 234         assert(stream);
 235
 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 }
 243
 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);
 253
 254         /* Remove every reference of the stream in the consumer. */
 255         consumer_stream_delete(stream, ht);
 256         /* Destroy tracer buffers of the stream. */
 257         consumer_stream_destroy_buffers(stream);
 258         /* Close down everything including the relayd if one. */
 259         consumer_stream_close(stream);
 260 }
 261
 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;
 278
 279                 pthread_mutex_lock(&consumer_data.lock);
 280                 pthread_mutex_lock(&stream->lock);
 281
 282                 destroy_monitor(stream, ht);
 283
 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);
 305         }
 306
 307         /* Free stream within a RCU call. */
 308         consumer_stream_free(stream);
 309 }