src/common/consumer/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 _LGPL_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/index/index.h>
  28 #include <common/kernel-consumer/kernel-consumer.h>
  29 #include <common/relayd/relayd.h>
  30 #include <common/ust-consumer/ust-consumer.h>
  31 #include <common/utils.h>
  32 #include <common/consumer/consumer.h>
  33 #include <common/consumer/consumer-timer.h>
  34 #include <common/consumer/metadata-bucket.h>
  35
  36 #include "consumer-stream.h"
  37
  38 /*
  39  * RCU call to free stream. MUST only be used with call_rcu().
  40  */
  41 static void free_stream_rcu(struct rcu_head *head)
  42 {
  43         struct lttng_ht_node_u64 *node =
  44                 caa_container_of(head, struct lttng_ht_node_u64, head);
  45         struct lttng_consumer_stream *stream =
  46                 caa_container_of(node, struct lttng_consumer_stream, node);
  47
  48         pthread_mutex_destroy(&stream->lock);
  49         free(stream);
  50 }
  51
  52 static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
  53 {
  54         pthread_mutex_lock(&stream->chan->lock);
  55         pthread_mutex_lock(&stream->lock);
  56 }
  57
  58 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
  59 {
  60         pthread_mutex_unlock(&stream->lock);
  61         pthread_mutex_unlock(&stream->chan->lock);
  62 }
  63
  64 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
  65 {
  66         consumer_stream_data_lock_all(stream);
  67         pthread_mutex_lock(&stream->metadata_rdv_lock);
  68 }
  69
  70 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
  71 {
  72         pthread_mutex_unlock(&stream->metadata_rdv_lock);
  73         consumer_stream_data_unlock_all(stream);
  74 }
  75
  76 /* Only used for data streams. */
  77 static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
  78                 const struct stream_subbuffer *subbuf)
  79 {
  80         int ret = 0;
  81         uint64_t sequence_number;
  82         const uint64_t discarded_events = subbuf->info.data.events_discarded;
  83
  84         if (!subbuf->info.data.sequence_number.is_set) {
  85                 /* Command not supported by the tracer. */
  86                 sequence_number = -1ULL;
  87                 stream->sequence_number_unavailable = true;
  88         } else {
  89                 sequence_number = subbuf->info.data.sequence_number.value;
  90         }
  91
  92         /*
  93          * Start the sequence when we extract the first packet in case we don't
  94          * start at 0 (for example if a consumer is not connected to the
  95          * session immediately after the beginning).
  96          */
  97         if (stream->last_sequence_number == -1ULL) {
  98                 stream->last_sequence_number = sequence_number;
  99         } else if (sequence_number > stream->last_sequence_number) {
 100                 stream->chan->lost_packets += sequence_number -
 101                                 stream->last_sequence_number - 1;
 102         } else {
 103                 /* seq <= last_sequence_number */
 104                 ERR("Sequence number inconsistent : prev = %" PRIu64
 105                     ", current = %" PRIu64,
 106                                 stream->last_sequence_number, sequence_number);
 107                 ret = -1;
 108                 goto end;
 109         }
 110         stream->last_sequence_number = sequence_number;
 111
 112         if (discarded_events < stream->last_discarded_events) {
 113                 /*
 114                  * Overflow has occurred. We assume only one wrap-around
 115                  * has occurred.
 116                  */
 117                 stream->chan->discarded_events +=
 118                                 (1ULL << (CAA_BITS_PER_LONG - 1)) -
 119                                 stream->last_discarded_events +
 120                                 discarded_events;
 121         } else {
 122                 stream->chan->discarded_events += discarded_events -
 123                                                   stream->last_discarded_events;
 124         }
 125         stream->last_discarded_events = discarded_events;
 126         ret = 0;
 127
 128 end:
 129         return ret;
 130 }
 131
 132 static
 133 void ctf_packet_index_populate(struct ctf_packet_index *index,
 134                 off_t offset, const struct stream_subbuffer *subbuffer)
 135 {
 136         *index = (typeof(*index)){
 137                 .offset = htobe64(offset),
 138                 .packet_size = htobe64(subbuffer->info.data.packet_size),
 139                 .content_size = htobe64(subbuffer->info.data.content_size),
 140                 .timestamp_begin = htobe64(
 141                                 subbuffer->info.data.timestamp_begin),
 142                 .timestamp_end = htobe64(
 143                                 subbuffer->info.data.timestamp_end),
 144                 .events_discarded = htobe64(
 145                                 subbuffer->info.data.events_discarded),
 146                 .stream_id = htobe64(subbuffer->info.data.stream_id),
 147                 .stream_instance_id = htobe64(
 148                                 subbuffer->info.data.stream_instance_id.is_set ?
 149                                 subbuffer->info.data.stream_instance_id.value : -1ULL),
 150                 .packet_seq_num = htobe64(
 151                                 subbuffer->info.data.sequence_number.is_set ?
 152                                 subbuffer->info.data.sequence_number.value : -1ULL),
 153         };
 154 }
 155
 156 static ssize_t consumer_stream_consume_mmap(
 157                 struct lttng_consumer_local_data *ctx,
 158                 struct lttng_consumer_stream *stream,
 159                 const struct stream_subbuffer *subbuffer)
 160 {
 161         const unsigned long padding_size =
 162                         subbuffer->info.data.padded_subbuf_size -
 163                         subbuffer->info.data.subbuf_size;
 164
 165         return lttng_consumer_on_read_subbuffer_mmap(
 166                         stream, &subbuffer->buffer.buffer, padding_size);
 167 }
 168
 169 static ssize_t consumer_stream_consume_splice(
 170                 struct lttng_consumer_local_data *ctx,
 171                 struct lttng_consumer_stream *stream,
 172                 const struct stream_subbuffer *subbuffer)
 173 {
 174         return lttng_consumer_on_read_subbuffer_splice(ctx, stream,
 175                         subbuffer->info.data.padded_subbuf_size, 0);
 176 }
 177
 178 static int consumer_stream_send_index(
 179                 struct lttng_consumer_stream *stream,
 180                 const struct stream_subbuffer *subbuffer,
 181                 struct lttng_consumer_local_data *ctx)
 182 {
 183         off_t packet_offset = 0;
 184         struct ctf_packet_index index = {};
 185
 186         /*
 187          * This is called after consuming the sub-buffer; substract the
 188          * effect this sub-buffer from the offset.
 189          */
 190         if (stream->net_seq_idx == (uint64_t) -1ULL) {
 191                 packet_offset = stream->out_fd_offset -
 192                                 subbuffer->info.data.padded_subbuf_size;
 193         }
 194
 195         ctf_packet_index_populate(&index, packet_offset, subbuffer);
 196         return consumer_stream_write_index(stream, &index);
 197 }
 198
 199 /*
 200  * Actually do the metadata sync using the given metadata stream.
 201  *
 202  * Return 0 on success else a negative value. ENODATA can be returned also
 203  * indicating that there is no metadata available for that stream.
 204  */
 205 static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 206                 struct lttng_consumer_local_data *ctx)
 207 {
 208         int ret;
 209
 210         assert(metadata);
 211         assert(metadata->metadata_flag);
 212         assert(ctx);
 213
 214         /*
 215          * In UST, since we have to write the metadata from the cache packet
 216          * by packet, we might need to start this procedure multiple times
 217          * until all the metadata from the cache has been extracted.
 218          */
 219         do {
 220                 /*
 221                  * Steps :
 222                  * - Lock the metadata stream
 223                  * - Check if metadata stream node was deleted before locking.
 224                  *   - if yes, release and return success
 225                  * - Check if new metadata is ready (flush + snapshot pos)
 226                  * - If nothing : release and return.
 227                  * - Lock the metadata_rdv_lock
 228                  * - Unlock the metadata stream
 229                  * - cond_wait on metadata_rdv to wait the wakeup from the
 230                  *   metadata thread
 231                  * - Unlock the metadata_rdv_lock
 232                  */
 233                 pthread_mutex_lock(&metadata->lock);
 234
 235                 /*
 236                  * There is a possibility that we were able to acquire a reference on the
 237                  * stream from the RCU hash table but between then and now, the node might
 238                  * have been deleted just before the lock is acquired. Thus, after locking,
 239                  * we make sure the metadata node has not been deleted which means that the
 240                  * buffers are closed.
 241                  *
 242                  * In that case, there is no need to sync the metadata hence returning a
 243                  * success return code.
 244                  */
 245                 ret = cds_lfht_is_node_deleted(&metadata->node.node);
 246                 if (ret) {
 247                         ret = 0;
 248                         goto end_unlock_mutex;
 249                 }
 250
 251                 switch (ctx->type) {
 252                 case LTTNG_CONSUMER_KERNEL:
 253                         /*
 254                          * Empty the metadata cache and flush the current stream.
 255                          */
 256                         ret = lttng_kconsumer_sync_metadata(metadata);
 257                         break;
 258                 case LTTNG_CONSUMER32_UST:
 259                 case LTTNG_CONSUMER64_UST:
 260                         /*
 261                          * Ask the sessiond if we have new metadata waiting and update the
 262                          * consumer metadata cache.
 263                          */
 264                         ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
 265                         break;
 266                 default:
 267                         assert(0);
 268                         ret = -1;
 269                         break;
 270                 }
 271                 /*
 272                  * Error or no new metadata, we exit here.
 273                  */
 274                 if (ret <= 0 || ret == ENODATA) {
 275                         goto end_unlock_mutex;
 276                 }
 277
 278                 /*
 279                  * At this point, new metadata have been flushed, so we wait on the
 280                  * rendez-vous point for the metadata thread to wake us up when it
 281                  * finishes consuming the metadata and continue execution.
 282                  */
 283
 284                 pthread_mutex_lock(&metadata->metadata_rdv_lock);
 285
 286                 /*
 287                  * Release metadata stream lock so the metadata thread can process it.
 288                  */
 289                 pthread_mutex_unlock(&metadata->lock);
 290
 291                 /*
 292                  * Wait on the rendez-vous point. Once woken up, it means the metadata was
 293                  * consumed and thus synchronization is achieved.
 294                  */
 295                 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
 296                 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
 297         } while (ret == EAGAIN);
 298
 299         /* Success */
 300         return 0;
 301
 302 end_unlock_mutex:
 303         pthread_mutex_unlock(&metadata->lock);
 304         return ret;
 305 }
 306
 307 /*
 308  * Synchronize the metadata using a given session ID. A successful acquisition
 309  * of a metadata stream will trigger a request to the session daemon and a
 310  * snapshot so the metadata thread can consume it.
 311  *
 312  * This function call is a rendez-vous point between the metadata thread and
 313  * the data thread.
 314  *
 315  * Return 0 on success or else a negative value.
 316  */
 317 int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
 318                 uint64_t session_id)
 319 {
 320         int ret;
 321         struct lttng_consumer_stream *stream = NULL;
 322         struct lttng_ht_iter iter;
 323         struct lttng_ht *ht;
 324
 325         assert(ctx);
 326
 327         /* Ease our life a bit. */
 328         ht = consumer_data.stream_list_ht;
 329
 330         rcu_read_lock();
 331
 332         /* Search the metadata associated with the session id of the given stream. */
 333
 334         cds_lfht_for_each_entry_duplicate(ht->ht,
 335                         ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
 336                         &session_id, &iter.iter, stream, node_session_id.node) {
 337                 if (!stream->metadata_flag) {
 338                         continue;
 339                 }
 340
 341                 ret = do_sync_metadata(stream, ctx);
 342                 if (ret < 0) {
 343                         goto end;
 344                 }
 345         }
 346
 347         /*
 348          * Force return code to 0 (success) since ret might be ENODATA for instance
 349          * which is not an error but rather that we should come back.
 350          */
 351         ret = 0;
 352
 353 end:
 354         rcu_read_unlock();
 355         return ret;
 356 }
 357
 358 static int consumer_stream_sync_metadata_index(
 359                 struct lttng_consumer_stream *stream,
 360                 const struct stream_subbuffer *subbuffer,
 361                 struct lttng_consumer_local_data *ctx)
 362 {
 363         int ret;
 364
 365         /* Block until all the metadata is sent. */
 366         pthread_mutex_lock(&stream->metadata_timer_lock);
 367         assert(!stream->missed_metadata_flush);
 368         stream->waiting_on_metadata = true;
 369         pthread_mutex_unlock(&stream->metadata_timer_lock);
 370
 371         ret = consumer_stream_sync_metadata(ctx, stream->session_id);
 372
 373         pthread_mutex_lock(&stream->metadata_timer_lock);
 374         stream->waiting_on_metadata = false;
 375         if (stream->missed_metadata_flush) {
 376                 stream->missed_metadata_flush = false;
 377                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 378                 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
 379         } else {
 380                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 381         }
 382         if (ret < 0) {
 383                 goto end;
 384         }
 385
 386         ret = consumer_stream_send_index(stream, subbuffer, ctx);
 387 end:
 388         return ret;
 389 }
 390
 391 /*
 392  * Check if the local version of the metadata stream matches with the version
 393  * of the metadata stream in the kernel. If it was updated, set the reset flag
 394  * on the stream.
 395  */
 396 static
 397 int metadata_stream_check_version(struct lttng_consumer_stream *stream,
 398         const struct stream_subbuffer *subbuffer)
 399 {
 400         if (stream->metadata_version == subbuffer->info.metadata.version) {
 401                 goto end;
 402         }
 403
 404         DBG("New metadata version detected");
 405         consumer_stream_metadata_set_version(stream,
 406                         subbuffer->info.metadata.version);
 407
 408         if (stream->read_subbuffer_ops.reset_metadata) {
 409                 stream->read_subbuffer_ops.reset_metadata(stream);
 410         }
 411
 412 end:
 413         return 0;
 414 }
 415
 416 struct lttng_consumer_stream *consumer_stream_create(
 417                 struct lttng_consumer_channel *channel,
 418                 uint64_t channel_key,
 419                 uint64_t stream_key,
 420                 const char *channel_name,
 421                 uint64_t relayd_id,
 422                 uint64_t session_id,
 423                 struct lttng_trace_chunk *trace_chunk,
 424                 int cpu,
 425                 int *alloc_ret,
 426                 enum consumer_channel_type type,
 427                 unsigned int monitor)
 428 {
 429         int ret;
 430         struct lttng_consumer_stream *stream;
 431
 432         stream = zmalloc(sizeof(*stream));
 433         if (stream == NULL) {
 434                 PERROR("malloc struct lttng_consumer_stream");
 435                 ret = -ENOMEM;
 436                 goto end;
 437         }
 438
 439         if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
 440                 ERR("Failed to acquire trace chunk reference during the creation of a stream");
 441                 ret = -1;
 442                 goto error;
 443         }
 444
 445         rcu_read_lock();
 446         stream->chan = channel;
 447         stream->key = stream_key;
 448         stream->trace_chunk = trace_chunk;
 449         stream->out_fd = -1;
 450         stream->out_fd_offset = 0;
 451         stream->output_written = 0;
 452         stream->net_seq_idx = relayd_id;
 453         stream->session_id = session_id;
 454         stream->monitor = monitor;
 455         stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
 456         stream->index_file = NULL;
 457         stream->last_sequence_number = -1ULL;
 458         stream->rotate_position = -1ULL;
 459         pthread_mutex_init(&stream->lock, NULL);
 460         pthread_mutex_init(&stream->metadata_timer_lock, NULL);
 461
 462         /* If channel is the metadata, flag this stream as metadata. */
 463         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 464                 stream->metadata_flag = 1;
 465                 /* Metadata is flat out. */
 466                 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
 467                 /* Live rendez-vous point. */
 468                 pthread_cond_init(&stream->metadata_rdv, NULL);
 469                 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
 470         } else {
 471                 /* Format stream name to <channel_name>_<cpu_number> */
 472                 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
 473                                 channel_name, cpu);
 474                 if (ret < 0) {
 475                         PERROR("snprintf stream name");
 476                         goto error;
 477                 }
 478         }
 479
 480         switch (channel->output) {
 481         case CONSUMER_CHANNEL_SPLICE:
 482                 stream->output = LTTNG_EVENT_SPLICE;
 483                 ret = utils_create_pipe(stream->splice_pipe);
 484                 if (ret < 0) {
 485                         goto error;
 486                 }
 487                 break;
 488         case CONSUMER_CHANNEL_MMAP:
 489                 stream->output = LTTNG_EVENT_MMAP;
 490                 break;
 491         default:
 492                 abort();
 493         }
 494
 495         /* Key is always the wait_fd for streams. */
 496         lttng_ht_node_init_u64(&stream->node, stream->key);
 497
 498         /* Init node per channel id key */
 499         lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
 500
 501         /* Init session id node with the stream session id */
 502         lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
 503
 504         DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
 505                         " relayd_id %" PRIu64 ", session_id %" PRIu64,
 506                         stream->name, stream->key, channel_key,
 507                         stream->net_seq_idx, stream->session_id);
 508
 509         rcu_read_unlock();
 510
 511         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 512                 stream->read_subbuffer_ops.lock =
 513                                 consumer_stream_metadata_lock_all;
 514                 stream->read_subbuffer_ops.unlock =
 515                                 consumer_stream_metadata_unlock_all;
 516                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 517                                 metadata_stream_check_version;
 518         } else {
 519                 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
 520                 stream->read_subbuffer_ops.unlock =
 521                                 consumer_stream_data_unlock_all;
 522                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 523                                 consumer_stream_update_stats;
 524                 if (channel->is_live) {
 525                         stream->read_subbuffer_ops.post_consume =
 526                                         consumer_stream_sync_metadata_index;
 527                 } else {
 528                         stream->read_subbuffer_ops.post_consume =
 529                                         consumer_stream_send_index;
 530                 }
 531         }
 532
 533         if (channel->output == CONSUMER_CHANNEL_MMAP) {
 534                 stream->read_subbuffer_ops.consume_subbuffer =
 535                                 consumer_stream_consume_mmap;
 536         } else {
 537                 stream->read_subbuffer_ops.consume_subbuffer =
 538                                 consumer_stream_consume_splice;
 539         }
 540
 541         return stream;
 542
 543 error:
 544         rcu_read_unlock();
 545         lttng_trace_chunk_put(stream->trace_chunk);
 546         free(stream);
 547 end:
 548         if (alloc_ret) {
 549                 *alloc_ret = ret;
 550         }
 551         return NULL;
 552 }
 553
 554 /*
 555  * Close stream on the relayd side. This call can destroy a relayd if the
 556  * conditions are met.
 557  *
 558  * A RCU read side lock MUST be acquired if the relayd object was looked up in
 559  * a hash table before calling this.
 560  */
 561 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
 562                 struct consumer_relayd_sock_pair *relayd)
 563 {
 564         int ret;
 565
 566         assert(stream);
 567         assert(relayd);
 568
 569         if (stream->sent_to_relayd) {
 570                 uatomic_dec(&relayd->refcount);
 571                 assert(uatomic_read(&relayd->refcount) >= 0);
 572         }
 573
 574         /* Closing streams requires to lock the control socket. */
 575         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 576         ret = relayd_send_close_stream(&relayd->control_sock,
 577                         stream->relayd_stream_id,
 578                         stream->next_net_seq_num - 1);
 579         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 580         if (ret < 0) {
 581                 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 582                 lttng_consumer_cleanup_relayd(relayd);
 583         }
 584
 585         /* Both conditions are met, we destroy the relayd. */
 586         if (uatomic_read(&relayd->refcount) == 0 &&
 587                         uatomic_read(&relayd->destroy_flag)) {
 588                 consumer_destroy_relayd(relayd);
 589         }
 590         stream->net_seq_idx = (uint64_t) -1ULL;
 591         stream->sent_to_relayd = 0;
 592 }
 593
 594 /*
 595  * Close stream's file descriptors and, if needed, close stream also on the
 596  * relayd side.
 597  *
 598  * The consumer data lock MUST be acquired.
 599  * The stream lock MUST be acquired.
 600  */
 601 void consumer_stream_close(struct lttng_consumer_stream *stream)
 602 {
 603         int ret;
 604         struct consumer_relayd_sock_pair *relayd;
 605
 606         assert(stream);
 607
 608         switch (consumer_data.type) {
 609         case LTTNG_CONSUMER_KERNEL:
 610                 if (stream->mmap_base != NULL) {
 611                         ret = munmap(stream->mmap_base, stream->mmap_len);
 612                         if (ret != 0) {
 613                                 PERROR("munmap");
 614                         }
 615                 }
 616
 617                 if (stream->wait_fd >= 0) {
 618                         ret = close(stream->wait_fd);
 619                         if (ret) {
 620                                 PERROR("close");
 621                         }
 622                         stream->wait_fd = -1;
 623                 }
 624                 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
 625                         utils_close_pipe(stream->splice_pipe);
 626                 }
 627                 break;
 628         case LTTNG_CONSUMER32_UST:
 629         case LTTNG_CONSUMER64_UST:
 630         {
 631                 /*
 632                  * Special case for the metadata since the wait fd is an internal pipe
 633                  * polled in the metadata thread.
 634                  */
 635                 if (stream->metadata_flag && stream->chan->monitor) {
 636                         int rpipe = stream->ust_metadata_poll_pipe[0];
 637
 638                         /*
 639                          * This will stop the channel timer if one and close the write side
 640                          * of the metadata poll pipe.
 641                          */
 642                         lttng_ustconsumer_close_metadata(stream->chan);
 643                         if (rpipe >= 0) {
 644                                 ret = close(rpipe);
 645                                 if (ret < 0) {
 646                                         PERROR("closing metadata pipe read side");
 647                                 }
 648                                 stream->ust_metadata_poll_pipe[0] = -1;
 649                         }
 650                 }
 651                 break;
 652         }
 653         default:
 654                 ERR("Unknown consumer_data type");
 655                 assert(0);
 656         }
 657
 658         /* Close output fd. Could be a socket or local file at this point. */
 659         if (stream->out_fd >= 0) {
 660                 ret = close(stream->out_fd);
 661                 if (ret) {
 662                         PERROR("close");
 663                 }
 664                 stream->out_fd = -1;
 665         }
 666
 667         if (stream->index_file) {
 668                 lttng_index_file_put(stream->index_file);
 669                 stream->index_file = NULL;
 670         }
 671
 672         lttng_trace_chunk_put(stream->trace_chunk);
 673         stream->trace_chunk = NULL;
 674
 675         /* Check and cleanup relayd if needed. */
 676         rcu_read_lock();
 677         relayd = consumer_find_relayd(stream->net_seq_idx);
 678         if (relayd != NULL) {
 679                 consumer_stream_relayd_close(stream, relayd);
 680         }
 681         rcu_read_unlock();
 682 }
 683
 684 /*
 685  * Delete the stream from all possible hash tables.
 686  *
 687  * The consumer data lock MUST be acquired.
 688  * The stream lock MUST be acquired.
 689  */
 690 void consumer_stream_delete(struct lttng_consumer_stream *stream,
 691                 struct lttng_ht *ht)
 692 {
 693         int ret;
 694         struct lttng_ht_iter iter;
 695
 696         assert(stream);
 697         /* Should NEVER be called not in monitor mode. */
 698         assert(stream->chan->monitor);
 699
 700         rcu_read_lock();
 701
 702         if (ht) {
 703                 iter.iter.node = &stream->node.node;
 704                 ret = lttng_ht_del(ht, &iter);
 705                 assert(!ret);
 706         }
 707
 708         /* Delete from stream per channel ID hash table. */
 709         iter.iter.node = &stream->node_channel_id.node;
 710         /*
 711          * The returned value is of no importance. Even if the node is NOT in the
 712          * hash table, we continue since we may have been called by a code path
 713          * that did not add the stream to a (all) hash table. Same goes for the
 714          * next call ht del call.
 715          */
 716         (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
 717
 718         /* Delete from the global stream list. */
 719         iter.iter.node = &stream->node_session_id.node;
 720         /* See the previous ht del on why we ignore the returned value. */
 721         (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
 722
 723         rcu_read_unlock();
 724
 725         if (!stream->metadata_flag) {
 726                 /* Decrement the stream count of the global consumer data. */
 727                 assert(consumer_data.stream_count > 0);
 728                 consumer_data.stream_count--;
 729         }
 730 }
 731
 732 /*
 733  * Free the given stream within a RCU call.
 734  */
 735 void consumer_stream_free(struct lttng_consumer_stream *stream)
 736 {
 737         assert(stream);
 738
 739         metadata_bucket_destroy(stream->metadata_bucket);
 740         call_rcu(&stream->node.head, free_stream_rcu);
 741 }
 742
 743 /*
 744  * Destroy the stream's buffers of the tracer.
 745  */
 746 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 747 {
 748         assert(stream);
 749
 750         switch (consumer_data.type) {
 751         case LTTNG_CONSUMER_KERNEL:
 752                 break;
 753         case LTTNG_CONSUMER32_UST:
 754         case LTTNG_CONSUMER64_UST:
 755                 lttng_ustconsumer_del_stream(stream);
 756                 break;
 757         default:
 758                 ERR("Unknown consumer_data type");
 759                 assert(0);
 760         }
 761 }
 762
 763 /*
 764  * Destroy and close a already created stream.
 765  */
 766 static void destroy_close_stream(struct lttng_consumer_stream *stream)
 767 {
 768         assert(stream);
 769
 770         DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
 771
 772         /* Destroy tracer buffers of the stream. */
 773         consumer_stream_destroy_buffers(stream);
 774         /* Close down everything including the relayd if one. */
 775         consumer_stream_close(stream);
 776 }
 777
 778 /*
 779  * Decrement the stream's channel refcount and if down to 0, return the channel
 780  * pointer so it can be destroyed by the caller or NULL if not.
 781  */
 782 static struct lttng_consumer_channel *unref_channel(
 783                 struct lttng_consumer_stream *stream)
 784 {
 785         struct lttng_consumer_channel *free_chan = NULL;
 786
 787         assert(stream);
 788         assert(stream->chan);
 789
 790         /* Update refcount of channel and see if we need to destroy it. */
 791         if (!uatomic_sub_return(&stream->chan->refcount, 1)
 792                         && !uatomic_read(&stream->chan->nb_init_stream_left)) {
 793                 free_chan = stream->chan;
 794         }
 795
 796         return free_chan;
 797 }
 798
 799 /*
 800  * Destroy a stream completely. This will delete, close and free the stream.
 801  * Once return, the stream is NO longer usable. Its channel may get destroyed
 802  * if conditions are met for a monitored stream.
 803  *
 804  * This MUST be called WITHOUT the consumer data and stream lock acquired if
 805  * the stream is in _monitor_ mode else it does not matter.
 806  */
 807 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 808                 struct lttng_ht *ht)
 809 {
 810         assert(stream);
 811
 812         /* Stream is in monitor mode. */
 813         if (stream->monitor) {
 814                 struct lttng_consumer_channel *free_chan = NULL;
 815
 816                 /*
 817                  * This means that the stream was successfully removed from the streams
 818                  * list of the channel and sent to the right thread managing this
 819                  * stream thus being globally visible.
 820                  */
 821                 if (stream->globally_visible) {
 822                         pthread_mutex_lock(&consumer_data.lock);
 823                         pthread_mutex_lock(&stream->chan->lock);
 824                         pthread_mutex_lock(&stream->lock);
 825                         /* Remove every reference of the stream in the consumer. */
 826                         consumer_stream_delete(stream, ht);
 827
 828                         destroy_close_stream(stream);
 829
 830                         /* Update channel's refcount of the stream. */
 831                         free_chan = unref_channel(stream);
 832
 833                         /* Indicates that the consumer data state MUST be updated after this. */
 834                         consumer_data.need_update = 1;
 835
 836                         pthread_mutex_unlock(&stream->lock);
 837                         pthread_mutex_unlock(&stream->chan->lock);
 838                         pthread_mutex_unlock(&consumer_data.lock);
 839                 } else {
 840                         /*
 841                          * If the stream is not visible globally, this needs to be done
 842                          * outside of the consumer data lock section.
 843                          */
 844                         free_chan = unref_channel(stream);
 845                 }
 846
 847                 if (free_chan) {
 848                         consumer_del_channel(free_chan);
 849                 }
 850         } else {
 851                 destroy_close_stream(stream);
 852         }
 853
 854         /* Free stream within a RCU call. */
 855         lttng_trace_chunk_put(stream->trace_chunk);
 856         stream->trace_chunk = NULL;
 857         consumer_stream_free(stream);
 858 }
 859
 860 /*
 861  * Write index of a specific stream either on the relayd or local disk.
 862  *
 863  * Return 0 on success or else a negative value.
 864  */
 865 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
 866                 struct ctf_packet_index *element)
 867 {
 868         int ret;
 869
 870         assert(stream);
 871         assert(element);
 872
 873         rcu_read_lock();
 874         if (stream->net_seq_idx != (uint64_t) -1ULL) {
 875                 struct consumer_relayd_sock_pair *relayd;
 876                 relayd = consumer_find_relayd(stream->net_seq_idx);
 877                 if (relayd) {
 878                         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 879                         ret = relayd_send_index(&relayd->control_sock, element,
 880                                 stream->relayd_stream_id, stream->next_net_seq_num - 1);
 881                         if (ret < 0) {
 882                                 /*
 883                                  * Communication error with lttng-relayd,
 884                                  * perform cleanup now
 885                                  */
 886                                 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 887                                 lttng_consumer_cleanup_relayd(relayd);
 888                                 ret = -1;
 889                         }
 890                         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 891                 } else {
 892                         ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
 893                                         stream->key, stream->net_seq_idx);
 894                         ret = -1;
 895                 }
 896         } else {
 897                 if (lttng_index_file_write(stream->index_file, element)) {
 898                         ret = -1;
 899                 } else {
 900                         ret = 0;
 901                 }
 902         }
 903         if (ret < 0) {
 904                 goto error;
 905         }
 906
 907 error:
 908         rcu_read_unlock();
 909         return ret;
 910 }
 911
 912 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
 913                 bool create_index)
 914 {
 915         int ret;
 916         enum lttng_trace_chunk_status chunk_status;
 917         const int flags = O_WRONLY | O_CREAT | O_TRUNC;
 918         const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
 919         char stream_path[LTTNG_PATH_MAX];
 920
 921         ASSERT_LOCKED(stream->lock);
 922         assert(stream->trace_chunk);
 923
 924         ret = utils_stream_file_path(stream->chan->pathname, stream->name,
 925                         stream->chan->tracefile_size,
 926                         stream->tracefile_count_current, NULL,
 927                         stream_path, sizeof(stream_path));
 928         if (ret < 0) {
 929                 goto end;
 930         }
 931
 932         if (stream->out_fd >= 0) {
 933                 ret = close(stream->out_fd);
 934                 if (ret < 0) {
 935                         PERROR("Failed to close stream file \"%s\"",
 936                                         stream->name);
 937                         goto end;
 938                 }
 939                 stream->out_fd = -1;
 940         }
 941
 942         DBG("Opening stream output file \"%s\"", stream_path);
 943         chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
 944                         flags, mode, &stream->out_fd);
 945         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 946                 ERR("Failed to open stream file \"%s\"", stream->name);
 947                 ret = -1;
 948                 goto end;
 949         }
 950
 951         if (!stream->metadata_flag && (create_index || stream->index_file)) {
 952                 if (stream->index_file) {
 953                         lttng_index_file_put(stream->index_file);
 954                 }
 955                 stream->index_file = lttng_index_file_create_from_trace_chunk(
 956                                 stream->trace_chunk,
 957                                 stream->chan->pathname,
 958                                 stream->name,
 959                                 stream->chan->tracefile_size,
 960                                 stream->tracefile_count_current,
 961                                 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
 962                                 false);
 963                 if (!stream->index_file) {
 964                         ret = -1;
 965                         goto end;
 966                 }
 967         }
 968
 969         /* Reset current size because we just perform a rotation. */
 970         stream->tracefile_size_current = 0;
 971         stream->out_fd_offset = 0;
 972 end:
 973         return ret;
 974 }
 975
 976 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
 977 {
 978         int ret;
 979
 980         stream->tracefile_count_current++;
 981         if (stream->chan->tracefile_count > 0) {
 982                 stream->tracefile_count_current %=
 983                                 stream->chan->tracefile_count;
 984         }
 985
 986         DBG("Rotating output files of stream \"%s\"", stream->name);
 987         ret = consumer_stream_create_output_files(stream, true);
 988         if (ret) {
 989                 goto end;
 990         }
 991
 992 end:
 993         return ret;
 994 }
 995
 996 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
 997 {
 998         /*
 999          * This function does not take a const stream since
1000          * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1001          */
1002         assert(stream);
1003         return cds_lfht_is_node_deleted(&stream->node.node);
1004 }
1005
1006 static ssize_t metadata_bucket_flush(
1007                 const struct stream_subbuffer *buffer, void *data)
1008 {
1009         ssize_t ret;
1010         struct lttng_consumer_stream *stream = data;
1011
1012         ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1013         if (ret < 0) {
1014                 goto end;
1015         }
1016 end:
1017         return ret;
1018 }
1019
1020 static ssize_t metadata_bucket_consume(
1021                 struct lttng_consumer_local_data *unused,
1022                 struct lttng_consumer_stream *stream,
1023                 const struct stream_subbuffer *subbuffer)
1024 {
1025         ssize_t ret;
1026         enum metadata_bucket_status status;
1027
1028         status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1029         switch (status) {
1030         case METADATA_BUCKET_STATUS_OK:
1031                 /* Return consumed size. */
1032                 ret = subbuffer->buffer.buffer.size;
1033                 break;
1034         default:
1035                 ret = -1;
1036         }
1037
1038         return ret;
1039 }
1040
1041 int consumer_stream_enable_metadata_bucketization(
1042                 struct lttng_consumer_stream *stream)
1043 {
1044         int ret = 0;
1045
1046         assert(stream->metadata_flag);
1047         assert(!stream->metadata_bucket);
1048         assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1049
1050         stream->metadata_bucket = metadata_bucket_create(
1051                         metadata_bucket_flush, stream);
1052         if (!stream->metadata_bucket) {
1053                 ret = -1;
1054                 goto end;
1055         }
1056
1057         stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1058 end:
1059         return ret;
1060 }
1061
1062 void consumer_stream_metadata_set_version(
1063                 struct lttng_consumer_stream *stream, uint64_t new_version)
1064 {
1065         assert(new_version > stream->metadata_version);
1066         stream->metadata_version = new_version;
1067         stream->reset_metadata_flag = 1;
1068
1069         if (stream->metadata_bucket) {
1070                 metadata_bucket_reset(stream->metadata_bucket);
1071         }
1072 }