Fix: work-around glibc __nptl_setxid vs clone hang
[lttng-tools.git] / src / common / consumer.c
CommitLineData
3bd1e081
MD
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 *
d14d33bf
AM
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
3bd1e081 8 *
d14d33bf
AM
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.
3bd1e081 13 *
d14d33bf
AM
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
17 */
18
19#define _GNU_SOURCE
20#include <assert.h>
3bd1e081
MD
21#include <poll.h>
22#include <pthread.h>
23#include <stdlib.h>
24#include <string.h>
25#include <sys/mman.h>
26#include <sys/socket.h>
27#include <sys/types.h>
28#include <unistd.h>
29
990570ed 30#include <common/common.h>
10a8a223
DG
31#include <common/kernel-ctl/kernel-ctl.h>
32#include <common/sessiond-comm/sessiond-comm.h>
33#include <common/kernel-consumer/kernel-consumer.h>
34#include <common/ust-consumer/ust-consumer.h>
35
36#include "consumer.h"
3bd1e081
MD
37
38struct lttng_consumer_global_data consumer_data = {
3bd1e081
MD
39 .stream_count = 0,
40 .need_update = 1,
41 .type = LTTNG_CONSUMER_UNKNOWN,
42};
43
44/* timeout parameter, to control the polling thread grace period. */
45int consumer_poll_timeout = -1;
46
47/*
48 * Flag to inform the polling thread to quit when all fd hung up. Updated by
49 * the consumer_thread_receive_fds when it notices that all fds has hung up.
50 * Also updated by the signal handler (consumer_should_exit()). Read by the
51 * polling threads.
52 */
53volatile int consumer_quit = 0;
54
55/*
56 * Find a stream. The consumer_data.lock must be locked during this
57 * call.
58 */
59static struct lttng_consumer_stream *consumer_find_stream(int key)
60{
e4421fec
DG
61 struct lttng_ht_iter iter;
62 struct lttng_ht_node_ulong *node;
63 struct lttng_consumer_stream *stream = NULL;
3bd1e081 64
7ad0a0cb
MD
65 /* Negative keys are lookup failures */
66 if (key < 0)
67 return NULL;
e4421fec 68
6065ceec
DG
69 rcu_read_lock();
70
e4421fec
DG
71 lttng_ht_lookup(consumer_data.stream_ht, (void *)((unsigned long) key),
72 &iter);
73 node = lttng_ht_iter_get_node_ulong(&iter);
74 if (node != NULL) {
75 stream = caa_container_of(node, struct lttng_consumer_stream, node);
3bd1e081 76 }
e4421fec 77
6065ceec
DG
78 rcu_read_unlock();
79
e4421fec 80 return stream;
3bd1e081
MD
81}
82
7ad0a0cb
MD
83static void consumer_steal_stream_key(int key)
84{
85 struct lttng_consumer_stream *stream;
86
04253271 87 rcu_read_lock();
7ad0a0cb 88 stream = consumer_find_stream(key);
04253271 89 if (stream) {
7ad0a0cb 90 stream->key = -1;
04253271
MD
91 /*
92 * We don't want the lookup to match, but we still need
93 * to iterate on this stream when iterating over the hash table. Just
94 * change the node key.
95 */
96 stream->node.key = -1;
97 }
98 rcu_read_unlock();
7ad0a0cb
MD
99}
100
3bd1e081
MD
101static struct lttng_consumer_channel *consumer_find_channel(int key)
102{
e4421fec
DG
103 struct lttng_ht_iter iter;
104 struct lttng_ht_node_ulong *node;
105 struct lttng_consumer_channel *channel = NULL;
3bd1e081 106
7ad0a0cb
MD
107 /* Negative keys are lookup failures */
108 if (key < 0)
109 return NULL;
e4421fec 110
6065ceec
DG
111 rcu_read_lock();
112
e4421fec
DG
113 lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
114 &iter);
115 node = lttng_ht_iter_get_node_ulong(&iter);
116 if (node != NULL) {
117 channel = caa_container_of(node, struct lttng_consumer_channel, node);
3bd1e081 118 }
e4421fec 119
6065ceec
DG
120 rcu_read_unlock();
121
e4421fec 122 return channel;
3bd1e081
MD
123}
124
7ad0a0cb
MD
125static void consumer_steal_channel_key(int key)
126{
127 struct lttng_consumer_channel *channel;
128
04253271 129 rcu_read_lock();
7ad0a0cb 130 channel = consumer_find_channel(key);
04253271 131 if (channel) {
7ad0a0cb 132 channel->key = -1;
04253271
MD
133 /*
134 * We don't want the lookup to match, but we still need
135 * to iterate on this channel when iterating over the hash table. Just
136 * change the node key.
137 */
138 channel->node.key = -1;
139 }
140 rcu_read_unlock();
7ad0a0cb
MD
141}
142
702b1ea4
MD
143static
144void consumer_free_stream(struct rcu_head *head)
145{
146 struct lttng_ht_node_ulong *node =
147 caa_container_of(head, struct lttng_ht_node_ulong, head);
148 struct lttng_consumer_stream *stream =
149 caa_container_of(node, struct lttng_consumer_stream, node);
150
151 free(stream);
152}
153
3bd1e081
MD
154/*
155 * Remove a stream from the global list protected by a mutex. This
156 * function is also responsible for freeing its data structures.
157 */
158void consumer_del_stream(struct lttng_consumer_stream *stream)
159{
160 int ret;
e4421fec 161 struct lttng_ht_iter iter;
3bd1e081
MD
162 struct lttng_consumer_channel *free_chan = NULL;
163
164 pthread_mutex_lock(&consumer_data.lock);
165
166 switch (consumer_data.type) {
167 case LTTNG_CONSUMER_KERNEL:
168 if (stream->mmap_base != NULL) {
169 ret = munmap(stream->mmap_base, stream->mmap_len);
170 if (ret != 0) {
171 perror("munmap");
172 }
173 }
174 break;
7753dea8
MD
175 case LTTNG_CONSUMER32_UST:
176 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
177 lttng_ustconsumer_del_stream(stream);
178 break;
179 default:
180 ERR("Unknown consumer_data type");
181 assert(0);
182 goto end;
183 }
184
6065ceec 185 rcu_read_lock();
04253271
MD
186 iter.iter.node = &stream->node.node;
187 ret = lttng_ht_del(consumer_data.stream_ht, &iter);
188 assert(!ret);
e4421fec 189
6065ceec
DG
190 rcu_read_unlock();
191
3bd1e081
MD
192 if (consumer_data.stream_count <= 0) {
193 goto end;
194 }
195 consumer_data.stream_count--;
196 if (!stream) {
197 goto end;
198 }
199 if (stream->out_fd >= 0) {
4c462e79
MD
200 ret = close(stream->out_fd);
201 if (ret) {
202 PERROR("close");
203 }
3bd1e081 204 }
b5c5fc29 205 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
4c462e79
MD
206 ret = close(stream->wait_fd);
207 if (ret) {
208 PERROR("close");
209 }
3bd1e081 210 }
2c1dd183 211 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
4c462e79
MD
212 ret = close(stream->shm_fd);
213 if (ret) {
214 PERROR("close");
215 }
3bd1e081
MD
216 }
217 if (!--stream->chan->refcount)
218 free_chan = stream->chan;
702b1ea4
MD
219
220 call_rcu(&stream->node.head, consumer_free_stream);
3bd1e081
MD
221end:
222 consumer_data.need_update = 1;
223 pthread_mutex_unlock(&consumer_data.lock);
224
225 if (free_chan)
226 consumer_del_channel(free_chan);
227}
228
229struct lttng_consumer_stream *consumer_allocate_stream(
230 int channel_key, int stream_key,
231 int shm_fd, int wait_fd,
232 enum lttng_consumer_stream_state state,
233 uint64_t mmap_len,
234 enum lttng_event_output output,
6df2e2c9
MD
235 const char *path_name,
236 uid_t uid,
237 gid_t gid)
3bd1e081
MD
238{
239 struct lttng_consumer_stream *stream;
240 int ret;
241
effcf122 242 stream = zmalloc(sizeof(*stream));
3bd1e081
MD
243 if (stream == NULL) {
244 perror("malloc struct lttng_consumer_stream");
245 goto end;
246 }
247 stream->chan = consumer_find_channel(channel_key);
248 if (!stream->chan) {
249 perror("Unable to find channel key");
250 goto end;
251 }
252 stream->chan->refcount++;
253 stream->key = stream_key;
254 stream->shm_fd = shm_fd;
255 stream->wait_fd = wait_fd;
256 stream->out_fd = -1;
257 stream->out_fd_offset = 0;
258 stream->state = state;
259 stream->mmap_len = mmap_len;
260 stream->mmap_base = NULL;
261 stream->output = output;
6df2e2c9
MD
262 stream->uid = uid;
263 stream->gid = gid;
3bd1e081
MD
264 strncpy(stream->path_name, path_name, PATH_MAX - 1);
265 stream->path_name[PATH_MAX - 1] = '\0';
e4421fec 266 lttng_ht_node_init_ulong(&stream->node, stream->key);
3bd1e081
MD
267
268 switch (consumer_data.type) {
269 case LTTNG_CONSUMER_KERNEL:
270 break;
7753dea8
MD
271 case LTTNG_CONSUMER32_UST:
272 case LTTNG_CONSUMER64_UST:
5af2f756 273 stream->cpu = stream->chan->cpucount++;
3bd1e081
MD
274 ret = lttng_ustconsumer_allocate_stream(stream);
275 if (ret) {
276 free(stream);
277 return NULL;
278 }
279 break;
280 default:
281 ERR("Unknown consumer_data type");
282 assert(0);
283 goto end;
284 }
285 DBG("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, out_fd %d)",
286 stream->path_name, stream->key,
287 stream->shm_fd,
288 stream->wait_fd,
289 (unsigned long long) stream->mmap_len,
290 stream->out_fd);
291end:
292 return stream;
293}
294
295/*
296 * Add a stream to the global list protected by a mutex.
297 */
298int consumer_add_stream(struct lttng_consumer_stream *stream)
299{
300 int ret = 0;
c77fc10a
DG
301 struct lttng_ht_node_ulong *node;
302 struct lttng_ht_iter iter;
3bd1e081
MD
303
304 pthread_mutex_lock(&consumer_data.lock);
7ad0a0cb
MD
305 /* Steal stream identifier, for UST */
306 consumer_steal_stream_key(stream->key);
6065ceec 307 rcu_read_lock();
c77fc10a
DG
308
309 lttng_ht_lookup(consumer_data.stream_ht,
310 (void *)((unsigned long) stream->key), &iter);
311 node = lttng_ht_iter_get_node_ulong(&iter);
312 if (node != NULL) {
313 rcu_read_unlock();
314 /* Stream already exist. Ignore the insertion */
315 goto end;
316 }
317
04253271 318 lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
6065ceec 319 rcu_read_unlock();
3bd1e081
MD
320 consumer_data.stream_count++;
321 consumer_data.need_update = 1;
322
323 switch (consumer_data.type) {
324 case LTTNG_CONSUMER_KERNEL:
325 break;
7753dea8
MD
326 case LTTNG_CONSUMER32_UST:
327 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
328 /* Streams are in CPU number order (we rely on this) */
329 stream->cpu = stream->chan->nr_streams++;
330 break;
331 default:
332 ERR("Unknown consumer_data type");
333 assert(0);
334 goto end;
335 }
336
337end:
338 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 339
3bd1e081
MD
340 return ret;
341}
342
343/*
344 * Update a stream according to what we just received.
345 */
346void consumer_change_stream_state(int stream_key,
347 enum lttng_consumer_stream_state state)
348{
349 struct lttng_consumer_stream *stream;
350
351 pthread_mutex_lock(&consumer_data.lock);
352 stream = consumer_find_stream(stream_key);
353 if (stream) {
354 stream->state = state;
355 }
356 consumer_data.need_update = 1;
357 pthread_mutex_unlock(&consumer_data.lock);
358}
359
702b1ea4
MD
360static
361void consumer_free_channel(struct rcu_head *head)
362{
363 struct lttng_ht_node_ulong *node =
364 caa_container_of(head, struct lttng_ht_node_ulong, head);
365 struct lttng_consumer_channel *channel =
366 caa_container_of(node, struct lttng_consumer_channel, node);
367
368 free(channel);
369}
370
3bd1e081
MD
371/*
372 * Remove a channel from the global list protected by a mutex. This
373 * function is also responsible for freeing its data structures.
374 */
375void consumer_del_channel(struct lttng_consumer_channel *channel)
376{
377 int ret;
e4421fec 378 struct lttng_ht_iter iter;
3bd1e081
MD
379
380 pthread_mutex_lock(&consumer_data.lock);
381
382 switch (consumer_data.type) {
383 case LTTNG_CONSUMER_KERNEL:
384 break;
7753dea8
MD
385 case LTTNG_CONSUMER32_UST:
386 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
387 lttng_ustconsumer_del_channel(channel);
388 break;
389 default:
390 ERR("Unknown consumer_data type");
391 assert(0);
392 goto end;
393 }
394
6065ceec 395 rcu_read_lock();
04253271
MD
396 iter.iter.node = &channel->node.node;
397 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
398 assert(!ret);
6065ceec
DG
399 rcu_read_unlock();
400
3bd1e081
MD
401 if (channel->mmap_base != NULL) {
402 ret = munmap(channel->mmap_base, channel->mmap_len);
403 if (ret != 0) {
404 perror("munmap");
405 }
406 }
b5c5fc29 407 if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
4c462e79
MD
408 ret = close(channel->wait_fd);
409 if (ret) {
410 PERROR("close");
411 }
3bd1e081 412 }
2c1dd183 413 if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
4c462e79
MD
414 ret = close(channel->shm_fd);
415 if (ret) {
416 PERROR("close");
417 }
3bd1e081 418 }
702b1ea4
MD
419
420 call_rcu(&channel->node.head, consumer_free_channel);
3bd1e081
MD
421end:
422 pthread_mutex_unlock(&consumer_data.lock);
423}
424
425struct lttng_consumer_channel *consumer_allocate_channel(
426 int channel_key,
427 int shm_fd, int wait_fd,
428 uint64_t mmap_len,
429 uint64_t max_sb_size)
430{
431 struct lttng_consumer_channel *channel;
432 int ret;
433
276b26d1 434 channel = zmalloc(sizeof(*channel));
3bd1e081
MD
435 if (channel == NULL) {
436 perror("malloc struct lttng_consumer_channel");
437 goto end;
438 }
439 channel->key = channel_key;
440 channel->shm_fd = shm_fd;
441 channel->wait_fd = wait_fd;
442 channel->mmap_len = mmap_len;
443 channel->max_sb_size = max_sb_size;
444 channel->refcount = 0;
445 channel->nr_streams = 0;
e4421fec 446 lttng_ht_node_init_ulong(&channel->node, channel->key);
3bd1e081
MD
447
448 switch (consumer_data.type) {
449 case LTTNG_CONSUMER_KERNEL:
450 channel->mmap_base = NULL;
451 channel->mmap_len = 0;
452 break;
7753dea8
MD
453 case LTTNG_CONSUMER32_UST:
454 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
455 ret = lttng_ustconsumer_allocate_channel(channel);
456 if (ret) {
457 free(channel);
458 return NULL;
459 }
460 break;
461 default:
462 ERR("Unknown consumer_data type");
463 assert(0);
464 goto end;
465 }
466 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
467 channel->key,
468 channel->shm_fd,
469 channel->wait_fd,
470 (unsigned long long) channel->mmap_len,
471 (unsigned long long) channel->max_sb_size);
472end:
473 return channel;
474}
475
476/*
477 * Add a channel to the global list protected by a mutex.
478 */
479int consumer_add_channel(struct lttng_consumer_channel *channel)
480{
c77fc10a
DG
481 struct lttng_ht_node_ulong *node;
482 struct lttng_ht_iter iter;
483
3bd1e081 484 pthread_mutex_lock(&consumer_data.lock);
7ad0a0cb
MD
485 /* Steal channel identifier, for UST */
486 consumer_steal_channel_key(channel->key);
6065ceec 487 rcu_read_lock();
c77fc10a
DG
488
489 lttng_ht_lookup(consumer_data.channel_ht,
490 (void *)((unsigned long) channel->key), &iter);
491 node = lttng_ht_iter_get_node_ulong(&iter);
492 if (node != NULL) {
493 /* Channel already exist. Ignore the insertion */
494 goto end;
495 }
496
04253271 497 lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
c77fc10a
DG
498
499end:
6065ceec 500 rcu_read_unlock();
3bd1e081 501 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 502
7ad0a0cb 503 return 0;
3bd1e081
MD
504}
505
506/*
507 * Allocate the pollfd structure and the local view of the out fds to avoid
508 * doing a lookup in the linked list and concurrency issues when writing is
509 * needed. Called with consumer_data.lock held.
510 *
511 * Returns the number of fds in the structures.
512 */
513int consumer_update_poll_array(
514 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
515 struct lttng_consumer_stream **local_stream)
516{
3bd1e081 517 int i = 0;
e4421fec
DG
518 struct lttng_ht_iter iter;
519 struct lttng_consumer_stream *stream;
3bd1e081
MD
520
521 DBG("Updating poll fd array");
481d6c57 522 rcu_read_lock();
e4421fec
DG
523 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
524 node.node) {
525 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
3bd1e081
MD
526 continue;
527 }
e4421fec
DG
528 DBG("Active FD %d", stream->wait_fd);
529 (*pollfd)[i].fd = stream->wait_fd;
3bd1e081 530 (*pollfd)[i].events = POLLIN | POLLPRI;
e4421fec 531 local_stream[i] = stream;
3bd1e081
MD
532 i++;
533 }
481d6c57 534 rcu_read_unlock();
3bd1e081
MD
535
536 /*
537 * Insert the consumer_poll_pipe at the end of the array and don't
538 * increment i so nb_fd is the number of real FD.
539 */
540 (*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
509bb1cf 541 (*pollfd)[i].events = POLLIN | POLLPRI;
3bd1e081
MD
542 return i;
543}
544
545/*
546 * Poll on the should_quit pipe and the command socket return -1 on error and
547 * should exit, 0 if data is available on the command socket
548 */
549int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
550{
551 int num_rdy;
552
88f2b785 553restart:
3bd1e081
MD
554 num_rdy = poll(consumer_sockpoll, 2, -1);
555 if (num_rdy == -1) {
88f2b785
MD
556 /*
557 * Restart interrupted system call.
558 */
559 if (errno == EINTR) {
560 goto restart;
561 }
3bd1e081
MD
562 perror("Poll error");
563 goto exit;
564 }
509bb1cf 565 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
3bd1e081
MD
566 DBG("consumer_should_quit wake up");
567 goto exit;
568 }
569 return 0;
570
571exit:
572 return -1;
573}
574
575/*
576 * Set the error socket.
577 */
578void lttng_consumer_set_error_sock(
579 struct lttng_consumer_local_data *ctx, int sock)
580{
581 ctx->consumer_error_socket = sock;
582}
583
584/*
585 * Set the command socket path.
586 */
587
588void lttng_consumer_set_command_sock_path(
589 struct lttng_consumer_local_data *ctx, char *sock)
590{
591 ctx->consumer_command_sock_path = sock;
592}
593
594/*
595 * Send return code to the session daemon.
596 * If the socket is not defined, we return 0, it is not a fatal error
597 */
598int lttng_consumer_send_error(
599 struct lttng_consumer_local_data *ctx, int cmd)
600{
601 if (ctx->consumer_error_socket > 0) {
602 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
603 sizeof(enum lttcomm_sessiond_command));
604 }
605
606 return 0;
607}
608
609/*
610 * Close all the tracefiles and stream fds, should be called when all instances
611 * are destroyed.
612 */
613void lttng_consumer_cleanup(void)
614{
e4421fec 615 struct lttng_ht_iter iter;
6065ceec
DG
616 struct lttng_ht_node_ulong *node;
617
618 rcu_read_lock();
3bd1e081
MD
619
620 /*
6065ceec
DG
621 * close all outfd. Called when there are no more threads running (after
622 * joining on the threads), no need to protect list iteration with mutex.
3bd1e081 623 */
6065ceec
DG
624 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
625 node) {
702b1ea4
MD
626 struct lttng_consumer_stream *stream =
627 caa_container_of(node, struct lttng_consumer_stream, node);
628 consumer_del_stream(stream);
3bd1e081 629 }
e4421fec 630
6065ceec
DG
631 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
632 node) {
702b1ea4
MD
633 struct lttng_consumer_channel *channel =
634 caa_container_of(node, struct lttng_consumer_channel, node);
635 consumer_del_channel(channel);
3bd1e081 636 }
6065ceec
DG
637
638 rcu_read_unlock();
3bd1e081
MD
639}
640
641/*
642 * Called from signal handler.
643 */
644void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
645{
646 int ret;
647 consumer_quit = 1;
648 ret = write(ctx->consumer_should_quit[1], "4", 1);
649 if (ret < 0) {
650 perror("write consumer quit");
651 }
652}
653
654void lttng_consumer_sync_trace_file(
655 struct lttng_consumer_stream *stream, off_t orig_offset)
656{
657 int outfd = stream->out_fd;
658
659 /*
660 * This does a blocking write-and-wait on any page that belongs to the
661 * subbuffer prior to the one we just wrote.
662 * Don't care about error values, as these are just hints and ways to
663 * limit the amount of page cache used.
664 */
665 if (orig_offset < stream->chan->max_sb_size) {
666 return;
667 }
b9182dd9 668 lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
3bd1e081
MD
669 stream->chan->max_sb_size,
670 SYNC_FILE_RANGE_WAIT_BEFORE
671 | SYNC_FILE_RANGE_WRITE
672 | SYNC_FILE_RANGE_WAIT_AFTER);
673 /*
674 * Give hints to the kernel about how we access the file:
675 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
676 * we write it.
677 *
678 * We need to call fadvise again after the file grows because the
679 * kernel does not seem to apply fadvise to non-existing parts of the
680 * file.
681 *
682 * Call fadvise _after_ having waited for the page writeback to
683 * complete because the dirty page writeback semantic is not well
684 * defined. So it can be expected to lead to lower throughput in
685 * streaming.
686 */
687 posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
688 stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
689}
690
691/*
692 * Initialise the necessary environnement :
693 * - create a new context
694 * - create the poll_pipe
695 * - create the should_quit pipe (for signal handler)
696 * - create the thread pipe (for splice)
697 *
698 * Takes a function pointer as argument, this function is called when data is
699 * available on a buffer. This function is responsible to do the
700 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
701 * buffer configuration and then kernctl_put_next_subbuf at the end.
702 *
703 * Returns a pointer to the new context or NULL on error.
704 */
705struct lttng_consumer_local_data *lttng_consumer_create(
706 enum lttng_consumer_type type,
4078b776 707 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 708 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
709 int (*recv_channel)(struct lttng_consumer_channel *channel),
710 int (*recv_stream)(struct lttng_consumer_stream *stream),
711 int (*update_stream)(int stream_key, uint32_t state))
712{
713 int ret, i;
714 struct lttng_consumer_local_data *ctx;
715
716 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
717 consumer_data.type == type);
718 consumer_data.type = type;
719
effcf122 720 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
3bd1e081
MD
721 if (ctx == NULL) {
722 perror("allocating context");
723 goto error;
724 }
725
726 ctx->consumer_error_socket = -1;
727 /* assign the callbacks */
728 ctx->on_buffer_ready = buffer_ready;
729 ctx->on_recv_channel = recv_channel;
730 ctx->on_recv_stream = recv_stream;
731 ctx->on_update_stream = update_stream;
732
733 ret = pipe(ctx->consumer_poll_pipe);
734 if (ret < 0) {
735 perror("Error creating poll pipe");
736 goto error_poll_pipe;
737 }
738
739 ret = pipe(ctx->consumer_should_quit);
740 if (ret < 0) {
741 perror("Error creating recv pipe");
742 goto error_quit_pipe;
743 }
744
745 ret = pipe(ctx->consumer_thread_pipe);
746 if (ret < 0) {
747 perror("Error creating thread pipe");
748 goto error_thread_pipe;
749 }
750
751 return ctx;
752
753
754error_thread_pipe:
755 for (i = 0; i < 2; i++) {
756 int err;
757
758 err = close(ctx->consumer_should_quit[i]);
4c462e79
MD
759 if (err) {
760 PERROR("close");
761 }
3bd1e081
MD
762 }
763error_quit_pipe:
764 for (i = 0; i < 2; i++) {
765 int err;
766
767 err = close(ctx->consumer_poll_pipe[i]);
4c462e79
MD
768 if (err) {
769 PERROR("close");
770 }
3bd1e081
MD
771 }
772error_poll_pipe:
773 free(ctx);
774error:
775 return NULL;
776}
777
778/*
779 * Close all fds associated with the instance and free the context.
780 */
781void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
782{
4c462e79
MD
783 int ret;
784
785 ret = close(ctx->consumer_error_socket);
786 if (ret) {
787 PERROR("close");
788 }
789 ret = close(ctx->consumer_thread_pipe[0]);
790 if (ret) {
791 PERROR("close");
792 }
793 ret = close(ctx->consumer_thread_pipe[1]);
794 if (ret) {
795 PERROR("close");
796 }
797 ret = close(ctx->consumer_poll_pipe[0]);
798 if (ret) {
799 PERROR("close");
800 }
801 ret = close(ctx->consumer_poll_pipe[1]);
802 if (ret) {
803 PERROR("close");
804 }
805 ret = close(ctx->consumer_should_quit[0]);
806 if (ret) {
807 PERROR("close");
808 }
809 ret = close(ctx->consumer_should_quit[1]);
810 if (ret) {
811 PERROR("close");
812 }
3bd1e081
MD
813 unlink(ctx->consumer_command_sock_path);
814 free(ctx);
815}
816
817/*
818 * Mmap the ring buffer, read it and write the data to the tracefile.
819 *
820 * Returns the number of bytes written
821 */
4078b776 822ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081
MD
823 struct lttng_consumer_local_data *ctx,
824 struct lttng_consumer_stream *stream, unsigned long len)
825{
826 switch (consumer_data.type) {
827 case LTTNG_CONSUMER_KERNEL:
828 return lttng_kconsumer_on_read_subbuffer_mmap(ctx, stream, len);
7753dea8
MD
829 case LTTNG_CONSUMER32_UST:
830 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
831 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx, stream, len);
832 default:
833 ERR("Unknown consumer_data type");
834 assert(0);
835 }
b9182dd9
DG
836
837 return 0;
3bd1e081
MD
838}
839
840/*
841 * Splice the data from the ring buffer to the tracefile.
842 *
843 * Returns the number of bytes spliced.
844 */
4078b776 845ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081
MD
846 struct lttng_consumer_local_data *ctx,
847 struct lttng_consumer_stream *stream, unsigned long len)
848{
849 switch (consumer_data.type) {
850 case LTTNG_CONSUMER_KERNEL:
851 return lttng_kconsumer_on_read_subbuffer_splice(ctx, stream, len);
7753dea8
MD
852 case LTTNG_CONSUMER32_UST:
853 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
854 return -ENOSYS;
855 default:
856 ERR("Unknown consumer_data type");
857 assert(0);
858 return -ENOSYS;
859 }
860
861}
862
863/*
864 * Take a snapshot for a specific fd
865 *
866 * Returns 0 on success, < 0 on error
867 */
868int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
869 struct lttng_consumer_stream *stream)
870{
871 switch (consumer_data.type) {
872 case LTTNG_CONSUMER_KERNEL:
873 return lttng_kconsumer_take_snapshot(ctx, stream);
7753dea8
MD
874 case LTTNG_CONSUMER32_UST:
875 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
876 return lttng_ustconsumer_take_snapshot(ctx, stream);
877 default:
878 ERR("Unknown consumer_data type");
879 assert(0);
880 return -ENOSYS;
881 }
882
883}
884
885/*
886 * Get the produced position
887 *
888 * Returns 0 on success, < 0 on error
889 */
890int lttng_consumer_get_produced_snapshot(
891 struct lttng_consumer_local_data *ctx,
892 struct lttng_consumer_stream *stream,
893 unsigned long *pos)
894{
895 switch (consumer_data.type) {
896 case LTTNG_CONSUMER_KERNEL:
897 return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
7753dea8
MD
898 case LTTNG_CONSUMER32_UST:
899 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
900 return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
901 default:
902 ERR("Unknown consumer_data type");
903 assert(0);
904 return -ENOSYS;
905 }
906}
907
908int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
909 int sock, struct pollfd *consumer_sockpoll)
910{
911 switch (consumer_data.type) {
912 case LTTNG_CONSUMER_KERNEL:
913 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
914 case LTTNG_CONSUMER32_UST:
915 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
916 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
917 default:
918 ERR("Unknown consumer_data type");
919 assert(0);
920 return -ENOSYS;
921 }
922}
923
924/*
e4421fec 925 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
926 * it to tracefile if necessary.
927 */
928void *lttng_consumer_thread_poll_fds(void *data)
929{
930 int num_rdy, num_hup, high_prio, ret, i;
931 struct pollfd *pollfd = NULL;
932 /* local view of the streams */
933 struct lttng_consumer_stream **local_stream = NULL;
934 /* local view of consumer_data.fds_count */
935 int nb_fd = 0;
936 char tmp;
937 int tmp2;
938 struct lttng_consumer_local_data *ctx = data;
939
e7b994a3
DG
940 rcu_register_thread();
941
effcf122 942 local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
3bd1e081
MD
943
944 while (1) {
945 high_prio = 0;
946 num_hup = 0;
947
948 /*
e4421fec 949 * the fds set has been updated, we need to update our
3bd1e081
MD
950 * local array as well
951 */
952 pthread_mutex_lock(&consumer_data.lock);
953 if (consumer_data.need_update) {
954 if (pollfd != NULL) {
955 free(pollfd);
956 pollfd = NULL;
957 }
958 if (local_stream != NULL) {
959 free(local_stream);
960 local_stream = NULL;
961 }
962
963 /* allocate for all fds + 1 for the consumer_poll_pipe */
effcf122 964 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
3bd1e081
MD
965 if (pollfd == NULL) {
966 perror("pollfd malloc");
967 pthread_mutex_unlock(&consumer_data.lock);
968 goto end;
969 }
970
971 /* allocate for all fds + 1 for the consumer_poll_pipe */
effcf122 972 local_stream = zmalloc((consumer_data.stream_count + 1) *
3bd1e081
MD
973 sizeof(struct lttng_consumer_stream));
974 if (local_stream == NULL) {
975 perror("local_stream malloc");
976 pthread_mutex_unlock(&consumer_data.lock);
977 goto end;
978 }
979 ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
980 if (ret < 0) {
981 ERR("Error in allocating pollfd or local_outfds");
982 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
983 pthread_mutex_unlock(&consumer_data.lock);
984 goto end;
985 }
986 nb_fd = ret;
987 consumer_data.need_update = 0;
988 }
989 pthread_mutex_unlock(&consumer_data.lock);
990
4078b776
MD
991 /* No FDs and consumer_quit, consumer_cleanup the thread */
992 if (nb_fd == 0 && consumer_quit == 1) {
993 goto end;
994 }
3bd1e081 995 /* poll on the array of fds */
88f2b785 996 restart:
3bd1e081
MD
997 DBG("polling on %d fd", nb_fd + 1);
998 num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
999 DBG("poll num_rdy : %d", num_rdy);
1000 if (num_rdy == -1) {
88f2b785
MD
1001 /*
1002 * Restart interrupted system call.
1003 */
1004 if (errno == EINTR) {
1005 goto restart;
1006 }
3bd1e081
MD
1007 perror("Poll error");
1008 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
1009 goto end;
1010 } else if (num_rdy == 0) {
1011 DBG("Polling thread timed out");
1012 goto end;
1013 }
1014
3bd1e081
MD
1015 /*
1016 * If the consumer_poll_pipe triggered poll go
1017 * directly to the beginning of the loop to update the
1018 * array. We want to prioritize array update over
1019 * low-priority reads.
1020 */
509bb1cf 1021 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
3bd1e081
MD
1022 DBG("consumer_poll_pipe wake up");
1023 tmp2 = read(ctx->consumer_poll_pipe[0], &tmp, 1);
1024 if (tmp2 < 0) {
d41f73b7 1025 perror("read consumer poll");
3bd1e081
MD
1026 }
1027 continue;
1028 }
1029
1030 /* Take care of high priority channels first. */
1031 for (i = 0; i < nb_fd; i++) {
d41f73b7 1032 if (pollfd[i].revents & POLLPRI) {
4078b776
MD
1033 ssize_t len;
1034
d41f73b7
MD
1035 DBG("Urgent read on fd %d", pollfd[i].fd);
1036 high_prio = 1;
4078b776 1037 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 1038 /* it's ok to have an unavailable sub-buffer */
4078b776
MD
1039 if (len < 0 && len != -EAGAIN) {
1040 goto end;
1041 } else if (len > 0) {
1042 local_stream[i]->data_read = 1;
d41f73b7 1043 }
3bd1e081
MD
1044 }
1045 }
1046
4078b776
MD
1047 /*
1048 * If we read high prio channel in this loop, try again
1049 * for more high prio data.
1050 */
1051 if (high_prio) {
3bd1e081
MD
1052 continue;
1053 }
1054
1055 /* Take care of low priority channels. */
4078b776
MD
1056 for (i = 0; i < nb_fd; i++) {
1057 if ((pollfd[i].revents & POLLIN) ||
1058 local_stream[i]->hangup_flush_done) {
1059 ssize_t len;
1060
4078b776
MD
1061 DBG("Normal read on fd %d", pollfd[i].fd);
1062 len = ctx->on_buffer_ready(local_stream[i], ctx);
1063 /* it's ok to have an unavailable sub-buffer */
1064 if (len < 0 && len != -EAGAIN) {
1065 goto end;
1066 } else if (len > 0) {
1067 local_stream[i]->data_read = 1;
1068 }
1069 }
1070 }
1071
1072 /* Handle hangup and errors */
1073 for (i = 0; i < nb_fd; i++) {
1074 if (!local_stream[i]->hangup_flush_done
1075 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
1076 && (consumer_data.type == LTTNG_CONSUMER32_UST
1077 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
1078 DBG("fd %d is hup|err|nval. Attempting flush and read.",
1079 pollfd[i].fd);
1080 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
1081 /* Attempt read again, for the data we just flushed. */
1082 local_stream[i]->data_read = 1;
1083 }
1084 /*
1085 * If the poll flag is HUP/ERR/NVAL and we have
1086 * read no data in this pass, we can remove the
1087 * stream from its hash table.
1088 */
1089 if ((pollfd[i].revents & POLLHUP)) {
1090 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
1091 if (!local_stream[i]->data_read) {
702b1ea4 1092 consumer_del_stream(local_stream[i]);
4078b776
MD
1093 num_hup++;
1094 }
1095 } else if (pollfd[i].revents & POLLERR) {
1096 ERR("Error returned in polling fd %d.", pollfd[i].fd);
1097 if (!local_stream[i]->data_read) {
702b1ea4 1098 consumer_del_stream(local_stream[i]);
4078b776
MD
1099 num_hup++;
1100 }
1101 } else if (pollfd[i].revents & POLLNVAL) {
1102 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
1103 if (!local_stream[i]->data_read) {
702b1ea4 1104 consumer_del_stream(local_stream[i]);
4078b776 1105 num_hup++;
3bd1e081
MD
1106 }
1107 }
4078b776 1108 local_stream[i]->data_read = 0;
3bd1e081
MD
1109 }
1110 }
1111end:
1112 DBG("polling thread exiting");
1113 if (pollfd != NULL) {
1114 free(pollfd);
1115 pollfd = NULL;
1116 }
1117 if (local_stream != NULL) {
1118 free(local_stream);
1119 local_stream = NULL;
1120 }
e7b994a3 1121 rcu_unregister_thread();
3bd1e081
MD
1122 return NULL;
1123}
1124
1125/*
1126 * This thread listens on the consumerd socket and receives the file
1127 * descriptors from the session daemon.
1128 */
1129void *lttng_consumer_thread_receive_fds(void *data)
1130{
1131 int sock, client_socket, ret;
1132 /*
1133 * structure to poll for incoming data on communication socket avoids
1134 * making blocking sockets.
1135 */
1136 struct pollfd consumer_sockpoll[2];
1137 struct lttng_consumer_local_data *ctx = data;
1138
e7b994a3
DG
1139 rcu_register_thread();
1140
3bd1e081
MD
1141 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
1142 unlink(ctx->consumer_command_sock_path);
1143 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
1144 if (client_socket < 0) {
1145 ERR("Cannot create command socket");
1146 goto end;
1147 }
1148
1149 ret = lttcomm_listen_unix_sock(client_socket);
1150 if (ret < 0) {
1151 goto end;
1152 }
1153
32258573 1154 DBG("Sending ready command to lttng-sessiond");
3bd1e081
MD
1155 ret = lttng_consumer_send_error(ctx, CONSUMERD_COMMAND_SOCK_READY);
1156 /* return < 0 on error, but == 0 is not fatal */
1157 if (ret < 0) {
32258573 1158 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
1159 goto end;
1160 }
1161
1162 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
1163 if (ret < 0) {
1164 perror("fcntl O_NONBLOCK");
1165 goto end;
1166 }
1167
1168 /* prepare the FDs to poll : to client socket and the should_quit pipe */
1169 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
1170 consumer_sockpoll[0].events = POLLIN | POLLPRI;
1171 consumer_sockpoll[1].fd = client_socket;
1172 consumer_sockpoll[1].events = POLLIN | POLLPRI;
1173
1174 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1175 goto end;
1176 }
1177 DBG("Connection on client_socket");
1178
1179 /* Blocking call, waiting for transmission */
1180 sock = lttcomm_accept_unix_sock(client_socket);
1181 if (sock <= 0) {
1182 WARN("On accept");
1183 goto end;
1184 }
1185 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
1186 if (ret < 0) {
1187 perror("fcntl O_NONBLOCK");
1188 goto end;
1189 }
1190
1191 /* update the polling structure to poll on the established socket */
1192 consumer_sockpoll[1].fd = sock;
1193 consumer_sockpoll[1].events = POLLIN | POLLPRI;
1194
1195 while (1) {
1196 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1197 goto end;
1198 }
1199 DBG("Incoming command on sock");
1200 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
1201 if (ret == -ENOENT) {
1202 DBG("Received STOP command");
1203 goto end;
1204 }
1205 if (ret < 0) {
1206 ERR("Communication interrupted on command socket");
1207 goto end;
1208 }
1209 if (consumer_quit) {
1210 DBG("consumer_thread_receive_fds received quit from signal");
1211 goto end;
1212 }
1213 DBG("received fds on sock");
1214 }
1215end:
1216 DBG("consumer_thread_receive_fds exiting");
1217
1218 /*
1219 * when all fds have hung up, the polling thread
1220 * can exit cleanly
1221 */
1222 consumer_quit = 1;
1223
1224 /*
1225 * 2s of grace period, if no polling events occur during
1226 * this period, the polling thread will exit even if there
1227 * are still open FDs (should not happen, but safety mechanism).
1228 */
1229 consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
1230
1231 /* wake up the polling thread */
1232 ret = write(ctx->consumer_poll_pipe[1], "4", 1);
1233 if (ret < 0) {
1234 perror("poll pipe write");
1235 }
e7b994a3 1236 rcu_unregister_thread();
3bd1e081
MD
1237 return NULL;
1238}
d41f73b7 1239
4078b776 1240ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
1241 struct lttng_consumer_local_data *ctx)
1242{
1243 switch (consumer_data.type) {
1244 case LTTNG_CONSUMER_KERNEL:
1245 return lttng_kconsumer_read_subbuffer(stream, ctx);
7753dea8
MD
1246 case LTTNG_CONSUMER32_UST:
1247 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
1248 return lttng_ustconsumer_read_subbuffer(stream, ctx);
1249 default:
1250 ERR("Unknown consumer_data type");
1251 assert(0);
1252 return -ENOSYS;
1253 }
1254}
1255
1256int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
1257{
1258 switch (consumer_data.type) {
1259 case LTTNG_CONSUMER_KERNEL:
1260 return lttng_kconsumer_on_recv_stream(stream);
7753dea8
MD
1261 case LTTNG_CONSUMER32_UST:
1262 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
1263 return lttng_ustconsumer_on_recv_stream(stream);
1264 default:
1265 ERR("Unknown consumer_data type");
1266 assert(0);
1267 return -ENOSYS;
1268 }
1269}
e4421fec
DG
1270
1271/*
1272 * Allocate and set consumer data hash tables.
1273 */
1274void lttng_consumer_init(void)
1275{
1276 consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1277 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1278}
1279
This page took 0.13947 seconds and 4 git commands to generate.