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