Fix: propagate archive id to the consumer daemon on stream creation
[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 564 enum consumer_channel_type type,
e098433c
JG
565 unsigned int monitor,
566 uint64_t trace_archive_id)
3bd1e081 567{
ffe60014 568 int ret;
3bd1e081 569 struct lttng_consumer_stream *stream;
3bd1e081 570
effcf122 571 stream = zmalloc(sizeof(*stream));
3bd1e081 572 if (stream == NULL) {
7a57cf92 573 PERROR("malloc struct lttng_consumer_stream");
ffe60014 574 ret = -ENOMEM;
7a57cf92 575 goto end;
3bd1e081 576 }
7a57cf92 577
d56db448
DG
578 rcu_read_lock();
579
3bd1e081 580 stream->key = stream_key;
3bd1e081
MD
581 stream->out_fd = -1;
582 stream->out_fd_offset = 0;
e5d1a9b3 583 stream->output_written = 0;
3bd1e081 584 stream->state = state;
6df2e2c9
MD
585 stream->uid = uid;
586 stream->gid = gid;
ffe60014 587 stream->net_seq_idx = relayd_id;
53632229 588 stream->session_id = session_id;
4891ece8 589 stream->monitor = monitor;
774d490c 590 stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
f8f3885c 591 stream->index_file = NULL;
fb83fe64 592 stream->last_sequence_number = -1ULL;
e098433c 593 stream->trace_archive_id = trace_archive_id;
53632229 594 pthread_mutex_init(&stream->lock, NULL);
c585821b 595 pthread_mutex_init(&stream->metadata_timer_lock, NULL);
58b1f425 596
ffe60014
DG
597 /* If channel is the metadata, flag this stream as metadata. */
598 if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
599 stream->metadata_flag = 1;
600 /* Metadata is flat out. */
601 strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
94d49140
JD
602 /* Live rendez-vous point. */
603 pthread_cond_init(&stream->metadata_rdv, NULL);
604 pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
58b1f425 605 } else {
ffe60014
DG
606 /* Format stream name to <channel_name>_<cpu_number> */
607 ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
608 channel_name, cpu);
609 if (ret < 0) {
610 PERROR("snprintf stream name");
611 goto error;
612 }
58b1f425 613 }
c30aaa51 614
ffe60014 615 /* Key is always the wait_fd for streams. */
d88aee68 616 lttng_ht_node_init_u64(&stream->node, stream->key);
ffe60014 617
d8ef542d
MD
618 /* Init node per channel id key */
619 lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
620
53632229 621 /* Init session id node with the stream session id */
d88aee68 622 lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
53632229 623
07b86b52
JD
624 DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
625 " relayd_id %" PRIu64 ", session_id %" PRIu64,
626 stream->name, stream->key, channel_key,
627 stream->net_seq_idx, stream->session_id);
d56db448
DG
628
629 rcu_read_unlock();
3bd1e081 630 return stream;
c80048c6
MD
631
632error:
d56db448 633 rcu_read_unlock();
c80048c6 634 free(stream);
7a57cf92 635end:
ffe60014
DG
636 if (alloc_ret) {
637 *alloc_ret = ret;
638 }
c80048c6 639 return NULL;
3bd1e081
MD
640}
641
642/*
643 * Add a stream to the global list protected by a mutex.
644 */
66d583dc 645void consumer_add_data_stream(struct lttng_consumer_stream *stream)
3bd1e081 646{
5ab66908 647 struct lttng_ht *ht = data_ht;
3bd1e081 648
e316aad5 649 assert(stream);
43c34bc3 650 assert(ht);
c77fc10a 651
d88aee68 652 DBG3("Adding consumer stream %" PRIu64, stream->key);
e316aad5
DG
653
654 pthread_mutex_lock(&consumer_data.lock);
a9838785 655 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 656 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 657 pthread_mutex_lock(&stream->lock);
b0b335c8 658 rcu_read_lock();
e316aad5 659
43c34bc3 660 /* Steal stream identifier to avoid having streams with the same key */
ffe60014 661 steal_stream_key(stream->key, ht);
43c34bc3 662
d88aee68 663 lttng_ht_add_unique_u64(ht, &stream->node);
00e2e675 664
d8ef542d
MD
665 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
666 &stream->node_channel_id);
667
ca22feea
DG
668 /*
669 * Add stream to the stream_list_ht of the consumer data. No need to steal
670 * the key since the HT does not use it and we allow to add redundant keys
671 * into this table.
672 */
d88aee68 673 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 674
e316aad5 675 /*
ffe60014
DG
676 * When nb_init_stream_left reaches 0, we don't need to trigger any action
677 * in terms of destroying the associated channel, because the action that
e316aad5
DG
678 * causes the count to become 0 also causes a stream to be added. The
679 * channel deletion will thus be triggered by the following removal of this
680 * stream.
681 */
ffe60014 682 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
683 /* Increment refcount before decrementing nb_init_stream_left */
684 cmm_smp_wmb();
ffe60014 685 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
686 }
687
688 /* Update consumer data once the node is inserted. */
3bd1e081
MD
689 consumer_data.stream_count++;
690 consumer_data.need_update = 1;
691
e316aad5 692 rcu_read_unlock();
2e818a6a 693 pthread_mutex_unlock(&stream->lock);
ec6ea7d0 694 pthread_mutex_unlock(&stream->chan->timer_lock);
a9838785 695 pthread_mutex_unlock(&stream->chan->lock);
3bd1e081 696 pthread_mutex_unlock(&consumer_data.lock);
3bd1e081
MD
697}
698
5ab66908
MD
699void consumer_del_data_stream(struct lttng_consumer_stream *stream)
700{
701 consumer_del_stream(stream, data_ht);
702}
703
00e2e675 704/*
3f8e211f
DG
705 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
706 * be acquired before calling this.
00e2e675 707 */
d09e1200 708static int add_relayd(struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
709{
710 int ret = 0;
d88aee68 711 struct lttng_ht_node_u64 *node;
00e2e675
DG
712 struct lttng_ht_iter iter;
713
ffe60014 714 assert(relayd);
00e2e675 715
00e2e675 716 lttng_ht_lookup(consumer_data.relayd_ht,
d88aee68
DG
717 &relayd->net_seq_idx, &iter);
718 node = lttng_ht_iter_get_node_u64(&iter);
00e2e675 719 if (node != NULL) {
00e2e675
DG
720 goto end;
721 }
d88aee68 722 lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
00e2e675 723
00e2e675
DG
724end:
725 return ret;
726}
727
728/*
729 * Allocate and return a consumer relayd socket.
730 */
027a694f 731static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
da009f2c 732 uint64_t net_seq_idx)
00e2e675
DG
733{
734 struct consumer_relayd_sock_pair *obj = NULL;
735
da009f2c
MD
736 /* net sequence index of -1 is a failure */
737 if (net_seq_idx == (uint64_t) -1ULL) {
00e2e675
DG
738 goto error;
739 }
740
741 obj = zmalloc(sizeof(struct consumer_relayd_sock_pair));
742 if (obj == NULL) {
743 PERROR("zmalloc relayd sock");
744 goto error;
745 }
746
747 obj->net_seq_idx = net_seq_idx;
748 obj->refcount = 0;
173af62f 749 obj->destroy_flag = 0;
f96e4545
MD
750 obj->control_sock.sock.fd = -1;
751 obj->data_sock.sock.fd = -1;
d88aee68 752 lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
00e2e675
DG
753 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
754
755error:
756 return obj;
757}
758
759/*
760 * Find a relayd socket pair in the global consumer data.
761 *
762 * Return the object if found else NULL.
b0b335c8
MD
763 * RCU read-side lock must be held across this call and while using the
764 * returned object.
00e2e675 765 */
d88aee68 766struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
00e2e675
DG
767{
768 struct lttng_ht_iter iter;
d88aee68 769 struct lttng_ht_node_u64 *node;
00e2e675
DG
770 struct consumer_relayd_sock_pair *relayd = NULL;
771
772 /* Negative keys are lookup failures */
d88aee68 773 if (key == (uint64_t) -1ULL) {
00e2e675
DG
774 goto error;
775 }
776
d88aee68 777 lttng_ht_lookup(consumer_data.relayd_ht, &key,
00e2e675 778 &iter);
d88aee68 779 node = lttng_ht_iter_get_node_u64(&iter);
00e2e675
DG
780 if (node != NULL) {
781 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
782 }
783
00e2e675
DG
784error:
785 return relayd;
786}
787
10a50311
JD
788/*
789 * Find a relayd and send the stream
790 *
791 * Returns 0 on success, < 0 on error
792 */
793int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
794 char *path)
795{
796 int ret = 0;
797 struct consumer_relayd_sock_pair *relayd;
798
799 assert(stream);
800 assert(stream->net_seq_idx != -1ULL);
801 assert(path);
802
803 /* The stream is not metadata. Get relayd reference if exists. */
804 rcu_read_lock();
805 relayd = consumer_find_relayd(stream->net_seq_idx);
806 if (relayd != NULL) {
807 /* Add stream on the relayd */
808 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
809 ret = relayd_add_stream(&relayd->control_sock, stream->name,
810 path, &stream->relayd_stream_id,
811 stream->chan->tracefile_size, stream->chan->tracefile_count);
812 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
813 if (ret < 0) {
814 goto end;
815 }
1c20f0e2 816
10a50311 817 uatomic_inc(&relayd->refcount);
d01178b6 818 stream->sent_to_relayd = 1;
10a50311
JD
819 } else {
820 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
821 stream->key, stream->net_seq_idx);
822 ret = -1;
823 goto end;
824 }
825
826 DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
827 stream->name, stream->key, stream->net_seq_idx);
828
829end:
830 rcu_read_unlock();
831 return ret;
832}
833
a4baae1b
JD
834/*
835 * Find a relayd and send the streams sent message
836 *
837 * Returns 0 on success, < 0 on error
838 */
839int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
840{
841 int ret = 0;
842 struct consumer_relayd_sock_pair *relayd;
843
844 assert(net_seq_idx != -1ULL);
845
846 /* The stream is not metadata. Get relayd reference if exists. */
847 rcu_read_lock();
848 relayd = consumer_find_relayd(net_seq_idx);
849 if (relayd != NULL) {
850 /* Add stream on the relayd */
851 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
852 ret = relayd_streams_sent(&relayd->control_sock);
853 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
854 if (ret < 0) {
855 goto end;
856 }
857 } else {
858 ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
859 net_seq_idx);
860 ret = -1;
861 goto end;
862 }
863
864 ret = 0;
865 DBG("All streams sent relayd id %" PRIu64, net_seq_idx);
866
867end:
868 rcu_read_unlock();
869 return ret;
870}
871
10a50311
JD
872/*
873 * Find a relayd and close the stream
874 */
875void close_relayd_stream(struct lttng_consumer_stream *stream)
876{
877 struct consumer_relayd_sock_pair *relayd;
878
879 /* The stream is not metadata. Get relayd reference if exists. */
880 rcu_read_lock();
881 relayd = consumer_find_relayd(stream->net_seq_idx);
882 if (relayd) {
883 consumer_stream_relayd_close(stream, relayd);
884 }
885 rcu_read_unlock();
886}
887
00e2e675
DG
888/*
889 * Handle stream for relayd transmission if the stream applies for network
890 * streaming where the net sequence index is set.
891 *
892 * Return destination file descriptor or negative value on error.
893 */
6197aea7 894static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
1d4dfdef
DG
895 size_t data_size, unsigned long padding,
896 struct consumer_relayd_sock_pair *relayd)
00e2e675
DG
897{
898 int outfd = -1, ret;
00e2e675
DG
899 struct lttcomm_relayd_data_hdr data_hdr;
900
901 /* Safety net */
902 assert(stream);
6197aea7 903 assert(relayd);
00e2e675
DG
904
905 /* Reset data header */
906 memset(&data_hdr, 0, sizeof(data_hdr));
907
00e2e675
DG
908 if (stream->metadata_flag) {
909 /* Caller MUST acquire the relayd control socket lock */
910 ret = relayd_send_metadata(&relayd->control_sock, data_size);
911 if (ret < 0) {
912 goto error;
913 }
914
915 /* Metadata are always sent on the control socket. */
6151a90f 916 outfd = relayd->control_sock.sock.fd;
00e2e675
DG
917 } else {
918 /* Set header with stream information */
919 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
920 data_hdr.data_size = htobe32(data_size);
1d4dfdef 921 data_hdr.padding_size = htobe32(padding);
39df6d9f
DG
922 /*
923 * Note that net_seq_num below is assigned with the *current* value of
924 * next_net_seq_num and only after that the next_net_seq_num will be
925 * increment. This is why when issuing a command on the relayd using
926 * this next value, 1 should always be substracted in order to compare
927 * the last seen sequence number on the relayd side to the last sent.
928 */
3604f373 929 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
00e2e675
DG
930 /* Other fields are zeroed previously */
931
932 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
933 sizeof(data_hdr));
934 if (ret < 0) {
935 goto error;
936 }
937
3604f373
DG
938 ++stream->next_net_seq_num;
939
00e2e675 940 /* Set to go on data socket */
6151a90f 941 outfd = relayd->data_sock.sock.fd;
00e2e675
DG
942 }
943
944error:
945 return outfd;
946}
947
3bd1e081 948/*
ffe60014
DG
949 * Allocate and return a new lttng_consumer_channel object using the given key
950 * to initialize the hash table node.
951 *
952 * On error, return NULL.
3bd1e081 953 */
886224ff 954struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
ffe60014
DG
955 uint64_t session_id,
956 const char *pathname,
957 const char *name,
958 uid_t uid,
959 gid_t gid,
57a269f2 960 uint64_t relayd_id,
1624d5b7
JD
961 enum lttng_event_output output,
962 uint64_t tracefile_size,
2bba9e53 963 uint64_t tracefile_count,
1950109e 964 uint64_t session_id_per_pid,
ecc48a90 965 unsigned int monitor,
d7ba1388 966 unsigned int live_timer_interval,
3d071855 967 const char *root_shm_path,
d7ba1388 968 const char *shm_path)
3bd1e081
MD
969{
970 struct lttng_consumer_channel *channel;
3bd1e081 971
276b26d1 972 channel = zmalloc(sizeof(*channel));
3bd1e081 973 if (channel == NULL) {
7a57cf92 974 PERROR("malloc struct lttng_consumer_channel");
3bd1e081
MD
975 goto end;
976 }
ffe60014
DG
977
978 channel->key = key;
3bd1e081 979 channel->refcount = 0;
ffe60014 980 channel->session_id = session_id;
1950109e 981 channel->session_id_per_pid = session_id_per_pid;
ffe60014
DG
982 channel->uid = uid;
983 channel->gid = gid;
984 channel->relayd_id = relayd_id;
1624d5b7
JD
985 channel->tracefile_size = tracefile_size;
986 channel->tracefile_count = tracefile_count;
2bba9e53 987 channel->monitor = monitor;
ecc48a90 988 channel->live_timer_interval = live_timer_interval;
a9838785 989 pthread_mutex_init(&channel->lock, NULL);
ec6ea7d0 990 pthread_mutex_init(&channel->timer_lock, NULL);
ffe60014 991
0c759fc9
DG
992 switch (output) {
993 case LTTNG_EVENT_SPLICE:
994 channel->output = CONSUMER_CHANNEL_SPLICE;
995 break;
996 case LTTNG_EVENT_MMAP:
997 channel->output = CONSUMER_CHANNEL_MMAP;
998 break;
999 default:
1000 assert(0);
1001 free(channel);
1002 channel = NULL;
1003 goto end;
1004 }
1005
07b86b52
JD
1006 /*
1007 * In monitor mode, the streams associated with the channel will be put in
1008 * a special list ONLY owned by this channel. So, the refcount is set to 1
1009 * here meaning that the channel itself has streams that are referenced.
1010 *
1011 * On a channel deletion, once the channel is no longer visible, the
1012 * refcount is decremented and checked for a zero value to delete it. With
1013 * streams in no monitor mode, it will now be safe to destroy the channel.
1014 */
1015 if (!channel->monitor) {
1016 channel->refcount = 1;
1017 }
1018
ffe60014
DG
1019 strncpy(channel->pathname, pathname, sizeof(channel->pathname));
1020 channel->pathname[sizeof(channel->pathname) - 1] = '\0';
1021
1022 strncpy(channel->name, name, sizeof(channel->name));
1023 channel->name[sizeof(channel->name) - 1] = '\0';
1024
3d071855
MD
1025 if (root_shm_path) {
1026 strncpy(channel->root_shm_path, root_shm_path, sizeof(channel->root_shm_path));
1027 channel->root_shm_path[sizeof(channel->root_shm_path) - 1] = '\0';
1028 }
d7ba1388
MD
1029 if (shm_path) {
1030 strncpy(channel->shm_path, shm_path, sizeof(channel->shm_path));
1031 channel->shm_path[sizeof(channel->shm_path) - 1] = '\0';
1032 }
1033
d88aee68 1034 lttng_ht_node_init_u64(&channel->node, channel->key);
d8ef542d
MD
1035
1036 channel->wait_fd = -1;
1037
ffe60014
DG
1038 CDS_INIT_LIST_HEAD(&channel->streams.head);
1039
62a7b8ed 1040 DBG("Allocated channel (key %" PRIu64 ")", channel->key);
3bd1e081 1041
3bd1e081
MD
1042end:
1043 return channel;
1044}
1045
1046/*
1047 * Add a channel to the global list protected by a mutex.
821fffb2 1048 *
b5a6470f 1049 * Always return 0 indicating success.
3bd1e081 1050 */
d8ef542d
MD
1051int consumer_add_channel(struct lttng_consumer_channel *channel,
1052 struct lttng_consumer_local_data *ctx)
3bd1e081 1053{
3bd1e081 1054 pthread_mutex_lock(&consumer_data.lock);
a9838785 1055 pthread_mutex_lock(&channel->lock);
ec6ea7d0 1056 pthread_mutex_lock(&channel->timer_lock);
c77fc10a 1057
b5a6470f
DG
1058 /*
1059 * This gives us a guarantee that the channel we are about to add to the
1060 * channel hash table will be unique. See this function comment on the why
1061 * we need to steel the channel key at this stage.
1062 */
1063 steal_channel_key(channel->key);
c77fc10a 1064
b5a6470f 1065 rcu_read_lock();
d88aee68 1066 lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
6065ceec 1067 rcu_read_unlock();
b5a6470f 1068
ec6ea7d0 1069 pthread_mutex_unlock(&channel->timer_lock);
a9838785 1070 pthread_mutex_unlock(&channel->lock);
3bd1e081 1071 pthread_mutex_unlock(&consumer_data.lock);
702b1ea4 1072
b5a6470f 1073 if (channel->wait_fd != -1 && channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
a0cbdd2e 1074 notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
d8ef542d 1075 }
b5a6470f
DG
1076
1077 return 0;
3bd1e081
MD
1078}
1079
1080/*
1081 * Allocate the pollfd structure and the local view of the out fds to avoid
1082 * doing a lookup in the linked list and concurrency issues when writing is
1083 * needed. Called with consumer_data.lock held.
1084 *
1085 * Returns the number of fds in the structures.
1086 */
ffe60014
DG
1087static int update_poll_array(struct lttng_consumer_local_data *ctx,
1088 struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
9a2fcf78 1089 struct lttng_ht *ht, int *nb_inactive_fd)
3bd1e081 1090{
3bd1e081 1091 int i = 0;
e4421fec
DG
1092 struct lttng_ht_iter iter;
1093 struct lttng_consumer_stream *stream;
3bd1e081 1094
ffe60014
DG
1095 assert(ctx);
1096 assert(ht);
1097 assert(pollfd);
1098 assert(local_stream);
1099
3bd1e081 1100 DBG("Updating poll fd array");
9a2fcf78 1101 *nb_inactive_fd = 0;
481d6c57 1102 rcu_read_lock();
43c34bc3 1103 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
8994307f
DG
1104 /*
1105 * Only active streams with an active end point can be added to the
1106 * poll set and local stream storage of the thread.
1107 *
1108 * There is a potential race here for endpoint_status to be updated
1109 * just after the check. However, this is OK since the stream(s) will
1110 * be deleted once the thread is notified that the end point state has
1111 * changed where this function will be called back again.
9a2fcf78
JD
1112 *
1113 * We track the number of inactive FDs because they still need to be
1114 * closed by the polling thread after a wakeup on the data_pipe or
1115 * metadata_pipe.
8994307f
DG
1116 */
1117 if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
79d4ffb7 1118 stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
9a2fcf78 1119 (*nb_inactive_fd)++;
3bd1e081
MD
1120 continue;
1121 }
7972aab2
DG
1122 /*
1123 * This clobbers way too much the debug output. Uncomment that if you
1124 * need it for debugging purposes.
1125 *
1126 * DBG("Active FD %d", stream->wait_fd);
1127 */
e4421fec 1128 (*pollfd)[i].fd = stream->wait_fd;
3bd1e081 1129 (*pollfd)[i].events = POLLIN | POLLPRI;
e4421fec 1130 local_stream[i] = stream;
3bd1e081
MD
1131 i++;
1132 }
481d6c57 1133 rcu_read_unlock();
3bd1e081
MD
1134
1135 /*
50f8ae69 1136 * Insert the consumer_data_pipe at the end of the array and don't
3bd1e081
MD
1137 * increment i so nb_fd is the number of real FD.
1138 */
acdb9057 1139 (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
509bb1cf 1140 (*pollfd)[i].events = POLLIN | POLLPRI;
02b3d176
DG
1141
1142 (*pollfd)[i + 1].fd = lttng_pipe_get_readfd(ctx->consumer_wakeup_pipe);
1143 (*pollfd)[i + 1].events = POLLIN | POLLPRI;
3bd1e081
MD
1144 return i;
1145}
1146
1147/*
84382d49
MD
1148 * Poll on the should_quit pipe and the command socket return -1 on
1149 * error, 1 if should exit, 0 if data is available on the command socket
3bd1e081
MD
1150 */
1151int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
1152{
1153 int num_rdy;
1154
88f2b785 1155restart:
3bd1e081
MD
1156 num_rdy = poll(consumer_sockpoll, 2, -1);
1157 if (num_rdy == -1) {
88f2b785
MD
1158 /*
1159 * Restart interrupted system call.
1160 */
1161 if (errno == EINTR) {
1162 goto restart;
1163 }
7a57cf92 1164 PERROR("Poll error");
84382d49 1165 return -1;
3bd1e081 1166 }
509bb1cf 1167 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
3bd1e081 1168 DBG("consumer_should_quit wake up");
84382d49 1169 return 1;
3bd1e081
MD
1170 }
1171 return 0;
3bd1e081
MD
1172}
1173
1174/*
1175 * Set the error socket.
1176 */
ffe60014
DG
1177void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
1178 int sock)
3bd1e081
MD
1179{
1180 ctx->consumer_error_socket = sock;
1181}
1182
1183/*
1184 * Set the command socket path.
1185 */
3bd1e081
MD
1186void lttng_consumer_set_command_sock_path(
1187 struct lttng_consumer_local_data *ctx, char *sock)
1188{
1189 ctx->consumer_command_sock_path = sock;
1190}
1191
1192/*
1193 * Send return code to the session daemon.
1194 * If the socket is not defined, we return 0, it is not a fatal error
1195 */
ffe60014 1196int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
3bd1e081
MD
1197{
1198 if (ctx->consumer_error_socket > 0) {
1199 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
1200 sizeof(enum lttcomm_sessiond_command));
1201 }
1202
1203 return 0;
1204}
1205
1206/*
228b5bf7
DG
1207 * Close all the tracefiles and stream fds and MUST be called when all
1208 * instances are destroyed i.e. when all threads were joined and are ended.
3bd1e081
MD
1209 */
1210void lttng_consumer_cleanup(void)
1211{
e4421fec 1212 struct lttng_ht_iter iter;
ffe60014 1213 struct lttng_consumer_channel *channel;
6065ceec
DG
1214
1215 rcu_read_lock();
3bd1e081 1216
ffe60014
DG
1217 cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
1218 node.node) {
702b1ea4 1219 consumer_del_channel(channel);
3bd1e081 1220 }
6065ceec
DG
1221
1222 rcu_read_unlock();
d6ce1df2 1223
d6ce1df2 1224 lttng_ht_destroy(consumer_data.channel_ht);
228b5bf7
DG
1225
1226 cleanup_relayd_ht();
1227
d8ef542d
MD
1228 lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
1229
228b5bf7
DG
1230 /*
1231 * This HT contains streams that are freed by either the metadata thread or
1232 * the data thread so we do *nothing* on the hash table and simply destroy
1233 * it.
1234 */
1235 lttng_ht_destroy(consumer_data.stream_list_ht);
3bd1e081
MD
1236}
1237
1238/*
1239 * Called from signal handler.
1240 */
1241void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
1242{
6cd525e8
MD
1243 ssize_t ret;
1244
10211f5c 1245 CMM_STORE_SHARED(consumer_quit, 1);
6cd525e8
MD
1246 ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
1247 if (ret < 1) {
7a57cf92 1248 PERROR("write consumer quit");
3bd1e081 1249 }
ab1027f4
DG
1250
1251 DBG("Consumer flag that it should quit");
3bd1e081
MD
1252}
1253
5199ffc4
JG
1254
1255/*
1256 * Flush pending writes to trace output disk file.
1257 */
1258static
00e2e675
DG
1259void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
1260 off_t orig_offset)
3bd1e081 1261{
c7a78aab 1262 int ret;
3bd1e081
MD
1263 int outfd = stream->out_fd;
1264
1265 /*
1266 * This does a blocking write-and-wait on any page that belongs to the
1267 * subbuffer prior to the one we just wrote.
1268 * Don't care about error values, as these are just hints and ways to
1269 * limit the amount of page cache used.
1270 */
ffe60014 1271 if (orig_offset < stream->max_sb_size) {
3bd1e081
MD
1272 return;
1273 }
ffe60014
DG
1274 lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
1275 stream->max_sb_size,
3bd1e081
MD
1276 SYNC_FILE_RANGE_WAIT_BEFORE
1277 | SYNC_FILE_RANGE_WRITE
1278 | SYNC_FILE_RANGE_WAIT_AFTER);
1279 /*
1280 * Give hints to the kernel about how we access the file:
1281 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1282 * we write it.
1283 *
1284 * We need to call fadvise again after the file grows because the
1285 * kernel does not seem to apply fadvise to non-existing parts of the
1286 * file.
1287 *
1288 * Call fadvise _after_ having waited for the page writeback to
1289 * complete because the dirty page writeback semantic is not well
1290 * defined. So it can be expected to lead to lower throughput in
1291 * streaming.
1292 */
c7a78aab 1293 ret = posix_fadvise(outfd, orig_offset - stream->max_sb_size,
ffe60014 1294 stream->max_sb_size, POSIX_FADV_DONTNEED);
a0d0e127 1295 if (ret && ret != -ENOSYS) {
a74a5f4a
JG
1296 errno = ret;
1297 PERROR("posix_fadvise on fd %i", outfd);
c7a78aab 1298 }
3bd1e081
MD
1299}
1300
1301/*
1302 * Initialise the necessary environnement :
1303 * - create a new context
1304 * - create the poll_pipe
1305 * - create the should_quit pipe (for signal handler)
1306 * - create the thread pipe (for splice)
1307 *
1308 * Takes a function pointer as argument, this function is called when data is
1309 * available on a buffer. This function is responsible to do the
1310 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1311 * buffer configuration and then kernctl_put_next_subbuf at the end.
1312 *
1313 * Returns a pointer to the new context or NULL on error.
1314 */
1315struct lttng_consumer_local_data *lttng_consumer_create(
1316 enum lttng_consumer_type type,
4078b776 1317 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 1318 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
1319 int (*recv_channel)(struct lttng_consumer_channel *channel),
1320 int (*recv_stream)(struct lttng_consumer_stream *stream),
30319bcb 1321 int (*update_stream)(uint64_t stream_key, uint32_t state))
3bd1e081 1322{
d8ef542d 1323 int ret;
3bd1e081
MD
1324 struct lttng_consumer_local_data *ctx;
1325
1326 assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
1327 consumer_data.type == type);
1328 consumer_data.type = type;
1329
effcf122 1330 ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
3bd1e081 1331 if (ctx == NULL) {
7a57cf92 1332 PERROR("allocating context");
3bd1e081
MD
1333 goto error;
1334 }
1335
1336 ctx->consumer_error_socket = -1;
331744e3 1337 ctx->consumer_metadata_socket = -1;
75d83e50 1338 pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
3bd1e081
MD
1339 /* assign the callbacks */
1340 ctx->on_buffer_ready = buffer_ready;
1341 ctx->on_recv_channel = recv_channel;
1342 ctx->on_recv_stream = recv_stream;
1343 ctx->on_update_stream = update_stream;
1344
acdb9057
DG
1345 ctx->consumer_data_pipe = lttng_pipe_open(0);
1346 if (!ctx->consumer_data_pipe) {
3bd1e081
MD
1347 goto error_poll_pipe;
1348 }
1349
02b3d176
DG
1350 ctx->consumer_wakeup_pipe = lttng_pipe_open(0);
1351 if (!ctx->consumer_wakeup_pipe) {
1352 goto error_wakeup_pipe;
1353 }
1354
3bd1e081
MD
1355 ret = pipe(ctx->consumer_should_quit);
1356 if (ret < 0) {
7a57cf92 1357 PERROR("Error creating recv pipe");
3bd1e081
MD
1358 goto error_quit_pipe;
1359 }
1360
d8ef542d
MD
1361 ret = pipe(ctx->consumer_channel_pipe);
1362 if (ret < 0) {
1363 PERROR("Error creating channel pipe");
1364 goto error_channel_pipe;
1365 }
1366
13886d2d
DG
1367 ctx->consumer_metadata_pipe = lttng_pipe_open(0);
1368 if (!ctx->consumer_metadata_pipe) {
fb3a43a9
DG
1369 goto error_metadata_pipe;
1370 }
3bd1e081 1371
e9404c27
JG
1372 ctx->channel_monitor_pipe = -1;
1373
fb3a43a9 1374 return ctx;
3bd1e081 1375
fb3a43a9 1376error_metadata_pipe:
d8ef542d
MD
1377 utils_close_pipe(ctx->consumer_channel_pipe);
1378error_channel_pipe:
d8ef542d 1379 utils_close_pipe(ctx->consumer_should_quit);
3bd1e081 1380error_quit_pipe:
02b3d176
DG
1381 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
1382error_wakeup_pipe:
acdb9057 1383 lttng_pipe_destroy(ctx->consumer_data_pipe);
3bd1e081
MD
1384error_poll_pipe:
1385 free(ctx);
1386error:
1387 return NULL;
1388}
1389
282dadbc
MD
1390/*
1391 * Iterate over all streams of the hashtable and free them properly.
1392 */
1393static void destroy_data_stream_ht(struct lttng_ht *ht)
1394{
1395 struct lttng_ht_iter iter;
1396 struct lttng_consumer_stream *stream;
1397
1398 if (ht == NULL) {
1399 return;
1400 }
1401
1402 rcu_read_lock();
1403 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1404 /*
1405 * Ignore return value since we are currently cleaning up so any error
1406 * can't be handled.
1407 */
1408 (void) consumer_del_stream(stream, ht);
1409 }
1410 rcu_read_unlock();
1411
1412 lttng_ht_destroy(ht);
1413}
1414
1415/*
1416 * Iterate over all streams of the metadata hashtable and free them
1417 * properly.
1418 */
1419static void destroy_metadata_stream_ht(struct lttng_ht *ht)
1420{
1421 struct lttng_ht_iter iter;
1422 struct lttng_consumer_stream *stream;
1423
1424 if (ht == NULL) {
1425 return;
1426 }
1427
1428 rcu_read_lock();
1429 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1430 /*
1431 * Ignore return value since we are currently cleaning up so any error
1432 * can't be handled.
1433 */
1434 (void) consumer_del_metadata_stream(stream, ht);
1435 }
1436 rcu_read_unlock();
1437
1438 lttng_ht_destroy(ht);
1439}
1440
3bd1e081
MD
1441/*
1442 * Close all fds associated with the instance and free the context.
1443 */
1444void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1445{
4c462e79
MD
1446 int ret;
1447
ab1027f4
DG
1448 DBG("Consumer destroying it. Closing everything.");
1449
4f2e75b9
DG
1450 if (!ctx) {
1451 return;
1452 }
1453
282dadbc
MD
1454 destroy_data_stream_ht(data_ht);
1455 destroy_metadata_stream_ht(metadata_ht);
1456
4c462e79
MD
1457 ret = close(ctx->consumer_error_socket);
1458 if (ret) {
1459 PERROR("close");
1460 }
331744e3
JD
1461 ret = close(ctx->consumer_metadata_socket);
1462 if (ret) {
1463 PERROR("close");
1464 }
d8ef542d 1465 utils_close_pipe(ctx->consumer_channel_pipe);
acdb9057 1466 lttng_pipe_destroy(ctx->consumer_data_pipe);
13886d2d 1467 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
02b3d176 1468 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
d8ef542d 1469 utils_close_pipe(ctx->consumer_should_quit);
fb3a43a9 1470
3bd1e081
MD
1471 unlink(ctx->consumer_command_sock_path);
1472 free(ctx);
1473}
1474
6197aea7
DG
1475/*
1476 * Write the metadata stream id on the specified file descriptor.
1477 */
1478static int write_relayd_metadata_id(int fd,
1479 struct lttng_consumer_stream *stream,
239f61af 1480 unsigned long padding)
6197aea7 1481{
6cd525e8 1482 ssize_t ret;
1d4dfdef 1483 struct lttcomm_relayd_metadata_payload hdr;
6197aea7 1484
1d4dfdef
DG
1485 hdr.stream_id = htobe64(stream->relayd_stream_id);
1486 hdr.padding_size = htobe32(padding);
6cd525e8
MD
1487 ret = lttng_write(fd, (void *) &hdr, sizeof(hdr));
1488 if (ret < sizeof(hdr)) {
d7b75ec8 1489 /*
6f04ed72 1490 * This error means that the fd's end is closed so ignore the PERROR
d7b75ec8
DG
1491 * not to clubber the error output since this can happen in a normal
1492 * code path.
1493 */
1494 if (errno != EPIPE) {
1495 PERROR("write metadata stream id");
1496 }
1497 DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
534d2592
DG
1498 /*
1499 * Set ret to a negative value because if ret != sizeof(hdr), we don't
1500 * handle writting the missing part so report that as an error and
1501 * don't lie to the caller.
1502 */
1503 ret = -1;
6197aea7
DG
1504 goto end;
1505 }
1d4dfdef
DG
1506 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1507 stream->relayd_stream_id, padding);
6197aea7
DG
1508
1509end:
6cd525e8 1510 return (int) ret;
6197aea7
DG
1511}
1512
3bd1e081 1513/*
09e26845
DG
1514 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1515 * core function for writing trace buffers to either the local filesystem or
1516 * the network.
1517 *
79d4ffb7
DG
1518 * It must be called with the stream lock held.
1519 *
09e26845 1520 * Careful review MUST be put if any changes occur!
3bd1e081
MD
1521 *
1522 * Returns the number of bytes written
1523 */
4078b776 1524ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081 1525 struct lttng_consumer_local_data *ctx,
1d4dfdef 1526 struct lttng_consumer_stream *stream, unsigned long len,
309167d2 1527 unsigned long padding,
50adc264 1528 struct ctf_packet_index *index)
3bd1e081 1529{
f02e1e8a 1530 unsigned long mmap_offset;
ffe60014 1531 void *mmap_base;
994ab360 1532 ssize_t ret = 0;
f02e1e8a
DG
1533 off_t orig_offset = stream->out_fd_offset;
1534 /* Default is on the disk */
1535 int outfd = stream->out_fd;
f02e1e8a 1536 struct consumer_relayd_sock_pair *relayd = NULL;
8994307f 1537 unsigned int relayd_hang_up = 0;
f02e1e8a
DG
1538
1539 /* RCU lock for the relayd pointer */
1540 rcu_read_lock();
1541
1542 /* Flag that the current stream if set for network streaming. */
da009f2c 1543 if (stream->net_seq_idx != (uint64_t) -1ULL) {
f02e1e8a
DG
1544 relayd = consumer_find_relayd(stream->net_seq_idx);
1545 if (relayd == NULL) {
56591bac 1546 ret = -EPIPE;
f02e1e8a
DG
1547 goto end;
1548 }
1549 }
1550
1551 /* get the offset inside the fd to mmap */
3bd1e081
MD
1552 switch (consumer_data.type) {
1553 case LTTNG_CONSUMER_KERNEL:
ffe60014 1554 mmap_base = stream->mmap_base;
f02e1e8a 1555 ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
994ab360 1556 if (ret < 0) {
56591bac 1557 PERROR("tracer ctl get_mmap_read_offset");
56591bac
MD
1558 goto end;
1559 }
f02e1e8a 1560 break;
7753dea8
MD
1561 case LTTNG_CONSUMER32_UST:
1562 case LTTNG_CONSUMER64_UST:
ffe60014
DG
1563 mmap_base = lttng_ustctl_get_mmap_base(stream);
1564 if (!mmap_base) {
1565 ERR("read mmap get mmap base for stream %s", stream->name);
994ab360 1566 ret = -EPERM;
ffe60014
DG
1567 goto end;
1568 }
1569 ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
56591bac
MD
1570 if (ret != 0) {
1571 PERROR("tracer ctl get_mmap_read_offset");
994ab360 1572 ret = -EINVAL;
56591bac
MD
1573 goto end;
1574 }
f02e1e8a 1575 break;
3bd1e081
MD
1576 default:
1577 ERR("Unknown consumer_data type");
1578 assert(0);
1579 }
b9182dd9 1580
f02e1e8a
DG
1581 /* Handle stream on the relayd if the output is on the network */
1582 if (relayd) {
1583 unsigned long netlen = len;
1584
1585 /*
1586 * Lock the control socket for the complete duration of the function
1587 * since from this point on we will use the socket.
1588 */
1589 if (stream->metadata_flag) {
1590 /* Metadata requires the control socket. */
1591 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
93ec662e
JD
1592 if (stream->reset_metadata_flag) {
1593 ret = relayd_reset_metadata(&relayd->control_sock,
1594 stream->relayd_stream_id,
1595 stream->metadata_version);
1596 if (ret < 0) {
1597 relayd_hang_up = 1;
1598 goto write_error;
1599 }
1600 stream->reset_metadata_flag = 0;
1601 }
1d4dfdef 1602 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
f02e1e8a
DG
1603 }
1604
1d4dfdef 1605 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
994ab360
DG
1606 if (ret < 0) {
1607 relayd_hang_up = 1;
1608 goto write_error;
1609 }
1610 /* Use the returned socket. */
1611 outfd = ret;
f02e1e8a 1612
994ab360
DG
1613 /* Write metadata stream id before payload */
1614 if (stream->metadata_flag) {
239f61af 1615 ret = write_relayd_metadata_id(outfd, stream, padding);
994ab360 1616 if (ret < 0) {
8994307f
DG
1617 relayd_hang_up = 1;
1618 goto write_error;
1619 }
f02e1e8a 1620 }
1d4dfdef
DG
1621 } else {
1622 /* No streaming, we have to set the len with the full padding */
1623 len += padding;
1624d5b7 1624
93ec662e
JD
1625 if (stream->metadata_flag && stream->reset_metadata_flag) {
1626 ret = utils_truncate_stream_file(stream->out_fd, 0);
1627 if (ret < 0) {
1628 ERR("Reset metadata file");
1629 goto end;
1630 }
1631 stream->reset_metadata_flag = 0;
1632 }
1633
1624d5b7
JD
1634 /*
1635 * Check if we need to change the tracefile before writing the packet.
1636 */
1637 if (stream->chan->tracefile_size > 0 &&
1638 (stream->tracefile_size_current + len) >
1639 stream->chan->tracefile_size) {
fe4477ee
JD
1640 ret = utils_rotate_stream_file(stream->chan->pathname,
1641 stream->name, stream->chan->tracefile_size,
1642 stream->chan->tracefile_count, stream->uid, stream->gid,
309167d2
JD
1643 stream->out_fd, &(stream->tracefile_count_current),
1644 &stream->out_fd);
1624d5b7
JD
1645 if (ret < 0) {
1646 ERR("Rotating output file");
1647 goto end;
1648 }
309167d2
JD
1649 outfd = stream->out_fd;
1650
f8f3885c
MD
1651 if (stream->index_file) {
1652 lttng_index_file_put(stream->index_file);
1653 stream->index_file = lttng_index_file_create(stream->chan->pathname,
309167d2
JD
1654 stream->name, stream->uid, stream->gid,
1655 stream->chan->tracefile_size,
f8f3885c
MD
1656 stream->tracefile_count_current,
1657 CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
1658 if (!stream->index_file) {
309167d2
JD
1659 goto end;
1660 }
309167d2
JD
1661 }
1662
a6976990
DG
1663 /* Reset current size because we just perform a rotation. */
1664 stream->tracefile_size_current = 0;
a1ae300f
JD
1665 stream->out_fd_offset = 0;
1666 orig_offset = 0;
1624d5b7
JD
1667 }
1668 stream->tracefile_size_current += len;
309167d2
JD
1669 if (index) {
1670 index->offset = htobe64(stream->out_fd_offset);
1671 }
f02e1e8a
DG
1672 }
1673
d02b8372
DG
1674 /*
1675 * This call guarantee that len or less is returned. It's impossible to
1676 * receive a ret value that is bigger than len.
1677 */
1678 ret = lttng_write(outfd, mmap_base + mmap_offset, len);
1679 DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
1680 if (ret < 0 || ((size_t) ret != len)) {
1681 /*
1682 * Report error to caller if nothing was written else at least send the
1683 * amount written.
1684 */
1685 if (ret < 0) {
994ab360 1686 ret = -errno;
f02e1e8a 1687 }
994ab360 1688 relayd_hang_up = 1;
f02e1e8a 1689
d02b8372 1690 /* Socket operation failed. We consider the relayd dead */
994ab360 1691 if (errno == EPIPE || errno == EINVAL || errno == EBADF) {
d02b8372
DG
1692 /*
1693 * This is possible if the fd is closed on the other side
1694 * (outfd) or any write problem. It can be verbose a bit for a
1695 * normal execution if for instance the relayd is stopped
1696 * abruptly. This can happen so set this to a DBG statement.
1697 */
1698 DBG("Consumer mmap write detected relayd hang up");
994ab360
DG
1699 } else {
1700 /* Unhandled error, print it and stop function right now. */
1701 PERROR("Error in write mmap (ret %zd != len %lu)", ret, len);
f02e1e8a 1702 }
994ab360 1703 goto write_error;
d02b8372
DG
1704 }
1705 stream->output_written += ret;
d02b8372
DG
1706
1707 /* This call is useless on a socket so better save a syscall. */
1708 if (!relayd) {
1709 /* This won't block, but will start writeout asynchronously */
1710 lttng_sync_file_range(outfd, stream->out_fd_offset, len,
1711 SYNC_FILE_RANGE_WRITE);
1712 stream->out_fd_offset += len;
f5dbe415 1713 lttng_consumer_sync_trace_file(stream, orig_offset);
f02e1e8a 1714 }
f02e1e8a 1715
8994307f
DG
1716write_error:
1717 /*
1718 * This is a special case that the relayd has closed its socket. Let's
1719 * cleanup the relayd object and all associated streams.
1720 */
1721 if (relayd && relayd_hang_up) {
1722 cleanup_relayd(relayd, ctx);
1723 }
1724
f02e1e8a
DG
1725end:
1726 /* Unlock only if ctrl socket used */
1727 if (relayd && stream->metadata_flag) {
1728 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1729 }
1730
1731 rcu_read_unlock();
994ab360 1732 return ret;
3bd1e081
MD
1733}
1734
1735/*
1736 * Splice the data from the ring buffer to the tracefile.
1737 *
79d4ffb7
DG
1738 * It must be called with the stream lock held.
1739 *
3bd1e081
MD
1740 * Returns the number of bytes spliced.
1741 */
4078b776 1742ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081 1743 struct lttng_consumer_local_data *ctx,
1d4dfdef 1744 struct lttng_consumer_stream *stream, unsigned long len,
309167d2 1745 unsigned long padding,
50adc264 1746 struct ctf_packet_index *index)
3bd1e081 1747{
f02e1e8a
DG
1748 ssize_t ret = 0, written = 0, ret_splice = 0;
1749 loff_t offset = 0;
1750 off_t orig_offset = stream->out_fd_offset;
1751 int fd = stream->wait_fd;
1752 /* Default is on the disk */
1753 int outfd = stream->out_fd;
f02e1e8a 1754 struct consumer_relayd_sock_pair *relayd = NULL;
fb3a43a9 1755 int *splice_pipe;
8994307f 1756 unsigned int relayd_hang_up = 0;
f02e1e8a 1757
3bd1e081
MD
1758 switch (consumer_data.type) {
1759 case LTTNG_CONSUMER_KERNEL:
f02e1e8a 1760 break;
7753dea8
MD
1761 case LTTNG_CONSUMER32_UST:
1762 case LTTNG_CONSUMER64_UST:
f02e1e8a 1763 /* Not supported for user space tracing */
3bd1e081
MD
1764 return -ENOSYS;
1765 default:
1766 ERR("Unknown consumer_data type");
1767 assert(0);
3bd1e081
MD
1768 }
1769
f02e1e8a
DG
1770 /* RCU lock for the relayd pointer */
1771 rcu_read_lock();
1772
1773 /* Flag that the current stream if set for network streaming. */
da009f2c 1774 if (stream->net_seq_idx != (uint64_t) -1ULL) {
f02e1e8a
DG
1775 relayd = consumer_find_relayd(stream->net_seq_idx);
1776 if (relayd == NULL) {
ad0b0d23 1777 written = -ret;
f02e1e8a
DG
1778 goto end;
1779 }
1780 }
a2361a61 1781 splice_pipe = stream->splice_pipe;
fb3a43a9 1782
f02e1e8a 1783 /* Write metadata stream id before payload */
1d4dfdef 1784 if (relayd) {
ad0b0d23 1785 unsigned long total_len = len;
f02e1e8a 1786
1d4dfdef
DG
1787 if (stream->metadata_flag) {
1788 /*
1789 * Lock the control socket for the complete duration of the function
1790 * since from this point on we will use the socket.
1791 */
1792 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1793
93ec662e
JD
1794 if (stream->reset_metadata_flag) {
1795 ret = relayd_reset_metadata(&relayd->control_sock,
1796 stream->relayd_stream_id,
1797 stream->metadata_version);
1798 if (ret < 0) {
1799 relayd_hang_up = 1;
1800 goto write_error;
1801 }
1802 stream->reset_metadata_flag = 0;
1803 }
239f61af 1804 ret = write_relayd_metadata_id(splice_pipe[1], stream,
1d4dfdef
DG
1805 padding);
1806 if (ret < 0) {
1807 written = ret;
ad0b0d23
DG
1808 relayd_hang_up = 1;
1809 goto write_error;
1d4dfdef
DG
1810 }
1811
1812 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1813 }
1814
1815 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
ad0b0d23
DG
1816 if (ret < 0) {
1817 written = ret;
1818 relayd_hang_up = 1;
1819 goto write_error;
f02e1e8a 1820 }
ad0b0d23
DG
1821 /* Use the returned socket. */
1822 outfd = ret;
1d4dfdef
DG
1823 } else {
1824 /* No streaming, we have to set the len with the full padding */
1825 len += padding;
1624d5b7 1826
93ec662e
JD
1827 if (stream->metadata_flag && stream->reset_metadata_flag) {
1828 ret = utils_truncate_stream_file(stream->out_fd, 0);
1829 if (ret < 0) {
1830 ERR("Reset metadata file");
1831 goto end;
1832 }
1833 stream->reset_metadata_flag = 0;
1834 }
1624d5b7
JD
1835 /*
1836 * Check if we need to change the tracefile before writing the packet.
1837 */
1838 if (stream->chan->tracefile_size > 0 &&
1839 (stream->tracefile_size_current + len) >
1840 stream->chan->tracefile_size) {
fe4477ee
JD
1841 ret = utils_rotate_stream_file(stream->chan->pathname,
1842 stream->name, stream->chan->tracefile_size,
1843 stream->chan->tracefile_count, stream->uid, stream->gid,
309167d2
JD
1844 stream->out_fd, &(stream->tracefile_count_current),
1845 &stream->out_fd);
1624d5b7 1846 if (ret < 0) {
ad0b0d23 1847 written = ret;
1624d5b7
JD
1848 ERR("Rotating output file");
1849 goto end;
1850 }
309167d2
JD
1851 outfd = stream->out_fd;
1852
f8f3885c
MD
1853 if (stream->index_file) {
1854 lttng_index_file_put(stream->index_file);
1855 stream->index_file = lttng_index_file_create(stream->chan->pathname,
309167d2
JD
1856 stream->name, stream->uid, stream->gid,
1857 stream->chan->tracefile_size,
f8f3885c
MD
1858 stream->tracefile_count_current,
1859 CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
1860 if (!stream->index_file) {
309167d2
JD
1861 goto end;
1862 }
309167d2
JD
1863 }
1864
a6976990
DG
1865 /* Reset current size because we just perform a rotation. */
1866 stream->tracefile_size_current = 0;
a1ae300f
JD
1867 stream->out_fd_offset = 0;
1868 orig_offset = 0;
1624d5b7
JD
1869 }
1870 stream->tracefile_size_current += len;
309167d2 1871 index->offset = htobe64(stream->out_fd_offset);
f02e1e8a
DG
1872 }
1873
1874 while (len > 0) {
1d4dfdef
DG
1875 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1876 (unsigned long)offset, len, fd, splice_pipe[1]);
fb3a43a9 1877 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
f02e1e8a
DG
1878 SPLICE_F_MOVE | SPLICE_F_MORE);
1879 DBG("splice chan to pipe, ret %zd", ret_splice);
1880 if (ret_splice < 0) {
d02b8372 1881 ret = errno;
ad0b0d23 1882 written = -ret;
d02b8372 1883 PERROR("Error in relay splice");
f02e1e8a
DG
1884 goto splice_error;
1885 }
1886
1887 /* Handle stream on the relayd if the output is on the network */
ad0b0d23
DG
1888 if (relayd && stream->metadata_flag) {
1889 size_t metadata_payload_size =
1890 sizeof(struct lttcomm_relayd_metadata_payload);
1891
1892 /* Update counter to fit the spliced data */
1893 ret_splice += metadata_payload_size;
1894 len += metadata_payload_size;
1895 /*
1896 * We do this so the return value can match the len passed as
1897 * argument to this function.
1898 */
1899 written -= metadata_payload_size;
f02e1e8a
DG
1900 }
1901
1902 /* Splice data out */
fb3a43a9 1903 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
f02e1e8a 1904 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
a2361a61
JD
1905 DBG("Consumer splice pipe to file (out_fd: %d), ret %zd",
1906 outfd, ret_splice);
f02e1e8a 1907 if (ret_splice < 0) {
d02b8372 1908 ret = errno;
ad0b0d23
DG
1909 written = -ret;
1910 relayd_hang_up = 1;
1911 goto write_error;
f02e1e8a 1912 } else if (ret_splice > len) {
d02b8372
DG
1913 /*
1914 * We don't expect this code path to be executed but you never know
1915 * so this is an extra protection agains a buggy splice().
1916 */
f02e1e8a 1917 ret = errno;
ad0b0d23 1918 written += ret_splice;
d02b8372
DG
1919 PERROR("Wrote more data than requested %zd (len: %lu)", ret_splice,
1920 len);
f02e1e8a 1921 goto splice_error;
d02b8372
DG
1922 } else {
1923 /* All good, update current len and continue. */
1924 len -= ret_splice;
f02e1e8a 1925 }
f02e1e8a
DG
1926
1927 /* This call is useless on a socket so better save a syscall. */
1928 if (!relayd) {
1929 /* This won't block, but will start writeout asynchronously */
1930 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
1931 SYNC_FILE_RANGE_WRITE);
1932 stream->out_fd_offset += ret_splice;
1933 }
e5d1a9b3 1934 stream->output_written += ret_splice;
f02e1e8a
DG
1935 written += ret_splice;
1936 }
f5dbe415
JG
1937 if (!relayd) {
1938 lttng_consumer_sync_trace_file(stream, orig_offset);
1939 }
f02e1e8a
DG
1940 goto end;
1941
8994307f
DG
1942write_error:
1943 /*
1944 * This is a special case that the relayd has closed its socket. Let's
1945 * cleanup the relayd object and all associated streams.
1946 */
1947 if (relayd && relayd_hang_up) {
1948 cleanup_relayd(relayd, ctx);
1949 /* Skip splice error so the consumer does not fail */
1950 goto end;
1951 }
1952
f02e1e8a
DG
1953splice_error:
1954 /* send the appropriate error description to sessiond */
1955 switch (ret) {
f02e1e8a 1956 case EINVAL:
f73fabfd 1957 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
f02e1e8a
DG
1958 break;
1959 case ENOMEM:
f73fabfd 1960 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
f02e1e8a
DG
1961 break;
1962 case ESPIPE:
f73fabfd 1963 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
f02e1e8a
DG
1964 break;
1965 }
1966
1967end:
1968 if (relayd && stream->metadata_flag) {
1969 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1970 }
1971
1972 rcu_read_unlock();
1973 return written;
3bd1e081
MD
1974}
1975
15055ce5
JD
1976/*
1977 * Sample the snapshot positions for a specific fd
1978 *
1979 * Returns 0 on success, < 0 on error
1980 */
1981int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
1982{
1983 switch (consumer_data.type) {
1984 case LTTNG_CONSUMER_KERNEL:
1985 return lttng_kconsumer_sample_snapshot_positions(stream);
1986 case LTTNG_CONSUMER32_UST:
1987 case LTTNG_CONSUMER64_UST:
1988 return lttng_ustconsumer_sample_snapshot_positions(stream);
1989 default:
1990 ERR("Unknown consumer_data type");
1991 assert(0);
1992 return -ENOSYS;
1993 }
1994}
3bd1e081
MD
1995/*
1996 * Take a snapshot for a specific fd
1997 *
1998 * Returns 0 on success, < 0 on error
1999 */
ffe60014 2000int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
3bd1e081
MD
2001{
2002 switch (consumer_data.type) {
2003 case LTTNG_CONSUMER_KERNEL:
ffe60014 2004 return lttng_kconsumer_take_snapshot(stream);
7753dea8
MD
2005 case LTTNG_CONSUMER32_UST:
2006 case LTTNG_CONSUMER64_UST:
ffe60014 2007 return lttng_ustconsumer_take_snapshot(stream);
3bd1e081
MD
2008 default:
2009 ERR("Unknown consumer_data type");
2010 assert(0);
2011 return -ENOSYS;
2012 }
3bd1e081
MD
2013}
2014
2015/*
2016 * Get the produced position
2017 *
2018 * Returns 0 on success, < 0 on error
2019 */
ffe60014 2020int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
3bd1e081
MD
2021 unsigned long *pos)
2022{
2023 switch (consumer_data.type) {
2024 case LTTNG_CONSUMER_KERNEL:
ffe60014 2025 return lttng_kconsumer_get_produced_snapshot(stream, pos);
7753dea8
MD
2026 case LTTNG_CONSUMER32_UST:
2027 case LTTNG_CONSUMER64_UST:
ffe60014 2028 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
3bd1e081
MD
2029 default:
2030 ERR("Unknown consumer_data type");
2031 assert(0);
2032 return -ENOSYS;
2033 }
2034}
2035
15055ce5
JD
2036/*
2037 * Get the consumed position (free-running counter position in bytes).
2038 *
2039 * Returns 0 on success, < 0 on error
2040 */
2041int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
2042 unsigned long *pos)
2043{
2044 switch (consumer_data.type) {
2045 case LTTNG_CONSUMER_KERNEL:
2046 return lttng_kconsumer_get_consumed_snapshot(stream, pos);
2047 case LTTNG_CONSUMER32_UST:
2048 case LTTNG_CONSUMER64_UST:
2049 return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
2050 default:
2051 ERR("Unknown consumer_data type");
2052 assert(0);
2053 return -ENOSYS;
2054 }
2055}
2056
3bd1e081
MD
2057int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
2058 int sock, struct pollfd *consumer_sockpoll)
2059{
2060 switch (consumer_data.type) {
2061 case LTTNG_CONSUMER_KERNEL:
2062 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
7753dea8
MD
2063 case LTTNG_CONSUMER32_UST:
2064 case LTTNG_CONSUMER64_UST:
3bd1e081
MD
2065 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
2066 default:
2067 ERR("Unknown consumer_data type");
2068 assert(0);
2069 return -ENOSYS;
2070 }
2071}
2072
6d574024 2073void lttng_consumer_close_all_metadata(void)
d88aee68
DG
2074{
2075 switch (consumer_data.type) {
2076 case LTTNG_CONSUMER_KERNEL:
2077 /*
2078 * The Kernel consumer has a different metadata scheme so we don't
2079 * close anything because the stream will be closed by the session
2080 * daemon.
2081 */
2082 break;
2083 case LTTNG_CONSUMER32_UST:
2084 case LTTNG_CONSUMER64_UST:
2085 /*
2086 * Close all metadata streams. The metadata hash table is passed and
2087 * this call iterates over it by closing all wakeup fd. This is safe
2088 * because at this point we are sure that the metadata producer is
2089 * either dead or blocked.
2090 */
6d574024 2091 lttng_ustconsumer_close_all_metadata(metadata_ht);
d88aee68
DG
2092 break;
2093 default:
2094 ERR("Unknown consumer_data type");
2095 assert(0);
2096 }
2097}
2098
fb3a43a9
DG
2099/*
2100 * Clean up a metadata stream and free its memory.
2101 */
e316aad5
DG
2102void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
2103 struct lttng_ht *ht)
fb3a43a9 2104{
e316aad5 2105 struct lttng_consumer_channel *free_chan = NULL;
fb3a43a9
DG
2106
2107 assert(stream);
2108 /*
2109 * This call should NEVER receive regular stream. It must always be
2110 * metadata stream and this is crucial for data structure synchronization.
2111 */
2112 assert(stream->metadata_flag);
2113
e316aad5
DG
2114 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
2115
74251bb8 2116 pthread_mutex_lock(&consumer_data.lock);
fb549e7a 2117 pthread_mutex_lock(&stream->chan->lock);
3dad2c0f 2118 pthread_mutex_lock(&stream->lock);
081424af
JG
2119 if (stream->chan->metadata_cache) {
2120 /* Only applicable to userspace consumers. */
2121 pthread_mutex_lock(&stream->chan->metadata_cache->lock);
2122 }
8994307f 2123
6d574024
DG
2124 /* Remove any reference to that stream. */
2125 consumer_stream_delete(stream, ht);
ca22feea 2126
6d574024
DG
2127 /* Close down everything including the relayd if one. */
2128 consumer_stream_close(stream);
2129 /* Destroy tracer buffers of the stream. */
2130 consumer_stream_destroy_buffers(stream);
fb3a43a9
DG
2131
2132 /* Atomically decrement channel refcount since other threads can use it. */
f2ad556d 2133 if (!uatomic_sub_return(&stream->chan->refcount, 1)
ffe60014 2134 && !uatomic_read(&stream->chan->nb_init_stream_left)) {
c30aaa51 2135 /* Go for channel deletion! */
e316aad5 2136 free_chan = stream->chan;
fb3a43a9
DG
2137 }
2138
73811ecc
DG
2139 /*
2140 * Nullify the stream reference so it is not used after deletion. The
6d574024
DG
2141 * channel lock MUST be acquired before being able to check for a NULL
2142 * pointer value.
73811ecc
DG
2143 */
2144 stream->chan->metadata_stream = NULL;
2145
081424af
JG
2146 if (stream->chan->metadata_cache) {
2147 pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
2148 }
3dad2c0f 2149 pthread_mutex_unlock(&stream->lock);
fb549e7a 2150 pthread_mutex_unlock(&stream->chan->lock);
74251bb8 2151 pthread_mutex_unlock(&consumer_data.lock);
e316aad5
DG
2152
2153 if (free_chan) {
2154 consumer_del_channel(free_chan);
2155 }
2156
6d574024 2157 consumer_stream_free(stream);
fb3a43a9
DG
2158}
2159
2160/*
2161 * Action done with the metadata stream when adding it to the consumer internal
2162 * data structures to handle it.
2163 */
66d583dc 2164void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
fb3a43a9 2165{
5ab66908 2166 struct lttng_ht *ht = metadata_ht;
76082088 2167 struct lttng_ht_iter iter;
d88aee68 2168 struct lttng_ht_node_u64 *node;
fb3a43a9 2169
e316aad5
DG
2170 assert(stream);
2171 assert(ht);
2172
d88aee68 2173 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
e316aad5
DG
2174
2175 pthread_mutex_lock(&consumer_data.lock);
a9838785 2176 pthread_mutex_lock(&stream->chan->lock);
ec6ea7d0 2177 pthread_mutex_lock(&stream->chan->timer_lock);
2e818a6a 2178 pthread_mutex_lock(&stream->lock);
e316aad5 2179
e316aad5
DG
2180 /*
2181 * From here, refcounts are updated so be _careful_ when returning an error
2182 * after this point.
2183 */
2184
fb3a43a9 2185 rcu_read_lock();
76082088
DG
2186
2187 /*
2188 * Lookup the stream just to make sure it does not exist in our internal
2189 * state. This should NEVER happen.
2190 */
d88aee68
DG
2191 lttng_ht_lookup(ht, &stream->key, &iter);
2192 node = lttng_ht_iter_get_node_u64(&iter);
76082088
DG
2193 assert(!node);
2194
e316aad5 2195 /*
ffe60014
DG
2196 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2197 * in terms of destroying the associated channel, because the action that
e316aad5
DG
2198 * causes the count to become 0 also causes a stream to be added. The
2199 * channel deletion will thus be triggered by the following removal of this
2200 * stream.
2201 */
ffe60014 2202 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
f2ad556d
MD
2203 /* Increment refcount before decrementing nb_init_stream_left */
2204 cmm_smp_wmb();
ffe60014 2205 uatomic_dec(&stream->chan->nb_init_stream_left);
e316aad5
DG
2206 }
2207
d88aee68 2208 lttng_ht_add_unique_u64(ht, &stream->node);
ca22feea 2209
446156b4 2210 lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
d8ef542d
MD
2211 &stream->node_channel_id);
2212
ca22feea
DG
2213 /*
2214 * Add stream to the stream_list_ht of the consumer data. No need to steal
2215 * the key since the HT does not use it and we allow to add redundant keys
2216 * into this table.
2217 */
d88aee68 2218 lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
ca22feea 2219
fb3a43a9 2220 rcu_read_unlock();
e316aad5 2221
2e818a6a 2222 pthread_mutex_unlock(&stream->lock);
a9838785 2223 pthread_mutex_unlock(&stream->chan->lock);
ec6ea7d0 2224 pthread_mutex_unlock(&stream->chan->timer_lock);
e316aad5 2225 pthread_mutex_unlock(&consumer_data.lock);
fb3a43a9
DG
2226}
2227
8994307f
DG
2228/*
2229 * Delete data stream that are flagged for deletion (endpoint_status).
2230 */
2231static void validate_endpoint_status_data_stream(void)
2232{
2233 struct lttng_ht_iter iter;
2234 struct lttng_consumer_stream *stream;
2235
2236 DBG("Consumer delete flagged data stream");
2237
2238 rcu_read_lock();
2239 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2240 /* Validate delete flag of the stream */
79d4ffb7 2241 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2242 continue;
2243 }
2244 /* Delete it right now */
2245 consumer_del_stream(stream, data_ht);
2246 }
2247 rcu_read_unlock();
2248}
2249
2250/*
2251 * Delete metadata stream that are flagged for deletion (endpoint_status).
2252 */
2253static void validate_endpoint_status_metadata_stream(
2254 struct lttng_poll_event *pollset)
2255{
2256 struct lttng_ht_iter iter;
2257 struct lttng_consumer_stream *stream;
2258
2259 DBG("Consumer delete flagged metadata stream");
2260
2261 assert(pollset);
2262
2263 rcu_read_lock();
2264 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2265 /* Validate delete flag of the stream */
79d4ffb7 2266 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
8994307f
DG
2267 continue;
2268 }
2269 /*
2270 * Remove from pollset so the metadata thread can continue without
2271 * blocking on a deleted stream.
2272 */
2273 lttng_poll_del(pollset, stream->wait_fd);
2274
2275 /* Delete it right now */
2276 consumer_del_metadata_stream(stream, metadata_ht);
2277 }
2278 rcu_read_unlock();
2279}
2280
d73bf3d7
JD
2281static
2282int rotate_notify_sessiond(struct lttng_consumer_local_data *ctx,
2283 uint64_t key)
2284{
2285 ssize_t ret;
2286
2287 do {
2288 ret = write(ctx->channel_rotate_pipe, &key, sizeof(key));
2289 } while (ret == -1 && errno == EINTR);
2290 if (ret == -1) {
2291 PERROR("Failed to write to the channel rotation pipe");
2292 } else {
2293 DBG("Sent channel rotation notification for channel key %"
2294 PRIu64, key);
2295 ret = 0;
2296 }
2297
2298 return (int) ret;
2299}
2300
2301/*
2302 * Perform operations that need to be done after a stream has
2303 * rotated and released the stream lock.
2304 *
2305 * Multiple rotations cannot occur simultaneously, so we know the state of the
2306 * "rotated" stream flag cannot change.
2307 *
2308 * This MUST be called WITHOUT the stream lock held.
2309 */
2310static
2311int consumer_post_rotation(struct lttng_consumer_stream *stream,
2312 struct lttng_consumer_local_data *ctx)
2313{
2314 int ret = 0;
2315
2316 pthread_mutex_lock(&stream->chan->lock);
2317
2318 switch (consumer_data.type) {
2319 case LTTNG_CONSUMER_KERNEL:
2320 break;
2321 case LTTNG_CONSUMER32_UST:
2322 case LTTNG_CONSUMER64_UST:
2323 /*
2324 * The ust_metadata_pushed counter has been reset to 0, so now
2325 * we can wakeup the metadata thread so it dumps the metadata
2326 * cache to the new file.
2327 */
2328 if (stream->metadata_flag) {
2329 consumer_metadata_wakeup_pipe(stream->chan);
2330 }
2331 break;
2332 default:
2333 ERR("Unknown consumer_data type");
2334 abort();
2335 }
2336
2337 if (--stream->chan->nr_stream_rotate_pending == 0) {
2338 DBG("Rotation of channel \"%s\" completed, notifying the session daemon",
2339 stream->chan->name);
2340 ret = rotate_notify_sessiond(ctx, stream->chan->key);
2341 }
2342 assert(stream->chan->nr_stream_rotate_pending >= 0);
2343 pthread_mutex_unlock(&stream->chan->lock);
2344
2345 return ret;
2346}
2347
fb3a43a9
DG
2348/*
2349 * Thread polls on metadata file descriptor and write them on disk or on the
2350 * network.
2351 */
7d980def 2352void *consumer_thread_metadata_poll(void *data)
fb3a43a9 2353{
1fc79fb4 2354 int ret, i, pollfd, err = -1;
fb3a43a9 2355 uint32_t revents, nb_fd;
e316aad5 2356 struct lttng_consumer_stream *stream = NULL;
fb3a43a9 2357 struct lttng_ht_iter iter;
d88aee68 2358 struct lttng_ht_node_u64 *node;
fb3a43a9
DG
2359 struct lttng_poll_event events;
2360 struct lttng_consumer_local_data *ctx = data;
2361 ssize_t len;
2362
2363 rcu_register_thread();
2364
1fc79fb4
MD
2365 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA);
2366
2d57de81
MD
2367 if (testpoint(consumerd_thread_metadata)) {
2368 goto error_testpoint;
2369 }
2370
9ce5646a
MD
2371 health_code_update();
2372
fb3a43a9
DG
2373 DBG("Thread metadata poll started");
2374
fb3a43a9
DG
2375 /* Size is set to 1 for the consumer_metadata pipe */
2376 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2377 if (ret < 0) {
2378 ERR("Poll set creation failed");
d8ef542d 2379 goto end_poll;
fb3a43a9
DG
2380 }
2381
13886d2d
DG
2382 ret = lttng_poll_add(&events,
2383 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
fb3a43a9
DG
2384 if (ret < 0) {
2385 goto end;
2386 }
2387
2388 /* Main loop */
2389 DBG("Metadata main loop started");
2390
2391 while (1) {
fb3a43a9 2392restart:
7fa2082e 2393 health_code_update();
9ce5646a 2394 health_poll_entry();
7fa2082e 2395 DBG("Metadata poll wait");
fb3a43a9 2396 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
2397 DBG("Metadata poll return from wait with %d fd(s)",
2398 LTTNG_POLL_GETNB(&events));
9ce5646a 2399 health_poll_exit();
40063ead 2400 DBG("Metadata event caught in thread");
fb3a43a9
DG
2401 if (ret < 0) {
2402 if (errno == EINTR) {
40063ead 2403 ERR("Poll EINTR caught");
fb3a43a9
DG
2404 goto restart;
2405 }
d9607cd7
MD
2406 if (LTTNG_POLL_GETNB(&events) == 0) {
2407 err = 0; /* All is OK */
2408 }
2409 goto end;
fb3a43a9
DG
2410 }
2411
0d9c5d77
DG
2412 nb_fd = ret;
2413
e316aad5 2414 /* From here, the event is a metadata wait fd */
fb3a43a9 2415 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2416 health_code_update();
2417
fb3a43a9
DG
2418 revents = LTTNG_POLL_GETEV(&events, i);
2419 pollfd = LTTNG_POLL_GETFD(&events, i);
2420
fd20dac9
MD
2421 if (!revents) {
2422 /* No activity for this FD (poll implementation). */
2423 continue;
2424 }
2425
13886d2d 2426 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
03e43155 2427 if (revents & LPOLLIN) {
13886d2d
DG
2428 ssize_t pipe_len;
2429
2430 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2431 &stream, sizeof(stream));
6cd525e8 2432 if (pipe_len < sizeof(stream)) {
03e43155
MD
2433 if (pipe_len < 0) {
2434 PERROR("read metadata stream");
2435 }
fb3a43a9 2436 /*
03e43155
MD
2437 * Remove the pipe from the poll set and continue the loop
2438 * since their might be data to consume.
fb3a43a9 2439 */
03e43155
MD
2440 lttng_poll_del(&events,
2441 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2442 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
fb3a43a9
DG
2443 continue;
2444 }
2445
8994307f
DG
2446 /* A NULL stream means that the state has changed. */
2447 if (stream == NULL) {
2448 /* Check for deleted streams. */
2449 validate_endpoint_status_metadata_stream(&events);
3714380f 2450 goto restart;
8994307f
DG
2451 }
2452
fb3a43a9
DG
2453 DBG("Adding metadata stream %d to poll set",
2454 stream->wait_fd);
2455
fb3a43a9
DG
2456 /* Add metadata stream to the global poll events list */
2457 lttng_poll_add(&events, stream->wait_fd,
6d574024 2458 LPOLLIN | LPOLLPRI | LPOLLHUP);
03e43155
MD
2459 } else if (revents & (LPOLLERR | LPOLLHUP)) {
2460 DBG("Metadata thread pipe hung up");
2461 /*
2462 * Remove the pipe from the poll set and continue the loop
2463 * since their might be data to consume.
2464 */
2465 lttng_poll_del(&events,
2466 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2467 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2468 continue;
2469 } else {
2470 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2471 goto end;
fb3a43a9
DG
2472 }
2473
e316aad5 2474 /* Handle other stream */
fb3a43a9
DG
2475 continue;
2476 }
2477
d09e1200 2478 rcu_read_lock();
d88aee68
DG
2479 {
2480 uint64_t tmp_id = (uint64_t) pollfd;
2481
2482 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2483 }
2484 node = lttng_ht_iter_get_node_u64(&iter);
e316aad5 2485 assert(node);
fb3a43a9
DG
2486
2487 stream = caa_container_of(node, struct lttng_consumer_stream,
58b1f425 2488 node);
fb3a43a9 2489
03e43155
MD
2490 if (revents & (LPOLLIN | LPOLLPRI)) {
2491 /* Get the data out of the metadata file descriptor */
2492 DBG("Metadata available on fd %d", pollfd);
2493 assert(stream->wait_fd == pollfd);
2494
2495 do {
2496 health_code_update();
2497
2498 len = ctx->on_buffer_ready(stream, ctx);
2499 /*
2500 * We don't check the return value here since if we get
83f4233d 2501 * a negative len, it means an error occurred thus we
03e43155
MD
2502 * simply remove it from the poll set and free the
2503 * stream.
2504 */
2505 } while (len > 0);
2506
2507 /* It's ok to have an unavailable sub-buffer */
2508 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2509 /* Clean up stream from consumer and free it. */
2510 lttng_poll_del(&events, stream->wait_fd);
2511 consumer_del_metadata_stream(stream, metadata_ht);
2512 }
2513 } else if (revents & (LPOLLERR | LPOLLHUP)) {
e316aad5 2514 DBG("Metadata fd %d is hup|err.", pollfd);
fb3a43a9
DG
2515 if (!stream->hangup_flush_done
2516 && (consumer_data.type == LTTNG_CONSUMER32_UST
2517 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2518 DBG("Attempting to flush and consume the UST buffers");
2519 lttng_ustconsumer_on_stream_hangup(stream);
2520
2521 /* We just flushed the stream now read it. */
4bb94b75 2522 do {
9ce5646a
MD
2523 health_code_update();
2524
4bb94b75
DG
2525 len = ctx->on_buffer_ready(stream, ctx);
2526 /*
2527 * We don't check the return value here since if we get
83f4233d 2528 * a negative len, it means an error occurred thus we
4bb94b75
DG
2529 * simply remove it from the poll set and free the
2530 * stream.
2531 */
2532 } while (len > 0);
fb3a43a9
DG
2533 }
2534
fb3a43a9 2535 lttng_poll_del(&events, stream->wait_fd);
e316aad5
DG
2536 /*
2537 * This call update the channel states, closes file descriptors
2538 * and securely free the stream.
2539 */
2540 consumer_del_metadata_stream(stream, metadata_ht);
03e43155
MD
2541 } else {
2542 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
6f2f1a70 2543 rcu_read_unlock();
03e43155 2544 goto end;
fb3a43a9 2545 }
e316aad5 2546 /* Release RCU lock for the stream looked up */
d09e1200 2547 rcu_read_unlock();
fb3a43a9
DG
2548 }
2549 }
2550
1fc79fb4
MD
2551 /* All is OK */
2552 err = 0;
fb3a43a9
DG
2553end:
2554 DBG("Metadata poll thread exiting");
fb3a43a9 2555
d8ef542d
MD
2556 lttng_poll_clean(&events);
2557end_poll:
2d57de81 2558error_testpoint:
1fc79fb4
MD
2559 if (err) {
2560 health_error();
2561 ERR("Health error occurred in %s", __func__);
2562 }
2563 health_unregister(health_consumerd);
fb3a43a9
DG
2564 rcu_unregister_thread();
2565 return NULL;
2566}
2567
3bd1e081 2568/*
e4421fec 2569 * This thread polls the fds in the set to consume the data and write
3bd1e081
MD
2570 * it to tracefile if necessary.
2571 */
7d980def 2572void *consumer_thread_data_poll(void *data)
3bd1e081 2573{
1fc79fb4 2574 int num_rdy, num_hup, high_prio, ret, i, err = -1;
3bd1e081
MD
2575 struct pollfd *pollfd = NULL;
2576 /* local view of the streams */
c869f647 2577 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
3bd1e081 2578 /* local view of consumer_data.fds_count */
261de637 2579 int nb_fd = 0, nb_pipes_fd;
9a2fcf78
JD
2580 /* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
2581 int nb_inactive_fd = 0;
3bd1e081 2582 struct lttng_consumer_local_data *ctx = data;
00e2e675 2583 ssize_t len;
3bd1e081 2584
e7b994a3
DG
2585 rcu_register_thread();
2586
1fc79fb4
MD
2587 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA);
2588
2d57de81
MD
2589 if (testpoint(consumerd_thread_data)) {
2590 goto error_testpoint;
2591 }
2592
9ce5646a
MD
2593 health_code_update();
2594
4df6c8cb
MD
2595 local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
2596 if (local_stream == NULL) {
2597 PERROR("local_stream malloc");
2598 goto end;
2599 }
3bd1e081
MD
2600
2601 while (1) {
9ce5646a
MD
2602 health_code_update();
2603
3bd1e081
MD
2604 high_prio = 0;
2605 num_hup = 0;
2606
2607 /*
e4421fec 2608 * the fds set has been updated, we need to update our
3bd1e081
MD
2609 * local array as well
2610 */
2611 pthread_mutex_lock(&consumer_data.lock);
2612 if (consumer_data.need_update) {
0e428499
DG
2613 free(pollfd);
2614 pollfd = NULL;
2615
2616 free(local_stream);
2617 local_stream = NULL;
3bd1e081 2618
02b3d176 2619 /*
261de637
JD
2620 * Allocate for all fds + 2:
2621 * +1 for the consumer_data_pipe
2622 * +1 for wake up pipe
02b3d176 2623 */
261de637
JD
2624 nb_pipes_fd = 2;
2625 pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
3bd1e081 2626 if (pollfd == NULL) {
7a57cf92 2627 PERROR("pollfd malloc");
3bd1e081
MD
2628 pthread_mutex_unlock(&consumer_data.lock);
2629 goto end;
2630 }
2631
261de637 2632 local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
747f8642 2633 sizeof(struct lttng_consumer_stream *));
3bd1e081 2634 if (local_stream == NULL) {
7a57cf92 2635 PERROR("local_stream malloc");
3bd1e081
MD
2636 pthread_mutex_unlock(&consumer_data.lock);
2637 goto end;
2638 }
ffe60014 2639 ret = update_poll_array(ctx, &pollfd, local_stream,
9a2fcf78 2640 data_ht, &nb_inactive_fd);
3bd1e081
MD
2641 if (ret < 0) {
2642 ERR("Error in allocating pollfd or local_outfds");
f73fabfd 2643 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2644 pthread_mutex_unlock(&consumer_data.lock);
2645 goto end;
2646 }
2647 nb_fd = ret;
2648 consumer_data.need_update = 0;
2649 }
2650 pthread_mutex_unlock(&consumer_data.lock);
2651
4078b776 2652 /* No FDs and consumer_quit, consumer_cleanup the thread */
9a2fcf78
JD
2653 if (nb_fd == 0 && nb_inactive_fd == 0 &&
2654 CMM_LOAD_SHARED(consumer_quit) == 1) {
1fc79fb4 2655 err = 0; /* All is OK */
4078b776
MD
2656 goto end;
2657 }
3bd1e081 2658 /* poll on the array of fds */
88f2b785 2659 restart:
261de637 2660 DBG("polling on %d fd", nb_fd + nb_pipes_fd);
cf0bcb51
JG
2661 if (testpoint(consumerd_thread_data_poll)) {
2662 goto end;
2663 }
9ce5646a 2664 health_poll_entry();
261de637 2665 num_rdy = poll(pollfd, nb_fd + nb_pipes_fd, -1);
9ce5646a 2666 health_poll_exit();
3bd1e081
MD
2667 DBG("poll num_rdy : %d", num_rdy);
2668 if (num_rdy == -1) {
88f2b785
MD
2669 /*
2670 * Restart interrupted system call.
2671 */
2672 if (errno == EINTR) {
2673 goto restart;
2674 }
7a57cf92 2675 PERROR("Poll error");
f73fabfd 2676 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3bd1e081
MD
2677 goto end;
2678 } else if (num_rdy == 0) {
2679 DBG("Polling thread timed out");
2680 goto end;
2681 }
2682
80957876
JG
2683 if (caa_unlikely(data_consumption_paused)) {
2684 DBG("Data consumption paused, sleeping...");
2685 sleep(1);
2686 goto restart;
2687 }
2688
3bd1e081 2689 /*
50f8ae69 2690 * If the consumer_data_pipe triggered poll go directly to the
00e2e675
DG
2691 * beginning of the loop to update the array. We want to prioritize
2692 * array update over low-priority reads.
3bd1e081 2693 */
509bb1cf 2694 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
ab30f567 2695 ssize_t pipe_readlen;
04fdd819 2696
50f8ae69 2697 DBG("consumer_data_pipe wake up");
acdb9057
DG
2698 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2699 &new_stream, sizeof(new_stream));
6cd525e8
MD
2700 if (pipe_readlen < sizeof(new_stream)) {
2701 PERROR("Consumer data pipe");
23f5f35d
DG
2702 /* Continue so we can at least handle the current stream(s). */
2703 continue;
2704 }
c869f647
DG
2705
2706 /*
2707 * If the stream is NULL, just ignore it. It's also possible that
2708 * the sessiond poll thread changed the consumer_quit state and is
2709 * waking us up to test it.
2710 */
2711 if (new_stream == NULL) {
8994307f 2712 validate_endpoint_status_data_stream();
c869f647
DG
2713 continue;
2714 }
2715
c869f647 2716 /* Continue to update the local streams and handle prio ones */
3bd1e081
MD
2717 continue;
2718 }
2719
02b3d176
DG
2720 /* Handle wakeup pipe. */
2721 if (pollfd[nb_fd + 1].revents & (POLLIN | POLLPRI)) {
2722 char dummy;
2723 ssize_t pipe_readlen;
2724
2725 pipe_readlen = lttng_pipe_read(ctx->consumer_wakeup_pipe, &dummy,
2726 sizeof(dummy));
2727 if (pipe_readlen < 0) {
2728 PERROR("Consumer data wakeup pipe");
2729 }
2730 /* We've been awakened to handle stream(s). */
2731 ctx->has_wakeup = 0;
2732 }
2733
3bd1e081
MD
2734 /* Take care of high priority channels first. */
2735 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2736 health_code_update();
2737
9617607b
DG
2738 if (local_stream[i] == NULL) {
2739 continue;
2740 }
fb3a43a9 2741 if (pollfd[i].revents & POLLPRI) {
d41f73b7
MD
2742 DBG("Urgent read on fd %d", pollfd[i].fd);
2743 high_prio = 1;
4078b776 2744 len = ctx->on_buffer_ready(local_stream[i], ctx);
d41f73b7 2745 /* it's ok to have an unavailable sub-buffer */
b64403e3 2746 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2747 /* Clean the stream and free it. */
2748 consumer_del_stream(local_stream[i], data_ht);
9617607b 2749 local_stream[i] = NULL;
4078b776
MD
2750 } else if (len > 0) {
2751 local_stream[i]->data_read = 1;
d41f73b7 2752 }
3bd1e081
MD
2753 }
2754 }
2755
4078b776
MD
2756 /*
2757 * If we read high prio channel in this loop, try again
2758 * for more high prio data.
2759 */
2760 if (high_prio) {
3bd1e081
MD
2761 continue;
2762 }
2763
2764 /* Take care of low priority channels. */
4078b776 2765 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2766 health_code_update();
2767
9617607b
DG
2768 if (local_stream[i] == NULL) {
2769 continue;
2770 }
4078b776 2771 if ((pollfd[i].revents & POLLIN) ||
02b3d176
DG
2772 local_stream[i]->hangup_flush_done ||
2773 local_stream[i]->has_data) {
4078b776
MD
2774 DBG("Normal read on fd %d", pollfd[i].fd);
2775 len = ctx->on_buffer_ready(local_stream[i], ctx);
2776 /* it's ok to have an unavailable sub-buffer */
b64403e3 2777 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
ab1027f4
DG
2778 /* Clean the stream and free it. */
2779 consumer_del_stream(local_stream[i], data_ht);
9617607b 2780 local_stream[i] = NULL;
4078b776
MD
2781 } else if (len > 0) {
2782 local_stream[i]->data_read = 1;
2783 }
2784 }
2785 }
2786
2787 /* Handle hangup and errors */
2788 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
2789 health_code_update();
2790
9617607b
DG
2791 if (local_stream[i] == NULL) {
2792 continue;
2793 }
4078b776
MD
2794 if (!local_stream[i]->hangup_flush_done
2795 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2796 && (consumer_data.type == LTTNG_CONSUMER32_UST
2797 || consumer_data.type == LTTNG_CONSUMER64_UST)) {
2798 DBG("fd %d is hup|err|nval. Attempting flush and read.",
9617607b 2799 pollfd[i].fd);
4078b776
MD
2800 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2801 /* Attempt read again, for the data we just flushed. */
2802 local_stream[i]->data_read = 1;
2803 }
2804 /*
2805 * If the poll flag is HUP/ERR/NVAL and we have
2806 * read no data in this pass, we can remove the
2807 * stream from its hash table.
2808 */
2809 if ((pollfd[i].revents & POLLHUP)) {
2810 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2811 if (!local_stream[i]->data_read) {
43c34bc3 2812 consumer_del_stream(local_stream[i], data_ht);
9617607b 2813 local_stream[i] = NULL;
4078b776
MD
2814 num_hup++;
2815 }
2816 } else if (pollfd[i].revents & POLLERR) {
2817 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2818 if (!local_stream[i]->data_read) {
43c34bc3 2819 consumer_del_stream(local_stream[i], data_ht);
9617607b 2820 local_stream[i] = NULL;
4078b776
MD
2821 num_hup++;
2822 }
2823 } else if (pollfd[i].revents & POLLNVAL) {
2824 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2825 if (!local_stream[i]->data_read) {
43c34bc3 2826 consumer_del_stream(local_stream[i], data_ht);
9617607b 2827 local_stream[i] = NULL;
4078b776 2828 num_hup++;
3bd1e081
MD
2829 }
2830 }
9617607b
DG
2831 if (local_stream[i] != NULL) {
2832 local_stream[i]->data_read = 0;
2833 }
3bd1e081
MD
2834 }
2835 }
1fc79fb4
MD
2836 /* All is OK */
2837 err = 0;
3bd1e081
MD
2838end:
2839 DBG("polling thread exiting");
0e428499
DG
2840 free(pollfd);
2841 free(local_stream);
fb3a43a9
DG
2842
2843 /*
2844 * Close the write side of the pipe so epoll_wait() in
7d980def
DG
2845 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2846 * read side of the pipe. If we close them both, epoll_wait strangely does
2847 * not return and could create a endless wait period if the pipe is the
2848 * only tracked fd in the poll set. The thread will take care of closing
2849 * the read side.
fb3a43a9 2850 */
13886d2d 2851 (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
fb3a43a9 2852
2d57de81 2853error_testpoint:
1fc79fb4
MD
2854 if (err) {
2855 health_error();
2856 ERR("Health error occurred in %s", __func__);
2857 }
2858 health_unregister(health_consumerd);
2859
e7b994a3 2860 rcu_unregister_thread();
3bd1e081
MD
2861 return NULL;
2862}
2863
d8ef542d
MD
2864/*
2865 * Close wake-up end of each stream belonging to the channel. This will
2866 * allow the poll() on the stream read-side to detect when the
2867 * write-side (application) finally closes them.
2868 */
2869static
2870void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
2871{
2872 struct lttng_ht *ht;
2873 struct lttng_consumer_stream *stream;
2874 struct lttng_ht_iter iter;
2875
2876 ht = consumer_data.stream_per_chan_id_ht;
2877
2878 rcu_read_lock();
2879 cds_lfht_for_each_entry_duplicate(ht->ht,
2880 ht->hash_fct(&channel->key, lttng_ht_seed),
2881 ht->match_fct, &channel->key,
2882 &iter.iter, stream, node_channel_id.node) {
f2ad556d
MD
2883 /*
2884 * Protect against teardown with mutex.
2885 */
2886 pthread_mutex_lock(&stream->lock);
2887 if (cds_lfht_is_node_deleted(&stream->node.node)) {
2888 goto next;
2889 }
d8ef542d
MD
2890 switch (consumer_data.type) {
2891 case LTTNG_CONSUMER_KERNEL:
2892 break;
2893 case LTTNG_CONSUMER32_UST:
2894 case LTTNG_CONSUMER64_UST:
b4a650f3
DG
2895 if (stream->metadata_flag) {
2896 /* Safe and protected by the stream lock. */
2897 lttng_ustconsumer_close_metadata(stream->chan);
2898 } else {
2899 /*
2900 * Note: a mutex is taken internally within
2901 * liblttng-ust-ctl to protect timer wakeup_fd
2902 * use from concurrent close.
2903 */
2904 lttng_ustconsumer_close_stream_wakeup(stream);
2905 }
d8ef542d
MD
2906 break;
2907 default:
2908 ERR("Unknown consumer_data type");
2909 assert(0);
2910 }
f2ad556d
MD
2911 next:
2912 pthread_mutex_unlock(&stream->lock);
d8ef542d
MD
2913 }
2914 rcu_read_unlock();
2915}
2916
2917static void destroy_channel_ht(struct lttng_ht *ht)
2918{
2919 struct lttng_ht_iter iter;
2920 struct lttng_consumer_channel *channel;
2921 int ret;
2922
2923 if (ht == NULL) {
2924 return;
2925 }
2926
2927 rcu_read_lock();
2928 cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
2929 ret = lttng_ht_del(ht, &iter);
2930 assert(ret != 0);
2931 }
2932 rcu_read_unlock();
2933
2934 lttng_ht_destroy(ht);
2935}
2936
2937/*
2938 * This thread polls the channel fds to detect when they are being
2939 * closed. It closes all related streams if the channel is detected as
2940 * closed. It is currently only used as a shim layer for UST because the
2941 * consumerd needs to keep the per-stream wakeup end of pipes open for
2942 * periodical flush.
2943 */
2944void *consumer_thread_channel_poll(void *data)
2945{
1fc79fb4 2946 int ret, i, pollfd, err = -1;
d8ef542d
MD
2947 uint32_t revents, nb_fd;
2948 struct lttng_consumer_channel *chan = NULL;
2949 struct lttng_ht_iter iter;
2950 struct lttng_ht_node_u64 *node;
2951 struct lttng_poll_event events;
2952 struct lttng_consumer_local_data *ctx = data;
2953 struct lttng_ht *channel_ht;
2954
2955 rcu_register_thread();
2956
1fc79fb4
MD
2957 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL);
2958
2d57de81
MD
2959 if (testpoint(consumerd_thread_channel)) {
2960 goto error_testpoint;
2961 }
2962
9ce5646a
MD
2963 health_code_update();
2964
d8ef542d
MD
2965 channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2966 if (!channel_ht) {
2967 /* ENOMEM at this point. Better to bail out. */
2968 goto end_ht;
2969 }
2970
2971 DBG("Thread channel poll started");
2972
2973 /* Size is set to 1 for the consumer_channel pipe */
2974 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2975 if (ret < 0) {
2976 ERR("Poll set creation failed");
2977 goto end_poll;
2978 }
2979
2980 ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
2981 if (ret < 0) {
2982 goto end;
2983 }
2984
2985 /* Main loop */
2986 DBG("Channel main loop started");
2987
2988 while (1) {
d8ef542d 2989restart:
7fa2082e
MD
2990 health_code_update();
2991 DBG("Channel poll wait");
9ce5646a 2992 health_poll_entry();
d8ef542d 2993 ret = lttng_poll_wait(&events, -1);
7fa2082e
MD
2994 DBG("Channel poll return from wait with %d fd(s)",
2995 LTTNG_POLL_GETNB(&events));
9ce5646a 2996 health_poll_exit();
40063ead 2997 DBG("Channel event caught in thread");
d8ef542d
MD
2998 if (ret < 0) {
2999 if (errno == EINTR) {
40063ead 3000 ERR("Poll EINTR caught");
d8ef542d
MD
3001 goto restart;
3002 }
d9607cd7
MD
3003 if (LTTNG_POLL_GETNB(&events) == 0) {
3004 err = 0; /* All is OK */
3005 }
d8ef542d
MD
3006 goto end;
3007 }
3008
3009 nb_fd = ret;
3010
3011 /* From here, the event is a channel wait fd */
3012 for (i = 0; i < nb_fd; i++) {
9ce5646a
MD
3013 health_code_update();
3014
d8ef542d
MD
3015 revents = LTTNG_POLL_GETEV(&events, i);
3016 pollfd = LTTNG_POLL_GETFD(&events, i);
3017
d8ef542d 3018 if (!revents) {
fd20dac9 3019 /* No activity for this FD (poll implementation). */
d8ef542d
MD
3020 continue;
3021 }
fd20dac9 3022
d8ef542d 3023 if (pollfd == ctx->consumer_channel_pipe[0]) {
03e43155 3024 if (revents & LPOLLIN) {
d8ef542d 3025 enum consumer_channel_action action;
a0cbdd2e 3026 uint64_t key;
d8ef542d 3027
a0cbdd2e 3028 ret = read_channel_pipe(ctx, &chan, &key, &action);
d8ef542d 3029 if (ret <= 0) {
03e43155
MD
3030 if (ret < 0) {
3031 ERR("Error reading channel pipe");
3032 }
3033 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
d8ef542d
MD
3034 continue;
3035 }
3036
3037 switch (action) {
3038 case CONSUMER_CHANNEL_ADD:
3039 DBG("Adding channel %d to poll set",
3040 chan->wait_fd);
3041
3042 lttng_ht_node_init_u64(&chan->wait_fd_node,
3043 chan->wait_fd);
c7260a81 3044 rcu_read_lock();
d8ef542d
MD
3045 lttng_ht_add_unique_u64(channel_ht,
3046 &chan->wait_fd_node);
c7260a81 3047 rcu_read_unlock();
d8ef542d
MD
3048 /* Add channel to the global poll events list */
3049 lttng_poll_add(&events, chan->wait_fd,
03e43155 3050 LPOLLERR | LPOLLHUP);
d8ef542d 3051 break;
a0cbdd2e
MD
3052 case CONSUMER_CHANNEL_DEL:
3053 {
b4a650f3
DG
3054 /*
3055 * This command should never be called if the channel
3056 * has streams monitored by either the data or metadata
3057 * thread. The consumer only notify this thread with a
3058 * channel del. command if it receives a destroy
3059 * channel command from the session daemon that send it
3060 * if a command prior to the GET_CHANNEL failed.
3061 */
3062
c7260a81 3063 rcu_read_lock();
a0cbdd2e
MD
3064 chan = consumer_find_channel(key);
3065 if (!chan) {
c7260a81 3066 rcu_read_unlock();
a0cbdd2e
MD
3067 ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
3068 break;
3069 }
3070 lttng_poll_del(&events, chan->wait_fd);
f623cc0b 3071 iter.iter.node = &chan->wait_fd_node.node;
a0cbdd2e
MD
3072 ret = lttng_ht_del(channel_ht, &iter);
3073 assert(ret == 0);
a0cbdd2e 3074
f2a444f1
DG
3075 switch (consumer_data.type) {
3076 case LTTNG_CONSUMER_KERNEL:
3077 break;
3078 case LTTNG_CONSUMER32_UST:
3079 case LTTNG_CONSUMER64_UST:
212d67a2
DG
3080 health_code_update();
3081 /* Destroy streams that might have been left in the stream list. */
3082 clean_channel_stream_list(chan);
f2a444f1
DG
3083 break;
3084 default:
3085 ERR("Unknown consumer_data type");
3086 assert(0);
3087 }
3088
a0cbdd2e
MD
3089 /*
3090 * Release our own refcount. Force channel deletion even if
3091 * streams were not initialized.
3092 */
3093 if (!uatomic_sub_return(&chan->refcount, 1)) {
3094 consumer_del_channel(chan);
3095 }
c7260a81 3096 rcu_read_unlock();
a0cbdd2e
MD
3097 goto restart;
3098 }
d8ef542d
MD
3099 case CONSUMER_CHANNEL_QUIT:
3100 /*
3101 * Remove the pipe from the poll set and continue the loop
3102 * since their might be data to consume.
3103 */
3104 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3105 continue;
3106 default:
3107 ERR("Unknown action");
3108 break;
3109 }
03e43155
MD
3110 } else if (revents & (LPOLLERR | LPOLLHUP)) {
3111 DBG("Channel thread pipe hung up");
3112 /*
3113 * Remove the pipe from the poll set and continue the loop
3114 * since their might be data to consume.
3115 */
3116 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3117 continue;
3118 } else {
3119 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3120 goto end;
d8ef542d
MD
3121 }
3122
3123 /* Handle other stream */
3124 continue;
3125 }
3126
3127 rcu_read_lock();
3128 {
3129 uint64_t tmp_id = (uint64_t) pollfd;
3130
3131 lttng_ht_lookup(channel_ht, &tmp_id, &iter);
3132 }
3133 node = lttng_ht_iter_get_node_u64(&iter);
3134 assert(node);
3135
3136 chan = caa_container_of(node, struct lttng_consumer_channel,
3137 wait_fd_node);
3138
3139 /* Check for error event */
3140 if (revents & (LPOLLERR | LPOLLHUP)) {
3141 DBG("Channel fd %d is hup|err.", pollfd);
3142
3143 lttng_poll_del(&events, chan->wait_fd);
3144 ret = lttng_ht_del(channel_ht, &iter);
3145 assert(ret == 0);
b4a650f3
DG
3146
3147 /*
3148 * This will close the wait fd for each stream associated to
3149 * this channel AND monitored by the data/metadata thread thus
3150 * will be clean by the right thread.
3151 */
d8ef542d 3152 consumer_close_channel_streams(chan);
f2ad556d
MD
3153
3154 /* Release our own refcount */
3155 if (!uatomic_sub_return(&chan->refcount, 1)
3156 && !uatomic_read(&chan->nb_init_stream_left)) {
3157 consumer_del_channel(chan);
3158 }
03e43155
MD
3159 } else {
3160 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3161 rcu_read_unlock();
3162 goto end;
d8ef542d
MD
3163 }
3164
3165 /* Release RCU lock for the channel looked up */
3166 rcu_read_unlock();
3167 }
3168 }
3169
1fc79fb4
MD
3170 /* All is OK */
3171 err = 0;
d8ef542d
MD
3172end:
3173 lttng_poll_clean(&events);
3174end_poll:
3175 destroy_channel_ht(channel_ht);
3176end_ht:
2d57de81 3177error_testpoint:
d8ef542d 3178 DBG("Channel poll thread exiting");
1fc79fb4
MD
3179 if (err) {
3180 health_error();
3181 ERR("Health error occurred in %s", __func__);
3182 }
3183 health_unregister(health_consumerd);
d8ef542d
MD
3184 rcu_unregister_thread();
3185 return NULL;
3186}
3187
331744e3
JD
3188static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
3189 struct pollfd *sockpoll, int client_socket)
3190{
3191 int ret;
3192
3193 assert(ctx);
3194 assert(sockpoll);
3195
84382d49
MD
3196 ret = lttng_consumer_poll_socket(sockpoll);
3197 if (ret) {
331744e3
JD
3198 goto error;
3199 }
3200 DBG("Metadata connection on client_socket");
3201
3202 /* Blocking call, waiting for transmission */
3203 ctx->consumer_metadata_socket = lttcomm_accept_unix_sock(client_socket);
3204 if (ctx->consumer_metadata_socket < 0) {
3205 WARN("On accept metadata");
3206 ret = -1;
3207 goto error;
3208 }
3209 ret = 0;
3210
3211error:
3212 return ret;
3213}
3214
3bd1e081
MD
3215/*
3216 * This thread listens on the consumerd socket and receives the file
3217 * descriptors from the session daemon.
3218 */
7d980def 3219void *consumer_thread_sessiond_poll(void *data)
3bd1e081 3220{
1fc79fb4 3221 int sock = -1, client_socket, ret, err = -1;
3bd1e081
MD
3222 /*
3223 * structure to poll for incoming data on communication socket avoids
3224 * making blocking sockets.
3225 */
3226 struct pollfd consumer_sockpoll[2];
3227 struct lttng_consumer_local_data *ctx = data;
3228
e7b994a3
DG
3229 rcu_register_thread();
3230
1fc79fb4
MD
3231 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_SESSIOND);
3232
2d57de81
MD
3233 if (testpoint(consumerd_thread_sessiond)) {
3234 goto error_testpoint;
3235 }
3236
9ce5646a
MD
3237 health_code_update();
3238
3bd1e081
MD
3239 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
3240 unlink(ctx->consumer_command_sock_path);
3241 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
3242 if (client_socket < 0) {
3243 ERR("Cannot create command socket");
3244 goto end;
3245 }
3246
3247 ret = lttcomm_listen_unix_sock(client_socket);
3248 if (ret < 0) {
3249 goto end;
3250 }
3251
32258573 3252 DBG("Sending ready command to lttng-sessiond");
f73fabfd 3253 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
3bd1e081
MD
3254 /* return < 0 on error, but == 0 is not fatal */
3255 if (ret < 0) {
32258573 3256 ERR("Error sending ready command to lttng-sessiond");
3bd1e081
MD
3257 goto end;
3258 }
3259
3bd1e081
MD
3260 /* prepare the FDs to poll : to client socket and the should_quit pipe */
3261 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
3262 consumer_sockpoll[0].events = POLLIN | POLLPRI;
3263 consumer_sockpoll[1].fd = client_socket;
3264 consumer_sockpoll[1].events = POLLIN | POLLPRI;
3265
84382d49
MD
3266 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3267 if (ret) {
3268 if (ret > 0) {
3269 /* should exit */
3270 err = 0;
3271 }
3bd1e081
MD
3272 goto end;
3273 }
3274 DBG("Connection on client_socket");
3275
3276 /* Blocking call, waiting for transmission */
3277 sock = lttcomm_accept_unix_sock(client_socket);
534d2592 3278 if (sock < 0) {
3bd1e081
MD
3279 WARN("On accept");
3280 goto end;
3281 }
3bd1e081 3282
331744e3
JD
3283 /*
3284 * Setup metadata socket which is the second socket connection on the
3285 * command unix socket.
3286 */
3287 ret = set_metadata_socket(ctx, consumer_sockpoll, client_socket);
84382d49
MD
3288 if (ret) {
3289 if (ret > 0) {
3290 /* should exit */
3291 err = 0;
3292 }
331744e3
JD
3293 goto end;
3294 }
3295
d96f09c6
DG
3296 /* This socket is not useful anymore. */
3297 ret = close(client_socket);
3298 if (ret < 0) {
3299 PERROR("close client_socket");
3300 }
3301 client_socket = -1;
3302
3bd1e081
MD
3303 /* update the polling structure to poll on the established socket */
3304 consumer_sockpoll[1].fd = sock;
3305 consumer_sockpoll[1].events = POLLIN | POLLPRI;
3306
3307 while (1) {
9ce5646a
MD
3308 health_code_update();
3309
3310 health_poll_entry();
3311 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3312 health_poll_exit();
84382d49
MD
3313 if (ret) {
3314 if (ret > 0) {
3315 /* should exit */
3316 err = 0;
3317 }
3bd1e081
MD
3318 goto end;
3319 }
3320 DBG("Incoming command on sock");
3321 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
4cbc1a04
DG
3322 if (ret <= 0) {
3323 /*
3324 * This could simply be a session daemon quitting. Don't output
3325 * ERR() here.
3326 */
3327 DBG("Communication interrupted on command socket");
41ba6035 3328 err = 0;
3bd1e081
MD
3329 goto end;
3330 }
10211f5c 3331 if (CMM_LOAD_SHARED(consumer_quit)) {
3bd1e081 3332 DBG("consumer_thread_receive_fds received quit from signal");
1fc79fb4 3333 err = 0; /* All is OK */
3bd1e081
MD
3334 goto end;
3335 }
ffe60014 3336 DBG("received command on sock");
3bd1e081 3337 }
1fc79fb4
MD
3338 /* All is OK */
3339 err = 0;
3340
3bd1e081 3341end:
ffe60014 3342 DBG("Consumer thread sessiond poll exiting");
3bd1e081 3343
d88aee68
DG
3344 /*
3345 * Close metadata streams since the producer is the session daemon which
3346 * just died.
3347 *
3348 * NOTE: for now, this only applies to the UST tracer.
3349 */
6d574024 3350 lttng_consumer_close_all_metadata();
d88aee68 3351
3bd1e081
MD
3352 /*
3353 * when all fds have hung up, the polling thread
3354 * can exit cleanly
3355 */
10211f5c 3356 CMM_STORE_SHARED(consumer_quit, 1);
3bd1e081 3357
04fdd819 3358 /*
c869f647 3359 * Notify the data poll thread to poll back again and test the
8994307f 3360 * consumer_quit state that we just set so to quit gracefully.
04fdd819 3361 */
acdb9057 3362 notify_thread_lttng_pipe(ctx->consumer_data_pipe);
c869f647 3363
a0cbdd2e 3364 notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
d8ef542d 3365
5c635c72
MD
3366 notify_health_quit_pipe(health_quit_pipe);
3367
d96f09c6
DG
3368 /* Cleaning up possibly open sockets. */
3369 if (sock >= 0) {
3370 ret = close(sock);
3371 if (ret < 0) {
3372 PERROR("close sock sessiond poll");
3373 }
3374 }
3375 if (client_socket >= 0) {
38476d24 3376 ret = close(client_socket);
d96f09c6
DG
3377 if (ret < 0) {
3378 PERROR("close client_socket sessiond poll");
3379 }
3380 }
3381
2d57de81 3382error_testpoint:
1fc79fb4
MD
3383 if (err) {
3384 health_error();
3385 ERR("Health error occurred in %s", __func__);
3386 }
3387 health_unregister(health_consumerd);
3388
e7b994a3 3389 rcu_unregister_thread();
3bd1e081
MD
3390 return NULL;
3391}
d41f73b7 3392
4078b776 3393ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
3394 struct lttng_consumer_local_data *ctx)
3395{
74251bb8 3396 ssize_t ret;
02d02e31
JD
3397 int rotate_ret;
3398 bool rotated = false;
74251bb8
DG
3399
3400 pthread_mutex_lock(&stream->lock);
94d49140
JD
3401 if (stream->metadata_flag) {
3402 pthread_mutex_lock(&stream->metadata_rdv_lock);
3403 }
74251bb8 3404
d41f73b7
MD
3405 switch (consumer_data.type) {
3406 case LTTNG_CONSUMER_KERNEL:
02d02e31 3407 ret = lttng_kconsumer_read_subbuffer(stream, ctx, &rotated);
74251bb8 3408 break;
7753dea8
MD
3409 case LTTNG_CONSUMER32_UST:
3410 case LTTNG_CONSUMER64_UST:
02d02e31 3411 ret = lttng_ustconsumer_read_subbuffer(stream, ctx, &rotated);
74251bb8 3412 break;
d41f73b7
MD
3413 default:
3414 ERR("Unknown consumer_data type");
3415 assert(0);
74251bb8
DG
3416 ret = -ENOSYS;
3417 break;
d41f73b7 3418 }
74251bb8 3419
94d49140
JD
3420 if (stream->metadata_flag) {
3421 pthread_cond_broadcast(&stream->metadata_rdv);
3422 pthread_mutex_unlock(&stream->metadata_rdv_lock);
3423 }
74251bb8 3424 pthread_mutex_unlock(&stream->lock);
02d02e31
JD
3425 if (rotated) {
3426 rotate_ret = consumer_post_rotation(stream, ctx);
3427 if (rotate_ret < 0) {
3428 ERR("Failed after a rotation");
3429 ret = -1;
3430 }
3431 }
3432
74251bb8 3433 return ret;
d41f73b7
MD
3434}
3435
3436int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
3437{
3438 switch (consumer_data.type) {
3439 case LTTNG_CONSUMER_KERNEL:
3440 return lttng_kconsumer_on_recv_stream(stream);
7753dea8
MD
3441 case LTTNG_CONSUMER32_UST:
3442 case LTTNG_CONSUMER64_UST:
d41f73b7
MD
3443 return lttng_ustconsumer_on_recv_stream(stream);
3444 default:
3445 ERR("Unknown consumer_data type");
3446 assert(0);
3447 return -ENOSYS;
3448 }
3449}
e4421fec
DG
3450
3451/*
3452 * Allocate and set consumer data hash tables.
3453 */
282dadbc 3454int lttng_consumer_init(void)
e4421fec 3455{
d88aee68 3456 consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3457 if (!consumer_data.channel_ht) {
3458 goto error;
3459 }
3460
d88aee68 3461 consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3462 if (!consumer_data.relayd_ht) {
3463 goto error;
3464 }
3465
d88aee68 3466 consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3467 if (!consumer_data.stream_list_ht) {
3468 goto error;
3469 }
3470
d8ef542d 3471 consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
282dadbc
MD
3472 if (!consumer_data.stream_per_chan_id_ht) {
3473 goto error;
3474 }
3475
3476 data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3477 if (!data_ht) {
3478 goto error;
3479 }
3480
3481 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3482 if (!metadata_ht) {
3483 goto error;
3484 }
3485
3486 return 0;
3487
3488error:
3489 return -1;
e4421fec 3490}
7735ef9e
DG
3491
3492/*
3493 * Process the ADD_RELAYD command receive by a consumer.
3494 *
3495 * This will create a relayd socket pair and add it to the relayd hash table.
3496 * The caller MUST acquire a RCU read side lock before calling it.
3497 */
2527bf85 3498 void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
7735ef9e 3499 struct lttng_consumer_local_data *ctx, int sock,
6151a90f 3500 struct pollfd *consumer_sockpoll,
d3e2ba59
JD
3501 struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
3502 uint64_t relayd_session_id)
7735ef9e 3503{
cd2b09ed 3504 int fd = -1, ret = -1, relayd_created = 0;
0c759fc9 3505 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
d4298c99 3506 struct consumer_relayd_sock_pair *relayd = NULL;
7735ef9e 3507
6151a90f
JD
3508 assert(ctx);
3509 assert(relayd_sock);
3510
da009f2c 3511 DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
7735ef9e
DG
3512
3513 /* Get relayd reference if exists. */
3514 relayd = consumer_find_relayd(net_seq_idx);
3515 if (relayd == NULL) {
da009f2c 3516 assert(sock_type == LTTNG_STREAM_CONTROL);
7735ef9e
DG
3517 /* Not found. Allocate one. */
3518 relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
3519 if (relayd == NULL) {
618a6a28
MD
3520 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3521 goto error;
0d08d75e 3522 } else {
30319bcb 3523 relayd->sessiond_session_id = sessiond_id;
0d08d75e 3524 relayd_created = 1;
7735ef9e 3525 }
0d08d75e
DG
3526
3527 /*
3528 * This code path MUST continue to the consumer send status message to
3529 * we can notify the session daemon and continue our work without
3530 * killing everything.
3531 */
da009f2c
MD
3532 } else {
3533 /*
3534 * relayd key should never be found for control socket.
3535 */
3536 assert(sock_type != LTTNG_STREAM_CONTROL);
0d08d75e
DG
3537 }
3538
3539 /* First send a status message before receiving the fds. */
0c759fc9 3540 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
618a6a28 3541 if (ret < 0) {
0d08d75e 3542 /* Somehow, the session daemon is not responding anymore. */
618a6a28
MD
3543 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3544 goto error_nosignal;
7735ef9e
DG
3545 }
3546
3547 /* Poll on consumer socket. */
84382d49
MD
3548 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3549 if (ret) {
3550 /* Needing to exit in the middle of a command: error. */
0d08d75e 3551 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
618a6a28 3552 goto error_nosignal;
7735ef9e
DG
3553 }
3554
3555 /* Get relayd socket from session daemon */
3556 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
3557 if (ret != sizeof(fd)) {
4028eeb9 3558 fd = -1; /* Just in case it gets set with an invalid value. */
0d08d75e
DG
3559
3560 /*
3561 * Failing to receive FDs might indicate a major problem such as
3562 * reaching a fd limit during the receive where the kernel returns a
3563 * MSG_CTRUNC and fails to cleanup the fd in the queue. Any case, we
3564 * don't take any chances and stop everything.
3565 *
3566 * XXX: Feature request #558 will fix that and avoid this possible
3567 * issue when reaching the fd limit.
3568 */
3569 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
618a6a28 3570 ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD;
f50f23d9
DG
3571 goto error;
3572 }
3573
7735ef9e
DG
3574 /* Copy socket information and received FD */
3575 switch (sock_type) {
3576 case LTTNG_STREAM_CONTROL:
3577 /* Copy received lttcomm socket */
6151a90f
JD
3578 lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
3579 ret = lttcomm_create_sock(&relayd->control_sock.sock);
4028eeb9 3580 /* Handle create_sock error. */
f66c074c 3581 if (ret < 0) {
618a6a28 3582 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
4028eeb9 3583 goto error;
f66c074c 3584 }
da009f2c
MD
3585 /*
3586 * Close the socket created internally by
3587 * lttcomm_create_sock, so we can replace it by the one
3588 * received from sessiond.
3589 */
3590 if (close(relayd->control_sock.sock.fd)) {
3591 PERROR("close");
3592 }
7735ef9e
DG
3593
3594 /* Assign new file descriptor */
6151a90f 3595 relayd->control_sock.sock.fd = fd;
4b29f1ce 3596 fd = -1; /* For error path */
6151a90f
JD
3597 /* Assign version values. */
3598 relayd->control_sock.major = relayd_sock->major;
3599 relayd->control_sock.minor = relayd_sock->minor;
c5b6f4f0 3600
d3e2ba59 3601 relayd->relayd_session_id = relayd_session_id;
c5b6f4f0 3602
7735ef9e
DG
3603 break;
3604 case LTTNG_STREAM_DATA:
3605 /* Copy received lttcomm socket */
6151a90f
JD
3606 lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
3607 ret = lttcomm_create_sock(&relayd->data_sock.sock);
4028eeb9 3608 /* Handle create_sock error. */
f66c074c 3609 if (ret < 0) {
618a6a28 3610 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
4028eeb9 3611 goto error;
f66c074c 3612 }
da009f2c
MD
3613 /*
3614 * Close the socket created internally by
3615 * lttcomm_create_sock, so we can replace it by the one
3616 * received from sessiond.
3617 */
3618 if (close(relayd->data_sock.sock.fd)) {
3619 PERROR("close");
3620 }
7735ef9e
DG
3621
3622 /* Assign new file descriptor */
6151a90f 3623 relayd->data_sock.sock.fd = fd;
4b29f1ce 3624 fd = -1; /* for eventual error paths */
6151a90f
JD
3625 /* Assign version values. */
3626 relayd->data_sock.major = relayd_sock->major;
3627 relayd->data_sock.minor = relayd_sock->minor;
7735ef9e
DG
3628 break;
3629 default:
3630 ERR("Unknown relayd socket type (%d)", sock_type);
618a6a28 3631 ret_code = LTTCOMM_CONSUMERD_FATAL;
7735ef9e
DG
3632 goto error;
3633 }
3634
d88aee68 3635 DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
7735ef9e
DG
3636 sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
3637 relayd->net_seq_idx, fd);
3638
618a6a28
MD
3639 /* We successfully added the socket. Send status back. */
3640 ret = consumer_send_status_msg(sock, ret_code);
3641 if (ret < 0) {
3642 /* Somehow, the session daemon is not responding anymore. */
3643 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3644 goto error_nosignal;
3645 }
3646
7735ef9e
DG
3647 /*
3648 * Add relayd socket pair to consumer data hashtable. If object already
3649 * exists or on error, the function gracefully returns.
3650 */
d09e1200 3651 add_relayd(relayd);
7735ef9e
DG
3652
3653 /* All good! */
2527bf85 3654 return;
7735ef9e
DG
3655
3656error:
618a6a28
MD
3657 if (consumer_send_status_msg(sock, ret_code) < 0) {
3658 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3659 }
3660
3661error_nosignal:
4028eeb9
DG
3662 /* Close received socket if valid. */
3663 if (fd >= 0) {
3664 if (close(fd)) {
3665 PERROR("close received socket");
3666 }
3667 }
cd2b09ed
DG
3668
3669 if (relayd_created) {
cd2b09ed
DG
3670 free(relayd);
3671 }
7735ef9e 3672}
ca22feea 3673
4e9a4686
DG
3674/*
3675 * Try to lock the stream mutex.
3676 *
3677 * On success, 1 is returned else 0 indicating that the mutex is NOT lock.
3678 */
3679static int stream_try_lock(struct lttng_consumer_stream *stream)
3680{
3681 int ret;
3682
3683 assert(stream);
3684
3685 /*
3686 * Try to lock the stream mutex. On failure, we know that the stream is
3687 * being used else where hence there is data still being extracted.
3688 */
3689 ret = pthread_mutex_trylock(&stream->lock);
3690 if (ret) {
3691 /* For both EBUSY and EINVAL error, the mutex is NOT locked. */
3692 ret = 0;
3693 goto end;
3694 }
3695
3696 ret = 1;
3697
3698end:
3699 return ret;
3700}
3701
f7079f67
DG
3702/*
3703 * Search for a relayd associated to the session id and return the reference.
3704 *
3705 * A rcu read side lock MUST be acquire before calling this function and locked
3706 * until the relayd object is no longer necessary.
3707 */
3708static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
3709{
3710 struct lttng_ht_iter iter;
f7079f67 3711 struct consumer_relayd_sock_pair *relayd = NULL;
f7079f67
DG
3712
3713 /* Iterate over all relayd since they are indexed by net_seq_idx. */
3714 cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
3715 node.node) {
18261bd1
DG
3716 /*
3717 * Check by sessiond id which is unique here where the relayd session
3718 * id might not be when having multiple relayd.
3719 */
3720 if (relayd->sessiond_session_id == id) {
f7079f67 3721 /* Found the relayd. There can be only one per id. */
18261bd1 3722 goto found;
f7079f67
DG
3723 }
3724 }
3725
18261bd1
DG
3726 return NULL;
3727
3728found:
f7079f67
DG
3729 return relayd;
3730}
3731
ca22feea
DG
3732/*
3733 * Check if for a given session id there is still data needed to be extract
3734 * from the buffers.
3735 *
6d805429 3736 * Return 1 if data is pending or else 0 meaning ready to be read.
ca22feea 3737 */
6d805429 3738int consumer_data_pending(uint64_t id)
ca22feea
DG
3739{
3740 int ret;
3741 struct lttng_ht_iter iter;
3742 struct lttng_ht *ht;
3743 struct lttng_consumer_stream *stream;
f7079f67 3744 struct consumer_relayd_sock_pair *relayd = NULL;
6d805429 3745 int (*data_pending)(struct lttng_consumer_stream *);
ca22feea 3746
6d805429 3747 DBG("Consumer data pending command on session id %" PRIu64, id);
ca22feea 3748
6f6eda74 3749 rcu_read_lock();
ca22feea
DG
3750 pthread_mutex_lock(&consumer_data.lock);
3751
3752 switch (consumer_data.type) {
3753 case LTTNG_CONSUMER_KERNEL:
6d805429 3754 data_pending = lttng_kconsumer_data_pending;
ca22feea
DG
3755 break;
3756 case LTTNG_CONSUMER32_UST:
3757 case LTTNG_CONSUMER64_UST:
6d805429 3758 data_pending = lttng_ustconsumer_data_pending;
ca22feea
DG
3759 break;
3760 default:
3761 ERR("Unknown consumer data type");
3762 assert(0);
3763 }
3764
3765 /* Ease our life a bit */
3766 ht = consumer_data.stream_list_ht;
3767
f7079f67
DG
3768 relayd = find_relayd_by_session_id(id);
3769 if (relayd) {
3770 /* Send init command for data pending. */
3771 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3772 ret = relayd_begin_data_pending(&relayd->control_sock,
3773 relayd->relayd_session_id);
3774 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3775 if (ret < 0) {
3776 /* Communication error thus the relayd so no data pending. */
3777 goto data_not_pending;
3778 }
3779 }
3780
c8f59ee5 3781 cds_lfht_for_each_entry_duplicate(ht->ht,
d88aee68
DG
3782 ht->hash_fct(&id, lttng_ht_seed),
3783 ht->match_fct, &id,
ca22feea 3784 &iter.iter, stream, node_session_id.node) {
4e9a4686
DG
3785 /* If this call fails, the stream is being used hence data pending. */
3786 ret = stream_try_lock(stream);
3787 if (!ret) {
f7079f67 3788 goto data_pending;
ca22feea 3789 }
ca22feea 3790
4e9a4686
DG
3791 /*
3792 * A removed node from the hash table indicates that the stream has
3793 * been deleted thus having a guarantee that the buffers are closed
3794 * on the consumer side. However, data can still be transmitted
3795 * over the network so don't skip the relayd check.
3796 */
3797 ret = cds_lfht_is_node_deleted(&stream->node.node);
3798 if (!ret) {
3799 /* Check the stream if there is data in the buffers. */
6d805429
DG
3800 ret = data_pending(stream);
3801 if (ret == 1) {
4e9a4686 3802 pthread_mutex_unlock(&stream->lock);
f7079f67 3803 goto data_pending;
4e9a4686
DG
3804 }
3805 }
3806
3807 /* Relayd check */
f7079f67 3808 if (relayd) {
c8f59ee5
DG
3809 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3810 if (stream->metadata_flag) {
ad7051c0
DG
3811 ret = relayd_quiescent_control(&relayd->control_sock,
3812 stream->relayd_stream_id);
c8f59ee5 3813 } else {
6d805429 3814 ret = relayd_data_pending(&relayd->control_sock,
39df6d9f
DG
3815 stream->relayd_stream_id,
3816 stream->next_net_seq_num - 1);
c8f59ee5
DG
3817 }
3818 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
6d805429 3819 if (ret == 1) {
4e9a4686 3820 pthread_mutex_unlock(&stream->lock);
f7079f67 3821 goto data_pending;
c8f59ee5
DG
3822 }
3823 }
4e9a4686 3824 pthread_mutex_unlock(&stream->lock);
c8f59ee5 3825 }
ca22feea 3826
f7079f67
DG
3827 if (relayd) {
3828 unsigned int is_data_inflight = 0;
3829
3830 /* Send init command for data pending. */
3831 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3832 ret = relayd_end_data_pending(&relayd->control_sock,
3833 relayd->relayd_session_id, &is_data_inflight);
3834 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
bdd88757 3835 if (ret < 0) {
f7079f67
DG
3836 goto data_not_pending;
3837 }
bdd88757
DG
3838 if (is_data_inflight) {
3839 goto data_pending;
3840 }
f7079f67
DG
3841 }
3842
ca22feea 3843 /*
f7079f67
DG
3844 * Finding _no_ node in the hash table and no inflight data means that the
3845 * stream(s) have been removed thus data is guaranteed to be available for
3846 * analysis from the trace files.
ca22feea
DG
3847 */
3848
f7079f67 3849data_not_pending:
ca22feea
DG
3850 /* Data is available to be read by a viewer. */
3851 pthread_mutex_unlock(&consumer_data.lock);
c8f59ee5 3852 rcu_read_unlock();
6d805429 3853 return 0;
ca22feea 3854
f7079f67 3855data_pending:
ca22feea
DG
3856 /* Data is still being extracted from buffers. */
3857 pthread_mutex_unlock(&consumer_data.lock);
c8f59ee5 3858 rcu_read_unlock();
6d805429 3859 return 1;
ca22feea 3860}
f50f23d9
DG
3861
3862/*
3863 * Send a ret code status message to the sessiond daemon.
3864 *
3865 * Return the sendmsg() return value.
3866 */
3867int consumer_send_status_msg(int sock, int ret_code)
3868{
3869 struct lttcomm_consumer_status_msg msg;
3870
53efb85a 3871 memset(&msg, 0, sizeof(msg));
f50f23d9
DG
3872 msg.ret_code = ret_code;
3873
3874 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3875}
ffe60014
DG
3876
3877/*
3878 * Send a channel status message to the sessiond daemon.
3879 *
3880 * Return the sendmsg() return value.
3881 */
3882int consumer_send_status_channel(int sock,
3883 struct lttng_consumer_channel *channel)
3884{
3885 struct lttcomm_consumer_status_channel msg;
3886
3887 assert(sock >= 0);
3888
53efb85a 3889 memset(&msg, 0, sizeof(msg));
ffe60014 3890 if (!channel) {
0c759fc9 3891 msg.ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
ffe60014 3892 } else {
0c759fc9 3893 msg.ret_code = LTTCOMM_CONSUMERD_SUCCESS;
ffe60014
DG
3894 msg.key = channel->key;
3895 msg.stream_count = channel->streams.count;
3896 }
3897
3898 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3899}
5c786ded 3900
d07ceecd
MD
3901unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
3902 unsigned long produced_pos, uint64_t nb_packets_per_stream,
3903 uint64_t max_sb_size)
5c786ded 3904{
d07ceecd 3905 unsigned long start_pos;
5c786ded 3906
d07ceecd
MD
3907 if (!nb_packets_per_stream) {
3908 return consumed_pos; /* Grab everything */
3909 }
3910 start_pos = produced_pos - offset_align_floor(produced_pos, max_sb_size);
3911 start_pos -= max_sb_size * nb_packets_per_stream;
3912 if ((long) (start_pos - consumed_pos) < 0) {
3913 return consumed_pos; /* Grab everything */
3914 }
3915 return start_pos;
5c786ded 3916}
a1ae2ea5 3917
b99a8d42
JD
3918static
3919int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active)
3920{
3921 int ret = 0;
3922
3923 switch (consumer_data.type) {
3924 case LTTNG_CONSUMER_KERNEL:
3925 ret = kernctl_buffer_flush(stream->wait_fd);
3926 if (ret < 0) {
3927 ERR("Failed to flush kernel stream");
3928 goto end;
3929 }
3930 break;
3931 case LTTNG_CONSUMER32_UST:
3932 case LTTNG_CONSUMER64_UST:
3933 lttng_ustctl_flush_buffer(stream, producer_active);
3934 break;
3935 default:
3936 ERR("Unknown consumer_data type");
3937 abort();
3938 }
3939
3940end:
3941 return ret;
3942}
3943
3944/*
3945 * Sample the rotate position for all the streams of a channel. If a stream
3946 * is already at the rotate position (produced == consumed), we flag it as
3947 * ready for rotation. The rotation of ready streams occurs after we have
3948 * replied to the session daemon that we have finished sampling the positions.
3949 *
3950 * Returns 0 on success, < 0 on error
3951 */
3952int lttng_consumer_rotate_channel(uint64_t key, const char *path,
3953 uint64_t relayd_id, uint32_t metadata, uint64_t new_chunk_id,
3954 struct lttng_consumer_local_data *ctx)
3955{
3956 int ret;
3957 struct lttng_consumer_channel *channel;
3958 struct lttng_consumer_stream *stream;
3959 struct lttng_ht_iter iter;
3960 struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
3961
3962 DBG("Consumer sample rotate position for channel %" PRIu64, key);
3963
3964 rcu_read_lock();
3965
3966 channel = consumer_find_channel(key);
3967 if (!channel) {
3968 ERR("No channel found for key %" PRIu64, key);
3969 ret = -1;
3970 goto end;
3971 }
3972
3973 pthread_mutex_lock(&channel->lock);
3974 channel->current_chunk_id = new_chunk_id;
3975
3976 ret = lttng_strncpy(channel->pathname, path, sizeof(channel->pathname));
3977 if (ret) {
3978 ERR("Failed to copy new path to channel during channel rotation");
3979 ret = -1;
3980 goto end_unlock_channel;
3981 }
3982
3983 if (relayd_id == -1ULL) {
3984 /*
3985 * The domain path (/ust or /kernel) has been created before, we
3986 * now need to create the last part of the path: the application/user
3987 * specific section (uid/1000/64-bit).
3988 */
3989 ret = utils_mkdir_recursive(channel->pathname, S_IRWXU | S_IRWXG,
3990 channel->uid, channel->gid);
3991 if (ret < 0) {
3992 ERR("Failed to create trace directory at %s during rotation",
3993 channel->pathname);
3994 ret = -1;
3995 goto end_unlock_channel;
3996 }
3997 }
3998
3999 cds_lfht_for_each_entry_duplicate(ht->ht,
4000 ht->hash_fct(&channel->key, lttng_ht_seed),
4001 ht->match_fct, &channel->key, &iter.iter,
4002 stream, node_channel_id.node) {
4003 unsigned long consumed_pos;
4004
4005 health_code_update();
4006
4007 /*
4008 * Lock stream because we are about to change its state.
4009 */
4010 pthread_mutex_lock(&stream->lock);
4011
4012 ret = lttng_strncpy(stream->channel_read_only_attributes.path,
4013 channel->pathname,
4014 sizeof(stream->channel_read_only_attributes.path));
4015 if (ret) {
4016 ERR("Failed to sample channel path name during channel rotation");
4017 goto end_unlock_stream;
4018 }
4019 ret = lttng_consumer_sample_snapshot_positions(stream);
4020 if (ret < 0) {
4021 ERR("Failed to sample snapshot position during channel rotation");
4022 goto end_unlock_stream;
4023 }
4024
4025 ret = lttng_consumer_get_produced_snapshot(stream,
4026 &stream->rotate_position);
4027 if (ret < 0) {
4028 ERR("Failed to sample produced position during channel rotation");
4029 goto end_unlock_stream;
4030 }
4031
4032 lttng_consumer_get_consumed_snapshot(stream,
4033 &consumed_pos);
4034 if (consumed_pos == stream->rotate_position) {
4035 stream->rotate_ready = true;
4036 }
4037 channel->nr_stream_rotate_pending++;
4038
4039 ret = consumer_flush_buffer(stream, 1);
4040 if (ret < 0) {
4041 ERR("Failed to flush stream %" PRIu64 " during channel rotation",
4042 stream->key);
4043 goto end_unlock_stream;
4044 }
4045
4046 pthread_mutex_unlock(&stream->lock);
4047 }
4048 pthread_mutex_unlock(&channel->lock);
4049
4050 ret = 0;
4051 goto end;
4052
4053end_unlock_stream:
4054 pthread_mutex_unlock(&stream->lock);
4055end_unlock_channel:
4056 pthread_mutex_unlock(&channel->lock);
4057end:
4058 rcu_read_unlock();
4059 return ret;
4060}
4061
02d02e31
JD
4062/*
4063 * Check if a stream is ready to be rotated after extracting it.
4064 *
4065 * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
4066 * error. Stream lock must be held.
4067 */
4068int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
4069{
4070 int ret;
4071 unsigned long consumed_pos;
4072
4073 if (!stream->rotate_position && !stream->rotate_ready) {
4074 ret = 0;
4075 goto end;
4076 }
4077
4078 if (stream->rotate_ready) {
4079 ret = 1;
4080 goto end;
4081 }
4082
4083 /*
4084 * If we don't have the rotate_ready flag, check the consumed position
4085 * to determine if we need to rotate.
4086 */
4087 ret = lttng_consumer_sample_snapshot_positions(stream);
4088 if (ret < 0) {
4089 ERR("Taking snapshot positions");
4090 goto end;
4091 }
4092
4093 ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos);
4094 if (ret < 0) {
4095 ERR("Consumed snapshot position");
4096 goto end;
4097 }
4098
4099 /* Rotate position not reached yet (with check for overflow). */
4100 if ((long) (consumed_pos - stream->rotate_position) < 0) {
4101 ret = 0;
4102 goto end;
4103 }
4104 ret = 1;
4105
4106end:
4107 return ret;
4108}
4109
d73bf3d7
JD
4110/*
4111 * Reset the state for a stream after a rotation occurred.
4112 */
4113void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream)
4114{
4115 stream->rotate_position = 0;
4116 stream->rotate_ready = false;
4117}
4118
4119/*
4120 * Perform the rotation a local stream file.
4121 */
4122int rotate_local_stream(struct lttng_consumer_local_data *ctx,
4123 struct lttng_consumer_stream *stream)
4124{
4125 int ret;
4126
4127 DBG("Rotate local stream: stream key %" PRIu64 ", channel key %" PRIu64 " at path %s",
4128 stream->key,
4129 stream->chan->key,
4130 stream->channel_read_only_attributes.path);
4131
4132 ret = close(stream->out_fd);
4133 if (ret < 0) {
4134 PERROR("Closing trace file (fd %d), stream %" PRIu64,
4135 stream->out_fd, stream->key);
4136 assert(0);
4137 goto error;
4138 }
4139
4140 ret = utils_create_stream_file(
4141 stream->channel_read_only_attributes.path,
4142 stream->name,
4143 stream->channel_read_only_attributes.tracefile_size,
4144 stream->tracefile_count_current,
4145 stream->uid, stream->gid, NULL);
4146 if (ret < 0) {
4147 ERR("Rotate create stream file");
4148 goto error;
4149 }
4150 stream->out_fd = ret;
4151 stream->tracefile_size_current = 0;
4152
4153 if (!stream->metadata_flag) {
4154 struct lttng_index_file *index_file;
4155
4156 lttng_index_file_put(stream->index_file);
4157
4158 index_file = lttng_index_file_create(
4159 stream->channel_read_only_attributes.path,
4160 stream->name, stream->uid, stream->gid,
4161 stream->channel_read_only_attributes.tracefile_size,
4162 stream->tracefile_count_current,
4163 CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
4164 if (!index_file) {
4165 ERR("Create index file during rotation");
4166 goto error;
4167 }
4168 stream->index_file = index_file;
4169 stream->out_fd_offset = 0;
4170 }
4171
4172 ret = 0;
4173 goto end;
4174
4175error:
4176 ret = -1;
4177end:
4178 return ret;
4179
4180}
4181
4182/*
4183 * Perform the rotation a stream file on the relay.
4184 */
4185int rotate_relay_stream(struct lttng_consumer_local_data *ctx,
4186 struct lttng_consumer_stream *stream)
4187{
4188 int ret;
4189 struct consumer_relayd_sock_pair *relayd;
4190
4191 DBG("Rotate relay stream");
4192 relayd = consumer_find_relayd(stream->net_seq_idx);
4193 if (!relayd) {
4194 ERR("Failed to find relayd");
4195 ret = -1;
4196 goto end;
4197 }
4198
4199 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4200 ret = relayd_rotate_stream(&relayd->control_sock,
4201 stream->relayd_stream_id,
4202 stream->channel_read_only_attributes.path,
4203 stream->chan->current_chunk_id,
4204 stream->last_sequence_number);
4205 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4206 if (ret) {
4207 ERR("Rotate relay stream");
4208 }
4209
4210end:
4211 return ret;
4212}
4213
4214/*
4215 * Performs the stream rotation for the rotate session feature if needed.
4216 * It must be called with the stream lock held.
4217 *
4218 * Return 0 on success, a negative number of error.
4219 */
4220int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
02d02e31 4221 struct lttng_consumer_stream *stream, bool *rotated)
d73bf3d7
JD
4222{
4223 int ret;
4224
4225 DBG("Consumer rotate stream %" PRIu64, stream->key);
4226
4227 if (stream->net_seq_idx != (uint64_t) -1ULL) {
4228 ret = rotate_relay_stream(ctx, stream);
4229 } else {
4230 ret = rotate_local_stream(ctx, stream);
4231 }
4232 if (ret < 0) {
4233 ERR("Rotate stream");
4234 goto error;
4235 }
4236
4237 if (stream->metadata_flag) {
4238 switch (consumer_data.type) {
4239 case LTTNG_CONSUMER_KERNEL:
4240 /*
4241 * Reset the position of what has been read from the metadata
4242 * cache to 0 so we can dump it again.
4243 */
4244 ret = kernctl_metadata_cache_dump(stream->wait_fd);
4245 if (ret < 0) {
4246 ERR("Failed to dump the kernel metadata cache after rotation");
4247 goto error;
4248 }
4249 break;
4250 case LTTNG_CONSUMER32_UST:
4251 case LTTNG_CONSUMER64_UST:
4252 /*
4253 * Reset the position pushed from the metadata cache so it
4254 * will write from the beginning on the next push.
4255 */
4256 stream->ust_metadata_pushed = 0;
4257 break;
4258 default:
4259 ERR("Unknown consumer_data type");
4260 abort();
4261 }
4262 }
4263 lttng_consumer_reset_stream_rotate_state(stream);
4264
02d02e31
JD
4265 if (rotated) {
4266 *rotated = true;
4267 }
4268
d73bf3d7
JD
4269 ret = 0;
4270
4271error:
4272 return ret;
4273}
4274
b99a8d42
JD
4275/*
4276 * Rotate all the ready streams now.
4277 *
4278 * This is especially important for low throughput streams that have already
4279 * been consumed, we cannot wait for their next packet to perform the
4280 * rotation.
4281 *
4282 * Returns 0 on success, < 0 on error
4283 */
4284int lttng_consumer_rotate_ready_streams(uint64_t key,
4285 struct lttng_consumer_local_data *ctx)
4286{
4287 int ret;
4288 struct lttng_consumer_channel *channel;
4289 struct lttng_consumer_stream *stream;
4290 struct lttng_ht_iter iter;
4291 struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
4292
4293 rcu_read_lock();
4294
4295 DBG("Consumer rotate ready streams in channel %" PRIu64, key);
4296
4297 channel = consumer_find_channel(key);
4298 if (!channel) {
4299 ERR("No channel found for key %" PRIu64, key);
4300 ret = -1;
4301 goto end;
4302 }
4303
4304 cds_lfht_for_each_entry_duplicate(ht->ht,
4305 ht->hash_fct(&channel->key, lttng_ht_seed),
4306 ht->match_fct, &channel->key, &iter.iter,
4307 stream, node_channel_id.node) {
4308 health_code_update();
4309
4310 pthread_mutex_lock(&stream->lock);
4311
4312 if (!stream->rotate_ready) {
4313 pthread_mutex_unlock(&stream->lock);
4314 continue;
4315 }
4316 DBG("Consumer rotate ready stream %" PRIu64, stream->key);
4317
4318 ret = lttng_consumer_rotate_stream(ctx, stream, NULL);
4319 pthread_mutex_unlock(&stream->lock);
4320 if (ret) {
4321 goto end;
4322 }
4323
4324 ret = consumer_post_rotation(stream, ctx);
4325 if (ret) {
4326 goto end;
4327 }
4328 }
4329
4330 ret = 0;
4331
4332end:
4333 rcu_read_unlock();
4334 return ret;
4335}
4336
00fb02ac
JD
4337static
4338int rotate_rename_local(const char *old_path, const char *new_path,
4339 uid_t uid, gid_t gid)
4340{
4341 int ret;
4342
4343 assert(old_path);
4344 assert(new_path);
4345
4346 ret = utils_mkdir_recursive(new_path, S_IRWXU | S_IRWXG, uid, gid);
4347 if (ret < 0) {
4348 ERR("Create directory on rotate");
4349 goto end;
4350 }
4351
4352 ret = rename(old_path, new_path);
4353 if (ret < 0 && errno != ENOENT) {
4354 PERROR("Rename completed rotation chunk");
4355 goto end;
4356 }
4357
4358 ret = 0;
4359end:
4360 return ret;
4361}
4362
4363static
4364int rotate_rename_relay(const char *old_path, const char *new_path,
4365 uint64_t relayd_id)
4366{
4367 int ret;
4368 struct consumer_relayd_sock_pair *relayd;
4369
4370 relayd = consumer_find_relayd(relayd_id);
4371 if (!relayd) {
4372 ERR("Failed to find relayd while running rotate_rename_relay command");
4373 ret = -1;
4374 goto end;
4375 }
4376
4377 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4378 ret = relayd_rotate_rename(&relayd->control_sock, old_path, new_path);
4379 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4380end:
4381 return ret;
4382}
4383
4384int lttng_consumer_rotate_rename(const char *old_path, const char *new_path,
4385 uid_t uid, gid_t gid, uint64_t relayd_id)
4386{
4387 if (relayd_id != -1ULL) {
4388 return rotate_rename_relay(old_path, new_path, relayd_id);
4389 } else {
4390 return rotate_rename_local(old_path, new_path, uid, gid);
4391 }
4392}
4393
d88744a4
JD
4394int lttng_consumer_rotate_pending_relay(uint64_t session_id,
4395 uint64_t relayd_id, uint64_t chunk_id)
4396{
4397 int ret;
4398 struct consumer_relayd_sock_pair *relayd;
4399
4400 relayd = consumer_find_relayd(relayd_id);
4401 if (!relayd) {
4402 ERR("Failed to find relayd");
4403 ret = -1;
4404 goto end;
4405 }
4406
4407 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4408 ret = relayd_rotate_pending(&relayd->control_sock, chunk_id);
4409 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4410
4411end:
4412 return ret;
4413}
4414
a1ae2ea5
JD
4415static
4416int mkdir_local(const char *path, uid_t uid, gid_t gid)
4417{
4418 int ret;
4419
4420 ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, uid, gid);
4421 if (ret < 0) {
4422 /* utils_mkdir_recursive logs an error. */
4423 goto end;
4424 }
4425
4426 ret = 0;
4427end:
4428 return ret;
4429}
4430
4431static
4432int mkdir_relay(const char *path, uint64_t relayd_id)
4433{
4434 int ret;
4435 struct consumer_relayd_sock_pair *relayd;
4436
4437 relayd = consumer_find_relayd(relayd_id);
4438 if (!relayd) {
4439 ERR("Failed to find relayd");
4440 ret = -1;
4441 goto end;
4442 }
4443
4444 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4445 ret = relayd_mkdir(&relayd->control_sock, path);
4446 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4447
4448end:
4449 return ret;
4450
4451}
4452
4453int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid,
4454 uint64_t relayd_id)
4455{
4456 if (relayd_id != -1ULL) {
4457 return mkdir_relay(path, relayd_id);
4458 } else {
4459 return mkdir_local(path, uid, gid);
4460 }
4461}
This page took 0.338799 seconds and 4 git commands to generate.