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