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