Fix: output number of bytes written by 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>
00e2e675 4 * 2012 - David Goulet <dgoulet@efficios.com>
3bd1e081 5 *
d14d33bf
AM
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
3bd1e081 9 *
d14d33bf
AM
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * more details.
3bd1e081 14 *
d14d33bf
AM
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
18 */
19
20#define _GNU_SOURCE
21#include <assert.h>
3bd1e081
MD
22#include <poll.h>
23#include <pthread.h>
24#include <stdlib.h>
25#include <string.h>
26#include <sys/mman.h>
27#include <sys/socket.h>
28#include <sys/types.h>
29#include <unistd.h>
77c7c900 30#include <inttypes.h>
3bd1e081 31
990570ed 32#include <common/common.h>
fb3a43a9
DG
33#include <common/utils.h>
34#include <common/compat/poll.h>
10a8a223 35#include <common/kernel-ctl/kernel-ctl.h>
00e2e675 36#include <common/sessiond-comm/relayd.h>
10a8a223
DG
37#include <common/sessiond-comm/sessiond-comm.h>
38#include <common/kernel-consumer/kernel-consumer.h>
00e2e675 39#include <common/relayd/relayd.h>
10a8a223
DG
40#include <common/ust-consumer/ust-consumer.h>
41
42#include "consumer.h"
3bd1e081
MD
43
44struct lttng_consumer_global_data consumer_data = {
3bd1e081
MD
45 .stream_count = 0,
46 .need_update = 1,
47 .type = LTTNG_CONSUMER_UNKNOWN,
48};
49
50/* timeout parameter, to control the polling thread grace period. */
51int consumer_poll_timeout = -1;
52
53/*
54 * Flag to inform the polling thread to quit when all fd hung up. Updated by
55 * the consumer_thread_receive_fds when it notices that all fds has hung up.
56 * Also updated by the signal handler (consumer_should_exit()). Read by the
57 * polling threads.
58 */
59volatile int consumer_quit = 0;
60
61/*
62 * Find a stream. The consumer_data.lock must be locked during this
63 * call.
64 */
8389e4f8
DG
65static struct lttng_consumer_stream *consumer_find_stream(int key,
66 struct lttng_ht *ht)
3bd1e081 67{
e4421fec
DG
68 struct lttng_ht_iter iter;
69 struct lttng_ht_node_ulong *node;
70 struct lttng_consumer_stream *stream = NULL;
3bd1e081 71
8389e4f8
DG
72 assert(ht);
73
7ad0a0cb 74 /* Negative keys are lookup failures */
7a57cf92 75 if (key < 0) {
7ad0a0cb 76 return NULL;
7a57cf92 77 }
e4421fec 78
6065ceec
DG
79 rcu_read_lock();
80
8389e4f8 81 lttng_ht_lookup(ht, (void *)((unsigned long) key), &iter);
e4421fec
DG
82 node = lttng_ht_iter_get_node_ulong(&iter);
83 if (node != NULL) {
84 stream = caa_container_of(node, struct lttng_consumer_stream, node);
3bd1e081 85 }
e4421fec 86
6065ceec
DG
87 rcu_read_unlock();
88
e4421fec 89 return stream;
3bd1e081
MD
90}
91
8389e4f8 92static void consumer_steal_stream_key(int key, struct lttng_ht *ht)
7ad0a0cb
MD
93{
94 struct lttng_consumer_stream *stream;
95
04253271 96 rcu_read_lock();
8389e4f8 97 stream = consumer_find_stream(key, ht);
04253271 98 if (stream) {
7ad0a0cb 99 stream->key = -1;
04253271
MD
100 /*
101 * We don't want the lookup to match, but we still need
102 * to iterate on this stream when iterating over the hash table. Just
103 * change the node key.
104 */
105 stream->node.key = -1;
106 }
107 rcu_read_unlock();
7ad0a0cb
MD
108}
109
3bd1e081
MD
110static struct lttng_consumer_channel *consumer_find_channel(int key)
111{
e4421fec
DG
112 struct lttng_ht_iter iter;
113 struct lttng_ht_node_ulong *node;
114 struct lttng_consumer_channel *channel = NULL;
3bd1e081 115
7ad0a0cb 116 /* Negative keys are lookup failures */
7a57cf92 117 if (key < 0) {
7ad0a0cb 118 return NULL;
7a57cf92 119 }
e4421fec 120
6065ceec
DG
121 rcu_read_lock();
122
e4421fec
DG
123 lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key),
124 &iter);
125 node = lttng_ht_iter_get_node_ulong(&iter);
126 if (node != NULL) {
127 channel = caa_container_of(node, struct lttng_consumer_channel, node);
3bd1e081 128 }
e4421fec 129
6065ceec
DG
130 rcu_read_unlock();
131
e4421fec 132 return channel;
3bd1e081
MD
133}
134
7ad0a0cb
MD
135static void consumer_steal_channel_key(int key)
136{
137 struct lttng_consumer_channel *channel;
138
04253271 139 rcu_read_lock();
7ad0a0cb 140 channel = consumer_find_channel(key);
04253271 141 if (channel) {
7ad0a0cb 142 channel->key = -1;
04253271
MD
143 /*
144 * We don't want the lookup to match, but we still need
145 * to iterate on this channel when iterating over the hash table. Just
146 * change the node key.
147 */
148 channel->node.key = -1;
149 }
150 rcu_read_unlock();
7ad0a0cb
MD
151}
152
702b1ea4
MD
153static
154void consumer_free_stream(struct rcu_head *head)
155{
156 struct lttng_ht_node_ulong *node =
157 caa_container_of(head, struct lttng_ht_node_ulong, head);
158 struct lttng_consumer_stream *stream =
159 caa_container_of(node, struct lttng_consumer_stream, node);
160
161 free(stream);
162}
163
00e2e675
DG
164/*
165 * RCU protected relayd socket pair free.
166 */
167static void consumer_rcu_free_relayd(struct rcu_head *head)
168{
169 struct lttng_ht_node_ulong *node =
170 caa_container_of(head, struct lttng_ht_node_ulong, head);
171 struct consumer_relayd_sock_pair *relayd =
172 caa_container_of(node, struct consumer_relayd_sock_pair, node);
173
174 free(relayd);
175}
176
177/*
178 * Destroy and free relayd socket pair object.
179 *
180 * This function MUST be called with the consumer_data lock acquired.
181 */
d09e1200 182static void destroy_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
183{
184 int ret;
185 struct lttng_ht_iter iter;
186
173af62f
DG
187 if (relayd == NULL) {
188 return;
189 }
190
00e2e675
DG
191 DBG("Consumer destroy and close relayd socket pair");
192
193 iter.iter.node = &relayd->node.node;
194 ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
173af62f
DG
195 if (ret != 0) {
196 /* We assume the relayd was already destroyed */
197 return;
198 }
00e2e675
DG
199
200 /* Close all sockets */
201 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
202 (void) relayd_close(&relayd->control_sock);
203 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
204 (void) relayd_close(&relayd->data_sock);
205
206 /* RCU free() call */
207 call_rcu(&relayd->node.head, consumer_rcu_free_relayd);
208}
209
a6ba4fe1
DG
210/*
211 * Flag a relayd socket pair for destruction. Destroy it if the refcount
212 * reaches zero.
213 *
214 * RCU read side lock MUST be aquired before calling this function.
215 */
216void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
217{
218 assert(relayd);
219
220 /* Set destroy flag for this object */
221 uatomic_set(&relayd->destroy_flag, 1);
222
223 /* Destroy the relayd if refcount is 0 */
224 if (uatomic_read(&relayd->refcount) == 0) {
d09e1200 225 destroy_relayd(relayd);
a6ba4fe1
DG
226 }
227}
228
3bd1e081
MD
229/*
230 * Remove a stream from the global list protected by a mutex. This
231 * function is also responsible for freeing its data structures.
232 */
233void consumer_del_stream(struct lttng_consumer_stream *stream)
234{
235 int ret;
e4421fec 236 struct lttng_ht_iter iter;
3bd1e081 237 struct lttng_consumer_channel *free_chan = NULL;
00e2e675
DG
238 struct consumer_relayd_sock_pair *relayd;
239
240 assert(stream);
3bd1e081
MD
241
242 pthread_mutex_lock(&consumer_data.lock);
243
244 switch (consumer_data.type) {
245 case LTTNG_CONSUMER_KERNEL:
246 if (stream->mmap_base != NULL) {
247 ret = munmap(stream->mmap_base, stream->mmap_len);
248 if (ret != 0) {
7a57cf92 249 PERROR("munmap");
3bd1e081
MD
250 }
251 }
252 break;
7753dea8
MD
253 case LTTNG_CONSUMER32_UST:
254 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
255 lttng_ustconsumer_del_stream(stream);
256 break;
257 default:
258 ERR("Unknown consumer_data type");
259 assert(0);
260 goto end;
261 }
262
6065ceec 263 rcu_read_lock();
04253271
MD
264 iter.iter.node = &stream->node.node;
265 ret = lttng_ht_del(consumer_data.stream_ht, &iter);
266 assert(!ret);
e4421fec 267
6065ceec
DG
268 rcu_read_unlock();
269
3bd1e081
MD
270 if (consumer_data.stream_count <= 0) {
271 goto end;
272 }
273 consumer_data.stream_count--;
274 if (!stream) {
275 goto end;
276 }
277 if (stream->out_fd >= 0) {
4c462e79
MD
278 ret = close(stream->out_fd);
279 if (ret) {
280 PERROR("close");
281 }
3bd1e081 282 }
b5c5fc29 283 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
4c462e79
MD
284 ret = close(stream->wait_fd);
285 if (ret) {
286 PERROR("close");
287 }
3bd1e081 288 }
2c1dd183 289 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
4c462e79
MD
290 ret = close(stream->shm_fd);
291 if (ret) {
292 PERROR("close");
293 }
3bd1e081 294 }
00e2e675
DG
295
296 /* Check and cleanup relayd */
b0b335c8 297 rcu_read_lock();
00e2e675
DG
298 relayd = consumer_find_relayd(stream->net_seq_idx);
299 if (relayd != NULL) {
b0b335c8
MD
300 uatomic_dec(&relayd->refcount);
301 assert(uatomic_read(&relayd->refcount) >= 0);
173af62f 302
3f8e211f
DG
303 /* Closing streams requires to lock the control socket. */
304 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
173af62f
DG
305 ret = relayd_send_close_stream(&relayd->control_sock,
306 stream->relayd_stream_id,
307 stream->next_net_seq_num - 1);
3f8e211f 308 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
173af62f 309 if (ret < 0) {
a4b92340
DG
310 DBG("Unable to close stream on the relayd. Continuing");
311 /*
312 * Continue here. There is nothing we can do for the relayd.
313 * Chances are that the relayd has closed the socket so we just
314 * continue cleaning up.
315 */
173af62f
DG
316 }
317
318 /* Both conditions are met, we destroy the relayd. */
319 if (uatomic_read(&relayd->refcount) == 0 &&
320 uatomic_read(&relayd->destroy_flag)) {
d09e1200 321 destroy_relayd(relayd);
00e2e675 322 }
00e2e675 323 }
b0b335c8 324 rcu_read_unlock();
00e2e675 325
c30aaa51
MD
326 uatomic_dec(&stream->chan->refcount);
327 if (!uatomic_read(&stream->chan->refcount)
328 && !uatomic_read(&stream->chan->nb_init_streams)) {
3bd1e081 329 free_chan = stream->chan;
00e2e675
DG
330 }
331
702b1ea4 332 call_rcu(&stream->node.head, consumer_free_stream);
3bd1e081
MD
333end:
334 consumer_data.need_update = 1;
335 pthread_mutex_unlock(&consumer_data.lock);
336
c30aaa51 337 if (free_chan) {
3bd1e081 338 consumer_del_channel(free_chan);
c30aaa51 339 }
3bd1e081
MD
340}
341
342struct lttng_consumer_stream *consumer_allocate_stream(
343 int channel_key, int stream_key,
344 int shm_fd, int wait_fd,
345 enum lttng_consumer_stream_state state,
346 uint64_t mmap_len,
347 enum lttng_event_output output,
6df2e2c9
MD
348 const char *path_name,
349 uid_t uid,
00e2e675
DG
350 gid_t gid,
351 int net_index,
c80048c6
MD
352 int metadata_flag,
353 int *alloc_ret)
3bd1e081
MD
354{
355 struct lttng_consumer_stream *stream;
356 int ret;
357
effcf122 358 stream = zmalloc(sizeof(*stream));
3bd1e081 359 if (stream == NULL) {
7a57cf92 360 PERROR("malloc struct lttng_consumer_stream");
c80048c6 361 *alloc_ret = -ENOMEM;
7a57cf92 362 goto end;
3bd1e081 363 }
7a57cf92
DG
364
365 /*
366 * Get stream's channel reference. Needed when adding the stream to the
367 * global hash table.
368 */
3bd1e081
MD
369 stream->chan = consumer_find_channel(channel_key);
370 if (!stream->chan) {
c80048c6 371 *alloc_ret = -ENOENT;
7a57cf92 372 ERR("Unable to find channel for stream %d", stream_key);
c80048c6 373 goto error;
3bd1e081
MD
374 }
375 stream->chan->refcount++;
376 stream->key = stream_key;
377 stream->shm_fd = shm_fd;
378 stream->wait_fd = wait_fd;
379 stream->out_fd = -1;
380 stream->out_fd_offset = 0;
381 stream->state = state;
382 stream->mmap_len = mmap_len;
383 stream->mmap_base = NULL;
384 stream->output = output;
6df2e2c9
MD
385 stream->uid = uid;
386 stream->gid = gid;
00e2e675
DG
387 stream->net_seq_idx = net_index;
388 stream->metadata_flag = metadata_flag;
389 strncpy(stream->path_name, path_name, sizeof(stream->path_name));
390 stream->path_name[sizeof(stream->path_name) - 1] = '\0';
e4421fec 391 lttng_ht_node_init_ulong(&stream->node, stream->key);
00e2e675 392 lttng_ht_node_init_ulong(&stream->waitfd_node, stream->wait_fd);
3bd1e081
MD
393
394 switch (consumer_data.type) {
395 case LTTNG_CONSUMER_KERNEL:
396 break;
7753dea8
MD
397 case LTTNG_CONSUMER32_UST:
398 case LTTNG_CONSUMER64_UST:
5af2f756 399 stream->cpu = stream->chan->cpucount++;
3bd1e081
MD
400 ret = lttng_ustconsumer_allocate_stream(stream);
401 if (ret) {
c80048c6
MD
402 *alloc_ret = -EINVAL;
403 goto error;
3bd1e081
MD
404 }
405 break;
406 default:
407 ERR("Unknown consumer_data type");
c80048c6
MD
408 *alloc_ret = -EINVAL;
409 goto error;
3bd1e081 410 }
c30aaa51
MD
411
412 /*
413 * When nb_init_streams reaches 0, we don't need to trigger any action in
414 * terms of destroying the associated channel, because the action that
415 * causes the count to become 0 also causes a stream to be added. The
416 * channel deletion will thus be triggered by the following removal of this
417 * stream.
418 */
419 if (uatomic_read(&stream->chan->nb_init_streams) > 0) {
420 uatomic_dec(&stream->chan->nb_init_streams);
421 }
422
423 DBG3("Allocated stream %s (key %d, shm_fd %d, wait_fd %d, mmap_len %llu,"
424 " out_fd %d, net_seq_idx %d)", stream->path_name, stream->key,
425 stream->shm_fd, stream->wait_fd,
426 (unsigned long long) stream->mmap_len, stream->out_fd,
00e2e675 427 stream->net_seq_idx);
3bd1e081 428 return stream;
c80048c6
MD
429
430error:
431 free(stream);
7a57cf92 432end:
c80048c6 433 return NULL;
3bd1e081
MD
434}
435
436/*
437 * Add a stream to the global list protected by a mutex.
438 */
439int consumer_add_stream(struct lttng_consumer_stream *stream)
440{
441 int ret = 0;
c77fc10a
DG
442 struct lttng_ht_node_ulong *node;
443 struct lttng_ht_iter iter;
00e2e675 444 struct consumer_relayd_sock_pair *relayd;
3bd1e081
MD
445
446 pthread_mutex_lock(&consumer_data.lock);
7ad0a0cb 447 /* Steal stream identifier, for UST */
8389e4f8 448 consumer_steal_stream_key(stream->key, consumer_data.stream_ht);
c77fc10a 449
b0b335c8 450 rcu_read_lock();
c77fc10a
DG
451 lttng_ht_lookup(consumer_data.stream_ht,
452 (void *)((unsigned long) stream->key), &iter);
453 node = lttng_ht_iter_get_node_ulong(&iter);
454 if (node != NULL) {
455 rcu_read_unlock();
456 /* Stream already exist. Ignore the insertion */
457 goto end;
458 }
459
04253271 460 lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node);
00e2e675
DG
461
462 /* Check and cleanup relayd */
463 relayd = consumer_find_relayd(stream->net_seq_idx);
464 if (relayd != NULL) {
b0b335c8 465 uatomic_inc(&relayd->refcount);
00e2e675 466 }
b0b335c8 467 rcu_read_unlock();
00e2e675
DG
468
469 /* Update consumer data */
3bd1e081
MD
470 consumer_data.stream_count++;
471 consumer_data.need_update = 1;
472
3bd1e081
MD
473end:
474 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 475
3bd1e081
MD
476 return ret;
477}
478
00e2e675 479/*
3f8e211f
DG
480 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
481 * be acquired before calling this.
00e2e675 482 */
d09e1200 483static int add_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
484{
485 int ret = 0;
486 struct lttng_ht_node_ulong *node;
487 struct lttng_ht_iter iter;
488
489 if (relayd == NULL) {
490 ret = -1;
491 goto end;
492 }
493
00e2e675
DG
494 lttng_ht_lookup(consumer_data.relayd_ht,
495 (void *)((unsigned long) relayd->net_seq_idx), &iter);
496 node = lttng_ht_iter_get_node_ulong(&iter);
497 if (node != NULL) {
00e2e675
DG
498 /* Relayd already exist. Ignore the insertion */
499 goto end;
500 }
501 lttng_ht_add_unique_ulong(consumer_data.relayd_ht, &relayd->node);
502
00e2e675
DG
503end:
504 return ret;
505}
506
507/*
508 * Allocate and return a consumer relayd socket.
509 */
510struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
511 int net_seq_idx)
512{
513 struct consumer_relayd_sock_pair *obj = NULL;
514
515 /* Negative net sequence index is a failure */
516 if (net_seq_idx < 0) {
517 goto error;
518 }
519
520 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
521 if (obj == NULL) {
522 PERROR("zmalloc relayd sock");
523 goto error;
524 }
525
526 obj->net_seq_idx = net_seq_idx;
527 obj->refcount = 0;
173af62f 528 obj->destroy_flag = 0;
00e2e675
DG
529 lttng_ht_node_init_ulong(&obj->node, obj->net_seq_idx);
530 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
531
532error:
533 return obj;
534}
535
536/*
537 * Find a relayd socket pair in the global consumer data.
538 *
539 * Return the object if found else NULL.
b0b335c8
MD
540 * RCU read-side lock must be held across this call and while using the
541 * returned object.
00e2e675
DG
542 */
543struct consumer_relayd_sock_pair *consumer_find_relayd(int key)
544{
545 struct lttng_ht_iter iter;
546 struct lttng_ht_node_ulong *node;
547 struct consumer_relayd_sock_pair *relayd = NULL;
548
549 /* Negative keys are lookup failures */
550 if (key < 0) {
551 goto error;
552 }
553
00e2e675
DG
554 lttng_ht_lookup(consumer_data.relayd_ht, (void *)((unsigned long) key),
555 &iter);
556 node = lttng_ht_iter_get_node_ulong(&iter);
557 if (node != NULL) {
558 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
559 }
560
00e2e675
DG
561error:
562 return relayd;
563}
564
565/*
566 * Handle stream for relayd transmission if the stream applies for network
567 * streaming where the net sequence index is set.
568 *
569 * Return destination file descriptor or negative value on error.
570 */
6197aea7 571static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
1d4dfdef
DG
572 size_t data_size, unsigned long padding,
573 struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
574{
575 int outfd = -1, ret;
00e2e675
DG
576 struct lttcomm_relayd_data_hdr data_hdr;
577
578 /* Safety net */
579 assert(stream);
6197aea7 580 assert(relayd);
00e2e675
DG
581
582 /* Reset data header */
583 memset(&data_hdr, 0, sizeof(data_hdr));
584
00e2e675
DG
585 if (stream->metadata_flag) {
586 /* Caller MUST acquire the relayd control socket lock */
587 ret = relayd_send_metadata(&relayd->control_sock, data_size);
588 if (ret < 0) {
589 goto error;
590 }
591
592 /* Metadata are always sent on the control socket. */
593 outfd = relayd->control_sock.fd;
594 } else {
595 /* Set header with stream information */
596 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
597 data_hdr.data_size = htobe32(data_size);
1d4dfdef 598 data_hdr.padding_size = htobe32(padding);
173af62f 599 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num++);
00e2e675
DG
600 /* Other fields are zeroed previously */
601
602 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
603 sizeof(data_hdr));
604 if (ret < 0) {
605 goto error;
606 }
607
608 /* Set to go on data socket */
609 outfd = relayd->data_sock.fd;
610 }
611
612error:
613 return outfd;
614}
615
3bd1e081
MD
616/*
617 * Update a stream according to what we just received.
618 */
619void consumer_change_stream_state(int stream_key,
620 enum lttng_consumer_stream_state state)
621{
622 struct lttng_consumer_stream *stream;
623
624 pthread_mutex_lock(&consumer_data.lock);
8389e4f8 625 stream = consumer_find_stream(stream_key, consumer_data.stream_ht);
3bd1e081
MD
626 if (stream) {
627 stream->state = state;
628 }
629 consumer_data.need_update = 1;
630 pthread_mutex_unlock(&consumer_data.lock);
631}
632
702b1ea4
MD
633static
634void consumer_free_channel(struct rcu_head *head)
635{
636 struct lttng_ht_node_ulong *node =
637 caa_container_of(head, struct lttng_ht_node_ulong, head);
638 struct lttng_consumer_channel *channel =
639 caa_container_of(node, struct lttng_consumer_channel, node);
640
641 free(channel);
642}
643
3bd1e081
MD
644/*
645 * Remove a channel from the global list protected by a mutex. This
646 * function is also responsible for freeing its data structures.
647 */
648void consumer_del_channel(struct lttng_consumer_channel *channel)
649{
650 int ret;
e4421fec 651 struct lttng_ht_iter iter;
3bd1e081
MD
652
653 pthread_mutex_lock(&consumer_data.lock);
654
655 switch (consumer_data.type) {
656 case LTTNG_CONSUMER_KERNEL:
657 break;
7753dea8
MD
658 case LTTNG_CONSUMER32_UST:
659 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
660 lttng_ustconsumer_del_channel(channel);
661 break;
662 default:
663 ERR("Unknown consumer_data type");
664 assert(0);
665 goto end;
666 }
667
6065ceec 668 rcu_read_lock();
04253271
MD
669 iter.iter.node = &channel->node.node;
670 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
671 assert(!ret);
6065ceec
DG
672 rcu_read_unlock();
673
3bd1e081
MD
674 if (channel->mmap_base != NULL) {
675 ret = munmap(channel->mmap_base, channel->mmap_len);
676 if (ret != 0) {
7a57cf92 677 PERROR("munmap");
3bd1e081
MD
678 }
679 }
b5c5fc29 680 if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
4c462e79
MD
681 ret = close(channel->wait_fd);
682 if (ret) {
683 PERROR("close");
684 }
3bd1e081 685 }
2c1dd183 686 if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
4c462e79
MD
687 ret = close(channel->shm_fd);
688 if (ret) {
689 PERROR("close");
690 }
3bd1e081 691 }
702b1ea4
MD
692
693 call_rcu(&channel->node.head, consumer_free_channel);
3bd1e081
MD
694end:
695 pthread_mutex_unlock(&consumer_data.lock);
696}
697
698struct lttng_consumer_channel *consumer_allocate_channel(
699 int channel_key,
700 int shm_fd, int wait_fd,
701 uint64_t mmap_len,
c30aaa51
MD
702 uint64_t max_sb_size,
703 unsigned int nb_init_streams)
3bd1e081
MD
704{
705 struct lttng_consumer_channel *channel;
706 int ret;
707
276b26d1 708 channel = zmalloc(sizeof(*channel));
3bd1e081 709 if (channel == NULL) {
7a57cf92 710 PERROR("malloc struct lttng_consumer_channel");
3bd1e081
MD
711 goto end;
712 }
713 channel->key = channel_key;
714 channel->shm_fd = shm_fd;
715 channel->wait_fd = wait_fd;
716 channel->mmap_len = mmap_len;
717 channel->max_sb_size = max_sb_size;
718 channel->refcount = 0;
c30aaa51 719 channel->nb_init_streams = nb_init_streams;
e4421fec 720 lttng_ht_node_init_ulong(&channel->node, channel->key);
3bd1e081
MD
721
722 switch (consumer_data.type) {
723 case LTTNG_CONSUMER_KERNEL:
724 channel->mmap_base = NULL;
725 channel->mmap_len = 0;
726 break;
7753dea8
MD
727 case LTTNG_CONSUMER32_UST:
728 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
729 ret = lttng_ustconsumer_allocate_channel(channel);
730 if (ret) {
731 free(channel);
732 return NULL;
733 }
734 break;
735 default:
736 ERR("Unknown consumer_data type");
737 assert(0);
738 goto end;
739 }
740 DBG("Allocated channel (key %d, shm_fd %d, wait_fd %d, mmap_len %llu, max_sb_size %llu)",
00e2e675 741 channel->key, channel->shm_fd, channel->wait_fd,
3bd1e081
MD
742 (unsigned long long) channel->mmap_len,
743 (unsigned long long) channel->max_sb_size);
744end:
745 return channel;
746}
747
748/*
749 * Add a channel to the global list protected by a mutex.
750 */
751int consumer_add_channel(struct lttng_consumer_channel *channel)
752{
c77fc10a
DG
753 struct lttng_ht_node_ulong *node;
754 struct lttng_ht_iter iter;
755
3bd1e081 756 pthread_mutex_lock(&consumer_data.lock);
7ad0a0cb
MD
757 /* Steal channel identifier, for UST */
758 consumer_steal_channel_key(channel->key);
6065ceec 759 rcu_read_lock();
c77fc10a
DG
760
761 lttng_ht_lookup(consumer_data.channel_ht,
762 (void *)((unsigned long) channel->key), &iter);
763 node = lttng_ht_iter_get_node_ulong(&iter);
764 if (node != NULL) {
765 /* Channel already exist. Ignore the insertion */
766 goto end;
767 }
768
04253271 769 lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
c77fc10a
DG
770
771end:
6065ceec 772 rcu_read_unlock();
3bd1e081 773 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 774
7ad0a0cb 775 return 0;
3bd1e081
MD
776}
777
778/*
779 * Allocate the pollfd structure and the local view of the out fds to avoid
780 * doing a lookup in the linked list and concurrency issues when writing is
781 * needed. Called with consumer_data.lock held.
782 *
783 * Returns the number of fds in the structures.
784 */
785int consumer_update_poll_array(
786 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
fb3a43a9 787 struct lttng_consumer_stream **local_stream)
3bd1e081 788{
3bd1e081 789 int i = 0;
e4421fec
DG
790 struct lttng_ht_iter iter;
791 struct lttng_consumer_stream *stream;
3bd1e081
MD
792
793 DBG("Updating poll fd array");
481d6c57 794 rcu_read_lock();
e4421fec
DG
795 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
796 node.node) {
797 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
3bd1e081
MD
798 continue;
799 }
e4421fec
DG
800 DBG("Active FD %d", stream->wait_fd);
801 (*pollfd)[i].fd = stream->wait_fd;
3bd1e081 802 (*pollfd)[i].events = POLLIN | POLLPRI;
e4421fec 803 local_stream[i] = stream;
3bd1e081
MD
804 i++;
805 }
481d6c57 806 rcu_read_unlock();
3bd1e081
MD
807
808 /*
809 * Insert the consumer_poll_pipe at the end of the array and don't
810 * increment i so nb_fd is the number of real FD.
811 */
812 (*pollfd)[i].fd = ctx->consumer_poll_pipe[0];
509bb1cf 813 (*pollfd)[i].events = POLLIN | POLLPRI;
3bd1e081
MD
814 return i;
815}
816
817/*
818 * Poll on the should_quit pipe and the command socket return -1 on error and
819 * should exit, 0 if data is available on the command socket
820 */
821int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
822{
823 int num_rdy;
824
88f2b785 825restart:
3bd1e081
MD
826 num_rdy = poll(consumer_sockpoll, 2, -1);
827 if (num_rdy == -1) {
88f2b785
MD
828 /*
829 * Restart interrupted system call.
830 */
831 if (errno == EINTR) {
832 goto restart;
833 }
7a57cf92 834 PERROR("Poll error");
3bd1e081
MD
835 goto exit;
836 }
509bb1cf 837 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
3bd1e081
MD
838 DBG("consumer_should_quit wake up");
839 goto exit;
840 }
841 return 0;
842
843exit:
844 return -1;
845}
846
847/*
848 * Set the error socket.
849 */
850void lttng_consumer_set_error_sock(
851 struct lttng_consumer_local_data *ctx, int sock)
852{
853 ctx->consumer_error_socket = sock;
854}
855
856/*
857 * Set the command socket path.
858 */
3bd1e081
MD
859void lttng_consumer_set_command_sock_path(
860 struct lttng_consumer_local_data *ctx, char *sock)
861{
862 ctx->consumer_command_sock_path = sock;
863}
864
865/*
866 * Send return code to the session daemon.
867 * If the socket is not defined, we return 0, it is not a fatal error
868 */
869int lttng_consumer_send_error(
870 struct lttng_consumer_local_data *ctx, int cmd)
871{
872 if (ctx->consumer_error_socket > 0) {
873 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
874 sizeof(enum lttcomm_sessiond_command));
875 }
876
877 return 0;
878}
879
880/*
881 * Close all the tracefiles and stream fds, should be called when all instances
882 * are destroyed.
883 */
884void lttng_consumer_cleanup(void)
885{
e4421fec 886 struct lttng_ht_iter iter;
6065ceec
DG
887 struct lttng_ht_node_ulong *node;
888
889 rcu_read_lock();
3bd1e081
MD
890
891 /*
6065ceec
DG
892 * close all outfd. Called when there are no more threads running (after
893 * joining on the threads), no need to protect list iteration with mutex.
3bd1e081 894 */
6065ceec
DG
895 cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
896 node) {
702b1ea4
MD
897 struct lttng_consumer_stream *stream =
898 caa_container_of(node, struct lttng_consumer_stream, node);
899 consumer_del_stream(stream);
3bd1e081 900 }
e4421fec 901
6065ceec
DG
902 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
903 node) {
702b1ea4
MD
904 struct lttng_consumer_channel *channel =
905 caa_container_of(node, struct lttng_consumer_channel, node);
906 consumer_del_channel(channel);
3bd1e081 907 }
6065ceec
DG
908
909 rcu_read_unlock();
d6ce1df2
MD
910
911 lttng_ht_destroy(consumer_data.stream_ht);
912 lttng_ht_destroy(consumer_data.channel_ht);
3bd1e081
MD
913}
914
915/*
916 * Called from signal handler.
917 */
918void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
919{
920 int ret;
921 consumer_quit = 1;
6f94560a
MD
922 do {
923 ret = write(ctx->consumer_should_quit[1], "4", 1);
924 } while (ret < 0 && errno == EINTR);
3bd1e081 925 if (ret < 0) {
7a57cf92 926 PERROR("write consumer quit");
3bd1e081
MD
927 }
928}
929
00e2e675
DG
930void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
931 off_t orig_offset)
3bd1e081
MD
932{
933 int outfd = stream->out_fd;
934
935 /*
936 * This does a blocking write-and-wait on any page that belongs to the
937 * subbuffer prior to the one we just wrote.
938 * Don't care about error values, as these are just hints and ways to
939 * limit the amount of page cache used.
940 */
941 if (orig_offset < stream->chan->max_sb_size) {
942 return;
943 }
b9182dd9 944 lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
3bd1e081
MD
945 stream->chan->max_sb_size,
946 SYNC_FILE_RANGE_WAIT_BEFORE
947 | SYNC_FILE_RANGE_WRITE
948 | SYNC_FILE_RANGE_WAIT_AFTER);
949 /*
950 * Give hints to the kernel about how we access the file:
951 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
952 * we write it.
953 *
954 * We need to call fadvise again after the file grows because the
955 * kernel does not seem to apply fadvise to non-existing parts of the
956 * file.
957 *
958 * Call fadvise _after_ having waited for the page writeback to
959 * complete because the dirty page writeback semantic is not well
960 * defined. So it can be expected to lead to lower throughput in
961 * streaming.
962 */
963 posix_fadvise(outfd, orig_offset - stream->chan->max_sb_size,
964 stream->chan->max_sb_size, POSIX_FADV_DONTNEED);
965}
966
967/*
968 * Initialise the necessary environnement :
969 * - create a new context
970 * - create the poll_pipe
971 * - create the should_quit pipe (for signal handler)
972 * - create the thread pipe (for splice)
973 *
974 * Takes a function pointer as argument, this function is called when data is
975 * available on a buffer. This function is responsible to do the
976 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
977 * buffer configuration and then kernctl_put_next_subbuf at the end.
978 *
979 * Returns a pointer to the new context or NULL on error.
980 */
981struct lttng_consumer_local_data *lttng_consumer_create(
982 enum lttng_consumer_type type,
4078b776 983 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 984 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
985 int (*recv_channel)(struct lttng_consumer_channel *channel),
986 int (*recv_stream)(struct lttng_consumer_stream *stream),
987 int (*update_stream)(int stream_key, uint32_t state))
988{
989 int ret, i;
990 struct lttng_consumer_local_data *ctx;
991
992 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
993 consumer_data.type == type);
994 consumer_data.type = type;
995
effcf122 996 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
3bd1e081 997 if (ctx == NULL) {
7a57cf92 998 PERROR("allocating context");
3bd1e081
MD
999 goto error;
1000 }
1001
1002 ctx->consumer_error_socket = -1;
1003 /* assign the callbacks */
1004 ctx->on_buffer_ready = buffer_ready;
1005 ctx->on_recv_channel = recv_channel;
1006 ctx->on_recv_stream = recv_stream;
1007 ctx->on_update_stream = update_stream;
1008
1009 ret = pipe(ctx->consumer_poll_pipe);
1010 if (ret < 0) {
7a57cf92 1011 PERROR("Error creating poll pipe");
3bd1e081
MD
1012 goto error_poll_pipe;
1013 }
1014
04fdd819
MD
1015 /* set read end of the pipe to non-blocking */
1016 ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK);
1017 if (ret < 0) {
7a57cf92 1018 PERROR("fcntl O_NONBLOCK");
04fdd819
MD
1019 goto error_poll_fcntl;
1020 }
1021
1022 /* set write end of the pipe to non-blocking */
1023 ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK);
1024 if (ret < 0) {
7a57cf92 1025 PERROR("fcntl O_NONBLOCK");
04fdd819
MD
1026 goto error_poll_fcntl;
1027 }
1028
3bd1e081
MD
1029 ret = pipe(ctx->consumer_should_quit);
1030 if (ret < 0) {
7a57cf92 1031 PERROR("Error creating recv pipe");
3bd1e081
MD
1032 goto error_quit_pipe;
1033 }
1034
1035 ret = pipe(ctx->consumer_thread_pipe);
1036 if (ret < 0) {
7a57cf92 1037 PERROR("Error creating thread pipe");
3bd1e081
MD
1038 goto error_thread_pipe;
1039 }
1040
fb3a43a9
DG
1041 ret = utils_create_pipe(ctx->consumer_metadata_pipe);
1042 if (ret < 0) {
1043 goto error_metadata_pipe;
1044 }
3bd1e081 1045
fb3a43a9
DG
1046 ret = utils_create_pipe(ctx->consumer_splice_metadata_pipe);
1047 if (ret < 0) {
1048 goto error_splice_pipe;
1049 }
1050
1051 return ctx;
3bd1e081 1052
fb3a43a9
DG
1053error_splice_pipe:
1054 utils_close_pipe(ctx->consumer_metadata_pipe);
1055error_metadata_pipe:
1056 utils_close_pipe(ctx->consumer_thread_pipe);
3bd1e081
MD
1057error_thread_pipe:
1058 for (i = 0; i < 2; i++) {
1059 int err;
1060
1061 err = close(ctx->consumer_should_quit[i]);
4c462e79
MD
1062 if (err) {
1063 PERROR("close");
1064 }
3bd1e081 1065 }
04fdd819 1066error_poll_fcntl:
3bd1e081
MD
1067error_quit_pipe:
1068 for (i = 0; i < 2; i++) {
1069 int err;
1070
1071 err = close(ctx->consumer_poll_pipe[i]);
4c462e79
MD
1072 if (err) {
1073 PERROR("close");
1074 }
3bd1e081
MD
1075 }
1076error_poll_pipe:
1077 free(ctx);
1078error:
1079 return NULL;
1080}
1081
1082/*
1083 * Close all fds associated with the instance and free the context.
1084 */
1085void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1086{
4c462e79
MD
1087 int ret;
1088
1089 ret = close(ctx->consumer_error_socket);
1090 if (ret) {
1091 PERROR("close");
1092 }
1093 ret = close(ctx->consumer_thread_pipe[0]);
1094 if (ret) {
1095 PERROR("close");
1096 }
1097 ret = close(ctx->consumer_thread_pipe[1]);
1098 if (ret) {
1099 PERROR("close");
1100 }
1101 ret = close(ctx->consumer_poll_pipe[0]);
1102 if (ret) {
1103 PERROR("close");
1104 }
1105 ret = close(ctx->consumer_poll_pipe[1]);
1106 if (ret) {
1107 PERROR("close");
1108 }
1109 ret = close(ctx->consumer_should_quit[0]);
1110 if (ret) {
1111 PERROR("close");
1112 }
1113 ret = close(ctx->consumer_should_quit[1]);
1114 if (ret) {
1115 PERROR("close");
1116 }
fb3a43a9
DG
1117 utils_close_pipe(ctx->consumer_splice_metadata_pipe);
1118
3bd1e081
MD
1119 unlink(ctx->consumer_command_sock_path);
1120 free(ctx);
1121}
1122
6197aea7
DG
1123/*
1124 * Write the metadata stream id on the specified file descriptor.
1125 */
1126static int write_relayd_metadata_id(int fd,
1127 struct lttng_consumer_stream *stream,
1d4dfdef
DG
1128 struct consumer_relayd_sock_pair *relayd,
1129 unsigned long padding)
6197aea7
DG
1130{
1131 int ret;
1d4dfdef 1132 struct lttcomm_relayd_metadata_payload hdr;
6197aea7 1133
1d4dfdef
DG
1134 hdr.stream_id = htobe64(stream->relayd_stream_id);
1135 hdr.padding_size = htobe32(padding);
6197aea7 1136 do {
1d4dfdef 1137 ret = write(fd, (void *) &hdr, sizeof(hdr));
6197aea7
DG
1138 } while (ret < 0 && errno == EINTR);
1139 if (ret < 0) {
1140 PERROR("write metadata stream id");
1141 goto end;
1142 }
1d4dfdef
DG
1143 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1144 stream->relayd_stream_id, padding);
6197aea7
DG
1145
1146end:
1147 return ret;
1148}
1149
3bd1e081 1150/*
09e26845
DG
1151 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1152 * core function for writing trace buffers to either the local filesystem or
1153 * the network.
1154 *
1155 * Careful review MUST be put if any changes occur!
3bd1e081
MD
1156 *
1157 * Returns the number of bytes written
1158 */
4078b776 1159ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081 1160 struct lttng_consumer_local_data *ctx,
1d4dfdef
DG
1161 struct lttng_consumer_stream *stream, unsigned long len,
1162 unsigned long padding)
3bd1e081 1163{
f02e1e8a
DG
1164 unsigned long mmap_offset;
1165 ssize_t ret = 0, written = 0;
1166 off_t orig_offset = stream->out_fd_offset;
1167 /* Default is on the disk */
1168 int outfd = stream->out_fd;
f02e1e8a
DG
1169 struct consumer_relayd_sock_pair *relayd = NULL;
1170
1171 /* RCU lock for the relayd pointer */
1172 rcu_read_lock();
1173
1174 /* Flag that the current stream if set for network streaming. */
1175 if (stream->net_seq_idx != -1) {
1176 relayd = consumer_find_relayd(stream->net_seq_idx);
1177 if (relayd == NULL) {
1178 goto end;
1179 }
1180 }
1181
1182 /* get the offset inside the fd to mmap */
3bd1e081
MD
1183 switch (consumer_data.type) {
1184 case LTTNG_CONSUMER_KERNEL:
f02e1e8a
DG
1185 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
1186 break;
7753dea8
MD
1187 case LTTNG_CONSUMER32_UST:
1188 case LTTNG_CONSUMER64_UST:
f02e1e8a
DG
1189 ret = lttng_ustctl_get_mmap_read_offset(stream->chan->handle,
1190 stream->buf, &mmap_offset);
1191 break;
3bd1e081
MD
1192 default:
1193 ERR("Unknown consumer_data type");
1194 assert(0);
1195 }
f02e1e8a
DG
1196 if (ret != 0) {
1197 errno = -ret;
1198 PERROR("tracer ctl get_mmap_read_offset");
1199 written = ret;
1200 goto end;
1201 }
b9182dd9 1202
f02e1e8a
DG
1203 /* Handle stream on the relayd if the output is on the network */
1204 if (relayd) {
1205 unsigned long netlen = len;
1206
1207 /*
1208 * Lock the control socket for the complete duration of the function
1209 * since from this point on we will use the socket.
1210 */
1211 if (stream->metadata_flag) {
1212 /* Metadata requires the control socket. */
1213 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1d4dfdef 1214 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
f02e1e8a
DG
1215 }
1216
1d4dfdef 1217 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
f02e1e8a
DG
1218 if (ret >= 0) {
1219 /* Use the returned socket. */
1220 outfd = ret;
1221
1222 /* Write metadata stream id before payload */
1223 if (stream->metadata_flag) {
1d4dfdef 1224 ret = write_relayd_metadata_id(outfd, stream, relayd, padding);
f02e1e8a 1225 if (ret < 0) {
f02e1e8a
DG
1226 written = ret;
1227 goto end;
1228 }
f02e1e8a
DG
1229 }
1230 }
1231 /* Else, use the default set before which is the filesystem. */
1d4dfdef
DG
1232 } else {
1233 /* No streaming, we have to set the len with the full padding */
1234 len += padding;
f02e1e8a
DG
1235 }
1236
1237 while (len > 0) {
1238 do {
1239 ret = write(outfd, stream->mmap_base + mmap_offset, len);
1240 } while (ret < 0 && errno == EINTR);
1d4dfdef 1241 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
f02e1e8a
DG
1242 if (ret < 0) {
1243 PERROR("Error in file write");
1244 if (written == 0) {
1245 written = ret;
1246 }
1247 goto end;
1248 } else if (ret > len) {
77c7c900 1249 PERROR("Error in file write (ret %zd > len %lu)", ret, len);
f02e1e8a
DG
1250 written += ret;
1251 goto end;
1252 } else {
1253 len -= ret;
1254 mmap_offset += ret;
1255 }
f02e1e8a
DG
1256
1257 /* This call is useless on a socket so better save a syscall. */
1258 if (!relayd) {
1259 /* This won't block, but will start writeout asynchronously */
1260 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
1261 SYNC_FILE_RANGE_WRITE);
1262 stream->out_fd_offset += ret;
1263 }
1264 written += ret;
1265 }
1266 lttng_consumer_sync_trace_file(stream, orig_offset);
1267
1268end:
1269 /* Unlock only if ctrl socket used */
1270 if (relayd && stream->metadata_flag) {
1271 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1272 }
1273
1274 rcu_read_unlock();
1275 return written;
3bd1e081
MD
1276}
1277
1278/*
1279 * Splice the data from the ring buffer to the tracefile.
1280 *
1281 * Returns the number of bytes spliced.
1282 */
4078b776 1283ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081 1284 struct lttng_consumer_local_data *ctx,
1d4dfdef
DG
1285 struct lttng_consumer_stream *stream, unsigned long len,
1286 unsigned long padding)
3bd1e081 1287{
f02e1e8a
DG
1288 ssize_t ret = 0, written = 0, ret_splice = 0;
1289 loff_t offset = 0;
1290 off_t orig_offset = stream->out_fd_offset;
1291 int fd = stream->wait_fd;
1292 /* Default is on the disk */
1293 int outfd = stream->out_fd;
f02e1e8a 1294 struct consumer_relayd_sock_pair *relayd = NULL;
fb3a43a9 1295 int *splice_pipe;
f02e1e8a 1296
3bd1e081
MD
1297 switch (consumer_data.type) {
1298 case LTTNG_CONSUMER_KERNEL:
f02e1e8a 1299 break;
7753dea8
MD
1300 case LTTNG_CONSUMER32_UST:
1301 case LTTNG_CONSUMER64_UST:
f02e1e8a 1302 /* Not supported for user space tracing */
3bd1e081
MD
1303 return -ENOSYS;
1304 default:
1305 ERR("Unknown consumer_data type");
1306 assert(0);
3bd1e081
MD
1307 }
1308
f02e1e8a
DG
1309 /* RCU lock for the relayd pointer */
1310 rcu_read_lock();
1311
1312 /* Flag that the current stream if set for network streaming. */
1313 if (stream->net_seq_idx != -1) {
1314 relayd = consumer_find_relayd(stream->net_seq_idx);
1315 if (relayd == NULL) {
1316 goto end;
1317 }
1318 }
1319
fb3a43a9
DG
1320 /*
1321 * Choose right pipe for splice. Metadata and trace data are handled by
1322 * different threads hence the use of two pipes in order not to race or
1323 * corrupt the written data.
1324 */
1325 if (stream->metadata_flag) {
1326 splice_pipe = ctx->consumer_splice_metadata_pipe;
1327 } else {
1328 splice_pipe = ctx->consumer_thread_pipe;
1329 }
1330
f02e1e8a 1331 /* Write metadata stream id before payload */
1d4dfdef
DG
1332 if (relayd) {
1333 int total_len = len;
f02e1e8a 1334
1d4dfdef
DG
1335 if (stream->metadata_flag) {
1336 /*
1337 * Lock the control socket for the complete duration of the function
1338 * since from this point on we will use the socket.
1339 */
1340 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1341
1342 ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
1343 padding);
1344 if (ret < 0) {
1345 written = ret;
1346 goto end;
1347 }
1348
1349 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1350 }
1351
1352 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
1353 if (ret >= 0) {
1354 /* Use the returned socket. */
1355 outfd = ret;
1356 } else {
1357 ERR("Remote relayd disconnected. Stopping");
f02e1e8a
DG
1358 goto end;
1359 }
1d4dfdef
DG
1360 } else {
1361 /* No streaming, we have to set the len with the full padding */
1362 len += padding;
f02e1e8a
DG
1363 }
1364
1365 while (len > 0) {
1d4dfdef
DG
1366 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1367 (unsigned long)offset, len, fd, splice_pipe[1]);
fb3a43a9 1368 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
f02e1e8a
DG
1369 SPLICE_F_MOVE | SPLICE_F_MORE);
1370 DBG("splice chan to pipe, ret %zd", ret_splice);
1371 if (ret_splice < 0) {
1372 PERROR("Error in relay splice");
1373 if (written == 0) {
1374 written = ret_splice;
1375 }
1376 ret = errno;
1377 goto splice_error;
1378 }
1379
1380 /* Handle stream on the relayd if the output is on the network */
1381 if (relayd) {
1382 if (stream->metadata_flag) {
1d4dfdef
DG
1383 size_t metadata_payload_size =
1384 sizeof(struct lttcomm_relayd_metadata_payload);
1385
f02e1e8a 1386 /* Update counter to fit the spliced data */
1d4dfdef
DG
1387 ret_splice += metadata_payload_size;
1388 len += metadata_payload_size;
f02e1e8a
DG
1389 /*
1390 * We do this so the return value can match the len passed as
1391 * argument to this function.
1392 */
1d4dfdef 1393 written -= metadata_payload_size;
f02e1e8a
DG
1394 }
1395 }
1396
1397 /* Splice data out */
fb3a43a9 1398 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
f02e1e8a 1399 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
1d4dfdef 1400 DBG("Consumer splice pipe to file, ret %zd", ret_splice);
f02e1e8a
DG
1401 if (ret_splice < 0) {
1402 PERROR("Error in file splice");
1403 if (written == 0) {
1404 written = ret_splice;
1405 }
1406 ret = errno;
1407 goto splice_error;
1408 } else if (ret_splice > len) {
1409 errno = EINVAL;
1410 PERROR("Wrote more data than requested %zd (len: %lu)",
1411 ret_splice, len);
1412 written += ret_splice;
1413 ret = errno;
1414 goto splice_error;
1415 }
1416 len -= ret_splice;
1417
1418 /* This call is useless on a socket so better save a syscall. */
1419 if (!relayd) {
1420 /* This won't block, but will start writeout asynchronously */
1421 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
1422 SYNC_FILE_RANGE_WRITE);
1423 stream->out_fd_offset += ret_splice;
1424 }
1425 written += ret_splice;
1426 }
1427 lttng_consumer_sync_trace_file(stream, orig_offset);
1428
1429 ret = ret_splice;
1430
1431 goto end;
1432
1433splice_error:
1434 /* send the appropriate error description to sessiond */
1435 switch (ret) {
1436 case EBADF:
f73fabfd 1437 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EBADF);
f02e1e8a
DG
1438 break;
1439 case EINVAL:
f73fabfd 1440 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
f02e1e8a
DG
1441 break;
1442 case ENOMEM:
f73fabfd 1443 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
f02e1e8a
DG
1444 break;
1445 case ESPIPE:
f73fabfd 1446 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
f02e1e8a
DG
1447 break;
1448 }
1449
1450end:
1451 if (relayd && stream->metadata_flag) {
1452 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1453 }
1454
1455 rcu_read_unlock();
1456 return written;
3bd1e081
MD
1457}
1458
1459/*
1460 * Take a snapshot for a specific fd
1461 *
1462 * Returns 0 on success, < 0 on error
1463 */
1464int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
1465 struct lttng_consumer_stream *stream)
1466{
1467 switch (consumer_data.type) {
1468 case LTTNG_CONSUMER_KERNEL:
1469 return lttng_kconsumer_take_snapshot(ctx, stream);
7753dea8
MD
1470 case LTTNG_CONSUMER32_UST:
1471 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
1472 return lttng_ustconsumer_take_snapshot(ctx, stream);
1473 default:
1474 ERR("Unknown consumer_data type");
1475 assert(0);
1476 return -ENOSYS;
1477 }
1478
1479}
1480
1481/*
1482 * Get the produced position
1483 *
1484 * Returns 0 on success, < 0 on error
1485 */
1486int lttng_consumer_get_produced_snapshot(
1487 struct lttng_consumer_local_data *ctx,
1488 struct lttng_consumer_stream *stream,
1489 unsigned long *pos)
1490{
1491 switch (consumer_data.type) {
1492 case LTTNG_CONSUMER_KERNEL:
1493 return lttng_kconsumer_get_produced_snapshot(ctx, stream, pos);
7753dea8
MD
1494 case LTTNG_CONSUMER32_UST:
1495 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
1496 return lttng_ustconsumer_get_produced_snapshot(ctx, stream, pos);
1497 default:
1498 ERR("Unknown consumer_data type");
1499 assert(0);
1500 return -ENOSYS;
1501 }
1502}
1503
1504int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
1505 int sock, struct pollfd *consumer_sockpoll)
1506{
1507 switch (consumer_data.type) {
1508 case LTTNG_CONSUMER_KERNEL:
1509 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
1510 case LTTNG_CONSUMER32_UST:
1511 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
1512 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
1513 default:
1514 ERR("Unknown consumer_data type");
1515 assert(0);
1516 return -ENOSYS;
1517 }
1518}
1519
fb3a43a9 1520/*
f724d81e 1521 * Iterate over all streams of the hashtable and free them properly.
fb3a43a9
DG
1522 */
1523static void destroy_stream_ht(struct lttng_ht *ht)
1524{
1525 int ret;
1526 struct lttng_ht_iter iter;
1527 struct lttng_consumer_stream *stream;
1528
1529 if (ht == NULL) {
1530 return;
1531 }
1532
d09e1200 1533 rcu_read_lock();
fb3a43a9
DG
1534 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1535 ret = lttng_ht_del(ht, &iter);
1536 assert(!ret);
1537
f724d81e 1538 call_rcu(&stream->node.head, consumer_free_stream);
fb3a43a9 1539 }
d09e1200 1540 rcu_read_unlock();
fb3a43a9
DG
1541
1542 lttng_ht_destroy(ht);
1543}
1544
1545/*
1546 * Clean up a metadata stream and free its memory.
1547 */
1548static void consumer_del_metadata_stream(struct lttng_consumer_stream *stream)
1549{
1550 int ret;
fb3a43a9
DG
1551 struct consumer_relayd_sock_pair *relayd;
1552
1553 assert(stream);
1554 /*
1555 * This call should NEVER receive regular stream. It must always be
1556 * metadata stream and this is crucial for data structure synchronization.
1557 */
1558 assert(stream->metadata_flag);
1559
1560 pthread_mutex_lock(&consumer_data.lock);
1561 switch (consumer_data.type) {
1562 case LTTNG_CONSUMER_KERNEL:
1563 if (stream->mmap_base != NULL) {
1564 ret = munmap(stream->mmap_base, stream->mmap_len);
1565 if (ret != 0) {
1566 PERROR("munmap metadata stream");
1567 }
1568 }
1569 break;
1570 case LTTNG_CONSUMER32_UST:
1571 case LTTNG_CONSUMER64_UST:
1572 lttng_ustconsumer_del_stream(stream);
1573 break;
1574 default:
1575 ERR("Unknown consumer_data type");
1576 assert(0);
1577 }
1578 pthread_mutex_unlock(&consumer_data.lock);
1579
1580 if (stream->out_fd >= 0) {
1581 ret = close(stream->out_fd);
1582 if (ret) {
1583 PERROR("close");
1584 }
1585 }
1586
1587 if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
1588 ret = close(stream->wait_fd);
1589 if (ret) {
1590 PERROR("close");
1591 }
1592 }
1593
1594 if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
1595 ret = close(stream->shm_fd);
1596 if (ret) {
1597 PERROR("close");
1598 }
1599 }
1600
1601 /* Check and cleanup relayd */
1602 rcu_read_lock();
1603 relayd = consumer_find_relayd(stream->net_seq_idx);
1604 if (relayd != NULL) {
1605 uatomic_dec(&relayd->refcount);
1606 assert(uatomic_read(&relayd->refcount) >= 0);
1607
1608 /* Closing streams requires to lock the control socket. */
1609 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1610 ret = relayd_send_close_stream(&relayd->control_sock,
1611 stream->relayd_stream_id, stream->next_net_seq_num - 1);
1612 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1613 if (ret < 0) {
1614 DBG("Unable to close stream on the relayd. Continuing");
1615 /*
1616 * Continue here. There is nothing we can do for the relayd.
1617 * Chances are that the relayd has closed the socket so we just
1618 * continue cleaning up.
1619 */
1620 }
1621
1622 /* Both conditions are met, we destroy the relayd. */
1623 if (uatomic_read(&relayd->refcount) == 0 &&
1624 uatomic_read(&relayd->destroy_flag)) {
d09e1200 1625 destroy_relayd(relayd);
fb3a43a9
DG
1626 }
1627 }
1628 rcu_read_unlock();
1629
1630 /* Atomically decrement channel refcount since other threads can use it. */
1631 uatomic_dec(&stream->chan->refcount);
c30aaa51
MD
1632 if (!uatomic_read(&stream->chan->refcount)
1633 && !uatomic_read(&stream->chan->nb_init_streams)) {
1634 /* Go for channel deletion! */
1635 consumer_del_channel(stream->chan);
fb3a43a9
DG
1636 }
1637
f724d81e 1638 call_rcu(&stream->node.head, consumer_free_stream);
fb3a43a9
DG
1639}
1640
1641/*
1642 * Action done with the metadata stream when adding it to the consumer internal
1643 * data structures to handle it.
1644 */
1645static void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
1646{
1647 struct consumer_relayd_sock_pair *relayd;
1648
1649 /* Find relayd and, if one is found, increment refcount. */
1650 rcu_read_lock();
1651 relayd = consumer_find_relayd(stream->net_seq_idx);
1652 if (relayd != NULL) {
1653 uatomic_inc(&relayd->refcount);
1654 }
1655 rcu_read_unlock();
1656}
1657
1658/*
1659 * Thread polls on metadata file descriptor and write them on disk or on the
1660 * network.
1661 */
1662void *lttng_consumer_thread_poll_metadata(void *data)
1663{
1664 int ret, i, pollfd;
1665 uint32_t revents, nb_fd;
1666 struct lttng_consumer_stream *stream;
1667 struct lttng_ht_iter iter;
1668 struct lttng_ht_node_ulong *node;
1669 struct lttng_ht *metadata_ht = NULL;
1670 struct lttng_poll_event events;
1671 struct lttng_consumer_local_data *ctx = data;
1672 ssize_t len;
1673
1674 rcu_register_thread();
1675
1676 DBG("Thread metadata poll started");
1677
1678 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1679 if (metadata_ht == NULL) {
1680 goto end;
1681 }
1682
1683 /* Size is set to 1 for the consumer_metadata pipe */
1684 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
1685 if (ret < 0) {
1686 ERR("Poll set creation failed");
1687 goto end;
1688 }
1689
1690 ret = lttng_poll_add(&events, ctx->consumer_metadata_pipe[0], LPOLLIN);
1691 if (ret < 0) {
1692 goto end;
1693 }
1694
1695 /* Main loop */
1696 DBG("Metadata main loop started");
1697
1698 while (1) {
1699 lttng_poll_reset(&events);
1700
1701 nb_fd = LTTNG_POLL_GETNB(&events);
1702
1703 /* Only the metadata pipe is set */
1704 if (nb_fd == 0 && consumer_quit == 1) {
1705 goto end;
1706 }
1707
1708restart:
1709 DBG("Metadata poll wait with %d fd(s)", nb_fd);
1710 ret = lttng_poll_wait(&events, -1);
1711 DBG("Metadata event catched in thread");
1712 if (ret < 0) {
1713 if (errno == EINTR) {
1714 goto restart;
1715 }
1716 goto error;
1717 }
1718
1719 for (i = 0; i < nb_fd; i++) {
1720 revents = LTTNG_POLL_GETEV(&events, i);
1721 pollfd = LTTNG_POLL_GETFD(&events, i);
1722
1723 /* Check the metadata pipe for incoming metadata. */
1724 if (pollfd == ctx->consumer_metadata_pipe[0]) {
4adabd61 1725 if (revents & (LPOLLERR | LPOLLHUP )) {
fb3a43a9
DG
1726 DBG("Metadata thread pipe hung up");
1727 /*
1728 * Remove the pipe from the poll set and continue the loop
1729 * since their might be data to consume.
1730 */
1731 lttng_poll_del(&events, ctx->consumer_metadata_pipe[0]);
1732 close(ctx->consumer_metadata_pipe[0]);
1733 continue;
1734 } else if (revents & LPOLLIN) {
fb3a43a9 1735 do {
633d0084
DG
1736 /* Get the stream pointer received */
1737 ret = read(pollfd, &stream, sizeof(stream));
fb3a43a9 1738 } while (ret < 0 && errno == EINTR);
633d0084
DG
1739 if (ret < 0 ||
1740 ret < sizeof(struct lttng_consumer_stream *)) {
fb3a43a9 1741 PERROR("read metadata stream");
fb3a43a9
DG
1742 /*
1743 * Let's continue here and hope we can still work
1744 * without stopping the consumer. XXX: Should we?
1745 */
1746 continue;
1747 }
1748
1749 DBG("Adding metadata stream %d to poll set",
1750 stream->wait_fd);
1751
d09e1200 1752 rcu_read_lock();
fb3a43a9
DG
1753 /* The node should be init at this point */
1754 lttng_ht_add_unique_ulong(metadata_ht,
1755 &stream->waitfd_node);
d09e1200 1756 rcu_read_unlock();
fb3a43a9
DG
1757
1758 /* Add metadata stream to the global poll events list */
1759 lttng_poll_add(&events, stream->wait_fd,
1760 LPOLLIN | LPOLLPRI);
1761
1762 consumer_add_metadata_stream(stream);
1763 }
1764
1765 /* Metadata pipe handled. Continue handling the others */
1766 continue;
1767 }
1768
1769 /* From here, the event is a metadata wait fd */
1770
d09e1200 1771 rcu_read_lock();
fb3a43a9
DG
1772 lttng_ht_lookup(metadata_ht, (void *)((unsigned long) pollfd),
1773 &iter);
1774 node = lttng_ht_iter_get_node_ulong(&iter);
1775 if (node == NULL) {
1776 /* FD not found, continue loop */
d09e1200 1777 rcu_read_unlock();
fb3a43a9
DG
1778 continue;
1779 }
1780
1781 stream = caa_container_of(node, struct lttng_consumer_stream,
1782 waitfd_node);
1783
1784 /* Get the data out of the metadata file descriptor */
1785 if (revents & (LPOLLIN | LPOLLPRI)) {
1786 DBG("Metadata available on fd %d", pollfd);
1787 assert(stream->wait_fd == pollfd);
1788
1789 len = ctx->on_buffer_ready(stream, ctx);
1790 /* It's ok to have an unavailable sub-buffer */
1791 if (len < 0 && len != -EAGAIN) {
d09e1200 1792 rcu_read_unlock();
fb3a43a9
DG
1793 goto end;
1794 } else if (len > 0) {
1795 stream->data_read = 1;
1796 }
1797 }
1798
1799 /*
1800 * Remove the stream from the hash table since there is no data
1801 * left on the fd because we previously did a read on the buffer.
1802 */
4adabd61 1803 if (revents & (LPOLLERR | LPOLLHUP)) {
fb3a43a9
DG
1804 DBG("Metadata fd %d is hup|err|nval.", pollfd);
1805 if (!stream->hangup_flush_done
1806 && (consumer_data.type == LTTNG_CONSUMER32_UST
1807 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
1808 DBG("Attempting to flush and consume the UST buffers");
1809 lttng_ustconsumer_on_stream_hangup(stream);
1810
1811 /* We just flushed the stream now read it. */
1812 len = ctx->on_buffer_ready(stream, ctx);
1813 /* It's ok to have an unavailable sub-buffer */
1814 if (len < 0 && len != -EAGAIN) {
d09e1200 1815 rcu_read_unlock();
fb3a43a9
DG
1816 goto end;
1817 }
1818 }
1819
1820 /* Removing it from hash table, poll set and free memory */
1821 lttng_ht_del(metadata_ht, &iter);
d09e1200 1822
fb3a43a9
DG
1823 lttng_poll_del(&events, stream->wait_fd);
1824 consumer_del_metadata_stream(stream);
1825 }
d09e1200 1826 rcu_read_unlock();
fb3a43a9
DG
1827 }
1828 }
1829
1830error:
1831end:
1832 DBG("Metadata poll thread exiting");
1833 lttng_poll_clean(&events);
1834
1835 if (metadata_ht) {
1836 destroy_stream_ht(metadata_ht);
1837 }
1838
1839 rcu_unregister_thread();
1840 return NULL;
1841}
1842
3bd1e081 1843/*
e4421fec 1844 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
1845 * it to tracefile if necessary.
1846 */
1847void *lttng_consumer_thread_poll_fds(void *data)
1848{
1849 int num_rdy, num_hup, high_prio, ret, i;
1850 struct pollfd *pollfd = NULL;
1851 /* local view of the streams */
1852 struct lttng_consumer_stream **local_stream = NULL;
1853 /* local view of consumer_data.fds_count */
1854 int nb_fd = 0;
3bd1e081 1855 struct lttng_consumer_local_data *ctx = data;
00e2e675 1856 ssize_t len;
fb3a43a9
DG
1857 pthread_t metadata_thread;
1858 void *status;
3bd1e081 1859
e7b994a3
DG
1860 rcu_register_thread();
1861
fb3a43a9
DG
1862 /* Start metadata polling thread */
1863 ret = pthread_create(&metadata_thread, NULL,
1864 lttng_consumer_thread_poll_metadata, (void *) ctx);
1865 if (ret < 0) {
1866 PERROR("pthread_create metadata thread");
1867 goto end;
1868 }
1869
effcf122 1870 local_stream = zmalloc(sizeof(struct lttng_consumer_stream));
3bd1e081
MD
1871
1872 while (1) {
1873 high_prio = 0;
1874 num_hup = 0;
1875
1876 /*
e4421fec 1877 * the fds set has been updated, we need to update our
3bd1e081
MD
1878 * local array as well
1879 */
1880 pthread_mutex_lock(&consumer_data.lock);
1881 if (consumer_data.need_update) {
1882 if (pollfd != NULL) {
1883 free(pollfd);
1884 pollfd = NULL;
1885 }
1886 if (local_stream != NULL) {
1887 free(local_stream);
1888 local_stream = NULL;
1889 }
1890
1891 /* allocate for all fds + 1 for the consumer_poll_pipe */
effcf122 1892 pollfd = zmalloc((consumer_data.stream_count + 1) * sizeof(struct pollfd));
3bd1e081 1893 if (pollfd == NULL) {
7a57cf92 1894 PERROR("pollfd malloc");
3bd1e081
MD
1895 pthread_mutex_unlock(&consumer_data.lock);
1896 goto end;
1897 }
1898
1899 /* allocate for all fds + 1 for the consumer_poll_pipe */
effcf122 1900 local_stream = zmalloc((consumer_data.stream_count + 1) *
3bd1e081
MD
1901 sizeof(struct lttng_consumer_stream));
1902 if (local_stream == NULL) {
7a57cf92 1903 PERROR("local_stream malloc");
3bd1e081
MD
1904 pthread_mutex_unlock(&consumer_data.lock);
1905 goto end;
1906 }
fb3a43a9 1907 ret = consumer_update_poll_array(ctx, &pollfd, local_stream);
3bd1e081
MD
1908 if (ret < 0) {
1909 ERR("Error in allocating pollfd or local_outfds");
f73fabfd 1910 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
1911 pthread_mutex_unlock(&consumer_data.lock);
1912 goto end;
1913 }
1914 nb_fd = ret;
1915 consumer_data.need_update = 0;
1916 }
1917 pthread_mutex_unlock(&consumer_data.lock);
1918
4078b776
MD
1919 /* No FDs and consumer_quit, consumer_cleanup the thread */
1920 if (nb_fd == 0 && consumer_quit == 1) {
1921 goto end;
1922 }
3bd1e081 1923 /* poll on the array of fds */
88f2b785 1924 restart:
3bd1e081
MD
1925 DBG("polling on %d fd", nb_fd + 1);
1926 num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
1927 DBG("poll num_rdy : %d", num_rdy);
1928 if (num_rdy == -1) {
88f2b785
MD
1929 /*
1930 * Restart interrupted system call.
1931 */
1932 if (errno == EINTR) {
1933 goto restart;
1934 }
7a57cf92 1935 PERROR("Poll error");
f73fabfd 1936 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
1937 goto end;
1938 } else if (num_rdy == 0) {
1939 DBG("Polling thread timed out");
1940 goto end;
1941 }
1942
3bd1e081 1943 /*
00e2e675
DG
1944 * If the consumer_poll_pipe triggered poll go directly to the
1945 * beginning of the loop to update the array. We want to prioritize
1946 * array update over low-priority reads.
3bd1e081 1947 */
509bb1cf 1948 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
04fdd819
MD
1949 size_t pipe_readlen;
1950 char tmp;
1951
3bd1e081 1952 DBG("consumer_poll_pipe wake up");
04fdd819
MD
1953 /* Consume 1 byte of pipe data */
1954 do {
1955 pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1);
1956 } while (pipe_readlen == -1 && errno == EINTR);
3bd1e081
MD
1957 continue;
1958 }
1959
1960 /* Take care of high priority channels first. */
1961 for (i = 0; i < nb_fd; i++) {
fb3a43a9 1962 if (pollfd[i].revents & POLLPRI) {
d41f73b7
MD
1963 DBG("Urgent read on fd %d", pollfd[i].fd);
1964 high_prio = 1;
4078b776 1965 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 1966 /* it's ok to have an unavailable sub-buffer */
4078b776
MD
1967 if (len < 0 && len != -EAGAIN) {
1968 goto end;
1969 } else if (len > 0) {
1970 local_stream[i]->data_read = 1;
d41f73b7 1971 }
3bd1e081
MD
1972 }
1973 }
1974
4078b776
MD
1975 /*
1976 * If we read high prio channel in this loop, try again
1977 * for more high prio data.
1978 */
1979 if (high_prio) {
3bd1e081
MD
1980 continue;
1981 }
1982
1983 /* Take care of low priority channels. */
4078b776
MD
1984 for (i = 0; i < nb_fd; i++) {
1985 if ((pollfd[i].revents & POLLIN) ||
1986 local_stream[i]->hangup_flush_done) {
4078b776
MD
1987 DBG("Normal read on fd %d", pollfd[i].fd);
1988 len = ctx->on_buffer_ready(local_stream[i], ctx);
1989 /* it's ok to have an unavailable sub-buffer */
1990 if (len < 0 && len != -EAGAIN) {
1991 goto end;
1992 } else if (len > 0) {
1993 local_stream[i]->data_read = 1;
1994 }
1995 }
1996 }
1997
1998 /* Handle hangup and errors */
1999 for (i = 0; i < nb_fd; i++) {
2000 if (!local_stream[i]->hangup_flush_done
2001 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2002 && (consumer_data.type == LTTNG_CONSUMER32_UST
2003 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2004 DBG("fd %d is hup|err|nval. Attempting flush and read.",
2005 pollfd[i].fd);
2006 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2007 /* Attempt read again, for the data we just flushed. */
2008 local_stream[i]->data_read = 1;
2009 }
2010 /*
2011 * If the poll flag is HUP/ERR/NVAL and we have
2012 * read no data in this pass, we can remove the
2013 * stream from its hash table.
2014 */
2015 if ((pollfd[i].revents & POLLHUP)) {
2016 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2017 if (!local_stream[i]->data_read) {
702b1ea4 2018 consumer_del_stream(local_stream[i]);
4078b776
MD
2019 num_hup++;
2020 }
2021 } else if (pollfd[i].revents & POLLERR) {
2022 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2023 if (!local_stream[i]->data_read) {
702b1ea4 2024 consumer_del_stream(local_stream[i]);
4078b776
MD
2025 num_hup++;
2026 }
2027 } else if (pollfd[i].revents & POLLNVAL) {
2028 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2029 if (!local_stream[i]->data_read) {
702b1ea4 2030 consumer_del_stream(local_stream[i]);
4078b776 2031 num_hup++;
3bd1e081
MD
2032 }
2033 }
4078b776 2034 local_stream[i]->data_read = 0;
3bd1e081
MD
2035 }
2036 }
2037end:
2038 DBG("polling thread exiting");
2039 if (pollfd != NULL) {
2040 free(pollfd);
2041 pollfd = NULL;
2042 }
2043 if (local_stream != NULL) {
2044 free(local_stream);
2045 local_stream = NULL;
2046 }
fb3a43a9
DG
2047
2048 /*
2049 * Close the write side of the pipe so epoll_wait() in
2050 * lttng_consumer_thread_poll_metadata can catch it. The thread is
2051 * monitoring the read side of the pipe. If we close them both, epoll_wait
2052 * strangely does not return and could create a endless wait period if the
2053 * pipe is the only tracked fd in the poll set. The thread will take care
2054 * of closing the read side.
2055 */
2056 close(ctx->consumer_metadata_pipe[1]);
2057 if (ret) {
2058 ret = pthread_join(metadata_thread, &status);
2059 if (ret < 0) {
2060 PERROR("pthread_join metadata thread");
2061 }
2062 }
2063
e7b994a3 2064 rcu_unregister_thread();
3bd1e081
MD
2065 return NULL;
2066}
2067
2068/*
2069 * This thread listens on the consumerd socket and receives the file
2070 * descriptors from the session daemon.
2071 */
2072void *lttng_consumer_thread_receive_fds(void *data)
2073{
2074 int sock, client_socket, ret;
2075 /*
2076 * structure to poll for incoming data on communication socket avoids
2077 * making blocking sockets.
2078 */
2079 struct pollfd consumer_sockpoll[2];
2080 struct lttng_consumer_local_data *ctx = data;
2081
e7b994a3
DG
2082 rcu_register_thread();
2083
3bd1e081
MD
2084 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
2085 unlink(ctx->consumer_command_sock_path);
2086 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
2087 if (client_socket < 0) {
2088 ERR("Cannot create command socket");
2089 goto end;
2090 }
2091
2092 ret = lttcomm_listen_unix_sock(client_socket);
2093 if (ret < 0) {
2094 goto end;
2095 }
2096
32258573 2097 DBG("Sending ready command to lttng-sessiond");
f73fabfd 2098 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
3bd1e081
MD
2099 /* return < 0 on error, but == 0 is not fatal */
2100 if (ret < 0) {
32258573 2101 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
2102 goto end;
2103 }
2104
2105 ret = fcntl(client_socket, F_SETFL, O_NONBLOCK);
2106 if (ret < 0) {
7a57cf92 2107 PERROR("fcntl O_NONBLOCK");
3bd1e081
MD
2108 goto end;
2109 }
2110
2111 /* prepare the FDs to poll : to client socket and the should_quit pipe */
2112 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
2113 consumer_sockpoll[0].events = POLLIN | POLLPRI;
2114 consumer_sockpoll[1].fd = client_socket;
2115 consumer_sockpoll[1].events = POLLIN | POLLPRI;
2116
2117 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2118 goto end;
2119 }
2120 DBG("Connection on client_socket");
2121
2122 /* Blocking call, waiting for transmission */
2123 sock = lttcomm_accept_unix_sock(client_socket);
2124 if (sock <= 0) {
2125 WARN("On accept");
2126 goto end;
2127 }
2128 ret = fcntl(sock, F_SETFL, O_NONBLOCK);
2129 if (ret < 0) {
7a57cf92 2130 PERROR("fcntl O_NONBLOCK");
3bd1e081
MD
2131 goto end;
2132 }
2133
2134 /* update the polling structure to poll on the established socket */
2135 consumer_sockpoll[1].fd = sock;
2136 consumer_sockpoll[1].events = POLLIN | POLLPRI;
2137
2138 while (1) {
2139 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2140 goto end;
2141 }
2142 DBG("Incoming command on sock");
2143 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
2144 if (ret == -ENOENT) {
2145 DBG("Received STOP command");
2146 goto end;
2147 }
4cbc1a04
DG
2148 if (ret <= 0) {
2149 /*
2150 * This could simply be a session daemon quitting. Don't output
2151 * ERR() here.
2152 */
2153 DBG("Communication interrupted on command socket");
3bd1e081
MD
2154 goto end;
2155 }
2156 if (consumer_quit) {
2157 DBG("consumer_thread_receive_fds received quit from signal");
2158 goto end;
2159 }
2160 DBG("received fds on sock");
2161 }
2162end:
2163 DBG("consumer_thread_receive_fds exiting");
2164
2165 /*
2166 * when all fds have hung up, the polling thread
2167 * can exit cleanly
2168 */
2169 consumer_quit = 1;
2170
2171 /*
2172 * 2s of grace period, if no polling events occur during
2173 * this period, the polling thread will exit even if there
2174 * are still open FDs (should not happen, but safety mechanism).
2175 */
2176 consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
2177
04fdd819
MD
2178 /*
2179 * Wake-up the other end by writing a null byte in the pipe
2180 * (non-blocking). Important note: Because writing into the
2181 * pipe is non-blocking (and therefore we allow dropping wakeup
2182 * data, as long as there is wakeup data present in the pipe
2183 * buffer to wake up the other end), the other end should
2184 * perform the following sequence for waiting:
2185 * 1) empty the pipe (reads).
2186 * 2) perform update operation.
2187 * 3) wait on the pipe (poll).
2188 */
2189 do {
2190 ret = write(ctx->consumer_poll_pipe[1], "", 1);
6f94560a 2191 } while (ret < 0 && errno == EINTR);
e7b994a3 2192 rcu_unregister_thread();
3bd1e081
MD
2193 return NULL;
2194}
d41f73b7 2195
4078b776 2196ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
2197 struct lttng_consumer_local_data *ctx)
2198{
2199 switch (consumer_data.type) {
2200 case LTTNG_CONSUMER_KERNEL:
2201 return lttng_kconsumer_read_subbuffer(stream, ctx);
7753dea8
MD
2202 case LTTNG_CONSUMER32_UST:
2203 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
2204 return lttng_ustconsumer_read_subbuffer(stream, ctx);
2205 default:
2206 ERR("Unknown consumer_data type");
2207 assert(0);
2208 return -ENOSYS;
2209 }
2210}
2211
2212int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
2213{
2214 switch (consumer_data.type) {
2215 case LTTNG_CONSUMER_KERNEL:
2216 return lttng_kconsumer_on_recv_stream(stream);
7753dea8
MD
2217 case LTTNG_CONSUMER32_UST:
2218 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
2219 return lttng_ustconsumer_on_recv_stream(stream);
2220 default:
2221 ERR("Unknown consumer_data type");
2222 assert(0);
2223 return -ENOSYS;
2224 }
2225}
e4421fec
DG
2226
2227/*
2228 * Allocate and set consumer data hash tables.
2229 */
2230void lttng_consumer_init(void)
2231{
2232 consumer_data.stream_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
2233 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675 2234 consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
e4421fec 2235}
7735ef9e
DG
2236
2237/*
2238 * Process the ADD_RELAYD command receive by a consumer.
2239 *
2240 * This will create a relayd socket pair and add it to the relayd hash table.
2241 * The caller MUST acquire a RCU read side lock before calling it.
2242 */
2243int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
2244 struct lttng_consumer_local_data *ctx, int sock,
2245 struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock)
2246{
2247 int fd, ret = -1;
2248 struct consumer_relayd_sock_pair *relayd;
2249
2250 DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
2251
2252 /* Get relayd reference if exists. */
2253 relayd = consumer_find_relayd(net_seq_idx);
2254 if (relayd == NULL) {
2255 /* Not found. Allocate one. */
2256 relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
2257 if (relayd == NULL) {
f73fabfd 2258 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
7735ef9e
DG
2259 goto error;
2260 }
2261 }
2262
2263 /* Poll on consumer socket. */
2264 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
2265 ret = -EINTR;
2266 goto error;
2267 }
2268
2269 /* Get relayd socket from session daemon */
2270 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
2271 if (ret != sizeof(fd)) {
f73fabfd 2272 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
7735ef9e
DG
2273 ret = -1;
2274 goto error;
2275 }
2276
2277 /* Copy socket information and received FD */
2278 switch (sock_type) {
2279 case LTTNG_STREAM_CONTROL:
2280 /* Copy received lttcomm socket */
2281 lttcomm_copy_sock(&relayd->control_sock, relayd_sock);
2282 ret = lttcomm_create_sock(&relayd->control_sock);
2283 if (ret < 0) {
2284 goto error;
2285 }
2286
2287 /* Close the created socket fd which is useless */
2288 close(relayd->control_sock.fd);
2289
2290 /* Assign new file descriptor */
2291 relayd->control_sock.fd = fd;
2292 break;
2293 case LTTNG_STREAM_DATA:
2294 /* Copy received lttcomm socket */
2295 lttcomm_copy_sock(&relayd->data_sock, relayd_sock);
2296 ret = lttcomm_create_sock(&relayd->data_sock);
2297 if (ret < 0) {
2298 goto error;
2299 }
2300
2301 /* Close the created socket fd which is useless */
2302 close(relayd->data_sock.fd);
2303
2304 /* Assign new file descriptor */
2305 relayd->data_sock.fd = fd;
2306 break;
2307 default:
2308 ERR("Unknown relayd socket type (%d)", sock_type);
2309 goto error;
2310 }
2311
2312 DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
2313 sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
2314 relayd->net_seq_idx, fd);
2315
2316 /*
2317 * Add relayd socket pair to consumer data hashtable. If object already
2318 * exists or on error, the function gracefully returns.
2319 */
d09e1200 2320 add_relayd(relayd);
7735ef9e
DG
2321
2322 /* All good! */
2323 ret = 0;
2324
2325error:
2326 return ret;
2327}
This page took 0.232889 seconds and 4 git commands to generate.