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