Fix: rotation of a stopped session hangs indifinitely
[lttng-tools.git] / src / common / consumer / consumer.c
CommitLineData
3bd1e081
MD
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
00e2e675 4 * 2012 - David Goulet <dgoulet@efficios.com>
3bd1e081 5 *
d14d33bf
AM
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
3bd1e081 9 *
d14d33bf
AM
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * more details.
3bd1e081 14 *
d14d33bf
AM
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
18 */
19
6c1c0768 20#define _LGPL_SOURCE
3bd1e081 21#include <assert.h>
3bd1e081
MD
22#include <poll.h>
23#include <pthread.h>
24#include <stdlib.h>
25#include <string.h>
26#include <sys/mman.h>
27#include <sys/socket.h>
28#include <sys/types.h>
29#include <unistd.h>
77c7c900 30#include <inttypes.h>
331744e3 31#include <signal.h>
3bd1e081 32
51a9e1c7 33#include <bin/lttng-consumerd/health-consumerd.h>
990570ed 34#include <common/common.h>
fb3a43a9 35#include <common/utils.h>
d2956687 36#include <common/time.h>
fb3a43a9 37#include <common/compat/poll.h>
f263b7fd 38#include <common/compat/endian.h>
309167d2 39#include <common/index/index.h>
10a8a223 40#include <common/kernel-ctl/kernel-ctl.h>
00e2e675 41#include <common/sessiond-comm/relayd.h>
10a8a223
DG
42#include <common/sessiond-comm/sessiond-comm.h>
43#include <common/kernel-consumer/kernel-consumer.h>
00e2e675 44#include <common/relayd/relayd.h>
10a8a223 45#include <common/ust-consumer/ust-consumer.h>
c8fea79c
JR
46#include <common/consumer/consumer-timer.h>
47#include <common/consumer/consumer.h>
48#include <common/consumer/consumer-stream.h>
49#include <common/consumer/consumer-testpoint.h>
50#include <common/align.h>
5feafd41 51#include <common/consumer/consumer-metadata-cache.h>
d2956687
JG
52#include <common/trace-chunk.h>
53#include <common/trace-chunk-registry.h>
54#include <common/string-utils/format.h>
c35f9726 55#include <common/dynamic-array.h>
3bd1e081
MD
56
57struct lttng_consumer_global_data consumer_data = {
3bd1e081
MD
58 .stream_count = 0,
59 .need_update = 1,
60 .type = LTTNG_CONSUMER_UNKNOWN,
61};
62
d8ef542d
MD
63enum consumer_channel_action {
64 CONSUMER_CHANNEL_ADD,
a0cbdd2e 65 CONSUMER_CHANNEL_DEL,
d8ef542d
MD
66 CONSUMER_CHANNEL_QUIT,
67};
68
69struct consumer_channel_msg {
70 enum consumer_channel_action action;
a0cbdd2e
MD
71 struct lttng_consumer_channel *chan; /* add */
72 uint64_t key; /* del */
d8ef542d
MD
73};
74
80957876 75/* Flag used to temporarily pause data consumption from testpoints. */
cf0bcb51
JG
76int data_consumption_paused;
77
3bd1e081
MD
78/*
79 * Flag to inform the polling thread to quit when all fd hung up. Updated by
80 * the consumer_thread_receive_fds when it notices that all fds has hung up.
81 * Also updated by the signal handler (consumer_should_exit()). Read by the
82 * polling threads.
83 */
10211f5c 84int consumer_quit;
3bd1e081 85
43c34bc3 86/*
43c34bc3
DG
87 * Global hash table containing respectively metadata and data streams. The
88 * stream element in this ht should only be updated by the metadata poll thread
89 * for the metadata and the data poll thread for the data.
90 */
40dc48e0
DG
91static struct lttng_ht *metadata_ht;
92static struct lttng_ht *data_ht;
43c34bc3 93
acdb9057
DG
94/*
95 * Notify a thread lttng pipe to poll back again. This usually means that some
96 * global state has changed so we just send back the thread in a poll wait
97 * call.
98 */
99static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
100{
101 struct lttng_consumer_stream *null_stream = NULL;
102
103 assert(pipe);
104
105 (void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
106}
107
5c635c72
MD
108static void notify_health_quit_pipe(int *pipe)
109{
6cd525e8 110 ssize_t ret;
5c635c72 111
6cd525e8
MD
112 ret = lttng_write(pipe[1], "4", 1);
113 if (ret < 1) {
5c635c72
MD
114 PERROR("write consumer health quit");
115 }
116}
117
d8ef542d
MD
118static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
119 struct lttng_consumer_channel *chan,
a0cbdd2e 120 uint64_t key,
d8ef542d
MD
121 enum consumer_channel_action action)
122{
123 struct consumer_channel_msg msg;
6cd525e8 124 ssize_t ret;
d8ef542d 125
e56251fc
DG
126 memset(&msg, 0, sizeof(msg));
127
d8ef542d
MD
128 msg.action = action;
129 msg.chan = chan;
f21dae48 130 msg.key = key;
6cd525e8
MD
131 ret = lttng_write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
132 if (ret < sizeof(msg)) {
133 PERROR("notify_channel_pipe write error");
134 }
d8ef542d
MD
135}
136
a0cbdd2e
MD
137void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
138 uint64_t key)
139{
140 notify_channel_pipe(ctx, NULL, key, CONSUMER_CHANNEL_DEL);
141}
142
d8ef542d
MD
143static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
144 struct lttng_consumer_channel **chan,
a0cbdd2e 145 uint64_t *key,
d8ef542d
MD
146 enum consumer_channel_action *action)
147{
148 struct consumer_channel_msg msg;
6cd525e8 149 ssize_t ret;
d8ef542d 150
6cd525e8
MD
151 ret = lttng_read(ctx->consumer_channel_pipe[0], &msg, sizeof(msg));
152 if (ret < sizeof(msg)) {
153 ret = -1;
154 goto error;
d8ef542d 155 }
6cd525e8
MD
156 *action = msg.action;
157 *chan = msg.chan;
158 *key = msg.key;
159error:
160 return (int) ret;
d8ef542d
MD
161}
162
212d67a2
DG
163/*
164 * Cleanup the stream list of a channel. Those streams are not yet globally
165 * visible
166 */
167static void clean_channel_stream_list(struct lttng_consumer_channel *channel)
168{
169 struct lttng_consumer_stream *stream, *stmp;
170
171 assert(channel);
172
173 /* Delete streams that might have been left in the stream list. */
174 cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
175 send_node) {
176 cds_list_del(&stream->send_node);
177 /*
178 * Once a stream is added to this list, the buffers were created so we
179 * have a guarantee that this call will succeed. Setting the monitor
180 * mode to 0 so we don't lock nor try to delete the stream from the
181 * global hash table.
182 */
183 stream->monitor = 0;
184 consumer_stream_destroy(stream, NULL);
185 }
186}
187
3bd1e081
MD
188/*
189 * Find a stream. The consumer_data.lock must be locked during this
190 * call.
191 */
d88aee68 192static struct lttng_consumer_stream *find_stream(uint64_t key,
8389e4f8 193 struct lttng_ht *ht)
3bd1e081 194{
e4421fec 195 struct lttng_ht_iter iter;
d88aee68 196 struct lttng_ht_node_u64 *node;
e4421fec 197 struct lttng_consumer_stream *stream = NULL;
3bd1e081 198
8389e4f8
DG
199 assert(ht);
200
d88aee68
DG
201 /* -1ULL keys are lookup failures */
202 if (key == (uint64_t) -1ULL) {
7ad0a0cb 203 return NULL;
7a57cf92 204 }
e4421fec 205
6065ceec
DG
206 rcu_read_lock();
207
d88aee68
DG
208 lttng_ht_lookup(ht, &key, &iter);
209 node = lttng_ht_iter_get_node_u64(&iter);
e4421fec
DG
210 if (node != NULL) {
211 stream = caa_container_of(node, struct lttng_consumer_stream, node);
3bd1e081 212 }
e4421fec 213
6065ceec
DG
214 rcu_read_unlock();
215
e4421fec 216 return stream;
3bd1e081
MD
217}
218
da009f2c 219static void steal_stream_key(uint64_t key, struct lttng_ht *ht)
7ad0a0cb
MD
220{
221 struct lttng_consumer_stream *stream;
222
04253271 223 rcu_read_lock();
ffe60014 224 stream = find_stream(key, ht);
04253271 225 if (stream) {
da009f2c 226 stream->key = (uint64_t) -1ULL;
04253271
MD
227 /*
228 * We don't want the lookup to match, but we still need
229 * to iterate on this stream when iterating over the hash table. Just
230 * change the node key.
231 */
da009f2c 232 stream->node.key = (uint64_t) -1ULL;
04253271
MD
233 }
234 rcu_read_unlock();
7ad0a0cb
MD
235}
236
d56db448
DG
237/*
238 * Return a channel object for the given key.
239 *
240 * RCU read side lock MUST be acquired before calling this function and
241 * protects the channel ptr.
242 */
d88aee68 243struct lttng_consumer_channel *consumer_find_channel(uint64_t key)
3bd1e081 244{
e4421fec 245 struct lttng_ht_iter iter;
d88aee68 246 struct lttng_ht_node_u64 *node;
e4421fec 247 struct lttng_consumer_channel *channel = NULL;
3bd1e081 248
d88aee68
DG
249 /* -1ULL keys are lookup failures */
250 if (key == (uint64_t) -1ULL) {
7ad0a0cb 251 return NULL;
7a57cf92 252 }
e4421fec 253
d88aee68
DG
254 lttng_ht_lookup(consumer_data.channel_ht, &key, &iter);
255 node = lttng_ht_iter_get_node_u64(&iter);
e4421fec
DG
256 if (node != NULL) {
257 channel = caa_container_of(node, struct lttng_consumer_channel, node);
3bd1e081 258 }
e4421fec
DG
259
260 return channel;
3bd1e081
MD
261}
262
b5a6470f
DG
263/*
264 * There is a possibility that the consumer does not have enough time between
265 * the close of the channel on the session daemon and the cleanup in here thus
266 * once we have a channel add with an existing key, we know for sure that this
267 * channel will eventually get cleaned up by all streams being closed.
268 *
269 * This function just nullifies the already existing channel key.
270 */
271static void steal_channel_key(uint64_t key)
272{
273 struct lttng_consumer_channel *channel;
274
275 rcu_read_lock();
276 channel = consumer_find_channel(key);
277 if (channel) {
278 channel->key = (uint64_t) -1ULL;
279 /*
280 * We don't want the lookup to match, but we still need to iterate on
281 * this channel when iterating over the hash table. Just change the
282 * node key.
283 */
284 channel->node.key = (uint64_t) -1ULL;
285 }
286 rcu_read_unlock();
287}
288
ffe60014 289static void free_channel_rcu(struct rcu_head *head)
702b1ea4 290{
d88aee68
DG
291 struct lttng_ht_node_u64 *node =
292 caa_container_of(head, struct lttng_ht_node_u64, head);
ffe60014
DG
293 struct lttng_consumer_channel *channel =
294 caa_container_of(node, struct lttng_consumer_channel, node);
702b1ea4 295
b83e03c4
MD
296 switch (consumer_data.type) {
297 case LTTNG_CONSUMER_KERNEL:
298 break;
299 case LTTNG_CONSUMER32_UST:
300 case LTTNG_CONSUMER64_UST:
301 lttng_ustconsumer_free_channel(channel);
302 break;
303 default:
304 ERR("Unknown consumer_data type");
305 abort();
306 }
ffe60014 307 free(channel);
702b1ea4
MD
308}
309
00e2e675
DG
310/*
311 * RCU protected relayd socket pair free.
312 */
ffe60014 313static void free_relayd_rcu(struct rcu_head *head)
00e2e675 314{
d88aee68
DG
315 struct lttng_ht_node_u64 *node =
316 caa_container_of(head, struct lttng_ht_node_u64, head);
00e2e675
DG
317 struct consumer_relayd_sock_pair *relayd =
318 caa_container_of(node, struct consumer_relayd_sock_pair, node);
319
8994307f
DG
320 /*
321 * Close all sockets. This is done in the call RCU since we don't want the
322 * socket fds to be reassigned thus potentially creating bad state of the
323 * relayd object.
324 *
325 * We do not have to lock the control socket mutex here since at this stage
326 * there is no one referencing to this relayd object.
327 */
328 (void) relayd_close(&relayd->control_sock);
329 (void) relayd_close(&relayd->data_sock);
330
3a84e2f3 331 pthread_mutex_destroy(&relayd->ctrl_sock_mutex);
00e2e675
DG
332 free(relayd);
333}
334
335/*
336 * Destroy and free relayd socket pair object.
00e2e675 337 */
51230d70 338void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
339{
340 int ret;
341 struct lttng_ht_iter iter;
342
173af62f
DG
343 if (relayd == NULL) {
344 return;
345 }
346
00e2e675
DG
347 DBG("Consumer destroy and close relayd socket pair");
348
349 iter.iter.node = &relayd->node.node;
350 ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
173af62f 351 if (ret != 0) {
8994307f 352 /* We assume the relayd is being or is destroyed */
173af62f
DG
353 return;
354 }
00e2e675 355
00e2e675 356 /* RCU free() call */
ffe60014
DG
357 call_rcu(&relayd->node.head, free_relayd_rcu);
358}
359
360/*
361 * Remove a channel from the global list protected by a mutex. This function is
362 * also responsible for freeing its data structures.
363 */
364void consumer_del_channel(struct lttng_consumer_channel *channel)
365{
ffe60014
DG
366 struct lttng_ht_iter iter;
367
d88aee68 368 DBG("Consumer delete channel key %" PRIu64, channel->key);
ffe60014
DG
369
370 pthread_mutex_lock(&consumer_data.lock);
a9838785 371 pthread_mutex_lock(&channel->lock);
ffe60014 372
212d67a2
DG
373 /* Destroy streams that might have been left in the stream list. */
374 clean_channel_stream_list(channel);
51e762e5 375
d3e2ba59
JD
376 if (channel->live_timer_enabled == 1) {
377 consumer_timer_live_stop(channel);
378 }
e9404c27
JG
379 if (channel->monitor_timer_enabled == 1) {
380 consumer_timer_monitor_stop(channel);
381 }
d3e2ba59 382
ffe60014
DG
383 switch (consumer_data.type) {
384 case LTTNG_CONSUMER_KERNEL:
385 break;
386 case LTTNG_CONSUMER32_UST:
387 case LTTNG_CONSUMER64_UST:
388 lttng_ustconsumer_del_channel(channel);
389 break;
390 default:
391 ERR("Unknown consumer_data type");
392 assert(0);
393 goto end;
394 }
395
d2956687
JG
396 lttng_trace_chunk_put(channel->trace_chunk);
397 channel->trace_chunk = NULL;
5c3892a6 398
d2956687
JG
399 if (channel->is_published) {
400 int ret;
401
402 rcu_read_lock();
403 iter.iter.node = &channel->node.node;
404 ret = lttng_ht_del(consumer_data.channel_ht, &iter);
405 assert(!ret);
ffe60014 406
d2956687
JG
407 iter.iter.node = &channel->channels_by_session_id_ht_node.node;
408 ret = lttng_ht_del(consumer_data.channels_by_session_id_ht,
409 &iter);
410 assert(!ret);
411 rcu_read_unlock();
412 }
413
414 call_rcu(&channel->node.head, free_channel_rcu);
ffe60014 415end:
a9838785 416 pthread_mutex_unlock(&channel->lock);
ffe60014 417 pthread_mutex_unlock(&consumer_data.lock);
00e2e675
DG
418}
419
228b5bf7
DG
420/*
421 * Iterate over the relayd hash table and destroy each element. Finally,
422 * destroy the whole hash table.
423 */
424static void cleanup_relayd_ht(void)
425{
426 struct lttng_ht_iter iter;
427 struct consumer_relayd_sock_pair *relayd;
428
429 rcu_read_lock();
430
431 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
432 node.node) {
51230d70 433 consumer_destroy_relayd(relayd);
228b5bf7
DG
434 }
435
228b5bf7 436 rcu_read_unlock();
36b588ed
MD
437
438 lttng_ht_destroy(consumer_data.relayd_ht);
228b5bf7
DG
439}
440
8994307f
DG
441/*
442 * Update the end point status of all streams having the given network sequence
443 * index (relayd index).
444 *
445 * It's atomically set without having the stream mutex locked which is fine
446 * because we handle the write/read race with a pipe wakeup for each thread.
447 */
da009f2c 448static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
8994307f
DG
449 enum consumer_endpoint_status status)
450{
451 struct lttng_ht_iter iter;
452 struct lttng_consumer_stream *stream;
453
da009f2c 454 DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
8994307f
DG
455
456 rcu_read_lock();
457
458 /* Let's begin with metadata */
459 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
460 if (stream->net_seq_idx == net_seq_idx) {
461 uatomic_set(&stream->endpoint_status, status);
462 DBG("Delete flag set to metadata stream %d", stream->wait_fd);
463 }
464 }
465
466 /* Follow up by the data streams */
467 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
468 if (stream->net_seq_idx == net_seq_idx) {
469 uatomic_set(&stream->endpoint_status, status);
470 DBG("Delete flag set to data stream %d", stream->wait_fd);
471 }
472 }
473 rcu_read_unlock();
474}
475
476/*
477 * Cleanup a relayd object by flagging every associated streams for deletion,
478 * destroying the object meaning removing it from the relayd hash table,
479 * closing the sockets and freeing the memory in a RCU call.
480 *
481 * If a local data context is available, notify the threads that the streams'
482 * state have changed.
483 */
9276e5c8 484void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
8994307f 485{
da009f2c 486 uint64_t netidx;
8994307f
DG
487
488 assert(relayd);
489
9276e5c8 490 DBG("Cleaning up relayd object ID %"PRIu64, relayd->net_seq_idx);
9617607b 491
8994307f
DG
492 /* Save the net sequence index before destroying the object */
493 netidx = relayd->net_seq_idx;
494
495 /*
496 * Delete the relayd from the relayd hash table, close the sockets and free
497 * the object in a RCU call.
498 */
51230d70 499 consumer_destroy_relayd(relayd);
8994307f
DG
500
501 /* Set inactive endpoint to all streams */
502 update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE);
503
504 /*
505 * With a local data context, notify the threads that the streams' state
506 * have changed. The write() action on the pipe acts as an "implicit"
507 * memory barrier ordering the updates of the end point status from the
508 * read of this status which happens AFTER receiving this notify.
509 */
9276e5c8
JR
510 notify_thread_lttng_pipe(relayd->ctx->consumer_data_pipe);
511 notify_thread_lttng_pipe(relayd->ctx->consumer_metadata_pipe);
8994307f
DG
512}
513
a6ba4fe1
DG
514/*
515 * Flag a relayd socket pair for destruction. Destroy it if the refcount
516 * reaches zero.
517 *
518 * RCU read side lock MUST be aquired before calling this function.
519 */
520void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
521{
522 assert(relayd);
523
524 /* Set destroy flag for this object */
525 uatomic_set(&relayd->destroy_flag, 1);
526
527 /* Destroy the relayd if refcount is 0 */
528 if (uatomic_read(&relayd->refcount) == 0) {
51230d70 529 consumer_destroy_relayd(relayd);
a6ba4fe1
DG
530 }
531}
532
3bd1e081 533/*
1d1a276c
DG
534 * Completly destroy stream from every visiable data structure and the given
535 * hash table if one.
536 *
537 * One this call returns, the stream object is not longer usable nor visible.
3bd1e081 538 */
e316aad5
DG
539void consumer_del_stream(struct lttng_consumer_stream *stream,
540 struct lttng_ht *ht)
3bd1e081 541{
1d1a276c 542 consumer_stream_destroy(stream, ht);
3bd1e081
MD
543}
544
5ab66908
MD
545/*
546 * XXX naming of del vs destroy is all mixed up.
547 */
548void consumer_del_stream_for_data(struct lttng_consumer_stream *stream)
549{
550 consumer_stream_destroy(stream, data_ht);
551}
552
553void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
554{
555 consumer_stream_destroy(stream, metadata_ht);
556}
557
d9a2e16e
JD
558void consumer_stream_update_channel_attributes(
559 struct lttng_consumer_stream *stream,
560 struct lttng_consumer_channel *channel)
561{
562 stream->channel_read_only_attributes.tracefile_size =
563 channel->tracefile_size;
d9a2e16e
JD
564}
565
d88aee68
DG
566struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
567 uint64_t stream_key,
ffe60014 568 const char *channel_name,
57a269f2 569 uint64_t relayd_id,
53632229 570 uint64_t session_id,
d2956687 571 struct lttng_trace_chunk *trace_chunk,
ffe60014
DG
572 int cpu,
573 int *alloc_ret,
4891ece8 574 enum consumer_channel_type type,
d2956687 575 unsigned int monitor)
3bd1e081 576{
ffe60014 577 int ret;
3bd1e081 578 struct lttng_consumer_stream *stream;
3bd1e081 579
effcf122 580 stream = zmalloc(sizeof(*stream));
3bd1e081 581 if (stream == NULL) {
7a57cf92 582 PERROR("malloc struct lttng_consumer_stream");
ffe60014 583 ret = -ENOMEM;
7a57cf92 584 goto end;
3bd1e081 585 }
7a57cf92 586
d2956687
JG
587 if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
588 ERR("Failed to acquire trace chunk reference during the creation of a stream");
589 ret = -1;
590 goto error;
591 }
d56db448 592
d2956687 593 rcu_read_lock();
3bd1e081 594 stream->key = stream_key;
d2956687 595 stream->trace_chunk = trace_chunk;
3bd1e081
MD
596 stream->out_fd = -1;
597 stream->out_fd_offset = 0;
e5d1a9b3 598 stream->output_written = 0;
ffe60014 599 stream->net_seq_idx = relayd_id;
53632229 600 stream->session_id = session_id;
4891ece8 601 stream->monitor = monitor;
774d490c 602 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
f8f3885c 603 stream->index_file = NULL;
fb83fe64 604 stream->last_sequence_number = -1ULL;
53632229 605 pthread_mutex_init(&stream->lock, NULL);
c585821b 606 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
58b1f425 607
ffe60014
DG
608 /* If channel is the metadata, flag this stream as metadata. */
609 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
610 stream->metadata_flag = 1;
611 /* Metadata is flat out. */
612 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
94d49140
JD
613 /* Live rendez-vous point. */
614 pthread_cond_init(&stream->metadata_rdv, NULL);
615 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
58b1f425 616 } else {
ffe60014
DG
617 /* Format stream name to <channel_name>_<cpu_number> */
618 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
619 channel_name, cpu);
620 if (ret < 0) {
621 PERROR("snprintf stream name");
622 goto error;
623 }
58b1f425 624 }
c30aaa51 625
ffe60014 626 /* Key is always the wait_fd for streams. */
d88aee68 627 lttng_ht_node_init_u64(&stream->node, stream->key);
ffe60014 628
d8ef542d
MD
629 /* Init node per channel id key */
630 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
631
53632229 632 /* Init session id node with the stream session id */
d88aee68 633 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
53632229 634
07b86b52
JD
635 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
636 " relayd_id %" PRIu64 ", session_id %" PRIu64,
637 stream->name, stream->key, channel_key,
638 stream->net_seq_idx, stream->session_id);
d56db448
DG
639
640 rcu_read_unlock();
3bd1e081 641 return stream;
c80048c6
MD
642
643error:
d56db448 644 rcu_read_unlock();
d2956687 645 lttng_trace_chunk_put(stream->trace_chunk);
c80048c6 646 free(stream);
7a57cf92 647end:
ffe60014
DG
648 if (alloc_ret) {
649 *alloc_ret = ret;
650 }
c80048c6 651 return NULL;
3bd1e081
MD
652}
653
654/*
655 * Add a stream to the global list protected by a mutex.
656 */
66d583dc 657void consumer_add_data_stream(struct lttng_consumer_stream *stream)
3bd1e081 658{
5ab66908 659 struct lttng_ht *ht = data_ht;
3bd1e081 660
e316aad5 661 assert(stream);
43c34bc3 662 assert(ht);
c77fc10a 663
d88aee68 664 DBG3("Adding consumer stream %" PRIu64, stream->key);
e316aad5
DG
665
666 pthread_mutex_lock(&consumer_data.lock);
a9838785 667 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 668 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 669 pthread_mutex_lock(&stream->lock);
b0b335c8 670 rcu_read_lock();
e316aad5 671
43c34bc3 672 /* Steal stream identifier to avoid having streams with the same key */
ffe60014 673 steal_stream_key(stream->key, ht);
43c34bc3 674
d88aee68 675 lttng_ht_add_unique_u64(ht, &stream->node);
00e2e675 676
d8ef542d
MD
677 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
678 &stream->node_channel_id);
679
ca22feea
DG
680 /*
681 * Add stream to the stream_list_ht of the consumer data. No need to steal
682 * the key since the HT does not use it and we allow to add redundant keys
683 * into this table.
684 */
d88aee68 685 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 686
e316aad5 687 /*
ffe60014
DG
688 * When nb_init_stream_left reaches 0, we don't need to trigger any action
689 * in terms of destroying the associated channel, because the action that
e316aad5
DG
690 * causes the count to become 0 also causes a stream to be added. The
691 * channel deletion will thus be triggered by the following removal of this
692 * stream.
693 */
ffe60014 694 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
695 /* Increment refcount before decrementing nb_init_stream_left */
696 cmm_smp_wmb();
ffe60014 697 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
698 }
699
700 /* Update consumer data once the node is inserted. */
3bd1e081
MD
701 consumer_data.stream_count++;
702 consumer_data.need_update = 1;
703
e316aad5 704 rcu_read_unlock();
2e818a6a 705 pthread_mutex_unlock(&stream->lock);
ec6ea7d0 706 pthread_mutex_unlock(&stream->chan->timer_lock);
a9838785 707 pthread_mutex_unlock(&stream->chan->lock);
3bd1e081 708 pthread_mutex_unlock(&consumer_data.lock);
3bd1e081
MD
709}
710
5ab66908
MD
711void consumer_del_data_stream(struct lttng_consumer_stream *stream)
712{
713 consumer_del_stream(stream, data_ht);
714}
715
00e2e675 716/*
3f8e211f
DG
717 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
718 * be acquired before calling this.
00e2e675 719 */
d09e1200 720static int add_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
721{
722 int ret = 0;
d88aee68 723 struct lttng_ht_node_u64 *node;
00e2e675
DG
724 struct lttng_ht_iter iter;
725
ffe60014 726 assert(relayd);
00e2e675 727
00e2e675 728 lttng_ht_lookup(consumer_data.relayd_ht,
d88aee68
DG
729 &relayd->net_seq_idx, &iter);
730 node = lttng_ht_iter_get_node_u64(&iter);
00e2e675 731 if (node != NULL) {
00e2e675
DG
732 goto end;
733 }
d88aee68 734 lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
00e2e675 735
00e2e675
DG
736end:
737 return ret;
738}
739
740/*
741 * Allocate and return a consumer relayd socket.
742 */
027a694f 743static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
da009f2c 744 uint64_t net_seq_idx)
00e2e675
DG
745{
746 struct consumer_relayd_sock_pair *obj = NULL;
747
da009f2c
MD
748 /* net sequence index of -1 is a failure */
749 if (net_seq_idx == (uint64_t) -1ULL) {
00e2e675
DG
750 goto error;
751 }
752
753 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
754 if (obj == NULL) {
755 PERROR("zmalloc relayd sock");
756 goto error;
757 }
758
759 obj->net_seq_idx = net_seq_idx;
760 obj->refcount = 0;
173af62f 761 obj->destroy_flag = 0;
f96e4545
MD
762 obj->control_sock.sock.fd = -1;
763 obj->data_sock.sock.fd = -1;
d88aee68 764 lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
00e2e675
DG
765 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
766
767error:
768 return obj;
769}
770
771/*
772 * Find a relayd socket pair in the global consumer data.
773 *
774 * Return the object if found else NULL.
b0b335c8
MD
775 * RCU read-side lock must be held across this call and while using the
776 * returned object.
00e2e675 777 */
d88aee68 778struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
00e2e675
DG
779{
780 struct lttng_ht_iter iter;
d88aee68 781 struct lttng_ht_node_u64 *node;
00e2e675
DG
782 struct consumer_relayd_sock_pair *relayd = NULL;
783
784 /* Negative keys are lookup failures */
d88aee68 785 if (key == (uint64_t) -1ULL) {
00e2e675
DG
786 goto error;
787 }
788
d88aee68 789 lttng_ht_lookup(consumer_data.relayd_ht, &key,
00e2e675 790 &iter);
d88aee68 791 node = lttng_ht_iter_get_node_u64(&iter);
00e2e675
DG
792 if (node != NULL) {
793 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
794 }
795
00e2e675
DG
796error:
797 return relayd;
798}
799
10a50311
JD
800/*
801 * Find a relayd and send the stream
802 *
803 * Returns 0 on success, < 0 on error
804 */
805int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
806 char *path)
807{
808 int ret = 0;
809 struct consumer_relayd_sock_pair *relayd;
810
811 assert(stream);
812 assert(stream->net_seq_idx != -1ULL);
813 assert(path);
814
815 /* The stream is not metadata. Get relayd reference if exists. */
816 rcu_read_lock();
817 relayd = consumer_find_relayd(stream->net_seq_idx);
818 if (relayd != NULL) {
819 /* Add stream on the relayd */
820 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
821 ret = relayd_add_stream(&relayd->control_sock, stream->name,
822 path, &stream->relayd_stream_id,
d2956687
JG
823 stream->chan->tracefile_size,
824 stream->chan->tracefile_count,
825 stream->trace_chunk);
10a50311
JD
826 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
827 if (ret < 0) {
9276e5c8
JR
828 ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
829 lttng_consumer_cleanup_relayd(relayd);
10a50311
JD
830 goto end;
831 }
1c20f0e2 832
10a50311 833 uatomic_inc(&relayd->refcount);
d01178b6 834 stream->sent_to_relayd = 1;
10a50311
JD
835 } else {
836 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
837 stream->key, stream->net_seq_idx);
838 ret = -1;
839 goto end;
840 }
841
842 DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
843 stream->name, stream->key, stream->net_seq_idx);
844
845end:
846 rcu_read_unlock();
847 return ret;
848}
849
a4baae1b
JD
850/*
851 * Find a relayd and send the streams sent message
852 *
853 * Returns 0 on success, < 0 on error
854 */
855int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
856{
857 int ret = 0;
858 struct consumer_relayd_sock_pair *relayd;
859
860 assert(net_seq_idx != -1ULL);
861
862 /* The stream is not metadata. Get relayd reference if exists. */
863 rcu_read_lock();
864 relayd = consumer_find_relayd(net_seq_idx);
865 if (relayd != NULL) {
866 /* Add stream on the relayd */
867 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
868 ret = relayd_streams_sent(&relayd->control_sock);
869 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
870 if (ret < 0) {
9276e5c8
JR
871 ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
872 lttng_consumer_cleanup_relayd(relayd);
a4baae1b
JD
873 goto end;
874 }
875 } else {
876 ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
877 net_seq_idx);
878 ret = -1;
879 goto end;
880 }
881
882 ret = 0;
883 DBG("All streams sent relayd id %" PRIu64, net_seq_idx);
884
885end:
886 rcu_read_unlock();
887 return ret;
888}
889
10a50311
JD
890/*
891 * Find a relayd and close the stream
892 */
893void close_relayd_stream(struct lttng_consumer_stream *stream)
894{
895 struct consumer_relayd_sock_pair *relayd;
896
897 /* The stream is not metadata. Get relayd reference if exists. */
898 rcu_read_lock();
899 relayd = consumer_find_relayd(stream->net_seq_idx);
900 if (relayd) {
901 consumer_stream_relayd_close(stream, relayd);
902 }
903 rcu_read_unlock();
904}
905
00e2e675
DG
906/*
907 * Handle stream for relayd transmission if the stream applies for network
908 * streaming where the net sequence index is set.
909 *
910 * Return destination file descriptor or negative value on error.
911 */
6197aea7 912static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
1d4dfdef
DG
913 size_t data_size, unsigned long padding,
914 struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
915{
916 int outfd = -1, ret;
00e2e675
DG
917 struct lttcomm_relayd_data_hdr data_hdr;
918
919 /* Safety net */
920 assert(stream);
6197aea7 921 assert(relayd);
00e2e675
DG
922
923 /* Reset data header */
924 memset(&data_hdr, 0, sizeof(data_hdr));
925
00e2e675
DG
926 if (stream->metadata_flag) {
927 /* Caller MUST acquire the relayd control socket lock */
928 ret = relayd_send_metadata(&relayd->control_sock, data_size);
929 if (ret < 0) {
930 goto error;
931 }
932
933 /* Metadata are always sent on the control socket. */
6151a90f 934 outfd = relayd->control_sock.sock.fd;
00e2e675
DG
935 } else {
936 /* Set header with stream information */
937 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
938 data_hdr.data_size = htobe32(data_size);
1d4dfdef 939 data_hdr.padding_size = htobe32(padding);
c35f9726 940
39df6d9f
DG
941 /*
942 * Note that net_seq_num below is assigned with the *current* value of
943 * next_net_seq_num and only after that the next_net_seq_num will be
944 * increment. This is why when issuing a command on the relayd using
945 * this next value, 1 should always be substracted in order to compare
946 * the last seen sequence number on the relayd side to the last sent.
947 */
3604f373 948 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
00e2e675
DG
949 /* Other fields are zeroed previously */
950
951 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
952 sizeof(data_hdr));
953 if (ret < 0) {
954 goto error;
955 }
956
3604f373
DG
957 ++stream->next_net_seq_num;
958
00e2e675 959 /* Set to go on data socket */
6151a90f 960 outfd = relayd->data_sock.sock.fd;
00e2e675
DG
961 }
962
963error:
964 return outfd;
965}
966
d2956687
JG
967/*
968 * Trigger a dump of the metadata content. Following/during the succesful
969 * completion of this call, the metadata poll thread will start receiving
970 * metadata packets to consume.
971 *
972 * The caller must hold the channel and stream locks.
973 */
974static
975int consumer_metadata_stream_dump(struct lttng_consumer_stream *stream)
976{
977 int ret;
978
979 ASSERT_LOCKED(stream->chan->lock);
980 ASSERT_LOCKED(stream->lock);
981 assert(stream->metadata_flag);
982 assert(stream->chan->trace_chunk);
983
984 switch (consumer_data.type) {
985 case LTTNG_CONSUMER_KERNEL:
986 /*
987 * Reset the position of what has been read from the
988 * metadata cache to 0 so we can dump it again.
989 */
990 ret = kernctl_metadata_cache_dump(stream->wait_fd);
991 break;
992 case LTTNG_CONSUMER32_UST:
993 case LTTNG_CONSUMER64_UST:
994 /*
995 * Reset the position pushed from the metadata cache so it
996 * will write from the beginning on the next push.
997 */
998 stream->ust_metadata_pushed = 0;
999 ret = consumer_metadata_wakeup_pipe(stream->chan);
1000 break;
1001 default:
1002 ERR("Unknown consumer_data type");
1003 abort();
1004 }
1005 if (ret < 0) {
1006 ERR("Failed to dump the metadata cache");
1007 }
1008 return ret;
1009}
1010
1011static
1012int lttng_consumer_channel_set_trace_chunk(
1013 struct lttng_consumer_channel *channel,
1014 struct lttng_trace_chunk *new_trace_chunk)
1015{
1016 int ret = 0;
1017 const bool is_local_trace = channel->relayd_id == -1ULL;
1018 bool update_stream_trace_chunk;
1019 struct cds_lfht_iter iter;
1020 struct lttng_consumer_stream *stream;
1021 unsigned long channel_hash;
1022
1023 pthread_mutex_lock(&channel->lock);
1024 /*
1025 * A stream can transition to a state where it and its channel
1026 * no longer belong to a trace chunk. For instance, this happens when
1027 * a session is rotated while it is inactive. After the rotation
1028 * of an inactive session completes, the channel and its streams no
1029 * longer belong to a trace chunk.
1030 *
1031 * However, if a session is stopped, rotated, and started again,
1032 * the session daemon will create a new chunk and send it to its peers.
1033 * In that case, the streams' transition to a new chunk can be performed
1034 * immediately.
1035 *
1036 * This trace chunk transition could also be performed lazily when
1037 * a buffer is consumed. However, creating the files here allows the
1038 * consumer daemon to report any creation error to the session daemon
1039 * and cause the start of the tracing session to fail.
1040 */
1041 update_stream_trace_chunk = !channel->trace_chunk && new_trace_chunk;
1042
1043 /*
1044 * The acquisition of the reference cannot fail (barring
1045 * a severe internal error) since a reference to the published
1046 * chunk is already held by the caller.
1047 */
1048 if (new_trace_chunk) {
1049 const bool acquired_reference = lttng_trace_chunk_get(
1050 new_trace_chunk);
1051
1052 assert(acquired_reference);
1053 }
1054
1055 lttng_trace_chunk_put(channel->trace_chunk);
1056 channel->trace_chunk = new_trace_chunk;
1057 if (!is_local_trace || !new_trace_chunk) {
1058 /* Not an error. */
1059 goto end;
1060 }
1061
1062 if (!update_stream_trace_chunk) {
1063 goto end;
1064 }
1065
1066 channel_hash = consumer_data.stream_per_chan_id_ht->hash_fct(
1067 &channel->key, lttng_ht_seed);
1068 rcu_read_lock();
1069 cds_lfht_for_each_entry_duplicate(consumer_data.stream_per_chan_id_ht->ht,
1070 channel_hash,
1071 consumer_data.stream_per_chan_id_ht->match_fct,
1072 &channel->key, &iter, stream, node_channel_id.node) {
1073 bool acquired_reference, should_regenerate_metadata = false;
1074
1075 acquired_reference = lttng_trace_chunk_get(channel->trace_chunk);
1076 assert(acquired_reference);
1077
1078 pthread_mutex_lock(&stream->lock);
1079
1080 /*
1081 * On a transition from "no-chunk" to a new chunk, a metadata
1082 * stream's content must be entirely dumped. This must occcur
1083 * _after_ the creation of the metadata stream's output files
1084 * as the consumption thread (not necessarily the one executing
1085 * this) may start to consume during the call to
1086 * consumer_metadata_stream_dump().
1087 */
1088 should_regenerate_metadata =
1089 stream->metadata_flag &&
1090 !stream->trace_chunk && channel->trace_chunk;
1091 stream->trace_chunk = channel->trace_chunk;
1092 ret = consumer_stream_create_output_files(stream, true);
1093 if (ret) {
1094 pthread_mutex_unlock(&stream->lock);
1095 goto end_rcu_unlock;
1096 }
1097 if (should_regenerate_metadata) {
1098 ret = consumer_metadata_stream_dump(stream);
1099 }
1100 pthread_mutex_unlock(&stream->lock);
1101 if (ret) {
1102 goto end_rcu_unlock;
1103 }
1104 }
1105end_rcu_unlock:
1106 rcu_read_unlock();
1107end:
1108 pthread_mutex_unlock(&channel->lock);
1109 return ret;
1110}
1111
3bd1e081 1112/*
ffe60014
DG
1113 * Allocate and return a new lttng_consumer_channel object using the given key
1114 * to initialize the hash table node.
1115 *
1116 * On error, return NULL.
3bd1e081 1117 */
886224ff 1118struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
ffe60014 1119 uint64_t session_id,
d2956687 1120 const uint64_t *chunk_id,
ffe60014
DG
1121 const char *pathname,
1122 const char *name,
57a269f2 1123 uint64_t relayd_id,
1624d5b7
JD
1124 enum lttng_event_output output,
1125 uint64_t tracefile_size,
2bba9e53 1126 uint64_t tracefile_count,
1950109e 1127 uint64_t session_id_per_pid,
ecc48a90 1128 unsigned int monitor,
d7ba1388 1129 unsigned int live_timer_interval,
3d071855 1130 const char *root_shm_path,
d7ba1388 1131 const char *shm_path)
3bd1e081 1132{
d2956687
JG
1133 struct lttng_consumer_channel *channel = NULL;
1134 struct lttng_trace_chunk *trace_chunk = NULL;
1135
1136 if (chunk_id) {
1137 trace_chunk = lttng_trace_chunk_registry_find_chunk(
1138 consumer_data.chunk_registry, session_id,
1139 *chunk_id);
1140 if (!trace_chunk) {
1141 ERR("Failed to find trace chunk reference during creation of channel");
1142 goto end;
1143 }
1144 }
3bd1e081 1145
276b26d1 1146 channel = zmalloc(sizeof(*channel));
3bd1e081 1147 if (channel == NULL) {
7a57cf92 1148 PERROR("malloc struct lttng_consumer_channel");
3bd1e081
MD
1149 goto end;
1150 }
ffe60014
DG
1151
1152 channel->key = key;
3bd1e081 1153 channel->refcount = 0;
ffe60014 1154 channel->session_id = session_id;
1950109e 1155 channel->session_id_per_pid = session_id_per_pid;
ffe60014 1156 channel->relayd_id = relayd_id;
1624d5b7
JD
1157 channel->tracefile_size = tracefile_size;
1158 channel->tracefile_count = tracefile_count;
2bba9e53 1159 channel->monitor = monitor;
ecc48a90 1160 channel->live_timer_interval = live_timer_interval;
a9838785 1161 pthread_mutex_init(&channel->lock, NULL);
ec6ea7d0 1162 pthread_mutex_init(&channel->timer_lock, NULL);
ffe60014 1163
0c759fc9
DG
1164 switch (output) {
1165 case LTTNG_EVENT_SPLICE:
1166 channel->output = CONSUMER_CHANNEL_SPLICE;
1167 break;
1168 case LTTNG_EVENT_MMAP:
1169 channel->output = CONSUMER_CHANNEL_MMAP;
1170 break;
1171 default:
1172 assert(0);
1173 free(channel);
1174 channel = NULL;
1175 goto end;
1176 }
1177
07b86b52
JD
1178 /*
1179 * In monitor mode, the streams associated with the channel will be put in
1180 * a special list ONLY owned by this channel. So, the refcount is set to 1
1181 * here meaning that the channel itself has streams that are referenced.
1182 *
1183 * On a channel deletion, once the channel is no longer visible, the
1184 * refcount is decremented and checked for a zero value to delete it. With
1185 * streams in no monitor mode, it will now be safe to destroy the channel.
1186 */
1187 if (!channel->monitor) {
1188 channel->refcount = 1;
1189 }
1190
ffe60014
DG
1191 strncpy(channel->pathname, pathname, sizeof(channel->pathname));
1192 channel->pathname[sizeof(channel->pathname) - 1] = '\0';
1193
1194 strncpy(channel->name, name, sizeof(channel->name));
1195 channel->name[sizeof(channel->name) - 1] = '\0';
1196
3d071855
MD
1197 if (root_shm_path) {
1198 strncpy(channel->root_shm_path, root_shm_path, sizeof(channel->root_shm_path));
1199 channel->root_shm_path[sizeof(channel->root_shm_path) - 1] = '\0';
1200 }
d7ba1388
MD
1201 if (shm_path) {
1202 strncpy(channel->shm_path, shm_path, sizeof(channel->shm_path));
1203 channel->shm_path[sizeof(channel->shm_path) - 1] = '\0';
1204 }
1205
d88aee68 1206 lttng_ht_node_init_u64(&channel->node, channel->key);
5c3892a6
JG
1207 lttng_ht_node_init_u64(&channel->channels_by_session_id_ht_node,
1208 channel->session_id);
d8ef542d
MD
1209
1210 channel->wait_fd = -1;
ffe60014
DG
1211 CDS_INIT_LIST_HEAD(&channel->streams.head);
1212
d2956687
JG
1213 if (trace_chunk) {
1214 int ret = lttng_consumer_channel_set_trace_chunk(channel,
1215 trace_chunk);
1216 if (ret) {
1217 goto error;
1218 }
1219 }
1220
62a7b8ed 1221 DBG("Allocated channel (key %" PRIu64 ")", channel->key);
3bd1e081 1222
3bd1e081 1223end:
d2956687 1224 lttng_trace_chunk_put(trace_chunk);
3bd1e081 1225 return channel;
d2956687
JG
1226error:
1227 consumer_del_channel(channel);
1228 channel = NULL;
1229 goto end;
3bd1e081
MD
1230}
1231
1232/*
1233 * Add a channel to the global list protected by a mutex.
821fffb2 1234 *
b5a6470f 1235 * Always return 0 indicating success.
3bd1e081 1236 */
d8ef542d
MD
1237int consumer_add_channel(struct lttng_consumer_channel *channel,
1238 struct lttng_consumer_local_data *ctx)
3bd1e081 1239{
3bd1e081 1240 pthread_mutex_lock(&consumer_data.lock);
a9838785 1241 pthread_mutex_lock(&channel->lock);
ec6ea7d0 1242 pthread_mutex_lock(&channel->timer_lock);
c77fc10a 1243
b5a6470f
DG
1244 /*
1245 * This gives us a guarantee that the channel we are about to add to the
1246 * channel hash table will be unique. See this function comment on the why
1247 * we need to steel the channel key at this stage.
1248 */
1249 steal_channel_key(channel->key);
c77fc10a 1250
b5a6470f 1251 rcu_read_lock();
d88aee68 1252 lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
5c3892a6
JG
1253 lttng_ht_add_u64(consumer_data.channels_by_session_id_ht,
1254 &channel->channels_by_session_id_ht_node);
6065ceec 1255 rcu_read_unlock();
d2956687 1256 channel->is_published = true;
b5a6470f 1257
ec6ea7d0 1258 pthread_mutex_unlock(&channel->timer_lock);
a9838785 1259 pthread_mutex_unlock(&channel->lock);
3bd1e081 1260 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 1261
b5a6470f 1262 if (channel->wait_fd != -1 && channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
a0cbdd2e 1263 notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
d8ef542d 1264 }
b5a6470f
DG
1265
1266 return 0;
3bd1e081
MD
1267}
1268
1269/*
1270 * Allocate the pollfd structure and the local view of the out fds to avoid
1271 * doing a lookup in the linked list and concurrency issues when writing is
1272 * needed. Called with consumer_data.lock held.
1273 *
1274 * Returns the number of fds in the structures.
1275 */
ffe60014
DG
1276static int update_poll_array(struct lttng_consumer_local_data *ctx,
1277 struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
9a2fcf78 1278 struct lttng_ht *ht, int *nb_inactive_fd)
3bd1e081 1279{
3bd1e081 1280 int i = 0;
e4421fec
DG
1281 struct lttng_ht_iter iter;
1282 struct lttng_consumer_stream *stream;
3bd1e081 1283
ffe60014
DG
1284 assert(ctx);
1285 assert(ht);
1286 assert(pollfd);
1287 assert(local_stream);
1288
3bd1e081 1289 DBG("Updating poll fd array");
9a2fcf78 1290 *nb_inactive_fd = 0;
481d6c57 1291 rcu_read_lock();
43c34bc3 1292 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
8994307f
DG
1293 /*
1294 * Only active streams with an active end point can be added to the
1295 * poll set and local stream storage of the thread.
1296 *
1297 * There is a potential race here for endpoint_status to be updated
1298 * just after the check. However, this is OK since the stream(s) will
1299 * be deleted once the thread is notified that the end point state has
1300 * changed where this function will be called back again.
9a2fcf78
JD
1301 *
1302 * We track the number of inactive FDs because they still need to be
1303 * closed by the polling thread after a wakeup on the data_pipe or
1304 * metadata_pipe.
8994307f 1305 */
d2956687 1306 if (stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
9a2fcf78 1307 (*nb_inactive_fd)++;
3bd1e081
MD
1308 continue;
1309 }
7972aab2
DG
1310 /*
1311 * This clobbers way too much the debug output. Uncomment that if you
1312 * need it for debugging purposes.
7972aab2 1313 */
e4421fec 1314 (*pollfd)[i].fd = stream->wait_fd;
3bd1e081 1315 (*pollfd)[i].events = POLLIN | POLLPRI;
e4421fec 1316 local_stream[i] = stream;
3bd1e081
MD
1317 i++;
1318 }
481d6c57 1319 rcu_read_unlock();
3bd1e081
MD
1320
1321 /*
50f8ae69 1322 * Insert the consumer_data_pipe at the end of the array and don't
3bd1e081
MD
1323 * increment i so nb_fd is the number of real FD.
1324 */
acdb9057 1325 (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
509bb1cf 1326 (*pollfd)[i].events = POLLIN | POLLPRI;
02b3d176
DG
1327
1328 (*pollfd)[i + 1].fd = lttng_pipe_get_readfd(ctx->consumer_wakeup_pipe);
1329 (*pollfd)[i + 1].events = POLLIN | POLLPRI;
3bd1e081
MD
1330 return i;
1331}
1332
1333/*
84382d49
MD
1334 * Poll on the should_quit pipe and the command socket return -1 on
1335 * error, 1 if should exit, 0 if data is available on the command socket
3bd1e081
MD
1336 */
1337int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
1338{
1339 int num_rdy;
1340
88f2b785 1341restart:
3bd1e081
MD
1342 num_rdy = poll(consumer_sockpoll, 2, -1);
1343 if (num_rdy == -1) {
88f2b785
MD
1344 /*
1345 * Restart interrupted system call.
1346 */
1347 if (errno == EINTR) {
1348 goto restart;
1349 }
7a57cf92 1350 PERROR("Poll error");
84382d49 1351 return -1;
3bd1e081 1352 }
509bb1cf 1353 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
3bd1e081 1354 DBG("consumer_should_quit wake up");
84382d49 1355 return 1;
3bd1e081
MD
1356 }
1357 return 0;
3bd1e081
MD
1358}
1359
1360/*
1361 * Set the error socket.
1362 */
ffe60014
DG
1363void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
1364 int sock)
3bd1e081
MD
1365{
1366 ctx->consumer_error_socket = sock;
1367}
1368
1369/*
1370 * Set the command socket path.
1371 */
3bd1e081
MD
1372void lttng_consumer_set_command_sock_path(
1373 struct lttng_consumer_local_data *ctx, char *sock)
1374{
1375 ctx->consumer_command_sock_path = sock;
1376}
1377
1378/*
1379 * Send return code to the session daemon.
1380 * If the socket is not defined, we return 0, it is not a fatal error
1381 */
ffe60014 1382int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
3bd1e081
MD
1383{
1384 if (ctx->consumer_error_socket > 0) {
1385 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
1386 sizeof(enum lttcomm_sessiond_command));
1387 }
1388
1389 return 0;
1390}
1391
1392/*
228b5bf7
DG
1393 * Close all the tracefiles and stream fds and MUST be called when all
1394 * instances are destroyed i.e. when all threads were joined and are ended.
3bd1e081
MD
1395 */
1396void lttng_consumer_cleanup(void)
1397{
e4421fec 1398 struct lttng_ht_iter iter;
ffe60014 1399 struct lttng_consumer_channel *channel;
6065ceec
DG
1400
1401 rcu_read_lock();
3bd1e081 1402
ffe60014
DG
1403 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
1404 node.node) {
702b1ea4 1405 consumer_del_channel(channel);
3bd1e081 1406 }
6065ceec
DG
1407
1408 rcu_read_unlock();
d6ce1df2 1409
d6ce1df2 1410 lttng_ht_destroy(consumer_data.channel_ht);
5c3892a6 1411 lttng_ht_destroy(consumer_data.channels_by_session_id_ht);
228b5bf7
DG
1412
1413 cleanup_relayd_ht();
1414
d8ef542d
MD
1415 lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
1416
228b5bf7
DG
1417 /*
1418 * This HT contains streams that are freed by either the metadata thread or
1419 * the data thread so we do *nothing* on the hash table and simply destroy
1420 * it.
1421 */
1422 lttng_ht_destroy(consumer_data.stream_list_ht);
28cc88f3
JG
1423
1424 lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
3bd1e081
MD
1425}
1426
1427/*
1428 * Called from signal handler.
1429 */
1430void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
1431{
6cd525e8
MD
1432 ssize_t ret;
1433
10211f5c 1434 CMM_STORE_SHARED(consumer_quit, 1);
6cd525e8
MD
1435 ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
1436 if (ret < 1) {
7a57cf92 1437 PERROR("write consumer quit");
3bd1e081 1438 }
ab1027f4
DG
1439
1440 DBG("Consumer flag that it should quit");
3bd1e081
MD
1441}
1442
5199ffc4
JG
1443
1444/*
1445 * Flush pending writes to trace output disk file.
1446 */
1447static
00e2e675
DG
1448void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
1449 off_t orig_offset)
3bd1e081 1450{
c7a78aab 1451 int ret;
3bd1e081
MD
1452 int outfd = stream->out_fd;
1453
1454 /*
1455 * This does a blocking write-and-wait on any page that belongs to the
1456 * subbuffer prior to the one we just wrote.
1457 * Don't care about error values, as these are just hints and ways to
1458 * limit the amount of page cache used.
1459 */
ffe60014 1460 if (orig_offset < stream->max_sb_size) {
3bd1e081
MD
1461 return;
1462 }
ffe60014
DG
1463 lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
1464 stream->max_sb_size,
3bd1e081
MD
1465 SYNC_FILE_RANGE_WAIT_BEFORE
1466 | SYNC_FILE_RANGE_WRITE
1467 | SYNC_FILE_RANGE_WAIT_AFTER);
1468 /*
1469 * Give hints to the kernel about how we access the file:
1470 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1471 * we write it.
1472 *
1473 * We need to call fadvise again after the file grows because the
1474 * kernel does not seem to apply fadvise to non-existing parts of the
1475 * file.
1476 *
1477 * Call fadvise _after_ having waited for the page writeback to
1478 * complete because the dirty page writeback semantic is not well
1479 * defined. So it can be expected to lead to lower throughput in
1480 * streaming.
1481 */
c7a78aab 1482 ret = posix_fadvise(outfd, orig_offset - stream->max_sb_size,
ffe60014 1483 stream->max_sb_size, POSIX_FADV_DONTNEED);
a0d0e127 1484 if (ret && ret != -ENOSYS) {
a74a5f4a
JG
1485 errno = ret;
1486 PERROR("posix_fadvise on fd %i", outfd);
c7a78aab 1487 }
3bd1e081
MD
1488}
1489
1490/*
1491 * Initialise the necessary environnement :
1492 * - create a new context
1493 * - create the poll_pipe
1494 * - create the should_quit pipe (for signal handler)
1495 * - create the thread pipe (for splice)
1496 *
1497 * Takes a function pointer as argument, this function is called when data is
1498 * available on a buffer. This function is responsible to do the
1499 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1500 * buffer configuration and then kernctl_put_next_subbuf at the end.
1501 *
1502 * Returns a pointer to the new context or NULL on error.
1503 */
1504struct lttng_consumer_local_data *lttng_consumer_create(
1505 enum lttng_consumer_type type,
4078b776 1506 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 1507 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
1508 int (*recv_channel)(struct lttng_consumer_channel *channel),
1509 int (*recv_stream)(struct lttng_consumer_stream *stream),
30319bcb 1510 int (*update_stream)(uint64_t stream_key, uint32_t state))
3bd1e081 1511{
d8ef542d 1512 int ret;
3bd1e081
MD
1513 struct lttng_consumer_local_data *ctx;
1514
1515 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
1516 consumer_data.type == type);
1517 consumer_data.type = type;
1518
effcf122 1519 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
3bd1e081 1520 if (ctx == NULL) {
7a57cf92 1521 PERROR("allocating context");
3bd1e081
MD
1522 goto error;
1523 }
1524
1525 ctx->consumer_error_socket = -1;
331744e3 1526 ctx->consumer_metadata_socket = -1;
75d83e50 1527 pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
3bd1e081
MD
1528 /* assign the callbacks */
1529 ctx->on_buffer_ready = buffer_ready;
1530 ctx->on_recv_channel = recv_channel;
1531 ctx->on_recv_stream = recv_stream;
1532 ctx->on_update_stream = update_stream;
1533
acdb9057
DG
1534 ctx->consumer_data_pipe = lttng_pipe_open(0);
1535 if (!ctx->consumer_data_pipe) {
3bd1e081
MD
1536 goto error_poll_pipe;
1537 }
1538
02b3d176
DG
1539 ctx->consumer_wakeup_pipe = lttng_pipe_open(0);
1540 if (!ctx->consumer_wakeup_pipe) {
1541 goto error_wakeup_pipe;
1542 }
1543
3bd1e081
MD
1544 ret = pipe(ctx->consumer_should_quit);
1545 if (ret < 0) {
7a57cf92 1546 PERROR("Error creating recv pipe");
3bd1e081
MD
1547 goto error_quit_pipe;
1548 }
1549
d8ef542d
MD
1550 ret = pipe(ctx->consumer_channel_pipe);
1551 if (ret < 0) {
1552 PERROR("Error creating channel pipe");
1553 goto error_channel_pipe;
1554 }
1555
13886d2d
DG
1556 ctx->consumer_metadata_pipe = lttng_pipe_open(0);
1557 if (!ctx->consumer_metadata_pipe) {
fb3a43a9
DG
1558 goto error_metadata_pipe;
1559 }
3bd1e081 1560
e9404c27
JG
1561 ctx->channel_monitor_pipe = -1;
1562
fb3a43a9 1563 return ctx;
3bd1e081 1564
fb3a43a9 1565error_metadata_pipe:
d8ef542d
MD
1566 utils_close_pipe(ctx->consumer_channel_pipe);
1567error_channel_pipe:
d8ef542d 1568 utils_close_pipe(ctx->consumer_should_quit);
3bd1e081 1569error_quit_pipe:
02b3d176
DG
1570 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
1571error_wakeup_pipe:
acdb9057 1572 lttng_pipe_destroy(ctx->consumer_data_pipe);
3bd1e081
MD
1573error_poll_pipe:
1574 free(ctx);
1575error:
1576 return NULL;
1577}
1578
282dadbc
MD
1579/*
1580 * Iterate over all streams of the hashtable and free them properly.
1581 */
1582static void destroy_data_stream_ht(struct lttng_ht *ht)
1583{
1584 struct lttng_ht_iter iter;
1585 struct lttng_consumer_stream *stream;
1586
1587 if (ht == NULL) {
1588 return;
1589 }
1590
1591 rcu_read_lock();
1592 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1593 /*
1594 * Ignore return value since we are currently cleaning up so any error
1595 * can't be handled.
1596 */
1597 (void) consumer_del_stream(stream, ht);
1598 }
1599 rcu_read_unlock();
1600
1601 lttng_ht_destroy(ht);
1602}
1603
1604/*
1605 * Iterate over all streams of the metadata hashtable and free them
1606 * properly.
1607 */
1608static void destroy_metadata_stream_ht(struct lttng_ht *ht)
1609{
1610 struct lttng_ht_iter iter;
1611 struct lttng_consumer_stream *stream;
1612
1613 if (ht == NULL) {
1614 return;
1615 }
1616
1617 rcu_read_lock();
1618 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1619 /*
1620 * Ignore return value since we are currently cleaning up so any error
1621 * can't be handled.
1622 */
1623 (void) consumer_del_metadata_stream(stream, ht);
1624 }
1625 rcu_read_unlock();
1626
1627 lttng_ht_destroy(ht);
1628}
1629
3bd1e081
MD
1630/*
1631 * Close all fds associated with the instance and free the context.
1632 */
1633void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1634{
4c462e79
MD
1635 int ret;
1636
ab1027f4
DG
1637 DBG("Consumer destroying it. Closing everything.");
1638
4f2e75b9
DG
1639 if (!ctx) {
1640 return;
1641 }
1642
282dadbc
MD
1643 destroy_data_stream_ht(data_ht);
1644 destroy_metadata_stream_ht(metadata_ht);
1645
4c462e79
MD
1646 ret = close(ctx->consumer_error_socket);
1647 if (ret) {
1648 PERROR("close");
1649 }
331744e3
JD
1650 ret = close(ctx->consumer_metadata_socket);
1651 if (ret) {
1652 PERROR("close");
1653 }
d8ef542d 1654 utils_close_pipe(ctx->consumer_channel_pipe);
acdb9057 1655 lttng_pipe_destroy(ctx->consumer_data_pipe);
13886d2d 1656 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
02b3d176 1657 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
d8ef542d 1658 utils_close_pipe(ctx->consumer_should_quit);
fb3a43a9 1659
3bd1e081
MD
1660 unlink(ctx->consumer_command_sock_path);
1661 free(ctx);
1662}
1663
6197aea7
DG
1664/*
1665 * Write the metadata stream id on the specified file descriptor.
1666 */
1667static int write_relayd_metadata_id(int fd,
1668 struct lttng_consumer_stream *stream,
239f61af 1669 unsigned long padding)
6197aea7 1670{
6cd525e8 1671 ssize_t ret;
1d4dfdef 1672 struct lttcomm_relayd_metadata_payload hdr;
6197aea7 1673
1d4dfdef
DG
1674 hdr.stream_id = htobe64(stream->relayd_stream_id);
1675 hdr.padding_size = htobe32(padding);
6cd525e8
MD
1676 ret = lttng_write(fd, (void *) &hdr, sizeof(hdr));
1677 if (ret < sizeof(hdr)) {
d7b75ec8 1678 /*
6f04ed72 1679 * This error means that the fd's end is closed so ignore the PERROR
d7b75ec8
DG
1680 * not to clubber the error output since this can happen in a normal
1681 * code path.
1682 */
1683 if (errno != EPIPE) {
1684 PERROR("write metadata stream id");
1685 }
1686 DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
534d2592
DG
1687 /*
1688 * Set ret to a negative value because if ret != sizeof(hdr), we don't
1689 * handle writting the missing part so report that as an error and
1690 * don't lie to the caller.
1691 */
1692 ret = -1;
6197aea7
DG
1693 goto end;
1694 }
1d4dfdef
DG
1695 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1696 stream->relayd_stream_id, padding);
6197aea7
DG
1697
1698end:
6cd525e8 1699 return (int) ret;
6197aea7
DG
1700}
1701
3bd1e081 1702/*
09e26845
DG
1703 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1704 * core function for writing trace buffers to either the local filesystem or
1705 * the network.
1706 *
d2956687 1707 * It must be called with the stream and the channel lock held.
79d4ffb7 1708 *
09e26845 1709 * Careful review MUST be put if any changes occur!
3bd1e081
MD
1710 *
1711 * Returns the number of bytes written
1712 */
4078b776 1713ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081 1714 struct lttng_consumer_local_data *ctx,
1d4dfdef 1715 struct lttng_consumer_stream *stream, unsigned long len,
309167d2 1716 unsigned long padding,
50adc264 1717 struct ctf_packet_index *index)
3bd1e081 1718{
f02e1e8a 1719 unsigned long mmap_offset;
ffe60014 1720 void *mmap_base;
994ab360 1721 ssize_t ret = 0;
f02e1e8a
DG
1722 off_t orig_offset = stream->out_fd_offset;
1723 /* Default is on the disk */
1724 int outfd = stream->out_fd;
f02e1e8a 1725 struct consumer_relayd_sock_pair *relayd = NULL;
8994307f 1726 unsigned int relayd_hang_up = 0;
f02e1e8a
DG
1727
1728 /* RCU lock for the relayd pointer */
1729 rcu_read_lock();
1730
7fd975c5
JG
1731 assert(stream->net_seq_idx != (uint64_t) -1ULL ||
1732 stream->chan->trace_chunk);
d2956687 1733
f02e1e8a 1734 /* Flag that the current stream if set for network streaming. */
da009f2c 1735 if (stream->net_seq_idx != (uint64_t) -1ULL) {
f02e1e8a
DG
1736 relayd = consumer_find_relayd(stream->net_seq_idx);
1737 if (relayd == NULL) {
56591bac 1738 ret = -EPIPE;
f02e1e8a
DG
1739 goto end;
1740 }
1741 }
1742
1743 /* get the offset inside the fd to mmap */
3bd1e081
MD
1744 switch (consumer_data.type) {
1745 case LTTNG_CONSUMER_KERNEL:
ffe60014 1746 mmap_base = stream->mmap_base;
f02e1e8a 1747 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
994ab360 1748 if (ret < 0) {
56591bac 1749 PERROR("tracer ctl get_mmap_read_offset");
56591bac
MD
1750 goto end;
1751 }
f02e1e8a 1752 break;
7753dea8
MD
1753 case LTTNG_CONSUMER32_UST:
1754 case LTTNG_CONSUMER64_UST:
ffe60014
DG
1755 mmap_base = lttng_ustctl_get_mmap_base(stream);
1756 if (!mmap_base) {
1757 ERR("read mmap get mmap base for stream %s", stream->name);
994ab360 1758 ret = -EPERM;
ffe60014
DG
1759 goto end;
1760 }
1761 ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
56591bac
MD
1762 if (ret != 0) {
1763 PERROR("tracer ctl get_mmap_read_offset");
994ab360 1764 ret = -EINVAL;
56591bac
MD
1765 goto end;
1766 }
f02e1e8a 1767 break;
3bd1e081
MD
1768 default:
1769 ERR("Unknown consumer_data type");
1770 assert(0);
1771 }
b9182dd9 1772
f02e1e8a
DG
1773 /* Handle stream on the relayd if the output is on the network */
1774 if (relayd) {
1775 unsigned long netlen = len;
1776
1777 /*
1778 * Lock the control socket for the complete duration of the function
1779 * since from this point on we will use the socket.
1780 */
1781 if (stream->metadata_flag) {
1782 /* Metadata requires the control socket. */
1783 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
93ec662e
JD
1784 if (stream->reset_metadata_flag) {
1785 ret = relayd_reset_metadata(&relayd->control_sock,
1786 stream->relayd_stream_id,
1787 stream->metadata_version);
1788 if (ret < 0) {
1789 relayd_hang_up = 1;
1790 goto write_error;
1791 }
1792 stream->reset_metadata_flag = 0;
1793 }
1d4dfdef 1794 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
f02e1e8a
DG
1795 }
1796
1d4dfdef 1797 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
994ab360
DG
1798 if (ret < 0) {
1799 relayd_hang_up = 1;
1800 goto write_error;
1801 }
1802 /* Use the returned socket. */
1803 outfd = ret;
f02e1e8a 1804
994ab360
DG
1805 /* Write metadata stream id before payload */
1806 if (stream->metadata_flag) {
239f61af 1807 ret = write_relayd_metadata_id(outfd, stream, padding);
994ab360 1808 if (ret < 0) {
8994307f
DG
1809 relayd_hang_up = 1;
1810 goto write_error;
1811 }
f02e1e8a 1812 }
1d4dfdef
DG
1813 } else {
1814 /* No streaming, we have to set the len with the full padding */
1815 len += padding;
1624d5b7 1816
93ec662e
JD
1817 if (stream->metadata_flag && stream->reset_metadata_flag) {
1818 ret = utils_truncate_stream_file(stream->out_fd, 0);
1819 if (ret < 0) {
1820 ERR("Reset metadata file");
1821 goto end;
1822 }
1823 stream->reset_metadata_flag = 0;
1824 }
1825
1624d5b7
JD
1826 /*
1827 * Check if we need to change the tracefile before writing the packet.
1828 */
1829 if (stream->chan->tracefile_size > 0 &&
1830 (stream->tracefile_size_current + len) >
1831 stream->chan->tracefile_size) {
d2956687
JG
1832 ret = consumer_stream_rotate_output_files(stream);
1833 if (ret) {
1624d5b7
JD
1834 goto end;
1835 }
309167d2 1836 outfd = stream->out_fd;
a1ae300f 1837 orig_offset = 0;
1624d5b7
JD
1838 }
1839 stream->tracefile_size_current += len;
309167d2
JD
1840 if (index) {
1841 index->offset = htobe64(stream->out_fd_offset);
1842 }
f02e1e8a
DG
1843 }
1844
d02b8372
DG
1845 /*
1846 * This call guarantee that len or less is returned. It's impossible to
1847 * receive a ret value that is bigger than len.
1848 */
1849 ret = lttng_write(outfd, mmap_base + mmap_offset, len);
1850 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
1851 if (ret < 0 || ((size_t) ret != len)) {
1852 /*
1853 * Report error to caller if nothing was written else at least send the
1854 * amount written.
1855 */
1856 if (ret < 0) {
994ab360 1857 ret = -errno;
f02e1e8a 1858 }
994ab360 1859 relayd_hang_up = 1;
f02e1e8a 1860
d02b8372 1861 /* Socket operation failed. We consider the relayd dead */
fcf0f774 1862 if (errno == EPIPE) {
d02b8372
DG
1863 /*
1864 * This is possible if the fd is closed on the other side
1865 * (outfd) or any write problem. It can be verbose a bit for a
1866 * normal execution if for instance the relayd is stopped
1867 * abruptly. This can happen so set this to a DBG statement.
1868 */
1869 DBG("Consumer mmap write detected relayd hang up");
994ab360
DG
1870 } else {
1871 /* Unhandled error, print it and stop function right now. */
1872 PERROR("Error in write mmap (ret %zd != len %lu)", ret, len);
f02e1e8a 1873 }
994ab360 1874 goto write_error;
d02b8372
DG
1875 }
1876 stream->output_written += ret;
d02b8372
DG
1877
1878 /* This call is useless on a socket so better save a syscall. */
1879 if (!relayd) {
1880 /* This won't block, but will start writeout asynchronously */
1881 lttng_sync_file_range(outfd, stream->out_fd_offset, len,
1882 SYNC_FILE_RANGE_WRITE);
1883 stream->out_fd_offset += len;
f5dbe415 1884 lttng_consumer_sync_trace_file(stream, orig_offset);
f02e1e8a 1885 }
f02e1e8a 1886
8994307f
DG
1887write_error:
1888 /*
1889 * This is a special case that the relayd has closed its socket. Let's
1890 * cleanup the relayd object and all associated streams.
1891 */
1892 if (relayd && relayd_hang_up) {
9276e5c8
JR
1893 ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
1894 lttng_consumer_cleanup_relayd(relayd);
8994307f
DG
1895 }
1896
f02e1e8a
DG
1897end:
1898 /* Unlock only if ctrl socket used */
1899 if (relayd && stream->metadata_flag) {
1900 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1901 }
1902
1903 rcu_read_unlock();
994ab360 1904 return ret;
3bd1e081
MD
1905}
1906
1907/*
1908 * Splice the data from the ring buffer to the tracefile.
1909 *
79d4ffb7
DG
1910 * It must be called with the stream lock held.
1911 *
3bd1e081
MD
1912 * Returns the number of bytes spliced.
1913 */
4078b776 1914ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081 1915 struct lttng_consumer_local_data *ctx,
1d4dfdef 1916 struct lttng_consumer_stream *stream, unsigned long len,
309167d2 1917 unsigned long padding,
50adc264 1918 struct ctf_packet_index *index)
3bd1e081 1919{
f02e1e8a
DG
1920 ssize_t ret = 0, written = 0, ret_splice = 0;
1921 loff_t offset = 0;
1922 off_t orig_offset = stream->out_fd_offset;
1923 int fd = stream->wait_fd;
1924 /* Default is on the disk */
1925 int outfd = stream->out_fd;
f02e1e8a 1926 struct consumer_relayd_sock_pair *relayd = NULL;
fb3a43a9 1927 int *splice_pipe;
8994307f 1928 unsigned int relayd_hang_up = 0;
f02e1e8a 1929
3bd1e081
MD
1930 switch (consumer_data.type) {
1931 case LTTNG_CONSUMER_KERNEL:
f02e1e8a 1932 break;
7753dea8
MD
1933 case LTTNG_CONSUMER32_UST:
1934 case LTTNG_CONSUMER64_UST:
f02e1e8a 1935 /* Not supported for user space tracing */
3bd1e081
MD
1936 return -ENOSYS;
1937 default:
1938 ERR("Unknown consumer_data type");
1939 assert(0);
3bd1e081
MD
1940 }
1941
f02e1e8a
DG
1942 /* RCU lock for the relayd pointer */
1943 rcu_read_lock();
1944
1945 /* Flag that the current stream if set for network streaming. */
da009f2c 1946 if (stream->net_seq_idx != (uint64_t) -1ULL) {
f02e1e8a
DG
1947 relayd = consumer_find_relayd(stream->net_seq_idx);
1948 if (relayd == NULL) {
ad0b0d23 1949 written = -ret;
f02e1e8a
DG
1950 goto end;
1951 }
1952 }
a2361a61 1953 splice_pipe = stream->splice_pipe;
fb3a43a9 1954
f02e1e8a 1955 /* Write metadata stream id before payload */
1d4dfdef 1956 if (relayd) {
ad0b0d23 1957 unsigned long total_len = len;
f02e1e8a 1958
1d4dfdef
DG
1959 if (stream->metadata_flag) {
1960 /*
1961 * Lock the control socket for the complete duration of the function
1962 * since from this point on we will use the socket.
1963 */
1964 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1965
93ec662e
JD
1966 if (stream->reset_metadata_flag) {
1967 ret = relayd_reset_metadata(&relayd->control_sock,
1968 stream->relayd_stream_id,
1969 stream->metadata_version);
1970 if (ret < 0) {
1971 relayd_hang_up = 1;
1972 goto write_error;
1973 }
1974 stream->reset_metadata_flag = 0;
1975 }
239f61af 1976 ret = write_relayd_metadata_id(splice_pipe[1], stream,
1d4dfdef
DG
1977 padding);
1978 if (ret < 0) {
1979 written = ret;
ad0b0d23
DG
1980 relayd_hang_up = 1;
1981 goto write_error;
1d4dfdef
DG
1982 }
1983
1984 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1985 }
1986
1987 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
ad0b0d23
DG
1988 if (ret < 0) {
1989 written = ret;
1990 relayd_hang_up = 1;
1991 goto write_error;
f02e1e8a 1992 }
ad0b0d23
DG
1993 /* Use the returned socket. */
1994 outfd = ret;
1d4dfdef
DG
1995 } else {
1996 /* No streaming, we have to set the len with the full padding */
1997 len += padding;
1624d5b7 1998
93ec662e
JD
1999 if (stream->metadata_flag && stream->reset_metadata_flag) {
2000 ret = utils_truncate_stream_file(stream->out_fd, 0);
2001 if (ret < 0) {
2002 ERR("Reset metadata file");
2003 goto end;
2004 }
2005 stream->reset_metadata_flag = 0;
2006 }
1624d5b7
JD
2007 /*
2008 * Check if we need to change the tracefile before writing the packet.
2009 */
2010 if (stream->chan->tracefile_size > 0 &&
2011 (stream->tracefile_size_current + len) >
2012 stream->chan->tracefile_size) {
d2956687 2013 ret = consumer_stream_rotate_output_files(stream);
1624d5b7 2014 if (ret < 0) {
ad0b0d23 2015 written = ret;
1624d5b7
JD
2016 goto end;
2017 }
309167d2 2018 outfd = stream->out_fd;
a1ae300f 2019 orig_offset = 0;
1624d5b7
JD
2020 }
2021 stream->tracefile_size_current += len;
309167d2 2022 index->offset = htobe64(stream->out_fd_offset);
f02e1e8a
DG
2023 }
2024
2025 while (len > 0) {
1d4dfdef
DG
2026 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
2027 (unsigned long)offset, len, fd, splice_pipe[1]);
fb3a43a9 2028 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
f02e1e8a
DG
2029 SPLICE_F_MOVE | SPLICE_F_MORE);
2030 DBG("splice chan to pipe, ret %zd", ret_splice);
2031 if (ret_splice < 0) {
d02b8372 2032 ret = errno;
ad0b0d23 2033 written = -ret;
d02b8372 2034 PERROR("Error in relay splice");
f02e1e8a
DG
2035 goto splice_error;
2036 }
2037
2038 /* Handle stream on the relayd if the output is on the network */
ad0b0d23
DG
2039 if (relayd && stream->metadata_flag) {
2040 size_t metadata_payload_size =
2041 sizeof(struct lttcomm_relayd_metadata_payload);
2042
2043 /* Update counter to fit the spliced data */
2044 ret_splice += metadata_payload_size;
2045 len += metadata_payload_size;
2046 /*
2047 * We do this so the return value can match the len passed as
2048 * argument to this function.
2049 */
2050 written -= metadata_payload_size;
f02e1e8a
DG
2051 }
2052
2053 /* Splice data out */
fb3a43a9 2054 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
f02e1e8a 2055 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
a2361a61
JD
2056 DBG("Consumer splice pipe to file (out_fd: %d), ret %zd",
2057 outfd, ret_splice);
f02e1e8a 2058 if (ret_splice < 0) {
d02b8372 2059 ret = errno;
ad0b0d23
DG
2060 written = -ret;
2061 relayd_hang_up = 1;
2062 goto write_error;
f02e1e8a 2063 } else if (ret_splice > len) {
d02b8372
DG
2064 /*
2065 * We don't expect this code path to be executed but you never know
2066 * so this is an extra protection agains a buggy splice().
2067 */
f02e1e8a 2068 ret = errno;
ad0b0d23 2069 written += ret_splice;
d02b8372
DG
2070 PERROR("Wrote more data than requested %zd (len: %lu)", ret_splice,
2071 len);
f02e1e8a 2072 goto splice_error;
d02b8372
DG
2073 } else {
2074 /* All good, update current len and continue. */
2075 len -= ret_splice;
f02e1e8a 2076 }
f02e1e8a
DG
2077
2078 /* This call is useless on a socket so better save a syscall. */
2079 if (!relayd) {
2080 /* This won't block, but will start writeout asynchronously */
2081 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
2082 SYNC_FILE_RANGE_WRITE);
2083 stream->out_fd_offset += ret_splice;
2084 }
e5d1a9b3 2085 stream->output_written += ret_splice;
f02e1e8a
DG
2086 written += ret_splice;
2087 }
f5dbe415
JG
2088 if (!relayd) {
2089 lttng_consumer_sync_trace_file(stream, orig_offset);
2090 }
f02e1e8a
DG
2091 goto end;
2092
8994307f
DG
2093write_error:
2094 /*
2095 * This is a special case that the relayd has closed its socket. Let's
2096 * cleanup the relayd object and all associated streams.
2097 */
2098 if (relayd && relayd_hang_up) {
9276e5c8
JR
2099 ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
2100 lttng_consumer_cleanup_relayd(relayd);
8994307f
DG
2101 /* Skip splice error so the consumer does not fail */
2102 goto end;
2103 }
2104
f02e1e8a
DG
2105splice_error:
2106 /* send the appropriate error description to sessiond */
2107 switch (ret) {
f02e1e8a 2108 case EINVAL:
f73fabfd 2109 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
f02e1e8a
DG
2110 break;
2111 case ENOMEM:
f73fabfd 2112 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
f02e1e8a
DG
2113 break;
2114 case ESPIPE:
f73fabfd 2115 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
f02e1e8a
DG
2116 break;
2117 }
2118
2119end:
2120 if (relayd && stream->metadata_flag) {
2121 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
2122 }
2123
2124 rcu_read_unlock();
2125 return written;
3bd1e081
MD
2126}
2127
15055ce5
JD
2128/*
2129 * Sample the snapshot positions for a specific fd
2130 *
2131 * Returns 0 on success, < 0 on error
2132 */
2133int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
2134{
2135 switch (consumer_data.type) {
2136 case LTTNG_CONSUMER_KERNEL:
2137 return lttng_kconsumer_sample_snapshot_positions(stream);
2138 case LTTNG_CONSUMER32_UST:
2139 case LTTNG_CONSUMER64_UST:
2140 return lttng_ustconsumer_sample_snapshot_positions(stream);
2141 default:
2142 ERR("Unknown consumer_data type");
2143 assert(0);
2144 return -ENOSYS;
2145 }
2146}
3bd1e081
MD
2147/*
2148 * Take a snapshot for a specific fd
2149 *
2150 * Returns 0 on success, < 0 on error
2151 */
ffe60014 2152int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081
MD
2153{
2154 switch (consumer_data.type) {
2155 case LTTNG_CONSUMER_KERNEL:
ffe60014 2156 return lttng_kconsumer_take_snapshot(stream);
7753dea8
MD
2157 case LTTNG_CONSUMER32_UST:
2158 case LTTNG_CONSUMER64_UST:
ffe60014 2159 return lttng_ustconsumer_take_snapshot(stream);
3bd1e081
MD
2160 default:
2161 ERR("Unknown consumer_data type");
2162 assert(0);
2163 return -ENOSYS;
2164 }
3bd1e081
MD
2165}
2166
2167/*
2168 * Get the produced position
2169 *
2170 * Returns 0 on success, < 0 on error
2171 */
ffe60014 2172int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
3bd1e081
MD
2173 unsigned long *pos)
2174{
2175 switch (consumer_data.type) {
2176 case LTTNG_CONSUMER_KERNEL:
ffe60014 2177 return lttng_kconsumer_get_produced_snapshot(stream, pos);
7753dea8
MD
2178 case LTTNG_CONSUMER32_UST:
2179 case LTTNG_CONSUMER64_UST:
ffe60014 2180 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
3bd1e081
MD
2181 default:
2182 ERR("Unknown consumer_data type");
2183 assert(0);
2184 return -ENOSYS;
2185 }
2186}
2187
15055ce5
JD
2188/*
2189 * Get the consumed position (free-running counter position in bytes).
2190 *
2191 * Returns 0 on success, < 0 on error
2192 */
2193int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
2194 unsigned long *pos)
2195{
2196 switch (consumer_data.type) {
2197 case LTTNG_CONSUMER_KERNEL:
2198 return lttng_kconsumer_get_consumed_snapshot(stream, pos);
2199 case LTTNG_CONSUMER32_UST:
2200 case LTTNG_CONSUMER64_UST:
2201 return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
2202 default:
2203 ERR("Unknown consumer_data type");
2204 assert(0);
2205 return -ENOSYS;
2206 }
2207}
2208
3bd1e081
MD
2209int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
2210 int sock, struct pollfd *consumer_sockpoll)
2211{
2212 switch (consumer_data.type) {
2213 case LTTNG_CONSUMER_KERNEL:
2214 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
2215 case LTTNG_CONSUMER32_UST:
2216 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
2217 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
2218 default:
2219 ERR("Unknown consumer_data type");
2220 assert(0);
2221 return -ENOSYS;
2222 }
2223}
2224
6d574024 2225void lttng_consumer_close_all_metadata(void)
d88aee68
DG
2226{
2227 switch (consumer_data.type) {
2228 case LTTNG_CONSUMER_KERNEL:
2229 /*
2230 * The Kernel consumer has a different metadata scheme so we don't
2231 * close anything because the stream will be closed by the session
2232 * daemon.
2233 */
2234 break;
2235 case LTTNG_CONSUMER32_UST:
2236 case LTTNG_CONSUMER64_UST:
2237 /*
2238 * Close all metadata streams. The metadata hash table is passed and
2239 * this call iterates over it by closing all wakeup fd. This is safe
2240 * because at this point we are sure that the metadata producer is
2241 * either dead or blocked.
2242 */
6d574024 2243 lttng_ustconsumer_close_all_metadata(metadata_ht);
d88aee68
DG
2244 break;
2245 default:
2246 ERR("Unknown consumer_data type");
2247 assert(0);
2248 }
2249}
2250
fb3a43a9
DG
2251/*
2252 * Clean up a metadata stream and free its memory.
2253 */
e316aad5
DG
2254void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
2255 struct lttng_ht *ht)
fb3a43a9 2256{
e316aad5 2257 struct lttng_consumer_channel *free_chan = NULL;
fb3a43a9
DG
2258
2259 assert(stream);
2260 /*
2261 * This call should NEVER receive regular stream. It must always be
2262 * metadata stream and this is crucial for data structure synchronization.
2263 */
2264 assert(stream->metadata_flag);
2265
e316aad5
DG
2266 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
2267
74251bb8 2268 pthread_mutex_lock(&consumer_data.lock);
fb549e7a 2269 pthread_mutex_lock(&stream->chan->lock);
3dad2c0f 2270 pthread_mutex_lock(&stream->lock);
081424af
JG
2271 if (stream->chan->metadata_cache) {
2272 /* Only applicable to userspace consumers. */
2273 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
2274 }
8994307f 2275
6d574024
DG
2276 /* Remove any reference to that stream. */
2277 consumer_stream_delete(stream, ht);
ca22feea 2278
6d574024
DG
2279 /* Close down everything including the relayd if one. */
2280 consumer_stream_close(stream);
2281 /* Destroy tracer buffers of the stream. */
2282 consumer_stream_destroy_buffers(stream);
fb3a43a9
DG
2283
2284 /* Atomically decrement channel refcount since other threads can use it. */
f2ad556d 2285 if (!uatomic_sub_return(&stream->chan->refcount, 1)
ffe60014 2286 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
c30aaa51 2287 /* Go for channel deletion! */
e316aad5 2288 free_chan = stream->chan;
fb3a43a9
DG
2289 }
2290
73811ecc
DG
2291 /*
2292 * Nullify the stream reference so it is not used after deletion. The
6d574024
DG
2293 * channel lock MUST be acquired before being able to check for a NULL
2294 * pointer value.
73811ecc
DG
2295 */
2296 stream->chan->metadata_stream = NULL;
2297
081424af
JG
2298 if (stream->chan->metadata_cache) {
2299 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2300 }
3dad2c0f 2301 pthread_mutex_unlock(&stream->lock);
fb549e7a 2302 pthread_mutex_unlock(&stream->chan->lock);
74251bb8 2303 pthread_mutex_unlock(&consumer_data.lock);
e316aad5
DG
2304
2305 if (free_chan) {
2306 consumer_del_channel(free_chan);
2307 }
2308
d2956687
JG
2309 lttng_trace_chunk_put(stream->trace_chunk);
2310 stream->trace_chunk = NULL;
6d574024 2311 consumer_stream_free(stream);
fb3a43a9
DG
2312}
2313
2314/*
2315 * Action done with the metadata stream when adding it to the consumer internal
2316 * data structures to handle it.
2317 */
66d583dc 2318void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
fb3a43a9 2319{
5ab66908 2320 struct lttng_ht *ht = metadata_ht;
76082088 2321 struct lttng_ht_iter iter;
d88aee68 2322 struct lttng_ht_node_u64 *node;
fb3a43a9 2323
e316aad5
DG
2324 assert(stream);
2325 assert(ht);
2326
d88aee68 2327 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
e316aad5
DG
2328
2329 pthread_mutex_lock(&consumer_data.lock);
a9838785 2330 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 2331 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 2332 pthread_mutex_lock(&stream->lock);
e316aad5 2333
e316aad5
DG
2334 /*
2335 * From here, refcounts are updated so be _careful_ when returning an error
2336 * after this point.
2337 */
2338
fb3a43a9 2339 rcu_read_lock();
76082088
DG
2340
2341 /*
2342 * Lookup the stream just to make sure it does not exist in our internal
2343 * state. This should NEVER happen.
2344 */
d88aee68
DG
2345 lttng_ht_lookup(ht, &stream->key, &iter);
2346 node = lttng_ht_iter_get_node_u64(&iter);
76082088
DG
2347 assert(!node);
2348
e316aad5 2349 /*
ffe60014
DG
2350 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2351 * in terms of destroying the associated channel, because the action that
e316aad5
DG
2352 * causes the count to become 0 also causes a stream to be added. The
2353 * channel deletion will thus be triggered by the following removal of this
2354 * stream.
2355 */
ffe60014 2356 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
2357 /* Increment refcount before decrementing nb_init_stream_left */
2358 cmm_smp_wmb();
ffe60014 2359 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
2360 }
2361
d88aee68 2362 lttng_ht_add_unique_u64(ht, &stream->node);
ca22feea 2363
446156b4 2364 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
d8ef542d
MD
2365 &stream->node_channel_id);
2366
ca22feea
DG
2367 /*
2368 * Add stream to the stream_list_ht of the consumer data. No need to steal
2369 * the key since the HT does not use it and we allow to add redundant keys
2370 * into this table.
2371 */
d88aee68 2372 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 2373
fb3a43a9 2374 rcu_read_unlock();
e316aad5 2375
2e818a6a 2376 pthread_mutex_unlock(&stream->lock);
a9838785 2377 pthread_mutex_unlock(&stream->chan->lock);
ec6ea7d0 2378 pthread_mutex_unlock(&stream->chan->timer_lock);
e316aad5 2379 pthread_mutex_unlock(&consumer_data.lock);
fb3a43a9
DG
2380}
2381
8994307f
DG
2382/*
2383 * Delete data stream that are flagged for deletion (endpoint_status).
2384 */
2385static void validate_endpoint_status_data_stream(void)
2386{
2387 struct lttng_ht_iter iter;
2388 struct lttng_consumer_stream *stream;
2389
2390 DBG("Consumer delete flagged data stream");
2391
2392 rcu_read_lock();
2393 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2394 /* Validate delete flag of the stream */
79d4ffb7 2395 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2396 continue;
2397 }
2398 /* Delete it right now */
2399 consumer_del_stream(stream, data_ht);
2400 }
2401 rcu_read_unlock();
2402}
2403
2404/*
2405 * Delete metadata stream that are flagged for deletion (endpoint_status).
2406 */
2407static void validate_endpoint_status_metadata_stream(
2408 struct lttng_poll_event *pollset)
2409{
2410 struct lttng_ht_iter iter;
2411 struct lttng_consumer_stream *stream;
2412
2413 DBG("Consumer delete flagged metadata stream");
2414
2415 assert(pollset);
2416
2417 rcu_read_lock();
2418 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2419 /* Validate delete flag of the stream */
79d4ffb7 2420 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2421 continue;
2422 }
2423 /*
2424 * Remove from pollset so the metadata thread can continue without
2425 * blocking on a deleted stream.
2426 */
2427 lttng_poll_del(pollset, stream->wait_fd);
2428
2429 /* Delete it right now */
2430 consumer_del_metadata_stream(stream, metadata_ht);
2431 }
2432 rcu_read_unlock();
2433}
2434
fb3a43a9
DG
2435/*
2436 * Thread polls on metadata file descriptor and write them on disk or on the
2437 * network.
2438 */
7d980def 2439void *consumer_thread_metadata_poll(void *data)
fb3a43a9 2440{
1fc79fb4 2441 int ret, i, pollfd, err = -1;
fb3a43a9 2442 uint32_t revents, nb_fd;
e316aad5 2443 struct lttng_consumer_stream *stream = NULL;
fb3a43a9 2444 struct lttng_ht_iter iter;
d88aee68 2445 struct lttng_ht_node_u64 *node;
fb3a43a9
DG
2446 struct lttng_poll_event events;
2447 struct lttng_consumer_local_data *ctx = data;
2448 ssize_t len;
2449
2450 rcu_register_thread();
2451
1fc79fb4
MD
2452 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA);
2453
2d57de81
MD
2454 if (testpoint(consumerd_thread_metadata)) {
2455 goto error_testpoint;
2456 }
2457
9ce5646a
MD
2458 health_code_update();
2459
fb3a43a9
DG
2460 DBG("Thread metadata poll started");
2461
fb3a43a9
DG
2462 /* Size is set to 1 for the consumer_metadata pipe */
2463 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2464 if (ret < 0) {
2465 ERR("Poll set creation failed");
d8ef542d 2466 goto end_poll;
fb3a43a9
DG
2467 }
2468
13886d2d
DG
2469 ret = lttng_poll_add(&events,
2470 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
fb3a43a9
DG
2471 if (ret < 0) {
2472 goto end;
2473 }
2474
2475 /* Main loop */
2476 DBG("Metadata main loop started");
2477
2478 while (1) {
fb3a43a9 2479restart:
7fa2082e 2480 health_code_update();
9ce5646a 2481 health_poll_entry();
7fa2082e 2482 DBG("Metadata poll wait");
fb3a43a9 2483 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
2484 DBG("Metadata poll return from wait with %d fd(s)",
2485 LTTNG_POLL_GETNB(&events));
9ce5646a 2486 health_poll_exit();
40063ead 2487 DBG("Metadata event caught in thread");
fb3a43a9
DG
2488 if (ret < 0) {
2489 if (errno == EINTR) {
40063ead 2490 ERR("Poll EINTR caught");
fb3a43a9
DG
2491 goto restart;
2492 }
d9607cd7
MD
2493 if (LTTNG_POLL_GETNB(&events) == 0) {
2494 err = 0; /* All is OK */
2495 }
2496 goto end;
fb3a43a9
DG
2497 }
2498
0d9c5d77
DG
2499 nb_fd = ret;
2500
e316aad5 2501 /* From here, the event is a metadata wait fd */
fb3a43a9 2502 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2503 health_code_update();
2504
fb3a43a9
DG
2505 revents = LTTNG_POLL_GETEV(&events, i);
2506 pollfd = LTTNG_POLL_GETFD(&events, i);
2507
13886d2d 2508 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
03e43155 2509 if (revents & LPOLLIN) {
13886d2d
DG
2510 ssize_t pipe_len;
2511
2512 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2513 &stream, sizeof(stream));
6cd525e8 2514 if (pipe_len < sizeof(stream)) {
03e43155
MD
2515 if (pipe_len < 0) {
2516 PERROR("read metadata stream");
2517 }
fb3a43a9 2518 /*
03e43155
MD
2519 * Remove the pipe from the poll set and continue the loop
2520 * since their might be data to consume.
fb3a43a9 2521 */
03e43155
MD
2522 lttng_poll_del(&events,
2523 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2524 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
fb3a43a9
DG
2525 continue;
2526 }
2527
8994307f
DG
2528 /* A NULL stream means that the state has changed. */
2529 if (stream == NULL) {
2530 /* Check for deleted streams. */
2531 validate_endpoint_status_metadata_stream(&events);
3714380f 2532 goto restart;
8994307f
DG
2533 }
2534
fb3a43a9
DG
2535 DBG("Adding metadata stream %d to poll set",
2536 stream->wait_fd);
2537
fb3a43a9
DG
2538 /* Add metadata stream to the global poll events list */
2539 lttng_poll_add(&events, stream->wait_fd,
6d574024 2540 LPOLLIN | LPOLLPRI | LPOLLHUP);
03e43155
MD
2541 } else if (revents & (LPOLLERR | LPOLLHUP)) {
2542 DBG("Metadata thread pipe hung up");
2543 /*
2544 * Remove the pipe from the poll set and continue the loop
2545 * since their might be data to consume.
2546 */
2547 lttng_poll_del(&events,
2548 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2549 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2550 continue;
2551 } else {
2552 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2553 goto end;
fb3a43a9
DG
2554 }
2555
e316aad5 2556 /* Handle other stream */
fb3a43a9
DG
2557 continue;
2558 }
2559
d09e1200 2560 rcu_read_lock();
d88aee68
DG
2561 {
2562 uint64_t tmp_id = (uint64_t) pollfd;
2563
2564 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2565 }
2566 node = lttng_ht_iter_get_node_u64(&iter);
e316aad5 2567 assert(node);
fb3a43a9
DG
2568
2569 stream = caa_container_of(node, struct lttng_consumer_stream,
58b1f425 2570 node);
fb3a43a9 2571
03e43155
MD
2572 if (revents & (LPOLLIN | LPOLLPRI)) {
2573 /* Get the data out of the metadata file descriptor */
2574 DBG("Metadata available on fd %d", pollfd);
2575 assert(stream->wait_fd == pollfd);
2576
2577 do {
2578 health_code_update();
2579
2580 len = ctx->on_buffer_ready(stream, ctx);
2581 /*
2582 * We don't check the return value here since if we get
83f4233d 2583 * a negative len, it means an error occurred thus we
03e43155
MD
2584 * simply remove it from the poll set and free the
2585 * stream.
2586 */
2587 } while (len > 0);
2588
2589 /* It's ok to have an unavailable sub-buffer */
2590 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2591 /* Clean up stream from consumer and free it. */
2592 lttng_poll_del(&events, stream->wait_fd);
2593 consumer_del_metadata_stream(stream, metadata_ht);
2594 }
2595 } else if (revents & (LPOLLERR | LPOLLHUP)) {
e316aad5 2596 DBG("Metadata fd %d is hup|err.", pollfd);
fb3a43a9
DG
2597 if (!stream->hangup_flush_done
2598 && (consumer_data.type == LTTNG_CONSUMER32_UST
2599 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2600 DBG("Attempting to flush and consume the UST buffers");
2601 lttng_ustconsumer_on_stream_hangup(stream);
2602
2603 /* We just flushed the stream now read it. */
4bb94b75 2604 do {
9ce5646a
MD
2605 health_code_update();
2606
4bb94b75
DG
2607 len = ctx->on_buffer_ready(stream, ctx);
2608 /*
2609 * We don't check the return value here since if we get
83f4233d 2610 * a negative len, it means an error occurred thus we
4bb94b75
DG
2611 * simply remove it from the poll set and free the
2612 * stream.
2613 */
2614 } while (len > 0);
fb3a43a9
DG
2615 }
2616
fb3a43a9 2617 lttng_poll_del(&events, stream->wait_fd);
e316aad5
DG
2618 /*
2619 * This call update the channel states, closes file descriptors
2620 * and securely free the stream.
2621 */
2622 consumer_del_metadata_stream(stream, metadata_ht);
03e43155
MD
2623 } else {
2624 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
6f2f1a70 2625 rcu_read_unlock();
03e43155 2626 goto end;
fb3a43a9 2627 }
e316aad5 2628 /* Release RCU lock for the stream looked up */
d09e1200 2629 rcu_read_unlock();
fb3a43a9
DG
2630 }
2631 }
2632
1fc79fb4
MD
2633 /* All is OK */
2634 err = 0;
fb3a43a9
DG
2635end:
2636 DBG("Metadata poll thread exiting");
fb3a43a9 2637
d8ef542d
MD
2638 lttng_poll_clean(&events);
2639end_poll:
2d57de81 2640error_testpoint:
1fc79fb4
MD
2641 if (err) {
2642 health_error();
2643 ERR("Health error occurred in %s", __func__);
2644 }
2645 health_unregister(health_consumerd);
fb3a43a9
DG
2646 rcu_unregister_thread();
2647 return NULL;
2648}
2649
3bd1e081 2650/*
e4421fec 2651 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
2652 * it to tracefile if necessary.
2653 */
7d980def 2654void *consumer_thread_data_poll(void *data)
3bd1e081 2655{
1fc79fb4 2656 int num_rdy, num_hup, high_prio, ret, i, err = -1;
3bd1e081
MD
2657 struct pollfd *pollfd = NULL;
2658 /* local view of the streams */
c869f647 2659 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
3bd1e081 2660 /* local view of consumer_data.fds_count */
8bdcc002
JG
2661 int nb_fd = 0;
2662 /* 2 for the consumer_data_pipe and wake up pipe */
2663 const int nb_pipes_fd = 2;
9a2fcf78
JD
2664 /* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
2665 int nb_inactive_fd = 0;
3bd1e081 2666 struct lttng_consumer_local_data *ctx = data;
00e2e675 2667 ssize_t len;
3bd1e081 2668
e7b994a3
DG
2669 rcu_register_thread();
2670
1fc79fb4
MD
2671 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA);
2672
2d57de81
MD
2673 if (testpoint(consumerd_thread_data)) {
2674 goto error_testpoint;
2675 }
2676
9ce5646a
MD
2677 health_code_update();
2678
4df6c8cb
MD
2679 local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
2680 if (local_stream == NULL) {
2681 PERROR("local_stream malloc");
2682 goto end;
2683 }
3bd1e081
MD
2684
2685 while (1) {
9ce5646a
MD
2686 health_code_update();
2687
3bd1e081
MD
2688 high_prio = 0;
2689 num_hup = 0;
2690
2691 /*
e4421fec 2692 * the fds set has been updated, we need to update our
3bd1e081
MD
2693 * local array as well
2694 */
2695 pthread_mutex_lock(&consumer_data.lock);
2696 if (consumer_data.need_update) {
0e428499
DG
2697 free(pollfd);
2698 pollfd = NULL;
2699
2700 free(local_stream);
2701 local_stream = NULL;
3bd1e081 2702
8bdcc002 2703 /* Allocate for all fds */
261de637 2704 pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
3bd1e081 2705 if (pollfd == NULL) {
7a57cf92 2706 PERROR("pollfd malloc");
3bd1e081
MD
2707 pthread_mutex_unlock(&consumer_data.lock);
2708 goto end;
2709 }
2710
261de637 2711 local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
747f8642 2712 sizeof(struct lttng_consumer_stream *));
3bd1e081 2713 if (local_stream == NULL) {
7a57cf92 2714 PERROR("local_stream malloc");
3bd1e081
MD
2715 pthread_mutex_unlock(&consumer_data.lock);
2716 goto end;
2717 }
ffe60014 2718 ret = update_poll_array(ctx, &pollfd, local_stream,
9a2fcf78 2719 data_ht, &nb_inactive_fd);
3bd1e081
MD
2720 if (ret < 0) {
2721 ERR("Error in allocating pollfd or local_outfds");
f73fabfd 2722 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2723 pthread_mutex_unlock(&consumer_data.lock);
2724 goto end;
2725 }
2726 nb_fd = ret;
2727 consumer_data.need_update = 0;
2728 }
2729 pthread_mutex_unlock(&consumer_data.lock);
2730
4078b776 2731 /* No FDs and consumer_quit, consumer_cleanup the thread */
9a2fcf78
JD
2732 if (nb_fd == 0 && nb_inactive_fd == 0 &&
2733 CMM_LOAD_SHARED(consumer_quit) == 1) {
1fc79fb4 2734 err = 0; /* All is OK */
4078b776
MD
2735 goto end;
2736 }
3bd1e081 2737 /* poll on the array of fds */
88f2b785 2738 restart:
261de637 2739 DBG("polling on %d fd", nb_fd + nb_pipes_fd);
cf0bcb51
JG
2740 if (testpoint(consumerd_thread_data_poll)) {
2741 goto end;
2742 }
9ce5646a 2743 health_poll_entry();
261de637 2744 num_rdy = poll(pollfd, nb_fd + nb_pipes_fd, -1);
9ce5646a 2745 health_poll_exit();
3bd1e081
MD
2746 DBG("poll num_rdy : %d", num_rdy);
2747 if (num_rdy == -1) {
88f2b785
MD
2748 /*
2749 * Restart interrupted system call.
2750 */
2751 if (errno == EINTR) {
2752 goto restart;
2753 }
7a57cf92 2754 PERROR("Poll error");
f73fabfd 2755 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2756 goto end;
2757 } else if (num_rdy == 0) {
2758 DBG("Polling thread timed out");
2759 goto end;
2760 }
2761
80957876
JG
2762 if (caa_unlikely(data_consumption_paused)) {
2763 DBG("Data consumption paused, sleeping...");
2764 sleep(1);
2765 goto restart;
2766 }
2767
3bd1e081 2768 /*
50f8ae69 2769 * If the consumer_data_pipe triggered poll go directly to the
00e2e675
DG
2770 * beginning of the loop to update the array. We want to prioritize
2771 * array update over low-priority reads.
3bd1e081 2772 */
509bb1cf 2773 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
ab30f567 2774 ssize_t pipe_readlen;
04fdd819 2775
50f8ae69 2776 DBG("consumer_data_pipe wake up");
acdb9057
DG
2777 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2778 &new_stream, sizeof(new_stream));
6cd525e8
MD
2779 if (pipe_readlen < sizeof(new_stream)) {
2780 PERROR("Consumer data pipe");
23f5f35d
DG
2781 /* Continue so we can at least handle the current stream(s). */
2782 continue;
2783 }
c869f647
DG
2784
2785 /*
2786 * If the stream is NULL, just ignore it. It's also possible that
2787 * the sessiond poll thread changed the consumer_quit state and is
2788 * waking us up to test it.
2789 */
2790 if (new_stream == NULL) {
8994307f 2791 validate_endpoint_status_data_stream();
c869f647
DG
2792 continue;
2793 }
2794
c869f647 2795 /* Continue to update the local streams and handle prio ones */
3bd1e081
MD
2796 continue;
2797 }
2798
02b3d176
DG
2799 /* Handle wakeup pipe. */
2800 if (pollfd[nb_fd + 1].revents & (POLLIN | POLLPRI)) {
2801 char dummy;
2802 ssize_t pipe_readlen;
2803
2804 pipe_readlen = lttng_pipe_read(ctx->consumer_wakeup_pipe, &dummy,
2805 sizeof(dummy));
2806 if (pipe_readlen < 0) {
2807 PERROR("Consumer data wakeup pipe");
2808 }
2809 /* We've been awakened to handle stream(s). */
2810 ctx->has_wakeup = 0;
2811 }
2812
3bd1e081
MD
2813 /* Take care of high priority channels first. */
2814 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2815 health_code_update();
2816
9617607b
DG
2817 if (local_stream[i] == NULL) {
2818 continue;
2819 }
fb3a43a9 2820 if (pollfd[i].revents & POLLPRI) {
d41f73b7
MD
2821 DBG("Urgent read on fd %d", pollfd[i].fd);
2822 high_prio = 1;
4078b776 2823 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 2824 /* it's ok to have an unavailable sub-buffer */
b64403e3 2825 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2826 /* Clean the stream and free it. */
2827 consumer_del_stream(local_stream[i], data_ht);
9617607b 2828 local_stream[i] = NULL;
4078b776
MD
2829 } else if (len > 0) {
2830 local_stream[i]->data_read = 1;
d41f73b7 2831 }
3bd1e081
MD
2832 }
2833 }
2834
4078b776
MD
2835 /*
2836 * If we read high prio channel in this loop, try again
2837 * for more high prio data.
2838 */
2839 if (high_prio) {
3bd1e081
MD
2840 continue;
2841 }
2842
2843 /* Take care of low priority channels. */
4078b776 2844 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2845 health_code_update();
2846
9617607b
DG
2847 if (local_stream[i] == NULL) {
2848 continue;
2849 }
4078b776 2850 if ((pollfd[i].revents & POLLIN) ||
02b3d176
DG
2851 local_stream[i]->hangup_flush_done ||
2852 local_stream[i]->has_data) {
4078b776
MD
2853 DBG("Normal read on fd %d", pollfd[i].fd);
2854 len = ctx->on_buffer_ready(local_stream[i], ctx);
2855 /* it's ok to have an unavailable sub-buffer */
b64403e3 2856 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2857 /* Clean the stream and free it. */
2858 consumer_del_stream(local_stream[i], data_ht);
9617607b 2859 local_stream[i] = NULL;
4078b776
MD
2860 } else if (len > 0) {
2861 local_stream[i]->data_read = 1;
2862 }
2863 }
2864 }
2865
2866 /* Handle hangup and errors */
2867 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2868 health_code_update();
2869
9617607b
DG
2870 if (local_stream[i] == NULL) {
2871 continue;
2872 }
4078b776
MD
2873 if (!local_stream[i]->hangup_flush_done
2874 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2875 && (consumer_data.type == LTTNG_CONSUMER32_UST
2876 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2877 DBG("fd %d is hup|err|nval. Attempting flush and read.",
9617607b 2878 pollfd[i].fd);
4078b776
MD
2879 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2880 /* Attempt read again, for the data we just flushed. */
2881 local_stream[i]->data_read = 1;
2882 }
2883 /*
2884 * If the poll flag is HUP/ERR/NVAL and we have
2885 * read no data in this pass, we can remove the
2886 * stream from its hash table.
2887 */
2888 if ((pollfd[i].revents & POLLHUP)) {
2889 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2890 if (!local_stream[i]->data_read) {
43c34bc3 2891 consumer_del_stream(local_stream[i], data_ht);
9617607b 2892 local_stream[i] = NULL;
4078b776
MD
2893 num_hup++;
2894 }
2895 } else if (pollfd[i].revents & POLLERR) {
2896 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2897 if (!local_stream[i]->data_read) {
43c34bc3 2898 consumer_del_stream(local_stream[i], data_ht);
9617607b 2899 local_stream[i] = NULL;
4078b776
MD
2900 num_hup++;
2901 }
2902 } else if (pollfd[i].revents & POLLNVAL) {
2903 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2904 if (!local_stream[i]->data_read) {
43c34bc3 2905 consumer_del_stream(local_stream[i], data_ht);
9617607b 2906 local_stream[i] = NULL;
4078b776 2907 num_hup++;
3bd1e081
MD
2908 }
2909 }
9617607b
DG
2910 if (local_stream[i] != NULL) {
2911 local_stream[i]->data_read = 0;
2912 }
3bd1e081
MD
2913 }
2914 }
1fc79fb4
MD
2915 /* All is OK */
2916 err = 0;
3bd1e081
MD
2917end:
2918 DBG("polling thread exiting");
0e428499
DG
2919 free(pollfd);
2920 free(local_stream);
fb3a43a9
DG
2921
2922 /*
2923 * Close the write side of the pipe so epoll_wait() in
7d980def
DG
2924 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2925 * read side of the pipe. If we close them both, epoll_wait strangely does
2926 * not return and could create a endless wait period if the pipe is the
2927 * only tracked fd in the poll set. The thread will take care of closing
2928 * the read side.
fb3a43a9 2929 */
13886d2d 2930 (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
fb3a43a9 2931
2d57de81 2932error_testpoint:
1fc79fb4
MD
2933 if (err) {
2934 health_error();
2935 ERR("Health error occurred in %s", __func__);
2936 }
2937 health_unregister(health_consumerd);
2938
e7b994a3 2939 rcu_unregister_thread();
3bd1e081
MD
2940 return NULL;
2941}
2942
d8ef542d
MD
2943/*
2944 * Close wake-up end of each stream belonging to the channel. This will
2945 * allow the poll() on the stream read-side to detect when the
2946 * write-side (application) finally closes them.
2947 */
2948static
2949void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
2950{
2951 struct lttng_ht *ht;
2952 struct lttng_consumer_stream *stream;
2953 struct lttng_ht_iter iter;
2954
2955 ht = consumer_data.stream_per_chan_id_ht;
2956
2957 rcu_read_lock();
2958 cds_lfht_for_each_entry_duplicate(ht->ht,
2959 ht->hash_fct(&channel->key, lttng_ht_seed),
2960 ht->match_fct, &channel->key,
2961 &iter.iter, stream, node_channel_id.node) {
f2ad556d
MD
2962 /*
2963 * Protect against teardown with mutex.
2964 */
2965 pthread_mutex_lock(&stream->lock);
2966 if (cds_lfht_is_node_deleted(&stream->node.node)) {
2967 goto next;
2968 }
d8ef542d
MD
2969 switch (consumer_data.type) {
2970 case LTTNG_CONSUMER_KERNEL:
2971 break;
2972 case LTTNG_CONSUMER32_UST:
2973 case LTTNG_CONSUMER64_UST:
b4a650f3
DG
2974 if (stream->metadata_flag) {
2975 /* Safe and protected by the stream lock. */
2976 lttng_ustconsumer_close_metadata(stream->chan);
2977 } else {
2978 /*
2979 * Note: a mutex is taken internally within
2980 * liblttng-ust-ctl to protect timer wakeup_fd
2981 * use from concurrent close.
2982 */
2983 lttng_ustconsumer_close_stream_wakeup(stream);
2984 }
d8ef542d
MD
2985 break;
2986 default:
2987 ERR("Unknown consumer_data type");
2988 assert(0);
2989 }
f2ad556d
MD
2990 next:
2991 pthread_mutex_unlock(&stream->lock);
d8ef542d
MD
2992 }
2993 rcu_read_unlock();
2994}
2995
2996static void destroy_channel_ht(struct lttng_ht *ht)
2997{
2998 struct lttng_ht_iter iter;
2999 struct lttng_consumer_channel *channel;
3000 int ret;
3001
3002 if (ht == NULL) {
3003 return;
3004 }
3005
3006 rcu_read_lock();
3007 cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
3008 ret = lttng_ht_del(ht, &iter);
3009 assert(ret != 0);
3010 }
3011 rcu_read_unlock();
3012
3013 lttng_ht_destroy(ht);
3014}
3015
3016/*
3017 * This thread polls the channel fds to detect when they are being
3018 * closed. It closes all related streams if the channel is detected as
3019 * closed. It is currently only used as a shim layer for UST because the
3020 * consumerd needs to keep the per-stream wakeup end of pipes open for
3021 * periodical flush.
3022 */
3023void *consumer_thread_channel_poll(void *data)
3024{
1fc79fb4 3025 int ret, i, pollfd, err = -1;
d8ef542d
MD
3026 uint32_t revents, nb_fd;
3027 struct lttng_consumer_channel *chan = NULL;
3028 struct lttng_ht_iter iter;
3029 struct lttng_ht_node_u64 *node;
3030 struct lttng_poll_event events;
3031 struct lttng_consumer_local_data *ctx = data;
3032 struct lttng_ht *channel_ht;
3033
3034 rcu_register_thread();
3035
1fc79fb4
MD
3036 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL);
3037
2d57de81
MD
3038 if (testpoint(consumerd_thread_channel)) {
3039 goto error_testpoint;
3040 }
3041
9ce5646a
MD
3042 health_code_update();
3043
d8ef542d
MD
3044 channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3045 if (!channel_ht) {
3046 /* ENOMEM at this point. Better to bail out. */
3047 goto end_ht;
3048 }
3049
3050 DBG("Thread channel poll started");
3051
3052 /* Size is set to 1 for the consumer_channel pipe */
3053 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
3054 if (ret < 0) {
3055 ERR("Poll set creation failed");
3056 goto end_poll;
3057 }
3058
3059 ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
3060 if (ret < 0) {
3061 goto end;
3062 }
3063
3064 /* Main loop */
3065 DBG("Channel main loop started");
3066
3067 while (1) {
d8ef542d 3068restart:
7fa2082e
MD
3069 health_code_update();
3070 DBG("Channel poll wait");
9ce5646a 3071 health_poll_entry();
d8ef542d 3072 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
3073 DBG("Channel poll return from wait with %d fd(s)",
3074 LTTNG_POLL_GETNB(&events));
9ce5646a 3075 health_poll_exit();
40063ead 3076 DBG("Channel event caught in thread");
d8ef542d
MD
3077 if (ret < 0) {
3078 if (errno == EINTR) {
40063ead 3079 ERR("Poll EINTR caught");
d8ef542d
MD
3080 goto restart;
3081 }
d9607cd7
MD
3082 if (LTTNG_POLL_GETNB(&events) == 0) {
3083 err = 0; /* All is OK */
3084 }
d8ef542d
MD
3085 goto end;
3086 }
3087
3088 nb_fd = ret;
3089
3090 /* From here, the event is a channel wait fd */
3091 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
3092 health_code_update();
3093
d8ef542d
MD
3094 revents = LTTNG_POLL_GETEV(&events, i);
3095 pollfd = LTTNG_POLL_GETFD(&events, i);
3096
d8ef542d 3097 if (pollfd == ctx->consumer_channel_pipe[0]) {
03e43155 3098 if (revents & LPOLLIN) {
d8ef542d 3099 enum consumer_channel_action action;
a0cbdd2e 3100 uint64_t key;
d8ef542d 3101
a0cbdd2e 3102 ret = read_channel_pipe(ctx, &chan, &key, &action);
d8ef542d 3103 if (ret <= 0) {
03e43155
MD
3104 if (ret < 0) {
3105 ERR("Error reading channel pipe");
3106 }
3107 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
d8ef542d
MD
3108 continue;
3109 }
3110
3111 switch (action) {
3112 case CONSUMER_CHANNEL_ADD:
3113 DBG("Adding channel %d to poll set",
3114 chan->wait_fd);
3115
3116 lttng_ht_node_init_u64(&chan->wait_fd_node,
3117 chan->wait_fd);
c7260a81 3118 rcu_read_lock();
d8ef542d
MD
3119 lttng_ht_add_unique_u64(channel_ht,
3120 &chan->wait_fd_node);
c7260a81 3121 rcu_read_unlock();
d8ef542d
MD
3122 /* Add channel to the global poll events list */
3123 lttng_poll_add(&events, chan->wait_fd,
03e43155 3124 LPOLLERR | LPOLLHUP);
d8ef542d 3125 break;
a0cbdd2e
MD
3126 case CONSUMER_CHANNEL_DEL:
3127 {
b4a650f3
DG
3128 /*
3129 * This command should never be called if the channel
3130 * has streams monitored by either the data or metadata
3131 * thread. The consumer only notify this thread with a
3132 * channel del. command if it receives a destroy
3133 * channel command from the session daemon that send it
3134 * if a command prior to the GET_CHANNEL failed.
3135 */
3136
c7260a81 3137 rcu_read_lock();
a0cbdd2e
MD
3138 chan = consumer_find_channel(key);
3139 if (!chan) {
c7260a81 3140 rcu_read_unlock();
a0cbdd2e
MD
3141 ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
3142 break;
3143 }
3144 lttng_poll_del(&events, chan->wait_fd);
f623cc0b 3145 iter.iter.node = &chan->wait_fd_node.node;
a0cbdd2e
MD
3146 ret = lttng_ht_del(channel_ht, &iter);
3147 assert(ret == 0);
a0cbdd2e 3148
f2a444f1
DG
3149 switch (consumer_data.type) {
3150 case LTTNG_CONSUMER_KERNEL:
3151 break;
3152 case LTTNG_CONSUMER32_UST:
3153 case LTTNG_CONSUMER64_UST:
212d67a2
DG
3154 health_code_update();
3155 /* Destroy streams that might have been left in the stream list. */
3156 clean_channel_stream_list(chan);
f2a444f1
DG
3157 break;
3158 default:
3159 ERR("Unknown consumer_data type");
3160 assert(0);
3161 }
3162
a0cbdd2e
MD
3163 /*
3164 * Release our own refcount. Force channel deletion even if
3165 * streams were not initialized.
3166 */
3167 if (!uatomic_sub_return(&chan->refcount, 1)) {
3168 consumer_del_channel(chan);
3169 }
c7260a81 3170 rcu_read_unlock();
a0cbdd2e
MD
3171 goto restart;
3172 }
d8ef542d
MD
3173 case CONSUMER_CHANNEL_QUIT:
3174 /*
3175 * Remove the pipe from the poll set and continue the loop
3176 * since their might be data to consume.
3177 */
3178 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3179 continue;
3180 default:
3181 ERR("Unknown action");
3182 break;
3183 }
03e43155
MD
3184 } else if (revents & (LPOLLERR | LPOLLHUP)) {
3185 DBG("Channel thread pipe hung up");
3186 /*
3187 * Remove the pipe from the poll set and continue the loop
3188 * since their might be data to consume.
3189 */
3190 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3191 continue;
3192 } else {
3193 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3194 goto end;
d8ef542d
MD
3195 }
3196
3197 /* Handle other stream */
3198 continue;
3199 }
3200
3201 rcu_read_lock();
3202 {
3203 uint64_t tmp_id = (uint64_t) pollfd;
3204
3205 lttng_ht_lookup(channel_ht, &tmp_id, &iter);
3206 }
3207 node = lttng_ht_iter_get_node_u64(&iter);
3208 assert(node);
3209
3210 chan = caa_container_of(node, struct lttng_consumer_channel,
3211 wait_fd_node);
3212
3213 /* Check for error event */
3214 if (revents & (LPOLLERR | LPOLLHUP)) {
3215 DBG("Channel fd %d is hup|err.", pollfd);
3216
3217 lttng_poll_del(&events, chan->wait_fd);
3218 ret = lttng_ht_del(channel_ht, &iter);
3219 assert(ret == 0);
b4a650f3
DG
3220
3221 /*
3222 * This will close the wait fd for each stream associated to
3223 * this channel AND monitored by the data/metadata thread thus
3224 * will be clean by the right thread.
3225 */
d8ef542d 3226 consumer_close_channel_streams(chan);
f2ad556d
MD
3227
3228 /* Release our own refcount */
3229 if (!uatomic_sub_return(&chan->refcount, 1)
3230 && !uatomic_read(&chan->nb_init_stream_left)) {
3231 consumer_del_channel(chan);
3232 }
03e43155
MD
3233 } else {
3234 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3235 rcu_read_unlock();
3236 goto end;
d8ef542d
MD
3237 }
3238
3239 /* Release RCU lock for the channel looked up */
3240 rcu_read_unlock();
3241 }
3242 }
3243
1fc79fb4
MD
3244 /* All is OK */
3245 err = 0;
d8ef542d
MD
3246end:
3247<