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