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