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