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