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