src/common/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 _GNU_SOURCE
  20 #include <assert.h>
  21 #include <inttypes.h>
  22 #include <signal.h>
  23
  24 #include <bin/lttng-consumerd/health-consumerd.h>
  25 #include <common/common.h>
  26 #include <common/kernel-ctl/kernel-ctl.h>
  27 #include <common/kernel-consumer/kernel-consumer.h>
  28 #include <common/consumer-stream.h>
  29
  30 #include "consumer-timer.h"
  31 #include "consumer-testpoint.h"
  32 #include "ust-consumer/ust-consumer.h"
  33
  34 static struct timer_signal_data timer_signal = {
  35         .tid = 0,
  36         .setup_done = 0,
  37         .qs_done = 0,
  38         .lock = PTHREAD_MUTEX_INITIALIZER,
  39 };
  40
  41 /*
  42  * Set custom signal mask to current thread.
  43  */
  44 static void setmask(sigset_t *mask)
  45 {
  46         int ret;
  47
  48         ret = sigemptyset(mask);
  49         if (ret) {
  50                 PERROR("sigemptyset");
  51         }
  52         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_SWITCH);
  53         if (ret) {
  54                 PERROR("sigaddset switch");
  55         }
  56         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_TEARDOWN);
  57         if (ret) {
  58                 PERROR("sigaddset teardown");
  59         }
  60         ret = sigaddset(mask, LTTNG_CONSUMER_SIG_LIVE);
  61         if (ret) {
  62                 PERROR("sigaddset live");
  63         }
  64 }
  65
  66 /*
  67  * Execute action on a timer switch.
  68  *
  69  * Beware: metadata_switch_timer() should *never* take a mutex also held
  70  * while consumer_timer_switch_stop() is called. It would result in
  71  * deadlocks.
  72  */
  73 static void metadata_switch_timer(struct lttng_consumer_local_data *ctx,
  74                 int sig, siginfo_t *si, void *uc)
  75 {
  76         int ret;
  77         struct lttng_consumer_channel *channel;
  78
  79         channel = si->si_value.sival_ptr;
  80         assert(channel);
  81
  82         if (channel->switch_timer_error) {
  83                 return;
  84         }
  85
  86         DBG("Switch timer for channel %" PRIu64, channel->key);
  87         switch (ctx->type) {
  88         case LTTNG_CONSUMER32_UST:
  89         case LTTNG_CONSUMER64_UST:
  90                 /*
  91                  * Locks taken by lttng_ustconsumer_request_metadata():
  92                  * - metadata_socket_lock
  93                  *   - Calling lttng_ustconsumer_recv_metadata():
  94                  *     - channel->metadata_cache->lock
  95                  *     - Calling consumer_metadata_cache_flushed():
  96                  *       - channel->timer_lock
  97                  *         - channel->metadata_cache->lock
  98                  *
  99                  * Ensure that neither consumer_data.lock nor
 100                  * channel->lock are taken within this function, since
 101                  * they are held while consumer_timer_switch_stop() is
 102                  * called.
 103                  */
 104                 ret = lttng_ustconsumer_request_metadata(ctx, channel, 1, 1);
 105                 if (ret < 0) {
 106                         channel->switch_timer_error = 1;
 107                 }
 108                 break;
 109         case LTTNG_CONSUMER_KERNEL:
 110         case LTTNG_CONSUMER_UNKNOWN:
 111                 assert(0);
 112                 break;
 113         }
 114 }
 115
 116 static int send_empty_index(struct lttng_consumer_stream *stream, uint64_t ts)
 117 {
 118         int ret;
 119         struct ctf_packet_index index;
 120
 121         memset(&index, 0, sizeof(index));
 122         index.timestamp_end = htobe64(ts);
 123         ret = consumer_stream_write_index(stream, &index);
 124         if (ret < 0) {
 125                 goto error;
 126         }
 127
 128 error:
 129         return ret;
 130 }
 131
 132 static int check_kernel_stream(struct lttng_consumer_stream *stream)
 133 {
 134         uint64_t ts;
 135         int ret;
 136
 137         /*
 138          * While holding the stream mutex, try to take a snapshot, if it
 139          * succeeds, it means that data is ready to be sent, just let the data
 140          * thread handle that. Otherwise, if the snapshot returns EAGAIN, it
 141          * means that there is no data to read after the flush, so we can
 142          * safely send the empty index.
 143          */
 144         pthread_mutex_lock(&stream->lock);
 145         ret = kernctl_get_current_timestamp(stream->wait_fd, &ts);
 146         if (ret < 0) {
 147                 ERR("Failed to get the current timestamp");
 148                 goto error_unlock;
 149         }
 150         ret = kernctl_buffer_flush(stream->wait_fd);
 151         if (ret < 0) {
 152                 ERR("Failed to flush kernel stream");
 153                 goto error_unlock;
 154         }
 155         ret = kernctl_snapshot(stream->wait_fd);
 156         if (ret < 0) {
 157                 if (errno != EAGAIN && errno != ENODATA) {
 158                         PERROR("live timer kernel snapshot");
 159                         ret = -1;
 160                         goto error_unlock;
 161                 }
 162                 DBG("Stream %" PRIu64 " empty, sending beacon", stream->key);
 163                 ret = send_empty_index(stream, ts);
 164                 if (ret < 0) {
 165                         goto error_unlock;
 166                 }
 167         }
 168         ret = 0;
 169
 170 error_unlock:
 171         pthread_mutex_unlock(&stream->lock);
 172         return ret;
 173 }
 174
 175 static int check_ust_stream(struct lttng_consumer_stream *stream)
 176 {
 177         uint64_t ts;
 178         int ret;
 179
 180         assert(stream);
 181         assert(stream->ustream);
 182         /*
 183          * While holding the stream mutex, try to take a snapshot, if it
 184          * succeeds, it means that data is ready to be sent, just let the data
 185          * thread handle that. Otherwise, if the snapshot returns EAGAIN, it
 186          * means that there is no data to read after the flush, so we can
 187          * safely send the empty index.
 188          */
 189         pthread_mutex_lock(&stream->lock);
 190         ret = cds_lfht_is_node_deleted(&stream->node.node);
 191         if (ret) {
 192                 goto error_unlock;
 193         }
 194
 195         ret = lttng_ustconsumer_get_current_timestamp(stream, &ts);
 196         if (ret < 0) {
 197                 ERR("Failed to get the current timestamp");
 198                 goto error_unlock;
 199         }
 200         lttng_ustconsumer_flush_buffer(stream, 1);
 201         ret = lttng_ustconsumer_take_snapshot(stream);
 202         if (ret < 0) {
 203                 if (ret != -EAGAIN) {
 204                         ERR("Taking UST snapshot");
 205                         ret = -1;
 206                         goto error_unlock;
 207                 }
 208                 DBG("Stream %" PRIu64 " empty, sending beacon", stream->key);
 209                 ret = send_empty_index(stream, ts);
 210                 if (ret < 0) {
 211                         goto error_unlock;
 212                 }
 213         }
 214         ret = 0;
 215
 216 error_unlock:
 217         pthread_mutex_unlock(&stream->lock);
 218         return ret;
 219 }
 220
 221 /*
 222  * Execute action on a live timer
 223  */
 224 static void live_timer(struct lttng_consumer_local_data *ctx,
 225                 int sig, siginfo_t *si, void *uc)
 226 {
 227         int ret;
 228         struct lttng_consumer_channel *channel;
 229         struct lttng_consumer_stream *stream;
 230         struct lttng_ht *ht;
 231         struct lttng_ht_iter iter;
 232
 233         channel = si->si_value.sival_ptr;
 234         assert(channel);
 235
 236         if (channel->switch_timer_error) {
 237                 goto error;
 238         }
 239         ht = consumer_data.stream_per_chan_id_ht;
 240
 241         DBG("Live timer for channel %" PRIu64, channel->key);
 242
 243         rcu_read_lock();
 244         switch (ctx->type) {
 245         case LTTNG_CONSUMER32_UST:
 246         case LTTNG_CONSUMER64_UST:
 247                 cds_lfht_for_each_entry_duplicate(ht->ht,
 248                                 ht->hash_fct(&channel->key, lttng_ht_seed),
 249                                 ht->match_fct, &channel->key, &iter.iter,
 250                                 stream, node_channel_id.node) {
 251                         ret = check_ust_stream(stream);
 252                         if (ret < 0) {
 253                                 goto error_unlock;
 254                         }
 255                 }
 256                 break;
 257         case LTTNG_CONSUMER_KERNEL:
 258                 cds_lfht_for_each_entry_duplicate(ht->ht,
 259                                 ht->hash_fct(&channel->key, lttng_ht_seed),
 260                                 ht->match_fct, &channel->key, &iter.iter,
 261                                 stream, node_channel_id.node) {
 262                         ret = check_kernel_stream(stream);
 263                         if (ret < 0) {
 264                                 goto error_unlock;
 265                         }
 266                 }
 267                 break;
 268         case LTTNG_CONSUMER_UNKNOWN:
 269                 assert(0);
 270                 break;
 271         }
 272
 273 error_unlock:
 274         rcu_read_unlock();
 275
 276 error:
 277         return;
 278 }
 279
 280 static
 281 void consumer_timer_signal_thread_qs(unsigned int signr)
 282 {
 283         sigset_t pending_set;
 284         int ret;
 285
 286         /*
 287          * We need to be the only thread interacting with the thread
 288          * that manages signals for teardown synchronization.
 289          */
 290         pthread_mutex_lock(&timer_signal.lock);
 291
 292         /* Ensure we don't have any signal queued for this channel. */
 293         for (;;) {
 294                 ret = sigemptyset(&pending_set);
 295                 if (ret == -1) {
 296                         PERROR("sigemptyset");
 297                 }
 298                 ret = sigpending(&pending_set);
 299                 if (ret == -1) {
 300                         PERROR("sigpending");
 301                 }
 302                 if (!sigismember(&pending_set, LTTNG_CONSUMER_SIG_SWITCH)) {
 303                         break;
 304                 }
 305                 caa_cpu_relax();
 306         }
 307
 308         /*
 309          * From this point, no new signal handler will be fired that would try to
 310          * access "chan". However, we still need to wait for any currently
 311          * executing handler to complete.
 312          */
 313         cmm_smp_mb();
 314         CMM_STORE_SHARED(timer_signal.qs_done, 0);
 315         cmm_smp_mb();
 316
 317         /*
 318          * Kill with LTTNG_CONSUMER_SIG_TEARDOWN, so signal management thread wakes
 319          * up.
 320          */
 321         kill(getpid(), LTTNG_CONSUMER_SIG_TEARDOWN);
 322
 323         while (!CMM_LOAD_SHARED(timer_signal.qs_done)) {
 324                 caa_cpu_relax();
 325         }
 326         cmm_smp_mb();
 327
 328         pthread_mutex_unlock(&timer_signal.lock);
 329 }
 330
 331 /*
 332  * Set the timer for periodical metadata flush.
 333  */
 334 void consumer_timer_switch_start(struct lttng_consumer_channel *channel,
 335                 unsigned int switch_timer_interval)
 336 {
 337         int ret;
 338         struct sigevent sev;
 339         struct itimerspec its;
 340
 341         assert(channel);
 342         assert(channel->key);
 343
 344         if (switch_timer_interval == 0) {
 345                 return;
 346         }
 347
 348         sev.sigev_notify = SIGEV_SIGNAL;
 349         sev.sigev_signo = LTTNG_CONSUMER_SIG_SWITCH;
 350         sev.sigev_value.sival_ptr = channel;
 351         ret = timer_create(CLOCKID, &sev, &channel->switch_timer);
 352         if (ret == -1) {
 353                 PERROR("timer_create");
 354         }
 355         channel->switch_timer_enabled = 1;
 356
 357         its.it_value.tv_sec = switch_timer_interval / 1000000;
 358         its.it_value.tv_nsec = switch_timer_interval % 1000000;
 359         its.it_interval.tv_sec = its.it_value.tv_sec;
 360         its.it_interval.tv_nsec = its.it_value.tv_nsec;
 361
 362         ret = timer_settime(channel->switch_timer, 0, &its, NULL);
 363         if (ret == -1) {
 364                 PERROR("timer_settime");
 365         }
 366 }
 367
 368 /*
 369  * Stop and delete timer.
 370  */
 371 void consumer_timer_switch_stop(struct lttng_consumer_channel *channel)
 372 {
 373         int ret;
 374
 375         assert(channel);
 376
 377         ret = timer_delete(channel->switch_timer);
 378         if (ret == -1) {
 379                 PERROR("timer_delete");
 380         }
 381
 382         consumer_timer_signal_thread_qs(LTTNG_CONSUMER_SIG_SWITCH);
 383
 384         channel->switch_timer = 0;
 385         channel->switch_timer_enabled = 0;
 386 }
 387
 388 /*
 389  * Set the timer for the live mode.
 390  */
 391 void consumer_timer_live_start(struct lttng_consumer_channel *channel,
 392                 int live_timer_interval)
 393 {
 394         int ret;
 395         struct sigevent sev;
 396         struct itimerspec its;
 397
 398         assert(channel);
 399         assert(channel->key);
 400
 401         if (live_timer_interval <= 0) {
 402                 return;
 403         }
 404
 405         sev.sigev_notify = SIGEV_SIGNAL;
 406         sev.sigev_signo = LTTNG_CONSUMER_SIG_LIVE;
 407         sev.sigev_value.sival_ptr = channel;
 408         ret = timer_create(CLOCKID, &sev, &channel->live_timer);
 409         if (ret == -1) {
 410                 PERROR("timer_create");
 411         }
 412         channel->live_timer_enabled = 1;
 413
 414         its.it_value.tv_sec = live_timer_interval / 1000000;
 415         its.it_value.tv_nsec = live_timer_interval % 1000000;
 416         its.it_interval.tv_sec = its.it_value.tv_sec;
 417         its.it_interval.tv_nsec = its.it_value.tv_nsec;
 418
 419         ret = timer_settime(channel->live_timer, 0, &its, NULL);
 420         if (ret == -1) {
 421                 PERROR("timer_settime");
 422         }
 423 }
 424
 425 /*
 426  * Stop and delete timer.
 427  */
 428 void consumer_timer_live_stop(struct lttng_consumer_channel *channel)
 429 {
 430         int ret;
 431
 432         assert(channel);
 433
 434         ret = timer_delete(channel->live_timer);
 435         if (ret == -1) {
 436                 PERROR("timer_delete");
 437         }
 438
 439         consumer_timer_signal_thread_qs(LTTNG_CONSUMER_SIG_LIVE);
 440
 441         channel->live_timer = 0;
 442         channel->live_timer_enabled = 0;
 443 }
 444
 445 /*
 446  * Block the RT signals for the entire process. It must be called from the
 447  * consumer main before creating the threads
 448  */
 449 void consumer_signal_init(void)
 450 {
 451         int ret;
 452         sigset_t mask;
 453
 454         /* Block signal for entire process, so only our thread processes it. */
 455         setmask(&mask);
 456         ret = pthread_sigmask(SIG_BLOCK, &mask, NULL);
 457         if (ret) {
 458                 errno = ret;
 459                 PERROR("pthread_sigmask");
 460         }
 461 }
 462
 463 /*
 464  * This thread is the sighandler for signals LTTNG_CONSUMER_SIG_SWITCH,
 465  * LTTNG_CONSUMER_SIG_TEARDOWN and LTTNG_CONSUMER_SIG_LIVE.
 466  */
 467 void *consumer_timer_thread(void *data)
 468 {
 469         int signr;
 470         sigset_t mask;
 471         siginfo_t info;
 472         struct lttng_consumer_local_data *ctx = data;
 473
 474         health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA_TIMER);
 475
 476         if (testpoint(consumerd_thread_metadata_timer)) {
 477                 goto error_testpoint;
 478         }
 479
 480         health_code_update();
 481
 482         /* Only self thread will receive signal mask. */
 483         setmask(&mask);
 484         CMM_STORE_SHARED(timer_signal.tid, pthread_self());
 485
 486         while (1) {
 487                 health_code_update();
 488
 489                 health_poll_entry();
 490                 signr = sigwaitinfo(&mask, &info);
 491                 health_poll_exit();
 492                 if (signr == -1) {
 493                         if (errno != EINTR) {
 494                                 PERROR("sigwaitinfo");
 495                         }
 496                         continue;
 497                 } else if (signr == LTTNG_CONSUMER_SIG_SWITCH) {
 498                         metadata_switch_timer(ctx, info.si_signo, &info, NULL);
 499                 } else if (signr == LTTNG_CONSUMER_SIG_TEARDOWN) {
 500                         cmm_smp_mb();
 501                         CMM_STORE_SHARED(timer_signal.qs_done, 1);
 502                         cmm_smp_mb();
 503                         DBG("Signal timer metadata thread teardown");
 504                 } else if (signr == LTTNG_CONSUMER_SIG_LIVE) {
 505                         live_timer(ctx, info.si_signo, &info, NULL);
 506                 } else {
 507                         ERR("Unexpected signal %d\n", info.si_signo);
 508                 }
 509         }
 510
 511 error_testpoint:
 512         /* Only reached in testpoint error */
 513         health_error();
 514         health_unregister(health_consumerd);
 515
 516         /* Never return */
 517         return NULL;
 518 }