Session consumed size notification
[lttng-tools.git] / 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 int sig, 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 int sig, 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;
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.si_signo, &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.si_signo, &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 }
This page took 0.077926 seconds and 5 git commands to generate.