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