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/index/index.h>
  18 #include <common/kernel-consumer/kernel-consumer.h>
  19 #include <common/relayd/relayd.h>
  20 #include <common/ust-consumer/ust-consumer.h>
  21 #include <common/utils.h>
  22
  23 #include "consumer-stream.h"
  24
  25 /*
  26  * RCU call to free stream. MUST only be used with call_rcu().
  27  */
  28 static void free_stream_rcu(struct rcu_head *head)
  29 {
  30         struct lttng_ht_node_u64 *node =
  31                 caa_container_of(head, struct lttng_ht_node_u64, head);
  32         struct lttng_consumer_stream *stream =
  33                 caa_container_of(node, struct lttng_consumer_stream, node);
  34
  35         pthread_mutex_destroy(&stream->lock);
  36         free(stream);
  37 }
  38
  39 /*
  40  * Close stream on the relayd side. This call can destroy a relayd if the
  41  * conditions are met.
  42  *
  43  * A RCU read side lock MUST be acquired if the relayd object was looked up in
  44  * a hash table before calling this.
  45  */
  46 void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
  47                 struct consumer_relayd_sock_pair *relayd)
  48 {
  49         int ret;
  50
  51         assert(stream);
  52         assert(relayd);
  53
  54         if (stream->sent_to_relayd) {
  55                 uatomic_dec(&relayd->refcount);
  56                 assert(uatomic_read(&relayd->refcount) >= 0);
  57         }
  58
  59         /* Closing streams requires to lock the control socket. */
  60         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
  61         ret = relayd_send_close_stream(&relayd->control_sock,
  62                         stream->relayd_stream_id,
  63                         stream->next_net_seq_num - 1);
  64         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
  65         if (ret < 0) {
  66                 ERR("Relayd send close stream failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
  67                 lttng_consumer_cleanup_relayd(relayd);
  68         }
  69
  70         /* Both conditions are met, we destroy the relayd. */
  71         if (uatomic_read(&relayd->refcount) == 0 &&
  72                         uatomic_read(&relayd->destroy_flag)) {
  73                 consumer_destroy_relayd(relayd);
  74         }
  75         stream->net_seq_idx = (uint64_t) -1ULL;
  76         stream->sent_to_relayd = 0;
  77 }
  78
  79 /*
  80  * Close stream's file descriptors and, if needed, close stream also on the
  81  * relayd side.
  82  *
  83  * The consumer data lock MUST be acquired.
  84  * The stream lock MUST be acquired.
  85  */
  86 void consumer_stream_close(struct lttng_consumer_stream *stream)
  87 {
  88         int ret;
  89         struct consumer_relayd_sock_pair *relayd;
  90
  91         assert(stream);
  92
  93         switch (consumer_data.type) {
  94         case LTTNG_CONSUMER_KERNEL:
  95                 if (stream->mmap_base != NULL) {
  96                         ret = munmap(stream->mmap_base, stream->mmap_len);
  97                         if (ret != 0) {
  98                                 PERROR("munmap");
  99                         }
 100                 }
 101
 102                 if (stream->wait_fd >= 0) {
 103                         ret = close(stream->wait_fd);
 104                         if (ret) {
 105                                 PERROR("close");
 106                         }
 107                         stream->wait_fd = -1;
 108                 }
 109                 if (stream->chan->output == CONSUMER_CHANNEL_SPLICE) {
 110                         utils_close_pipe(stream->splice_pipe);
 111                 }
 112                 break;
 113         case LTTNG_CONSUMER32_UST:
 114         case LTTNG_CONSUMER64_UST:
 115         {
 116                 /*
 117                  * Special case for the metadata since the wait fd is an internal pipe
 118                  * polled in the metadata thread.
 119                  */
 120                 if (stream->metadata_flag && stream->chan->monitor) {
 121                         int rpipe = stream->ust_metadata_poll_pipe[0];
 122
 123                         /*
 124                          * This will stop the channel timer if one and close the write side
 125                          * of the metadata poll pipe.
 126                          */
 127                         lttng_ustconsumer_close_metadata(stream->chan);
 128                         if (rpipe >= 0) {
 129                                 ret = close(rpipe);
 130                                 if (ret < 0) {
 131                                         PERROR("closing metadata pipe read side");
 132                                 }
 133                                 stream->ust_metadata_poll_pipe[0] = -1;
 134                         }
 135                 }
 136                 break;
 137         }
 138         default:
 139                 ERR("Unknown consumer_data type");
 140                 assert(0);
 141         }
 142
 143         /* Close output fd. Could be a socket or local file at this point. */
 144         if (stream->out_fd >= 0) {
 145                 ret = close(stream->out_fd);
 146                 if (ret) {
 147                         PERROR("close");
 148                 }
 149                 stream->out_fd = -1;
 150         }
 151
 152         if (stream->index_file) {
 153                 lttng_index_file_put(stream->index_file);
 154                 stream->index_file = NULL;
 155         }
 156
 157         lttng_trace_chunk_put(stream->trace_chunk);
 158         stream->trace_chunk = NULL;
 159
 160         /* Check and cleanup relayd if needed. */
 161         rcu_read_lock();
 162         relayd = consumer_find_relayd(stream->net_seq_idx);
 163         if (relayd != NULL) {
 164                 consumer_stream_relayd_close(stream, relayd);
 165         }
 166         rcu_read_unlock();
 167 }
 168
 169 /*
 170  * Delete the stream from all possible hash tables.
 171  *
 172  * The consumer data lock MUST be acquired.
 173  * The stream lock MUST be acquired.
 174  */
 175 void consumer_stream_delete(struct lttng_consumer_stream *stream,
 176                 struct lttng_ht *ht)
 177 {
 178         int ret;
 179         struct lttng_ht_iter iter;
 180
 181         assert(stream);
 182         /* Should NEVER be called not in monitor mode. */
 183         assert(stream->chan->monitor);
 184
 185         rcu_read_lock();
 186
 187         if (ht) {
 188                 iter.iter.node = &stream->node.node;
 189                 ret = lttng_ht_del(ht, &iter);
 190                 assert(!ret);
 191         }
 192
 193         /* Delete from stream per channel ID hash table. */
 194         iter.iter.node = &stream->node_channel_id.node;
 195         /*
 196          * The returned value is of no importance. Even if the node is NOT in the
 197          * hash table, we continue since we may have been called by a code path
 198          * that did not add the stream to a (all) hash table. Same goes for the
 199          * next call ht del call.
 200          */
 201         (void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
 202
 203         /* Delete from the global stream list. */
 204         iter.iter.node = &stream->node_session_id.node;
 205         /* See the previous ht del on why we ignore the returned value. */
 206         (void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
 207
 208         rcu_read_unlock();
 209
 210         if (!stream->metadata_flag) {
 211                 /* Decrement the stream count of the global consumer data. */
 212                 assert(consumer_data.stream_count > 0);
 213                 consumer_data.stream_count--;
 214         }
 215 }
 216
 217 /*
 218  * Free the given stream within a RCU call.
 219  */
 220 void consumer_stream_free(struct lttng_consumer_stream *stream)
 221 {
 222         assert(stream);
 223
 224         call_rcu(&stream->node.head, free_stream_rcu);
 225 }
 226
 227 /*
 228  * Destroy the stream's buffers of the tracer.
 229  */
 230 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 231 {
 232         assert(stream);
 233
 234         switch (consumer_data.type) {
 235         case LTTNG_CONSUMER_KERNEL:
 236                 break;
 237         case LTTNG_CONSUMER32_UST:
 238         case LTTNG_CONSUMER64_UST:
 239                 lttng_ustconsumer_del_stream(stream);
 240                 break;
 241         default:
 242                 ERR("Unknown consumer_data type");
 243                 assert(0);
 244         }
 245 }
 246
 247 /*
 248  * Destroy and close a already created stream.
 249  */
 250 static void destroy_close_stream(struct lttng_consumer_stream *stream)
 251 {
 252         assert(stream);
 253
 254         DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
 255
 256         /* Destroy tracer buffers of the stream. */
 257         consumer_stream_destroy_buffers(stream);
 258         /* Close down everything including the relayd if one. */
 259         consumer_stream_close(stream);
 260 }
 261
 262 /*
 263  * Decrement the stream's channel refcount and if down to 0, return the channel
 264  * pointer so it can be destroyed by the caller or NULL if not.
 265  */
 266 static struct lttng_consumer_channel *unref_channel(
 267                 struct lttng_consumer_stream *stream)
 268 {
 269         struct lttng_consumer_channel *free_chan = NULL;
 270
 271         assert(stream);
 272         assert(stream->chan);
 273
 274         /* Update refcount of channel and see if we need to destroy it. */
 275         if (!uatomic_sub_return(&stream->chan->refcount, 1)
 276                         && !uatomic_read(&stream->chan->nb_init_stream_left)) {
 277                 free_chan = stream->chan;
 278         }
 279
 280         return free_chan;
 281 }
 282
 283 /*
 284  * Destroy a stream completely. This will delete, close and free the stream.
 285  * Once return, the stream is NO longer usable. Its channel may get destroyed
 286  * if conditions are met for a monitored stream.
 287  *
 288  * This MUST be called WITHOUT the consumer data and stream lock acquired if
 289  * the stream is in _monitor_ mode else it does not matter.
 290  */
 291 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 292                 struct lttng_ht *ht)
 293 {
 294         assert(stream);
 295
 296         /* Stream is in monitor mode. */
 297         if (stream->monitor) {
 298                 struct lttng_consumer_channel *free_chan = NULL;
 299
 300                 /*
 301                  * This means that the stream was successfully removed from the streams
 302                  * list of the channel and sent to the right thread managing this
 303                  * stream thus being globally visible.
 304                  */
 305                 if (stream->globally_visible) {
 306                         pthread_mutex_lock(&consumer_data.lock);
 307                         pthread_mutex_lock(&stream->chan->lock);
 308                         pthread_mutex_lock(&stream->lock);
 309                         /* Remove every reference of the stream in the consumer. */
 310                         consumer_stream_delete(stream, ht);
 311
 312                         destroy_close_stream(stream);
 313
 314                         /* Update channel's refcount of the stream. */
 315                         free_chan = unref_channel(stream);
 316
 317                         /* Indicates that the consumer data state MUST be updated after this. */
 318                         consumer_data.need_update = 1;
 319
 320                         pthread_mutex_unlock(&stream->lock);
 321                         pthread_mutex_unlock(&stream->chan->lock);
 322                         pthread_mutex_unlock(&consumer_data.lock);
 323                 } else {
 324                         /*
 325                          * If the stream is not visible globally, this needs to be done
 326                          * outside of the consumer data lock section.
 327                          */
 328                         free_chan = unref_channel(stream);
 329                 }
 330
 331                 if (free_chan) {
 332                         consumer_del_channel(free_chan);
 333                 }
 334         } else {
 335                 destroy_close_stream(stream);
 336         }
 337
 338         /* Free stream within a RCU call. */
 339         lttng_trace_chunk_put(stream->trace_chunk);
 340         stream->trace_chunk = NULL;
 341         consumer_stream_free(stream);
 342 }
 343
 344 /*
 345  * Write index of a specific stream either on the relayd or local disk.
 346  *
 347  * Return 0 on success or else a negative value.
 348  */
 349 int consumer_stream_write_index(struct lttng_consumer_stream *stream,
 350                 struct ctf_packet_index *element)
 351 {
 352         int ret;
 353
 354         assert(stream);
 355         assert(element);
 356
 357         rcu_read_lock();
 358         if (stream->net_seq_idx != (uint64_t) -1ULL) {
 359                 struct consumer_relayd_sock_pair *relayd;
 360                 relayd = consumer_find_relayd(stream->net_seq_idx);
 361                 if (relayd) {
 362                         pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 363                         ret = relayd_send_index(&relayd->control_sock, element,
 364                                 stream->relayd_stream_id, stream->next_net_seq_num - 1);
 365                         if (ret < 0) {
 366                                 /*
 367                                  * Communication error with lttng-relayd,
 368                                  * perform cleanup now
 369                                  */
 370                                 ERR("Relayd send index failed. Cleaning up relayd %" PRIu64 ".", relayd->net_seq_idx);
 371                                 lttng_consumer_cleanup_relayd(relayd);
 372                                 ret = -1;
 373                         }
 374                         pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 375                 } else {
 376                         ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't write index.",
 377                                         stream->key, stream->net_seq_idx);
 378                         ret = -1;
 379                 }
 380         } else {
 381                 if (lttng_index_file_write(stream->index_file, element)) {
 382                         ret = -1;
 383                 } else {
 384                         ret = 0;
 385                 }
 386         }
 387         if (ret < 0) {
 388                 goto error;
 389         }
 390
 391 error:
 392         rcu_read_unlock();
 393         return ret;
 394 }
 395
 396 /*
 397  * Actually do the metadata sync using the given metadata stream.
 398  *
 399  * Return 0 on success else a negative value. ENODATA can be returned also
 400  * indicating that there is no metadata available for that stream.
 401  */
 402 static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 403                 struct lttng_consumer_local_data *ctx)
 404 {
 405         int ret;
 406
 407         assert(metadata);
 408         assert(metadata->metadata_flag);
 409         assert(ctx);
 410
 411         /*
 412          * In UST, since we have to write the metadata from the cache packet
 413          * by packet, we might need to start this procedure multiple times
 414          * until all the metadata from the cache has been extracted.
 415          */
 416         do {
 417                 /*
 418                  * Steps :
 419                  * - Lock the metadata stream
 420                  * - Check if metadata stream node was deleted before locking.
 421                  *   - if yes, release and return success
 422                  * - Check if new metadata is ready (flush + snapshot pos)
 423                  * - If nothing : release and return.
 424                  * - Lock the metadata_rdv_lock
 425                  * - Unlock the metadata stream
 426                  * - cond_wait on metadata_rdv to wait the wakeup from the
 427                  *   metadata thread
 428                  * - Unlock the metadata_rdv_lock
 429                  */
 430                 pthread_mutex_lock(&metadata->lock);
 431
 432                 /*
 433                  * There is a possibility that we were able to acquire a reference on the
 434                  * stream from the RCU hash table but between then and now, the node might
 435                  * have been deleted just before the lock is acquired. Thus, after locking,
 436                  * we make sure the metadata node has not been deleted which means that the
 437                  * buffers are closed.
 438                  *
 439                  * In that case, there is no need to sync the metadata hence returning a
 440                  * success return code.
 441                  */
 442                 ret = cds_lfht_is_node_deleted(&metadata->node.node);
 443                 if (ret) {
 444                         ret = 0;
 445                         goto end_unlock_mutex;
 446                 }
 447
 448                 switch (ctx->type) {
 449                 case LTTNG_CONSUMER_KERNEL:
 450                         /*
 451                          * Empty the metadata cache and flush the current stream.
 452                          */
 453                         ret = lttng_kconsumer_sync_metadata(metadata);
 454                         break;
 455                 case LTTNG_CONSUMER32_UST:
 456                 case LTTNG_CONSUMER64_UST:
 457                         /*
 458                          * Ask the sessiond if we have new metadata waiting and update the
 459                          * consumer metadata cache.
 460                          */
 461                         ret = lttng_ustconsumer_sync_metadata(ctx, metadata);
 462                         break;
 463                 default:
 464                         assert(0);
 465                         ret = -1;
 466                         break;
 467                 }
 468                 /*
 469                  * Error or no new metadata, we exit here.
 470                  */
 471                 if (ret <= 0 || ret == ENODATA) {
 472                         goto end_unlock_mutex;
 473                 }
 474
 475                 /*
 476                  * At this point, new metadata have been flushed, so we wait on the
 477                  * rendez-vous point for the metadata thread to wake us up when it
 478                  * finishes consuming the metadata and continue execution.
 479                  */
 480
 481                 pthread_mutex_lock(&metadata->metadata_rdv_lock);
 482
 483                 /*
 484                  * Release metadata stream lock so the metadata thread can process it.
 485                  */
 486                 pthread_mutex_unlock(&metadata->lock);
 487
 488                 /*
 489                  * Wait on the rendez-vous point. Once woken up, it means the metadata was
 490                  * consumed and thus synchronization is achieved.
 491                  */
 492                 pthread_cond_wait(&metadata->metadata_rdv, &metadata->metadata_rdv_lock);
 493                 pthread_mutex_unlock(&metadata->metadata_rdv_lock);
 494         } while (ret == EAGAIN);
 495
 496         /* Success */
 497         return 0;
 498
 499 end_unlock_mutex:
 500         pthread_mutex_unlock(&metadata->lock);
 501         return ret;
 502 }
 503
 504 /*
 505  * Synchronize the metadata using a given session ID. A successful acquisition
 506  * of a metadata stream will trigger a request to the session daemon and a
 507  * snapshot so the metadata thread can consume it.
 508  *
 509  * This function call is a rendez-vous point between the metadata thread and
 510  * the data thread.
 511  *
 512  * Return 0 on success or else a negative value.
 513  */
 514 int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
 515                 uint64_t session_id)
 516 {
 517         int ret;
 518         struct lttng_consumer_stream *stream = NULL;
 519         struct lttng_ht_iter iter;
 520         struct lttng_ht *ht;
 521
 522         assert(ctx);
 523
 524         /* Ease our life a bit. */
 525         ht = consumer_data.stream_list_ht;
 526
 527         rcu_read_lock();
 528
 529         /* Search the metadata associated with the session id of the given stream. */
 530
 531         cds_lfht_for_each_entry_duplicate(ht->ht,
 532                         ht->hash_fct(&session_id, lttng_ht_seed), ht->match_fct,
 533                         &session_id, &iter.iter, stream, node_session_id.node) {
 534                 if (!stream->metadata_flag) {
 535                         continue;
 536                 }
 537
 538                 ret = do_sync_metadata(stream, ctx);
 539                 if (ret < 0) {
 540                         goto end;
 541                 }
 542         }
 543
 544         /*
 545          * Force return code to 0 (success) since ret might be ENODATA for instance
 546          * which is not an error but rather that we should come back.
 547          */
 548         ret = 0;
 549
 550 end:
 551         rcu_read_unlock();
 552         return ret;
 553 }
 554
 555 int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
 556                 bool create_index)
 557 {
 558         int ret;
 559         enum lttng_trace_chunk_status chunk_status;
 560         const int flags = O_WRONLY | O_CREAT | O_TRUNC;
 561         const mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;
 562         char stream_path[LTTNG_PATH_MAX];
 563
 564         ASSERT_LOCKED(stream->lock);
 565         assert(stream->trace_chunk);
 566
 567         ret = utils_stream_file_path(stream->chan->pathname, stream->name,
 568                         stream->chan->tracefile_size,
 569                         stream->tracefile_count_current, NULL,
 570                         stream_path, sizeof(stream_path));
 571         if (ret < 0) {
 572                 goto end;
 573         }
 574
 575         if (stream->out_fd >= 0) {
 576                 ret = close(stream->out_fd);
 577                 if (ret < 0) {
 578                         PERROR("Failed to close stream file \"%s\"",
 579                                         stream->name);
 580                         goto end;
 581                 }
 582                 stream->out_fd = -1;
 583         }
 584
 585         DBG("Opening stream output file \"%s\"", stream_path);
 586         chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
 587                         flags, mode, &stream->out_fd, false);
 588         if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 589                 ERR("Failed to open stream file \"%s\"", stream->name);
 590                 ret = -1;
 591                 goto end;
 592         }
 593
 594         if (!stream->metadata_flag && (create_index || stream->index_file)) {
 595                 if (stream->index_file) {
 596                         lttng_index_file_put(stream->index_file);
 597                 }
 598                 chunk_status = lttng_index_file_create_from_trace_chunk(
 599                                 stream->trace_chunk,
 600                                 stream->chan->pathname,
 601                                 stream->name,
 602                                 stream->chan->tracefile_size,
 603                                 stream->tracefile_count_current,
 604                                 CTF_INDEX_MAJOR, CTF_INDEX_MINOR,
 605                                 false, &stream->index_file);
 606                 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 607                         ret = -1;
 608                         goto end;
 609                 }
 610         }
 611
 612         /* Reset current size because we just perform a rotation. */
 613         stream->tracefile_size_current = 0;
 614         stream->out_fd_offset = 0;
 615 end:
 616         return ret;
 617 }
 618
 619 int consumer_stream_rotate_output_files(struct lttng_consumer_stream *stream)
 620 {
 621         int ret;
 622
 623         stream->tracefile_count_current++;
 624         if (stream->chan->tracefile_count > 0) {
 625                 stream->tracefile_count_current %=
 626                                 stream->chan->tracefile_count;
 627         }
 628
 629         DBG("Rotating output files of stream \"%s\"", stream->name);
 630         ret = consumer_stream_create_output_files(stream, true);
 631         if (ret) {
 632                 goto end;
 633         }
 634
 635 end:
 636         return ret;
 637 }
 638
 639 bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
 640 {
 641         /*
 642          * This function does not take a const stream since
 643          * cds_lfht_is_node_deleted was not const before liburcu 0.12.
 644          */
 645         assert(stream);
 646         return cds_lfht_is_node_deleted(&stream->node.node);
 647 }