Create userspace buffers using ua_sess effective credentials
[lttng-tools.git] / src / common / consumer / 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
6c1c0768 20#define _LGPL_SOURCE
3bd1e081 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>
331744e3 31#include <signal.h>
3bd1e081 32
51a9e1c7 33#include <bin/lttng-consumerd/health-consumerd.h>
990570ed 34#include <common/common.h>
fb3a43a9 35#include <common/utils.h>
d2956687 36#include <common/time.h>
fb3a43a9 37#include <common/compat/poll.h>
f263b7fd 38#include <common/compat/endian.h>
309167d2 39#include <common/index/index.h>
10a8a223 40#include <common/kernel-ctl/kernel-ctl.h>
00e2e675 41#include <common/sessiond-comm/relayd.h>
10a8a223
DG
42#include <common/sessiond-comm/sessiond-comm.h>
43#include <common/kernel-consumer/kernel-consumer.h>
00e2e675 44#include <common/relayd/relayd.h>
10a8a223 45#include <common/ust-consumer/ust-consumer.h>
c8fea79c
JR
46#include <common/consumer/consumer-timer.h>
47#include <common/consumer/consumer.h>
48#include <common/consumer/consumer-stream.h>
49#include <common/consumer/consumer-testpoint.h>
50#include <common/align.h>
5feafd41 51#include <common/consumer/consumer-metadata-cache.h>
d2956687
JG
52#include <common/trace-chunk.h>
53#include <common/trace-chunk-registry.h>
54#include <common/string-utils/format.h>
3bd1e081
MD
55
56struct lttng_consumer_global_data consumer_data = {
3bd1e081
MD
57 .stream_count = 0,
58 .need_update = 1,
59 .type = LTTNG_CONSUMER_UNKNOWN,
60};
61
d8ef542d
MD
62enum consumer_channel_action {
63 CONSUMER_CHANNEL_ADD,
a0cbdd2e 64 CONSUMER_CHANNEL_DEL,
d8ef542d
MD
65 CONSUMER_CHANNEL_QUIT,
66};
67
68struct consumer_channel_msg {
69 enum consumer_channel_action action;
a0cbdd2e
MD
70 struct lttng_consumer_channel *chan; /* add */
71 uint64_t key; /* del */
d8ef542d
MD
72};
73
80957876 74/* Flag used to temporarily pause data consumption from testpoints. */
cf0bcb51
JG
75int data_consumption_paused;
76
3bd1e081
MD
77/*
78 * Flag to inform the polling thread to quit when all fd hung up. Updated by
79 * the consumer_thread_receive_fds when it notices that all fds has hung up.
80 * Also updated by the signal handler (consumer_should_exit()). Read by the
81 * polling threads.
82 */
10211f5c 83int consumer_quit;
3bd1e081 84
43c34bc3 85/*
43c34bc3
DG
86 * Global hash table containing respectively metadata and data streams. The
87 * stream element in this ht should only be updated by the metadata poll thread
88 * for the metadata and the data poll thread for the data.
89 */
40dc48e0
DG
90static struct lttng_ht *metadata_ht;
91static struct lttng_ht *data_ht;
43c34bc3 92
acdb9057
DG
93/*
94 * Notify a thread lttng pipe to poll back again. This usually means that some
95 * global state has changed so we just send back the thread in a poll wait
96 * call.
97 */
98static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
99{
100 struct lttng_consumer_stream *null_stream = NULL;
101
102 assert(pipe);
103
104 (void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
105}
106
5c635c72
MD
107static void notify_health_quit_pipe(int *pipe)
108{
6cd525e8 109 ssize_t ret;
5c635c72 110
6cd525e8
MD
111 ret = lttng_write(pipe[1], "4", 1);
112 if (ret < 1) {
5c635c72
MD
113 PERROR("write consumer health quit");
114 }
115}
116
d8ef542d
MD
117static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
118 struct lttng_consumer_channel *chan,
a0cbdd2e 119 uint64_t key,
d8ef542d
MD
120 enum consumer_channel_action action)
121{
122 struct consumer_channel_msg msg;
6cd525e8 123 ssize_t ret;
d8ef542d 124
e56251fc
DG
125 memset(&msg, 0, sizeof(msg));
126
d8ef542d
MD
127 msg.action = action;
128 msg.chan = chan;
f21dae48 129 msg.key = key;
6cd525e8
MD
130 ret = lttng_write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
131 if (ret < sizeof(msg)) {
132 PERROR("notify_channel_pipe write error");
133 }
d8ef542d
MD
134}
135
a0cbdd2e
MD
136void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
137 uint64_t key)
138{
139 notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
140}
141
d8ef542d
MD
142static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
143 struct lttng_consumer_channel **chan,
a0cbdd2e 144 uint64_t *key,
d8ef542d
MD
145 enum consumer_channel_action *action)
146{
147 struct consumer_channel_msg msg;
6cd525e8 148 ssize_t ret;
d8ef542d 149
6cd525e8
MD
150 ret = lttng_read(ctx->consumer_channel_pipe[0], &msg, sizeof(msg));
151 if (ret < sizeof(msg)) {
152 ret = -1;
153 goto error;
d8ef542d 154 }
6cd525e8
MD
155 *action = msg.action;
156 *chan = msg.chan;
157 *key = msg.key;
158error:
159 return (int) ret;
d8ef542d
MD
160}
161
212d67a2
DG
162/*
163 * Cleanup the stream list of a channel. Those streams are not yet globally
164 * visible
165 */
166static void clean_channel_stream_list(struct lttng_consumer_channel *channel)
167{
168 struct lttng_consumer_stream *stream, *stmp;
169
170 assert(channel);
171
172 /* Delete streams that might have been left in the stream list. */
173 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
174 send_node) {
175 cds_list_del(&stream->send_node);
176 /*
177 * Once a stream is added to this list, the buffers were created so we
178 * have a guarantee that this call will succeed. Setting the monitor
179 * mode to 0 so we don't lock nor try to delete the stream from the
180 * global hash table.
181 */
182 stream->monitor = 0;
183 consumer_stream_destroy(stream, NULL);
184 }
185}
186
3bd1e081
MD
187/*
188 * Find a stream. The consumer_data.lock must be locked during this
189 * call.
190 */
d88aee68 191static struct lttng_consumer_stream *find_stream(uint64_t key,
8389e4f8 192 struct lttng_ht *ht)
3bd1e081 193{
e4421fec 194 struct lttng_ht_iter iter;
d88aee68 195 struct lttng_ht_node_u64 *node;
e4421fec 196 struct lttng_consumer_stream *stream = NULL;
3bd1e081 197
8389e4f8
DG
198 assert(ht);
199
d88aee68
DG
200 /* -1ULL keys are lookup failures */
201 if (key == (uint64_t) -1ULL) {
7ad0a0cb 202 return NULL;
7a57cf92 203 }
e4421fec 204
6065ceec
DG
205 rcu_read_lock();
206
d88aee68
DG
207 lttng_ht_lookup(ht, &key, &iter);
208 node = lttng_ht_iter_get_node_u64(&iter);
e4421fec
DG
209 if (node != NULL) {
210 stream = caa_container_of(node, struct lttng_consumer_stream, node);
3bd1e081 211 }
e4421fec 212
6065ceec
DG
213 rcu_read_unlock();
214
e4421fec 215 return stream;
3bd1e081
MD
216}
217
da009f2c 218static void steal_stream_key(uint64_t key, struct lttng_ht *ht)
7ad0a0cb
MD
219{
220 struct lttng_consumer_stream *stream;
221
04253271 222 rcu_read_lock();
ffe60014 223 stream = find_stream(key, ht);
04253271 224 if (stream) {
da009f2c 225 stream->key = (uint64_t) -1ULL;
04253271
MD
226 /*
227 * We don't want the lookup to match, but we still need
228 * to iterate on this stream when iterating over the hash table. Just
229 * change the node key.
230 */
da009f2c 231 stream->node.key = (uint64_t) -1ULL;
04253271
MD
232 }
233 rcu_read_unlock();
7ad0a0cb
MD
234}
235
d56db448
DG
236/*
237 * Return a channel object for the given key.
238 *
239 * RCU read side lock MUST be acquired before calling this function and
240 * protects the channel ptr.
241 */
d88aee68 242struct lttng_consumer_channel *consumer_find_channel(uint64_t key)
3bd1e081 243{
e4421fec 244 struct lttng_ht_iter iter;
d88aee68 245 struct lttng_ht_node_u64 *node;
e4421fec 246 struct lttng_consumer_channel *channel = NULL;
3bd1e081 247
d88aee68
DG
248 /* -1ULL keys are lookup failures */
249 if (key == (uint64_t) -1ULL) {
7ad0a0cb 250 return NULL;
7a57cf92 251 }
e4421fec 252
d88aee68
DG
253 lttng_ht_lookup(consumer_data.channel_ht, &key, &iter);
254 node = lttng_ht_iter_get_node_u64(&iter);
e4421fec
DG
255 if (node != NULL) {
256 channel = caa_container_of(node, struct lttng_consumer_channel, node);
3bd1e081 257 }
e4421fec
DG
258
259 return channel;
3bd1e081
MD
260}
261
b5a6470f
DG
262/*
263 * There is a possibility that the consumer does not have enough time between
264 * the close of the channel on the session daemon and the cleanup in here thus
265 * once we have a channel add with an existing key, we know for sure that this
266 * channel will eventually get cleaned up by all streams being closed.
267 *
268 * This function just nullifies the already existing channel key.
269 */
270static void steal_channel_key(uint64_t key)
271{
272 struct lttng_consumer_channel *channel;
273
274 rcu_read_lock();
275 channel = consumer_find_channel(key);
276 if (channel) {
277 channel->key = (uint64_t) -1ULL;
278 /*
279 * We don't want the lookup to match, but we still need to iterate on
280 * this channel when iterating over the hash table. Just change the
281 * node key.
282 */
283 channel->node.key = (uint64_t) -1ULL;
284 }
285 rcu_read_unlock();
286}
287
ffe60014 288static void free_channel_rcu(struct rcu_head *head)
702b1ea4 289{
d88aee68
DG
290 struct lttng_ht_node_u64 *node =
291 caa_container_of(head, struct lttng_ht_node_u64, head);
ffe60014
DG
292 struct lttng_consumer_channel *channel =
293 caa_container_of(node, struct lttng_consumer_channel, node);
702b1ea4 294
b83e03c4
MD
295 switch (consumer_data.type) {
296 case LTTNG_CONSUMER_KERNEL:
297 break;
298 case LTTNG_CONSUMER32_UST:
299 case LTTNG_CONSUMER64_UST:
300 lttng_ustconsumer_free_channel(channel);
301 break;
302 default:
303 ERR("Unknown consumer_data type");
304 abort();
305 }
ffe60014 306 free(channel);
702b1ea4
MD
307}
308
00e2e675
DG
309/*
310 * RCU protected relayd socket pair free.
311 */
ffe60014 312static void free_relayd_rcu(struct rcu_head *head)
00e2e675 313{
d88aee68
DG
314 struct lttng_ht_node_u64 *node =
315 caa_container_of(head, struct lttng_ht_node_u64, head);
00e2e675
DG
316 struct consumer_relayd_sock_pair *relayd =
317 caa_container_of(node, struct consumer_relayd_sock_pair, node);
318
8994307f
DG
319 /*
320 * Close all sockets. This is done in the call RCU since we don't want the
321 * socket fds to be reassigned thus potentially creating bad state of the
322 * relayd object.
323 *
324 * We do not have to lock the control socket mutex here since at this stage
325 * there is no one referencing to this relayd object.
326 */
327 (void) relayd_close(&relayd->control_sock);
328 (void) relayd_close(&relayd->data_sock);
329
3a84e2f3 330 pthread_mutex_destroy(&relayd->ctrl_sock_mutex);
00e2e675
DG
331 free(relayd);
332}
333
334/*
335 * Destroy and free relayd socket pair object.
00e2e675 336 */
51230d70 337void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
338{
339 int ret;
340 struct lttng_ht_iter iter;
341
173af62f
DG
342 if (relayd == NULL) {
343 return;
344 }
345
00e2e675
DG
346 DBG("Consumer destroy and close relayd socket pair");
347
348 iter.iter.node = &relayd->node.node;
349 ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
173af62f 350 if (ret != 0) {
8994307f 351 /* We assume the relayd is being or is destroyed */
173af62f
DG
352 return;
353 }
00e2e675 354
00e2e675 355 /* RCU free() call */
ffe60014
DG
356 call_rcu(&relayd->node.head, free_relayd_rcu);
357}
358
359/*
360 * Remove a channel from the global list protected by a mutex. This function is
361 * also responsible for freeing its data structures.
362 */
363void consumer_del_channel(struct lttng_consumer_channel *channel)
364{
ffe60014
DG
365 struct lttng_ht_iter iter;
366
d88aee68 367 DBG("Consumer delete channel key %" PRIu64, channel->key);
ffe60014
DG
368
369 pthread_mutex_lock(&consumer_data.lock);
a9838785 370 pthread_mutex_lock(&channel->lock);
ffe60014 371
212d67a2
DG
372 /* Destroy streams that might have been left in the stream list. */
373 clean_channel_stream_list(channel);
51e762e5 374
d3e2ba59
JD
375 if (channel->live_timer_enabled == 1) {
376 consumer_timer_live_stop(channel);
377 }
e9404c27
JG
378 if (channel->monitor_timer_enabled == 1) {
379 consumer_timer_monitor_stop(channel);
380 }
d3e2ba59 381
ffe60014
DG
382 switch (consumer_data.type) {
383 case LTTNG_CONSUMER_KERNEL:
384 break;
385 case LTTNG_CONSUMER32_UST:
386 case LTTNG_CONSUMER64_UST:
387 lttng_ustconsumer_del_channel(channel);
388 break;
389 default:
390 ERR("Unknown consumer_data type");
391 assert(0);
392 goto end;
393 }
394
d2956687
JG
395 lttng_trace_chunk_put(channel->trace_chunk);
396 channel->trace_chunk = NULL;
5c3892a6 397
d2956687
JG
398 if (channel->is_published) {
399 int ret;
400
401 rcu_read_lock();
402 iter.iter.node = &channel->node.node;
403 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
404 assert(!ret);
ffe60014 405
d2956687
JG
406 iter.iter.node = &channel->channels_by_session_id_ht_node.node;
407 ret = lttng_ht_del(consumer_data.channels_by_session_id_ht,
408 &iter);
409 assert(!ret);
410 rcu_read_unlock();
411 }
412
413 call_rcu(&channel->node.head, free_channel_rcu);
ffe60014 414end:
a9838785 415 pthread_mutex_unlock(&channel->lock);
ffe60014 416 pthread_mutex_unlock(&consumer_data.lock);
00e2e675
DG
417}
418
228b5bf7
DG
419/*
420 * Iterate over the relayd hash table and destroy each element. Finally,
421 * destroy the whole hash table.
422 */
423static void cleanup_relayd_ht(void)
424{
425 struct lttng_ht_iter iter;
426 struct consumer_relayd_sock_pair *relayd;
427
428 rcu_read_lock();
429
430 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
431 node.node) {
51230d70 432 consumer_destroy_relayd(relayd);
228b5bf7
DG
433 }
434
228b5bf7 435 rcu_read_unlock();
36b588ed
MD
436
437 lttng_ht_destroy(consumer_data.relayd_ht);
228b5bf7
DG
438}
439
8994307f
DG
440/*
441 * Update the end point status of all streams having the given network sequence
442 * index (relayd index).
443 *
444 * It's atomically set without having the stream mutex locked which is fine
445 * because we handle the write/read race with a pipe wakeup for each thread.
446 */
da009f2c 447static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
8994307f
DG
448 enum consumer_endpoint_status status)
449{
450 struct lttng_ht_iter iter;
451 struct lttng_consumer_stream *stream;
452
da009f2c 453 DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
8994307f
DG
454
455 rcu_read_lock();
456
457 /* Let's begin with metadata */
458 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
459 if (stream->net_seq_idx == net_seq_idx) {
460 uatomic_set(&stream->endpoint_status, status);
461 DBG("Delete flag set to metadata stream %d", stream->wait_fd);
462 }
463 }
464
465 /* Follow up by the data streams */
466 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
467 if (stream->net_seq_idx == net_seq_idx) {
468 uatomic_set(&stream->endpoint_status, status);
469 DBG("Delete flag set to data stream %d", stream->wait_fd);
470 }
471 }
472 rcu_read_unlock();
473}
474
475/*
476 * Cleanup a relayd object by flagging every associated streams for deletion,
477 * destroying the object meaning removing it from the relayd hash table,
478 * closing the sockets and freeing the memory in a RCU call.
479 *
480 * If a local data context is available, notify the threads that the streams'
481 * state have changed.
482 */
9276e5c8 483void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
8994307f 484{
da009f2c 485 uint64_t netidx;
8994307f
DG
486
487 assert(relayd);
488
9276e5c8 489 DBG("Cleaning up relayd object ID %"PRIu64, relayd->net_seq_idx);
9617607b 490
8994307f
DG
491 /* Save the net sequence index before destroying the object */
492 netidx = relayd->net_seq_idx;
493
494 /*
495 * Delete the relayd from the relayd hash table, close the sockets and free
496 * the object in a RCU call.
497 */
51230d70 498 consumer_destroy_relayd(relayd);
8994307f
DG
499
500 /* Set inactive endpoint to all streams */
501 update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE);
502
503 /*
504 * With a local data context, notify the threads that the streams' state
505 * have changed. The write() action on the pipe acts as an "implicit"
506 * memory barrier ordering the updates of the end point status from the
507 * read of this status which happens AFTER receiving this notify.
508 */
9276e5c8
JR
509 notify_thread_lttng_pipe(relayd->ctx->consumer_data_pipe);
510 notify_thread_lttng_pipe(relayd->ctx->consumer_metadata_pipe);
8994307f
DG
511}
512
a6ba4fe1
DG
513/*
514 * Flag a relayd socket pair for destruction. Destroy it if the refcount
515 * reaches zero.
516 *
517 * RCU read side lock MUST be aquired before calling this function.
518 */
519void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
520{
521 assert(relayd);
522
523 /* Set destroy flag for this object */
524 uatomic_set(&relayd->destroy_flag, 1);
525
526 /* Destroy the relayd if refcount is 0 */
527 if (uatomic_read(&relayd->refcount) == 0) {
51230d70 528 consumer_destroy_relayd(relayd);
a6ba4fe1
DG
529 }
530}
531
3bd1e081 532/*
1d1a276c
DG
533 * Completly destroy stream from every visiable data structure and the given
534 * hash table if one.
535 *
536 * One this call returns, the stream object is not longer usable nor visible.
3bd1e081 537 */
e316aad5
DG
538void consumer_del_stream(struct lttng_consumer_stream *stream,
539 struct lttng_ht *ht)
3bd1e081 540{
1d1a276c 541 consumer_stream_destroy(stream, ht);
3bd1e081
MD
542}
543
5ab66908
MD
544/*
545 * XXX naming of del vs destroy is all mixed up.
546 */
547void consumer_del_stream_for_data(struct lttng_consumer_stream *stream)
548{
549 consumer_stream_destroy(stream, data_ht);
550}
551
552void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
553{
554 consumer_stream_destroy(stream, metadata_ht);
555}
556
d9a2e16e
JD
557void consumer_stream_update_channel_attributes(
558 struct lttng_consumer_stream *stream,
559 struct lttng_consumer_channel *channel)
560{
561 stream->channel_read_only_attributes.tracefile_size =
562 channel->tracefile_size;
d9a2e16e
JD
563}
564
d88aee68
DG
565struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
566 uint64_t stream_key,
ffe60014 567 const char *channel_name,
57a269f2 568 uint64_t relayd_id,
53632229 569 uint64_t session_id,
d2956687 570 struct lttng_trace_chunk *trace_chunk,
ffe60014
DG
571 int cpu,
572 int *alloc_ret,
4891ece8 573 enum consumer_channel_type type,
d2956687 574 unsigned int monitor)
3bd1e081 575{
ffe60014 576 int ret;
3bd1e081 577 struct lttng_consumer_stream *stream;
3bd1e081 578
effcf122 579 stream = zmalloc(sizeof(*stream));
3bd1e081 580 if (stream == NULL) {
7a57cf92 581 PERROR("malloc struct lttng_consumer_stream");
ffe60014 582 ret = -ENOMEM;
7a57cf92 583 goto end;
3bd1e081 584 }
7a57cf92 585
d2956687
JG
586 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
587 ERR("Failed to acquire trace chunk reference during the creation of a stream");
588 ret = -1;
589 goto error;
590 }
d56db448 591
d2956687 592 rcu_read_lock();
3bd1e081 593 stream->key = stream_key;
d2956687 594 stream->trace_chunk = trace_chunk;
3bd1e081
MD
595 stream->out_fd = -1;
596 stream->out_fd_offset = 0;
e5d1a9b3 597 stream->output_written = 0;
ffe60014 598 stream->net_seq_idx = relayd_id;
53632229 599 stream->session_id = session_id;
4891ece8 600 stream->monitor = monitor;
774d490c 601 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
f8f3885c 602 stream->index_file = NULL;
fb83fe64 603 stream->last_sequence_number = -1ULL;
53632229 604 pthread_mutex_init(&stream->lock, NULL);
c585821b 605 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
58b1f425 606
ffe60014
DG
607 /* If channel is the metadata, flag this stream as metadata. */
608 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
609 stream->metadata_flag = 1;
610 /* Metadata is flat out. */
611 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
94d49140
JD
612 /* Live rendez-vous point. */
613 pthread_cond_init(&stream->metadata_rdv, NULL);
614 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
58b1f425 615 } else {
ffe60014
DG
616 /* Format stream name to <channel_name>_<cpu_number> */
617 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
618 channel_name, cpu);
619 if (ret < 0) {
620 PERROR("snprintf stream name");
621 goto error;
622 }
58b1f425 623 }
c30aaa51 624
ffe60014 625 /* Key is always the wait_fd for streams. */
d88aee68 626 lttng_ht_node_init_u64(&stream->node, stream->key);
ffe60014 627
d8ef542d
MD
628 /* Init node per channel id key */
629 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
630
53632229 631 /* Init session id node with the stream session id */
d88aee68 632 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
53632229 633
07b86b52
JD
634 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
635 " relayd_id %" PRIu64 ", session_id %" PRIu64,
636 stream->name, stream->key, channel_key,
637 stream->net_seq_idx, stream->session_id);
d56db448
DG
638
639 rcu_read_unlock();
3bd1e081 640 return stream;
c80048c6
MD
641
642error:
d56db448 643 rcu_read_unlock();
d2956687 644 lttng_trace_chunk_put(stream->trace_chunk);
c80048c6 645 free(stream);
7a57cf92 646end:
ffe60014
DG
647 if (alloc_ret) {
648 *alloc_ret = ret;
649 }
c80048c6 650 return NULL;
3bd1e081
MD
651}
652
653/*
654 * Add a stream to the global list protected by a mutex.
655 */
66d583dc 656void consumer_add_data_stream(struct lttng_consumer_stream *stream)
3bd1e081 657{
5ab66908 658 struct lttng_ht *ht = data_ht;
3bd1e081 659
e316aad5 660 assert(stream);
43c34bc3 661 assert(ht);
c77fc10a 662
d88aee68 663 DBG3("Adding consumer stream %" PRIu64, stream->key);
e316aad5
DG
664
665 pthread_mutex_lock(&consumer_data.lock);
a9838785 666 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 667 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 668 pthread_mutex_lock(&stream->lock);
b0b335c8 669 rcu_read_lock();
e316aad5 670
43c34bc3 671 /* Steal stream identifier to avoid having streams with the same key */
ffe60014 672 steal_stream_key(stream->key, ht);
43c34bc3 673
d88aee68 674 lttng_ht_add_unique_u64(ht, &stream->node);
00e2e675 675
d8ef542d
MD
676 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
677 &stream->node_channel_id);
678
ca22feea
DG
679 /*
680 * Add stream to the stream_list_ht of the consumer data. No need to steal
681 * the key since the HT does not use it and we allow to add redundant keys
682 * into this table.
683 */
d88aee68 684 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 685
e316aad5 686 /*
ffe60014
DG
687 * When nb_init_stream_left reaches 0, we don't need to trigger any action
688 * in terms of destroying the associated channel, because the action that
e316aad5
DG
689 * causes the count to become 0 also causes a stream to be added. The
690 * channel deletion will thus be triggered by the following removal of this
691 * stream.
692 */
ffe60014 693 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
694 /* Increment refcount before decrementing nb_init_stream_left */
695 cmm_smp_wmb();
ffe60014 696 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
697 }
698
699 /* Update consumer data once the node is inserted. */
3bd1e081
MD
700 consumer_data.stream_count++;
701 consumer_data.need_update = 1;
702
e316aad5 703 rcu_read_unlock();
2e818a6a 704 pthread_mutex_unlock(&stream->lock);
ec6ea7d0 705 pthread_mutex_unlock(&stream->chan->timer_lock);
a9838785 706 pthread_mutex_unlock(&stream->chan->lock);
3bd1e081 707 pthread_mutex_unlock(&consumer_data.lock);
3bd1e081
MD
708}
709
5ab66908
MD
710void consumer_del_data_stream(struct lttng_consumer_stream *stream)
711{
712 consumer_del_stream(stream, data_ht);
713}
714
00e2e675 715/*
3f8e211f
DG
716 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
717 * be acquired before calling this.
00e2e675 718 */
d09e1200 719static int add_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
720{
721 int ret = 0;
d88aee68 722 struct lttng_ht_node_u64 *node;
00e2e675
DG
723 struct lttng_ht_iter iter;
724
ffe60014 725 assert(relayd);
00e2e675 726
00e2e675 727 lttng_ht_lookup(consumer_data.relayd_ht,
d88aee68
DG
728 &relayd->net_seq_idx, &iter);
729 node = lttng_ht_iter_get_node_u64(&iter);
00e2e675 730 if (node != NULL) {
00e2e675
DG
731 goto end;
732 }
d88aee68 733 lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
00e2e675 734
00e2e675
DG
735end:
736 return ret;
737}
738
739/*
740 * Allocate and return a consumer relayd socket.
741 */
027a694f 742static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
da009f2c 743 uint64_t net_seq_idx)
00e2e675
DG
744{
745 struct consumer_relayd_sock_pair *obj = NULL;
746
da009f2c
MD
747 /* net sequence index of -1 is a failure */
748 if (net_seq_idx == (uint64_t) -1ULL) {
00e2e675
DG
749 goto error;
750 }
751
752 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
753 if (obj == NULL) {
754 PERROR("zmalloc relayd sock");
755 goto error;
756 }
757
758 obj->net_seq_idx = net_seq_idx;
759 obj->refcount = 0;
173af62f 760 obj->destroy_flag = 0;
f96e4545
MD
761 obj->control_sock.sock.fd = -1;
762 obj->data_sock.sock.fd = -1;
d88aee68 763 lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
00e2e675
DG
764 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
765
766error:
767 return obj;
768}
769
770/*
771 * Find a relayd socket pair in the global consumer data.
772 *
773 * Return the object if found else NULL.
b0b335c8
MD
774 * RCU read-side lock must be held across this call and while using the
775 * returned object.
00e2e675 776 */
d88aee68 777struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
00e2e675
DG
778{
779 struct lttng_ht_iter iter;
d88aee68 780 struct lttng_ht_node_u64 *node;
00e2e675
DG
781 struct consumer_relayd_sock_pair *relayd = NULL;
782
783 /* Negative keys are lookup failures */
d88aee68 784 if (key == (uint64_t) -1ULL) {
00e2e675
DG
785 goto error;
786 }
787
d88aee68 788 lttng_ht_lookup(consumer_data.relayd_ht, &key,
00e2e675 789 &iter);
d88aee68 790 node = lttng_ht_iter_get_node_u64(&iter);
00e2e675
DG
791 if (node != NULL) {
792 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
793 }
794
00e2e675
DG
795error:
796 return relayd;
797}
798
10a50311
JD
799/*
800 * Find a relayd and send the stream
801 *
802 * Returns 0 on success, < 0 on error
803 */
804int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
805 char *path)
806{
807 int ret = 0;
808 struct consumer_relayd_sock_pair *relayd;
809
810 assert(stream);
811 assert(stream->net_seq_idx != -1ULL);
812 assert(path);
813
814 /* The stream is not metadata. Get relayd reference if exists. */
815 rcu_read_lock();
816 relayd = consumer_find_relayd(stream->net_seq_idx);
817 if (relayd != NULL) {
818 /* Add stream on the relayd */
819 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
820 ret = relayd_add_stream(&relayd->control_sock, stream->name,
821 path, &stream->relayd_stream_id,
d2956687
JG
822 stream->chan->tracefile_size,
823 stream->chan->tracefile_count,
824 stream->trace_chunk);
10a50311
JD
825 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
826 if (ret < 0) {
9276e5c8
JR
827 ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
828 lttng_consumer_cleanup_relayd(relayd);
10a50311
JD
829 goto end;
830 }
1c20f0e2 831
10a50311 832 uatomic_inc(&relayd->refcount);
d01178b6 833 stream->sent_to_relayd = 1;
10a50311
JD
834 } else {
835 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
836 stream->key, stream->net_seq_idx);
837 ret = -1;
838 goto end;
839 }
840
841 DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
842 stream->name, stream->key, stream->net_seq_idx);
843
844end:
845 rcu_read_unlock();
846 return ret;
847}
848
a4baae1b
JD
849/*
850 * Find a relayd and send the streams sent message
851 *
852 * Returns 0 on success, < 0 on error
853 */
854int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
855{
856 int ret = 0;
857 struct consumer_relayd_sock_pair *relayd;
858
859 assert(net_seq_idx != -1ULL);
860
861 /* The stream is not metadata. Get relayd reference if exists. */
862 rcu_read_lock();
863 relayd = consumer_find_relayd(net_seq_idx);
864 if (relayd != NULL) {
865 /* Add stream on the relayd */
866 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
867 ret = relayd_streams_sent(&relayd->control_sock);
868 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
869 if (ret < 0) {
9276e5c8
JR
870 ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
871 lttng_consumer_cleanup_relayd(relayd);
a4baae1b
JD
872 goto end;
873 }
874 } else {
875 ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
876 net_seq_idx);
877 ret = -1;
878 goto end;
879 }
880
881 ret = 0;
882 DBG("All streams sent relayd id %" PRIu64, net_seq_idx);
883
884end:
885 rcu_read_unlock();
886 return ret;
887}
888
10a50311
JD
889/*
890 * Find a relayd and close the stream
891 */
892void close_relayd_stream(struct lttng_consumer_stream *stream)
893{
894 struct consumer_relayd_sock_pair *relayd;
895
896 /* The stream is not metadata. Get relayd reference if exists. */
897 rcu_read_lock();
898 relayd = consumer_find_relayd(stream->net_seq_idx);
899 if (relayd) {
900 consumer_stream_relayd_close(stream, relayd);
901 }
902 rcu_read_unlock();
903}
904
00e2e675
DG
905/*
906 * Handle stream for relayd transmission if the stream applies for network
907 * streaming where the net sequence index is set.
908 *
909 * Return destination file descriptor or negative value on error.
910 */
6197aea7 911static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
1d4dfdef
DG
912 size_t data_size, unsigned long padding,
913 struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
914{
915 int outfd = -1, ret;
00e2e675
DG
916 struct lttcomm_relayd_data_hdr data_hdr;
917
918 /* Safety net */
919 assert(stream);
6197aea7 920 assert(relayd);
00e2e675
DG
921
922 /* Reset data header */
923 memset(&data_hdr, 0, sizeof(data_hdr));
924
00e2e675
DG
925 if (stream->metadata_flag) {
926 /* Caller MUST acquire the relayd control socket lock */
927 ret = relayd_send_metadata(&relayd->control_sock, data_size);
928 if (ret < 0) {
929 goto error;
930 }
931
932 /* Metadata are always sent on the control socket. */
6151a90f 933 outfd = relayd->control_sock.sock.fd;
00e2e675
DG
934 } else {
935 /* Set header with stream information */
936 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
937 data_hdr.data_size = htobe32(data_size);
1d4dfdef 938 data_hdr.padding_size = htobe32(padding);
39df6d9f
DG
939 /*
940 * Note that net_seq_num below is assigned with the *current* value of
941 * next_net_seq_num and only after that the next_net_seq_num will be
942 * increment. This is why when issuing a command on the relayd using
943 * this next value, 1 should always be substracted in order to compare
944 * the last seen sequence number on the relayd side to the last sent.
945 */
3604f373 946 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
00e2e675
DG
947 /* Other fields are zeroed previously */
948
949 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
950 sizeof(data_hdr));
951 if (ret < 0) {
952 goto error;
953 }
954
3604f373
DG
955 ++stream->next_net_seq_num;
956
00e2e675 957 /* Set to go on data socket */
6151a90f 958 outfd = relayd->data_sock.sock.fd;
00e2e675
DG
959 }
960
961error:
962 return outfd;
963}
964
d2956687
JG
965/*
966 * Trigger a dump of the metadata content. Following/during the succesful
967 * completion of this call, the metadata poll thread will start receiving
968 * metadata packets to consume.
969 *
970 * The caller must hold the channel and stream locks.
971 */
972static
973int consumer_metadata_stream_dump(struct lttng_consumer_stream *stream)
974{
975 int ret;
976
977 ASSERT_LOCKED(stream->chan->lock);
978 ASSERT_LOCKED(stream->lock);
979 assert(stream->metadata_flag);
980 assert(stream->chan->trace_chunk);
981
982 switch (consumer_data.type) {
983 case LTTNG_CONSUMER_KERNEL:
984 /*
985 * Reset the position of what has been read from the
986 * metadata cache to 0 so we can dump it again.
987 */
988 ret = kernctl_metadata_cache_dump(stream->wait_fd);
989 break;
990 case LTTNG_CONSUMER32_UST:
991 case LTTNG_CONSUMER64_UST:
992 /*
993 * Reset the position pushed from the metadata cache so it
994 * will write from the beginning on the next push.
995 */
996 stream->ust_metadata_pushed = 0;
997 ret = consumer_metadata_wakeup_pipe(stream->chan);
998 break;
999 default:
1000 ERR("Unknown consumer_data type");
1001 abort();
1002 }
1003 if (ret < 0) {
1004 ERR("Failed to dump the metadata cache");
1005 }
1006 return ret;
1007}
1008
1009static
1010int lttng_consumer_channel_set_trace_chunk(
1011 struct lttng_consumer_channel *channel,
1012 struct lttng_trace_chunk *new_trace_chunk)
1013{
1014 int ret = 0;
1015 const bool is_local_trace = channel->relayd_id == -1ULL;
1016 bool update_stream_trace_chunk;
1017 struct cds_lfht_iter iter;
1018 struct lttng_consumer_stream *stream;
1019 unsigned long channel_hash;
1020
1021 pthread_mutex_lock(&channel->lock);
1022 /*
1023 * A stream can transition to a state where it and its channel
1024 * no longer belong to a trace chunk. For instance, this happens when
1025 * a session is rotated while it is inactive. After the rotation
1026 * of an inactive session completes, the channel and its streams no
1027 * longer belong to a trace chunk.
1028 *
1029 * However, if a session is stopped, rotated, and started again,
1030 * the session daemon will create a new chunk and send it to its peers.
1031 * In that case, the streams' transition to a new chunk can be performed
1032 * immediately.
1033 *
1034 * This trace chunk transition could also be performed lazily when
1035 * a buffer is consumed. However, creating the files here allows the
1036 * consumer daemon to report any creation error to the session daemon
1037 * and cause the start of the tracing session to fail.
1038 */
1039 update_stream_trace_chunk = !channel->trace_chunk && new_trace_chunk;
1040
1041 /*
1042 * The acquisition of the reference cannot fail (barring
1043 * a severe internal error) since a reference to the published
1044 * chunk is already held by the caller.
1045 */
1046 if (new_trace_chunk) {
1047 const bool acquired_reference = lttng_trace_chunk_get(
1048 new_trace_chunk);
1049
1050 assert(acquired_reference);
1051 }
1052
1053 lttng_trace_chunk_put(channel->trace_chunk);
1054 channel->trace_chunk = new_trace_chunk;
1055 if (!is_local_trace || !new_trace_chunk) {
1056 /* Not an error. */
1057 goto end;
1058 }
1059
1060 if (!update_stream_trace_chunk) {
1061 goto end;
1062 }
1063
1064 channel_hash = consumer_data.stream_per_chan_id_ht->hash_fct(
1065 &channel->key, lttng_ht_seed);
1066 rcu_read_lock();
1067 cds_lfht_for_each_entry_duplicate(consumer_data.stream_per_chan_id_ht->ht,
1068 channel_hash,
1069 consumer_data.stream_per_chan_id_ht->match_fct,
1070 &channel->key, &iter, stream, node_channel_id.node) {
1071 bool acquired_reference, should_regenerate_metadata = false;
1072
1073 acquired_reference = lttng_trace_chunk_get(channel->trace_chunk);
1074 assert(acquired_reference);
1075
1076 pthread_mutex_lock(&stream->lock);
1077
1078 /*
1079 * On a transition from "no-chunk" to a new chunk, a metadata
1080 * stream's content must be entirely dumped. This must occcur
1081 * _after_ the creation of the metadata stream's output files
1082 * as the consumption thread (not necessarily the one executing
1083 * this) may start to consume during the call to
1084 * consumer_metadata_stream_dump().
1085 */
1086 should_regenerate_metadata =
1087 stream->metadata_flag &&
1088 !stream->trace_chunk && channel->trace_chunk;
1089 stream->trace_chunk = channel->trace_chunk;
1090 ret = consumer_stream_create_output_files(stream, true);
1091 if (ret) {
1092 pthread_mutex_unlock(&stream->lock);
1093 goto end_rcu_unlock;
1094 }
1095 if (should_regenerate_metadata) {
1096 ret = consumer_metadata_stream_dump(stream);
1097 }
1098 pthread_mutex_unlock(&stream->lock);
1099 if (ret) {
1100 goto end_rcu_unlock;
1101 }
1102 }
1103end_rcu_unlock:
1104 rcu_read_unlock();
1105end:
1106 pthread_mutex_unlock(&channel->lock);
1107 return ret;
1108}
1109
3bd1e081 1110/*
ffe60014
DG
1111 * Allocate and return a new lttng_consumer_channel object using the given key
1112 * to initialize the hash table node.
1113 *
1114 * On error, return NULL.
3bd1e081 1115 */
886224ff 1116struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
ffe60014 1117 uint64_t session_id,
d2956687 1118 const uint64_t *chunk_id,
ffe60014
DG
1119 const char *pathname,
1120 const char *name,
57a269f2 1121 uint64_t relayd_id,
1624d5b7
JD
1122 enum lttng_event_output output,
1123 uint64_t tracefile_size,
2bba9e53 1124 uint64_t tracefile_count,
1950109e 1125 uint64_t session_id_per_pid,
ecc48a90 1126 unsigned int monitor,
d7ba1388 1127 unsigned int live_timer_interval,
3d071855 1128 const char *root_shm_path,
d7ba1388 1129 const char *shm_path)
3bd1e081 1130{
d2956687
JG
1131 struct lttng_consumer_channel *channel = NULL;
1132 struct lttng_trace_chunk *trace_chunk = NULL;
1133
1134 if (chunk_id) {
1135 trace_chunk = lttng_trace_chunk_registry_find_chunk(
1136 consumer_data.chunk_registry, session_id,
1137 *chunk_id);
1138 if (!trace_chunk) {
1139 ERR("Failed to find trace chunk reference during creation of channel");
1140 goto end;
1141 }
1142 }
3bd1e081 1143
276b26d1 1144 channel = zmalloc(sizeof(*channel));
3bd1e081 1145 if (channel == NULL) {
7a57cf92 1146 PERROR("malloc struct lttng_consumer_channel");
3bd1e081
MD
1147 goto end;
1148 }
ffe60014
DG
1149
1150 channel->key = key;
3bd1e081 1151 channel->refcount = 0;
ffe60014 1152 channel->session_id = session_id;
1950109e 1153 channel->session_id_per_pid = session_id_per_pid;
ffe60014 1154 channel->relayd_id = relayd_id;
1624d5b7
JD
1155 channel->tracefile_size = tracefile_size;
1156 channel->tracefile_count = tracefile_count;
2bba9e53 1157 channel->monitor = monitor;
ecc48a90 1158 channel->live_timer_interval = live_timer_interval;
a9838785 1159 pthread_mutex_init(&channel->lock, NULL);
ec6ea7d0 1160 pthread_mutex_init(&channel->timer_lock, NULL);
ffe60014 1161
0c759fc9
DG
1162 switch (output) {
1163 case LTTNG_EVENT_SPLICE:
1164 channel->output = CONSUMER_CHANNEL_SPLICE;
1165 break;
1166 case LTTNG_EVENT_MMAP:
1167 channel->output = CONSUMER_CHANNEL_MMAP;
1168 break;
1169 default:
1170 assert(0);
1171 free(channel);
1172 channel = NULL;
1173 goto end;
1174 }
1175
07b86b52
JD
1176 /*
1177 * In monitor mode, the streams associated with the channel will be put in
1178 * a special list ONLY owned by this channel. So, the refcount is set to 1
1179 * here meaning that the channel itself has streams that are referenced.
1180 *
1181 * On a channel deletion, once the channel is no longer visible, the
1182 * refcount is decremented and checked for a zero value to delete it. With
1183 * streams in no monitor mode, it will now be safe to destroy the channel.
1184 */
1185 if (!channel->monitor) {
1186 channel->refcount = 1;
1187 }
1188
ffe60014
DG
1189 strncpy(channel->pathname, pathname, sizeof(channel->pathname));
1190 channel->pathname[sizeof(channel->pathname) - 1] = '\0';
1191
1192 strncpy(channel->name, name, sizeof(channel->name));
1193 channel->name[sizeof(channel->name) - 1] = '\0';
1194
3d071855
MD
1195 if (root_shm_path) {
1196 strncpy(channel->root_shm_path, root_shm_path, sizeof(channel->root_shm_path));
1197 channel->root_shm_path[sizeof(channel->root_shm_path) - 1] = '\0';
1198 }
d7ba1388
MD
1199 if (shm_path) {
1200 strncpy(channel->shm_path, shm_path, sizeof(channel->shm_path));
1201 channel->shm_path[sizeof(channel->shm_path) - 1] = '\0';
1202 }
1203
d88aee68 1204 lttng_ht_node_init_u64(&channel->node, channel->key);
5c3892a6
JG
1205 lttng_ht_node_init_u64(&channel->channels_by_session_id_ht_node,
1206 channel->session_id);
d8ef542d
MD
1207
1208 channel->wait_fd = -1;
ffe60014
DG
1209 CDS_INIT_LIST_HEAD(&channel->streams.head);
1210
d2956687
JG
1211 if (trace_chunk) {
1212 int ret = lttng_consumer_channel_set_trace_chunk(channel,
1213 trace_chunk);
1214 if (ret) {
1215 goto error;
1216 }
1217 }
1218
62a7b8ed 1219 DBG("Allocated channel (key %" PRIu64 ")", channel->key);
3bd1e081 1220
3bd1e081 1221end:
d2956687 1222 lttng_trace_chunk_put(trace_chunk);
3bd1e081 1223 return channel;
d2956687
JG
1224error:
1225 consumer_del_channel(channel);
1226 channel = NULL;
1227 goto end;
3bd1e081
MD
1228}
1229
1230/*
1231 * Add a channel to the global list protected by a mutex.
821fffb2 1232 *
b5a6470f 1233 * Always return 0 indicating success.
3bd1e081 1234 */
d8ef542d
MD
1235int consumer_add_channel(struct lttng_consumer_channel *channel,
1236 struct lttng_consumer_local_data *ctx)
3bd1e081 1237{
3bd1e081 1238 pthread_mutex_lock(&consumer_data.lock);
a9838785 1239 pthread_mutex_lock(&channel->lock);
ec6ea7d0 1240 pthread_mutex_lock(&channel->timer_lock);
c77fc10a 1241
b5a6470f
DG
1242 /*
1243 * This gives us a guarantee that the channel we are about to add to the
1244 * channel hash table will be unique. See this function comment on the why
1245 * we need to steel the channel key at this stage.
1246 */
1247 steal_channel_key(channel->key);
c77fc10a 1248
b5a6470f 1249 rcu_read_lock();
d88aee68 1250 lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
5c3892a6
JG
1251 lttng_ht_add_u64(consumer_data.channels_by_session_id_ht,
1252 &channel->channels_by_session_id_ht_node);
6065ceec 1253 rcu_read_unlock();
d2956687 1254 channel->is_published = true;
b5a6470f 1255
ec6ea7d0 1256 pthread_mutex_unlock(&channel->timer_lock);
a9838785 1257 pthread_mutex_unlock(&channel->lock);
3bd1e081 1258 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 1259
b5a6470f 1260 if (channel->wait_fd != -1 && channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
a0cbdd2e 1261 notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
d8ef542d 1262 }
b5a6470f
DG
1263
1264 return 0;
3bd1e081
MD
1265}
1266
1267/*
1268 * Allocate the pollfd structure and the local view of the out fds to avoid
1269 * doing a lookup in the linked list and concurrency issues when writing is
1270 * needed. Called with consumer_data.lock held.
1271 *
1272 * Returns the number of fds in the structures.
1273 */
ffe60014
DG
1274static int update_poll_array(struct lttng_consumer_local_data *ctx,
1275 struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
9a2fcf78 1276 struct lttng_ht *ht, int *nb_inactive_fd)
3bd1e081 1277{
3bd1e081 1278 int i = 0;
e4421fec
DG
1279 struct lttng_ht_iter iter;
1280 struct lttng_consumer_stream *stream;
3bd1e081 1281
ffe60014
DG
1282 assert(ctx);
1283 assert(ht);
1284 assert(pollfd);
1285 assert(local_stream);
1286
3bd1e081 1287 DBG("Updating poll fd array");
9a2fcf78 1288 *nb_inactive_fd = 0;
481d6c57 1289 rcu_read_lock();
43c34bc3 1290 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
8994307f
DG
1291 /*
1292 * Only active streams with an active end point can be added to the
1293 * poll set and local stream storage of the thread.
1294 *
1295 * There is a potential race here for endpoint_status to be updated
1296 * just after the check. However, this is OK since the stream(s) will
1297 * be deleted once the thread is notified that the end point state has
1298 * changed where this function will be called back again.
9a2fcf78
JD
1299 *
1300 * We track the number of inactive FDs because they still need to be
1301 * closed by the polling thread after a wakeup on the data_pipe or
1302 * metadata_pipe.
8994307f 1303 */
d2956687 1304 if (stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
9a2fcf78 1305 (*nb_inactive_fd)++;
3bd1e081
MD
1306 continue;
1307 }
7972aab2
DG
1308 /*
1309 * This clobbers way too much the debug output. Uncomment that if you
1310 * need it for debugging purposes.
7972aab2 1311 */
e4421fec 1312 (*pollfd)[i].fd = stream->wait_fd;
3bd1e081 1313 (*pollfd)[i].events = POLLIN | POLLPRI;
e4421fec 1314 local_stream[i] = stream;
3bd1e081
MD
1315 i++;
1316 }
481d6c57 1317 rcu_read_unlock();
3bd1e081
MD
1318
1319 /*
50f8ae69 1320 * Insert the consumer_data_pipe at the end of the array and don't
3bd1e081
MD
1321 * increment i so nb_fd is the number of real FD.
1322 */
acdb9057 1323 (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
509bb1cf 1324 (*pollfd)[i].events = POLLIN | POLLPRI;
02b3d176
DG
1325
1326 (*pollfd)[i + 1].fd = lttng_pipe_get_readfd(ctx->consumer_wakeup_pipe);
1327 (*pollfd)[i + 1].events = POLLIN | POLLPRI;
3bd1e081
MD
1328 return i;
1329}
1330
1331/*
84382d49
MD
1332 * Poll on the should_quit pipe and the command socket return -1 on
1333 * error, 1 if should exit, 0 if data is available on the command socket
3bd1e081
MD
1334 */
1335int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
1336{
1337 int num_rdy;
1338
88f2b785 1339restart:
3bd1e081
MD
1340 num_rdy = poll(consumer_sockpoll, 2, -1);
1341 if (num_rdy == -1) {
88f2b785
MD
1342 /*
1343 * Restart interrupted system call.
1344 */
1345 if (errno == EINTR) {
1346 goto restart;
1347 }
7a57cf92 1348 PERROR("Poll error");
84382d49 1349 return -1;
3bd1e081 1350 }
509bb1cf 1351 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
3bd1e081 1352 DBG("consumer_should_quit wake up");
84382d49 1353 return 1;
3bd1e081
MD
1354 }
1355 return 0;
3bd1e081
MD
1356}
1357
1358/*
1359 * Set the error socket.
1360 */
ffe60014
DG
1361void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
1362 int sock)
3bd1e081
MD
1363{
1364 ctx->consumer_error_socket = sock;
1365}
1366
1367/*
1368 * Set the command socket path.
1369 */
3bd1e081
MD
1370void lttng_consumer_set_command_sock_path(
1371 struct lttng_consumer_local_data *ctx, char *sock)
1372{
1373 ctx->consumer_command_sock_path = sock;
1374}
1375
1376/*
1377 * Send return code to the session daemon.
1378 * If the socket is not defined, we return 0, it is not a fatal error
1379 */
ffe60014 1380int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
3bd1e081
MD
1381{
1382 if (ctx->consumer_error_socket > 0) {
1383 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
1384 sizeof(enum lttcomm_sessiond_command));
1385 }
1386
1387 return 0;
1388}
1389
1390/*
228b5bf7
DG
1391 * Close all the tracefiles and stream fds and MUST be called when all
1392 * instances are destroyed i.e. when all threads were joined and are ended.
3bd1e081
MD
1393 */
1394void lttng_consumer_cleanup(void)
1395{
e4421fec 1396 struct lttng_ht_iter iter;
ffe60014 1397 struct lttng_consumer_channel *channel;
6065ceec
DG
1398
1399 rcu_read_lock();
3bd1e081 1400
ffe60014
DG
1401 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
1402 node.node) {
702b1ea4 1403 consumer_del_channel(channel);
3bd1e081 1404 }
6065ceec
DG
1405
1406 rcu_read_unlock();
d6ce1df2 1407
d6ce1df2 1408 lttng_ht_destroy(consumer_data.channel_ht);
5c3892a6 1409 lttng_ht_destroy(consumer_data.channels_by_session_id_ht);
228b5bf7
DG
1410
1411 cleanup_relayd_ht();
1412
d8ef542d
MD
1413 lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
1414
228b5bf7
DG
1415 /*
1416 * This HT contains streams that are freed by either the metadata thread or
1417 * the data thread so we do *nothing* on the hash table and simply destroy
1418 * it.
1419 */
1420 lttng_ht_destroy(consumer_data.stream_list_ht);
28cc88f3
JG
1421
1422 lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
3bd1e081
MD
1423}
1424
1425/*
1426 * Called from signal handler.
1427 */
1428void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
1429{
6cd525e8
MD
1430 ssize_t ret;
1431
10211f5c 1432 CMM_STORE_SHARED(consumer_quit, 1);
6cd525e8
MD
1433 ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
1434 if (ret < 1) {
7a57cf92 1435 PERROR("write consumer quit");
3bd1e081 1436 }
ab1027f4
DG
1437
1438 DBG("Consumer flag that it should quit");
3bd1e081
MD
1439}
1440
5199ffc4
JG
1441
1442/*
1443 * Flush pending writes to trace output disk file.
1444 */
1445static
00e2e675
DG
1446void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
1447 off_t orig_offset)
3bd1e081 1448{
c7a78aab 1449 int ret;
3bd1e081
MD
1450 int outfd = stream->out_fd;
1451
1452 /*
1453 * This does a blocking write-and-wait on any page that belongs to the
1454 * subbuffer prior to the one we just wrote.
1455 * Don't care about error values, as these are just hints and ways to
1456 * limit the amount of page cache used.
1457 */
ffe60014 1458 if (orig_offset < stream->max_sb_size) {
3bd1e081
MD
1459 return;
1460 }
ffe60014
DG
1461 lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
1462 stream->max_sb_size,
3bd1e081
MD
1463 SYNC_FILE_RANGE_WAIT_BEFORE
1464 | SYNC_FILE_RANGE_WRITE
1465 | SYNC_FILE_RANGE_WAIT_AFTER);
1466 /*
1467 * Give hints to the kernel about how we access the file:
1468 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1469 * we write it.
1470 *
1471 * We need to call fadvise again after the file grows because the
1472 * kernel does not seem to apply fadvise to non-existing parts of the
1473 * file.
1474 *
1475 * Call fadvise _after_ having waited for the page writeback to
1476 * complete because the dirty page writeback semantic is not well
1477 * defined. So it can be expected to lead to lower throughput in
1478 * streaming.
1479 */
c7a78aab 1480 ret = posix_fadvise(outfd, orig_offset - stream->max_sb_size,
ffe60014 1481 stream->max_sb_size, POSIX_FADV_DONTNEED);
a0d0e127 1482 if (ret && ret != -ENOSYS) {
a74a5f4a
JG
1483 errno = ret;
1484 PERROR("posix_fadvise on fd %i", outfd);
c7a78aab 1485 }
3bd1e081
MD
1486}
1487
1488/*
1489 * Initialise the necessary environnement :
1490 * - create a new context
1491 * - create the poll_pipe
1492 * - create the should_quit pipe (for signal handler)
1493 * - create the thread pipe (for splice)
1494 *
1495 * Takes a function pointer as argument, this function is called when data is
1496 * available on a buffer. This function is responsible to do the
1497 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1498 * buffer configuration and then kernctl_put_next_subbuf at the end.
1499 *
1500 * Returns a pointer to the new context or NULL on error.
1501 */
1502struct lttng_consumer_local_data *lttng_consumer_create(
1503 enum lttng_consumer_type type,
4078b776 1504 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 1505 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
1506 int (*recv_channel)(struct lttng_consumer_channel *channel),
1507 int (*recv_stream)(struct lttng_consumer_stream *stream),
30319bcb 1508 int (*update_stream)(uint64_t stream_key, uint32_t state))
3bd1e081 1509{
d8ef542d 1510 int ret;
3bd1e081
MD
1511 struct lttng_consumer_local_data *ctx;
1512
1513 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
1514 consumer_data.type == type);
1515 consumer_data.type = type;
1516
effcf122 1517 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
3bd1e081 1518 if (ctx == NULL) {
7a57cf92 1519 PERROR("allocating context");
3bd1e081
MD
1520 goto error;
1521 }
1522
1523 ctx->consumer_error_socket = -1;
331744e3 1524 ctx->consumer_metadata_socket = -1;
75d83e50 1525 pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
3bd1e081
MD
1526 /* assign the callbacks */
1527 ctx->on_buffer_ready = buffer_ready;
1528 ctx->on_recv_channel = recv_channel;
1529 ctx->on_recv_stream = recv_stream;
1530 ctx->on_update_stream = update_stream;
1531
acdb9057
DG
1532 ctx->consumer_data_pipe = lttng_pipe_open(0);
1533 if (!ctx->consumer_data_pipe) {
3bd1e081
MD
1534 goto error_poll_pipe;
1535 }
1536
02b3d176
DG
1537 ctx->consumer_wakeup_pipe = lttng_pipe_open(0);
1538 if (!ctx->consumer_wakeup_pipe) {
1539 goto error_wakeup_pipe;
1540 }
1541
3bd1e081
MD
1542 ret = pipe(ctx->consumer_should_quit);
1543 if (ret < 0) {
7a57cf92 1544 PERROR("Error creating recv pipe");
3bd1e081
MD
1545 goto error_quit_pipe;
1546 }
1547
d8ef542d
MD
1548 ret = pipe(ctx->consumer_channel_pipe);
1549 if (ret < 0) {
1550 PERROR("Error creating channel pipe");
1551 goto error_channel_pipe;
1552 }
1553
13886d2d
DG
1554 ctx->consumer_metadata_pipe = lttng_pipe_open(0);
1555 if (!ctx->consumer_metadata_pipe) {
fb3a43a9
DG
1556 goto error_metadata_pipe;
1557 }
3bd1e081 1558
e9404c27
JG
1559 ctx->channel_monitor_pipe = -1;
1560
fb3a43a9 1561 return ctx;
3bd1e081 1562
fb3a43a9 1563error_metadata_pipe:
d8ef542d
MD
1564 utils_close_pipe(ctx->consumer_channel_pipe);
1565error_channel_pipe:
d8ef542d 1566 utils_close_pipe(ctx->consumer_should_quit);
3bd1e081 1567error_quit_pipe:
02b3d176
DG
1568 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
1569error_wakeup_pipe:
acdb9057 1570 lttng_pipe_destroy(ctx->consumer_data_pipe);
3bd1e081
MD
1571error_poll_pipe:
1572 free(ctx);
1573error:
1574 return NULL;
1575}
1576
282dadbc
MD
1577/*
1578 * Iterate over all streams of the hashtable and free them properly.
1579 */
1580static void destroy_data_stream_ht(struct lttng_ht *ht)
1581{
1582 struct lttng_ht_iter iter;
1583 struct lttng_consumer_stream *stream;
1584
1585 if (ht == NULL) {
1586 return;
1587 }
1588
1589 rcu_read_lock();
1590 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1591 /*
1592 * Ignore return value since we are currently cleaning up so any error
1593 * can't be handled.
1594 */
1595 (void) consumer_del_stream(stream, ht);
1596 }
1597 rcu_read_unlock();
1598
1599 lttng_ht_destroy(ht);
1600}
1601
1602/*
1603 * Iterate over all streams of the metadata hashtable and free them
1604 * properly.
1605 */
1606static void destroy_metadata_stream_ht(struct lttng_ht *ht)
1607{
1608 struct lttng_ht_iter iter;
1609 struct lttng_consumer_stream *stream;
1610
1611 if (ht == NULL) {
1612 return;
1613 }
1614
1615 rcu_read_lock();
1616 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1617 /*
1618 * Ignore return value since we are currently cleaning up so any error
1619 * can't be handled.
1620 */
1621 (void) consumer_del_metadata_stream(stream, ht);
1622 }
1623 rcu_read_unlock();
1624
1625 lttng_ht_destroy(ht);
1626}
1627
3bd1e081
MD
1628/*
1629 * Close all fds associated with the instance and free the context.
1630 */
1631void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1632{
4c462e79
MD
1633 int ret;
1634
ab1027f4
DG
1635 DBG("Consumer destroying it. Closing everything.");
1636
4f2e75b9
DG
1637 if (!ctx) {
1638 return;
1639 }
1640
282dadbc
MD
1641 destroy_data_stream_ht(data_ht);
1642 destroy_metadata_stream_ht(metadata_ht);
1643
4c462e79
MD
1644 ret = close(ctx->consumer_error_socket);
1645 if (ret) {
1646 PERROR("close");
1647 }
331744e3
JD
1648 ret = close(ctx->consumer_metadata_socket);
1649 if (ret) {
1650 PERROR("close");
1651 }
d8ef542d 1652 utils_close_pipe(ctx->consumer_channel_pipe);
acdb9057 1653 lttng_pipe_destroy(ctx->consumer_data_pipe);
13886d2d 1654 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
02b3d176 1655 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
d8ef542d 1656 utils_close_pipe(ctx->consumer_should_quit);
fb3a43a9 1657
3bd1e081
MD
1658 unlink(ctx->consumer_command_sock_path);
1659 free(ctx);
1660}
1661
6197aea7
DG
1662/*
1663 * Write the metadata stream id on the specified file descriptor.
1664 */
1665static int write_relayd_metadata_id(int fd,
1666 struct lttng_consumer_stream *stream,
239f61af 1667 unsigned long padding)
6197aea7 1668{
6cd525e8 1669 ssize_t ret;
1d4dfdef 1670 struct lttcomm_relayd_metadata_payload hdr;
6197aea7 1671
1d4dfdef
DG
1672 hdr.stream_id = htobe64(stream->relayd_stream_id);
1673 hdr.padding_size = htobe32(padding);
6cd525e8
MD
1674 ret = lttng_write(fd, (void *) &hdr, sizeof(hdr));
1675 if (ret < sizeof(hdr)) {
d7b75ec8 1676 /*
6f04ed72 1677 * This error means that the fd's end is closed so ignore the PERROR
d7b75ec8
DG
1678 * not to clubber the error output since this can happen in a normal
1679 * code path.
1680 */
1681 if (errno != EPIPE) {
1682 PERROR("write metadata stream id");
1683 }
1684 DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
534d2592
DG
1685 /*
1686 * Set ret to a negative value because if ret != sizeof(hdr), we don't
1687 * handle writting the missing part so report that as an error and
1688 * don't lie to the caller.
1689 */
1690 ret = -1;
6197aea7
DG
1691 goto end;
1692 }
1d4dfdef
DG
1693 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1694 stream->relayd_stream_id, padding);
6197aea7
DG
1695
1696end:
6cd525e8 1697 return (int) ret;
6197aea7
DG
1698}
1699
3bd1e081 1700/*
09e26845
DG
1701 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1702 * core function for writing trace buffers to either the local filesystem or
1703 * the network.
1704 *
d2956687 1705 * It must be called with the stream and the channel lock held.
79d4ffb7 1706 *
09e26845 1707 * Careful review MUST be put if any changes occur!
3bd1e081
MD
1708 *
1709 * Returns the number of bytes written
1710 */
4078b776 1711ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081 1712 struct lttng_consumer_local_data *ctx,
1d4dfdef 1713 struct lttng_consumer_stream *stream, unsigned long len,
309167d2 1714 unsigned long padding,
50adc264 1715 struct ctf_packet_index *index)
3bd1e081 1716{
f02e1e8a 1717 unsigned long mmap_offset;
ffe60014 1718 void *mmap_base;
994ab360 1719 ssize_t ret = 0;
f02e1e8a
DG
1720 off_t orig_offset = stream->out_fd_offset;
1721 /* Default is on the disk */
1722 int outfd = stream->out_fd;
f02e1e8a 1723 struct consumer_relayd_sock_pair *relayd = NULL;
8994307f 1724 unsigned int relayd_hang_up = 0;
f02e1e8a
DG
1725
1726 /* RCU lock for the relayd pointer */
1727 rcu_read_lock();
1728
d2956687
JG
1729 assert(stream->chan->trace_chunk);
1730
f02e1e8a 1731 /* Flag that the current stream if set for network streaming. */
da009f2c 1732 if (stream->net_seq_idx != (uint64_t) -1ULL) {
f02e1e8a
DG
1733 relayd = consumer_find_relayd(stream->net_seq_idx);
1734 if (relayd == NULL) {
56591bac 1735 ret = -EPIPE;
f02e1e8a
DG
1736 goto end;
1737 }
1738 }
1739
1740 /* get the offset inside the fd to mmap */
3bd1e081
MD
1741 switch (consumer_data.type) {
1742 case LTTNG_CONSUMER_KERNEL:
ffe60014 1743 mmap_base = stream->mmap_base;
f02e1e8a 1744 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
994ab360 1745 if (ret < 0) {
56591bac 1746 PERROR("tracer ctl get_mmap_read_offset");
56591bac
MD
1747 goto end;
1748 }
f02e1e8a 1749 break;
7753dea8
MD
1750 case LTTNG_CONSUMER32_UST:
1751 case LTTNG_CONSUMER64_UST:
ffe60014
DG
1752 mmap_base = lttng_ustctl_get_mmap_base(stream);
1753 if (!mmap_base) {
1754 ERR("read mmap get mmap base for stream %s", stream->name);
994ab360 1755 ret = -EPERM;
ffe60014
DG
1756 goto end;
1757 }
1758 ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
56591bac
MD
1759 if (ret != 0) {
1760 PERROR("tracer ctl get_mmap_read_offset");
994ab360 1761 ret = -EINVAL;
56591bac
MD
1762 goto end;
1763 }
f02e1e8a 1764 break;
3bd1e081
MD
1765 default:
1766 ERR("Unknown consumer_data type");
1767 assert(0);
1768 }
b9182dd9 1769
f02e1e8a
DG
1770 /* Handle stream on the relayd if the output is on the network */
1771 if (relayd) {
1772 unsigned long netlen = len;
1773
1774 /*
1775 * Lock the control socket for the complete duration of the function
1776 * since from this point on we will use the socket.
1777 */
1778 if (stream->metadata_flag) {
1779 /* Metadata requires the control socket. */
1780 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
93ec662e
JD
1781 if (stream->reset_metadata_flag) {
1782 ret = relayd_reset_metadata(&relayd->control_sock,
1783 stream->relayd_stream_id,
1784 stream->metadata_version);
1785 if (ret < 0) {
1786 relayd_hang_up = 1;
1787 goto write_error;
1788 }
1789 stream->reset_metadata_flag = 0;
1790 }
1d4dfdef 1791 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
f02e1e8a
DG
1792 }
1793
1d4dfdef 1794 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
994ab360
DG
1795 if (ret < 0) {
1796 relayd_hang_up = 1;
1797 goto write_error;
1798 }
1799 /* Use the returned socket. */
1800 outfd = ret;
f02e1e8a 1801
994ab360
DG
1802 /* Write metadata stream id before payload */
1803 if (stream->metadata_flag) {
239f61af 1804 ret = write_relayd_metadata_id(outfd, stream, padding);
994ab360 1805 if (ret < 0) {
8994307f
DG
1806 relayd_hang_up = 1;
1807 goto write_error;
1808 }
f02e1e8a 1809 }
1d4dfdef
DG
1810 } else {
1811 /* No streaming, we have to set the len with the full padding */
1812 len += padding;
1624d5b7 1813
93ec662e
JD
1814 if (stream->metadata_flag && stream->reset_metadata_flag) {
1815 ret = utils_truncate_stream_file(stream->out_fd, 0);
1816 if (ret < 0) {
1817 ERR("Reset metadata file");
1818 goto end;
1819 }
1820 stream->reset_metadata_flag = 0;
1821 }
1822
1624d5b7
JD
1823 /*
1824 * Check if we need to change the tracefile before writing the packet.
1825 */
1826 if (stream->chan->tracefile_size > 0 &&
1827 (stream->tracefile_size_current + len) >
1828 stream->chan->tracefile_size) {
d2956687
JG
1829 ret = consumer_stream_rotate_output_files(stream);
1830 if (ret) {
1624d5b7
JD
1831 goto end;
1832 }
309167d2 1833 outfd = stream->out_fd;
a1ae300f 1834 orig_offset = 0;
1624d5b7
JD
1835 }
1836 stream->tracefile_size_current += len;
309167d2
JD
1837 if (index) {
1838 index->offset = htobe64(stream->out_fd_offset);
1839 }
f02e1e8a
DG
1840 }
1841
d02b8372
DG
1842 /*
1843 * This call guarantee that len or less is returned. It's impossible to
1844 * receive a ret value that is bigger than len.
1845 */
1846 ret = lttng_write(outfd, mmap_base + mmap_offset, len);
1847 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
1848 if (ret < 0 || ((size_t) ret != len)) {
1849 /*
1850 * Report error to caller if nothing was written else at least send the
1851 * amount written.
1852 */
1853 if (ret < 0) {
994ab360 1854 ret = -errno;
f02e1e8a 1855 }
994ab360 1856 relayd_hang_up = 1;
f02e1e8a 1857
d02b8372 1858 /* Socket operation failed. We consider the relayd dead */
fcf0f774 1859 if (errno == EPIPE) {
d02b8372
DG
1860 /*
1861 * This is possible if the fd is closed on the other side
1862 * (outfd) or any write problem. It can be verbose a bit for a
1863 * normal execution if for instance the relayd is stopped
1864 * abruptly. This can happen so set this to a DBG statement.
1865 */
1866 DBG("Consumer mmap write detected relayd hang up");
994ab360
DG
1867 } else {
1868 /* Unhandled error, print it and stop function right now. */
1869 PERROR("Error in write mmap (ret %zd != len %lu)", ret, len);
f02e1e8a 1870 }
994ab360 1871 goto write_error;
d02b8372
DG
1872 }
1873 stream->output_written += ret;
d02b8372
DG
1874
1875 /* This call is useless on a socket so better save a syscall. */
1876 if (!relayd) {
1877 /* This won't block, but will start writeout asynchronously */
1878 lttng_sync_file_range(outfd, stream->out_fd_offset, len,
1879 SYNC_FILE_RANGE_WRITE);
1880 stream->out_fd_offset += len;
f5dbe415 1881 lttng_consumer_sync_trace_file(stream, orig_offset);
f02e1e8a 1882 }
f02e1e8a 1883
8994307f
DG
1884write_error:
1885 /*
1886 * This is a special case that the relayd has closed its socket. Let's
1887 * cleanup the relayd object and all associated streams.
1888 */
1889 if (relayd && relayd_hang_up) {
9276e5c8
JR
1890 ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
1891 lttng_consumer_cleanup_relayd(relayd);
8994307f
DG
1892 }
1893
f02e1e8a
DG
1894end:
1895 /* Unlock only if ctrl socket used */
1896 if (relayd && stream->metadata_flag) {
1897 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1898 }
1899
1900 rcu_read_unlock();
994ab360 1901 return ret;
3bd1e081
MD
1902}
1903
1904/*
1905 * Splice the data from the ring buffer to the tracefile.
1906 *
79d4ffb7
DG
1907 * It must be called with the stream lock held.
1908 *
3bd1e081
MD
1909 * Returns the number of bytes spliced.
1910 */
4078b776 1911ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081 1912 struct lttng_consumer_local_data *ctx,
1d4dfdef 1913 struct lttng_consumer_stream *stream, unsigned long len,
309167d2 1914 unsigned long padding,
50adc264 1915 struct ctf_packet_index *index)
3bd1e081 1916{
f02e1e8a
DG
1917 ssize_t ret = 0, written = 0, ret_splice = 0;
1918 loff_t offset = 0;
1919 off_t orig_offset = stream->out_fd_offset;
1920 int fd = stream->wait_fd;
1921 /* Default is on the disk */
1922 int outfd = stream->out_fd;
f02e1e8a 1923 struct consumer_relayd_sock_pair *relayd = NULL;
fb3a43a9 1924 int *splice_pipe;
8994307f 1925 unsigned int relayd_hang_up = 0;
f02e1e8a 1926
3bd1e081
MD
1927 switch (consumer_data.type) {
1928 case LTTNG_CONSUMER_KERNEL:
f02e1e8a 1929 break;
7753dea8
MD
1930 case LTTNG_CONSUMER32_UST:
1931 case LTTNG_CONSUMER64_UST:
f02e1e8a 1932 /* Not supported for user space tracing */
3bd1e081
MD
1933 return -ENOSYS;
1934 default:
1935 ERR("Unknown consumer_data type");
1936 assert(0);
3bd1e081
MD
1937 }
1938
f02e1e8a
DG
1939 /* RCU lock for the relayd pointer */
1940 rcu_read_lock();
1941
1942 /* Flag that the current stream if set for network streaming. */
da009f2c 1943 if (stream->net_seq_idx != (uint64_t) -1ULL) {
f02e1e8a
DG
1944 relayd = consumer_find_relayd(stream->net_seq_idx);
1945 if (relayd == NULL) {
ad0b0d23 1946 written = -ret;
f02e1e8a
DG
1947 goto end;
1948 }
1949 }
a2361a61 1950 splice_pipe = stream->splice_pipe;
fb3a43a9 1951
f02e1e8a 1952 /* Write metadata stream id before payload */
1d4dfdef 1953 if (relayd) {
ad0b0d23 1954 unsigned long total_len = len;
f02e1e8a 1955
1d4dfdef
DG
1956 if (stream->metadata_flag) {
1957 /*
1958 * Lock the control socket for the complete duration of the function
1959 * since from this point on we will use the socket.
1960 */
1961 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1962
93ec662e
JD
1963 if (stream->reset_metadata_flag) {
1964 ret = relayd_reset_metadata(&relayd->control_sock,
1965 stream->relayd_stream_id,
1966 stream->metadata_version);
1967 if (ret < 0) {
1968 relayd_hang_up = 1;
1969 goto write_error;
1970 }
1971 stream->reset_metadata_flag = 0;
1972 }
239f61af 1973 ret = write_relayd_metadata_id(splice_pipe[1], stream,
1d4dfdef
DG
1974 padding);
1975 if (ret < 0) {
1976 written = ret;
ad0b0d23
DG
1977 relayd_hang_up = 1;
1978 goto write_error;
1d4dfdef
DG
1979 }
1980
1981 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1982 }
1983
1984 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
ad0b0d23
DG
1985 if (ret < 0) {
1986 written = ret;
1987 relayd_hang_up = 1;
1988 goto write_error;
f02e1e8a 1989 }
ad0b0d23
DG
1990 /* Use the returned socket. */
1991 outfd = ret;
1d4dfdef
DG
1992 } else {
1993 /* No streaming, we have to set the len with the full padding */
1994 len += padding;
1624d5b7 1995
93ec662e
JD
1996 if (stream->metadata_flag && stream->reset_metadata_flag) {
1997 ret = utils_truncate_stream_file(stream->out_fd, 0);
1998 if (ret < 0) {
1999 ERR("Reset metadata file");
2000 goto end;
2001 }
2002 stream->reset_metadata_flag = 0;
2003 }
1624d5b7
JD
2004 /*
2005 * Check if we need to change the tracefile before writing the packet.
2006 */
2007 if (stream->chan->tracefile_size > 0 &&
2008 (stream->tracefile_size_current + len) >
2009 stream->chan->tracefile_size) {
d2956687 2010 ret = consumer_stream_rotate_output_files(stream);
1624d5b7 2011 if (ret < 0) {
ad0b0d23 2012 written = ret;
1624d5b7
JD
2013 goto end;
2014 }
309167d2 2015 outfd = stream->out_fd;
a1ae300f 2016 orig_offset = 0;
1624d5b7
JD
2017 }
2018 stream->tracefile_size_current += len;
309167d2 2019 index->offset = htobe64(stream->out_fd_offset);
f02e1e8a
DG
2020 }
2021
2022 while (len > 0) {
1d4dfdef
DG
2023 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
2024 (unsigned long)offset, len, fd, splice_pipe[1]);
fb3a43a9 2025 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
f02e1e8a
DG
2026 SPLICE_F_MOVE | SPLICE_F_MORE);
2027 DBG("splice chan to pipe, ret %zd", ret_splice);
2028 if (ret_splice < 0) {
d02b8372 2029 ret = errno;
ad0b0d23 2030 written = -ret;
d02b8372 2031 PERROR("Error in relay splice");
f02e1e8a
DG
2032 goto splice_error;
2033 }
2034
2035 /* Handle stream on the relayd if the output is on the network */
ad0b0d23
DG
2036 if (relayd && stream->metadata_flag) {
2037 size_t metadata_payload_size =
2038 sizeof(struct lttcomm_relayd_metadata_payload);
2039
2040 /* Update counter to fit the spliced data */
2041 ret_splice += metadata_payload_size;
2042 len += metadata_payload_size;
2043 /*
2044 * We do this so the return value can match the len passed as
2045 * argument to this function.
2046 */
2047 written -= metadata_payload_size;
f02e1e8a
DG
2048 }
2049
2050 /* Splice data out */
fb3a43a9 2051 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
f02e1e8a 2052 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
a2361a61
JD
2053 DBG("Consumer splice pipe to file (out_fd: %d), ret %zd",
2054 outfd, ret_splice);
f02e1e8a 2055 if (ret_splice < 0) {
d02b8372 2056 ret = errno;
ad0b0d23
DG
2057 written = -ret;
2058 relayd_hang_up = 1;
2059 goto write_error;
f02e1e8a 2060 } else if (ret_splice > len) {
d02b8372
DG
2061 /*
2062 * We don't expect this code path to be executed but you never know
2063 * so this is an extra protection agains a buggy splice().
2064 */
f02e1e8a 2065 ret = errno;
ad0b0d23 2066 written += ret_splice;
d02b8372
DG
2067 PERROR("Wrote more data than requested %zd (len: %lu)", ret_splice,
2068 len);
f02e1e8a 2069 goto splice_error;
d02b8372
DG
2070 } else {
2071 /* All good, update current len and continue. */
2072 len -= ret_splice;
f02e1e8a 2073 }
f02e1e8a
DG
2074
2075 /* This call is useless on a socket so better save a syscall. */
2076 if (!relayd) {
2077 /* This won't block, but will start writeout asynchronously */
2078 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
2079 SYNC_FILE_RANGE_WRITE);
2080 stream->out_fd_offset += ret_splice;
2081 }
e5d1a9b3 2082 stream->output_written += ret_splice;
f02e1e8a
DG
2083 written += ret_splice;
2084 }
f5dbe415
JG
2085 if (!relayd) {
2086 lttng_consumer_sync_trace_file(stream, orig_offset);
2087 }
f02e1e8a
DG
2088 goto end;
2089
8994307f
DG
2090write_error:
2091 /*
2092 * This is a special case that the relayd has closed its socket. Let's
2093 * cleanup the relayd object and all associated streams.
2094 */
2095 if (relayd && relayd_hang_up) {
9276e5c8
JR
2096 ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
2097 lttng_consumer_cleanup_relayd(relayd);
8994307f
DG
2098 /* Skip splice error so the consumer does not fail */
2099 goto end;
2100 }
2101
f02e1e8a
DG
2102splice_error:
2103 /* send the appropriate error description to sessiond */
2104 switch (ret) {
f02e1e8a 2105 case EINVAL:
f73fabfd 2106 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
f02e1e8a
DG
2107 break;
2108 case ENOMEM:
f73fabfd 2109 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
f02e1e8a
DG
2110 break;
2111 case ESPIPE:
f73fabfd 2112 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
f02e1e8a
DG
2113 break;
2114 }
2115
2116end:
2117 if (relayd && stream->metadata_flag) {
2118 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
2119 }
2120
2121 rcu_read_unlock();
2122 return written;
3bd1e081
MD
2123}
2124
15055ce5
JD
2125/*
2126 * Sample the snapshot positions for a specific fd
2127 *
2128 * Returns 0 on success, < 0 on error
2129 */
2130int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
2131{
2132 switch (consumer_data.type) {
2133 case LTTNG_CONSUMER_KERNEL:
2134 return lttng_kconsumer_sample_snapshot_positions(stream);
2135 case LTTNG_CONSUMER32_UST:
2136 case LTTNG_CONSUMER64_UST:
2137 return lttng_ustconsumer_sample_snapshot_positions(stream);
2138 default:
2139 ERR("Unknown consumer_data type");
2140 assert(0);
2141 return -ENOSYS;
2142 }
2143}
3bd1e081
MD
2144/*
2145 * Take a snapshot for a specific fd
2146 *
2147 * Returns 0 on success, < 0 on error
2148 */
ffe60014 2149int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081
MD
2150{
2151 switch (consumer_data.type) {
2152 case LTTNG_CONSUMER_KERNEL:
ffe60014 2153 return lttng_kconsumer_take_snapshot(stream);
7753dea8
MD
2154 case LTTNG_CONSUMER32_UST:
2155 case LTTNG_CONSUMER64_UST:
ffe60014 2156 return lttng_ustconsumer_take_snapshot(stream);
3bd1e081
MD
2157 default:
2158 ERR("Unknown consumer_data type");
2159 assert(0);
2160 return -ENOSYS;
2161 }
3bd1e081
MD
2162}
2163
2164/*
2165 * Get the produced position
2166 *
2167 * Returns 0 on success, < 0 on error
2168 */
ffe60014 2169int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
3bd1e081
MD
2170 unsigned long *pos)
2171{
2172 switch (consumer_data.type) {
2173 case LTTNG_CONSUMER_KERNEL:
ffe60014 2174 return lttng_kconsumer_get_produced_snapshot(stream, pos);
7753dea8
MD
2175 case LTTNG_CONSUMER32_UST:
2176 case LTTNG_CONSUMER64_UST:
ffe60014 2177 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
3bd1e081
MD
2178 default:
2179 ERR("Unknown consumer_data type");
2180 assert(0);
2181 return -ENOSYS;
2182 }
2183}
2184
15055ce5
JD
2185/*
2186 * Get the consumed position (free-running counter position in bytes).
2187 *
2188 * Returns 0 on success, < 0 on error
2189 */
2190int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
2191 unsigned long *pos)
2192{
2193 switch (consumer_data.type) {
2194 case LTTNG_CONSUMER_KERNEL:
2195 return lttng_kconsumer_get_consumed_snapshot(stream, pos);
2196 case LTTNG_CONSUMER32_UST:
2197 case LTTNG_CONSUMER64_UST:
2198 return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
2199 default:
2200 ERR("Unknown consumer_data type");
2201 assert(0);
2202 return -ENOSYS;
2203 }
2204}
2205
3bd1e081
MD
2206int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
2207 int sock, struct pollfd *consumer_sockpoll)
2208{
2209 switch (consumer_data.type) {
2210 case LTTNG_CONSUMER_KERNEL:
2211 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
2212 case LTTNG_CONSUMER32_UST:
2213 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
2214 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
2215 default:
2216 ERR("Unknown consumer_data type");
2217 assert(0);
2218 return -ENOSYS;
2219 }
2220}
2221
6d574024 2222void lttng_consumer_close_all_metadata(void)
d88aee68
DG
2223{
2224 switch (consumer_data.type) {
2225 case LTTNG_CONSUMER_KERNEL:
2226 /*
2227 * The Kernel consumer has a different metadata scheme so we don't
2228 * close anything because the stream will be closed by the session
2229 * daemon.
2230 */
2231 break;
2232 case LTTNG_CONSUMER32_UST:
2233 case LTTNG_CONSUMER64_UST:
2234 /*
2235 * Close all metadata streams. The metadata hash table is passed and
2236 * this call iterates over it by closing all wakeup fd. This is safe
2237 * because at this point we are sure that the metadata producer is
2238 * either dead or blocked.
2239 */
6d574024 2240 lttng_ustconsumer_close_all_metadata(metadata_ht);
d88aee68
DG
2241 break;
2242 default:
2243 ERR("Unknown consumer_data type");
2244 assert(0);
2245 }
2246}
2247
fb3a43a9
DG
2248/*
2249 * Clean up a metadata stream and free its memory.
2250 */
e316aad5
DG
2251void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
2252 struct lttng_ht *ht)
fb3a43a9 2253{
e316aad5 2254 struct lttng_consumer_channel *free_chan = NULL;
fb3a43a9
DG
2255
2256 assert(stream);
2257 /*
2258 * This call should NEVER receive regular stream. It must always be
2259 * metadata stream and this is crucial for data structure synchronization.
2260 */
2261 assert(stream->metadata_flag);
2262
e316aad5
DG
2263 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
2264
74251bb8 2265 pthread_mutex_lock(&consumer_data.lock);
fb549e7a 2266 pthread_mutex_lock(&stream->chan->lock);
3dad2c0f 2267 pthread_mutex_lock(&stream->lock);
081424af
JG
2268 if (stream->chan->metadata_cache) {
2269 /* Only applicable to userspace consumers. */
2270 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
2271 }
8994307f 2272
6d574024
DG
2273 /* Remove any reference to that stream. */
2274 consumer_stream_delete(stream, ht);
ca22feea 2275
6d574024
DG
2276 /* Close down everything including the relayd if one. */
2277 consumer_stream_close(stream);
2278 /* Destroy tracer buffers of the stream. */
2279 consumer_stream_destroy_buffers(stream);
fb3a43a9
DG
2280
2281 /* Atomically decrement channel refcount since other threads can use it. */
f2ad556d 2282 if (!uatomic_sub_return(&stream->chan->refcount, 1)
ffe60014 2283 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
c30aaa51 2284 /* Go for channel deletion! */
e316aad5 2285 free_chan = stream->chan;
fb3a43a9
DG
2286 }
2287
73811ecc
DG
2288 /*
2289 * Nullify the stream reference so it is not used after deletion. The
6d574024
DG
2290 * channel lock MUST be acquired before being able to check for a NULL
2291 * pointer value.
73811ecc
DG
2292 */
2293 stream->chan->metadata_stream = NULL;
2294
081424af
JG
2295 if (stream->chan->metadata_cache) {
2296 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2297 }
3dad2c0f 2298 pthread_mutex_unlock(&stream->lock);
fb549e7a 2299 pthread_mutex_unlock(&stream->chan->lock);
74251bb8 2300 pthread_mutex_unlock(&consumer_data.lock);
e316aad5
DG
2301
2302 if (free_chan) {
2303 consumer_del_channel(free_chan);
2304 }
2305
d2956687
JG
2306 lttng_trace_chunk_put(stream->trace_chunk);
2307 stream->trace_chunk = NULL;
6d574024 2308 consumer_stream_free(stream);
fb3a43a9
DG
2309}
2310
2311/*
2312 * Action done with the metadata stream when adding it to the consumer internal
2313 * data structures to handle it.
2314 */
66d583dc 2315void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
fb3a43a9 2316{
5ab66908 2317 struct lttng_ht *ht = metadata_ht;
76082088 2318 struct lttng_ht_iter iter;
d88aee68 2319 struct lttng_ht_node_u64 *node;
fb3a43a9 2320
e316aad5
DG
2321 assert(stream);
2322 assert(ht);
2323
d88aee68 2324 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
e316aad5
DG
2325
2326 pthread_mutex_lock(&consumer_data.lock);
a9838785 2327 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 2328 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 2329 pthread_mutex_lock(&stream->lock);
e316aad5 2330
e316aad5
DG
2331 /*
2332 * From here, refcounts are updated so be _careful_ when returning an error
2333 * after this point.
2334 */
2335
fb3a43a9 2336 rcu_read_lock();
76082088
DG
2337
2338 /*
2339 * Lookup the stream just to make sure it does not exist in our internal
2340 * state. This should NEVER happen.
2341 */
d88aee68
DG
2342 lttng_ht_lookup(ht, &stream->key, &iter);
2343 node = lttng_ht_iter_get_node_u64(&iter);
76082088
DG
2344 assert(!node);
2345
e316aad5 2346 /*
ffe60014
DG
2347 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2348 * in terms of destroying the associated channel, because the action that
e316aad5
DG
2349 * causes the count to become 0 also causes a stream to be added. The
2350 * channel deletion will thus be triggered by the following removal of this
2351 * stream.
2352 */
ffe60014 2353 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
2354 /* Increment refcount before decrementing nb_init_stream_left */
2355 cmm_smp_wmb();
ffe60014 2356 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
2357 }
2358
d88aee68 2359 lttng_ht_add_unique_u64(ht, &stream->node);
ca22feea 2360
446156b4 2361 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
d8ef542d
MD
2362 &stream->node_channel_id);
2363
ca22feea
DG
2364 /*
2365 * Add stream to the stream_list_ht of the consumer data. No need to steal
2366 * the key since the HT does not use it and we allow to add redundant keys
2367 * into this table.
2368 */
d88aee68 2369 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 2370
fb3a43a9 2371 rcu_read_unlock();
e316aad5 2372
2e818a6a 2373 pthread_mutex_unlock(&stream->lock);
a9838785 2374 pthread_mutex_unlock(&stream->chan->lock);
ec6ea7d0 2375 pthread_mutex_unlock(&stream->chan->timer_lock);
e316aad5 2376 pthread_mutex_unlock(&consumer_data.lock);
fb3a43a9
DG
2377}
2378
8994307f
DG
2379/*
2380 * Delete data stream that are flagged for deletion (endpoint_status).
2381 */
2382static void validate_endpoint_status_data_stream(void)
2383{
2384 struct lttng_ht_iter iter;
2385 struct lttng_consumer_stream *stream;
2386
2387 DBG("Consumer delete flagged data stream");
2388
2389 rcu_read_lock();
2390 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2391 /* Validate delete flag of the stream */
79d4ffb7 2392 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2393 continue;
2394 }
2395 /* Delete it right now */
2396 consumer_del_stream(stream, data_ht);
2397 }
2398 rcu_read_unlock();
2399}
2400
2401/*
2402 * Delete metadata stream that are flagged for deletion (endpoint_status).
2403 */
2404static void validate_endpoint_status_metadata_stream(
2405 struct lttng_poll_event *pollset)
2406{
2407 struct lttng_ht_iter iter;
2408 struct lttng_consumer_stream *stream;
2409
2410 DBG("Consumer delete flagged metadata stream");
2411
2412 assert(pollset);
2413
2414 rcu_read_lock();
2415 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2416 /* Validate delete flag of the stream */
79d4ffb7 2417 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2418 continue;
2419 }
2420 /*
2421 * Remove from pollset so the metadata thread can continue without
2422 * blocking on a deleted stream.
2423 */
2424 lttng_poll_del(pollset, stream->wait_fd);
2425
2426 /* Delete it right now */
2427 consumer_del_metadata_stream(stream, metadata_ht);
2428 }
2429 rcu_read_unlock();
2430}
2431
fb3a43a9
DG
2432/*
2433 * Thread polls on metadata file descriptor and write them on disk or on the
2434 * network.
2435 */
7d980def 2436void *consumer_thread_metadata_poll(void *data)
fb3a43a9 2437{
1fc79fb4 2438 int ret, i, pollfd, err = -1;
fb3a43a9 2439 uint32_t revents, nb_fd;
e316aad5 2440 struct lttng_consumer_stream *stream = NULL;
fb3a43a9 2441 struct lttng_ht_iter iter;
d88aee68 2442 struct lttng_ht_node_u64 *node;
fb3a43a9
DG
2443 struct lttng_poll_event events;
2444 struct lttng_consumer_local_data *ctx = data;
2445 ssize_t len;
2446
2447 rcu_register_thread();
2448
1fc79fb4
MD
2449 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA);
2450
2d57de81
MD
2451 if (testpoint(consumerd_thread_metadata)) {
2452 goto error_testpoint;
2453 }
2454
9ce5646a
MD
2455 health_code_update();
2456
fb3a43a9
DG
2457 DBG("Thread metadata poll started");
2458
fb3a43a9
DG
2459 /* Size is set to 1 for the consumer_metadata pipe */
2460 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2461 if (ret < 0) {
2462 ERR("Poll set creation failed");
d8ef542d 2463 goto end_poll;
fb3a43a9
DG
2464 }
2465
13886d2d
DG
2466 ret = lttng_poll_add(&events,
2467 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
fb3a43a9
DG
2468 if (ret < 0) {
2469 goto end;
2470 }
2471
2472 /* Main loop */
2473 DBG("Metadata main loop started");
2474
2475 while (1) {
fb3a43a9 2476restart:
7fa2082e 2477 health_code_update();
9ce5646a 2478 health_poll_entry();
7fa2082e 2479 DBG("Metadata poll wait");
fb3a43a9 2480 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
2481 DBG("Metadata poll return from wait with %d fd(s)",
2482 LTTNG_POLL_GETNB(&events));
9ce5646a 2483 health_poll_exit();
40063ead 2484 DBG("Metadata event caught in thread");
fb3a43a9
DG
2485 if (ret < 0) {
2486 if (errno == EINTR) {
40063ead 2487 ERR("Poll EINTR caught");
fb3a43a9
DG
2488 goto restart;
2489 }
d9607cd7
MD
2490 if (LTTNG_POLL_GETNB(&events) == 0) {
2491 err = 0; /* All is OK */
2492 }
2493 goto end;
fb3a43a9
DG
2494 }
2495
0d9c5d77
DG
2496 nb_fd = ret;
2497
e316aad5 2498 /* From here, the event is a metadata wait fd */
fb3a43a9 2499 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2500 health_code_update();
2501
fb3a43a9
DG
2502 revents = LTTNG_POLL_GETEV(&events, i);
2503 pollfd = LTTNG_POLL_GETFD(&events, i);
2504
13886d2d 2505 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
03e43155 2506 if (revents & LPOLLIN) {
13886d2d
DG
2507 ssize_t pipe_len;
2508
2509 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2510 &stream, sizeof(stream));
6cd525e8 2511 if (pipe_len < sizeof(stream)) {
03e43155
MD
2512 if (pipe_len < 0) {
2513 PERROR("read metadata stream");
2514 }
fb3a43a9 2515 /*
03e43155
MD
2516 * Remove the pipe from the poll set and continue the loop
2517 * since their might be data to consume.
fb3a43a9 2518 */
03e43155
MD
2519 lttng_poll_del(&events,
2520 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2521 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
fb3a43a9
DG
2522 continue;
2523 }
2524
8994307f
DG
2525 /* A NULL stream means that the state has changed. */
2526 if (stream == NULL) {
2527 /* Check for deleted streams. */
2528 validate_endpoint_status_metadata_stream(&events);
3714380f 2529 goto restart;
8994307f
DG
2530 }
2531
fb3a43a9
DG
2532 DBG("Adding metadata stream %d to poll set",
2533 stream->wait_fd);
2534
fb3a43a9
DG
2535 /* Add metadata stream to the global poll events list */
2536 lttng_poll_add(&events, stream->wait_fd,
6d574024 2537 LPOLLIN | LPOLLPRI | LPOLLHUP);
03e43155
MD
2538 } else if (revents & (LPOLLERR | LPOLLHUP)) {
2539 DBG("Metadata thread pipe hung up");
2540 /*
2541 * Remove the pipe from the poll set and continue the loop
2542 * since their might be data to consume.
2543 */
2544 lttng_poll_del(&events,
2545 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2546 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2547 continue;
2548 } else {
2549 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2550 goto end;
fb3a43a9
DG
2551 }
2552
e316aad5 2553 /* Handle other stream */
fb3a43a9
DG
2554 continue;
2555 }
2556
d09e1200 2557 rcu_read_lock();
d88aee68
DG
2558 {
2559 uint64_t tmp_id = (uint64_t) pollfd;
2560
2561 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2562 }
2563 node = lttng_ht_iter_get_node_u64(&iter);
e316aad5 2564 assert(node);
fb3a43a9
DG
2565
2566 stream = caa_container_of(node, struct lttng_consumer_stream,
58b1f425 2567 node);
fb3a43a9 2568
03e43155
MD
2569 if (revents & (LPOLLIN | LPOLLPRI)) {
2570 /* Get the data out of the metadata file descriptor */
2571 DBG("Metadata available on fd %d", pollfd);
2572 assert(stream->wait_fd == pollfd);
2573
2574 do {
2575 health_code_update();
2576
2577 len = ctx->on_buffer_ready(stream, ctx);
2578 /*
2579 * We don't check the return value here since if we get
83f4233d 2580 * a negative len, it means an error occurred thus we
03e43155
MD
2581 * simply remove it from the poll set and free the
2582 * stream.
2583 */
2584 } while (len > 0);
2585
2586 /* It's ok to have an unavailable sub-buffer */
2587 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2588 /* Clean up stream from consumer and free it. */
2589 lttng_poll_del(&events, stream->wait_fd);
2590 consumer_del_metadata_stream(stream, metadata_ht);
2591 }
2592 } else if (revents & (LPOLLERR | LPOLLHUP)) {
e316aad5 2593 DBG("Metadata fd %d is hup|err.", pollfd);
fb3a43a9
DG
2594 if (!stream->hangup_flush_done
2595 && (consumer_data.type == LTTNG_CONSUMER32_UST
2596 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2597 DBG("Attempting to flush and consume the UST buffers");
2598 lttng_ustconsumer_on_stream_hangup(stream);
2599
2600 /* We just flushed the stream now read it. */
4bb94b75 2601 do {
9ce5646a
MD
2602 health_code_update();
2603
4bb94b75
DG
2604 len = ctx->on_buffer_ready(stream, ctx);
2605 /*
2606 * We don't check the return value here since if we get
83f4233d 2607 * a negative len, it means an error occurred thus we
4bb94b75
DG
2608 * simply remove it from the poll set and free the
2609 * stream.
2610 */
2611 } while (len > 0);
fb3a43a9
DG
2612 }
2613
fb3a43a9 2614 lttng_poll_del(&events, stream->wait_fd);
e316aad5
DG
2615 /*
2616 * This call update the channel states, closes file descriptors
2617 * and securely free the stream.
2618 */
2619 consumer_del_metadata_stream(stream, metadata_ht);
03e43155
MD
2620 } else {
2621 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
6f2f1a70 2622 rcu_read_unlock();
03e43155 2623 goto end;
fb3a43a9 2624 }
e316aad5 2625 /* Release RCU lock for the stream looked up */
d09e1200 2626 rcu_read_unlock();
fb3a43a9
DG
2627 }
2628 }
2629
1fc79fb4
MD
2630 /* All is OK */
2631 err = 0;
fb3a43a9
DG
2632end:
2633 DBG("Metadata poll thread exiting");
fb3a43a9 2634
d8ef542d
MD
2635 lttng_poll_clean(&events);
2636end_poll:
2d57de81 2637error_testpoint:
1fc79fb4
MD
2638 if (err) {
2639 health_error();
2640 ERR("Health error occurred in %s", __func__);
2641 }
2642 health_unregister(health_consumerd);
fb3a43a9
DG
2643 rcu_unregister_thread();
2644 return NULL;
2645}
2646
3bd1e081 2647/*
e4421fec 2648 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
2649 * it to tracefile if necessary.
2650 */
7d980def 2651void *consumer_thread_data_poll(void *data)
3bd1e081 2652{
1fc79fb4 2653 int num_rdy, num_hup, high_prio, ret, i, err = -1;
3bd1e081
MD
2654 struct pollfd *pollfd = NULL;
2655 /* local view of the streams */
c869f647 2656 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
3bd1e081 2657 /* local view of consumer_data.fds_count */
8bdcc002
JG
2658 int nb_fd = 0;
2659 /* 2 for the consumer_data_pipe and wake up pipe */
2660 const int nb_pipes_fd = 2;
9a2fcf78
JD
2661 /* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
2662 int nb_inactive_fd = 0;
3bd1e081 2663 struct lttng_consumer_local_data *ctx = data;
00e2e675 2664 ssize_t len;
3bd1e081 2665
e7b994a3
DG
2666 rcu_register_thread();
2667
1fc79fb4
MD
2668 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA);
2669
2d57de81
MD
2670 if (testpoint(consumerd_thread_data)) {
2671 goto error_testpoint;
2672 }
2673
9ce5646a
MD
2674 health_code_update();
2675
4df6c8cb
MD
2676 local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
2677 if (local_stream == NULL) {
2678 PERROR("local_stream malloc");
2679 goto end;
2680 }
3bd1e081
MD
2681
2682 while (1) {
9ce5646a
MD
2683 health_code_update();
2684
3bd1e081
MD
2685 high_prio = 0;
2686 num_hup = 0;
2687
2688 /*
e4421fec 2689 * the fds set has been updated, we need to update our
3bd1e081
MD
2690 * local array as well
2691 */
2692 pthread_mutex_lock(&consumer_data.lock);
2693 if (consumer_data.need_update) {
0e428499
DG
2694 free(pollfd);
2695 pollfd = NULL;
2696
2697 free(local_stream);
2698 local_stream = NULL;
3bd1e081 2699
8bdcc002 2700 /* Allocate for all fds */
261de637 2701 pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
3bd1e081 2702 if (pollfd == NULL) {
7a57cf92 2703 PERROR("pollfd malloc");
3bd1e081
MD
2704 pthread_mutex_unlock(&consumer_data.lock);
2705 goto end;
2706 }
2707
261de637 2708 local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
747f8642 2709 sizeof(struct lttng_consumer_stream *));
3bd1e081 2710 if (local_stream == NULL) {
7a57cf92 2711 PERROR("local_stream malloc");
3bd1e081
MD
2712 pthread_mutex_unlock(&consumer_data.lock);
2713 goto end;
2714 }
ffe60014 2715 ret = update_poll_array(ctx, &pollfd, local_stream,
9a2fcf78 2716 data_ht, &nb_inactive_fd);
3bd1e081
MD
2717 if (ret < 0) {
2718 ERR("Error in allocating pollfd or local_outfds");
f73fabfd 2719 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2720 pthread_mutex_unlock(&consumer_data.lock);
2721 goto end;
2722 }
2723 nb_fd = ret;
2724 consumer_data.need_update = 0;
2725 }
2726 pthread_mutex_unlock(&consumer_data.lock);
2727
4078b776 2728 /* No FDs and consumer_quit, consumer_cleanup the thread */
9a2fcf78
JD
2729 if (nb_fd == 0 && nb_inactive_fd == 0 &&
2730 CMM_LOAD_SHARED(consumer_quit) == 1) {
1fc79fb4 2731 err = 0; /* All is OK */
4078b776
MD
2732 goto end;
2733 }
3bd1e081 2734 /* poll on the array of fds */
88f2b785 2735 restart:
261de637 2736 DBG("polling on %d fd", nb_fd + nb_pipes_fd);
cf0bcb51
JG
2737 if (testpoint(consumerd_thread_data_poll)) {
2738 goto end;
2739 }
9ce5646a 2740 health_poll_entry();
261de637 2741 num_rdy = poll(pollfd, nb_fd + nb_pipes_fd, -1);
9ce5646a 2742 health_poll_exit();
3bd1e081
MD
2743 DBG("poll num_rdy : %d", num_rdy);
2744 if (num_rdy == -1) {
88f2b785
MD
2745 /*
2746 * Restart interrupted system call.
2747 */
2748 if (errno == EINTR) {
2749 goto restart;
2750 }
7a57cf92 2751 PERROR("Poll error");
f73fabfd 2752 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2753 goto end;
2754 } else if (num_rdy == 0) {
2755 DBG("Polling thread timed out");
2756 goto end;
2757 }
2758
80957876
JG
2759 if (caa_unlikely(data_consumption_paused)) {
2760 DBG("Data consumption paused, sleeping...");
2761 sleep(1);
2762 goto restart;
2763 }
2764
3bd1e081 2765 /*
50f8ae69 2766 * If the consumer_data_pipe triggered poll go directly to the
00e2e675
DG
2767 * beginning of the loop to update the array. We want to prioritize
2768 * array update over low-priority reads.
3bd1e081 2769 */
509bb1cf 2770 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
ab30f567 2771 ssize_t pipe_readlen;
04fdd819 2772
50f8ae69 2773 DBG("consumer_data_pipe wake up");
acdb9057
DG
2774 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2775 &new_stream, sizeof(new_stream));
6cd525e8
MD
2776 if (pipe_readlen < sizeof(new_stream)) {
2777 PERROR("Consumer data pipe");
23f5f35d
DG
2778 /* Continue so we can at least handle the current stream(s). */
2779 continue;
2780 }
c869f647
DG
2781
2782 /*
2783 * If the stream is NULL, just ignore it. It's also possible that
2784 * the sessiond poll thread changed the consumer_quit state and is
2785 * waking us up to test it.
2786 */
2787 if (new_stream == NULL) {
8994307f 2788 validate_endpoint_status_data_stream();
c869f647
DG
2789 continue;
2790 }
2791
c869f647 2792 /* Continue to update the local streams and handle prio ones */
3bd1e081
MD
2793 continue;
2794 }
2795
02b3d176
DG
2796 /* Handle wakeup pipe. */
2797 if (pollfd[nb_fd + 1].revents & (POLLIN | POLLPRI)) {
2798 char dummy;
2799 ssize_t pipe_readlen;
2800
2801 pipe_readlen = lttng_pipe_read(ctx->consumer_wakeup_pipe, &dummy,
2802 sizeof(dummy));
2803 if (pipe_readlen < 0) {
2804 PERROR("Consumer data wakeup pipe");
2805 }
2806 /* We've been awakened to handle stream(s). */
2807 ctx->has_wakeup = 0;
2808 }
2809
3bd1e081
MD
2810 /* Take care of high priority channels first. */
2811 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2812 health_code_update();
2813
9617607b
DG
2814 if (local_stream[i] == NULL) {
2815 continue;
2816 }
fb3a43a9 2817 if (pollfd[i].revents & POLLPRI) {
d41f73b7
MD
2818 DBG("Urgent read on fd %d", pollfd[i].fd);
2819 high_prio = 1;
4078b776 2820 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 2821 /* it's ok to have an unavailable sub-buffer */
b64403e3 2822 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2823 /* Clean the stream and free it. */
2824 consumer_del_stream(local_stream[i], data_ht);
9617607b 2825 local_stream[i] = NULL;
4078b776
MD
2826 } else if (len > 0) {
2827 local_stream[i]->data_read = 1;
d41f73b7 2828 }
3bd1e081
MD
2829 }
2830 }
2831
4078b776
MD
2832 /*
2833 * If we read high prio channel in this loop, try again
2834 * for more high prio data.
2835 */
2836 if (high_prio) {
3bd1e081
MD
2837 continue;
2838 }
2839
2840 /* Take care of low priority channels. */
4078b776 2841 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2842 health_code_update();
2843
9617607b
DG
2844 if (local_stream[i] == NULL) {
2845 continue;
2846 }
4078b776 2847 if ((pollfd[i].revents & POLLIN) ||
02b3d176
DG
2848 local_stream[i]->hangup_flush_done ||
2849 local_stream[i]->has_data) {
4078b776
MD
2850 DBG("Normal read on fd %d", pollfd[i].fd);
2851 len = ctx->on_buffer_ready(local_stream[i], ctx);
2852 /* it's ok to have an unavailable sub-buffer */
b64403e3 2853 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2854 /* Clean the stream and free it. */
2855 consumer_del_stream(local_stream[i], data_ht);
9617607b 2856 local_stream[i] = NULL;
4078b776
MD
2857 } else if (len > 0) {
2858 local_stream[i]->data_read = 1;
2859 }
2860 }
2861 }
2862
2863 /* Handle hangup and errors */
2864 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2865 health_code_update();
2866
9617607b
DG
2867 if (local_stream[i] == NULL) {
2868 continue;
2869 }
4078b776
MD
2870 if (!local_stream[i]->hangup_flush_done
2871 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2872 && (consumer_data.type == LTTNG_CONSUMER32_UST
2873 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2874 DBG("fd %d is hup|err|nval. Attempting flush and read.",
9617607b 2875 pollfd[i].fd);
4078b776
MD
2876 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2877 /* Attempt read again, for the data we just flushed. */
2878 local_stream[i]->data_read = 1;
2879 }
2880 /*
2881 * If the poll flag is HUP/ERR/NVAL and we have
2882 * read no data in this pass, we can remove the
2883 * stream from its hash table.
2884 */
2885 if ((pollfd[i].revents & POLLHUP)) {
2886 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2887 if (!local_stream[i]->data_read) {
43c34bc3 2888 consumer_del_stream(local_stream[i], data_ht);
9617607b 2889 local_stream[i] = NULL;
4078b776
MD
2890 num_hup++;
2891 }
2892 } else if (pollfd[i].revents & POLLERR) {
2893 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2894 if (!local_stream[i]->data_read) {
43c34bc3 2895 consumer_del_stream(local_stream[i], data_ht);
9617607b 2896 local_stream[i] = NULL;
4078b776
MD
2897 num_hup++;
2898 }
2899 } else if (pollfd[i].revents & POLLNVAL) {
2900 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2901 if (!local_stream[i]->data_read) {
43c34bc3 2902 consumer_del_stream(local_stream[i], data_ht);
9617607b 2903 local_stream[i] = NULL;
4078b776 2904 num_hup++;
3bd1e081
MD
2905 }
2906 }
9617607b
DG
2907 if (local_stream[i] != NULL) {
2908 local_stream[i]->data_read = 0;
2909 }
3bd1e081
MD
2910 }
2911 }
1fc79fb4
MD
2912 /* All is OK */
2913 err = 0;
3bd1e081
MD
2914end:
2915 DBG("polling thread exiting");
0e428499
DG
2916 free(pollfd);
2917 free(local_stream);
fb3a43a9
DG
2918
2919 /*
2920 * Close the write side of the pipe so epoll_wait() in
7d980def
DG
2921 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2922 * read side of the pipe. If we close them both, epoll_wait strangely does
2923 * not return and could create a endless wait period if the pipe is the
2924 * only tracked fd in the poll set. The thread will take care of closing
2925 * the read side.
fb3a43a9 2926 */
13886d2d 2927 (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
fb3a43a9 2928
2d57de81 2929error_testpoint:
1fc79fb4
MD
2930 if (err) {
2931 health_error();
2932 ERR("Health error occurred in %s", __func__);
2933 }
2934 health_unregister(health_consumerd);
2935
e7b994a3 2936 rcu_unregister_thread();
3bd1e081
MD
2937 return NULL;
2938}
2939
d8ef542d
MD
2940/*
2941 * Close wake-up end of each stream belonging to the channel. This will
2942 * allow the poll() on the stream read-side to detect when the
2943 * write-side (application) finally closes them.
2944 */
2945static
2946void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
2947{
2948 struct lttng_ht *ht;
2949 struct lttng_consumer_stream *stream;
2950 struct lttng_ht_iter iter;
2951
2952 ht = consumer_data.stream_per_chan_id_ht;
2953
2954 rcu_read_lock();
2955 cds_lfht_for_each_entry_duplicate(ht->ht,
2956 ht->hash_fct(&channel->key, lttng_ht_seed),
2957 ht->match_fct, &channel->key,
2958 &iter.iter, stream, node_channel_id.node) {
f2ad556d
MD
2959 /*
2960 * Protect against teardown with mutex.
2961 */
2962 pthread_mutex_lock(&stream->lock);
2963 if (cds_lfht_is_node_deleted(&stream->node.node)) {
2964 goto next;
2965 }
d8ef542d
MD
2966 switch (consumer_data.type) {
2967 case LTTNG_CONSUMER_KERNEL:
2968 break;
2969 case LTTNG_CONSUMER32_UST:
2970 case LTTNG_CONSUMER64_UST:
b4a650f3
DG
2971 if (stream->metadata_flag) {
2972 /* Safe and protected by the stream lock. */
2973 lttng_ustconsumer_close_metadata(stream->chan);
2974 } else {
2975 /*
2976 * Note: a mutex is taken internally within
2977 * liblttng-ust-ctl to protect timer wakeup_fd
2978 * use from concurrent close.
2979 */
2980 lttng_ustconsumer_close_stream_wakeup(stream);
2981 }
d8ef542d
MD
2982 break;
2983 default:
2984 ERR("Unknown consumer_data type");
2985 assert(0);
2986 }
f2ad556d
MD
2987 next:
2988 pthread_mutex_unlock(&stream->lock);
d8ef542d
MD
2989 }
2990 rcu_read_unlock();
2991}
2992
2993static void destroy_channel_ht(struct lttng_ht *ht)
2994{
2995 struct lttng_ht_iter iter;
2996 struct lttng_consumer_channel *channel;
2997 int ret;
2998
2999 if (ht == NULL) {
3000 return;
3001 }
3002
3003 rcu_read_lock();
3004 cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
3005 ret = lttng_ht_del(ht, &iter);
3006 assert(ret != 0);
3007 }
3008 rcu_read_unlock();
3009
3010 lttng_ht_destroy(ht);
3011}
3012
3013/*
3014 * This thread polls the channel fds to detect when they are being
3015 * closed. It closes all related streams if the channel is detected as
3016 * closed. It is currently only used as a shim layer for UST because the
3017 * consumerd needs to keep the per-stream wakeup end of pipes open for
3018 * periodical flush.
3019 */
3020void *consumer_thread_channel_poll(void *data)
3021{
1fc79fb4 3022 int ret, i, pollfd, err = -1;
d8ef542d
MD
3023 uint32_t revents, nb_fd;
3024 struct lttng_consumer_channel *chan = NULL;
3025 struct lttng_ht_iter iter;
3026 struct lttng_ht_node_u64 *node;
3027 struct lttng_poll_event events;
3028 struct lttng_consumer_local_data *ctx = data;
3029 struct lttng_ht *channel_ht;
3030
3031 rcu_register_thread();
3032
1fc79fb4
MD
3033 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL);
3034
2d57de81
MD
3035 if (testpoint(consumerd_thread_channel)) {
3036 goto error_testpoint;
3037 }
3038
9ce5646a
MD
3039 health_code_update();
3040
d8ef542d
MD
3041 channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3042 if (!channel_ht) {
3043 /* ENOMEM at this point. Better to bail out. */
3044 goto end_ht;
3045 }
3046
3047 DBG("Thread channel poll started");
3048
3049 /* Size is set to 1 for the consumer_channel pipe */
3050 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
3051 if (ret < 0) {
3052 ERR("Poll set creation failed");
3053 goto end_poll;
3054 }
3055
3056 ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
3057 if (ret < 0) {
3058 goto end;
3059 }
3060
3061 /* Main loop */
3062 DBG("Channel main loop started");
3063
3064 while (1) {
d8ef542d 3065restart:
7fa2082e
MD
3066 health_code_update();
3067 DBG("Channel poll wait");
9ce5646a 3068 health_poll_entry();
d8ef542d 3069 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
3070 DBG("Channel poll return from wait with %d fd(s)",
3071 LTTNG_POLL_GETNB(&events));
9ce5646a 3072 health_poll_exit();
40063ead 3073 DBG("Channel event caught in thread");
d8ef542d
MD
3074 if (ret < 0) {
3075 if (errno == EINTR) {
40063ead 3076 ERR("Poll EINTR caught");
d8ef542d
MD
3077 goto restart;
3078 }
d9607cd7
MD
3079 if (LTTNG_POLL_GETNB(&events) == 0) {
3080 err = 0; /* All is OK */
3081 }
d8ef542d
MD
3082 goto end;
3083 }
3084
3085 nb_fd = ret;
3086
3087 /* From here, the event is a channel wait fd */
3088 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
3089 health_code_update();
3090
d8ef542d
MD
3091 revents = LTTNG_POLL_GETEV(&events, i);
3092 pollfd = LTTNG_POLL_GETFD(&events, i);
3093
d8ef542d 3094 if (pollfd == ctx->consumer_channel_pipe[0]) {
03e43155 3095 if (revents & LPOLLIN) {
d8ef542d 3096 enum consumer_channel_action action;
a0cbdd2e 3097 uint64_t key;
d8ef542d 3098
a0cbdd2e 3099 ret = read_channel_pipe(ctx, &chan, &key, &action);
d8ef542d 3100 if (ret <= 0) {
03e43155
MD
3101 if (ret < 0) {
3102 ERR("Error reading channel pipe");
3103 }
3104 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
d8ef542d
MD
3105 continue;
3106 }
3107
3108 switch (action) {
3109 case CONSUMER_CHANNEL_ADD:
3110 DBG("Adding channel %d to poll set",
3111 chan->wait_fd);
3112
3113 lttng_ht_node_init_u64(&chan->wait_fd_node,
3114 chan->wait_fd);
c7260a81 3115 rcu_read_lock();
d8ef542d
MD
3116 lttng_ht_add_unique_u64(channel_ht,
3117 &chan->wait_fd_node);
c7260a81 3118 rcu_read_unlock();
d8ef542d
MD
3119 /* Add channel to the global poll events list */
3120 lttng_poll_add(&events, chan->wait_fd,
03e43155 3121 LPOLLERR | LPOLLHUP);
d8ef542d 3122 break;
a0cbdd2e
MD
3123 case CONSUMER_CHANNEL_DEL:
3124 {
b4a650f3
DG
3125 /*
3126 * This command should never be called if the channel
3127 * has streams monitored by either the data or metadata
3128 * thread. The consumer only notify this thread with a
3129 * channel del. command if it receives a destroy
3130 * channel command from the session daemon that send it
3131 * if a command prior to the GET_CHANNEL failed.
3132 */
3133
c7260a81 3134 rcu_read_lock();
a0cbdd2e
MD
3135 chan = consumer_find_channel(key);
3136 if (!chan) {
c7260a81 3137 rcu_read_unlock();
a0cbdd2e
MD
3138 ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
3139 break;
3140 }
3141 lttng_poll_del(&events, chan->wait_fd);
f623cc0b 3142 iter.iter.node = &chan->wait_fd_node.node;
a0cbdd2e
MD
3143 ret = lttng_ht_del(channel_ht, &iter);
3144 assert(ret == 0);
a0cbdd2e 3145
f2a444f1
DG
3146 switch (consumer_data.type) {
3147 case LTTNG_CONSUMER_KERNEL:
3148 break;
3149 case LTTNG_CONSUMER32_UST:
3150 case LTTNG_CONSUMER64_UST:
212d67a2
DG
3151 health_code_update();
3152 /* Destroy streams that might have been left in the stream list. */
3153 clean_channel_stream_list(chan);
f2a444f1
DG
3154 break;
3155 default:
3156 ERR("Unknown consumer_data type");
3157 assert(0);
3158 }
3159
a0cbdd2e
MD
3160 /*
3161 * Release our own refcount. Force channel deletion even if
3162 * streams were not initialized.
3163 */
3164 if (!uatomic_sub_return(&chan->refcount, 1)) {
3165 consumer_del_channel(chan);
3166 }
c7260a81 3167 rcu_read_unlock();
a0cbdd2e
MD
3168 goto restart;
3169 }
d8ef542d
MD
3170 case CONSUMER_CHANNEL_QUIT:
3171 /*
3172 * Remove the pipe from the poll set and continue the loop
3173 * since their might be data to consume.
3174 */
3175 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3176 continue;
3177 default:
3178 ERR("Unknown action");
3179 break;
3180 }
03e43155
MD
3181 } else if (revents & (LPOLLERR | LPOLLHUP)) {
3182 DBG("Channel thread pipe hung up");
3183 /*
3184 * Remove the pipe from the poll set and continue the loop
3185 * since their might be data to consume.
3186 */
3187 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3188 continue;
3189 } else {
3190 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3191 goto end;
d8ef542d
MD
3192 }
3193
3194 /* Handle other stream */
3195 continue;
3196 }
3197
3198 rcu_read_lock();
3199 {
3200 uint64_t tmp_id = (uint64_t) pollfd;
3201
3202 lttng_ht_lookup(channel_ht, &tmp_id, &iter);
3203 }
3204 node = lttng_ht_iter_get_node_u64(&iter);
3205 assert(node);
3206
3207 chan = caa_container_of(node, struct lttng_consumer_channel,
3208 wait_fd_node);
3209
3210 /* Check for error event */
3211 if (revents & (LPOLLERR | LPOLLHUP)) {
3212 DBG("Channel fd %d is hup|err.", pollfd);
3213
3214 lttng_poll_del(&events, chan->wait_fd);
3215 ret = lttng_ht_del(channel_ht, &iter);
3216 assert(ret == 0);
b4a650f3
DG
3217
3218 /*
3219 * This will close the wait fd for each stream associated to
3220 * this channel AND monitored by the data/metadata thread thus
3221 * will be clean by the right thread.
3222 */
d8ef542d 3223 consumer_close_channel_streams(chan);
f2ad556d
MD
3224
3225 /* Release our own refcount */
3226 if (!uatomic_sub_return(&chan->refcount, 1)
3227 && !uatomic_read(&chan->nb_init_stream_left)) {
3228 consumer_del_channel(chan);
3229 }
03e43155
MD
3230 } else {
3231 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3232 rcu_read_unlock();
3233 goto end;
d8ef542d
MD
3234 }
3235
3236 /* Release RCU lock for the channel looked up */
3237 rcu_read_unlock();
3238 }
3239 }
3240
1fc79fb4
MD
3241 /* All is OK */
3242 err = 0;
d8ef542d
MD
3243end:
3244 lttng_poll_clean(&events);
3245end_poll:
3246 destroy_channel_ht(channel_ht);
3247end_ht:
2d57de81 3248error_testpoint:
d8ef542d 3249 DBG("Channel poll thread exiting");
1fc79fb4
MD
3250 if (err) {
3251 health_error();
3252 ERR("Health error occurred in %s", __func__);
3253 }
3254 health_unregister(health_consumerd);
d8ef542d
MD
3255 rcu_unregister_thread();
3256 return NULL;
3257}
3258
331744e3
JD
3259static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
3260 struct pollfd *sockpoll, int client_socket)
3261{
3262 int ret;
3263
3264 assert(ctx);
3265 assert(sockpoll);
3266
84382d49
MD
3267 ret = lttng_consumer_poll_socket(sockpoll);
3268 if (ret) {
331744e3
JD
3269 goto error;
3270 }
3271 DBG("Metadata connection on client_socket");
3272
3273 /* Blocking call, waiting for transmission */
3274 ctx->consumer_metadata_socket = lttcomm_accept_unix_sock(client_socket);
3275 if (ctx->consumer_metadata_socket < 0) {
3276 WARN("On accept metadata");
3277 ret = -1;
3278 goto error;
3279 }
3280 ret = 0;
3281
3282error:
3283 return ret;
3284}
3285
3bd1e081
MD
3286/*
3287 * This thread listens on the consumerd socket and receives the file
3288 * descriptors from the session daemon.
3289 */
7d980def 3290void *consumer_thread_sessiond_poll(void *data)
3bd1e081 3291{
1fc79fb4 3292 int sock = -1, client_socket, ret, err = -1;
3bd1e081
MD
3293 /*
3294 * structure to poll for incoming data on communication socket avoids
3295 * making blocking sockets.
3296 */
3297 struct pollfd consumer_sockpoll[2];
3298 struct lttng_consumer_local_data *ctx = data;
3299
e7b994a3
DG
3300 rcu_register_thread();
3301
1fc79fb4
MD
3302 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_SESSIOND);
3303
2d57de81
MD
3304 if (testpoint(consumerd_thread_sessiond)) {
3305 goto error_testpoint;
3306 }
3307
9ce5646a
MD
3308 health_code_update();
3309
3bd1e081
MD
3310 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
3311 unlink(ctx->consumer_command_sock_path);
3312 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
3313 if (client_socket < 0) {
3314 ERR("Cannot create command socket");
3315 goto end;
3316 }
3317
3318 ret = lttcomm_listen_unix_sock(client_socket);
3319 if (ret < 0) {
3320 goto end;
3321 }
3322
32258573 3323 DBG("Sending ready command to lttng-sessiond");
f73fabfd 3324 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
3bd1e081
MD
3325 /* return < 0 on error, but == 0 is not fatal */
3326 if (ret < 0) {
32258573 3327 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
3328 goto end;
3329 }
3330
3bd1e081
MD
3331 /* prepare the FDs to poll : to client socket and the should_quit pipe */
3332 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
3333 consumer_sockpoll[0].events = POLLIN | POLLPRI;
3334 consumer_sockpoll[1].fd = client_socket;
3335 consumer_sockpoll[1].events = POLLIN | POLLPRI;
3336
84382d49
MD
3337 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3338 if (ret) {
3339 if (ret > 0) {
3340 /* should exit */
3341 err = 0;
3342 }
3bd1e081
MD
3343 goto end;
3344 }
3345 DBG("Connection on client_socket");
3346
3347 /* Blocking call, waiting for transmission */
3348 sock = lttcomm_accept_unix_sock(client_socket);
534d2592 3349 if (sock < 0) {
3bd1e081
MD
3350 WARN("On accept");
3351 goto end;
3352 }
3bd1e081 3353
331744e3
JD
3354 /*
3355 * Setup metadata socket which is the second socket connection on the
3356 * command unix socket.
3357 */
3358 ret = set_metadata_socket(ctx, consumer_sockpoll, client_socket);
84382d49
MD
3359 if (ret) {
3360 if (ret > 0) {
3361 /* should exit */
3362 err = 0;
3363 }
331744e3
JD
3364 goto end;
3365 }
3366
d96f09c6
DG
3367 /* This socket is not useful anymore. */
3368 ret = close(client_socket);
3369 if (ret < 0) {
3370 PERROR("close client_socket");
3371 }
3372 client_socket = -1;
3373
3bd1e081
MD
3374 /* update the polling structure to poll on the established socket */
3375 consumer_sockpoll[1].fd = sock;
3376 consumer_sockpoll[1].events = POLLIN | POLLPRI;
3377
3378 while (1) {
9ce5646a
MD
3379 health_code_update();
3380
3381 health_poll_entry();
3382 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3383 health_poll_exit();
84382d49
MD
3384 if (ret) {
3385 if (ret > 0) {
3386 /* should exit */
3387 err = 0;
3388 }
3bd1e081
MD
3389 goto end;
3390 }
3391 DBG("Incoming command on sock");
3392 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
4cbc1a04
DG
3393 if (ret <= 0) {
3394 /*
3395 * This could simply be a session daemon quitting. Don't output
3396 * ERR() here.
3397 */
3398 DBG("Communication interrupted on command socket");
41ba6035 3399 err = 0;
3bd1e081
MD
3400 goto end;
3401 }
10211f5c 3402 if (CMM_LOAD_SHARED(consumer_quit)) {
3bd1e081 3403 DBG("consumer_thread_receive_fds received quit from signal");
1fc79fb4 3404 err = 0; /* All is OK */
3bd1e081
MD
3405 goto end;
3406 }
ffe60014 3407 DBG("received command on sock");
3bd1e081 3408 }
1fc79fb4
MD
3409 /* All is OK */
3410 err = 0;
3411
3bd1e081 3412end:
ffe60014 3413 DBG("Consumer thread sessiond poll exiting");
3bd1e081 3414
d88aee68
DG
3415 /*
3416 * Close metadata streams since the producer is the session daemon which
3417 * just died.
3418 *
3419 * NOTE: for now, this only applies to the UST tracer.
3420 */
6d574024 3421 lttng_consumer_close_all_metadata();
d88aee68 3422
3bd1e081
MD
3423 /*
3424 * when all fds have hung up, the polling thread
3425 * can exit cleanly
3426 */
10211f5c 3427 CMM_STORE_SHARED(consumer_quit, 1);
3bd1e081 3428
04fdd819 3429 /*
c869f647 3430 * Notify the data poll thread to poll back again and test the
8994307f 3431 * consumer_quit state that we just set so to quit gracefully.
04fdd819 3432 */
acdb9057 3433 notify_thread_lttng_pipe(ctx->consumer_data_pipe);
c869f647 3434
a0cbdd2e 3435 notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
d8ef542d 3436
5c635c72
MD
3437 notify_health_quit_pipe(health_quit_pipe);
3438
d96f09c6
DG
3439 /* Cleaning up possibly open sockets. */
3440 if (sock >= 0) {
3441 ret = close(sock);
3442 if (ret < 0) {
3443 PERROR("close sock sessiond poll");
3444 }
3445 }
3446 if (client_socket >= 0) {
38476d24 3447 ret = close(client_socket);
d96f09c6
DG
3448 if (ret < 0) {
3449 PERROR("close client_socket sessiond poll");
3450 }
3451 }
3452
2d57de81 3453error_testpoint:
1fc79fb4
MD
3454 if (err) {
3455 health_error();
3456 ERR("Health error occurred in %s", __func__);
3457 }
3458 health_unregister(health_consumerd);
3459
e7b994a3 3460 rcu_unregister_thread();
3bd1e081
MD
3461 return NULL;
3462}
d41f73b7 3463
4078b776 3464ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
3465 struct lttng_consumer_local_data *ctx)
3466{
74251bb8
DG
3467 ssize_t ret;
3468
d2956687 3469 pthread_mutex_lock(&stream->chan->lock);
74251bb8 3470 pthread_mutex_lock(&stream->lock);
94d49140
JD
3471 if (stream->metadata_flag) {
3472 pthread_mutex_lock(&stream->metadata_rdv_lock);
3473 }
74251bb8 3474
d41f73b7
MD
3475 switch (consumer_data.type) {
3476 case LTTNG_CONSUMER_KERNEL:
d2956687 3477 ret = lttng_kconsumer_read_subbuffer(stream, ctx);
74251bb8 3478 break;
7753dea8
MD
3479 case LTTNG_CONSUMER32_UST:
3480 case LTTNG_CONSUMER64_UST:
d2956687 3481 ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
74251bb8 3482 break;
d41f73b7
MD
3483 default:
3484 ERR("Unknown consumer_data type");
3485 assert(0);
74251bb8
DG
3486 ret = -ENOSYS;
3487 break;
d41f73b7 3488 }
74251bb8 3489
94d49140
JD
3490 if (stream->metadata_flag) {
3491 pthread_cond_broadcast(&stream->metadata_rdv);
3492 pthread_mutex_unlock(&stream->metadata_rdv_lock);
3493 }
74251bb8 3494 pthread_mutex_unlock(&stream->lock);
d2956687 3495 pthread_mutex_unlock(&stream->chan->lock);
02d02e31 3496
74251bb8 3497 return ret;
d41f73b7
MD
3498}
3499
3500int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
3501{
3502 switch (consumer_data.type) {
3503 case LTTNG_CONSUMER_KERNEL:
3504 return lttng_kconsumer_on_recv_stream(stream);
7753dea8
MD
3505 case LTTNG_CONSUMER32_UST:
3506 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
3507 return lttng_ustconsumer_on_recv_stream(stream);
3508 default:
3509 ERR("Unknown consumer_data type");
3510 assert(0);
3511 return -ENOSYS;
3512 }
3513}
e4421fec
DG
3514
3515/*
3516 * Allocate and set consumer data hash tables.
3517 */
282dadbc 3518int lttng_consumer_init(void)
e4421fec 3519{
d88aee68 3520 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3521 if (!consumer_data.channel_ht) {
3522 goto error;
3523 }
3524
5c3892a6
JG
3525 consumer_data.channels_by_session_id_ht =
3526 lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3527 if (!consumer_data.channels_by_session_id_ht) {
3528 goto error;
3529 }
3530
d88aee68 3531 consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3532 if (!consumer_data.relayd_ht) {
3533 goto error;
3534 }
3535
d88aee68 3536 consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3537 if (!consumer_data.stream_list_ht) {
3538 goto error;
3539 }
3540
d8ef542d 3541 consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3542 if (!consumer_data.stream_per_chan_id_ht) {
3543 goto error;
3544 }
3545
3546 data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3547 if (!data_ht) {
3548 goto error;
3549 }
3550
3551 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3552 if (!metadata_ht) {
3553 goto error;
3554 }
3555
28cc88f3
JG
3556 consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
3557 if (!consumer_data.chunk_registry) {
3558 goto error;
3559 }
3560
282dadbc
MD
3561 return 0;
3562
3563error:
3564 return -1;
e4421fec 3565}
7735ef9e
DG
3566
3567/*
3568 * Process the ADD_RELAYD command receive by a consumer.
3569 *
3570 * This will create a relayd socket pair and add it to the relayd hash table.
3571 * The caller MUST acquire a RCU read side lock before calling it.
3572 */
2527bf85 3573 void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
7735ef9e 3574 struct lttng_consumer_local_data *ctx, int sock,
6151a90f 3575 struct pollfd *consumer_sockpoll,
d3e2ba59
JD
3576 struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
3577 uint64_t relayd_session_id)
7735ef9e 3578{
cd2b09ed 3579 int fd = -1, ret = -1, relayd_created = 0;
0c759fc9 3580 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
d4298c99 3581 struct consumer_relayd_sock_pair *relayd = NULL;
7735ef9e 3582
6151a90f
JD
3583 assert(ctx);
3584 assert(relayd_sock);
3585
da009f2c 3586 DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
7735ef9e
DG
3587
3588 /* Get relayd reference if exists. */
3589 relayd = consumer_find_relayd(net_seq_idx);
3590 if (relayd == NULL) {
da009f2c 3591 assert(sock_type == LTTNG_STREAM_CONTROL);
7735ef9e
DG
3592 /* Not found. Allocate one. */
3593 relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
3594 if (relayd == NULL) {
618a6a28
MD
3595 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3596 goto error;
0d08d75e 3597 } else {
30319bcb 3598 relayd->sessiond_session_id = sessiond_id;
0d08d75e 3599 relayd_created = 1;
7735ef9e 3600 }
0d08d75e
DG
3601
3602 /*
3603 * This code path MUST continue to the consumer send status message to
3604 * we can notify the session daemon and continue our work without
3605 * killing everything.
3606 */
da009f2c
MD
3607 } else {
3608 /*
3609 * relayd key should never be found for control socket.
3610 */
3611 assert(sock_type != LTTNG_STREAM_CONTROL);
0d08d75e
DG
3612 }
3613
3614 /* First send a status message before receiving the fds. */
0c759fc9 3615 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
618a6a28 3616 if (ret < 0) {
0d08d75e 3617 /* Somehow, the session daemon is not responding anymore. */
618a6a28
MD
3618 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3619 goto error_nosignal;
7735ef9e
DG
3620 }
3621
3622 /* Poll on consumer socket. */
84382d49
MD
3623 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3624 if (ret) {
3625 /* Needing to exit in the middle of a command: error. */
0d08d75e 3626 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
618a6a28 3627 goto error_nosignal;
7735ef9e
DG
3628 }
3629
3630 /* Get relayd socket from session daemon */
3631 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
3632 if (ret != sizeof(fd)) {
4028eeb9 3633 fd = -1; /* Just in case it gets set with an invalid value. */
0d08d75e
DG
3634
3635 /*
3636 * Failing to receive FDs might indicate a major problem such as
3637 * reaching a fd limit during the receive where the kernel returns a
3638 * MSG_CTRUNC and fails to cleanup the fd in the queue. Any case, we
3639 * don't take any chances and stop everything.
3640 *
3641 * XXX: Feature request #558 will fix that and avoid this possible
3642 * issue when reaching the fd limit.
3643 */
3644 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
618a6a28 3645 ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD;
f50f23d9
DG
3646 goto error;
3647 }
3648
7735ef9e
DG
3649 /* Copy socket information and received FD */
3650 switch (sock_type) {
3651 case LTTNG_STREAM_CONTROL:
3652 /* Copy received lttcomm socket */
6151a90f
JD
3653 lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
3654 ret = lttcomm_create_sock(&relayd->control_sock.sock);
4028eeb9 3655 /* Handle create_sock error. */
f66c074c 3656 if (ret < 0) {
618a6a28 3657 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
4028eeb9 3658 goto error;
f66c074c 3659 }
da009f2c
MD
3660 /*
3661 * Close the socket created internally by
3662 * lttcomm_create_sock, so we can replace it by the one
3663 * received from sessiond.
3664 */
3665 if (close(relayd->control_sock.sock.fd)) {
3666 PERROR("close");
3667 }
7735ef9e
DG
3668
3669 /* Assign new file descriptor */
6151a90f
JD
3670 relayd->control_sock.sock.fd = fd;
3671 /* Assign version values. */
3672 relayd->control_sock.major = relayd_sock->major;
3673 relayd->control_sock.minor = relayd_sock->minor;
c5b6f4f0 3674
d3e2ba59 3675 relayd->relayd_session_id = relayd_session_id;
c5b6f4f0 3676
7735ef9e
DG
3677 break;
3678 case LTTNG_STREAM_DATA:
3679 /* Copy received lttcomm socket */
6151a90f
JD
3680 lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
3681 ret = lttcomm_create_sock(&relayd->data_sock.sock);
4028eeb9 3682 /* Handle create_sock error. */
f66c074c 3683 if (ret < 0) {
618a6a28 3684 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
4028eeb9 3685 goto error;
f66c074c 3686 }
da009f2c
MD
3687 /*
3688 * Close the socket created internally by
3689 * lttcomm_create_sock, so we can replace it by the one
3690 * received from sessiond.
3691 */
3692 if (close(relayd->data_sock.sock.fd)) {
3693 PERROR("close");
3694 }
7735ef9e
DG
3695
3696 /* Assign new file descriptor */
6151a90f
JD
3697 relayd->data_sock.sock.fd = fd;
3698 /* Assign version values. */
3699 relayd->data_sock.major = relayd_sock->major;
3700 relayd->data_sock.minor = relayd_sock->minor;
7735ef9e
DG
3701 break;
3702 default:
3703 ERR("Unknown relayd socket type (%d)", sock_type);
618a6a28 3704 ret_code = LTTCOMM_CONSUMERD_FATAL;
7735ef9e
DG
3705 goto error;
3706 }
3707
d88aee68 3708 DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
7735ef9e
DG
3709 sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
3710 relayd->net_seq_idx, fd);
39d9954c
FD
3711 /*
3712 * We gave the ownership of the fd to the relayd structure. Set the
3713 * fd to -1 so we don't call close() on it in the error path below.
3714 */
3715 fd = -1;
7735ef9e 3716
618a6a28
MD
3717 /* We successfully added the socket. Send status back. */
3718 ret = consumer_send_status_msg(sock, ret_code);
3719 if (ret < 0) {
3720 /* Somehow, the session daemon is not responding anymore. */
3721 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3722 goto error_nosignal;
3723 }
3724
7735ef9e
DG
3725 /*
3726 * Add relayd socket pair to consumer data hashtable. If object already
3727 * exists or on error, the function gracefully returns.
3728 */
9276e5c8 3729 relayd->ctx = ctx;
d09e1200 3730 add_relayd(relayd);
7735ef9e
DG
3731
3732 /* All good! */
2527bf85 3733 return;
7735ef9e
DG
3734
3735error:
618a6a28
MD
3736 if (consumer_send_status_msg(sock, ret_code) < 0) {
3737 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3738 }
3739
3740error_nosignal:
4028eeb9
DG
3741 /* Close received socket if valid. */
3742 if (fd >= 0) {
3743 if (close(fd)) {
3744 PERROR("close received socket");
3745 }
3746 }
cd2b09ed
DG
3747
3748 if (relayd_created) {
cd2b09ed
DG
3749 free(relayd);
3750 }
7735ef9e 3751}
ca22feea 3752
f7079f67
DG
3753/*
3754 * Search for a relayd associated to the session id and return the reference.
3755 *
3756 * A rcu read side lock MUST be acquire before calling this function and locked
3757 * until the relayd object is no longer necessary.
3758 */
3759static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
3760{
3761 struct lttng_ht_iter iter;
f7079f67 3762 struct consumer_relayd_sock_pair *relayd = NULL;
f7079f67
DG
3763
3764 /* Iterate over all relayd since they are indexed by net_seq_idx. */
3765 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
3766 node.node) {
18261bd1
DG
3767 /*
3768 * Check by sessiond id which is unique here where the relayd session
3769 * id might not be when having multiple relayd.
3770 */
3771 if (relayd->sessiond_session_id == id) {
f7079f67 3772 /* Found the relayd. There can be only one per id. */
18261bd1 3773 goto found;
f7079f67
DG
3774 }
3775 }
3776
18261bd1
DG
3777 return NULL;
3778
3779found:
f7079f67
DG
3780 return relayd;
3781}
3782
ca22feea
DG
3783/*
3784 * Check if for a given session id there is still data needed to be extract
3785 * from the buffers.
3786 *
6d805429 3787 * Return 1 if data is pending or else 0 meaning ready to be read.
ca22feea 3788 */
6d805429 3789int consumer_data_pending(uint64_t id)
ca22feea
DG
3790{
3791 int ret;
3792 struct lttng_ht_iter iter;
3793 struct lttng_ht *ht;
3794 struct lttng_consumer_stream *stream;
f7079f67 3795 struct consumer_relayd_sock_pair *relayd = NULL;
6d805429 3796 int (*data_pending)(struct lttng_consumer_stream *);
ca22feea 3797
6d805429 3798 DBG("Consumer data pending command on session id %" PRIu64, id);
ca22feea 3799
6f6eda74 3800 rcu_read_lock();
ca22feea
DG
3801 pthread_mutex_lock(&consumer_data.lock);
3802
3803 switch (consumer_data.type) {
3804 case LTTNG_CONSUMER_KERNEL:
6d805429 3805 data_pending = lttng_kconsumer_data_pending;
ca22feea
DG
3806 break;
3807 case LTTNG_CONSUMER32_UST:
3808 case LTTNG_CONSUMER64_UST:
6d805429 3809 data_pending = lttng_ustconsumer_data_pending;
ca22feea
DG
3810 break;
3811 default:
3812 ERR("Unknown consumer data type");
3813 assert(0);
3814 }
3815
3816 /* Ease our life a bit */
3817 ht = consumer_data.stream_list_ht;
3818
c8f59ee5 3819 cds_lfht_for_each_entry_duplicate(ht->ht,
d88aee68
DG
3820 ht->hash_fct(&id, lttng_ht_seed),
3821 ht->match_fct, &id,
ca22feea 3822 &iter.iter, stream, node_session_id.node) {
bb586a6e 3823 pthread_mutex_lock(&stream->lock);
ca22feea 3824
4e9a4686
DG
3825 /*
3826 * A removed node from the hash table indicates that the stream has
3827 * been deleted thus having a guarantee that the buffers are closed
3828 * on the consumer side. However, data can still be transmitted
3829 * over the network so don't skip the relayd check.
3830 */
3831 ret = cds_lfht_is_node_deleted(&stream->node.node);
3832 if (!ret) {
3833 /* Check the stream if there is data in the buffers. */
6d805429
DG
3834 ret = data_pending(stream);
3835 if (ret == 1) {
4e9a4686 3836 pthread_mutex_unlock(&stream->lock);
f7079f67 3837 goto data_pending;
4e9a4686
DG
3838 }
3839 }
3840
d9f0c7c7
JR
3841 pthread_mutex_unlock(&stream->lock);
3842 }
3843
3844 relayd = find_relayd_by_session_id(id);
3845 if (relayd) {
3846 unsigned int is_data_inflight = 0;
3847
3848 /* Send init command for data pending. */
3849 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3850 ret = relayd_begin_data_pending(&relayd->control_sock,
3851 relayd->relayd_session_id);
3852 if (ret < 0) {
3853 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3854 /* Communication error thus the relayd so no data pending. */
3855 goto data_not_pending;
3856 }
3857
3858 cds_lfht_for_each_entry_duplicate(ht->ht,
3859 ht->hash_fct(&id, lttng_ht_seed),
3860 ht->match_fct, &id,
3861 &iter.iter, stream, node_session_id.node) {
c8f59ee5 3862 if (stream->metadata_flag) {
ad7051c0
DG
3863 ret = relayd_quiescent_control(&relayd->control_sock,
3864 stream->relayd_stream_id);
c8f59ee5 3865 } else {
6d805429 3866 ret = relayd_data_pending(&relayd->control_sock,
39df6d9f
DG
3867 stream->relayd_stream_id,
3868 stream->next_net_seq_num - 1);
c8f59ee5 3869 }
d9f0c7c7
JR
3870
3871 if (ret == 1) {
3872 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3873 goto data_pending;
3874 } else if (ret < 0) {
9276e5c8
JR
3875 ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
3876 lttng_consumer_cleanup_relayd(relayd);
3877 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
9276e5c8
JR
3878 goto data_not_pending;
3879 }
c8f59ee5 3880 }
f7079f67 3881
d9f0c7c7 3882 /* Send end command for data pending. */
f7079f67
DG
3883 ret = relayd_end_data_pending(&relayd->control_sock,
3884 relayd->relayd_session_id, &is_data_inflight);
3885 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
bdd88757 3886 if (ret < 0) {
9276e5c8
JR
3887 ERR("Relayd end data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
3888 lttng_consumer_cleanup_relayd(relayd);
f7079f67
DG
3889 goto data_not_pending;
3890 }
bdd88757
DG
3891 if (is_data_inflight) {
3892 goto data_pending;
3893 }
f7079f67
DG
3894 }
3895
ca22feea 3896 /*
f7079f67
DG
3897 * Finding _no_ node in the hash table and no inflight data means that the
3898 * stream(s) have been removed thus data is guaranteed to be available for
3899 * analysis from the trace files.
ca22feea
DG
3900 */
3901
f7079f67 3902data_not_pending:
ca22feea
DG
3903 /* Data is available to be read by a viewer. */
3904 pthread_mutex_unlock(&consumer_data.lock);
c8f59ee5 3905 rcu_read_unlock();
6d805429 3906 return 0;
ca22feea 3907
f7079f67 3908data_pending:
ca22feea
DG
3909 /* Data is still being extracted from buffers. */
3910 pthread_mutex_unlock(&consumer_data.lock);
c8f59ee5 3911 rcu_read_unlock();
6d805429 3912 return 1;
ca22feea 3913}
f50f23d9
DG
3914
3915/*
3916 * Send a ret code status message to the sessiond daemon.
3917 *
3918 * Return the sendmsg() return value.
3919 */
3920int consumer_send_status_msg(int sock, int ret_code)
3921{
3922 struct lttcomm_consumer_status_msg msg;
3923
53efb85a 3924 memset(&msg, 0, sizeof(msg));
f50f23d9
DG
3925 msg.ret_code = ret_code;
3926
3927 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3928}
ffe60014
DG
3929
3930/*
3931 * Send a channel status message to the sessiond daemon.
3932 *
3933 * Return the sendmsg() return value.
3934 */
3935int consumer_send_status_channel(int sock,
3936 struct lttng_consumer_channel *channel)
3937{
3938 struct lttcomm_consumer_status_channel msg;
3939
3940 assert(sock >= 0);
3941
53efb85a 3942 memset(&msg, 0, sizeof(msg));
ffe60014 3943 if (!channel) {
0c759fc9 3944 msg.ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
ffe60014 3945 } else {
0c759fc9 3946 msg.ret_code = LTTCOMM_CONSUMERD_SUCCESS;
ffe60014
DG
3947 msg.key = channel->key;
3948 msg.stream_count = channel->streams.count;
3949 }
3950
3951 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3952}
5c786ded 3953
d07ceecd
MD
3954unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
3955 unsigned long produced_pos, uint64_t nb_packets_per_stream,
3956 uint64_t max_sb_size)
5c786ded 3957{
d07ceecd 3958 unsigned long start_pos;
5c786ded 3959
d07ceecd
MD
3960 if (!nb_packets_per_stream) {
3961 return consumed_pos; /* Grab everything */
3962 }
3963 start_pos = produced_pos - offset_align_floor(produced_pos, max_sb_size);
3964 start_pos -= max_sb_size * nb_packets_per_stream;
3965 if ((long) (start_pos - consumed_pos) < 0) {
3966 return consumed_pos; /* Grab everything */
3967 }
3968 return start_pos;
5c786ded 3969}
a1ae2ea5 3970
b99a8d42
JD
3971static
3972int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active)
3973{
3974 int ret = 0;
3975
3976 switch (consumer_data.type) {
3977 case LTTNG_CONSUMER_KERNEL:
3978 ret = kernctl_buffer_flush(stream->wait_fd);
3979 if (ret < 0) {
3980 ERR("Failed to flush kernel stream");
3981 goto end;
3982 }
3983 break;
3984 case LTTNG_CONSUMER32_UST:
3985 case LTTNG_CONSUMER64_UST:
3986 lttng_ustctl_flush_buffer(stream, producer_active);
3987 break;
3988 default:
3989 ERR("Unknown consumer_data type");
3990 abort();
3991 }
3992
3993end:
3994 return ret;
3995}
3996
3997/*
3998 * Sample the rotate position for all the streams of a channel. If a stream
3999 * is already at the rotate position (produced == consumed), we flag it as
4000 * ready for rotation. The rotation of ready streams occurs after we have
4001 * replied to the session daemon that we have finished sampling the positions.
92b7a7f8 4002 * Must be called with RCU read-side lock held to ensure existence of channel.
b99a8d42
JD
4003 *
4004 * Returns 0 on success, < 0 on error
4005 */
92b7a7f8 4006int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
d2956687 4007 uint64_t key, uint64_t relayd_id, uint32_t metadata,
b99a8d42
JD
4008 struct lttng_consumer_local_data *ctx)
4009{
4010 int ret;
b99a8d42
JD
4011 struct lttng_consumer_stream *stream;
4012 struct lttng_ht_iter iter;
4013 struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
4014
4015 DBG("Consumer sample rotate position for channel %" PRIu64, key);
4016
4017 rcu_read_lock();
4018
b99a8d42 4019 pthread_mutex_lock(&channel->lock);
b99a8d42
JD
4020
4021 cds_lfht_for_each_entry_duplicate(ht->ht,
4022 ht->hash_fct(&channel->key, lttng_ht_seed),
4023 ht->match_fct, &channel->key, &iter.iter,
4024 stream, node_channel_id.node) {
4025 unsigned long consumed_pos;
4026
4027 health_code_update();
4028
4029 /*
4030 * Lock stream because we are about to change its state.
4031 */
4032 pthread_mutex_lock(&stream->lock);
4033
b99a8d42
JD
4034 ret = lttng_consumer_sample_snapshot_positions(stream);
4035 if (ret < 0) {
4036 ERR("Failed to sample snapshot position during channel rotation");
4037 goto end_unlock_stream;
4038 }
4039
4040 ret = lttng_consumer_get_produced_snapshot(stream,
4041 &stream->rotate_position);
4042 if (ret < 0) {
4043 ERR("Failed to sample produced position during channel rotation");
4044 goto end_unlock_stream;
4045 }
4046
4047 lttng_consumer_get_consumed_snapshot(stream,
4048 &consumed_pos);
4049 if (consumed_pos == stream->rotate_position) {
4050 stream->rotate_ready = true;
4051 }
b99a8d42
JD
4052
4053 ret = consumer_flush_buffer(stream, 1);
4054 if (ret < 0) {
4055 ERR("Failed to flush stream %" PRIu64 " during channel rotation",
4056 stream->key);
4057 goto end_unlock_stream;
4058 }
4059
4060 pthread_mutex_unlock(&stream->lock);
4061 }
4062 pthread_mutex_unlock(&channel->lock);
4063
4064 ret = 0;
4065 goto end;
4066
4067end_unlock_stream:
4068 pthread_mutex_unlock(&stream->lock);
b99a8d42
JD
4069 pthread_mutex_unlock(&channel->lock);
4070end:
4071 rcu_read_unlock();
4072 return ret;
4073}
4074
02d02e31
JD
4075/*
4076 * Check if a stream is ready to be rotated after extracting it.
4077 *
4078 * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
4079 * error. Stream lock must be held.
4080 */
4081int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
4082{
4083 int ret;
4084 unsigned long consumed_pos;
4085
4086 if (!stream->rotate_position && !stream->rotate_ready) {
4087 ret = 0;
4088 goto end;
4089 }
4090
4091 if (stream->rotate_ready) {
4092 ret = 1;
4093 goto end;
4094 }
4095
4096 /*
4097 * If we don't have the rotate_ready flag, check the consumed position
4098 * to determine if we need to rotate.
4099 */
4100 ret = lttng_consumer_sample_snapshot_positions(stream);
4101 if (ret < 0) {
4102 ERR("Taking snapshot positions");
4103 goto end;
4104 }
4105
4106 ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos);
4107 if (ret < 0) {
4108 ERR("Consumed snapshot position");
4109 goto end;
4110 }
4111
4112 /* Rotate position not reached yet (with check for overflow). */
4113 if ((long) (consumed_pos - stream->rotate_position) < 0) {
4114 ret = 0;
4115 goto end;
4116 }
4117 ret = 1;
4118
4119end:
4120 return ret;
4121}
4122
d73bf3d7
JD
4123/*
4124 * Reset the state for a stream after a rotation occurred.
4125 */
4126void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream)
4127{
4128 stream->rotate_position = 0;
4129 stream->rotate_ready = false;
4130}
4131
4132/*
4133 * Perform the rotation a local stream file.
4134 */
d2956687 4135static
d73bf3d7
JD
4136int rotate_local_stream(struct lttng_consumer_local_data *ctx,
4137 struct lttng_consumer_stream *stream)
4138{
d2956687 4139 int ret = 0;
d73bf3d7 4140
d2956687 4141 DBG("Rotate local stream: stream key %" PRIu64 ", channel key %" PRIu64,
d73bf3d7 4142 stream->key,
d2956687 4143 stream->chan->key);
d73bf3d7 4144 stream->tracefile_size_current = 0;
d2956687 4145 stream->tracefile_count_current = 0;
d73bf3d7 4146
d2956687
JG
4147 if (stream->out_fd >= 0) {
4148 ret = close(stream->out_fd);
4149 if (ret) {
4150 PERROR("Failed to close stream out_fd of channel \"%s\"",
4151 stream->chan->name);
4152 }
4153 stream->out_fd = -1;
4154 }
d73bf3d7 4155
d2956687 4156 if (stream->index_file) {
d73bf3d7 4157 lttng_index_file_put(stream->index_file);
d2956687 4158 stream->index_file = NULL;
d73bf3d7
JD
4159 }
4160
d2956687
JG
4161 if (!stream->trace_chunk) {
4162 goto end;
4163 }
d73bf3d7 4164
d2956687 4165 ret = consumer_stream_create_output_files(stream, true);
d73bf3d7
JD
4166end:
4167 return ret;
d73bf3d7
JD
4168}
4169
4170/*
4171 * Perform the rotation a stream file on the relay.
4172 */
4173int rotate_relay_stream(struct lttng_consumer_local_data *ctx,
4174 struct lttng_consumer_stream *stream)
4175{
4176 int ret;
4177 struct consumer_relayd_sock_pair *relayd;
d2956687
JG
4178 uint64_t chunk_id;
4179 enum lttng_trace_chunk_status chunk_status;
d73bf3d7
JD
4180
4181 DBG("Rotate relay stream");
4182 relayd = consumer_find_relayd(stream->net_seq_idx);
4183 if (!relayd) {
4184 ERR("Failed to find relayd");
4185 ret = -1;
4186 goto end;
4187 }
4188
d2956687
JG
4189 chunk_status = lttng_trace_chunk_get_id(stream->chan->trace_chunk,
4190 &chunk_id);
4191 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4192 ERR("Failed to retrieve the id of the current trace chunk of channel \"%s\"",
4193 stream->chan->name);
4194 ret = -1;
4195 goto end;
4196 }
4197
d73bf3d7
JD
4198 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4199 ret = relayd_rotate_stream(&relayd->control_sock,
4200 stream->relayd_stream_id,
d2956687 4201 chunk_id,
d73bf3d7
JD
4202 stream->last_sequence_number);
4203 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
45f1d9a1
JR
4204 if (ret < 0) {
4205 ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
4206 lttng_consumer_cleanup_relayd(relayd);
4207 }
d73bf3d7
JD
4208 if (ret) {
4209 ERR("Rotate relay stream");
4210 }
4211
4212end:
4213 return ret;
4214}
4215
4216/*
4217 * Performs the stream rotation for the rotate session feature if needed.
d2956687 4218 * It must be called with the channel and stream locks held.
d73bf3d7
JD
4219 *
4220 * Return 0 on success, a negative number of error.
4221 */
4222int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
d2956687 4223 struct lttng_consumer_stream *stream)
d73bf3d7
JD
4224{
4225 int ret;
4226
4227 DBG("Consumer rotate stream %" PRIu64, stream->key);
4228
d2956687
JG
4229 /*
4230 * Update the stream's 'current' chunk to the session's (channel)
4231 * now-current chunk.
4232 */
4233 lttng_trace_chunk_put(stream->trace_chunk);
4234 if (stream->chan->trace_chunk == stream->trace_chunk) {
4235 /*
4236 * A channel can be rotated and not have a "next" chunk
4237 * to transition to. In that case, the channel's "current chunk"
4238 * has not been closed yet, but it has not been updated to
4239 * a "next" trace chunk either. Hence, the stream, like its
4240 * parent channel, becomes part of no chunk and can't output
4241 * anything until a new trace chunk is created.
4242 */
4243 stream->trace_chunk = NULL;
4244 } else if (stream->chan->trace_chunk &&
4245 !lttng_trace_chunk_get(stream->chan->trace_chunk)) {
4246 ERR("Failed to acquire a reference to channel's trace chunk during stream rotation");
4247 ret = -1;
4248 goto error;
4249 } else {
4250 /*
4251 * Update the stream's trace chunk to its parent channel's
4252 * current trace chunk.
4253 */
4254 stream->trace_chunk = stream->chan->trace_chunk;
4255 }
4256
d73bf3d7
JD
4257 if (stream->net_seq_idx != (uint64_t) -1ULL) {
4258 ret = rotate_relay_stream(ctx, stream);
4259 } else {
4260 ret = rotate_local_stream(ctx, stream);
4261 }
4262 if (ret < 0) {
92816cc3 4263 ERR("Failed to rotate stream, ret = %i", ret);
d73bf3d7
JD
4264 goto error;
4265 }
4266
d2956687
JG
4267 if (stream->metadata_flag && stream->trace_chunk) {
4268 /*
4269 * If the stream has transitioned to a new trace
4270 * chunk, the metadata should be re-dumped to the
4271 * newest chunk.
4272 *
4273 * However, it is possible for a stream to transition to
4274 * a "no-chunk" state. This can happen if a rotation
4275 * occurs on an inactive session. In such cases, the metadata
4276 * regeneration will happen when the next trace chunk is
4277 * created.
4278 */
4279 ret = consumer_metadata_stream_dump(stream);
4280 if (ret) {
4281 goto error;
d73bf3d7
JD
4282 }
4283 }
4284 lttng_consumer_reset_stream_rotate_state(stream);
4285
4286 ret = 0;
4287
4288error:
4289 return ret;
4290}
4291
b99a8d42
JD
4292/*
4293 * Rotate all the ready streams now.
4294 *
4295 * This is especially important for low throughput streams that have already
4296 * been consumed, we cannot wait for their next packet to perform the
4297 * rotation.
92b7a7f8
MD
4298 * Need to be called with RCU read-side lock held to ensure existence of
4299 * channel.
b99a8d42
JD
4300 *
4301 * Returns 0 on success, < 0 on error
4302 */
92b7a7f8
MD
4303int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel,
4304 uint64_t key, struct lttng_consumer_local_data *ctx)
b99a8d42
JD
4305{
4306 int ret;
b99a8d42
JD
4307 struct lttng_consumer_stream *stream;
4308 struct lttng_ht_iter iter;
4309 struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
4310
4311 rcu_read_lock();
4312
4313 DBG("Consumer rotate ready streams in channel %" PRIu64, key);
4314
b99a8d42
JD
4315 cds_lfht_for_each_entry_duplicate(ht->ht,
4316 ht->hash_fct(&channel->key, lttng_ht_seed),
4317 ht->match_fct, &channel->key, &iter.iter,
4318 stream, node_channel_id.node) {
4319 health_code_update();
4320
d2956687 4321 pthread_mutex_lock(&stream->chan->lock);
b99a8d42
JD
4322 pthread_mutex_lock(&stream->lock);
4323
4324 if (!stream->rotate_ready) {
4325 pthread_mutex_unlock(&stream->lock);
d2956687 4326 pthread_mutex_unlock(&stream->chan->lock);
b99a8d42
JD
4327 continue;
4328 }
4329 DBG("Consumer rotate ready stream %" PRIu64, stream->key);
4330
d2956687 4331 ret = lttng_consumer_rotate_stream(ctx, stream);
b99a8d42 4332 pthread_mutex_unlock(&stream->lock);
d2956687 4333 pthread_mutex_unlock(&stream->chan->lock);
b99a8d42
JD
4334 if (ret) {
4335 goto end;
4336 }
4337 }
4338
4339 ret = 0;
4340
4341end:
4342 rcu_read_unlock();
4343 return ret;
4344}
4345
d2956687
JG
4346enum lttcomm_return_code lttng_consumer_init_command(
4347 struct lttng_consumer_local_data *ctx,
4348 const lttng_uuid sessiond_uuid)
00fb02ac 4349{
d2956687
JG
4350 enum lttcomm_return_code ret;
4351 char uuid_str[UUID_STR_LEN];
00fb02ac 4352
d2956687
JG
4353 if (ctx->sessiond_uuid.is_set) {
4354 ret = LTTCOMM_CONSUMERD_ALREADY_SET;
00fb02ac
JD
4355 goto end;
4356 }
4357
d2956687
JG
4358 ctx->sessiond_uuid.is_set = true;
4359 memcpy(ctx->sessiond_uuid.value, sessiond_uuid, sizeof(lttng_uuid));
4360 ret = LTTCOMM_CONSUMERD_SUCCESS;
4361 lttng_uuid_to_str(sessiond_uuid, uuid_str);
4362 DBG("Received session daemon UUID: %s", uuid_str);
00fb02ac
JD
4363end:
4364 return ret;
4365}
4366
d2956687
JG
4367enum lttcomm_return_code lttng_consumer_create_trace_chunk(
4368 const uint64_t *relayd_id, uint64_t session_id,
4369 uint64_t chunk_id,
4370 time_t chunk_creation_timestamp,
4371 const char *chunk_override_name,
4372 const struct lttng_credentials *credentials,
4373 struct lttng_directory_handle *chunk_directory_handle)
00fb02ac
JD
4374{
4375 int ret;
d2956687
JG
4376 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
4377 struct lttng_trace_chunk *created_chunk, *published_chunk;
4378 enum lttng_trace_chunk_status chunk_status;
4379 char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
4380 char creation_timestamp_buffer[ISO8601_STR_LEN];
4381 const char *relayd_id_str = "(none)";
4382 const char *creation_timestamp_str;
4383 struct lttng_ht_iter iter;
4384 struct lttng_consumer_channel *channel;
92816cc3 4385
d2956687
JG
4386 if (relayd_id) {
4387 /* Only used for logging purposes. */
4388 ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
4389 "%" PRIu64, *relayd_id);
4390 if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
4391 relayd_id_str = relayd_id_buffer;
4392 } else {
4393 relayd_id_str = "(formatting error)";
4394 }
4395 }
4396
4397 /* Local protocol error. */
4398 assert(chunk_creation_timestamp);
4399 ret = time_to_iso8601_str(chunk_creation_timestamp,
4400 creation_timestamp_buffer,
4401 sizeof(creation_timestamp_buffer));
4402 creation_timestamp_str = !ret ? creation_timestamp_buffer :
4403 "(formatting error)";
4404
4405 DBG("Consumer create trace chunk command: relay_id = %s"
4406 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
4407 ", chunk_override_name = %s"
4408 ", chunk_creation_timestamp = %s",
4409 relayd_id_str, session_id, chunk_id,
4410 chunk_override_name ? : "(none)",
4411 creation_timestamp_str);
92816cc3
JG
4412
4413 /*
d2956687
JG
4414 * The trace chunk registry, as used by the consumer daemon, implicitly
4415 * owns the trace chunks. This is only needed in the consumer since
4416 * the consumer has no notion of a session beyond session IDs being
4417 * used to identify other objects.
4418 *
4419 * The lttng_trace_chunk_registry_publish() call below provides a
4420 * reference which is not released; it implicitly becomes the session
4421 * daemon's reference to the chunk in the consumer daemon.
4422 *
4423 * The lifetime of trace chunks in the consumer daemon is managed by
4424 * the session daemon through the LTTNG_CONSUMER_CREATE_TRACE_CHUNK
4425 * and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands.
92816cc3 4426 */
d2956687
JG
4427 created_chunk = lttng_trace_chunk_create(chunk_id,
4428 chunk_creation_timestamp);
4429 if (!created_chunk) {
4430 ERR("Failed to create trace chunk");
4431 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4432 goto end;
4433 }
92816cc3 4434
d2956687
JG
4435 if (chunk_override_name) {
4436 chunk_status = lttng_trace_chunk_override_name(created_chunk,
4437 chunk_override_name);
4438 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4439 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
92816cc3
JG
4440 goto end;
4441 }
4442 }
4443
d2956687
JG
4444 if (chunk_directory_handle) {
4445 chunk_status = lttng_trace_chunk_set_credentials(created_chunk,
4446 credentials);
4447 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4448 ERR("Failed to set trace chunk credentials");
4449 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4450 goto end;
4451 }
4452 /*
4453 * The consumer daemon has no ownership of the chunk output
4454 * directory.
4455 */
4456 chunk_status = lttng_trace_chunk_set_as_user(created_chunk,
4457 chunk_directory_handle);
4458 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4459 ERR("Failed to set trace chunk's directory handle");
4460 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
92816cc3
JG
4461 goto end;
4462 }
4463 }
4464
d2956687
JG
4465 published_chunk = lttng_trace_chunk_registry_publish_chunk(
4466 consumer_data.chunk_registry, session_id,
4467 created_chunk);
4468 lttng_trace_chunk_put(created_chunk);
4469 created_chunk = NULL;
4470 if (!published_chunk) {
4471 ERR("Failed to publish trace chunk");
4472 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
d88744a4
JD
4473 goto end;
4474 }
4475
d2956687
JG
4476 rcu_read_lock();
4477 cds_lfht_for_each_entry_duplicate(consumer_data.channels_by_session_id_ht->ht,
4478 consumer_data.channels_by_session_id_ht->hash_fct(
4479 &session_id, lttng_ht_seed),
4480 consumer_data.channels_by_session_id_ht->match_fct,
4481 &session_id, &iter.iter, channel,
4482 channels_by_session_id_ht_node.node) {
4483 ret = lttng_consumer_channel_set_trace_chunk(channel,
4484 published_chunk);
4485 if (ret) {
4486 /*
4487 * Roll-back the creation of this chunk.
4488 *
4489 * This is important since the session daemon will
4490 * assume that the creation of this chunk failed and
4491 * will never ask for it to be closed, resulting
4492 * in a leak and an inconsistent state for some
4493 * channels.
4494 */
4495 enum lttcomm_return_code close_ret;
4496
4497 DBG("Failed to set new trace chunk on existing channels, rolling back");
4498 close_ret = lttng_consumer_close_trace_chunk(relayd_id,
4499 session_id, chunk_id,
4500 chunk_creation_timestamp);
4501 if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
4502 ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
4503 session_id, chunk_id);
4504 }
a1ae2ea5 4505
d2956687
JG
4506 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4507 break;
4508 }
a1ae2ea5
JD
4509 }
4510
e5add6d0
JG
4511 if (relayd_id) {
4512 struct consumer_relayd_sock_pair *relayd;
4513
4514 relayd = consumer_find_relayd(*relayd_id);
4515 if (relayd) {
4516 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4517 ret = relayd_create_trace_chunk(
4518 &relayd->control_sock, published_chunk);
4519 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4520 } else {
4521 ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64, *relayd_id);
4522 }
4523
4524 if (!relayd || ret) {
4525 enum lttcomm_return_code close_ret;
4526
4527 close_ret = lttng_consumer_close_trace_chunk(relayd_id,
4528 session_id,
4529 chunk_id,
4530 chunk_creation_timestamp);
4531 if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
4532 ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
4533 session_id,
4534 chunk_id);
4535 }
4536
4537 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4538 goto error;
4539 }
4540 }
4541error:
4542 rcu_read_unlock();
d2956687
JG
4543 /* Release the reference returned by the "publish" operation. */
4544 lttng_trace_chunk_put(published_chunk);
a1ae2ea5 4545end:
d2956687 4546 return ret_code;
a1ae2ea5
JD
4547}
4548
d2956687
JG
4549enum lttcomm_return_code lttng_consumer_close_trace_chunk(
4550 const uint64_t *relayd_id, uint64_t session_id,
4551 uint64_t chunk_id, time_t chunk_close_timestamp)
a1ae2ea5 4552{
d2956687
JG
4553 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
4554 struct lttng_trace_chunk *chunk;
4555 char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
4556 const char *relayd_id_str = "(none)";
4557 struct lttng_ht_iter iter;
4558 struct lttng_consumer_channel *channel;
4559 enum lttng_trace_chunk_status chunk_status;
a1ae2ea5 4560
d2956687
JG
4561 if (relayd_id) {
4562 int ret;
4563
4564 /* Only used for logging purposes. */
4565 ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
4566 "%" PRIu64, *relayd_id);
4567 if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
4568 relayd_id_str = relayd_id_buffer;
4569 } else {
4570 relayd_id_str = "(formatting error)";
4571 }
4572 }
4573
4574 DBG("Consumer close trace chunk command: relayd_id = %s"
4575 ", session_id = %" PRIu64
4576 ", chunk_id = %" PRIu64, relayd_id_str,
4577 session_id, chunk_id);
4578 chunk = lttng_trace_chunk_registry_find_chunk(
4579 consumer_data.chunk_registry, session_id,
4580 chunk_id);
4581 if (!chunk) {
4582 ERR("Failed to find chunk: session_id = %" PRIu64
4583 ", chunk_id = %" PRIu64,
4584 session_id, chunk_id);
4585 ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
a1ae2ea5
JD
4586 goto end;
4587 }
4588
d2956687
JG
4589 chunk_status = lttng_trace_chunk_set_close_timestamp(chunk,
4590 chunk_close_timestamp);
4591 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4592 ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
4593 goto end;
45f1d9a1 4594 }
d2956687
JG
4595 /*
4596 * Release the reference returned by the "find" operation and
4597 * the session daemon's implicit reference to the chunk.
4598 */
4599 lttng_trace_chunk_put(chunk);
4600 lttng_trace_chunk_put(chunk);
a1ae2ea5 4601
d2956687
JG
4602 /*
4603 * chunk is now invalid to access as we no longer hold a reference to
4604 * it; it is only kept around to compare it (by address) to the
4605 * current chunk found in the session's channels.
4606 */
4607 rcu_read_lock();
4608 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter,
4609 channel, node.node) {
4610 int ret;
a1ae2ea5 4611
d2956687
JG
4612 /*
4613 * Only change the channel's chunk to NULL if it still
4614 * references the chunk being closed. The channel may
4615 * reference a newer channel in the case of a session
4616 * rotation. When a session rotation occurs, the "next"
4617 * chunk is created before the "current" chunk is closed.
4618 */
4619 if (channel->trace_chunk != chunk) {
4620 continue;
4621 }
4622 ret = lttng_consumer_channel_set_trace_chunk(channel, NULL);
4623 if (ret) {
4624 /*
4625 * Attempt to close the chunk on as many channels as
4626 * possible.
4627 */
4628 ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
4629 }
a1ae2ea5 4630 }
d2956687
JG
4631 rcu_read_unlock();
4632end:
4633 return ret_code;
a1ae2ea5 4634}
3654ed19 4635
d2956687
JG
4636enum lttcomm_return_code lttng_consumer_trace_chunk_exists(
4637 const uint64_t *relayd_id, uint64_t session_id,
4638 uint64_t chunk_id)
3654ed19 4639{
d2956687
JG
4640 enum lttcomm_return_code ret_code;
4641 struct lttng_trace_chunk *chunk;
4642 char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
4643 const char *relayd_id_str = "(none)";
4644
4645 if (relayd_id) {
4646 int ret;
4647
4648 /* Only used for logging purposes. */
4649 ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
4650 "%" PRIu64, *relayd_id);
4651 if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
4652 relayd_id_str = relayd_id_buffer;
4653 } else {
4654 relayd_id_str = "(formatting error)";
4655 }
4656 }
4657
4658 DBG("Consumer trace chunk exists command: relayd_id = %s"
4659 ", session_id = %" PRIu64
4660 ", chunk_id = %" PRIu64, relayd_id_str,
4661 session_id, chunk_id);
4662 chunk = lttng_trace_chunk_registry_find_chunk(
4663 consumer_data.chunk_registry, session_id,
4664 chunk_id);
4665 DBG("Trace chunk %s locally", chunk ? "exists" : "does not exist");
4666 ret_code = chunk ? LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL :
4667 LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
4668
4669 lttng_trace_chunk_put(chunk);
4670 return ret_code;
3654ed19 4671}
This page took 0.349311 seconds and 4 git commands to generate.