Initial import of the new binary lttng-relayd
[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();
d6ce1df2
MD
639
640 lttng_ht_destroy(consumer_data.stream_ht);
641 lttng_ht_destroy(consumer_data.channel_ht);
3bd1e081
MD
642}
643
644/*
645 * Called from signal handler.
646 */
647void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
648{
649 int ret;
650 consumer_quit = 1;
651 ret = write(ctx->consumer_should_quit[1], "4", 1);
652 if (ret < 0) {
653 perror("write consumer quit");
654 }
655}
656
657void lttng_consumer_sync_trace_file(
658 struct lttng_consumer_stream *stream, off_t orig_offset)
659{
660 int outfd = stream->out_fd;
661
662 /*
663 * This does a blocking write-and-wait on any page that belongs to the
664 * subbuffer prior to the one we just wrote.
665 * Don't care about error values, as these are just hints and ways to
666 * limit the amount of page cache used.
667 */
668 if (orig_offset < stream->chan->max_sb_size) {
669 return;
670 }
b9182dd9 671 lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
3bd1e081
MD
672 stream->chan->max_sb_size,
673 SYNC_FILE_RANGE_WAIT_BEFORE
674 | SYNC_FILE_RANGE_WRITE
675 | SYNC_FILE_RANGE_WAIT_AFTER);
676 /*
677 * Give hints to the kernel about how we access the file:
678 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
679 * we write it.
680 *
681 * We need to call fadvise again after the file grows because the
682 * kernel does not seem to apply fadvise to non-existing parts of the
683 * file.
684 *
685 * Call fadvise _after_ having waited for the page writeback to
686 * complete because the dirty page writeback semantic is not well
687 * defined. So it can be expected to lead to lower throughput in
688 * streaming.
689 */
690 posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
691 stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
692}
693
694/*
695 * Initialise the necessary environnement :
696 * - create a new context
697 * - create the poll_pipe
698 * - create the should_quit pipe (for signal handler)
699 * - create the thread pipe (for splice)
700 *
701 * Takes a function pointer as argument, this function is called when data is
702 * available on a buffer. This function is responsible to do the
703 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
704 * buffer configuration and then kernctl_put_next_subbuf at the end.
705 *
706 * Returns a pointer to the new context or NULL on error.
707 */
708struct lttng_consumer_local_data *lttng_consumer_create(
709 enum lttng_consumer_type type,
4078b776 710 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 711 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
712 int (*recv_channel)(struct lttng_consumer_channel *channel),
713 int (*recv_stream)(struct lttng_consumer_stream *stream),
714 int (*update_stream)(int stream_key, uint32_t state))
715{
716 int ret, i;
717 struct lttng_consumer_local_data *ctx;
718
719 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
720 consumer_data.type == type);
721 consumer_data.type = type;
722
effcf122 723 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
3bd1e081
MD
724 if (ctx == NULL) {
725 perror("allocating context");
726 goto error;
727 }
728
729 ctx->consumer_error_socket = -1;
730 /* assign the callbacks */
731 ctx->on_buffer_ready = buffer_ready;
732 ctx->on_recv_channel = recv_channel;
733 ctx->on_recv_stream = recv_stream;
734 ctx->on_update_stream = update_stream;
735
736 ret = pipe(ctx->consumer_poll_pipe);
737 if (ret < 0) {
738 perror("Error creating poll pipe");
739 goto error_poll_pipe;
740 }
741
04fdd819
MD
742 /* set read end of the pipe to non-blocking */
743 ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
744 if (ret < 0) {
745 perror("fcntl O_NONBLOCK");
746 goto error_poll_fcntl;
747 }
748
749 /* set write end of the pipe to non-blocking */
750 ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
751 if (ret < 0) {
752 perror("fcntl O_NONBLOCK");
753 goto error_poll_fcntl;
754 }
755
3bd1e081
MD
756 ret = pipe(ctx->consumer_should_quit);
757 if (ret < 0) {
758 perror("Error creating recv pipe");
759 goto error_quit_pipe;
760 }
761
762 ret = pipe(ctx->consumer_thread_pipe);
763 if (ret < 0) {
764 perror("Error creating thread pipe");
765 goto error_thread_pipe;
766 }
767
768 return ctx;
769
770
771error_thread_pipe:
772 for (i = 0; i < 2; i++) {
773 int err;
774
775 err = close(ctx->consumer_should_quit[i]);
4c462e79
MD
776 if (err) {
777 PERROR("close");
778 }
3bd1e081 779 }
04fdd819 780error_poll_fcntl:
3bd1e081
MD
781error_quit_pipe:
782 for (i = 0; i < 2; i++) {
783 int err;
784
785 err = close(ctx->consumer_poll_pipe[i]);
4c462e79
MD
786 if (err) {
787 PERROR("close");
788 }
3bd1e081
MD
789 }
790error_poll_pipe:
791 free(ctx);
792error:
793 return NULL;
794}
795
796/*
797 * Close all fds associated with the instance and free the context.
798 */
799void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
800{
4c462e79
MD
801 int ret;
802
803 ret = close(ctx->consumer_error_socket);
804 if (ret) {
805 PERROR("close");
806 }
807 ret = close(ctx->consumer_thread_pipe[0]);
808 if (ret) {
809 PERROR("close");
810 }
811 ret = close(ctx->consumer_thread_pipe[1]);
812 if (ret) {
813 PERROR("close");
814 }
815 ret = close(ctx->consumer_poll_pipe[0]);
816 if (ret) {
817 PERROR("close");
818 }
819 ret = close(ctx->consumer_poll_pipe[1]);
820 if (ret) {
821 PERROR("close");
822 }
823 ret = close(ctx->consumer_should_quit[0]);
824 if (ret) {
825 PERROR("close");
826 }
827 ret = close(ctx->consumer_should_quit[1]);
828 if (ret) {
829 PERROR("close");
830 }
3bd1e081
MD
831 unlink(ctx->consumer_command_sock_path);
832 free(ctx);
833}
834
835/*
836 * Mmap the ring buffer, read it and write the data to the tracefile.
837 *
838 * Returns the number of bytes written
839 */
4078b776 840ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081
MD
841 struct lttng_consumer_local_data *ctx,
842 struct lttng_consumer_stream *stream, unsigned long len)
843{
844 switch (consumer_data.type) {
845 case LTTNG_CONSUMER_KERNEL:
846 return lttng_kconsumer_on_read_subbuffer_mmap(ctx, stream, len);
7753dea8
MD
847 case LTTNG_CONSUMER32_UST:
848 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
849 return lttng_ustconsumer_on_read_subbuffer_mmap(ctx, stream, len);
850 default:
851 ERR("Unknown consumer_data type");
852 assert(0);
853 }
b9182dd9
DG
854
855 return 0;
3bd1e081
MD
856}
857
858/*
859 * Splice the data from the ring buffer to the tracefile.
860 *
861 * Returns the number of bytes spliced.
862 */
4078b776 863ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081
MD
864 struct lttng_consumer_local_data *ctx,
865 struct lttng_consumer_stream *stream, unsigned long len)
866{
867 switch (consumer_data.type) {
868 case LTTNG_CONSUMER_KERNEL:
869 return lttng_kconsumer_on_read_subbuffer_splice(ctx, stream, len);
7753dea8
MD
870 case LTTNG_CONSUMER32_UST:
871 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
872 return -ENOSYS;
873 default:
874 ERR("Unknown consumer_data type");
875 assert(0);
876 return -ENOSYS;
877 }
878
879}
880
881/*
882 * Take a snapshot for a specific fd
883 *
884 * Returns 0 on success, < 0 on error
885 */
886int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
887 struct lttng_consumer_stream *stream)
888{
889 switch (consumer_data.type) {
890 case LTTNG_CONSUMER_KERNEL:
891 return lttng_kconsumer_take_snapshot(ctx, stream);
7753dea8
MD
892 case LTTNG_CONSUMER32_UST:
893 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
894 return lttng_ustconsumer_take_snapshot(ctx, stream);
895 default:
896 ERR("Unknown consumer_data type");
897 assert(0);
898 return -ENOSYS;
899 }
900
901}
902
903/*
904 * Get the produced position
905 *
906 * Returns 0 on success, < 0 on error
907 */
908int lttng_consumer_get_produced_snapshot(
909 struct lttng_consumer_local_data *ctx,
910 struct lttng_consumer_stream *stream,
911 unsigned long *pos)
912{
913 switch (consumer_data.type) {
914 case LTTNG_CONSUMER_KERNEL:
915 return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
7753dea8
MD
916 case LTTNG_CONSUMER32_UST:
917 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
918 return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
919 default:
920 ERR("Unknown consumer_data type");
921 assert(0);
922 return -ENOSYS;
923 }
924}
925
926int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
927 int sock, struct pollfd *consumer_sockpoll)
928{
929 switch (consumer_data.type) {
930 case LTTNG_CONSUMER_KERNEL:
931 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
932 case LTTNG_CONSUMER32_UST:
933 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
934 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
935 default:
936 ERR("Unknown consumer_data type");
937 assert(0);
938 return -ENOSYS;
939 }
940}
941
942/*
e4421fec 943 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
944 * it to tracefile if necessary.
945 */
946void *lttng_consumer_thread_poll_fds(void *data)
947{
948 int num_rdy, num_hup, high_prio, ret, i;
949 struct pollfd *pollfd = NULL;
950 /* local view of the streams */
951 struct lttng_consumer_stream **local_stream = NULL;
952 /* local view of consumer_data.fds_count */
953 int nb_fd = 0;
3bd1e081
MD
954 struct lttng_consumer_local_data *ctx = data;
955
e7b994a3
DG
956 rcu_register_thread();
957
effcf122 958 local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
3bd1e081
MD
959
960 while (1) {
961 high_prio = 0;
962 num_hup = 0;
963
964 /*
e4421fec 965 * the fds set has been updated, we need to update our
3bd1e081
MD
966 * local array as well
967 */
968 pthread_mutex_lock(&consumer_data.lock);
969 if (consumer_data.need_update) {
970 if (pollfd != NULL) {
971 free(pollfd);
972 pollfd = NULL;
973 }
974 if (local_stream != NULL) {
975 free(local_stream);
976 local_stream = NULL;
977 }
978
979 /* allocate for all fds + 1 for the consumer_poll_pipe */
effcf122 980 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
3bd1e081
MD
981 if (pollfd == NULL) {
982 perror("pollfd malloc");
983 pthread_mutex_unlock(&consumer_data.lock);
984 goto end;
985 }
986
987 /* allocate for all fds + 1 for the consumer_poll_pipe */
effcf122 988 local_stream = zmalloc((consumer_data.stream_count + 1) *
3bd1e081
MD
989 sizeof(struct lttng_consumer_stream));
990 if (local_stream == NULL) {
991 perror("local_stream malloc");
992 pthread_mutex_unlock(&consumer_data.lock);
993 goto end;
994 }
995 ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
996 if (ret < 0) {
997 ERR("Error in allocating pollfd or local_outfds");
998 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
999 pthread_mutex_unlock(&consumer_data.lock);
1000 goto end;
1001 }
1002 nb_fd = ret;
1003 consumer_data.need_update = 0;
1004 }
1005 pthread_mutex_unlock(&consumer_data.lock);
1006
4078b776
MD
1007 /* No FDs and consumer_quit, consumer_cleanup the thread */
1008 if (nb_fd == 0 && consumer_quit == 1) {
1009 goto end;
1010 }
3bd1e081 1011 /* poll on the array of fds */
88f2b785 1012 restart:
3bd1e081
MD
1013 DBG("polling on %d fd", nb_fd + 1);
1014 num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
1015 DBG("poll num_rdy : %d", num_rdy);
1016 if (num_rdy == -1) {
88f2b785
MD
1017 /*
1018 * Restart interrupted system call.
1019 */
1020 if (errno == EINTR) {
1021 goto restart;
1022 }
3bd1e081
MD
1023 perror("Poll error");
1024 lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
1025 goto end;
1026 } else if (num_rdy == 0) {
1027 DBG("Polling thread timed out");
1028 goto end;
1029 }
1030
3bd1e081
MD
1031 /*
1032 * If the consumer_poll_pipe triggered poll go
1033 * directly to the beginning of the loop to update the
1034 * array. We want to prioritize array update over
1035 * low-priority reads.
1036 */
509bb1cf 1037 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
04fdd819
MD
1038 size_t pipe_readlen;
1039 char tmp;
1040
3bd1e081 1041 DBG("consumer_poll_pipe wake up");
04fdd819
MD
1042 /* Consume 1 byte of pipe data */
1043 do {
1044 pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1);
1045 } while (pipe_readlen == -1 && errno == EINTR);
3bd1e081
MD
1046 continue;
1047 }
1048
1049 /* Take care of high priority channels first. */
1050 for (i = 0; i < nb_fd; i++) {
d41f73b7 1051 if (pollfd[i].revents & POLLPRI) {
4078b776
MD
1052 ssize_t len;
1053
d41f73b7
MD
1054 DBG("Urgent read on fd %d", pollfd[i].fd);
1055 high_prio = 1;
4078b776 1056 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 1057 /* it's ok to have an unavailable sub-buffer */
4078b776
MD
1058 if (len < 0 && len != -EAGAIN) {
1059 goto end;
1060 } else if (len > 0) {
1061 local_stream[i]->data_read = 1;
d41f73b7 1062 }
3bd1e081
MD
1063 }
1064 }
1065
4078b776
MD
1066 /*
1067 * If we read high prio channel in this loop, try again
1068 * for more high prio data.
1069 */
1070 if (high_prio) {
3bd1e081
MD
1071 continue;
1072 }
1073
1074 /* Take care of low priority channels. */
4078b776
MD
1075 for (i = 0; i < nb_fd; i++) {
1076 if ((pollfd[i].revents & POLLIN) ||
1077 local_stream[i]->hangup_flush_done) {
1078 ssize_t len;
1079
4078b776
MD
1080 DBG("Normal read on fd %d", pollfd[i].fd);
1081 len = ctx->on_buffer_ready(local_stream[i], ctx);
1082 /* it's ok to have an unavailable sub-buffer */
1083 if (len < 0 && len != -EAGAIN) {
1084 goto end;
1085 } else if (len > 0) {
1086 local_stream[i]->data_read = 1;
1087 }
1088 }
1089 }
1090
1091 /* Handle hangup and errors */
1092 for (i = 0; i < nb_fd; i++) {
1093 if (!local_stream[i]->hangup_flush_done
1094 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
1095 && (consumer_data.type == LTTNG_CONSUMER32_UST
1096 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
1097 DBG("fd %d is hup|err|nval. Attempting flush and read.",
1098 pollfd[i].fd);
1099 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
1100 /* Attempt read again, for the data we just flushed. */
1101 local_stream[i]->data_read = 1;
1102 }
1103 /*
1104 * If the poll flag is HUP/ERR/NVAL and we have
1105 * read no data in this pass, we can remove the
1106 * stream from its hash table.
1107 */
1108 if ((pollfd[i].revents & POLLHUP)) {
1109 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
1110 if (!local_stream[i]->data_read) {
702b1ea4 1111 consumer_del_stream(local_stream[i]);
4078b776
MD
1112 num_hup++;
1113 }
1114 } else if (pollfd[i].revents & POLLERR) {
1115 ERR("Error returned in polling fd %d.", pollfd[i].fd);
1116 if (!local_stream[i]->data_read) {
702b1ea4 1117 consumer_del_stream(local_stream[i]);
4078b776
MD
1118 num_hup++;
1119 }
1120 } else if (pollfd[i].revents & POLLNVAL) {
1121 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
1122 if (!local_stream[i]->data_read) {
702b1ea4 1123 consumer_del_stream(local_stream[i]);
4078b776 1124 num_hup++;
3bd1e081
MD
1125 }
1126 }
4078b776 1127 local_stream[i]->data_read = 0;
3bd1e081
MD
1128 }
1129 }
1130end:
1131 DBG("polling thread exiting");
1132 if (pollfd != NULL) {
1133 free(pollfd);
1134 pollfd = NULL;
1135 }
1136 if (local_stream != NULL) {
1137 free(local_stream);
1138 local_stream = NULL;
1139 }
e7b994a3 1140 rcu_unregister_thread();
3bd1e081
MD
1141 return NULL;
1142}
1143
1144/*
1145 * This thread listens on the consumerd socket and receives the file
1146 * descriptors from the session daemon.
1147 */
1148void *lttng_consumer_thread_receive_fds(void *data)
1149{
1150 int sock, client_socket, ret;
1151 /*
1152 * structure to poll for incoming data on communication socket avoids
1153 * making blocking sockets.
1154 */
1155 struct pollfd consumer_sockpoll[2];
1156 struct lttng_consumer_local_data *ctx = data;
1157
e7b994a3
DG
1158 rcu_register_thread();
1159
3bd1e081
MD
1160 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
1161 unlink(ctx->consumer_command_sock_path);
1162 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
1163 if (client_socket < 0) {
1164 ERR("Cannot create command socket");
1165 goto end;
1166 }
1167
1168 ret = lttcomm_listen_unix_sock(client_socket);
1169 if (ret < 0) {
1170 goto end;
1171 }
1172
32258573 1173 DBG("Sending ready command to lttng-sessiond");
3bd1e081
MD
1174 ret = lttng_consumer_send_error(ctx, CONSUMERD_COMMAND_SOCK_READY);
1175 /* return < 0 on error, but == 0 is not fatal */
1176 if (ret < 0) {
32258573 1177 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
1178 goto end;
1179 }
1180
1181 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
1182 if (ret < 0) {
1183 perror("fcntl O_NONBLOCK");
1184 goto end;
1185 }
1186
1187 /* prepare the FDs to poll : to client socket and the should_quit pipe */
1188 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
1189 consumer_sockpoll[0].events = POLLIN | POLLPRI;
1190 consumer_sockpoll[1].fd = client_socket;
1191 consumer_sockpoll[1].events = POLLIN | POLLPRI;
1192
1193 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1194 goto end;
1195 }
1196 DBG("Connection on client_socket");
1197
1198 /* Blocking call, waiting for transmission */
1199 sock = lttcomm_accept_unix_sock(client_socket);
1200 if (sock <= 0) {
1201 WARN("On accept");
1202 goto end;
1203 }
1204 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
1205 if (ret < 0) {
1206 perror("fcntl O_NONBLOCK");
1207 goto end;
1208 }
1209
1210 /* update the polling structure to poll on the established socket */
1211 consumer_sockpoll[1].fd = sock;
1212 consumer_sockpoll[1].events = POLLIN | POLLPRI;
1213
1214 while (1) {
1215 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
1216 goto end;
1217 }
1218 DBG("Incoming command on sock");
1219 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
1220 if (ret == -ENOENT) {
1221 DBG("Received STOP command");
1222 goto end;
1223 }
1224 if (ret < 0) {
1225 ERR("Communication interrupted on command socket");
1226 goto end;
1227 }
1228 if (consumer_quit) {
1229 DBG("consumer_thread_receive_fds received quit from signal");
1230 goto end;
1231 }
1232 DBG("received fds on sock");
1233 }
1234end:
1235 DBG("consumer_thread_receive_fds exiting");
1236
1237 /*
1238 * when all fds have hung up, the polling thread
1239 * can exit cleanly
1240 */
1241 consumer_quit = 1;
1242
1243 /*
1244 * 2s of grace period, if no polling events occur during
1245 * this period, the polling thread will exit even if there
1246 * are still open FDs (should not happen, but safety mechanism).
1247 */
1248 consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
1249
04fdd819
MD
1250 /*
1251 * Wake-up the other end by writing a null byte in the pipe
1252 * (non-blocking). Important note: Because writing into the
1253 * pipe is non-blocking (and therefore we allow dropping wakeup
1254 * data, as long as there is wakeup data present in the pipe
1255 * buffer to wake up the other end), the other end should
1256 * perform the following sequence for waiting:
1257 * 1) empty the pipe (reads).
1258 * 2) perform update operation.
1259 * 3) wait on the pipe (poll).
1260 */
1261 do {
1262 ret = write(ctx->consumer_poll_pipe[1], "", 1);
1263 } while (ret == -1UL && errno == EINTR);
e7b994a3 1264 rcu_unregister_thread();
3bd1e081
MD
1265 return NULL;
1266}
d41f73b7 1267
4078b776 1268ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
1269 struct lttng_consumer_local_data *ctx)
1270{
1271 switch (consumer_data.type) {
1272 case LTTNG_CONSUMER_KERNEL:
1273 return lttng_kconsumer_read_subbuffer(stream, ctx);
7753dea8
MD
1274 case LTTNG_CONSUMER32_UST:
1275 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
1276 return lttng_ustconsumer_read_subbuffer(stream, ctx);
1277 default:
1278 ERR("Unknown consumer_data type");
1279 assert(0);
1280 return -ENOSYS;
1281 }
1282}
1283
1284int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
1285{
1286 switch (consumer_data.type) {
1287 case LTTNG_CONSUMER_KERNEL:
1288 return lttng_kconsumer_on_recv_stream(stream);
7753dea8
MD
1289 case LTTNG_CONSUMER32_UST:
1290 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
1291 return lttng_ustconsumer_on_recv_stream(stream);
1292 default:
1293 ERR("Unknown consumer_data type");
1294 assert(0);
1295 return -ENOSYS;
1296 }
1297}
e4421fec
DG
1298
1299/*
1300 * Allocate and set consumer data hash tables.
1301 */
1302void lttng_consumer_init(void)
1303{
1304 consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1305 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1306}
1307
This page took 0.142449 seconds and 4 git commands to generate.