Cleanup: relayd id is never used by write_relayd_metadata_id()
[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,
239f61af 1478 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) {
239f61af 1613 ret = write_relayd_metadata_id(outfd, stream, padding);
994ab360 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 }
239f61af 1802 ret = write_relayd_metadata_id(splice_pipe[1], stream,
1d4dfdef
DG
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
15055ce5
JD
1974/*
1975 * Sample the snapshot positions for a specific fd
1976 *
1977 * Returns 0 on success, < 0 on error
1978 */
1979int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
1980{
1981 switch (consumer_data.type) {
1982 case LTTNG_CONSUMER_KERNEL:
1983 return lttng_kconsumer_sample_snapshot_positions(stream);
1984 case LTTNG_CONSUMER32_UST:
1985 case LTTNG_CONSUMER64_UST:
1986 return lttng_ustconsumer_sample_snapshot_positions(stream);
1987 default:
1988 ERR("Unknown consumer_data type");
1989 assert(0);
1990 return -ENOSYS;
1991 }
1992}
3bd1e081
MD
1993/*
1994 * Take a snapshot for a specific fd
1995 *
1996 * Returns 0 on success, < 0 on error
1997 */
ffe60014 1998int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081
MD
1999{
2000 switch (consumer_data.type) {
2001 case LTTNG_CONSUMER_KERNEL:
ffe60014 2002 return lttng_kconsumer_take_snapshot(stream);
7753dea8
MD
2003 case LTTNG_CONSUMER32_UST:
2004 case LTTNG_CONSUMER64_UST:
ffe60014 2005 return lttng_ustconsumer_take_snapshot(stream);
3bd1e081
MD
2006 default:
2007 ERR("Unknown consumer_data type");
2008 assert(0);
2009 return -ENOSYS;
2010 }
3bd1e081
MD
2011}
2012
2013/*
2014 * Get the produced position
2015 *
2016 * Returns 0 on success, < 0 on error
2017 */
ffe60014 2018int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
3bd1e081
MD
2019 unsigned long *pos)
2020{
2021 switch (consumer_data.type) {
2022 case LTTNG_CONSUMER_KERNEL:
ffe60014 2023 return lttng_kconsumer_get_produced_snapshot(stream, pos);
7753dea8
MD
2024 case LTTNG_CONSUMER32_UST:
2025 case LTTNG_CONSUMER64_UST:
ffe60014 2026 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
3bd1e081
MD
2027 default:
2028 ERR("Unknown consumer_data type");
2029 assert(0);
2030 return -ENOSYS;
2031 }
2032}
2033
15055ce5
JD
2034/*
2035 * Get the consumed position (free-running counter position in bytes).
2036 *
2037 * Returns 0 on success, < 0 on error
2038 */
2039int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
2040 unsigned long *pos)
2041{
2042 switch (consumer_data.type) {
2043 case LTTNG_CONSUMER_KERNEL:
2044 return lttng_kconsumer_get_consumed_snapshot(stream, pos);
2045 case LTTNG_CONSUMER32_UST:
2046 case LTTNG_CONSUMER64_UST:
2047 return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
2048 default:
2049 ERR("Unknown consumer_data type");
2050 assert(0);
2051 return -ENOSYS;
2052 }
2053}
2054
3bd1e081
MD
2055int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
2056 int sock, struct pollfd *consumer_sockpoll)
2057{
2058 switch (consumer_data.type) {
2059 case LTTNG_CONSUMER_KERNEL:
2060 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
2061 case LTTNG_CONSUMER32_UST:
2062 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
2063 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
2064 default:
2065 ERR("Unknown consumer_data type");
2066 assert(0);
2067 return -ENOSYS;
2068 }
2069}
2070
6d574024 2071void lttng_consumer_close_all_metadata(void)
d88aee68
DG
2072{
2073 switch (consumer_data.type) {
2074 case LTTNG_CONSUMER_KERNEL:
2075 /*
2076 * The Kernel consumer has a different metadata scheme so we don't
2077 * close anything because the stream will be closed by the session
2078 * daemon.
2079 */
2080 break;
2081 case LTTNG_CONSUMER32_UST:
2082 case LTTNG_CONSUMER64_UST:
2083 /*
2084 * Close all metadata streams. The metadata hash table is passed and
2085 * this call iterates over it by closing all wakeup fd. This is safe
2086 * because at this point we are sure that the metadata producer is
2087 * either dead or blocked.
2088 */
6d574024 2089 lttng_ustconsumer_close_all_metadata(metadata_ht);
d88aee68
DG
2090 break;
2091 default:
2092 ERR("Unknown consumer_data type");
2093 assert(0);
2094 }
2095}
2096
fb3a43a9
DG
2097/*
2098 * Clean up a metadata stream and free its memory.
2099 */
e316aad5
DG
2100void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
2101 struct lttng_ht *ht)
fb3a43a9 2102{
e316aad5 2103 struct lttng_consumer_channel *free_chan = NULL;
fb3a43a9
DG
2104
2105 assert(stream);
2106 /*
2107 * This call should NEVER receive regular stream. It must always be
2108 * metadata stream and this is crucial for data structure synchronization.
2109 */
2110 assert(stream->metadata_flag);
2111
e316aad5
DG
2112 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
2113
74251bb8 2114 pthread_mutex_lock(&consumer_data.lock);
fb549e7a 2115 pthread_mutex_lock(&stream->chan->lock);
3dad2c0f 2116 pthread_mutex_lock(&stream->lock);
081424af
JG
2117 if (stream->chan->metadata_cache) {
2118 /* Only applicable to userspace consumers. */
2119 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
2120 }
8994307f 2121
6d574024
DG
2122 /* Remove any reference to that stream. */
2123 consumer_stream_delete(stream, ht);
ca22feea 2124
6d574024
DG
2125 /* Close down everything including the relayd if one. */
2126 consumer_stream_close(stream);
2127 /* Destroy tracer buffers of the stream. */
2128 consumer_stream_destroy_buffers(stream);
fb3a43a9
DG
2129
2130 /* Atomically decrement channel refcount since other threads can use it. */
f2ad556d 2131 if (!uatomic_sub_return(&stream->chan->refcount, 1)
ffe60014 2132 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
c30aaa51 2133 /* Go for channel deletion! */
e316aad5 2134 free_chan = stream->chan;
fb3a43a9
DG
2135 }
2136
73811ecc
DG
2137 /*
2138 * Nullify the stream reference so it is not used after deletion. The
6d574024
DG
2139 * channel lock MUST be acquired before being able to check for a NULL
2140 * pointer value.
73811ecc
DG
2141 */
2142 stream->chan->metadata_stream = NULL;
2143
081424af
JG
2144 if (stream->chan->metadata_cache) {
2145 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2146 }
3dad2c0f 2147 pthread_mutex_unlock(&stream->lock);
fb549e7a 2148 pthread_mutex_unlock(&stream->chan->lock);
74251bb8 2149 pthread_mutex_unlock(&consumer_data.lock);
e316aad5
DG
2150
2151 if (free_chan) {
2152 consumer_del_channel(free_chan);
2153 }
2154
6d574024 2155 consumer_stream_free(stream);
fb3a43a9
DG
2156}
2157
2158/*
2159 * Action done with the metadata stream when adding it to the consumer internal
2160 * data structures to handle it.
2161 */
66d583dc 2162void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
fb3a43a9 2163{
5ab66908 2164 struct lttng_ht *ht = metadata_ht;
76082088 2165 struct lttng_ht_iter iter;
d88aee68 2166 struct lttng_ht_node_u64 *node;
fb3a43a9 2167
e316aad5
DG
2168 assert(stream);
2169 assert(ht);
2170
d88aee68 2171 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
e316aad5
DG
2172
2173 pthread_mutex_lock(&consumer_data.lock);
a9838785 2174 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 2175 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 2176 pthread_mutex_lock(&stream->lock);
e316aad5 2177
e316aad5
DG
2178 /*
2179 * From here, refcounts are updated so be _careful_ when returning an error
2180 * after this point.
2181 */
2182
fb3a43a9 2183 rcu_read_lock();
76082088
DG
2184
2185 /*
2186 * Lookup the stream just to make sure it does not exist in our internal
2187 * state. This should NEVER happen.
2188 */
d88aee68
DG
2189 lttng_ht_lookup(ht, &stream->key, &iter);
2190 node = lttng_ht_iter_get_node_u64(&iter);
76082088
DG
2191 assert(!node);
2192
e316aad5 2193 /*
ffe60014
DG
2194 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2195 * in terms of destroying the associated channel, because the action that
e316aad5
DG
2196 * causes the count to become 0 also causes a stream to be added. The
2197 * channel deletion will thus be triggered by the following removal of this
2198 * stream.
2199 */
ffe60014 2200 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
2201 /* Increment refcount before decrementing nb_init_stream_left */
2202 cmm_smp_wmb();
ffe60014 2203 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
2204 }
2205
d88aee68 2206 lttng_ht_add_unique_u64(ht, &stream->node);
ca22feea 2207
446156b4 2208 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
d8ef542d
MD
2209 &stream->node_channel_id);
2210
ca22feea
DG
2211 /*
2212 * Add stream to the stream_list_ht of the consumer data. No need to steal
2213 * the key since the HT does not use it and we allow to add redundant keys
2214 * into this table.
2215 */
d88aee68 2216 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 2217
fb3a43a9 2218 rcu_read_unlock();
e316aad5 2219
2e818a6a 2220 pthread_mutex_unlock(&stream->lock);
a9838785 2221 pthread_mutex_unlock(&stream->chan->lock);
ec6ea7d0 2222 pthread_mutex_unlock(&stream->chan->timer_lock);
e316aad5 2223 pthread_mutex_unlock(&consumer_data.lock);
fb3a43a9
DG
2224}
2225
8994307f
DG
2226/*
2227 * Delete data stream that are flagged for deletion (endpoint_status).
2228 */
2229static void validate_endpoint_status_data_stream(void)
2230{
2231 struct lttng_ht_iter iter;
2232 struct lttng_consumer_stream *stream;
2233
2234 DBG("Consumer delete flagged data stream");
2235
2236 rcu_read_lock();
2237 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2238 /* Validate delete flag of the stream */
79d4ffb7 2239 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2240 continue;
2241 }
2242 /* Delete it right now */
2243 consumer_del_stream(stream, data_ht);
2244 }
2245 rcu_read_unlock();
2246}
2247
2248/*
2249 * Delete metadata stream that are flagged for deletion (endpoint_status).
2250 */
2251static void validate_endpoint_status_metadata_stream(
2252 struct lttng_poll_event *pollset)
2253{
2254 struct lttng_ht_iter iter;
2255 struct lttng_consumer_stream *stream;
2256
2257 DBG("Consumer delete flagged metadata stream");
2258
2259 assert(pollset);
2260
2261 rcu_read_lock();
2262 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2263 /* Validate delete flag of the stream */
79d4ffb7 2264 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2265 continue;
2266 }
2267 /*
2268 * Remove from pollset so the metadata thread can continue without
2269 * blocking on a deleted stream.
2270 */
2271 lttng_poll_del(pollset, stream->wait_fd);
2272
2273 /* Delete it right now */
2274 consumer_del_metadata_stream(stream, metadata_ht);
2275 }
2276 rcu_read_unlock();
2277}
2278
d73bf3d7
JD
2279static
2280int rotate_notify_sessiond(struct lttng_consumer_local_data *ctx,
2281 uint64_t key)
2282{
2283 ssize_t ret;
2284
2285 do {
2286 ret = write(ctx->channel_rotate_pipe, &key, sizeof(key));
2287 } while (ret == -1 && errno == EINTR);
2288 if (ret == -1) {
2289 PERROR("Failed to write to the channel rotation pipe");
2290 } else {
2291 DBG("Sent channel rotation notification for channel key %"
2292 PRIu64, key);
2293 ret = 0;
2294 }
2295
2296 return (int) ret;
2297}
2298
2299/*
2300 * Perform operations that need to be done after a stream has
2301 * rotated and released the stream lock.
2302 *
2303 * Multiple rotations cannot occur simultaneously, so we know the state of the
2304 * "rotated" stream flag cannot change.
2305 *
2306 * This MUST be called WITHOUT the stream lock held.
2307 */
2308static
2309int consumer_post_rotation(struct lttng_consumer_stream *stream,
2310 struct lttng_consumer_local_data *ctx)
2311{
2312 int ret = 0;
2313
2314 pthread_mutex_lock(&stream->chan->lock);
2315
2316 switch (consumer_data.type) {
2317 case LTTNG_CONSUMER_KERNEL:
2318 break;
2319 case LTTNG_CONSUMER32_UST:
2320 case LTTNG_CONSUMER64_UST:
2321 /*
2322 * The ust_metadata_pushed counter has been reset to 0, so now
2323 * we can wakeup the metadata thread so it dumps the metadata
2324 * cache to the new file.
2325 */
2326 if (stream->metadata_flag) {
2327 consumer_metadata_wakeup_pipe(stream->chan);
2328 }
2329 break;
2330 default:
2331 ERR("Unknown consumer_data type");
2332 abort();
2333 }
2334
2335 if (--stream->chan->nr_stream_rotate_pending == 0) {
2336 DBG("Rotation of channel \"%s\" completed, notifying the session daemon",
2337 stream->chan->name);
2338 ret = rotate_notify_sessiond(ctx, stream->chan->key);
2339 }
2340 assert(stream->chan->nr_stream_rotate_pending >= 0);
2341 pthread_mutex_unlock(&stream->chan->lock);
2342
2343 return ret;
2344}
2345
fb3a43a9
DG
2346/*
2347 * Thread polls on metadata file descriptor and write them on disk or on the
2348 * network.
2349 */
7d980def 2350void *consumer_thread_metadata_poll(void *data)
fb3a43a9 2351{
1fc79fb4 2352 int ret, i, pollfd, err = -1;
fb3a43a9 2353 uint32_t revents, nb_fd;
e316aad5 2354 struct lttng_consumer_stream *stream = NULL;
fb3a43a9 2355 struct lttng_ht_iter iter;
d88aee68 2356 struct lttng_ht_node_u64 *node;
fb3a43a9
DG
2357 struct lttng_poll_event events;
2358 struct lttng_consumer_local_data *ctx = data;
2359 ssize_t len;
2360
2361 rcu_register_thread();
2362
1fc79fb4
MD
2363 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA);
2364
2d57de81
MD
2365 if (testpoint(consumerd_thread_metadata)) {
2366 goto error_testpoint;
2367 }
2368
9ce5646a
MD
2369 health_code_update();
2370
fb3a43a9
DG
2371 DBG("Thread metadata poll started");
2372
fb3a43a9
DG
2373 /* Size is set to 1 for the consumer_metadata pipe */
2374 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2375 if (ret < 0) {
2376 ERR("Poll set creation failed");
d8ef542d 2377 goto end_poll;
fb3a43a9
DG
2378 }
2379
13886d2d
DG
2380 ret = lttng_poll_add(&events,
2381 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
fb3a43a9
DG
2382 if (ret < 0) {
2383 goto end;
2384 }
2385
2386 /* Main loop */
2387 DBG("Metadata main loop started");
2388
2389 while (1) {
fb3a43a9 2390restart:
7fa2082e 2391 health_code_update();
9ce5646a 2392 health_poll_entry();
7fa2082e 2393 DBG("Metadata poll wait");
fb3a43a9 2394 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
2395 DBG("Metadata poll return from wait with %d fd(s)",
2396 LTTNG_POLL_GETNB(&events));
9ce5646a 2397 health_poll_exit();
40063ead 2398 DBG("Metadata event caught in thread");
fb3a43a9
DG
2399 if (ret < 0) {
2400 if (errno == EINTR) {
40063ead 2401 ERR("Poll EINTR caught");
fb3a43a9
DG
2402 goto restart;
2403 }
d9607cd7
MD
2404 if (LTTNG_POLL_GETNB(&events) == 0) {
2405 err = 0; /* All is OK */
2406 }
2407 goto end;
fb3a43a9
DG
2408 }
2409
0d9c5d77
DG
2410 nb_fd = ret;
2411
e316aad5 2412 /* From here, the event is a metadata wait fd */
fb3a43a9 2413 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2414 health_code_update();
2415
fb3a43a9
DG
2416 revents = LTTNG_POLL_GETEV(&events, i);
2417 pollfd = LTTNG_POLL_GETFD(&events, i);
2418
fd20dac9
MD
2419 if (!revents) {
2420 /* No activity for this FD (poll implementation). */
2421 continue;
2422 }
2423
13886d2d 2424 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
03e43155 2425 if (revents & LPOLLIN) {
13886d2d
DG
2426 ssize_t pipe_len;
2427
2428 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2429 &stream, sizeof(stream));
6cd525e8 2430 if (pipe_len < sizeof(stream)) {
03e43155
MD
2431 if (pipe_len < 0) {
2432 PERROR("read metadata stream");
2433 }
fb3a43a9 2434 /*
03e43155
MD
2435 * Remove the pipe from the poll set and continue the loop
2436 * since their might be data to consume.
fb3a43a9 2437 */
03e43155
MD
2438 lttng_poll_del(&events,
2439 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2440 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
fb3a43a9
DG
2441 continue;
2442 }
2443
8994307f
DG
2444 /* A NULL stream means that the state has changed. */
2445 if (stream == NULL) {
2446 /* Check for deleted streams. */
2447 validate_endpoint_status_metadata_stream(&events);
3714380f 2448 goto restart;
8994307f
DG
2449 }
2450
fb3a43a9
DG
2451 DBG("Adding metadata stream %d to poll set",
2452 stream->wait_fd);
2453
fb3a43a9
DG
2454 /* Add metadata stream to the global poll events list */
2455 lttng_poll_add(&events, stream->wait_fd,
6d574024 2456 LPOLLIN | LPOLLPRI | LPOLLHUP);
03e43155
MD
2457 } else if (revents & (LPOLLERR | LPOLLHUP)) {
2458 DBG("Metadata thread pipe hung up");
2459 /*
2460 * Remove the pipe from the poll set and continue the loop
2461 * since their might be data to consume.
2462 */
2463 lttng_poll_del(&events,
2464 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2465 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2466 continue;
2467 } else {
2468 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2469 goto end;
fb3a43a9
DG
2470 }
2471
e316aad5 2472 /* Handle other stream */
fb3a43a9
DG
2473 continue;
2474 }
2475
d09e1200 2476 rcu_read_lock();
d88aee68
DG
2477 {
2478 uint64_t tmp_id = (uint64_t) pollfd;
2479
2480 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2481 }
2482 node = lttng_ht_iter_get_node_u64(&iter);
e316aad5 2483 assert(node);
fb3a43a9
DG
2484
2485 stream = caa_container_of(node, struct lttng_consumer_stream,
58b1f425 2486 node);
fb3a43a9 2487
03e43155
MD
2488 if (revents & (LPOLLIN | LPOLLPRI)) {
2489 /* Get the data out of the metadata file descriptor */
2490 DBG("Metadata available on fd %d", pollfd);
2491 assert(stream->wait_fd == pollfd);
2492
2493 do {
2494 health_code_update();
2495
2496 len = ctx->on_buffer_ready(stream, ctx);
2497 /*
2498 * We don't check the return value here since if we get
83f4233d 2499 * a negative len, it means an error occurred thus we
03e43155
MD
2500 * simply remove it from the poll set and free the
2501 * stream.
2502 */
2503 } while (len > 0);
2504
2505 /* It's ok to have an unavailable sub-buffer */
2506 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2507 /* Clean up stream from consumer and free it. */
2508 lttng_poll_del(&events, stream->wait_fd);
2509 consumer_del_metadata_stream(stream, metadata_ht);
2510 }
2511 } else if (revents & (LPOLLERR | LPOLLHUP)) {
e316aad5 2512 DBG("Metadata fd %d is hup|err.", pollfd);
fb3a43a9
DG
2513 if (!stream->hangup_flush_done
2514 && (consumer_data.type == LTTNG_CONSUMER32_UST
2515 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2516 DBG("Attempting to flush and consume the UST buffers");
2517 lttng_ustconsumer_on_stream_hangup(stream);
2518
2519 /* We just flushed the stream now read it. */
4bb94b75 2520 do {
9ce5646a
MD
2521 health_code_update();
2522
4bb94b75
DG
2523 len = ctx->on_buffer_ready(stream, ctx);
2524 /*
2525 * We don't check the return value here since if we get
83f4233d 2526 * a negative len, it means an error occurred thus we
4bb94b75
DG
2527 * simply remove it from the poll set and free the
2528 * stream.
2529 */
2530 } while (len > 0);
fb3a43a9
DG
2531 }
2532
fb3a43a9 2533 lttng_poll_del(&events, stream->wait_fd);
e316aad5
DG
2534 /*
2535 * This call update the channel states, closes file descriptors
2536 * and securely free the stream.
2537 */
2538 consumer_del_metadata_stream(stream, metadata_ht);
03e43155
MD
2539 } else {
2540 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
6f2f1a70 2541 rcu_read_unlock();
03e43155 2542 goto end;
fb3a43a9 2543 }
e316aad5 2544 /* Release RCU lock for the stream looked up */
d09e1200 2545 rcu_read_unlock();
fb3a43a9
DG
2546 }
2547 }
2548
1fc79fb4
MD
2549 /* All is OK */
2550 err = 0;
fb3a43a9
DG
2551end:
2552 DBG("Metadata poll thread exiting");
fb3a43a9 2553
d8ef542d
MD
2554 lttng_poll_clean(&events);
2555end_poll:
2d57de81 2556error_testpoint:
1fc79fb4
MD
2557 if (err) {
2558 health_error();
2559 ERR("Health error occurred in %s", __func__);
2560 }
2561 health_unregister(health_consumerd);
fb3a43a9
DG
2562 rcu_unregister_thread();
2563 return NULL;
2564}
2565
3bd1e081 2566/*
e4421fec 2567 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
2568 * it to tracefile if necessary.
2569 */
7d980def 2570void *consumer_thread_data_poll(void *data)
3bd1e081 2571{
1fc79fb4 2572 int num_rdy, num_hup, high_prio, ret, i, err = -1;
3bd1e081
MD
2573 struct pollfd *pollfd = NULL;
2574 /* local view of the streams */
c869f647 2575 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
3bd1e081 2576 /* local view of consumer_data.fds_count */
261de637 2577 int nb_fd = 0, nb_pipes_fd;
9a2fcf78
JD
2578 /* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
2579 int nb_inactive_fd = 0;
3bd1e081 2580 struct lttng_consumer_local_data *ctx = data;
00e2e675 2581 ssize_t len;
3bd1e081 2582
e7b994a3
DG
2583 rcu_register_thread();
2584
1fc79fb4
MD
2585 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA);
2586
2d57de81
MD
2587 if (testpoint(consumerd_thread_data)) {
2588 goto error_testpoint;
2589 }
2590
9ce5646a
MD
2591 health_code_update();
2592
4df6c8cb
MD
2593 local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
2594 if (local_stream == NULL) {
2595 PERROR("local_stream malloc");
2596 goto end;
2597 }
3bd1e081
MD
2598
2599 while (1) {
9ce5646a
MD
2600 health_code_update();
2601
3bd1e081
MD
2602 high_prio = 0;
2603 num_hup = 0;
2604
2605 /*
e4421fec 2606 * the fds set has been updated, we need to update our
3bd1e081
MD
2607 * local array as well
2608 */
2609 pthread_mutex_lock(&consumer_data.lock);
2610 if (consumer_data.need_update) {
0e428499
DG
2611 free(pollfd);
2612 pollfd = NULL;
2613
2614 free(local_stream);
2615 local_stream = NULL;
3bd1e081 2616
02b3d176 2617 /*
261de637
JD
2618 * Allocate for all fds + 2:
2619 * +1 for the consumer_data_pipe
2620 * +1 for wake up pipe
02b3d176 2621 */
261de637
JD
2622 nb_pipes_fd = 2;
2623 pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
3bd1e081 2624 if (pollfd == NULL) {
7a57cf92 2625 PERROR("pollfd malloc");
3bd1e081
MD
2626 pthread_mutex_unlock(&consumer_data.lock);
2627 goto end;
2628 }
2629
261de637 2630 local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
747f8642 2631 sizeof(struct lttng_consumer_stream *));
3bd1e081 2632 if (local_stream == NULL) {
7a57cf92 2633 PERROR("local_stream malloc");
3bd1e081
MD
2634 pthread_mutex_unlock(&consumer_data.lock);
2635 goto end;
2636 }
ffe60014 2637 ret = update_poll_array(ctx, &pollfd, local_stream,
9a2fcf78 2638 data_ht, &nb_inactive_fd);
3bd1e081
MD
2639 if (ret < 0) {
2640 ERR("Error in allocating pollfd or local_outfds");
f73fabfd 2641 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2642 pthread_mutex_unlock(&consumer_data.lock);
2643 goto end;
2644 }
2645 nb_fd = ret;
2646 consumer_data.need_update = 0;
2647 }
2648 pthread_mutex_unlock(&consumer_data.lock);
2649
4078b776 2650 /* No FDs and consumer_quit, consumer_cleanup the thread */
9a2fcf78
JD
2651 if (nb_fd == 0 && nb_inactive_fd == 0 &&
2652 CMM_LOAD_SHARED(consumer_quit) == 1) {
1fc79fb4 2653 err = 0; /* All is OK */
4078b776
MD
2654 goto end;
2655 }
3bd1e081 2656 /* poll on the array of fds */
88f2b785 2657 restart:
261de637 2658 DBG("polling on %d fd", nb_fd + nb_pipes_fd);
cf0bcb51
JG
2659 if (testpoint(consumerd_thread_data_poll)) {
2660 goto end;
2661 }
9ce5646a 2662 health_poll_entry();
261de637 2663 num_rdy = poll(pollfd, nb_fd + nb_pipes_fd, -1);
9ce5646a 2664 health_poll_exit();
3bd1e081
MD
2665 DBG("poll num_rdy : %d", num_rdy);
2666 if (num_rdy == -1) {
88f2b785
MD
2667 /*
2668 * Restart interrupted system call.
2669 */
2670 if (errno == EINTR) {
2671 goto restart;
2672 }
7a57cf92 2673 PERROR("Poll error");
f73fabfd 2674 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2675 goto end;
2676 } else if (num_rdy == 0) {
2677 DBG("Polling thread timed out");
2678 goto end;
2679 }
2680
80957876
JG
2681 if (caa_unlikely(data_consumption_paused)) {
2682 DBG("Data consumption paused, sleeping...");
2683 sleep(1);
2684 goto restart;
2685 }
2686
3bd1e081 2687 /*
50f8ae69 2688 * If the consumer_data_pipe triggered poll go directly to the
00e2e675
DG
2689 * beginning of the loop to update the array. We want to prioritize
2690 * array update over low-priority reads.
3bd1e081 2691 */
509bb1cf 2692 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
ab30f567 2693 ssize_t pipe_readlen;
04fdd819 2694
50f8ae69 2695 DBG("consumer_data_pipe wake up");
acdb9057
DG
2696 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2697 &new_stream, sizeof(new_stream));
6cd525e8
MD
2698 if (pipe_readlen < sizeof(new_stream)) {
2699 PERROR("Consumer data pipe");
23f5f35d
DG
2700 /* Continue so we can at least handle the current stream(s). */
2701 continue;
2702 }
c869f647
DG
2703
2704 /*
2705 * If the stream is NULL, just ignore it. It's also possible that
2706 * the sessiond poll thread changed the consumer_quit state and is
2707 * waking us up to test it.
2708 */
2709 if (new_stream == NULL) {
8994307f 2710 validate_endpoint_status_data_stream();
c869f647
DG
2711 continue;
2712 }
2713
c869f647 2714 /* Continue to update the local streams and handle prio ones */
3bd1e081
MD
2715 continue;
2716 }
2717
02b3d176
DG
2718 /* Handle wakeup pipe. */
2719 if (pollfd[nb_fd + 1].revents & (POLLIN | POLLPRI)) {
2720 char dummy;
2721 ssize_t pipe_readlen;
2722
2723 pipe_readlen = lttng_pipe_read(ctx->consumer_wakeup_pipe, &dummy,
2724 sizeof(dummy));
2725 if (pipe_readlen < 0) {
2726 PERROR("Consumer data wakeup pipe");
2727 }
2728 /* We've been awakened to handle stream(s). */
2729 ctx->has_wakeup = 0;
2730 }
2731
3bd1e081
MD
2732 /* Take care of high priority channels first. */
2733 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2734 health_code_update();
2735
9617607b
DG
2736 if (local_stream[i] == NULL) {
2737 continue;
2738 }
fb3a43a9 2739 if (pollfd[i].revents & POLLPRI) {
d41f73b7
MD
2740 DBG("Urgent read on fd %d", pollfd[i].fd);
2741 high_prio = 1;
4078b776 2742 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 2743 /* it's ok to have an unavailable sub-buffer */
b64403e3 2744 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2745 /* Clean the stream and free it. */
2746 consumer_del_stream(local_stream[i], data_ht);
9617607b 2747 local_stream[i] = NULL;
4078b776
MD
2748 } else if (len > 0) {
2749 local_stream[i]->data_read = 1;
d41f73b7 2750 }
3bd1e081
MD
2751 }
2752 }
2753
4078b776
MD
2754 /*
2755 * If we read high prio channel in this loop, try again
2756 * for more high prio data.
2757 */
2758 if (high_prio) {
3bd1e081
MD
2759 continue;
2760 }
2761
2762 /* Take care of low priority channels. */
4078b776 2763 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2764 health_code_update();
2765
9617607b
DG
2766 if (local_stream[i] == NULL) {
2767 continue;
2768 }
4078b776 2769 if ((pollfd[i].revents & POLLIN) ||
02b3d176
DG
2770 local_stream[i]->hangup_flush_done ||
2771 local_stream[i]->has_data) {
4078b776
MD
2772 DBG("Normal read on fd %d", pollfd[i].fd);
2773 len = ctx->on_buffer_ready(local_stream[i], ctx);
2774 /* it's ok to have an unavailable sub-buffer */
b64403e3 2775 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2776 /* Clean the stream and free it. */
2777 consumer_del_stream(local_stream[i], data_ht);
9617607b 2778 local_stream[i] = NULL;
4078b776
MD
2779 } else if (len > 0) {
2780 local_stream[i]->data_read = 1;
2781 }
2782 }
2783 }
2784
2785 /* Handle hangup and errors */
2786 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2787 health_code_update();
2788
9617607b
DG
2789 if (local_stream[i] == NULL) {
2790 continue;
2791 }
4078b776
MD
2792 if (!local_stream[i]->hangup_flush_done
2793 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2794 && (consumer_data.type == LTTNG_CONSUMER32_UST
2795 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2796 DBG("fd %d is hup|err|nval. Attempting flush and read.",
9617607b 2797 pollfd[i].fd);
4078b776
MD
2798 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2799 /* Attempt read again, for the data we just flushed. */
2800 local_stream[i]->data_read = 1;
2801 }
2802 /*
2803 * If the poll flag is HUP/ERR/NVAL and we have
2804 * read no data in this pass, we can remove the
2805 * stream from its hash table.
2806 */
2807 if ((pollfd[i].revents & POLLHUP)) {
2808 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2809 if (!local_stream[i]->data_read) {
43c34bc3 2810 consumer_del_stream(local_stream[i], data_ht);
9617607b 2811 local_stream[i] = NULL;
4078b776
MD
2812 num_hup++;
2813 }
2814 } else if (pollfd[i].revents & POLLERR) {
2815 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2816 if (!local_stream[i]->data_read) {
43c34bc3 2817 consumer_del_stream(local_stream[i], data_ht);
9617607b 2818 local_stream[i] = NULL;
4078b776
MD
2819 num_hup++;
2820 }
2821 } else if (pollfd[i].revents & POLLNVAL) {
2822 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2823 if (!local_stream[i]->data_read) {
43c34bc3 2824 consumer_del_stream(local_stream[i], data_ht);
9617607b 2825 local_stream[i] = NULL;
4078b776 2826 num_hup++;
3bd1e081
MD
2827 }
2828 }
9617607b
DG
2829 if (local_stream[i] != NULL) {
2830 local_stream[i]->data_read = 0;
2831 }
3bd1e081
MD
2832 }
2833 }
1fc79fb4
MD
2834 /* All is OK */
2835 err = 0;
3bd1e081
MD
2836end:
2837 DBG("polling thread exiting");
0e428499
DG
2838 free(pollfd);
2839 free(local_stream);
fb3a43a9
DG
2840
2841 /*
2842 * Close the write side of the pipe so epoll_wait() in
7d980def
DG
2843 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2844 * read side of the pipe. If we close them both, epoll_wait strangely does
2845 * not return and could create a endless wait period if the pipe is the
2846 * only tracked fd in the poll set. The thread will take care of closing
2847 * the read side.
fb3a43a9 2848 */
13886d2d 2849 (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
fb3a43a9 2850
2d57de81 2851error_testpoint:
1fc79fb4
MD
2852 if (err) {
2853 health_error();
2854 ERR("Health error occurred in %s", __func__);
2855 }
2856 health_unregister(health_consumerd);
2857
e7b994a3 2858 rcu_unregister_thread();
3bd1e081
MD
2859 return NULL;
2860}
2861
d8ef542d
MD
2862/*
2863 * Close wake-up end of each stream belonging to the channel. This will
2864 * allow the poll() on the stream read-side to detect when the
2865 * write-side (application) finally closes them.
2866 */
2867static
2868void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
2869{
2870 struct lttng_ht *ht;
2871 struct lttng_consumer_stream *stream;
2872 struct lttng_ht_iter iter;
2873
2874 ht = consumer_data.stream_per_chan_id_ht;
2875
2876 rcu_read_lock();
2877 cds_lfht_for_each_entry_duplicate(ht->ht,
2878 ht->hash_fct(&channel->key, lttng_ht_seed),
2879 ht->match_fct, &channel->key,
2880 &iter.iter, stream, node_channel_id.node) {
f2ad556d
MD
2881 /*
2882 * Protect against teardown with mutex.
2883 */
2884 pthread_mutex_lock(&stream->lock);
2885 if (cds_lfht_is_node_deleted(&stream->node.node)) {
2886 goto next;
2887 }
d8ef542d
MD
2888 switch (consumer_data.type) {
2889 case LTTNG_CONSUMER_KERNEL:
2890 break;
2891 case LTTNG_CONSUMER32_UST:
2892 case LTTNG_CONSUMER64_UST:
b4a650f3
DG
2893 if (stream->metadata_flag) {
2894 /* Safe and protected by the stream lock. */
2895 lttng_ustconsumer_close_metadata(stream->chan);
2896 } else {
2897 /*
2898 * Note: a mutex is taken internally within
2899 * liblttng-ust-ctl to protect timer wakeup_fd
2900 * use from concurrent close.
2901 */
2902 lttng_ustconsumer_close_stream_wakeup(stream);
2903 }
d8ef542d
MD
2904 break;
2905 default:
2906 ERR("Unknown consumer_data type");
2907 assert(0);
2908 }
f2ad556d
MD
2909 next:
2910 pthread_mutex_unlock(&stream->lock);
d8ef542d
MD
2911 }
2912 rcu_read_unlock();
2913}
2914
2915static void destroy_channel_ht(struct lttng_ht *ht)
2916{
2917 struct lttng_ht_iter iter;
2918 struct lttng_consumer_channel *channel;
2919 int ret;
2920
2921 if (ht == NULL) {
2922 return;
2923 }
2924
2925 rcu_read_lock();
2926 cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
2927 ret = lttng_ht_del(ht, &iter);
2928 assert(ret != 0);
2929 }
2930 rcu_read_unlock();
2931
2932 lttng_ht_destroy(ht);
2933}
2934
2935/*
2936 * This thread polls the channel fds to detect when they are being
2937 * closed. It closes all related streams if the channel is detected as
2938 * closed. It is currently only used as a shim layer for UST because the
2939 * consumerd needs to keep the per-stream wakeup end of pipes open for
2940 * periodical flush.
2941 */
2942void *consumer_thread_channel_poll(void *data)
2943{
1fc79fb4 2944 int ret, i, pollfd, err = -1;
d8ef542d
MD
2945 uint32_t revents, nb_fd;
2946 struct lttng_consumer_channel *chan = NULL;
2947 struct lttng_ht_iter iter;
2948 struct lttng_ht_node_u64 *node;
2949 struct lttng_poll_event events;
2950 struct lttng_consumer_local_data *ctx = data;
2951 struct lttng_ht *channel_ht;
2952
2953 rcu_register_thread();
2954
1fc79fb4
MD
2955 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL);
2956
2d57de81
MD
2957 if (testpoint(consumerd_thread_channel)) {
2958 goto error_testpoint;
2959 }
2960
9ce5646a
MD
2961 health_code_update();
2962
d8ef542d
MD
2963 channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2964 if (!channel_ht) {
2965 /* ENOMEM at this point. Better to bail out. */
2966 goto end_ht;
2967 }
2968
2969 DBG("Thread channel poll started");
2970
2971 /* Size is set to 1 for the consumer_channel pipe */
2972 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2973 if (ret < 0) {
2974 ERR("Poll set creation failed");
2975 goto end_poll;
2976 }
2977
2978 ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
2979 if (ret < 0) {
2980 goto end;
2981 }
2982
2983 /* Main loop */
2984 DBG("Channel main loop started");
2985
2986 while (1) {
d8ef542d 2987restart:
7fa2082e
MD
2988 health_code_update();
2989 DBG("Channel poll wait");
9ce5646a 2990 health_poll_entry();
d8ef542d 2991 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
2992 DBG("Channel poll return from wait with %d fd(s)",
2993 LTTNG_POLL_GETNB(&events));
9ce5646a 2994 health_poll_exit();
40063ead 2995 DBG("Channel event caught in thread");
d8ef542d
MD
2996 if (ret < 0) {
2997 if (errno == EINTR) {
40063ead 2998 ERR("Poll EINTR caught");
d8ef542d
MD
2999 goto restart;
3000 }
d9607cd7
MD
3001 if (LTTNG_POLL_GETNB(&events) == 0) {
3002 err = 0; /* All is OK */
3003 }
d8ef542d
MD
3004 goto end;
3005 }
3006
3007 nb_fd = ret;
3008
3009 /* From here, the event is a channel wait fd */
3010 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
3011 health_code_update();
3012
d8ef542d
MD
3013 revents = LTTNG_POLL_GETEV(&events, i);
3014 pollfd = LTTNG_POLL_GETFD(&events, i);
3015
d8ef542d 3016 if (!revents) {
fd20dac9 3017 /* No activity for this FD (poll implementation). */
d8ef542d
MD
3018 continue;
3019 }
fd20dac9 3020
d8ef542d 3021 if (pollfd == ctx->consumer_channel_pipe[0]) {
03e43155 3022 if (revents & LPOLLIN) {
d8ef542d 3023 enum consumer_channel_action action;
a0cbdd2e 3024 uint64_t key;
d8ef542d 3025
a0cbdd2e 3026 ret = read_channel_pipe(ctx, &chan, &key, &action);
d8ef542d 3027 if (ret <= 0) {
03e43155
MD
3028 if (ret < 0) {
3029 ERR("Error reading channel pipe");
3030 }
3031 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
d8ef542d
MD
3032 continue;
3033 }
3034
3035 switch (action) {
3036 case CONSUMER_CHANNEL_ADD:
3037 DBG("Adding channel %d to poll set",
3038 chan->wait_fd);
3039
3040 lttng_ht_node_init_u64(&chan->wait_fd_node,
3041 chan->wait_fd);
c7260a81 3042 rcu_read_lock();
d8ef542d
MD
3043 lttng_ht_add_unique_u64(channel_ht,
3044 &chan->wait_fd_node);
c7260a81 3045 rcu_read_unlock();
d8ef542d
MD
3046 /* Add channel to the global poll events list */
3047 lttng_poll_add(&events, chan->wait_fd,
03e43155 3048 LPOLLERR | LPOLLHUP);
d8ef542d 3049 break;
a0cbdd2e
MD
3050 case CONSUMER_CHANNEL_DEL:
3051 {
b4a650f3
DG
3052 /*
3053 * This command should never be called if the channel
3054 * has streams monitored by either the data or metadata
3055 * thread. The consumer only notify this thread with a
3056 * channel del. command if it receives a destroy
3057 * channel command from the session daemon that send it
3058 * if a command prior to the GET_CHANNEL failed.
3059 */
3060
c7260a81 3061 rcu_read_lock();
a0cbdd2e
MD
3062 chan = consumer_find_channel(key);
3063 if (!chan) {
c7260a81 3064 rcu_read_unlock();
a0cbdd2e
MD
3065 ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
3066 break;
3067 }
3068 lttng_poll_del(&events, chan->wait_fd);
f623cc0b 3069 iter.iter.node = &chan->wait_fd_node.node;
a0cbdd2e
MD
3070 ret = lttng_ht_del(channel_ht, &iter);
3071 assert(ret == 0);
a0cbdd2e 3072
f2a444f1
DG
3073 switch (consumer_data.type) {
3074 case LTTNG_CONSUMER_KERNEL:
3075 break;
3076 case LTTNG_CONSUMER32_UST:
3077 case LTTNG_CONSUMER64_UST:
212d67a2
DG
3078 health_code_update();
3079 /* Destroy streams that might have been left in the stream list. */
3080 clean_channel_stream_list(chan);
f2a444f1
DG
3081 break;
3082 default:
3083 ERR("Unknown consumer_data type");
3084 assert(0);
3085 }
3086
a0cbdd2e
MD
3087 /*
3088 * Release our own refcount. Force channel deletion even if
3089 * streams were not initialized.
3090 */
3091 if (!uatomic_sub_return(&chan->refcount, 1)) {
3092 consumer_del_channel(chan);
3093 }
c7260a81 3094 rcu_read_unlock();
a0cbdd2e
MD
3095 goto restart;
3096 }
d8ef542d
MD
3097 case CONSUMER_CHANNEL_QUIT:
3098 /*
3099 * Remove the pipe from the poll set and continue the loop
3100 * since their might be data to consume.
3101 */
3102 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3103 continue;
3104 default:
3105 ERR("Unknown action");
3106 break;
3107 }
03e43155
MD
3108 } else if (revents & (LPOLLERR | LPOLLHUP)) {
3109 DBG("Channel thread pipe hung up");
3110 /*
3111 * Remove the pipe from the poll set and continue the loop
3112 * since their might be data to consume.
3113 */
3114 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3115 continue;
3116 } else {
3117 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3118 goto end;
d8ef542d
MD
3119 }
3120
3121 /* Handle other stream */
3122 continue;
3123 }
3124
3125 rcu_read_lock();
3126 {
3127 uint64_t tmp_id = (uint64_t) pollfd;
3128
3129 lttng_ht_lookup(channel_ht, &tmp_id, &iter);
3130 }
3131 node = lttng_ht_iter_get_node_u64(&iter);
3132 assert(node);
3133
3134 chan = caa_container_of(node, struct lttng_consumer_channel,
3135 wait_fd_node);
3136
3137 /* Check for error event */
3138 if (revents & (LPOLLERR | LPOLLHUP)) {
3139 DBG("Channel fd %d is hup|err.", pollfd);
3140
3141 lttng_poll_del(&events, chan->wait_fd);
3142 ret = lttng_ht_del(channel_ht, &iter);
3143 assert(ret == 0);
b4a650f3
DG
3144
3145 /*
3146 * This will close the wait fd for each stream associated to
3147 * this channel AND monitored by the data/metadata thread thus
3148 * will be clean by the right thread.
3149 */
d8ef542d 3150 consumer_close_channel_streams(chan);
f2ad556d
MD
3151
3152 /* Release our own refcount */
3153 if (!uatomic_sub_return(&chan->refcount, 1)
3154 && !uatomic_read(&chan->nb_init_stream_left)) {
3155 consumer_del_channel(chan);
3156 }
03e43155
MD
3157 } else {
3158 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3159 rcu_read_unlock();
3160 goto end;
d8ef542d
MD
3161 }
3162
3163 /* Release RCU lock for the channel looked up */
3164 rcu_read_unlock();
3165 }
3166 }
3167
1fc79fb4
MD
3168 /* All is OK */
3169 err = 0;
d8ef542d
MD
3170end:
3171 lttng_poll_clean(&events);
3172end_poll:
3173 destroy_channel_ht(channel_ht);
3174end_ht:
2d57de81 3175error_testpoint:
d8ef542d 3176 DBG("Channel poll thread exiting");
1fc79fb4
MD
3177 if (err) {
3178 health_error();
3179 ERR("Health error occurred in %s", __func__);
3180 }
3181 health_unregister(health_consumerd);
d8ef542d
MD
3182 rcu_unregister_thread();
3183 return NULL;
3184}
3185
331744e3
JD
3186static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
3187 struct pollfd *sockpoll, int client_socket)
3188{
3189 int ret;
3190
3191 assert(ctx);
3192 assert(sockpoll);
3193
84382d49
MD
3194 ret = lttng_consumer_poll_socket(sockpoll);
3195 if (ret) {
331744e3
JD
3196 goto error;
3197 }
3198 DBG("Metadata connection on client_socket");
3199
3200 /* Blocking call, waiting for transmission */
3201 ctx->consumer_metadata_socket = lttcomm_accept_unix_sock(client_socket);
3202 if (ctx->consumer_metadata_socket < 0) {
3203 WARN("On accept metadata");
3204 ret = -1;
3205 goto error;
3206 }
3207 ret = 0;
3208
3209error:
3210 return ret;
3211}
3212
3bd1e081
MD
3213/*
3214 * This thread listens on the consumerd socket and receives the file
3215 * descriptors from the session daemon.
3216 */
7d980def 3217void *consumer_thread_sessiond_poll(void *data)
3bd1e081 3218{
1fc79fb4 3219 int sock = -1, client_socket, ret, err = -1;
3bd1e081
MD
3220 /*
3221 * structure to poll for incoming data on communication socket avoids
3222 * making blocking sockets.
3223 */
3224 struct pollfd consumer_sockpoll[2];
3225 struct lttng_consumer_local_data *ctx = data;
3226
e7b994a3
DG
3227 rcu_register_thread();
3228
1fc79fb4
MD
3229 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_SESSIOND);
3230
2d57de81
MD
3231 if (testpoint(consumerd_thread_sessiond)) {
3232 goto error_testpoint;
3233 }
3234
9ce5646a
MD
3235 health_code_update();
3236
3bd1e081
MD
3237 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
3238 unlink(ctx->consumer_command_sock_path);
3239 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
3240 if (client_socket < 0) {
3241 ERR("Cannot create command socket");
3242 goto end;
3243 }
3244
3245 ret = lttcomm_listen_unix_sock(client_socket);
3246 if (ret < 0) {
3247 goto end;
3248 }
3249
32258573 3250 DBG("Sending ready command to lttng-sessiond");
f73fabfd 3251 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
3bd1e081
MD
3252 /* return < 0 on error, but == 0 is not fatal */
3253 if (ret < 0) {
32258573 3254 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
3