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