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