src/common/consumer/consumer-timer.cpp

   1 /*
   2  * Copyright (C) 2012 Julien Desfossez <julien.desfossez@efficios.com>
   3  * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
   4  *
   5  * SPDX-License-Identifier: GPL-2.0-only
   6  *
   7  */
   8
   9 #define _LGPL_SOURCE
  10 #include <inttypes.h>
  11 #include <signal.h>
  12
  13 #include <bin/lttng-consumerd/health-consumerd.hpp>
  14 #include <common/common.hpp>
  15 #include <common/compat/endian.hpp>
  16 #include <common/kernel-ctl/kernel-ctl.hpp>
  17 #include <common/kernel-consumer/kernel-consumer.hpp>
  18 #include <common/consumer/consumer-stream.hpp>
  19 #include <common/consumer/consumer-timer.hpp>
  20 #include <common/consumer/consumer-testpoint.hpp>
  21 #include <common/ust-consumer/ust-consumer.hpp>
  22
  23 typedef int (*sample_positions_cb)(struct lttng_consumer_stream *stream);
  24 typedef int (*get_consumed_cb)(struct lttng_consumer_stream *stream,
  25                 unsigned long *consumed);
  26 typedef int (*get_produced_cb)(struct lttng_consumer_stream *stream,
  27                 unsigned long *produced);
  28 typedef int (*flush_index_cb)(struct lttng_consumer_stream *stream);
  29
  30 static struct timer_signal_data timer_signal = {
  31         .tid = 0,
  32         .setup_done = 0,
  33         .qs_done = 0,
  34         .lock = PTHREAD_MUTEX_INITIALIZER,
  35 };
  36
  37 /*
  38  * Set custom signal mask to current thread.
  39  */
  40 static void setmask(sigset_t *mask)
  41 {
  42         int ret;
  43
  44         ret = sigemptyset(mask);
  45         if (ret) {
  46                 PERROR("sigemptyset");
  47         }
  48         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_SWITCH);
  49         if (ret) {
  50                 PERROR("sigaddset switch");
  51         }
  52         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_TEARDOWN);
  53         if (ret) {
  54                 PERROR("sigaddset teardown");
  55         }
  56         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_LIVE);
  57         if (ret) {
  58                 PERROR("sigaddset live");
  59         }
  60         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_MONITOR);
  61         if (ret) {
  62                 PERROR("sigaddset monitor");
  63         }
  64         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_EXIT);
  65         if (ret) {
  66                 PERROR("sigaddset exit");
  67         }
  68 }
  69
  70 static int the_channel_monitor_pipe = -1;
  71
  72 /*
  73  * Execute action on a timer switch.
  74  *
  75  * Beware: metadata_switch_timer() should *never* take a mutex also held
  76  * while consumer_timer_switch_stop() is called. It would result in
  77  * deadlocks.
  78  */
  79 static void metadata_switch_timer(struct lttng_consumer_local_data *ctx,
  80                 siginfo_t *si)
  81 {
  82         int ret;
  83         struct lttng_consumer_channel *channel;
  84
  85         channel = (lttng_consumer_channel *) si->si_value.sival_ptr;
  86         LTTNG_ASSERT(channel);
  87
  88         if (channel->switch_timer_error) {
  89                 return;
  90         }
  91
  92         DBG("Switch timer for channel %" PRIu64, channel->key);
  93         switch (ctx->type) {
  94         case LTTNG_CONSUMER32_UST:
  95         case LTTNG_CONSUMER64_UST:
  96                 /*
  97                  * Locks taken by lttng_ustconsumer_request_metadata():
  98                  * - metadata_socket_lock
  99                  *   - Calling lttng_ustconsumer_recv_metadata():
 100                  *     - channel->metadata_cache->lock
 101                  *     - Calling consumer_metadata_cache_flushed():
 102                  *       - channel->timer_lock
 103                  *         - channel->metadata_cache->lock
 104                  *
 105                  * Ensure that neither consumer_data.lock nor
 106                  * channel->lock are taken within this function, since
 107                  * they are held while consumer_timer_switch_stop() is
 108                  * called.
 109                  */
 110                 ret = lttng_ustconsumer_request_metadata(ctx, channel, 1, 1);
 111                 if (ret < 0) {
 112                         channel->switch_timer_error = 1;
 113                 }
 114                 break;
 115         case LTTNG_CONSUMER_KERNEL:
 116         case LTTNG_CONSUMER_UNKNOWN:
 117                 abort();
 118                 break;
 119         }
 120 }
 121
 122 static int send_empty_index(struct lttng_consumer_stream *stream, uint64_t ts,
 123                 uint64_t stream_id)
 124 {
 125         int ret;
 126         struct ctf_packet_index index;
 127
 128         memset(&index, 0, sizeof(index));
 129         index.stream_id = htobe64(stream_id);
 130         index.timestamp_end = htobe64(ts);
 131         ret = consumer_stream_write_index(stream, &index);
 132         if (ret < 0) {
 133                 goto error;
 134         }
 135
 136 error:
 137         return ret;
 138 }
 139
 140 int consumer_flush_kernel_index(struct lttng_consumer_stream *stream)
 141 {
 142         uint64_t ts, stream_id;
 143         int ret;
 144
 145         ret = kernctl_get_current_timestamp(stream->wait_fd, &ts);
 146         if (ret < 0) {
 147                 ERR("Failed to get the current timestamp");
 148                 goto end;
 149         }
 150         ret = kernctl_buffer_flush(stream->wait_fd);
 151         if (ret < 0) {
 152                 ERR("Failed to flush kernel stream");
 153                 goto end;
 154         }
 155         ret = kernctl_snapshot(stream->wait_fd);
 156         if (ret < 0) {
 157                 if (ret != -EAGAIN && ret != -ENODATA) {
 158                         PERROR("live timer kernel snapshot");
 159                         ret = -1;
 160                         goto end;
 161                 }
 162                 ret = kernctl_get_stream_id(stream->wait_fd, &stream_id);
 163                 if (ret < 0) {
 164                         PERROR("kernctl_get_stream_id");
 165                         goto end;
 166                 }
 167                 DBG("Stream %" PRIu64 " empty, sending beacon", stream->key);
 168                 ret = send_empty_index(stream, ts, stream_id);
 169                 if (ret < 0) {
 170                         goto end;
 171                 }
 172         }
 173         ret = 0;
 174 end:
 175         return ret;
 176 }
 177
 178 static int check_stream(struct lttng_consumer_stream *stream,
 179                 flush_index_cb flush_index)
 180 {
 181         int ret;
 182
 183         /*
 184          * While holding the stream mutex, try to take a snapshot, if it
 185          * succeeds, it means that data is ready to be sent, just let the data
 186          * thread handle that. Otherwise, if the snapshot returns EAGAIN, it
 187          * means that there is no data to read after the flush, so we can
 188          * safely send the empty index.
 189          *
 190          * Doing a trylock and checking if waiting on metadata if
 191          * trylock fails. Bail out of the stream is indeed waiting for
 192          * metadata to be pushed. Busy wait on trylock otherwise.
 193          */
 194         for (;;) {
 195                 ret = pthread_mutex_trylock(&stream->lock);
 196                 switch (ret) {
 197                 case 0:
 198                         break;  /* We have the lock. */
 199                 case EBUSY:
 200                         pthread_mutex_lock(&stream->metadata_timer_lock);
 201                         if (stream->waiting_on_metadata) {
 202                                 ret = 0;
 203                                 stream->missed_metadata_flush = true;
 204                                 pthread_mutex_unlock(&stream->metadata_timer_lock);
 205                                 goto end;       /* Bail out. */
 206                         }
 207                         pthread_mutex_unlock(&stream->metadata_timer_lock);
 208                         /* Try again. */
 209                         caa_cpu_relax();
 210                         continue;
 211                 default:
 212                         ERR("Unexpected pthread_mutex_trylock error %d", ret);
 213                         ret = -1;
 214                         goto end;
 215                 }
 216                 break;
 217         }
 218         ret = flush_index(stream);
 219         pthread_mutex_unlock(&stream->lock);
 220 end:
 221         return ret;
 222 }
 223
 224 int consumer_flush_ust_index(struct lttng_consumer_stream *stream)
 225 {
 226         uint64_t ts, stream_id;
 227         int ret;
 228
 229         ret = cds_lfht_is_node_deleted(&stream->node.node);
 230         if (ret) {
 231                 goto end;
 232         }
 233
 234         ret = lttng_ustconsumer_get_current_timestamp(stream, &ts);
 235         if (ret < 0) {
 236                 ERR("Failed to get the current timestamp");
 237                 goto end;
 238         }
 239         ret = lttng_ustconsumer_flush_buffer(stream, 1);
 240         if (ret < 0) {
 241                 ERR("Failed to flush buffer while flushing index");
 242                 goto end;
 243         }
 244         ret = lttng_ustconsumer_take_snapshot(stream);
 245         if (ret < 0) {
 246                 if (ret != -EAGAIN) {
 247                         ERR("Taking UST snapshot");
 248                         ret = -1;
 249                         goto end;
 250                 }
 251                 ret = lttng_ustconsumer_get_stream_id(stream, &stream_id);
 252                 if (ret < 0) {
 253                         PERROR("lttng_ust_ctl_get_stream_id");
 254                         goto end;
 255                 }
 256                 DBG("Stream %" PRIu64 " empty, sending beacon", stream->key);
 257                 ret = send_empty_index(stream, ts, stream_id);
 258                 if (ret < 0) {
 259                         goto end;
 260                 }
 261         }
 262         ret = 0;
 263 end:
 264         return ret;
 265 }
 266
 267 /*
 268  * Execute action on a live timer
 269  */
 270 static void live_timer(struct lttng_consumer_local_data *ctx,
 271                 siginfo_t *si)
 272 {
 273         int ret;
 274         struct lttng_consumer_channel *channel;
 275         struct lttng_consumer_stream *stream;
 276         struct lttng_ht_iter iter;
 277         const struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
 278         const flush_index_cb flush_index =
 279                         ctx->type == LTTNG_CONSUMER_KERNEL ?
 280                                         consumer_flush_kernel_index :
 281                                         consumer_flush_ust_index;
 282
 283         channel = (lttng_consumer_channel *) si->si_value.sival_ptr;
 284         LTTNG_ASSERT(channel);
 285
 286         if (channel->switch_timer_error) {
 287                 goto error;
 288         }
 289
 290         DBG("Live timer for channel %" PRIu64, channel->key);
 291
 292         rcu_read_lock();
 293         cds_lfht_for_each_entry_duplicate(ht->ht,
 294                         ht->hash_fct(&channel->key, lttng_ht_seed),
 295                         ht->match_fct, &channel->key, &iter.iter,
 296                         stream, node_channel_id.node) {
 297                 ret = check_stream(stream, flush_index);
 298                 if (ret < 0) {
 299                         goto error_unlock;
 300                 }
 301         }
 302
 303 error_unlock:
 304         rcu_read_unlock();
 305
 306 error:
 307         return;
 308 }
 309
 310 static
 311 void consumer_timer_signal_thread_qs(unsigned int signr)
 312 {
 313         sigset_t pending_set;
 314         int ret;
 315
 316         /*
 317          * We need to be the only thread interacting with the thread
 318          * that manages signals for teardown synchronization.
 319          */
 320         pthread_mutex_lock(&timer_signal.lock);
 321
 322         /* Ensure we don't have any signal queued for this channel. */
 323         for (;;) {
 324                 ret = sigemptyset(&pending_set);
 325                 if (ret == -1) {
 326                         PERROR("sigemptyset");
 327                 }
 328                 ret = sigpending(&pending_set);
 329                 if (ret == -1) {
 330                         PERROR("sigpending");
 331                 }
 332                 if (!sigismember(&pending_set, signr)) {
 333                         break;
 334                 }
 335                 caa_cpu_relax();
 336         }
 337
 338         /*
 339          * From this point, no new signal handler will be fired that would try to
 340          * access "chan". However, we still need to wait for any currently
 341          * executing handler to complete.
 342          */
 343         cmm_smp_mb();
 344         CMM_STORE_SHARED(timer_signal.qs_done, 0);
 345         cmm_smp_mb();
 346
 347         /*
 348          * Kill with LTTNG_CONSUMER_SIG_TEARDOWN, so signal management thread wakes
 349          * up.
 350          */
 351         kill(getpid(), LTTNG_CONSUMER_SIG_TEARDOWN);
 352
 353         while (!CMM_LOAD_SHARED(timer_signal.qs_done)) {
 354                 caa_cpu_relax();
 355         }
 356         cmm_smp_mb();
 357
 358         pthread_mutex_unlock(&timer_signal.lock);
 359 }
 360
 361 /*
 362  * Start a timer channel timer which will fire at a given interval
 363  * (timer_interval_us)and fire a given signal (signal).
 364  *
 365  * Returns a negative value on error, 0 if a timer was created, and
 366  * a positive value if no timer was created (not an error).
 367  */
 368 static
 369 int consumer_channel_timer_start(timer_t *timer_id,
 370                 struct lttng_consumer_channel *channel,
 371                 unsigned int timer_interval_us, int signal)
 372 {
 373         int ret = 0, delete_ret;
 374         struct sigevent sev = {};
 375         struct itimerspec its;
 376
 377         LTTNG_ASSERT(channel);
 378         LTTNG_ASSERT(channel->key);
 379
 380         if (timer_interval_us == 0) {
 381                 /* No creation needed; not an error. */
 382                 ret = 1;
 383                 goto end;
 384         }
 385
 386         sev.sigev_notify = SIGEV_SIGNAL;
 387         sev.sigev_signo = signal;
 388         sev.sigev_value.sival_ptr = channel;
 389         ret = timer_create(CLOCKID, &sev, timer_id);
 390         if (ret == -1) {
 391                 PERROR("timer_create");
 392                 goto end;
 393         }
 394
 395         its.it_value.tv_sec = timer_interval_us / 1000000;
 396         its.it_value.tv_nsec = (timer_interval_us % 1000000) * 1000;
 397         its.it_interval.tv_sec = its.it_value.tv_sec;
 398         its.it_interval.tv_nsec = its.it_value.tv_nsec;
 399
 400         ret = timer_settime(*timer_id, 0, &its, NULL);
 401         if (ret == -1) {
 402                 PERROR("timer_settime");
 403                 goto error_destroy_timer;
 404         }
 405 end:
 406         return ret;
 407 error_destroy_timer:
 408         delete_ret = timer_delete(*timer_id);
 409         if (delete_ret == -1) {
 410                 PERROR("timer_delete");
 411         }
 412         goto end;
 413 }
 414
 415 static
 416 int consumer_channel_timer_stop(timer_t *timer_id, int signal)
 417 {
 418         int ret = 0;
 419
 420         ret = timer_delete(*timer_id);
 421         if (ret == -1) {
 422                 PERROR("timer_delete");
 423                 goto end;
 424         }
 425
 426         consumer_timer_signal_thread_qs(signal);
 427         *timer_id = 0;
 428 end:
 429         return ret;
 430 }
 431
 432 /*
 433  * Set the channel's switch timer.
 434  */
 435 void consumer_timer_switch_start(struct lttng_consumer_channel *channel,
 436                 unsigned int switch_timer_interval_us)
 437 {
 438         int ret;
 439
 440         LTTNG_ASSERT(channel);
 441         LTTNG_ASSERT(channel->key);
 442
 443         ret = consumer_channel_timer_start(&channel->switch_timer, channel,
 444                         switch_timer_interval_us, LTTNG_CONSUMER_SIG_SWITCH);
 445
 446         channel->switch_timer_enabled = !!(ret == 0);
 447 }
 448
 449 /*
 450  * Stop and delete the channel's switch timer.
 451  */
 452 void consumer_timer_switch_stop(struct lttng_consumer_channel *channel)
 453 {
 454         int ret;
 455
 456         LTTNG_ASSERT(channel);
 457
 458         ret = consumer_channel_timer_stop(&channel->switch_timer,
 459                         LTTNG_CONSUMER_SIG_SWITCH);
 460         if (ret == -1) {
 461                 ERR("Failed to stop switch timer");
 462         }
 463
 464         channel->switch_timer_enabled = 0;
 465 }
 466
 467 /*
 468  * Set the channel's live timer.
 469  */
 470 void consumer_timer_live_start(struct lttng_consumer_channel *channel,
 471                 unsigned int live_timer_interval_us)
 472 {
 473         int ret;
 474
 475         LTTNG_ASSERT(channel);
 476         LTTNG_ASSERT(channel->key);
 477
 478         ret = consumer_channel_timer_start(&channel->live_timer, channel,
 479                         live_timer_interval_us, LTTNG_CONSUMER_SIG_LIVE);
 480
 481         channel->live_timer_enabled = !!(ret == 0);
 482 }
 483
 484 /*
 485  * Stop and delete the channel's live timer.
 486  */
 487 void consumer_timer_live_stop(struct lttng_consumer_channel *channel)
 488 {
 489         int ret;
 490
 491         LTTNG_ASSERT(channel);
 492
 493         ret = consumer_channel_timer_stop(&channel->live_timer,
 494                         LTTNG_CONSUMER_SIG_LIVE);
 495         if (ret == -1) {
 496                 ERR("Failed to stop live timer");
 497         }
 498
 499         channel->live_timer_enabled = 0;
 500 }
 501
 502 /*
 503  * Set the channel's monitoring timer.
 504  *
 505  * Returns a negative value on error, 0 if a timer was created, and
 506  * a positive value if no timer was created (not an error).
 507  */
 508 int consumer_timer_monitor_start(struct lttng_consumer_channel *channel,
 509                 unsigned int monitor_timer_interval_us)
 510 {
 511         int ret;
 512
 513         LTTNG_ASSERT(channel);
 514         LTTNG_ASSERT(channel->key);
 515         LTTNG_ASSERT(!channel->monitor_timer_enabled);
 516
 517         ret = consumer_channel_timer_start(&channel->monitor_timer, channel,
 518                         monitor_timer_interval_us, LTTNG_CONSUMER_SIG_MONITOR);
 519         channel->monitor_timer_enabled = !!(ret == 0);
 520         return ret;
 521 }
 522
 523 /*
 524  * Stop and delete the channel's monitoring timer.
 525  */
 526 int consumer_timer_monitor_stop(struct lttng_consumer_channel *channel)
 527 {
 528         int ret;
 529
 530         LTTNG_ASSERT(channel);
 531         LTTNG_ASSERT(channel->monitor_timer_enabled);
 532
 533         ret = consumer_channel_timer_stop(&channel->monitor_timer,
 534                         LTTNG_CONSUMER_SIG_MONITOR);
 535         if (ret == -1) {
 536                 ERR("Failed to stop live timer");
 537                 goto end;
 538         }
 539
 540         channel->monitor_timer_enabled = 0;
 541 end:
 542         return ret;
 543 }
 544
 545 /*
 546  * Block the RT signals for the entire process. It must be called from the
 547  * consumer main before creating the threads
 548  */
 549 int consumer_signal_init(void)
 550 {
 551         int ret;
 552         sigset_t mask;
 553
 554         /* Block signal for entire process, so only our thread processes it. */
 555         setmask(&mask);
 556         ret = pthread_sigmask(SIG_BLOCK, &mask, NULL);
 557         if (ret) {
 558                 errno = ret;
 559                 PERROR("pthread_sigmask");
 560                 return -1;
 561         }
 562         return 0;
 563 }
 564
 565 static
 566 int sample_channel_positions(struct lttng_consumer_channel *channel,
 567                 uint64_t *_highest_use, uint64_t *_lowest_use, uint64_t *_total_consumed,
 568                 sample_positions_cb sample, get_consumed_cb get_consumed,
 569                 get_produced_cb get_produced)
 570 {
 571         int ret = 0;
 572         struct lttng_ht_iter iter;
 573         struct lttng_consumer_stream *stream;
 574         bool empty_channel = true;
 575         uint64_t high = 0, low = UINT64_MAX;
 576         struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
 577
 578         *_total_consumed = 0;
 579
 580         rcu_read_lock();
 581
 582         cds_lfht_for_each_entry_duplicate(ht->ht,
 583                         ht->hash_fct(&channel->key, lttng_ht_seed),
 584                         ht->match_fct, &channel->key,
 585                         &iter.iter, stream, node_channel_id.node) {
 586                 unsigned long produced, consumed, usage;
 587
 588                 empty_channel = false;
 589
 590                 pthread_mutex_lock(&stream->lock);
 591                 if (cds_lfht_is_node_deleted(&stream->node.node)) {
 592                         goto next;
 593                 }
 594
 595                 ret = sample(stream);
 596                 if (ret) {
 597                         ERR("Failed to take buffer position snapshot in monitor timer (ret = %d)", ret);
 598                         pthread_mutex_unlock(&stream->lock);
 599                         goto end;
 600                 }
 601                 ret = get_consumed(stream, &consumed);
 602                 if (ret) {
 603                         ERR("Failed to get buffer consumed position in monitor timer");
 604                         pthread_mutex_unlock(&stream->lock);
 605                         goto end;
 606                 }
 607                 ret = get_produced(stream, &produced);
 608                 if (ret) {
 609                         ERR("Failed to get buffer produced position in monitor timer");
 610                         pthread_mutex_unlock(&stream->lock);
 611                         goto end;
 612                 }
 613
 614                 usage = produced - consumed;
 615                 high = (usage > high) ? usage : high;
 616                 low = (usage < low) ? usage : low;
 617
 618                 /*
 619                  * We don't use consumed here for 2 reasons:
 620                  *  - output_written takes into account the padding written in the
 621                  *    tracefiles when we stop the session;
 622                  *  - the consumed position is not the accurate representation of what
 623                  *    was extracted from a buffer in overwrite mode.
 624                  */
 625                 *_total_consumed += stream->output_written;
 626         next:
 627                 pthread_mutex_unlock(&stream->lock);
 628         }
 629
 630         *_highest_use = high;
 631         *_lowest_use = low;
 632 end:
 633         rcu_read_unlock();
 634         if (empty_channel) {
 635                 ret = -1;
 636         }
 637         return ret;
 638 }
 639
 640 /*
 641  * Execute action on a monitor timer.
 642  */
 643 static
 644 void monitor_timer(struct lttng_consumer_channel *channel)
 645 {
 646         int ret;
 647         int channel_monitor_pipe =
 648                         consumer_timer_thread_get_channel_monitor_pipe();
 649         struct lttcomm_consumer_channel_monitor_msg msg = {
 650                 .key = channel->key,
 651                 .lowest = 0,
 652                 .highest = 0,
 653                 .total_consumed = 0,
 654         };
 655         sample_positions_cb sample;
 656         get_consumed_cb get_consumed;
 657         get_produced_cb get_produced;
 658         uint64_t lowest = 0, highest = 0, total_consumed = 0;
 659
 660         LTTNG_ASSERT(channel);
 661
 662         if (channel_monitor_pipe < 0) {
 663                 return;
 664         }
 665
 666         switch (the_consumer_data.type) {
 667         case LTTNG_CONSUMER_KERNEL:
 668                 sample = lttng_kconsumer_sample_snapshot_positions;
 669                 get_consumed = lttng_kconsumer_get_consumed_snapshot;
 670                 get_produced = lttng_kconsumer_get_produced_snapshot;
 671                 break;
 672         case LTTNG_CONSUMER32_UST:
 673         case LTTNG_CONSUMER64_UST:
 674                 sample = lttng_ustconsumer_sample_snapshot_positions;
 675                 get_consumed = lttng_ustconsumer_get_consumed_snapshot;
 676                 get_produced = lttng_ustconsumer_get_produced_snapshot;
 677                 break;
 678         default:
 679                 abort();
 680         }
 681
 682         ret = sample_channel_positions(channel, &highest, &lowest,
 683                         &total_consumed, sample, get_consumed, get_produced);
 684         if (ret) {
 685                 return;
 686         }
 687         msg.highest = highest;
 688         msg.lowest = lowest;
 689         msg.total_consumed = total_consumed;
 690
 691         /*
 692          * Writes performed here are assumed to be atomic which is only
 693          * guaranteed for sizes < than PIPE_BUF.
 694          */
 695         LTTNG_ASSERT(sizeof(msg) <= PIPE_BUF);
 696
 697         do {
 698                 ret = write(channel_monitor_pipe, &msg, sizeof(msg));
 699         } while (ret == -1 && errno == EINTR);
 700         if (ret == -1) {
 701                 if (errno == EAGAIN) {
 702                         /* Not an error, the sample is merely dropped. */
 703                         DBG("Channel monitor pipe is full; dropping sample for channel key = %" PRIu64,
 704                                         channel->key);
 705                 } else {
 706                         PERROR("write to the channel monitor pipe");
 707                 }
 708         } else {
 709                 DBG("Sent channel monitoring sample for channel key %" PRIu64
 710                                 ", (highest = %" PRIu64 ", lowest = %" PRIu64 ")",
 711                                 channel->key, msg.highest, msg.lowest);
 712         }
 713 }
 714
 715 int consumer_timer_thread_get_channel_monitor_pipe(void)
 716 {
 717         return uatomic_read(&the_channel_monitor_pipe);
 718 }
 719
 720 int consumer_timer_thread_set_channel_monitor_pipe(int fd)
 721 {
 722         int ret;
 723
 724         ret = uatomic_cmpxchg(&the_channel_monitor_pipe, -1, fd);
 725         if (ret != -1) {
 726                 ret = -1;
 727                 goto end;
 728         }
 729         ret = 0;
 730 end:
 731         return ret;
 732 }
 733
 734 /*
 735  * This thread is the sighandler for signals LTTNG_CONSUMER_SIG_SWITCH,
 736  * LTTNG_CONSUMER_SIG_TEARDOWN, LTTNG_CONSUMER_SIG_LIVE, and
 737  * LTTNG_CONSUMER_SIG_MONITOR, LTTNG_CONSUMER_SIG_EXIT.
 738  */
 739 void *consumer_timer_thread(void *data)
 740 {
 741         int signr;
 742         sigset_t mask;
 743         siginfo_t info;
 744         struct lttng_consumer_local_data *ctx = (lttng_consumer_local_data *) data;
 745
 746         rcu_register_thread();
 747
 748         health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA_TIMER);
 749
 750         if (testpoint(consumerd_thread_metadata_timer)) {
 751                 goto error_testpoint;
 752         }
 753
 754         health_code_update();
 755
 756         /* Only self thread will receive signal mask. */
 757         setmask(&mask);
 758         CMM_STORE_SHARED(timer_signal.tid, pthread_self());
 759
 760         while (1) {
 761                 health_code_update();
 762
 763                 health_poll_entry();
 764                 signr = sigwaitinfo(&mask, &info);
 765                 health_poll_exit();
 766
 767                 /*
 768                  * NOTE: cascading conditions are used instead of a switch case
 769                  * since the use of SIGRTMIN in the definition of the signals'
 770                  * values prevents the reduction to an integer constant.
 771                  */
 772                 if (signr == -1) {
 773                         if (errno != EINTR) {
 774                                 PERROR("sigwaitinfo");
 775                         }
 776                         continue;
 777                 } else if (signr == LTTNG_CONSUMER_SIG_SWITCH) {
 778                         metadata_switch_timer(ctx, &info);
 779                 } else if (signr == LTTNG_CONSUMER_SIG_TEARDOWN) {
 780                         cmm_smp_mb();
 781                         CMM_STORE_SHARED(timer_signal.qs_done, 1);
 782                         cmm_smp_mb();
 783                         DBG("Signal timer metadata thread teardown");
 784                 } else if (signr == LTTNG_CONSUMER_SIG_LIVE) {
 785                         live_timer(ctx, &info);
 786                 } else if (signr == LTTNG_CONSUMER_SIG_MONITOR) {
 787                         struct lttng_consumer_channel *channel;
 788
 789                         channel = (lttng_consumer_channel *) info.si_value.sival_ptr;
 790                         monitor_timer(channel);
 791                 } else if (signr == LTTNG_CONSUMER_SIG_EXIT) {
 792                         LTTNG_ASSERT(CMM_LOAD_SHARED(consumer_quit));
 793                         goto end;
 794                 } else {
 795                         ERR("Unexpected signal %d\n", info.si_signo);
 796                 }
 797         }
 798
 799 error_testpoint:
 800         /* Only reached in testpoint error */
 801         health_error();
 802 end:
 803         health_unregister(health_consumerd);
 804         rcu_unregister_thread();
 805         return NULL;
 806 }