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