src/common/consumer/consumer-timer.c

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