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