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