src/common/consumer/consumer-stream.c

   1 /*
   2  * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
   3  * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
   5  *
   6  * SPDX-License-Identifier: GPL-2.0-only
   7  *
   8  */
   9
  10 #define _LGPL_SOURCE
  11 #include <assert.h>
  12 #include <inttypes.h>
  13 #include <sys/mman.h>
  14 #include <unistd.h>
  15
  16 #include <common/common.h>
  17 #include <common/consumer/consumer-timer.h>
  18 #include <common/consumer/consumer-timer.h>
  19 #include <common/consumer/consumer.h>
  20 #include <common/consumer/consumer.h>
  21 #include <common/consumer/metadata-bucket.h>
  22 #include <common/consumer/metadata-bucket.h>
  23 #include <common/index/index.h>
  24 #include <common/kernel-consumer/kernel-consumer.h>
  25 #include <common/kernel-ctl/kernel-ctl.h>
  26 #include <common/macros.h>
  27 #include <common/relayd/relayd.h>
  28 #include <common/ust-consumer/ust-consumer.h>
  29 #include <common/utils.h>
  30
  31 #include "consumer-stream.h"
  32
  33 /*
  34  * RCU call to free stream. MUST only be used with call_rcu().
  35  */
  36 static void free_stream_rcu(struct rcu_head *head)
  37 {
  38         struct lttng_ht_node_u64 *node =
  39                 caa_container_of(head, struct lttng_ht_node_u64, head);
  40         struct lttng_consumer_stream *stream =
  41                 caa_container_of(node, struct lttng_consumer_stream, node);
  42
  43         pthread_mutex_destroy(&stream->lock);
  44         free(stream);
  45 }
  46
  47 static void consumer_stream_data_lock_all(struct lttng_consumer_stream *stream)
  48 {
  49         pthread_mutex_lock(&stream->chan->lock);
  50         pthread_mutex_lock(&stream->lock);
  51 }
  52
  53 static void consumer_stream_data_unlock_all(struct lttng_consumer_stream *stream)
  54 {
  55         pthread_mutex_unlock(&stream->lock);
  56         pthread_mutex_unlock(&stream->chan->lock);
  57 }
  58
  59 static void consumer_stream_data_assert_locked_all(struct lttng_consumer_stream *stream)
  60 {
  61         ASSERT_LOCKED(stream->lock);
  62         ASSERT_LOCKED(stream->chan->lock);
  63 }
  64
  65 static void consumer_stream_metadata_lock_all(struct lttng_consumer_stream *stream)
  66 {
  67         consumer_stream_data_lock_all(stream);
  68         pthread_mutex_lock(&stream->metadata_rdv_lock);
  69 }
  70
  71 static void consumer_stream_metadata_unlock_all(struct lttng_consumer_stream *stream)
  72 {
  73         pthread_mutex_unlock(&stream->metadata_rdv_lock);
  74         consumer_stream_data_unlock_all(stream);
  75 }
  76
  77 static void consumer_stream_metadata_assert_locked_all(struct lttng_consumer_stream *stream)
  78 {
  79         ASSERT_LOCKED(stream->metadata_rdv_lock);
  80         consumer_stream_data_assert_locked_all(stream);
  81 }
  82
  83 /* Only used for data streams. */
  84 static int consumer_stream_update_stats(struct lttng_consumer_stream *stream,
  85                 const struct stream_subbuffer *subbuf)
  86 {
  87         int ret = 0;
  88         uint64_t sequence_number;
  89         const uint64_t discarded_events = subbuf->info.data.events_discarded;
  90
  91         if (!subbuf->info.data.sequence_number.is_set) {
  92                 /* Command not supported by the tracer. */
  93                 sequence_number = -1ULL;
  94                 stream->sequence_number_unavailable = true;
  95         } else {
  96                 sequence_number = subbuf->info.data.sequence_number.value;
  97         }
  98
  99         /*
 100          * Start the sequence when we extract the first packet in case we don't
 101          * start at 0 (for example if a consumer is not connected to the
 102          * session immediately after the beginning).
 103          */
 104         if (stream->last_sequence_number == -1ULL) {
 105                 stream->last_sequence_number = sequence_number;
 106         } else if (sequence_number > stream->last_sequence_number) {
 107                 stream->chan->lost_packets += sequence_number -
 108                                 stream->last_sequence_number - 1;
 109         } else {
 110                 /* seq <= last_sequence_number */
 111                 ERR("Sequence number inconsistent : prev = %" PRIu64
 112                     ", current = %" PRIu64,
 113                                 stream->last_sequence_number, sequence_number);
 114                 ret = -1;
 115                 goto end;
 116         }
 117         stream->last_sequence_number = sequence_number;
 118
 119         if (discarded_events < stream->last_discarded_events) {
 120                 /*
 121                  * Overflow has occurred. We assume only one wrap-around
 122                  * has occurred.
 123                  */
 124                 stream->chan->discarded_events +=
 125                                 (1ULL << (CAA_BITS_PER_LONG - 1)) -
 126                                 stream->last_discarded_events +
 127                                 discarded_events;
 128         } else {
 129                 stream->chan->discarded_events += discarded_events -
 130                                                   stream->last_discarded_events;
 131         }
 132         stream->last_discarded_events = discarded_events;
 133         ret = 0;
 134
 135 end:
 136         return ret;
 137 }
 138
 139 static
 140 void ctf_packet_index_populate(struct ctf_packet_index *index,
 141                 off_t offset, const struct stream_subbuffer *subbuffer)
 142 {
 143         *index = (typeof(*index)){
 144                 .offset = htobe64(offset),
 145                 .packet_size = htobe64(subbuffer->info.data.packet_size),
 146                 .content_size = htobe64(subbuffer->info.data.content_size),
 147                 .timestamp_begin = htobe64(
 148                                 subbuffer->info.data.timestamp_begin),
 149                 .timestamp_end = htobe64(
 150                                 subbuffer->info.data.timestamp_end),
 151                 .events_discarded = htobe64(
 152                                 subbuffer->info.data.events_discarded),
 153                 .stream_id = htobe64(subbuffer->info.data.stream_id),
 154                 .stream_instance_id = htobe64(
 155                                 subbuffer->info.data.stream_instance_id.is_set ?
 156                                 subbuffer->info.data.stream_instance_id.value : -1ULL),
 157                 .packet_seq_num = htobe64(
 158                                 subbuffer->info.data.sequence_number.is_set ?
 159                                 subbuffer->info.data.sequence_number.value : -1ULL),
 160         };
 161 }
 162
 163 static ssize_t consumer_stream_consume_mmap(
 164                 struct lttng_consumer_local_data *ctx,
 165                 struct lttng_consumer_stream *stream,
 166                 const struct stream_subbuffer *subbuffer)
 167 {
 168         const unsigned long padding_size =
 169                         subbuffer->info.data.padded_subbuf_size -
 170                         subbuffer->info.data.subbuf_size;
 171         const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_mmap(
 172                         stream, &subbuffer->buffer.buffer, padding_size);
 173
 174         if (stream->net_seq_idx == -1ULL) {
 175                 /*
 176                  * When writing on disk, check that only the subbuffer (no
 177                  * padding) was written to disk.
 178                  */
 179                 if (written_bytes != subbuffer->info.data.padded_subbuf_size) {
 180                         DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
 181                                         written_bytes,
 182                                         subbuffer->info.data.padded_subbuf_size);
 183                 }
 184         } else {
 185                 /*
 186                  * When streaming over the network, check that the entire
 187                  * subbuffer including padding was successfully written.
 188                  */
 189                 if (written_bytes != subbuffer->info.data.subbuf_size) {
 190                         DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
 191                                         written_bytes,
 192                                         subbuffer->info.data.subbuf_size);
 193                 }
 194         }
 195
 196         /*
 197          * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
 198          * it along to the caller, else return zero.
 199          */
 200         if (written_bytes < 0) {
 201                 ERR("Error reading mmap subbuffer: %zd", written_bytes);
 202         }
 203
 204         return written_bytes;
 205 }
 206
 207 static ssize_t consumer_stream_consume_splice(
 208                 struct lttng_consumer_local_data *ctx,
 209                 struct lttng_consumer_stream *stream,
 210                 const struct stream_subbuffer *subbuffer)
 211 {
 212         const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_splice(
 213                         ctx, stream, subbuffer->info.data.padded_subbuf_size, 0);
 214
 215         if (written_bytes != subbuffer->info.data.padded_subbuf_size) {
 216                 DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
 217                                 written_bytes,
 218                                 subbuffer->info.data.padded_subbuf_size);
 219         }
 220
 221         /*
 222          * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
 223          * pass it along to the caller, else return zero.
 224          */
 225         if (written_bytes < 0) {
 226                 ERR("Error reading splice subbuffer: %zd", written_bytes);
 227         }
 228
 229         return written_bytes;
 230 }
 231
 232 static int consumer_stream_send_index(
 233                 struct lttng_consumer_stream *stream,
 234                 const struct stream_subbuffer *subbuffer,
 235                 struct lttng_consumer_local_data *ctx)
 236 {
 237         off_t packet_offset = 0;
 238         struct ctf_packet_index index = {};
 239
 240         /*
 241          * This is called after consuming the sub-buffer; substract the
 242          * effect this sub-buffer from the offset.
 243          */
 244         if (stream->net_seq_idx == (uint64_t) -1ULL) {
 245                 packet_offset = stream->out_fd_offset -
 246                                 subbuffer->info.data.padded_subbuf_size;
 247         }
 248
 249         ctf_packet_index_populate(&index, packet_offset, subbuffer);
 250         return consumer_stream_write_index(stream, &index);
 251 }
 252
 253 /*
 254  * Actually do the metadata sync using the given metadata stream.
 255  *
 256  * Return 0 on success else a negative value. ENODATA can be returned also
 257  * indicating that there is no metadata available for that stream.
 258  */
 259 static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 260                 struct lttng_consumer_local_data *ctx)
 261 {
 262         int ret;
 263         enum sync_metadata_status status;
 264
 265         assert(metadata);
 266         assert(metadata->metadata_flag);
 267         assert(ctx);
 268
 269         /*
 270          * In UST, since we have to write the metadata from the cache packet
 271          * by packet, we might need to start this procedure multiple times
 272          * until all the metadata from the cache has been extracted.
 273          */
 274         do {
 275                 /*
 276                  * Steps :
 277                  * - Lock the metadata stream
 278                  * - Check if metadata stream node was deleted before locking.
 279                  *   - if yes, release and return success
 280                  * - Check if new metadata is ready (flush + snapshot pos)
 281                  * - If nothing : release and return.
 282                  * - Lock the metadata_rdv_lock
 283                  * - Unlock the metadata stream
 284                  * - cond_wait on metadata_rdv to wait the wakeup from the
 285                  *   metadata thread
 286                  * - Unlock the metadata_rdv_lock
 287                  */
 288                 pthread_mutex_lock(&metadata->lock);
 289
 290                 /*
 291                  * There is a possibility that we were able to acquire a reference on the
 292                  * stream from the RCU hash table but between then and now, the node might
 293                  * have been deleted just before the lock is acquired. Thus, after locking,
 294                  * we make sure the metadata node has not been deleted which means that the
 295                  * buffers are closed.
 296                  *
 297                  * In that case, there is no need to sync the metadata hence returning a
 298                  * success return code.
 299                  */
 300                 ret = cds_lfht_is_node_deleted(&metadata->node.node);
 301                 if (ret) {
 302                         ret = 0;
 303                         goto end_unlock_mutex;
 304                 }
 305
 306                 switch (ctx->type) {
 307                 case LTTNG_CONSUMER_KERNEL:
 308                         /*
 309                          * Empty the metadata cache and flush the current stream.
 310                          */
 311                         status = lttng_kconsumer_sync_metadata(metadata);
 312                         break;
 313                 case LTTNG_CONSUMER32_UST:
 314                 case LTTNG_CONSUMER64_UST:
 315                         /*
 316                          * Ask the sessiond if we have new metadata waiting and update the
 317                          * consumer metadata cache.
 318                          */
 319                         status = lttng_ustconsumer_sync_metadata(ctx, metadata);
 320                         break;
 321                 default:
 322                         abort();
 323                 }
 324
 325                 switch (status) {
 326                 case SYNC_METADATA_STATUS_NEW_DATA:
 327                         break;
 328                 case SYNC_METADATA_STATUS_NO_DATA:
 329                         ret = 0;
 330                         goto end_unlock_mutex;
 331                 case SYNC_METADATA_STATUS_ERROR:
 332                         ret = -1;
 333                         goto end_unlock_mutex;
 334                 default:
 335                         abort();
 336                 }
 337
 338                 /*
 339                  * At this point, new metadata have been flushed, so we wait on the
 340                  * rendez-vous point for the metadata thread to wake us up when it
 341                  * finishes consuming the metadata and continue execution.
 342                  */
 343
 344                 pthread_mutex_lock(&metadata->metadata_rdv_lock);
 345
 346                 /*
 347                  * Release metadata stream lock so the metadata thread can process it.
 348                  */
 349                 pthread_mutex_unlock(&metadata->lock);
 350
 351                 /*
 352                  * Wait on the rendez-vous point. Once woken up, it means the metadata was
 353                  * consumed and thus synchronization is achieved.
 354                  */
 355                 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
 356                 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
 357         } while (status == SYNC_METADATA_STATUS_NEW_DATA);
 358
 359         /* Success */
 360         return 0;
 361
 362 end_unlock_mutex:
 363         pthread_mutex_unlock(&metadata->lock);
 364         return ret;
 365 }
 366
 367 /*
 368  * Synchronize the metadata using a given session ID. A successful acquisition
 369  * of a metadata stream will trigger a request to the session daemon and a
 370  * snapshot so the metadata thread can consume it.
 371  *
 372  * This function call is a rendez-vous point between the metadata thread and
 373  * the data thread.
 374  *
 375  * Return 0 on success or else a negative value.
 376  */
 377 int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
 378                 uint64_t session_id)
 379 {
 380         int ret;
 381         struct lttng_consumer_stream *stream = NULL;
 382         struct lttng_ht_iter iter;
 383         struct lttng_ht *ht;
 384
 385         assert(ctx);
 386
 387         /* Ease our life a bit. */
 388         ht = the_consumer_data.stream_list_ht;
 389
 390         rcu_read_lock();
 391
 392         /* Search the metadata associated with the session id of the given stream. */
 393
 394         cds_lfht_for_each_entry_duplicate(ht->ht,
 395                         ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
 396                         &session_id, &iter.iter, stream, node_session_id.node) {
 397                 if (!stream->metadata_flag) {
 398                         continue;
 399                 }
 400
 401                 ret = do_sync_metadata(stream, ctx);
 402                 if (ret < 0) {
 403                         goto end;
 404                 }
 405         }
 406
 407         /*
 408          * Force return code to 0 (success) since ret might be ENODATA for instance
 409          * which is not an error but rather that we should come back.
 410          */
 411         ret = 0;
 412
 413 end:
 414         rcu_read_unlock();
 415         return ret;
 416 }
 417
 418 static int consumer_stream_sync_metadata_index(
 419                 struct lttng_consumer_stream *stream,
 420                 const struct stream_subbuffer *subbuffer,
 421                 struct lttng_consumer_local_data *ctx)
 422 {
 423         int ret;
 424
 425         /* Block until all the metadata is sent. */
 426         pthread_mutex_lock(&stream->metadata_timer_lock);
 427         assert(!stream->missed_metadata_flush);
 428         stream->waiting_on_metadata = true;
 429         pthread_mutex_unlock(&stream->metadata_timer_lock);
 430
 431         ret = consumer_stream_sync_metadata(ctx, stream->session_id);
 432
 433         pthread_mutex_lock(&stream->metadata_timer_lock);
 434         stream->waiting_on_metadata = false;
 435         if (stream->missed_metadata_flush) {
 436                 stream->missed_metadata_flush = false;
 437                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 438                 (void) stream->read_subbuffer_ops.send_live_beacon(stream);
 439         } else {
 440                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 441         }
 442         if (ret < 0) {
 443                 goto end;
 444         }
 445
 446         ret = consumer_stream_send_index(stream, subbuffer, ctx);
 447 end:
 448         return ret;
 449 }
 450
 451 /*
 452  * Check if the local version of the metadata stream matches with the version
 453  * of the metadata stream in the kernel. If it was updated, set the reset flag
 454  * on the stream.
 455  */
 456 static
 457 int metadata_stream_check_version(struct lttng_consumer_stream *stream,
 458         const struct stream_subbuffer *subbuffer)
 459 {
 460         if (stream->metadata_version == subbuffer->info.metadata.version) {
 461                 goto end;
 462         }
 463
 464         DBG("New metadata version detected");
 465         consumer_stream_metadata_set_version(stream,
 466                         subbuffer->info.metadata.version);
 467
 468         if (stream->read_subbuffer_ops.reset_metadata) {
 469                 stream->read_subbuffer_ops.reset_metadata(stream);
 470         }
 471
 472 end:
 473         return 0;
 474 }
 475
 476 static
 477 bool stream_is_rotating_to_null_chunk(
 478                 const struct lttng_consumer_stream *stream)
 479 {
 480         bool rotating_to_null_chunk = false;
 481
 482         if (stream->rotate_position == -1ULL) {
 483                 /* No rotation ongoing. */
 484                 goto end;
 485         }
 486
 487         if (stream->trace_chunk == stream->chan->trace_chunk ||
 488                         !stream->chan->trace_chunk) {
 489                 rotating_to_null_chunk = true;
 490         }
 491 end:
 492         return rotating_to_null_chunk;
 493 }
 494
 495 enum consumer_stream_open_packet_status consumer_stream_open_packet(
 496                 struct lttng_consumer_stream *stream)
 497 {
 498         int ret;
 499         enum consumer_stream_open_packet_status status;
 500         unsigned long produced_pos_before, produced_pos_after;
 501
 502         ret = lttng_consumer_sample_snapshot_positions(stream);
 503         if (ret < 0) {
 504                 ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
 505                                 ", channel name = %s, session id = %" PRIu64,
 506                                 stream->key, stream->chan->name,
 507                                 stream->chan->session_id);
 508                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 509                 goto end;
 510         }
 511
 512         ret = lttng_consumer_get_produced_snapshot(
 513                         stream, &produced_pos_before);
 514         if (ret < 0) {
 515                 ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
 516                                 ", channel name = %s, session id = %" PRIu64,
 517                                 stream->key, stream->chan->name,
 518                                 stream->chan->session_id);
 519                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 520                 goto end;
 521         }
 522
 523         ret = consumer_stream_flush_buffer(stream, 0);
 524         if (ret) {
 525                 ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
 526                                 ", channel name = %s, session id = %" PRIu64,
 527                                 stream->key, stream->chan->name,
 528                                 stream->chan->session_id);
 529                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 530                 goto end;
 531         }
 532
 533         ret = lttng_consumer_sample_snapshot_positions(stream);
 534         if (ret < 0) {
 535                 ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
 536                                 ", channel name = %s, session id = %" PRIu64,
 537                                 stream->key, stream->chan->name,
 538                                 stream->chan->session_id);
 539                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 540                 goto end;
 541         }
 542
 543         ret = lttng_consumer_get_produced_snapshot(stream, &produced_pos_after);
 544         if (ret < 0) {
 545                 ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
 546                                 ", channel name = %s, session id = %" PRIu64,
 547                                 stream->key, stream->chan->name,
 548                                 stream->chan->session_id);
 549                 status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
 550                 goto end;
 551         }
 552
 553         /*
 554          * Determine if the flush had an effect by comparing the produced
 555          * positons before and after the flush.
 556          */
 557         status = produced_pos_before != produced_pos_after ?
 558                         CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED :
 559                         CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE;
 560         if (status == CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED) {
 561                 stream->opened_packet_in_current_trace_chunk = true;
 562         }
 563
 564 end:
 565         return status;
 566 }
 567
 568 /*
 569  * An attempt to open a new packet is performed after a rotation completes to
 570  * get a begin timestamp as close as possible to the rotation point.
 571  *
 572  * However, that initial attempt at opening a packet can fail due to a full
 573  * ring-buffer. In that case, a second attempt is performed after consuming
 574  * a packet since that will have freed enough space in the ring-buffer.
 575  */
 576 static
 577 int post_consume_open_new_packet(struct lttng_consumer_stream *stream,
 578                 const struct stream_subbuffer *subbuffer,
 579                 struct lttng_consumer_local_data *ctx)
 580 {
 581         int ret = 0;
 582
 583         if (!stream->opened_packet_in_current_trace_chunk &&
 584                         stream->trace_chunk &&
 585                         !stream_is_rotating_to_null_chunk(stream)) {
 586                 const enum consumer_stream_open_packet_status status =
 587                                 consumer_stream_open_packet(stream);
 588
 589                 switch (status) {
 590                 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
 591                         DBG("Opened a packet after consuming a packet rotation: stream id = %" PRIu64
 592                                         ", channel name = %s, session id = %" PRIu64,
 593                                         stream->key, stream->chan->name,
 594                                         stream->chan->session_id);
 595                         stream->opened_packet_in_current_trace_chunk = true;
 596                         break;
 597                 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
 598                         /*
 599                          * Can't open a packet as there is no space left.
 600                          * This means that new events were produced, resulting
 601                          * in a packet being opened, which is what we want
 602                          * anyhow.
 603                          */
 604                         DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
 605                                         ", channel name = %s, session id = %" PRIu64,
 606                                         stream->key, stream->chan->name,
 607                                         stream->chan->session_id);
 608                         stream->opened_packet_in_current_trace_chunk = true;
 609                         break;
 610                 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
 611                         /* Logged by callee. */
 612                         ret = -1;
 613                         goto end;
 614                 default:
 615                         abort();
 616                 }
 617
 618                 stream->opened_packet_in_current_trace_chunk = true;
 619         }
 620
 621 end:
 622         return ret;
 623 }
 624
 625 struct lttng_consumer_stream *consumer_stream_create(
 626                 struct lttng_consumer_channel *channel,
 627                 uint64_t channel_key,
 628                 uint64_t stream_key,
 629                 const char *channel_name,
 630                 uint64_t relayd_id,
 631                 uint64_t session_id,
 632                 struct lttng_trace_chunk *trace_chunk,
 633                 int cpu,
 634                 int *alloc_ret,
 635                 enum consumer_channel_type type,
 636                 unsigned int monitor)
 637 {
 638         int ret;
 639         struct lttng_consumer_stream *stream;
 640
 641         stream = zmalloc(sizeof(*stream));
 642         if (stream == NULL) {
 643                 PERROR("malloc struct lttng_consumer_stream");
 644                 ret = -ENOMEM;
 645                 goto end;
 646         }
 647
 648         rcu_read_lock();
 649
 650         if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
 651                 ERR("Failed to acquire trace chunk reference during the creation of a stream");
 652                 ret = -1;
 653                 goto error;
 654         }
 655
 656         stream->chan = channel;
 657         stream->key = stream_key;
 658         stream->trace_chunk = trace_chunk;
 659         stream->out_fd = -1;
 660         stream->out_fd_offset = 0;
 661         stream->output_written = 0;
 662         stream->net_seq_idx = relayd_id;
 663         stream->session_id = session_id;
 664         stream->monitor = monitor;
 665         stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
 666         stream->index_file = NULL;
 667         stream->last_sequence_number = -1ULL;
 668         stream->rotate_position = -1ULL;
 669         /* Buffer is created with an open packet. */
 670         stream->opened_packet_in_current_trace_chunk = true;
 671         pthread_mutex_init(&stream->lock, NULL);
 672         pthread_mutex_init(&stream->metadata_timer_lock, NULL);
 673
 674         /* If channel is the metadata, flag this stream as metadata. */
 675         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 676                 stream->metadata_flag = 1;
 677                 /* Metadata is flat out. */
 678                 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
 679                 /* Live rendez-vous point. */
 680                 pthread_cond_init(&stream->metadata_rdv, NULL);
 681                 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
 682         } else {
 683                 /* Format stream name to <channel_name>_<cpu_number> */
 684                 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
 685                                 channel_name, cpu);
 686                 if (ret < 0) {
 687                         PERROR("snprintf stream name");
 688                         goto error;
 689                 }
 690         }
 691
 692         switch (channel->output) {
 693         case CONSUMER_CHANNEL_SPLICE:
 694                 stream->output = LTTNG_EVENT_SPLICE;
 695                 ret = utils_create_pipe(stream->splice_pipe);
 696                 if (ret < 0) {
 697                         goto error;
 698                 }
 699                 break;
 700         case CONSUMER_CHANNEL_MMAP:
 701                 stream->output = LTTNG_EVENT_MMAP;
 702                 break;
 703         default:
 704                 abort();
 705         }
 706
 707         /* Key is always the wait_fd for streams. */
 708         lttng_ht_node_init_u64(&stream->node, stream->key);
 709
 710         /* Init node per channel id key */
 711         lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
 712
 713         /* Init session id node with the stream session id */
 714         lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
 715
 716         DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
 717                         " relayd_id %" PRIu64 ", session_id %" PRIu64,
 718                         stream->name, stream->key, channel_key,
 719                         stream->net_seq_idx, stream->session_id);
 720
 721         rcu_read_unlock();
 722
 723         lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
 724                         sizeof(post_consume_cb), NULL);
 725
 726         if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 727                 stream->read_subbuffer_ops.lock =
 728                                 consumer_stream_metadata_lock_all;
 729                 stream->read_subbuffer_ops.unlock =
 730                                 consumer_stream_metadata_unlock_all;
 731                 stream->read_subbuffer_ops.assert_locked =
 732                                 consumer_stream_metadata_assert_locked_all;
 733                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 734                                 metadata_stream_check_version;
 735         } else {
 736                 const post_consume_cb post_consume_index_op = channel->is_live ?
 737                                 consumer_stream_sync_metadata_index :
 738                                 consumer_stream_send_index;
 739
 740                 ret = lttng_dynamic_array_add_element(
 741                                 &stream->read_subbuffer_ops.post_consume_cbs,
 742                                 &post_consume_index_op);
 743                 if (ret) {
 744                         PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
 745                         goto error;
 746                 }
 747
 748                 ret = lttng_dynamic_array_add_element(
 749                                 &stream->read_subbuffer_ops.post_consume_cbs,
 750                                 &(post_consume_cb) { post_consume_open_new_packet });
 751                 if (ret) {
 752                         PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
 753                         goto error;
 754                 }
 755
 756                 stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
 757                 stream->read_subbuffer_ops.unlock =
 758                                 consumer_stream_data_unlock_all;
 759                 stream->read_subbuffer_ops.assert_locked =
 760                                 consumer_stream_data_assert_locked_all;
 761                 stream->read_subbuffer_ops.pre_consume_subbuffer =
 762                                 consumer_stream_update_stats;
 763         }
 764
 765         if (channel->output == CONSUMER_CHANNEL_MMAP) {
 766                 stream->read_subbuffer_ops.consume_subbuffer =
 767                                 consumer_stream_consume_mmap;
 768         } else {
 769                 stream->read_subbuffer_ops.consume_subbuffer =
 770                                 consumer_stream_consume_splice;
 771         }
 772
 773         return stream;
 774
 775 error:
 776         rcu_read_unlock();
 777         lttng_trace_chunk_put(stream->trace_chunk);
 778         lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
 779         free(stream);
 780 end:
 781         if (alloc_ret) {
 782                 *alloc_ret = ret;
 783         }
 784         return NULL;
 785 }
 786
 787 /*
 788  * Close stream on the relayd side. This call can destroy a relayd if the
 789  * conditions are met.
 790  *
 791  * A RCU read side lock MUST be acquired if the relayd object was looked up in
 792  * a hash table before calling this.
 793  */
 794 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
 795                 struct consumer_relayd_sock_pair *relayd)
 796 {
 797         int ret;
 798
 799         assert(stream);
 800         assert(relayd);
 801
 802         if (stream->sent_to_relayd) {
 803                 uatomic_dec(&relayd->refcount);
 804                 assert(uatomic_read(&relayd->refcount) >= 0);
 805         }
 806
 807         /* Closing streams requires to lock the control socket. */
 808         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 809         ret = relayd_send_close_stream(&relayd->control_sock,
 810                         stream->relayd_stream_id,
 811                         stream->next_net_seq_num - 1);
 812         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 813         if (ret < 0) {
 814                 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 815                 lttng_consumer_cleanup_relayd(relayd);
 816         }
 817
 818         /* Both conditions are met, we destroy the relayd. */
 819         if (uatomic_read(&relayd->refcount) == 0 &&
 820                         uatomic_read(&relayd->destroy_flag)) {
 821                 consumer_destroy_relayd(relayd);
 822         }
 823         stream->net_seq_idx = (uint64_t) -1ULL;
 824         stream->sent_to_relayd = 0;
 825 }
 826
 827 /*
 828  * Close stream's file descriptors and, if needed, close stream also on the
 829  * relayd side.
 830  *
 831  * The consumer data lock MUST be acquired.
 832  * The stream lock MUST be acquired.
 833  */
 834 void consumer_stream_close(struct lttng_consumer_stream *stream)
 835 {
 836         int ret;
 837         struct consumer_relayd_sock_pair *relayd;
 838
 839         assert(stream);
 840
 841         switch (the_consumer_data.type) {
 842         case LTTNG_CONSUMER_KERNEL:
 843                 if (stream->mmap_base != NULL) {
 844                         ret = munmap(stream->mmap_base, stream->mmap_len);
 845                         if (ret != 0) {
 846                                 PERROR("munmap");
 847                         }
 848                 }
 849
 850                 if (stream->wait_fd >= 0) {
 851                         ret = close(stream->wait_fd);
 852                         if (ret) {
 853                                 PERROR("close");
 854                         }
 855                         stream->wait_fd = -1;
 856                 }
 857                 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
 858                         utils_close_pipe(stream->splice_pipe);
 859                 }
 860                 break;
 861         case LTTNG_CONSUMER32_UST:
 862         case LTTNG_CONSUMER64_UST:
 863         {
 864                 /*
 865                  * Special case for the metadata since the wait fd is an internal pipe
 866                  * polled in the metadata thread.
 867                  */
 868                 if (stream->metadata_flag && stream->chan->monitor) {
 869                         int rpipe = stream->ust_metadata_poll_pipe[0];
 870
 871                         /*
 872                          * This will stop the channel timer if one and close the write side
 873                          * of the metadata poll pipe.
 874                          */
 875                         lttng_ustconsumer_close_metadata(stream->chan);
 876                         if (rpipe >= 0) {
 877                                 ret = close(rpipe);
 878                                 if (ret < 0) {
 879                                         PERROR("closing metadata pipe read side");
 880                                 }
 881                                 stream->ust_metadata_poll_pipe[0] = -1;
 882                         }
 883                 }
 884                 break;
 885         }
 886         default:
 887                 ERR("Unknown consumer_data type");
 888                 assert(0);
 889         }
 890
 891         /* Close output fd. Could be a socket or local file at this point. */
 892         if (stream->out_fd >= 0) {
 893                 ret = close(stream->out_fd);
 894                 if (ret) {
 895                         PERROR("close");
 896                 }
 897                 stream->out_fd = -1;
 898         }
 899
 900         if (stream->index_file) {
 901                 lttng_index_file_put(stream->index_file);
 902                 stream->index_file = NULL;
 903         }
 904
 905         lttng_trace_chunk_put(stream->trace_chunk);
 906         stream->trace_chunk = NULL;
 907
 908         /* Check and cleanup relayd if needed. */
 909         rcu_read_lock();
 910         relayd = consumer_find_relayd(stream->net_seq_idx);
 911         if (relayd != NULL) {
 912                 consumer_stream_relayd_close(stream, relayd);
 913         }
 914         rcu_read_unlock();
 915 }
 916
 917 /*
 918  * Delete the stream from all possible hash tables.
 919  *
 920  * The consumer data lock MUST be acquired.
 921  * The stream lock MUST be acquired.
 922  */
 923 void consumer_stream_delete(struct lttng_consumer_stream *stream,
 924                 struct lttng_ht *ht)
 925 {
 926         int ret;
 927         struct lttng_ht_iter iter;
 928
 929         assert(stream);
 930         /* Should NEVER be called not in monitor mode. */
 931         assert(stream->chan->monitor);
 932
 933         rcu_read_lock();
 934
 935         if (ht) {
 936                 iter.iter.node = &stream->node.node;
 937                 ret = lttng_ht_del(ht, &iter);
 938                 assert(!ret);
 939         }
 940
 941         /* Delete from stream per channel ID hash table. */
 942         iter.iter.node = &stream->node_channel_id.node;
 943         /*
 944          * The returned value is of no importance. Even if the node is NOT in the
 945          * hash table, we continue since we may have been called by a code path
 946          * that did not add the stream to a (all) hash table. Same goes for the
 947          * next call ht del call.
 948          */
 949         (void) lttng_ht_del(the_consumer_data.stream_per_chan_id_ht, &iter);
 950
 951         /* Delete from the global stream list. */
 952         iter.iter.node = &stream->node_session_id.node;
 953         /* See the previous ht del on why we ignore the returned value. */
 954         (void) lttng_ht_del(the_consumer_data.stream_list_ht, &iter);
 955
 956         rcu_read_unlock();
 957
 958         if (!stream->metadata_flag) {
 959                 /* Decrement the stream count of the global consumer data. */
 960                 assert(the_consumer_data.stream_count > 0);
 961                 the_consumer_data.stream_count--;
 962         }
 963 }
 964
 965 /*
 966  * Free the given stream within a RCU call.
 967  */
 968 void consumer_stream_free(struct lttng_consumer_stream *stream)
 969 {
 970         assert(stream);
 971
 972         metadata_bucket_destroy(stream->metadata_bucket);
 973         call_rcu(&stream->node.head, free_stream_rcu);
 974 }
 975
 976 /*
 977  * Destroy the stream's buffers of the tracer.
 978  */
 979 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 980 {
 981         assert(stream);
 982
 983         switch (the_consumer_data.type) {
 984         case LTTNG_CONSUMER_KERNEL:
 985                 break;
 986         case LTTNG_CONSUMER32_UST:
 987         case LTTNG_CONSUMER64_UST:
 988                 lttng_ustconsumer_del_stream(stream);
 989                 break;
 990         default:
 991                 ERR("Unknown consumer_data type");
 992                 assert(0);
 993         }
 994 }
 995
 996 /*
 997  * Destroy and close a already created stream.
 998  */
 999 static void destroy_close_stream(struct lttng_consumer_stream *stream)
1000 {
1001         assert(stream);
1002
1003         DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
1004
1005         /* Destroy tracer buffers of the stream. */
1006         consumer_stream_destroy_buffers(stream);
1007         /* Close down everything including the relayd if one. */
1008         consumer_stream_close(stream);
1009 }
1010
1011 /*
1012  * Decrement the stream's channel refcount and if down to 0, return the channel
1013  * pointer so it can be destroyed by the caller or NULL if not.
1014  */
1015 static struct lttng_consumer_channel *unref_channel(
1016                 struct lttng_consumer_stream *stream)
1017 {
1018         struct lttng_consumer_channel *free_chan = NULL;
1019
1020         assert(stream);
1021         assert(stream->chan);
1022
1023         /* Update refcount of channel and see if we need to destroy it. */
1024         if (!uatomic_sub_return(&stream->chan->refcount, 1)
1025                         && !uatomic_read(&stream->chan->nb_init_stream_left)) {
1026                 free_chan = stream->chan;
1027         }
1028
1029         return free_chan;
1030 }
1031
1032 /*
1033  * Destroy a stream completely. This will delete, close and free the stream.
1034  * Once return, the stream is NO longer usable. Its channel may get destroyed
1035  * if conditions are met for a monitored stream.
1036  *
1037  * This MUST be called WITHOUT the consumer data and stream lock acquired if
1038  * the stream is in _monitor_ mode else it does not matter.
1039  */
1040 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
1041                 struct lttng_ht *ht)
1042 {
1043         assert(stream);
1044
1045         /* Stream is in monitor mode. */
1046         if (stream->monitor) {
1047                 struct lttng_consumer_channel *free_chan = NULL;
1048
1049                 /*
1050                  * This means that the stream was successfully removed from the streams
1051                  * list of the channel and sent to the right thread managing this
1052                  * stream thus being globally visible.
1053                  */
1054                 if (stream->globally_visible) {
1055                         pthread_mutex_lock(&the_consumer_data.lock);
1056                         pthread_mutex_lock(&stream->chan->lock);
1057                         pthread_mutex_lock(&stream->lock);
1058                         /* Remove every reference of the stream in the consumer. */
1059                         consumer_stream_delete(stream, ht);
1060
1061                         destroy_close_stream(stream);
1062
1063                         /* Update channel's refcount of the stream. */
1064                         free_chan = unref_channel(stream);
1065
1066                         /* Indicates that the consumer data state MUST be updated after this. */
1067                         the_consumer_data.need_update = 1;
1068
1069                         pthread_mutex_unlock(&stream->lock);
1070                         pthread_mutex_unlock(&stream->chan->lock);
1071                         pthread_mutex_unlock(&the_consumer_data.lock);
1072                 } else {
1073                         /*
1074                          * If the stream is not visible globally, this needs to be done
1075                          * outside of the consumer data lock section.
1076                          */
1077                         free_chan = unref_channel(stream);
1078                 }
1079
1080                 if (free_chan) {
1081                         consumer_del_channel(free_chan);
1082                 }
1083         } else {
1084                 destroy_close_stream(stream);
1085         }
1086
1087         /* Free stream within a RCU call. */
1088         lttng_trace_chunk_put(stream->trace_chunk);
1089         stream->trace_chunk = NULL;
1090         lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
1091         consumer_stream_free(stream);
1092 }
1093
1094 /*
1095  * Write index of a specific stream either on the relayd or local disk.
1096  *
1097  * Return 0 on success or else a negative value.
1098  */
1099 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
1100                 struct ctf_packet_index *element)
1101 {
1102         int ret;
1103
1104         assert(stream);
1105         assert(element);
1106
1107         rcu_read_lock();
1108         if (stream->net_seq_idx != (uint64_t) -1ULL) {
1109                 struct consumer_relayd_sock_pair *relayd;
1110                 relayd = consumer_find_relayd(stream->net_seq_idx);
1111                 if (relayd) {
1112                         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1113                         ret = relayd_send_index(&relayd->control_sock, element,
1114                                 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1115                         if (ret < 0) {
1116                                 /*
1117                                  * Communication error with lttng-relayd,
1118                                  * perform cleanup now
1119                                  */
1120                                 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
1121                                 lttng_consumer_cleanup_relayd(relayd);
1122                                 ret = -1;
1123                         }
1124                         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1125                 } else {
1126                         ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
1127                                         stream->key, stream->net_seq_idx);
1128                         ret = -1;
1129                 }
1130         } else {
1131                 if (lttng_index_file_write(stream->index_file, element)) {
1132                         ret = -1;
1133                 } else {
1134                         ret = 0;
1135                 }
1136         }
1137         if (ret < 0) {
1138                 goto error;
1139         }
1140
1141 error:
1142         rcu_read_unlock();
1143         return ret;
1144 }
1145
1146 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
1147                 bool create_index)
1148 {
1149         int ret;
1150         enum lttng_trace_chunk_status chunk_status;
1151         const int flags = O_WRONLY | O_CREAT | O_TRUNC;
1152         const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
1153         char stream_path[LTTNG_PATH_MAX];
1154
1155         ASSERT_LOCKED(stream->lock);
1156         assert(stream->trace_chunk);
1157
1158         ret = utils_stream_file_path(stream->chan->pathname, stream->name,
1159                         stream->chan->tracefile_size,
1160                         stream->tracefile_count_current, NULL,
1161                         stream_path, sizeof(stream_path));
1162         if (ret < 0) {
1163                 goto end;
1164         }
1165
1166         if (stream->out_fd >= 0) {
1167                 ret = close(stream->out_fd);
1168                 if (ret < 0) {
1169                         PERROR("Failed to close stream file \"%s\"",
1170                                         stream->name);
1171                         goto end;
1172                 }
1173                 stream->out_fd = -1;
1174         }
1175
1176         DBG("Opening stream output file \"%s\"", stream_path);
1177         chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
1178                         flags, mode, &stream->out_fd, false);
1179         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1180                 ERR("Failed to open stream file \"%s\"", stream->name);
1181                 ret = -1;
1182                 goto end;
1183         }
1184
1185         if (!stream->metadata_flag && (create_index || stream->index_file)) {
1186                 if (stream->index_file) {
1187                         lttng_index_file_put(stream->index_file);
1188                 }
1189                 chunk_status = lttng_index_file_create_from_trace_chunk(
1190                                 stream->trace_chunk,
1191                                 stream->chan->pathname,
1192                                 stream->name,
1193                                 stream->chan->tracefile_size,
1194                                 stream->tracefile_count_current,
1195                                 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
1196                                 false, &stream->index_file);
1197                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1198                         ret = -1;
1199                         goto end;
1200                 }
1201         }
1202
1203         /* Reset current size because we just perform a rotation. */
1204         stream->tracefile_size_current = 0;
1205         stream->out_fd_offset = 0;
1206 end:
1207         return ret;
1208 }
1209
1210 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
1211 {
1212         int ret;
1213
1214         stream->tracefile_count_current++;
1215         if (stream->chan->tracefile_count > 0) {
1216                 stream->tracefile_count_current %=
1217                                 stream->chan->tracefile_count;
1218         }
1219
1220         DBG("Rotating output files of stream \"%s\"", stream->name);
1221         ret = consumer_stream_create_output_files(stream, true);
1222         if (ret) {
1223                 goto end;
1224         }
1225
1226 end:
1227         return ret;
1228 }
1229
1230 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
1231 {
1232         /*
1233          * This function does not take a const stream since
1234          * cds_lfht_is_node_deleted was not const before liburcu 0.12.
1235          */
1236         assert(stream);
1237         return cds_lfht_is_node_deleted(&stream->node.node);
1238 }
1239
1240 static ssize_t metadata_bucket_flush(
1241                 const struct stream_subbuffer *buffer, void *data)
1242 {
1243         ssize_t ret;
1244         struct lttng_consumer_stream *stream = data;
1245
1246         ret = consumer_stream_consume_mmap(NULL, stream, buffer);
1247         if (ret < 0) {
1248                 goto end;
1249         }
1250 end:
1251         return ret;
1252 }
1253
1254 static ssize_t metadata_bucket_consume(
1255                 struct lttng_consumer_local_data *unused,
1256                 struct lttng_consumer_stream *stream,
1257                 const struct stream_subbuffer *subbuffer)
1258 {
1259         ssize_t ret;
1260         enum metadata_bucket_status status;
1261
1262         status = metadata_bucket_fill(stream->metadata_bucket, subbuffer);
1263         switch (status) {
1264         case METADATA_BUCKET_STATUS_OK:
1265                 /* Return consumed size. */
1266                 ret = subbuffer->buffer.buffer.size;
1267                 break;
1268         default:
1269                 ret = -1;
1270         }
1271
1272         return ret;
1273 }
1274
1275 int consumer_stream_enable_metadata_bucketization(
1276                 struct lttng_consumer_stream *stream)
1277 {
1278         int ret = 0;
1279
1280         assert(stream->metadata_flag);
1281         assert(!stream->metadata_bucket);
1282         assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
1283
1284         stream->metadata_bucket = metadata_bucket_create(
1285                         metadata_bucket_flush, stream);
1286         if (!stream->metadata_bucket) {
1287                 ret = -1;
1288                 goto end;
1289         }
1290
1291         stream->read_subbuffer_ops.consume_subbuffer = metadata_bucket_consume;
1292 end:
1293         return ret;
1294 }
1295
1296 void consumer_stream_metadata_set_version(
1297                 struct lttng_consumer_stream *stream, uint64_t new_version)
1298 {
1299         assert(new_version > stream->metadata_version);
1300         stream->metadata_version = new_version;
1301         stream->reset_metadata_flag = 1;
1302
1303         if (stream->metadata_bucket) {
1304                 metadata_bucket_reset(stream->metadata_bucket);
1305         }
1306 }
1307
1308 int consumer_stream_flush_buffer(struct lttng_consumer_stream *stream,
1309                 bool producer_active)
1310 {
1311         int ret = 0;
1312
1313         switch (the_consumer_data.type) {
1314         case LTTNG_CONSUMER_KERNEL:
1315                 if (producer_active) {
1316                         ret = kernctl_buffer_flush(stream->wait_fd);
1317                         if (ret < 0) {
1318                                 ERR("Failed to flush kernel stream");
1319                                 goto end;
1320                         }
1321                 } else {
1322                         ret = kernctl_buffer_flush_empty(stream->wait_fd);
1323                         if (ret < 0) {
1324                                 /*
1325                                  * Doing a buffer flush which does not take into
1326                                  * account empty packets. This is not perfect,
1327                                  * but required as a fall-back when
1328                                  * "flush_empty" is not implemented by
1329                                  * lttng-modules.
1330                                  */
1331                                 ret = kernctl_buffer_flush(stream->wait_fd);
1332                                 if (ret < 0) {
1333                                         ERR("Failed to flush kernel stream");
1334                                         goto end;
1335                                 }
1336                         }
1337                 }
1338                 break;
1339         case LTTNG_CONSUMER32_UST:
1340         case LTTNG_CONSUMER64_UST:
1341                 ret = lttng_ustconsumer_flush_buffer(stream, (int) producer_active);
1342                 break;
1343         default:
1344                 ERR("Unknown consumer_data type");
1345                 abort();
1346         }
1347
1348 end:
1349         return ret;
1350 }