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