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