src/common: use single Makefile for parallel builds
[lttng-tools.git] / src / common / consumer / consumer.cpp
1 /*
2 * Copyright (C) 2011 EfficiOS Inc.
3 * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
5 *
6 * SPDX-License-Identifier: GPL-2.0-only
7 *
8 */
9
10 #include "common/index/ctf-index.h"
11 #define _LGPL_SOURCE
12 #include <poll.h>
13 #include <pthread.h>
14 #include <stdlib.h>
15 #include <string.h>
16 #include <sys/mman.h>
17 #include <sys/socket.h>
18 #include <sys/types.h>
19 #include <unistd.h>
20 #include <inttypes.h>
21 #include <signal.h>
22
23 #include <bin/lttng-consumerd/health-consumerd.h>
24 #include <common/common.h>
25 #include <common/utils.h>
26 #include <common/time.h>
27 #include <common/compat/poll.h>
28 #include <common/compat/endian.h>
29 #include <common/index/index.h>
30 #include <common/kernel-ctl/kernel-ctl.h>
31 #include <common/sessiond-comm/relayd.h>
32 #include <common/sessiond-comm/sessiond-comm.h>
33 #include <common/kernel-consumer/kernel-consumer.h>
34 #include <common/relayd/relayd.h>
35 #include <common/ust-consumer/ust-consumer.h>
36 #include <common/consumer/consumer-timer.h>
37 #include <common/consumer/consumer.h>
38 #include <common/consumer/consumer-stream.h>
39 #include <common/consumer/consumer-testpoint.h>
40 #include <common/align.h>
41 #include <common/consumer/consumer-metadata-cache.h>
42 #include <common/trace-chunk.h>
43 #include <common/trace-chunk-registry.h>
44 #include <common/string-utils/format.h>
45 #include <common/dynamic-array.h>
46
47 lttng_consumer_global_data the_consumer_data;
48
49 enum consumer_channel_action {
50 CONSUMER_CHANNEL_ADD,
51 CONSUMER_CHANNEL_DEL,
52 CONSUMER_CHANNEL_QUIT,
53 };
54
55 struct consumer_channel_msg {
56 enum consumer_channel_action action;
57 struct lttng_consumer_channel *chan; /* add */
58 uint64_t key; /* del */
59 };
60
61 /* Flag used to temporarily pause data consumption from testpoints. */
62 int data_consumption_paused;
63
64 /*
65 * Flag to inform the polling thread to quit when all fd hung up. Updated by
66 * the consumer_thread_receive_fds when it notices that all fds has hung up.
67 * Also updated by the signal handler (consumer_should_exit()). Read by the
68 * polling threads.
69 */
70 int consumer_quit;
71
72 /*
73 * Global hash table containing respectively metadata and data streams. The
74 * stream element in this ht should only be updated by the metadata poll thread
75 * for the metadata and the data poll thread for the data.
76 */
77 static struct lttng_ht *metadata_ht;
78 static struct lttng_ht *data_ht;
79
80 static const char *get_consumer_domain(void)
81 {
82 switch (the_consumer_data.type) {
83 case LTTNG_CONSUMER_KERNEL:
84 return DEFAULT_KERNEL_TRACE_DIR;
85 case LTTNG_CONSUMER64_UST:
86 /* Fall-through. */
87 case LTTNG_CONSUMER32_UST:
88 return DEFAULT_UST_TRACE_DIR;
89 default:
90 abort();
91 }
92 }
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 LTTNG_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 LTTNG_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 LTTNG_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 ASSERT_RCU_READ_LOCKED();
250
251 /* -1ULL keys are lookup failures */
252 if (key == (uint64_t) -1ULL) {
253 return NULL;
254 }
255
256 lttng_ht_lookup(the_consumer_data.channel_ht, &key, &iter);
257 node = lttng_ht_iter_get_node_u64(&iter);
258 if (node != NULL) {
259 channel = caa_container_of(node, struct lttng_consumer_channel, node);
260 }
261
262 return channel;
263 }
264
265 /*
266 * There is a possibility that the consumer does not have enough time between
267 * the close of the channel on the session daemon and the cleanup in here thus
268 * once we have a channel add with an existing key, we know for sure that this
269 * channel will eventually get cleaned up by all streams being closed.
270 *
271 * This function just nullifies the already existing channel key.
272 */
273 static void steal_channel_key(uint64_t key)
274 {
275 struct lttng_consumer_channel *channel;
276
277 rcu_read_lock();
278 channel = consumer_find_channel(key);
279 if (channel) {
280 channel->key = (uint64_t) -1ULL;
281 /*
282 * We don't want the lookup to match, but we still need to iterate on
283 * this channel when iterating over the hash table. Just change the
284 * node key.
285 */
286 channel->node.key = (uint64_t) -1ULL;
287 }
288 rcu_read_unlock();
289 }
290
291 static void free_channel_rcu(struct rcu_head *head)
292 {
293 struct lttng_ht_node_u64 *node =
294 caa_container_of(head, struct lttng_ht_node_u64, head);
295 struct lttng_consumer_channel *channel =
296 caa_container_of(node, struct lttng_consumer_channel, node);
297
298 switch (the_consumer_data.type) {
299 case LTTNG_CONSUMER_KERNEL:
300 break;
301 case LTTNG_CONSUMER32_UST:
302 case LTTNG_CONSUMER64_UST:
303 lttng_ustconsumer_free_channel(channel);
304 break;
305 default:
306 ERR("Unknown consumer_data type");
307 abort();
308 }
309 free(channel);
310 }
311
312 /*
313 * RCU protected relayd socket pair free.
314 */
315 static void free_relayd_rcu(struct rcu_head *head)
316 {
317 struct lttng_ht_node_u64 *node =
318 caa_container_of(head, struct lttng_ht_node_u64, head);
319 struct consumer_relayd_sock_pair *relayd =
320 caa_container_of(node, struct consumer_relayd_sock_pair, node);
321
322 /*
323 * Close all sockets. This is done in the call RCU since we don't want the
324 * socket fds to be reassigned thus potentially creating bad state of the
325 * relayd object.
326 *
327 * We do not have to lock the control socket mutex here since at this stage
328 * there is no one referencing to this relayd object.
329 */
330 (void) relayd_close(&relayd->control_sock);
331 (void) relayd_close(&relayd->data_sock);
332
333 pthread_mutex_destroy(&relayd->ctrl_sock_mutex);
334 free(relayd);
335 }
336
337 /*
338 * Destroy and free relayd socket pair object.
339 */
340 void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
341 {
342 int ret;
343 struct lttng_ht_iter iter;
344
345 if (relayd == NULL) {
346 return;
347 }
348
349 DBG("Consumer destroy and close relayd socket pair");
350
351 iter.iter.node = &relayd->node.node;
352 ret = lttng_ht_del(the_consumer_data.relayd_ht, &iter);
353 if (ret != 0) {
354 /* We assume the relayd is being or is destroyed */
355 return;
356 }
357
358 /* RCU free() call */
359 call_rcu(&relayd->node.head, free_relayd_rcu);
360 }
361
362 /*
363 * Remove a channel from the global list protected by a mutex. This function is
364 * also responsible for freeing its data structures.
365 */
366 void consumer_del_channel(struct lttng_consumer_channel *channel)
367 {
368 struct lttng_ht_iter iter;
369
370 DBG("Consumer delete channel key %" PRIu64, channel->key);
371
372 pthread_mutex_lock(&the_consumer_data.lock);
373 pthread_mutex_lock(&channel->lock);
374
375 /* Destroy streams that might have been left in the stream list. */
376 clean_channel_stream_list(channel);
377
378 if (channel->live_timer_enabled == 1) {
379 consumer_timer_live_stop(channel);
380 }
381 if (channel->monitor_timer_enabled == 1) {
382 consumer_timer_monitor_stop(channel);
383 }
384
385 switch (the_consumer_data.type) {
386 case LTTNG_CONSUMER_KERNEL:
387 break;
388 case LTTNG_CONSUMER32_UST:
389 case LTTNG_CONSUMER64_UST:
390 lttng_ustconsumer_del_channel(channel);
391 break;
392 default:
393 ERR("Unknown consumer_data type");
394 abort();
395 goto end;
396 }
397
398 lttng_trace_chunk_put(channel->trace_chunk);
399 channel->trace_chunk = NULL;
400
401 if (channel->is_published) {
402 int ret;
403
404 rcu_read_lock();
405 iter.iter.node = &channel->node.node;
406 ret = lttng_ht_del(the_consumer_data.channel_ht, &iter);
407 LTTNG_ASSERT(!ret);
408
409 iter.iter.node = &channel->channels_by_session_id_ht_node.node;
410 ret = lttng_ht_del(the_consumer_data.channels_by_session_id_ht,
411 &iter);
412 LTTNG_ASSERT(!ret);
413 rcu_read_unlock();
414 }
415
416 channel->is_deleted = true;
417 call_rcu(&channel->node.head, free_channel_rcu);
418 end:
419 pthread_mutex_unlock(&channel->lock);
420 pthread_mutex_unlock(&the_consumer_data.lock);
421 }
422
423 /*
424 * Iterate over the relayd hash table and destroy each element. Finally,
425 * destroy the whole hash table.
426 */
427 static void cleanup_relayd_ht(void)
428 {
429 struct lttng_ht_iter iter;
430 struct consumer_relayd_sock_pair *relayd;
431
432 rcu_read_lock();
433
434 cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
435 relayd, node.node) {
436 consumer_destroy_relayd(relayd);
437 }
438
439 rcu_read_unlock();
440
441 lttng_ht_destroy(the_consumer_data.relayd_ht);
442 }
443
444 /*
445 * Update the end point status of all streams having the given network sequence
446 * index (relayd index).
447 *
448 * It's atomically set without having the stream mutex locked which is fine
449 * because we handle the write/read race with a pipe wakeup for each thread.
450 */
451 static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
452 enum consumer_endpoint_status status)
453 {
454 struct lttng_ht_iter iter;
455 struct lttng_consumer_stream *stream;
456
457 DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
458
459 rcu_read_lock();
460
461 /* Let's begin with metadata */
462 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
463 if (stream->net_seq_idx == net_seq_idx) {
464 uatomic_set(&stream->endpoint_status, status);
465 DBG("Delete flag set to metadata stream %d", stream->wait_fd);
466 }
467 }
468
469 /* Follow up by the data streams */
470 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
471 if (stream->net_seq_idx == net_seq_idx) {
472 uatomic_set(&stream->endpoint_status, status);
473 DBG("Delete flag set to data stream %d", stream->wait_fd);
474 }
475 }
476 rcu_read_unlock();
477 }
478
479 /*
480 * Cleanup a relayd object by flagging every associated streams for deletion,
481 * destroying the object meaning removing it from the relayd hash table,
482 * closing the sockets and freeing the memory in a RCU call.
483 *
484 * If a local data context is available, notify the threads that the streams'
485 * state have changed.
486 */
487 void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
488 {
489 uint64_t netidx;
490
491 LTTNG_ASSERT(relayd);
492
493 DBG("Cleaning up relayd object ID %" PRIu64, relayd->net_seq_idx);
494
495 /* Save the net sequence index before destroying the object */
496 netidx = relayd->net_seq_idx;
497
498 /*
499 * Delete the relayd from the relayd hash table, close the sockets and free
500 * the object in a RCU call.
501 */
502 consumer_destroy_relayd(relayd);
503
504 /* Set inactive endpoint to all streams */
505 update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE);
506
507 /*
508 * With a local data context, notify the threads that the streams' state
509 * have changed. The write() action on the pipe acts as an "implicit"
510 * memory barrier ordering the updates of the end point status from the
511 * read of this status which happens AFTER receiving this notify.
512 */
513 notify_thread_lttng_pipe(relayd->ctx->consumer_data_pipe);
514 notify_thread_lttng_pipe(relayd->ctx->consumer_metadata_pipe);
515 }
516
517 /*
518 * Flag a relayd socket pair for destruction. Destroy it if the refcount
519 * reaches zero.
520 *
521 * RCU read side lock MUST be aquired before calling this function.
522 */
523 void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
524 {
525 LTTNG_ASSERT(relayd);
526 ASSERT_RCU_READ_LOCKED();
527
528 /* Set destroy flag for this object */
529 uatomic_set(&relayd->destroy_flag, 1);
530
531 /* Destroy the relayd if refcount is 0 */
532 if (uatomic_read(&relayd->refcount) == 0) {
533 consumer_destroy_relayd(relayd);
534 }
535 }
536
537 /*
538 * Completly destroy stream from every visiable data structure and the given
539 * hash table if one.
540 *
541 * One this call returns, the stream object is not longer usable nor visible.
542 */
543 void consumer_del_stream(struct lttng_consumer_stream *stream,
544 struct lttng_ht *ht)
545 {
546 consumer_stream_destroy(stream, ht);
547 }
548
549 /*
550 * XXX naming of del vs destroy is all mixed up.
551 */
552 void consumer_del_stream_for_data(struct lttng_consumer_stream *stream)
553 {
554 consumer_stream_destroy(stream, data_ht);
555 }
556
557 void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream)
558 {
559 consumer_stream_destroy(stream, metadata_ht);
560 }
561
562 void consumer_stream_update_channel_attributes(
563 struct lttng_consumer_stream *stream,
564 struct lttng_consumer_channel *channel)
565 {
566 stream->channel_read_only_attributes.tracefile_size =
567 channel->tracefile_size;
568 }
569
570 /*
571 * Add a stream to the global list protected by a mutex.
572 */
573 void consumer_add_data_stream(struct lttng_consumer_stream *stream)
574 {
575 struct lttng_ht *ht = data_ht;
576
577 LTTNG_ASSERT(stream);
578 LTTNG_ASSERT(ht);
579
580 DBG3("Adding consumer stream %" PRIu64, stream->key);
581
582 pthread_mutex_lock(&the_consumer_data.lock);
583 pthread_mutex_lock(&stream->chan->lock);
584 pthread_mutex_lock(&stream->chan->timer_lock);
585 pthread_mutex_lock(&stream->lock);
586 rcu_read_lock();
587
588 /* Steal stream identifier to avoid having streams with the same key */
589 steal_stream_key(stream->key, ht);
590
591 lttng_ht_add_unique_u64(ht, &stream->node);
592
593 lttng_ht_add_u64(the_consumer_data.stream_per_chan_id_ht,
594 &stream->node_channel_id);
595
596 /*
597 * Add stream to the stream_list_ht of the consumer data. No need to steal
598 * the key since the HT does not use it and we allow to add redundant keys
599 * into this table.
600 */
601 lttng_ht_add_u64(the_consumer_data.stream_list_ht,
602 &stream->node_session_id);
603
604 /*
605 * When nb_init_stream_left reaches 0, we don't need to trigger any action
606 * in terms of destroying the associated channel, because the action that
607 * causes the count to become 0 also causes a stream to be added. The
608 * channel deletion will thus be triggered by the following removal of this
609 * stream.
610 */
611 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
612 /* Increment refcount before decrementing nb_init_stream_left */
613 cmm_smp_wmb();
614 uatomic_dec(&stream->chan->nb_init_stream_left);
615 }
616
617 /* Update consumer data once the node is inserted. */
618 the_consumer_data.stream_count++;
619 the_consumer_data.need_update = 1;
620
621 rcu_read_unlock();
622 pthread_mutex_unlock(&stream->lock);
623 pthread_mutex_unlock(&stream->chan->timer_lock);
624 pthread_mutex_unlock(&stream->chan->lock);
625 pthread_mutex_unlock(&the_consumer_data.lock);
626 }
627
628 /*
629 * Add relayd socket to global consumer data hashtable. RCU read side lock MUST
630 * be acquired before calling this.
631 */
632 static int add_relayd(struct consumer_relayd_sock_pair *relayd)
633 {
634 int ret = 0;
635 struct lttng_ht_node_u64 *node;
636 struct lttng_ht_iter iter;
637
638 LTTNG_ASSERT(relayd);
639 ASSERT_RCU_READ_LOCKED();
640
641 lttng_ht_lookup(the_consumer_data.relayd_ht, &relayd->net_seq_idx,
642 &iter);
643 node = lttng_ht_iter_get_node_u64(&iter);
644 if (node != NULL) {
645 goto end;
646 }
647 lttng_ht_add_unique_u64(the_consumer_data.relayd_ht, &relayd->node);
648
649 end:
650 return ret;
651 }
652
653 /*
654 * Allocate and return a consumer relayd socket.
655 */
656 static struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
657 uint64_t net_seq_idx)
658 {
659 struct consumer_relayd_sock_pair *obj = NULL;
660
661 /* net sequence index of -1 is a failure */
662 if (net_seq_idx == (uint64_t) -1ULL) {
663 goto error;
664 }
665
666 obj = (consumer_relayd_sock_pair *) zmalloc(sizeof(struct consumer_relayd_sock_pair));
667 if (obj == NULL) {
668 PERROR("zmalloc relayd sock");
669 goto error;
670 }
671
672 obj->net_seq_idx = net_seq_idx;
673 obj->refcount = 0;
674 obj->destroy_flag = 0;
675 obj->control_sock.sock.fd = -1;
676 obj->data_sock.sock.fd = -1;
677 lttng_ht_node_init_u64(&obj->node, obj->net_seq_idx);
678 pthread_mutex_init(&obj->ctrl_sock_mutex, NULL);
679
680 error:
681 return obj;
682 }
683
684 /*
685 * Find a relayd socket pair in the global consumer data.
686 *
687 * Return the object if found else NULL.
688 * RCU read-side lock must be held across this call and while using the
689 * returned object.
690 */
691 struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
692 {
693 struct lttng_ht_iter iter;
694 struct lttng_ht_node_u64 *node;
695 struct consumer_relayd_sock_pair *relayd = NULL;
696
697 ASSERT_RCU_READ_LOCKED();
698
699 /* Negative keys are lookup failures */
700 if (key == (uint64_t) -1ULL) {
701 goto error;
702 }
703
704 lttng_ht_lookup(the_consumer_data.relayd_ht, &key, &iter);
705 node = lttng_ht_iter_get_node_u64(&iter);
706 if (node != NULL) {
707 relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
708 }
709
710 error:
711 return relayd;
712 }
713
714 /*
715 * Find a relayd and send the stream
716 *
717 * Returns 0 on success, < 0 on error
718 */
719 int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
720 char *path)
721 {
722 int ret = 0;
723 struct consumer_relayd_sock_pair *relayd;
724
725 LTTNG_ASSERT(stream);
726 LTTNG_ASSERT(stream->net_seq_idx != -1ULL);
727 LTTNG_ASSERT(path);
728
729 /* The stream is not metadata. Get relayd reference if exists. */
730 rcu_read_lock();
731 relayd = consumer_find_relayd(stream->net_seq_idx);
732 if (relayd != NULL) {
733 /* Add stream on the relayd */
734 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
735 ret = relayd_add_stream(&relayd->control_sock, stream->name,
736 get_consumer_domain(), path, &stream->relayd_stream_id,
737 stream->chan->tracefile_size,
738 stream->chan->tracefile_count,
739 stream->trace_chunk);
740 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
741 if (ret < 0) {
742 ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
743 lttng_consumer_cleanup_relayd(relayd);
744 goto end;
745 }
746
747 uatomic_inc(&relayd->refcount);
748 stream->sent_to_relayd = 1;
749 } else {
750 ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.",
751 stream->key, stream->net_seq_idx);
752 ret = -1;
753 goto end;
754 }
755
756 DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64,
757 stream->name, stream->key, stream->net_seq_idx);
758
759 end:
760 rcu_read_unlock();
761 return ret;
762 }
763
764 /*
765 * Find a relayd and send the streams sent message
766 *
767 * Returns 0 on success, < 0 on error
768 */
769 int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
770 {
771 int ret = 0;
772 struct consumer_relayd_sock_pair *relayd;
773
774 LTTNG_ASSERT(net_seq_idx != -1ULL);
775
776 /* The stream is not metadata. Get relayd reference if exists. */
777 rcu_read_lock();
778 relayd = consumer_find_relayd(net_seq_idx);
779 if (relayd != NULL) {
780 /* Add stream on the relayd */
781 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
782 ret = relayd_streams_sent(&relayd->control_sock);
783 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
784 if (ret < 0) {
785 ERR("Relayd streams sent failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
786 lttng_consumer_cleanup_relayd(relayd);
787 goto end;
788 }
789 } else {
790 ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
791 net_seq_idx);
792 ret = -1;
793 goto end;
794 }
795
796 ret = 0;
797 DBG("All streams sent relayd id %" PRIu64, net_seq_idx);
798
799 end:
800 rcu_read_unlock();
801 return ret;
802 }
803
804 /*
805 * Find a relayd and close the stream
806 */
807 void close_relayd_stream(struct lttng_consumer_stream *stream)
808 {
809 struct consumer_relayd_sock_pair *relayd;
810
811 /* The stream is not metadata. Get relayd reference if exists. */
812 rcu_read_lock();
813 relayd = consumer_find_relayd(stream->net_seq_idx);
814 if (relayd) {
815 consumer_stream_relayd_close(stream, relayd);
816 }
817 rcu_read_unlock();
818 }
819
820 /*
821 * Handle stream for relayd transmission if the stream applies for network
822 * streaming where the net sequence index is set.
823 *
824 * Return destination file descriptor or negative value on error.
825 */
826 static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
827 size_t data_size, unsigned long padding,
828 struct consumer_relayd_sock_pair *relayd)
829 {
830 int outfd = -1, ret;
831 struct lttcomm_relayd_data_hdr data_hdr;
832
833 /* Safety net */
834 LTTNG_ASSERT(stream);
835 LTTNG_ASSERT(relayd);
836
837 /* Reset data header */
838 memset(&data_hdr, 0, sizeof(data_hdr));
839
840 if (stream->metadata_flag) {
841 /* Caller MUST acquire the relayd control socket lock */
842 ret = relayd_send_metadata(&relayd->control_sock, data_size);
843 if (ret < 0) {
844 goto error;
845 }
846
847 /* Metadata are always sent on the control socket. */
848 outfd = relayd->control_sock.sock.fd;
849 } else {
850 /* Set header with stream information */
851 data_hdr.stream_id = htobe64(stream->relayd_stream_id);
852 data_hdr.data_size = htobe32(data_size);
853 data_hdr.padding_size = htobe32(padding);
854
855 /*
856 * Note that net_seq_num below is assigned with the *current* value of
857 * next_net_seq_num and only after that the next_net_seq_num will be
858 * increment. This is why when issuing a command on the relayd using
859 * this next value, 1 should always be substracted in order to compare
860 * the last seen sequence number on the relayd side to the last sent.
861 */
862 data_hdr.net_seq_num = htobe64(stream->next_net_seq_num);
863 /* Other fields are zeroed previously */
864
865 ret = relayd_send_data_hdr(&relayd->data_sock, &data_hdr,
866 sizeof(data_hdr));
867 if (ret < 0) {
868 goto error;
869 }
870
871 ++stream->next_net_seq_num;
872
873 /* Set to go on data socket */
874 outfd = relayd->data_sock.sock.fd;
875 }
876
877 error:
878 return outfd;
879 }
880
881 /*
882 * Write a character on the metadata poll pipe to wake the metadata thread.
883 * Returns 0 on success, -1 on error.
884 */
885 int consumer_metadata_wakeup_pipe(const struct lttng_consumer_channel *channel)
886 {
887 int ret = 0;
888
889 DBG("Waking up metadata poll thread (writing to pipe): channel name = '%s'",
890 channel->name);
891 if (channel->monitor && channel->metadata_stream) {
892 const char dummy = 'c';
893 const ssize_t write_ret = lttng_write(
894 channel->metadata_stream->ust_metadata_poll_pipe[1],
895 &dummy, 1);
896
897 if (write_ret < 1) {
898 if (errno == EWOULDBLOCK) {
899 /*
900 * This is fine, the metadata poll thread
901 * is having a hard time keeping-up, but
902 * it will eventually wake-up and consume
903 * the available data.
904 */
905 ret = 0;
906 } else {
907 PERROR("Failed to write to UST metadata pipe while attempting to wake-up the metadata poll thread");
908 ret = -1;
909 goto end;
910 }
911 }
912 }
913
914 end:
915 return ret;
916 }
917
918 /*
919 * Trigger a dump of the metadata content. Following/during the succesful
920 * completion of this call, the metadata poll thread will start receiving
921 * metadata packets to consume.
922 *
923 * The caller must hold the channel and stream locks.
924 */
925 static
926 int consumer_metadata_stream_dump(struct lttng_consumer_stream *stream)
927 {
928 int ret;
929
930 ASSERT_LOCKED(stream->chan->lock);
931 ASSERT_LOCKED(stream->lock);
932 LTTNG_ASSERT(stream->metadata_flag);
933 LTTNG_ASSERT(stream->chan->trace_chunk);
934
935 switch (the_consumer_data.type) {
936 case LTTNG_CONSUMER_KERNEL:
937 /*
938 * Reset the position of what has been read from the
939 * metadata cache to 0 so we can dump it again.
940 */
941 ret = kernctl_metadata_cache_dump(stream->wait_fd);
942 break;
943 case LTTNG_CONSUMER32_UST:
944 case LTTNG_CONSUMER64_UST:
945 /*
946 * Reset the position pushed from the metadata cache so it
947 * will write from the beginning on the next push.
948 */
949 stream->ust_metadata_pushed = 0;
950 ret = consumer_metadata_wakeup_pipe(stream->chan);
951 break;
952 default:
953 ERR("Unknown consumer_data type");
954 abort();
955 }
956 if (ret < 0) {
957 ERR("Failed to dump the metadata cache");
958 }
959 return ret;
960 }
961
962 static
963 int lttng_consumer_channel_set_trace_chunk(
964 struct lttng_consumer_channel *channel,
965 struct lttng_trace_chunk *new_trace_chunk)
966 {
967 pthread_mutex_lock(&channel->lock);
968 if (channel->is_deleted) {
969 /*
970 * The channel has been logically deleted and should no longer
971 * be used. It has released its reference to its current trace
972 * chunk and should not acquire a new one.
973 *
974 * Return success as there is nothing for the caller to do.
975 */
976 goto end;
977 }
978
979 /*
980 * The acquisition of the reference cannot fail (barring
981 * a severe internal error) since a reference to the published
982 * chunk is already held by the caller.
983 */
984 if (new_trace_chunk) {
985 const bool acquired_reference = lttng_trace_chunk_get(
986 new_trace_chunk);
987
988 LTTNG_ASSERT(acquired_reference);
989 }
990
991 lttng_trace_chunk_put(channel->trace_chunk);
992 channel->trace_chunk = new_trace_chunk;
993 end:
994 pthread_mutex_unlock(&channel->lock);
995 return 0;
996 }
997
998 /*
999 * Allocate and return a new lttng_consumer_channel object using the given key
1000 * to initialize the hash table node.
1001 *
1002 * On error, return NULL.
1003 */
1004 struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
1005 uint64_t session_id,
1006 const uint64_t *chunk_id,
1007 const char *pathname,
1008 const char *name,
1009 uint64_t relayd_id,
1010 enum lttng_event_output output,
1011 uint64_t tracefile_size,
1012 uint64_t tracefile_count,
1013 uint64_t session_id_per_pid,
1014 unsigned int monitor,
1015 unsigned int live_timer_interval,
1016 bool is_in_live_session,
1017 const char *root_shm_path,
1018 const char *shm_path)
1019 {
1020 struct lttng_consumer_channel *channel = NULL;
1021 struct lttng_trace_chunk *trace_chunk = NULL;
1022
1023 if (chunk_id) {
1024 trace_chunk = lttng_trace_chunk_registry_find_chunk(
1025 the_consumer_data.chunk_registry, session_id,
1026 *chunk_id);
1027 if (!trace_chunk) {
1028 ERR("Failed to find trace chunk reference during creation of channel");
1029 goto end;
1030 }
1031 }
1032
1033 channel = (lttng_consumer_channel *) zmalloc(sizeof(*channel));
1034 if (channel == NULL) {
1035 PERROR("malloc struct lttng_consumer_channel");
1036 goto end;
1037 }
1038
1039 channel->key = key;
1040 channel->refcount = 0;
1041 channel->session_id = session_id;
1042 channel->session_id_per_pid = session_id_per_pid;
1043 channel->relayd_id = relayd_id;
1044 channel->tracefile_size = tracefile_size;
1045 channel->tracefile_count = tracefile_count;
1046 channel->monitor = monitor;
1047 channel->live_timer_interval = live_timer_interval;
1048 channel->is_live = is_in_live_session;
1049 pthread_mutex_init(&channel->lock, NULL);
1050 pthread_mutex_init(&channel->timer_lock, NULL);
1051
1052 switch (output) {
1053 case LTTNG_EVENT_SPLICE:
1054 channel->output = CONSUMER_CHANNEL_SPLICE;
1055 break;
1056 case LTTNG_EVENT_MMAP:
1057 channel->output = CONSUMER_CHANNEL_MMAP;
1058 break;
1059 default:
1060 abort();
1061 free(channel);
1062 channel = NULL;
1063 goto end;
1064 }
1065
1066 /*
1067 * In monitor mode, the streams associated with the channel will be put in
1068 * a special list ONLY owned by this channel. So, the refcount is set to 1
1069 * here meaning that the channel itself has streams that are referenced.
1070 *
1071 * On a channel deletion, once the channel is no longer visible, the
1072 * refcount is decremented and checked for a zero value to delete it. With
1073 * streams in no monitor mode, it will now be safe to destroy the channel.
1074 */
1075 if (!channel->monitor) {
1076 channel->refcount = 1;
1077 }
1078
1079 strncpy(channel->pathname, pathname, sizeof(channel->pathname));
1080 channel->pathname[sizeof(channel->pathname) - 1] = '\0';
1081
1082 strncpy(channel->name, name, sizeof(channel->name));
1083 channel->name[sizeof(channel->name) - 1] = '\0';
1084
1085 if (root_shm_path) {
1086 strncpy(channel->root_shm_path, root_shm_path, sizeof(channel->root_shm_path));
1087 channel->root_shm_path[sizeof(channel->root_shm_path) - 1] = '\0';
1088 }
1089 if (shm_path) {
1090 strncpy(channel->shm_path, shm_path, sizeof(channel->shm_path));
1091 channel->shm_path[sizeof(channel->shm_path) - 1] = '\0';
1092 }
1093
1094 lttng_ht_node_init_u64(&channel->node, channel->key);
1095 lttng_ht_node_init_u64(&channel->channels_by_session_id_ht_node,
1096 channel->session_id);
1097
1098 channel->wait_fd = -1;
1099 CDS_INIT_LIST_HEAD(&channel->streams.head);
1100
1101 if (trace_chunk) {
1102 int ret = lttng_consumer_channel_set_trace_chunk(channel,
1103 trace_chunk);
1104 if (ret) {
1105 goto error;
1106 }
1107 }
1108
1109 DBG("Allocated channel (key %" PRIu64 ")", channel->key);
1110
1111 end:
1112 lttng_trace_chunk_put(trace_chunk);
1113 return channel;
1114 error:
1115 consumer_del_channel(channel);
1116 channel = NULL;
1117 goto end;
1118 }
1119
1120 /*
1121 * Add a channel to the global list protected by a mutex.
1122 *
1123 * Always return 0 indicating success.
1124 */
1125 int consumer_add_channel(struct lttng_consumer_channel *channel,
1126 struct lttng_consumer_local_data *ctx)
1127 {
1128 pthread_mutex_lock(&the_consumer_data.lock);
1129 pthread_mutex_lock(&channel->lock);
1130 pthread_mutex_lock(&channel->timer_lock);
1131
1132 /*
1133 * This gives us a guarantee that the channel we are about to add to the
1134 * channel hash table will be unique. See this function comment on the why
1135 * we need to steel the channel key at this stage.
1136 */
1137 steal_channel_key(channel->key);
1138
1139 rcu_read_lock();
1140 lttng_ht_add_unique_u64(the_consumer_data.channel_ht, &channel->node);
1141 lttng_ht_add_u64(the_consumer_data.channels_by_session_id_ht,
1142 &channel->channels_by_session_id_ht_node);
1143 rcu_read_unlock();
1144 channel->is_published = true;
1145
1146 pthread_mutex_unlock(&channel->timer_lock);
1147 pthread_mutex_unlock(&channel->lock);
1148 pthread_mutex_unlock(&the_consumer_data.lock);
1149
1150 if (channel->wait_fd != -1 && channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
1151 notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
1152 }
1153
1154 return 0;
1155 }
1156
1157 /*
1158 * Allocate the pollfd structure and the local view of the out fds to avoid
1159 * doing a lookup in the linked list and concurrency issues when writing is
1160 * needed. Called with consumer_data.lock held.
1161 *
1162 * Returns the number of fds in the structures.
1163 */
1164 static int update_poll_array(struct lttng_consumer_local_data *ctx,
1165 struct pollfd **pollfd, struct lttng_consumer_stream **local_stream,
1166 struct lttng_ht *ht, int *nb_inactive_fd)
1167 {
1168 int i = 0;
1169 struct lttng_ht_iter iter;
1170 struct lttng_consumer_stream *stream;
1171
1172 LTTNG_ASSERT(ctx);
1173 LTTNG_ASSERT(ht);
1174 LTTNG_ASSERT(pollfd);
1175 LTTNG_ASSERT(local_stream);
1176
1177 DBG("Updating poll fd array");
1178 *nb_inactive_fd = 0;
1179 rcu_read_lock();
1180 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1181 /*
1182 * Only active streams with an active end point can be added to the
1183 * poll set and local stream storage of the thread.
1184 *
1185 * There is a potential race here for endpoint_status to be updated
1186 * just after the check. However, this is OK since the stream(s) will
1187 * be deleted once the thread is notified that the end point state has
1188 * changed where this function will be called back again.
1189 *
1190 * We track the number of inactive FDs because they still need to be
1191 * closed by the polling thread after a wakeup on the data_pipe or
1192 * metadata_pipe.
1193 */
1194 if (stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
1195 (*nb_inactive_fd)++;
1196 continue;
1197 }
1198 /*
1199 * This clobbers way too much the debug output. Uncomment that if you
1200 * need it for debugging purposes.
1201 */
1202 (*pollfd)[i].fd = stream->wait_fd;
1203 (*pollfd)[i].events = POLLIN | POLLPRI;
1204 local_stream[i] = stream;
1205 i++;
1206 }
1207 rcu_read_unlock();
1208
1209 /*
1210 * Insert the consumer_data_pipe at the end of the array and don't
1211 * increment i so nb_fd is the number of real FD.
1212 */
1213 (*pollfd)[i].fd = lttng_pipe_get_readfd(ctx->consumer_data_pipe);
1214 (*pollfd)[i].events = POLLIN | POLLPRI;
1215
1216 (*pollfd)[i + 1].fd = lttng_pipe_get_readfd(ctx->consumer_wakeup_pipe);
1217 (*pollfd)[i + 1].events = POLLIN | POLLPRI;
1218 return i;
1219 }
1220
1221 /*
1222 * Poll on the should_quit pipe and the command socket return -1 on
1223 * error, 1 if should exit, 0 if data is available on the command socket
1224 */
1225 int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
1226 {
1227 int num_rdy;
1228
1229 restart:
1230 num_rdy = poll(consumer_sockpoll, 2, -1);
1231 if (num_rdy == -1) {
1232 /*
1233 * Restart interrupted system call.
1234 */
1235 if (errno == EINTR) {
1236 goto restart;
1237 }
1238 PERROR("Poll error");
1239 return -1;
1240 }
1241 if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) {
1242 DBG("consumer_should_quit wake up");
1243 return 1;
1244 }
1245 return 0;
1246 }
1247
1248 /*
1249 * Set the error socket.
1250 */
1251 void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
1252 int sock)
1253 {
1254 ctx->consumer_error_socket = sock;
1255 }
1256
1257 /*
1258 * Set the command socket path.
1259 */
1260 void lttng_consumer_set_command_sock_path(
1261 struct lttng_consumer_local_data *ctx, char *sock)
1262 {
1263 ctx->consumer_command_sock_path = sock;
1264 }
1265
1266 /*
1267 * Send return code to the session daemon.
1268 * If the socket is not defined, we return 0, it is not a fatal error
1269 */
1270 int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
1271 {
1272 if (ctx->consumer_error_socket > 0) {
1273 return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
1274 sizeof(enum lttcomm_sessiond_command));
1275 }
1276
1277 return 0;
1278 }
1279
1280 /*
1281 * Close all the tracefiles and stream fds and MUST be called when all
1282 * instances are destroyed i.e. when all threads were joined and are ended.
1283 */
1284 void lttng_consumer_cleanup(void)
1285 {
1286 struct lttng_ht_iter iter;
1287 struct lttng_consumer_channel *channel;
1288 unsigned int trace_chunks_left;
1289
1290 rcu_read_lock();
1291
1292 cds_lfht_for_each_entry(the_consumer_data.channel_ht->ht, &iter.iter,
1293 channel, node.node) {
1294 consumer_del_channel(channel);
1295 }
1296
1297 rcu_read_unlock();
1298
1299 lttng_ht_destroy(the_consumer_data.channel_ht);
1300 lttng_ht_destroy(the_consumer_data.channels_by_session_id_ht);
1301
1302 cleanup_relayd_ht();
1303
1304 lttng_ht_destroy(the_consumer_data.stream_per_chan_id_ht);
1305
1306 /*
1307 * This HT contains streams that are freed by either the metadata thread or
1308 * the data thread so we do *nothing* on the hash table and simply destroy
1309 * it.
1310 */
1311 lttng_ht_destroy(the_consumer_data.stream_list_ht);
1312
1313 /*
1314 * Trace chunks in the registry may still exist if the session
1315 * daemon has encountered an internal error and could not
1316 * tear down its sessions and/or trace chunks properly.
1317 *
1318 * Release the session daemon's implicit reference to any remaining
1319 * trace chunk and print an error if any trace chunk was found. Note
1320 * that there are _no_ legitimate cases for trace chunks to be left,
1321 * it is a leak. However, it can happen following a crash of the
1322 * session daemon and not emptying the registry would cause an assertion
1323 * to hit.
1324 */
1325 trace_chunks_left = lttng_trace_chunk_registry_put_each_chunk(
1326 the_consumer_data.chunk_registry);
1327 if (trace_chunks_left) {
1328 ERR("%u trace chunks are leaked by lttng-consumerd. "
1329 "This can be caused by an internal error of the session daemon.",
1330 trace_chunks_left);
1331 }
1332 /* Run all callbacks freeing each chunk. */
1333 rcu_barrier();
1334 lttng_trace_chunk_registry_destroy(the_consumer_data.chunk_registry);
1335 }
1336
1337 /*
1338 * Called from signal handler.
1339 */
1340 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
1341 {
1342 ssize_t ret;
1343
1344 CMM_STORE_SHARED(consumer_quit, 1);
1345 ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
1346 if (ret < 1) {
1347 PERROR("write consumer quit");
1348 }
1349
1350 DBG("Consumer flag that it should quit");
1351 }
1352
1353
1354 /*
1355 * Flush pending writes to trace output disk file.
1356 */
1357 static
1358 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
1359 off_t orig_offset)
1360 {
1361 int ret;
1362 int outfd = stream->out_fd;
1363
1364 /*
1365 * This does a blocking write-and-wait on any page that belongs to the
1366 * subbuffer prior to the one we just wrote.
1367 * Don't care about error values, as these are just hints and ways to
1368 * limit the amount of page cache used.
1369 */
1370 if (orig_offset < stream->max_sb_size) {
1371 return;
1372 }
1373 lttng_sync_file_range(outfd, orig_offset - stream->max_sb_size,
1374 stream->max_sb_size,
1375 SYNC_FILE_RANGE_WAIT_BEFORE
1376 | SYNC_FILE_RANGE_WRITE
1377 | SYNC_FILE_RANGE_WAIT_AFTER);
1378 /*
1379 * Give hints to the kernel about how we access the file:
1380 * POSIX_FADV_DONTNEED : we won't re-access data in a near future after
1381 * we write it.
1382 *
1383 * We need to call fadvise again after the file grows because the
1384 * kernel does not seem to apply fadvise to non-existing parts of the
1385 * file.
1386 *
1387 * Call fadvise _after_ having waited for the page writeback to
1388 * complete because the dirty page writeback semantic is not well
1389 * defined. So it can be expected to lead to lower throughput in
1390 * streaming.
1391 */
1392 ret = posix_fadvise(outfd, orig_offset - stream->max_sb_size,
1393 stream->max_sb_size, POSIX_FADV_DONTNEED);
1394 if (ret && ret != -ENOSYS) {
1395 errno = ret;
1396 PERROR("posix_fadvise on fd %i", outfd);
1397 }
1398 }
1399
1400 /*
1401 * Initialise the necessary environnement :
1402 * - create a new context
1403 * - create the poll_pipe
1404 * - create the should_quit pipe (for signal handler)
1405 * - create the thread pipe (for splice)
1406 *
1407 * Takes a function pointer as argument, this function is called when data is
1408 * available on a buffer. This function is responsible to do the
1409 * kernctl_get_next_subbuf, read the data with mmap or splice depending on the
1410 * buffer configuration and then kernctl_put_next_subbuf at the end.
1411 *
1412 * Returns a pointer to the new context or NULL on error.
1413 */
1414 struct lttng_consumer_local_data *lttng_consumer_create(
1415 enum lttng_consumer_type type,
1416 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
1417 struct lttng_consumer_local_data *ctx, bool locked_by_caller),
1418 int (*recv_channel)(struct lttng_consumer_channel *channel),
1419 int (*recv_stream)(struct lttng_consumer_stream *stream),
1420 int (*update_stream)(uint64_t stream_key, uint32_t state))
1421 {
1422 int ret;
1423 struct lttng_consumer_local_data *ctx;
1424
1425 LTTNG_ASSERT(the_consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
1426 the_consumer_data.type == type);
1427 the_consumer_data.type = type;
1428
1429 ctx = (lttng_consumer_local_data *) zmalloc(sizeof(struct lttng_consumer_local_data));
1430 if (ctx == NULL) {
1431 PERROR("allocating context");
1432 goto error;
1433 }
1434
1435 ctx->consumer_error_socket = -1;
1436 ctx->consumer_metadata_socket = -1;
1437 pthread_mutex_init(&ctx->metadata_socket_lock, NULL);
1438 /* assign the callbacks */
1439 ctx->on_buffer_ready = buffer_ready;
1440 ctx->on_recv_channel = recv_channel;
1441 ctx->on_recv_stream = recv_stream;
1442 ctx->on_update_stream = update_stream;
1443
1444 ctx->consumer_data_pipe = lttng_pipe_open(0);
1445 if (!ctx->consumer_data_pipe) {
1446 goto error_poll_pipe;
1447 }
1448
1449 ctx->consumer_wakeup_pipe = lttng_pipe_open(0);
1450 if (!ctx->consumer_wakeup_pipe) {
1451 goto error_wakeup_pipe;
1452 }
1453
1454 ret = pipe(ctx->consumer_should_quit);
1455 if (ret < 0) {
1456 PERROR("Error creating recv pipe");
1457 goto error_quit_pipe;
1458 }
1459
1460 ret = pipe(ctx->consumer_channel_pipe);
1461 if (ret < 0) {
1462 PERROR("Error creating channel pipe");
1463 goto error_channel_pipe;
1464 }
1465
1466 ctx->consumer_metadata_pipe = lttng_pipe_open(0);
1467 if (!ctx->consumer_metadata_pipe) {
1468 goto error_metadata_pipe;
1469 }
1470
1471 ctx->channel_monitor_pipe = -1;
1472
1473 return ctx;
1474
1475 error_metadata_pipe:
1476 utils_close_pipe(ctx->consumer_channel_pipe);
1477 error_channel_pipe:
1478 utils_close_pipe(ctx->consumer_should_quit);
1479 error_quit_pipe:
1480 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
1481 error_wakeup_pipe:
1482 lttng_pipe_destroy(ctx->consumer_data_pipe);
1483 error_poll_pipe:
1484 free(ctx);
1485 error:
1486 return NULL;
1487 }
1488
1489 /*
1490 * Iterate over all streams of the hashtable and free them properly.
1491 */
1492 static void destroy_data_stream_ht(struct lttng_ht *ht)
1493 {
1494 struct lttng_ht_iter iter;
1495 struct lttng_consumer_stream *stream;
1496
1497 if (ht == NULL) {
1498 return;
1499 }
1500
1501 rcu_read_lock();
1502 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1503 /*
1504 * Ignore return value since we are currently cleaning up so any error
1505 * can't be handled.
1506 */
1507 (void) consumer_del_stream(stream, ht);
1508 }
1509 rcu_read_unlock();
1510
1511 lttng_ht_destroy(ht);
1512 }
1513
1514 /*
1515 * Iterate over all streams of the metadata hashtable and free them
1516 * properly.
1517 */
1518 static void destroy_metadata_stream_ht(struct lttng_ht *ht)
1519 {
1520 struct lttng_ht_iter iter;
1521 struct lttng_consumer_stream *stream;
1522
1523 if (ht == NULL) {
1524 return;
1525 }
1526
1527 rcu_read_lock();
1528 cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
1529 /*
1530 * Ignore return value since we are currently cleaning up so any error
1531 * can't be handled.
1532 */
1533 (void) consumer_del_metadata_stream(stream, ht);
1534 }
1535 rcu_read_unlock();
1536
1537 lttng_ht_destroy(ht);
1538 }
1539
1540 /*
1541 * Close all fds associated with the instance and free the context.
1542 */
1543 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
1544 {
1545 int ret;
1546
1547 DBG("Consumer destroying it. Closing everything.");
1548
1549 if (!ctx) {
1550 return;
1551 }
1552
1553 destroy_data_stream_ht(data_ht);
1554 destroy_metadata_stream_ht(metadata_ht);
1555
1556 ret = close(ctx->consumer_error_socket);
1557 if (ret) {
1558 PERROR("close");
1559 }
1560 ret = close(ctx->consumer_metadata_socket);
1561 if (ret) {
1562 PERROR("close");
1563 }
1564 utils_close_pipe(ctx->consumer_channel_pipe);
1565 lttng_pipe_destroy(ctx->consumer_data_pipe);
1566 lttng_pipe_destroy(ctx->consumer_metadata_pipe);
1567 lttng_pipe_destroy(ctx->consumer_wakeup_pipe);
1568 utils_close_pipe(ctx->consumer_should_quit);
1569
1570 unlink(ctx->consumer_command_sock_path);
1571 free(ctx);
1572 }
1573
1574 /*
1575 * Write the metadata stream id on the specified file descriptor.
1576 */
1577 static int write_relayd_metadata_id(int fd,
1578 struct lttng_consumer_stream *stream,
1579 unsigned long padding)
1580 {
1581 ssize_t ret;
1582 struct lttcomm_relayd_metadata_payload hdr;
1583
1584 hdr.stream_id = htobe64(stream->relayd_stream_id);
1585 hdr.padding_size = htobe32(padding);
1586 ret = lttng_write(fd, (void *) &hdr, sizeof(hdr));
1587 if (ret < sizeof(hdr)) {
1588 /*
1589 * This error means that the fd's end is closed so ignore the PERROR
1590 * not to clubber the error output since this can happen in a normal
1591 * code path.
1592 */
1593 if (errno != EPIPE) {
1594 PERROR("write metadata stream id");
1595 }
1596 DBG3("Consumer failed to write relayd metadata id (errno: %d)", errno);
1597 /*
1598 * Set ret to a negative value because if ret != sizeof(hdr), we don't
1599 * handle writting the missing part so report that as an error and
1600 * don't lie to the caller.
1601 */
1602 ret = -1;
1603 goto end;
1604 }
1605 DBG("Metadata stream id %" PRIu64 " with padding %lu written before data",
1606 stream->relayd_stream_id, padding);
1607
1608 end:
1609 return (int) ret;
1610 }
1611
1612 /*
1613 * Mmap the ring buffer, read it and write the data to the tracefile. This is a
1614 * core function for writing trace buffers to either the local filesystem or
1615 * the network.
1616 *
1617 * It must be called with the stream and the channel lock held.
1618 *
1619 * Careful review MUST be put if any changes occur!
1620 *
1621 * Returns the number of bytes written
1622 */
1623 ssize_t lttng_consumer_on_read_subbuffer_mmap(
1624 struct lttng_consumer_stream *stream,
1625 const struct lttng_buffer_view *buffer,
1626 unsigned long padding)
1627 {
1628 ssize_t ret = 0;
1629 off_t orig_offset = stream->out_fd_offset;
1630 /* Default is on the disk */
1631 int outfd = stream->out_fd;
1632 struct consumer_relayd_sock_pair *relayd = NULL;
1633 unsigned int relayd_hang_up = 0;
1634 const size_t subbuf_content_size = buffer->size - padding;
1635 size_t write_len;
1636
1637 /* RCU lock for the relayd pointer */
1638 rcu_read_lock();
1639 LTTNG_ASSERT(stream->net_seq_idx != (uint64_t) -1ULL ||
1640 stream->trace_chunk);
1641
1642 /* Flag that the current stream if set for network streaming. */
1643 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1644 relayd = consumer_find_relayd(stream->net_seq_idx);
1645 if (relayd == NULL) {
1646 ret = -EPIPE;
1647 goto end;
1648 }
1649 }
1650
1651 /* Handle stream on the relayd if the output is on the network */
1652 if (relayd) {
1653 unsigned long netlen = subbuf_content_size;
1654
1655 /*
1656 * Lock the control socket for the complete duration of the function
1657 * since from this point on we will use the socket.
1658 */
1659 if (stream->metadata_flag) {
1660 /* Metadata requires the control socket. */
1661 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1662 if (stream->reset_metadata_flag) {
1663 ret = relayd_reset_metadata(&relayd->control_sock,
1664 stream->relayd_stream_id,
1665 stream->metadata_version);
1666 if (ret < 0) {
1667 relayd_hang_up = 1;
1668 goto write_error;
1669 }
1670 stream->reset_metadata_flag = 0;
1671 }
1672 netlen += sizeof(struct lttcomm_relayd_metadata_payload);
1673 }
1674
1675 ret = write_relayd_stream_header(stream, netlen, padding, relayd);
1676 if (ret < 0) {
1677 relayd_hang_up = 1;
1678 goto write_error;
1679 }
1680 /* Use the returned socket. */
1681 outfd = ret;
1682
1683 /* Write metadata stream id before payload */
1684 if (stream->metadata_flag) {
1685 ret = write_relayd_metadata_id(outfd, stream, padding);
1686 if (ret < 0) {
1687 relayd_hang_up = 1;
1688 goto write_error;
1689 }
1690 }
1691
1692 write_len = subbuf_content_size;
1693 } else {
1694 /* No streaming; we have to write the full padding. */
1695 if (stream->metadata_flag && stream->reset_metadata_flag) {
1696 ret = utils_truncate_stream_file(stream->out_fd, 0);
1697 if (ret < 0) {
1698 ERR("Reset metadata file");
1699 goto end;
1700 }
1701 stream->reset_metadata_flag = 0;
1702 }
1703
1704 /*
1705 * Check if we need to change the tracefile before writing the packet.
1706 */
1707 if (stream->chan->tracefile_size > 0 &&
1708 (stream->tracefile_size_current + buffer->size) >
1709 stream->chan->tracefile_size) {
1710 ret = consumer_stream_rotate_output_files(stream);
1711 if (ret) {
1712 goto end;
1713 }
1714 outfd = stream->out_fd;
1715 orig_offset = 0;
1716 }
1717 stream->tracefile_size_current += buffer->size;
1718 write_len = buffer->size;
1719 }
1720
1721 /*
1722 * This call guarantee that len or less is returned. It's impossible to
1723 * receive a ret value that is bigger than len.
1724 */
1725 ret = lttng_write(outfd, buffer->data, write_len);
1726 DBG("Consumer mmap write() ret %zd (len %zu)", ret, write_len);
1727 if (ret < 0 || ((size_t) ret != write_len)) {
1728 /*
1729 * Report error to caller if nothing was written else at least send the
1730 * amount written.
1731 */
1732 if (ret < 0) {
1733 ret = -errno;
1734 }
1735 relayd_hang_up = 1;
1736
1737 /* Socket operation failed. We consider the relayd dead */
1738 if (errno == EPIPE) {
1739 /*
1740 * This is possible if the fd is closed on the other side
1741 * (outfd) or any write problem. It can be verbose a bit for a
1742 * normal execution if for instance the relayd is stopped
1743 * abruptly. This can happen so set this to a DBG statement.
1744 */
1745 DBG("Consumer mmap write detected relayd hang up");
1746 } else {
1747 /* Unhandled error, print it and stop function right now. */
1748 PERROR("Error in write mmap (ret %zd != write_len %zu)", ret,
1749 write_len);
1750 }
1751 goto write_error;
1752 }
1753 stream->output_written += ret;
1754
1755 /* This call is useless on a socket so better save a syscall. */
1756 if (!relayd) {
1757 /* This won't block, but will start writeout asynchronously */
1758 lttng_sync_file_range(outfd, stream->out_fd_offset, write_len,
1759 SYNC_FILE_RANGE_WRITE);
1760 stream->out_fd_offset += write_len;
1761 lttng_consumer_sync_trace_file(stream, orig_offset);
1762 }
1763
1764 write_error:
1765 /*
1766 * This is a special case that the relayd has closed its socket. Let's
1767 * cleanup the relayd object and all associated streams.
1768 */
1769 if (relayd && relayd_hang_up) {
1770 ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
1771 lttng_consumer_cleanup_relayd(relayd);
1772 }
1773
1774 end:
1775 /* Unlock only if ctrl socket used */
1776 if (relayd && stream->metadata_flag) {
1777 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1778 }
1779
1780 rcu_read_unlock();
1781 return ret;
1782 }
1783
1784 /*
1785 * Splice the data from the ring buffer to the tracefile.
1786 *
1787 * It must be called with the stream lock held.
1788 *
1789 * Returns the number of bytes spliced.
1790 */
1791 ssize_t lttng_consumer_on_read_subbuffer_splice(
1792 struct lttng_consumer_local_data *ctx,
1793 struct lttng_consumer_stream *stream, unsigned long len,
1794 unsigned long padding)
1795 {
1796 ssize_t ret = 0, written = 0, ret_splice = 0;
1797 loff_t offset = 0;
1798 off_t orig_offset = stream->out_fd_offset;
1799 int fd = stream->wait_fd;
1800 /* Default is on the disk */
1801 int outfd = stream->out_fd;
1802 struct consumer_relayd_sock_pair *relayd = NULL;
1803 int *splice_pipe;
1804 unsigned int relayd_hang_up = 0;
1805
1806 switch (the_consumer_data.type) {
1807 case LTTNG_CONSUMER_KERNEL:
1808 break;
1809 case LTTNG_CONSUMER32_UST:
1810 case LTTNG_CONSUMER64_UST:
1811 /* Not supported for user space tracing */
1812 return -ENOSYS;
1813 default:
1814 ERR("Unknown consumer_data type");
1815 abort();
1816 }
1817
1818 /* RCU lock for the relayd pointer */
1819 rcu_read_lock();
1820
1821 /* Flag that the current stream if set for network streaming. */
1822 if (stream->net_seq_idx != (uint64_t) -1ULL) {
1823 relayd = consumer_find_relayd(stream->net_seq_idx);
1824 if (relayd == NULL) {
1825 written = -ret;
1826 goto end;
1827 }
1828 }
1829 splice_pipe = stream->splice_pipe;
1830
1831 /* Write metadata stream id before payload */
1832 if (relayd) {
1833 unsigned long total_len = len;
1834
1835 if (stream->metadata_flag) {
1836 /*
1837 * Lock the control socket for the complete duration of the function
1838 * since from this point on we will use the socket.
1839 */
1840 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
1841
1842 if (stream->reset_metadata_flag) {
1843 ret = relayd_reset_metadata(&relayd->control_sock,
1844 stream->relayd_stream_id,
1845 stream->metadata_version);
1846 if (ret < 0) {
1847 relayd_hang_up = 1;
1848 goto write_error;
1849 }
1850 stream->reset_metadata_flag = 0;
1851 }
1852 ret = write_relayd_metadata_id(splice_pipe[1], stream,
1853 padding);
1854 if (ret < 0) {
1855 written = ret;
1856 relayd_hang_up = 1;
1857 goto write_error;
1858 }
1859
1860 total_len += sizeof(struct lttcomm_relayd_metadata_payload);
1861 }
1862
1863 ret = write_relayd_stream_header(stream, total_len, padding, relayd);
1864 if (ret < 0) {
1865 written = ret;
1866 relayd_hang_up = 1;
1867 goto write_error;
1868 }
1869 /* Use the returned socket. */
1870 outfd = ret;
1871 } else {
1872 /* No streaming, we have to set the len with the full padding */
1873 len += padding;
1874
1875 if (stream->metadata_flag && stream->reset_metadata_flag) {
1876 ret = utils_truncate_stream_file(stream->out_fd, 0);
1877 if (ret < 0) {
1878 ERR("Reset metadata file");
1879 goto end;
1880 }
1881 stream->reset_metadata_flag = 0;
1882 }
1883 /*
1884 * Check if we need to change the tracefile before writing the packet.
1885 */
1886 if (stream->chan->tracefile_size > 0 &&
1887 (stream->tracefile_size_current + len) >
1888 stream->chan->tracefile_size) {
1889 ret = consumer_stream_rotate_output_files(stream);
1890 if (ret < 0) {
1891 written = ret;
1892 goto end;
1893 }
1894 outfd = stream->out_fd;
1895 orig_offset = 0;
1896 }
1897 stream->tracefile_size_current += len;
1898 }
1899
1900 while (len > 0) {
1901 DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
1902 (unsigned long)offset, len, fd, splice_pipe[1]);
1903 ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
1904 SPLICE_F_MOVE | SPLICE_F_MORE);
1905 DBG("splice chan to pipe, ret %zd", ret_splice);
1906 if (ret_splice < 0) {
1907 ret = errno;
1908 written = -ret;
1909 PERROR("Error in relay splice");
1910 goto splice_error;
1911 }
1912
1913 /* Handle stream on the relayd if the output is on the network */
1914 if (relayd && stream->metadata_flag) {
1915 size_t metadata_payload_size =
1916 sizeof(struct lttcomm_relayd_metadata_payload);
1917
1918 /* Update counter to fit the spliced data */
1919 ret_splice += metadata_payload_size;
1920 len += metadata_payload_size;
1921 /*
1922 * We do this so the return value can match the len passed as
1923 * argument to this function.
1924 */
1925 written -= metadata_payload_size;
1926 }
1927
1928 /* Splice data out */
1929 ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
1930 ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
1931 DBG("Consumer splice pipe to file (out_fd: %d), ret %zd",
1932 outfd, ret_splice);
1933 if (ret_splice < 0) {
1934 ret = errno;
1935 written = -ret;
1936 relayd_hang_up = 1;
1937 goto write_error;
1938 } else if (ret_splice > len) {
1939 /*
1940 * We don't expect this code path to be executed but you never know
1941 * so this is an extra protection agains a buggy splice().
1942 */
1943 ret = errno;
1944 written += ret_splice;
1945 PERROR("Wrote more data than requested %zd (len: %lu)", ret_splice,
1946 len);
1947 goto splice_error;
1948 } else {
1949 /* All good, update current len and continue. */
1950 len -= ret_splice;
1951 }
1952
1953 /* This call is useless on a socket so better save a syscall. */
1954 if (!relayd) {
1955 /* This won't block, but will start writeout asynchronously */
1956 lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
1957 SYNC_FILE_RANGE_WRITE);
1958 stream->out_fd_offset += ret_splice;
1959 }
1960 stream->output_written += ret_splice;
1961 written += ret_splice;
1962 }
1963 if (!relayd) {
1964 lttng_consumer_sync_trace_file(stream, orig_offset);
1965 }
1966 goto end;
1967
1968 write_error:
1969 /*
1970 * This is a special case that the relayd has closed its socket. Let's
1971 * cleanup the relayd object and all associated streams.
1972 */
1973 if (relayd && relayd_hang_up) {
1974 ERR("Relayd hangup. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
1975 lttng_consumer_cleanup_relayd(relayd);
1976 /* Skip splice error so the consumer does not fail */
1977 goto end;
1978 }
1979
1980 splice_error:
1981 /* send the appropriate error description to sessiond */
1982 switch (ret) {
1983 case EINVAL:
1984 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
1985 break;
1986 case ENOMEM:
1987 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
1988 break;
1989 case ESPIPE:
1990 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
1991 break;
1992 }
1993
1994 end:
1995 if (relayd && stream->metadata_flag) {
1996 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
1997 }
1998
1999 rcu_read_unlock();
2000 return written;
2001 }
2002
2003 /*
2004 * Sample the snapshot positions for a specific fd
2005 *
2006 * Returns 0 on success, < 0 on error
2007 */
2008 int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
2009 {
2010 switch (the_consumer_data.type) {
2011 case LTTNG_CONSUMER_KERNEL:
2012 return lttng_kconsumer_sample_snapshot_positions(stream);
2013 case LTTNG_CONSUMER32_UST:
2014 case LTTNG_CONSUMER64_UST:
2015 return lttng_ustconsumer_sample_snapshot_positions(stream);
2016 default:
2017 ERR("Unknown consumer_data type");
2018 abort();
2019 return -ENOSYS;
2020 }
2021 }
2022 /*
2023 * Take a snapshot for a specific fd
2024 *
2025 * Returns 0 on success, < 0 on error
2026 */
2027 int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
2028 {
2029 switch (the_consumer_data.type) {
2030 case LTTNG_CONSUMER_KERNEL:
2031 return lttng_kconsumer_take_snapshot(stream);
2032 case LTTNG_CONSUMER32_UST:
2033 case LTTNG_CONSUMER64_UST:
2034 return lttng_ustconsumer_take_snapshot(stream);
2035 default:
2036 ERR("Unknown consumer_data type");
2037 abort();
2038 return -ENOSYS;
2039 }
2040 }
2041
2042 /*
2043 * Get the produced position
2044 *
2045 * Returns 0 on success, < 0 on error
2046 */
2047 int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
2048 unsigned long *pos)
2049 {
2050 switch (the_consumer_data.type) {
2051 case LTTNG_CONSUMER_KERNEL:
2052 return lttng_kconsumer_get_produced_snapshot(stream, pos);
2053 case LTTNG_CONSUMER32_UST:
2054 case LTTNG_CONSUMER64_UST:
2055 return lttng_ustconsumer_get_produced_snapshot(stream, pos);
2056 default:
2057 ERR("Unknown consumer_data type");
2058 abort();
2059 return -ENOSYS;
2060 }
2061 }
2062
2063 /*
2064 * Get the consumed position (free-running counter position in bytes).
2065 *
2066 * Returns 0 on success, < 0 on error
2067 */
2068 int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
2069 unsigned long *pos)
2070 {
2071 switch (the_consumer_data.type) {
2072 case LTTNG_CONSUMER_KERNEL:
2073 return lttng_kconsumer_get_consumed_snapshot(stream, pos);
2074 case LTTNG_CONSUMER32_UST:
2075 case LTTNG_CONSUMER64_UST:
2076 return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
2077 default:
2078 ERR("Unknown consumer_data type");
2079 abort();
2080 return -ENOSYS;
2081 }
2082 }
2083
2084 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
2085 int sock, struct pollfd *consumer_sockpoll)
2086 {
2087 switch (the_consumer_data.type) {
2088 case LTTNG_CONSUMER_KERNEL:
2089 return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
2090 case LTTNG_CONSUMER32_UST:
2091 case LTTNG_CONSUMER64_UST:
2092 return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
2093 default:
2094 ERR("Unknown consumer_data type");
2095 abort();
2096 return -ENOSYS;
2097 }
2098 }
2099
2100 static
2101 void lttng_consumer_close_all_metadata(void)
2102 {
2103 switch (the_consumer_data.type) {
2104 case LTTNG_CONSUMER_KERNEL:
2105 /*
2106 * The Kernel consumer has a different metadata scheme so we don't
2107 * close anything because the stream will be closed by the session
2108 * daemon.
2109 */
2110 break;
2111 case LTTNG_CONSUMER32_UST:
2112 case LTTNG_CONSUMER64_UST:
2113 /*
2114 * Close all metadata streams. The metadata hash table is passed and
2115 * this call iterates over it by closing all wakeup fd. This is safe
2116 * because at this point we are sure that the metadata producer is
2117 * either dead or blocked.
2118 */
2119 lttng_ustconsumer_close_all_metadata(metadata_ht);
2120 break;
2121 default:
2122 ERR("Unknown consumer_data type");
2123 abort();
2124 }
2125 }
2126
2127 /*
2128 * Clean up a metadata stream and free its memory.
2129 */
2130 void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
2131 struct lttng_ht *ht)
2132 {
2133 struct lttng_consumer_channel *channel = NULL;
2134 bool free_channel = false;
2135
2136 LTTNG_ASSERT(stream);
2137 /*
2138 * This call should NEVER receive regular stream. It must always be
2139 * metadata stream and this is crucial for data structure synchronization.
2140 */
2141 LTTNG_ASSERT(stream->metadata_flag);
2142
2143 DBG3("Consumer delete metadata stream %d", stream->wait_fd);
2144
2145 pthread_mutex_lock(&the_consumer_data.lock);
2146 /*
2147 * Note that this assumes that a stream's channel is never changed and
2148 * that the stream's lock doesn't need to be taken to sample its
2149 * channel.
2150 */
2151 channel = stream->chan;
2152 pthread_mutex_lock(&channel->lock);
2153 pthread_mutex_lock(&stream->lock);
2154 if (channel->metadata_cache) {
2155 /* Only applicable to userspace consumers. */
2156 pthread_mutex_lock(&channel->metadata_cache->lock);
2157 }
2158
2159 /* Remove any reference to that stream. */
2160 consumer_stream_delete(stream, ht);
2161
2162 /* Close down everything including the relayd if one. */
2163 consumer_stream_close(stream);
2164 /* Destroy tracer buffers of the stream. */
2165 consumer_stream_destroy_buffers(stream);
2166
2167 /* Atomically decrement channel refcount since other threads can use it. */
2168 if (!uatomic_sub_return(&channel->refcount, 1)
2169 && !uatomic_read(&channel->nb_init_stream_left)) {
2170 /* Go for channel deletion! */
2171 free_channel = true;
2172 }
2173 stream->chan = NULL;
2174
2175 /*
2176 * Nullify the stream reference so it is not used after deletion. The
2177 * channel lock MUST be acquired before being able to check for a NULL
2178 * pointer value.
2179 */
2180 channel->metadata_stream = NULL;
2181
2182 if (channel->metadata_cache) {
2183 pthread_mutex_unlock(&channel->metadata_cache->lock);
2184 }
2185 pthread_mutex_unlock(&stream->lock);
2186 pthread_mutex_unlock(&channel->lock);
2187 pthread_mutex_unlock(&the_consumer_data.lock);
2188
2189 if (free_channel) {
2190 consumer_del_channel(channel);
2191 }
2192
2193 lttng_trace_chunk_put(stream->trace_chunk);
2194 stream->trace_chunk = NULL;
2195 consumer_stream_free(stream);
2196 }
2197
2198 /*
2199 * Action done with the metadata stream when adding it to the consumer internal
2200 * data structures to handle it.
2201 */
2202 void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
2203 {
2204 struct lttng_ht *ht = metadata_ht;
2205 struct lttng_ht_iter iter;
2206 struct lttng_ht_node_u64 *node;
2207
2208 LTTNG_ASSERT(stream);
2209 LTTNG_ASSERT(ht);
2210
2211 DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
2212
2213 pthread_mutex_lock(&the_consumer_data.lock);
2214 pthread_mutex_lock(&stream->chan->lock);
2215 pthread_mutex_lock(&stream->chan->timer_lock);
2216 pthread_mutex_lock(&stream->lock);
2217
2218 /*
2219 * From here, refcounts are updated so be _careful_ when returning an error
2220 * after this point.
2221 */
2222
2223 rcu_read_lock();
2224
2225 /*
2226 * Lookup the stream just to make sure it does not exist in our internal
2227 * state. This should NEVER happen.
2228 */
2229 lttng_ht_lookup(ht, &stream->key, &iter);
2230 node = lttng_ht_iter_get_node_u64(&iter);
2231 LTTNG_ASSERT(!node);
2232
2233 /*
2234 * When nb_init_stream_left reaches 0, we don't need to trigger any action
2235 * in terms of destroying the associated channel, because the action that
2236 * causes the count to become 0 also causes a stream to be added. The
2237 * channel deletion will thus be triggered by the following removal of this
2238 * stream.
2239 */
2240 if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
2241 /* Increment refcount before decrementing nb_init_stream_left */
2242 cmm_smp_wmb();
2243 uatomic_dec(&stream->chan->nb_init_stream_left);
2244 }
2245
2246 lttng_ht_add_unique_u64(ht, &stream->node);
2247
2248 lttng_ht_add_u64(the_consumer_data.stream_per_chan_id_ht,
2249 &stream->node_channel_id);
2250
2251 /*
2252 * Add stream to the stream_list_ht of the consumer data. No need to steal
2253 * the key since the HT does not use it and we allow to add redundant keys
2254 * into this table.
2255 */
2256 lttng_ht_add_u64(the_consumer_data.stream_list_ht,
2257 &stream->node_session_id);
2258
2259 rcu_read_unlock();
2260
2261 pthread_mutex_unlock(&stream->lock);
2262 pthread_mutex_unlock(&stream->chan->lock);
2263 pthread_mutex_unlock(&stream->chan->timer_lock);
2264 pthread_mutex_unlock(&the_consumer_data.lock);
2265 }
2266
2267 /*
2268 * Delete data stream that are flagged for deletion (endpoint_status).
2269 */
2270 static void validate_endpoint_status_data_stream(void)
2271 {
2272 struct lttng_ht_iter iter;
2273 struct lttng_consumer_stream *stream;
2274
2275 DBG("Consumer delete flagged data stream");
2276
2277 rcu_read_lock();
2278 cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
2279 /* Validate delete flag of the stream */
2280 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
2281 continue;
2282 }
2283 /* Delete it right now */
2284 consumer_del_stream(stream, data_ht);
2285 }
2286 rcu_read_unlock();
2287 }
2288
2289 /*
2290 * Delete metadata stream that are flagged for deletion (endpoint_status).
2291 */
2292 static void validate_endpoint_status_metadata_stream(
2293 struct lttng_poll_event *pollset)
2294 {
2295 struct lttng_ht_iter iter;
2296 struct lttng_consumer_stream *stream;
2297
2298 DBG("Consumer delete flagged metadata stream");
2299
2300 LTTNG_ASSERT(pollset);
2301
2302 rcu_read_lock();
2303 cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
2304 /* Validate delete flag of the stream */
2305 if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) {
2306 continue;
2307 }
2308 /*
2309 * Remove from pollset so the metadata thread can continue without
2310 * blocking on a deleted stream.
2311 */
2312 lttng_poll_del(pollset, stream->wait_fd);
2313
2314 /* Delete it right now */
2315 consumer_del_metadata_stream(stream, metadata_ht);
2316 }
2317 rcu_read_unlock();
2318 }
2319
2320 /*
2321 * Thread polls on metadata file descriptor and write them on disk or on the
2322 * network.
2323 */
2324 void *consumer_thread_metadata_poll(void *data)
2325 {
2326 int ret, i, pollfd, err = -1;
2327 uint32_t revents, nb_fd;
2328 struct lttng_consumer_stream *stream = NULL;
2329 struct lttng_ht_iter iter;
2330 struct lttng_ht_node_u64 *node;
2331 struct lttng_poll_event events;
2332 struct lttng_consumer_local_data *ctx = (lttng_consumer_local_data *) data;
2333 ssize_t len;
2334
2335 rcu_register_thread();
2336
2337 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA);
2338
2339 if (testpoint(consumerd_thread_metadata)) {
2340 goto error_testpoint;
2341 }
2342
2343 health_code_update();
2344
2345 DBG("Thread metadata poll started");
2346
2347 /* Size is set to 1 for the consumer_metadata pipe */
2348 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2349 if (ret < 0) {
2350 ERR("Poll set creation failed");
2351 goto end_poll;
2352 }
2353
2354 ret = lttng_poll_add(&events,
2355 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe), LPOLLIN);
2356 if (ret < 0) {
2357 goto end;
2358 }
2359
2360 /* Main loop */
2361 DBG("Metadata main loop started");
2362
2363 while (1) {
2364 restart:
2365 health_code_update();
2366 health_poll_entry();
2367 DBG("Metadata poll wait");
2368 ret = lttng_poll_wait(&events, -1);
2369 DBG("Metadata poll return from wait with %d fd(s)",
2370 LTTNG_POLL_GETNB(&events));
2371 health_poll_exit();
2372 DBG("Metadata event caught in thread");
2373 if (ret < 0) {
2374 if (errno == EINTR) {
2375 ERR("Poll EINTR caught");
2376 goto restart;
2377 }
2378 if (LTTNG_POLL_GETNB(&events) == 0) {
2379 err = 0; /* All is OK */
2380 }
2381 goto end;
2382 }
2383
2384 nb_fd = ret;
2385
2386 /* From here, the event is a metadata wait fd */
2387 for (i = 0; i < nb_fd; i++) {
2388 health_code_update();
2389
2390 revents = LTTNG_POLL_GETEV(&events, i);
2391 pollfd = LTTNG_POLL_GETFD(&events, i);
2392
2393 if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
2394 if (revents & LPOLLIN) {
2395 ssize_t pipe_len;
2396
2397 pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
2398 &stream, sizeof(stream));
2399 if (pipe_len < sizeof(stream)) {
2400 if (pipe_len < 0) {
2401 PERROR("read metadata stream");
2402 }
2403 /*
2404 * Remove the pipe from the poll set and continue the loop
2405 * since their might be data to consume.
2406 */
2407 lttng_poll_del(&events,
2408 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2409 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2410 continue;
2411 }
2412
2413 /* A NULL stream means that the state has changed. */
2414 if (stream == NULL) {
2415 /* Check for deleted streams. */
2416 validate_endpoint_status_metadata_stream(&events);
2417 goto restart;
2418 }
2419
2420 DBG("Adding metadata stream %d to poll set",
2421 stream->wait_fd);
2422
2423 /* Add metadata stream to the global poll events list */
2424 lttng_poll_add(&events, stream->wait_fd,
2425 LPOLLIN | LPOLLPRI | LPOLLHUP);
2426 } else if (revents & (LPOLLERR | LPOLLHUP)) {
2427 DBG("Metadata thread pipe hung up");
2428 /*
2429 * Remove the pipe from the poll set and continue the loop
2430 * since their might be data to consume.
2431 */
2432 lttng_poll_del(&events,
2433 lttng_pipe_get_readfd(ctx->consumer_metadata_pipe));
2434 lttng_pipe_read_close(ctx->consumer_metadata_pipe);
2435 continue;
2436 } else {
2437 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2438 goto end;
2439 }
2440
2441 /* Handle other stream */
2442 continue;
2443 }
2444
2445 rcu_read_lock();
2446 {
2447 uint64_t tmp_id = (uint64_t) pollfd;
2448
2449 lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
2450 }
2451 node = lttng_ht_iter_get_node_u64(&iter);
2452 LTTNG_ASSERT(node);
2453
2454 stream = caa_container_of(node, struct lttng_consumer_stream,
2455 node);
2456
2457 if (revents & (LPOLLIN | LPOLLPRI)) {
2458 /* Get the data out of the metadata file descriptor */
2459 DBG("Metadata available on fd %d", pollfd);
2460 LTTNG_ASSERT(stream->wait_fd == pollfd);
2461
2462 do {
2463 health_code_update();
2464
2465 len = ctx->on_buffer_ready(stream, ctx, false);
2466 /*
2467 * We don't check the return value here since if we get
2468 * a negative len, it means an error occurred thus we
2469 * simply remove it from the poll set and free the
2470 * stream.
2471 */
2472 } while (len > 0);
2473
2474 /* It's ok to have an unavailable sub-buffer */
2475 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2476 /* Clean up stream from consumer and free it. */
2477 lttng_poll_del(&events, stream->wait_fd);
2478 consumer_del_metadata_stream(stream, metadata_ht);
2479 }
2480 } else if (revents & (LPOLLERR | LPOLLHUP)) {
2481 DBG("Metadata fd %d is hup|err.", pollfd);
2482 if (!stream->hangup_flush_done &&
2483 (the_consumer_data.type == LTTNG_CONSUMER32_UST ||
2484 the_consumer_data.type ==
2485 LTTNG_CONSUMER64_UST)) {
2486 DBG("Attempting to flush and consume the UST buffers");
2487 lttng_ustconsumer_on_stream_hangup(stream);
2488
2489 /* We just flushed the stream now read it. */
2490 do {
2491 health_code_update();
2492
2493 len = ctx->on_buffer_ready(stream, ctx, false);
2494 /*
2495 * We don't check the return value here since if we get
2496 * a negative len, it means an error occurred thus we
2497 * simply remove it from the poll set and free the
2498 * stream.
2499 */
2500 } while (len > 0);
2501 }
2502
2503 lttng_poll_del(&events, stream->wait_fd);
2504 /*
2505 * This call update the channel states, closes file descriptors
2506 * and securely free the stream.
2507 */
2508 consumer_del_metadata_stream(stream, metadata_ht);
2509 } else {
2510 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2511 rcu_read_unlock();
2512 goto end;
2513 }
2514 /* Release RCU lock for the stream looked up */
2515 rcu_read_unlock();
2516 }
2517 }
2518
2519 /* All is OK */
2520 err = 0;
2521 end:
2522 DBG("Metadata poll thread exiting");
2523
2524 lttng_poll_clean(&events);
2525 end_poll:
2526 error_testpoint:
2527 if (err) {
2528 health_error();
2529 ERR("Health error occurred in %s", __func__);
2530 }
2531 health_unregister(health_consumerd);
2532 rcu_unregister_thread();
2533 return NULL;
2534 }
2535
2536 /*
2537 * This thread polls the fds in the set to consume the data and write
2538 * it to tracefile if necessary.
2539 */
2540 void *consumer_thread_data_poll(void *data)
2541 {
2542 int num_rdy, num_hup, high_prio, ret, i, err = -1;
2543 struct pollfd *pollfd = NULL;
2544 /* local view of the streams */
2545 struct lttng_consumer_stream **local_stream = NULL, *new_stream = NULL;
2546 /* local view of consumer_data.fds_count */
2547 int nb_fd = 0;
2548 /* 2 for the consumer_data_pipe and wake up pipe */
2549 const int nb_pipes_fd = 2;
2550 /* Number of FDs with CONSUMER_ENDPOINT_INACTIVE but still open. */
2551 int nb_inactive_fd = 0;
2552 struct lttng_consumer_local_data *ctx = (lttng_consumer_local_data *) data;
2553 ssize_t len;
2554
2555 rcu_register_thread();
2556
2557 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA);
2558
2559 if (testpoint(consumerd_thread_data)) {
2560 goto error_testpoint;
2561 }
2562
2563 health_code_update();
2564
2565 local_stream = (lttng_consumer_stream **) zmalloc(sizeof(struct lttng_consumer_stream *));
2566 if (local_stream == NULL) {
2567 PERROR("local_stream malloc");
2568 goto end;
2569 }
2570
2571 while (1) {
2572 health_code_update();
2573
2574 high_prio = 0;
2575 num_hup = 0;
2576
2577 /*
2578 * the fds set has been updated, we need to update our
2579 * local array as well
2580 */
2581 pthread_mutex_lock(&the_consumer_data.lock);
2582 if (the_consumer_data.need_update) {
2583 free(pollfd);
2584 pollfd = NULL;
2585
2586 free(local_stream);
2587 local_stream = NULL;
2588
2589 /* Allocate for all fds */
2590 pollfd = (struct pollfd *) zmalloc((the_consumer_data.stream_count +
2591 nb_pipes_fd) *
2592 sizeof(struct pollfd));
2593 if (pollfd == NULL) {
2594 PERROR("pollfd malloc");
2595 pthread_mutex_unlock(&the_consumer_data.lock);
2596 goto end;
2597 }
2598
2599 local_stream = (lttng_consumer_stream **) zmalloc((the_consumer_data.stream_count +
2600 nb_pipes_fd) *
2601 sizeof(struct lttng_consumer_stream *));
2602 if (local_stream == NULL) {
2603 PERROR("local_stream malloc");
2604 pthread_mutex_unlock(&the_consumer_data.lock);
2605 goto end;
2606 }
2607 ret = update_poll_array(ctx, &pollfd, local_stream,
2608 data_ht, &nb_inactive_fd);
2609 if (ret < 0) {
2610 ERR("Error in allocating pollfd or local_outfds");
2611 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
2612 pthread_mutex_unlock(&the_consumer_data.lock);
2613 goto end;
2614 }
2615 nb_fd = ret;
2616 the_consumer_data.need_update = 0;
2617 }
2618 pthread_mutex_unlock(&the_consumer_data.lock);
2619
2620 /* No FDs and consumer_quit, consumer_cleanup the thread */
2621 if (nb_fd == 0 && nb_inactive_fd == 0 &&
2622 CMM_LOAD_SHARED(consumer_quit) == 1) {
2623 err = 0; /* All is OK */
2624 goto end;
2625 }
2626 /* poll on the array of fds */
2627 restart:
2628 DBG("polling on %d fd", nb_fd + nb_pipes_fd);
2629 if (testpoint(consumerd_thread_data_poll)) {
2630 goto end;
2631 }
2632 health_poll_entry();
2633 num_rdy = poll(pollfd, nb_fd + nb_pipes_fd, -1);
2634 health_poll_exit();
2635 DBG("poll num_rdy : %d", num_rdy);
2636 if (num_rdy == -1) {
2637 /*
2638 * Restart interrupted system call.
2639 */
2640 if (errno == EINTR) {
2641 goto restart;
2642 }
2643 PERROR("Poll error");
2644 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
2645 goto end;
2646 } else if (num_rdy == 0) {
2647 DBG("Polling thread timed out");
2648 goto end;
2649 }
2650
2651 if (caa_unlikely(data_consumption_paused)) {
2652 DBG("Data consumption paused, sleeping...");
2653 sleep(1);
2654 goto restart;
2655 }
2656
2657 /*
2658 * If the consumer_data_pipe triggered poll go directly to the
2659 * beginning of the loop to update the array. We want to prioritize
2660 * array update over low-priority reads.
2661 */
2662 if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) {
2663 ssize_t pipe_readlen;
2664
2665 DBG("consumer_data_pipe wake up");
2666 pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
2667 &new_stream, sizeof(new_stream));
2668 if (pipe_readlen < sizeof(new_stream)) {
2669 PERROR("Consumer data pipe");
2670 /* Continue so we can at least handle the current stream(s). */
2671 continue;
2672 }
2673
2674 /*
2675 * If the stream is NULL, just ignore it. It's also possible that
2676 * the sessiond poll thread changed the consumer_quit state and is
2677 * waking us up to test it.
2678 */
2679 if (new_stream == NULL) {
2680 validate_endpoint_status_data_stream();
2681 continue;
2682 }
2683
2684 /* Continue to update the local streams and handle prio ones */
2685 continue;
2686 }
2687
2688 /* Handle wakeup pipe. */
2689 if (pollfd[nb_fd + 1].revents & (POLLIN | POLLPRI)) {
2690 char dummy;
2691 ssize_t pipe_readlen;
2692
2693 pipe_readlen = lttng_pipe_read(ctx->consumer_wakeup_pipe, &dummy,
2694 sizeof(dummy));
2695 if (pipe_readlen < 0) {
2696 PERROR("Consumer data wakeup pipe");
2697 }
2698 /* We've been awakened to handle stream(s). */
2699 ctx->has_wakeup = 0;
2700 }
2701
2702 /* Take care of high priority channels first. */
2703 for (i = 0; i < nb_fd; i++) {
2704 health_code_update();
2705
2706 if (local_stream[i] == NULL) {
2707 continue;
2708 }
2709 if (pollfd[i].revents & POLLPRI) {
2710 DBG("Urgent read on fd %d", pollfd[i].fd);
2711 high_prio = 1;
2712 len = ctx->on_buffer_ready(local_stream[i], ctx, false);
2713 /* it's ok to have an unavailable sub-buffer */
2714 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2715 /* Clean the stream and free it. */
2716 consumer_del_stream(local_stream[i], data_ht);
2717 local_stream[i] = NULL;
2718 } else if (len > 0) {
2719 local_stream[i]->data_read = 1;
2720 }
2721 }
2722 }
2723
2724 /*
2725 * If we read high prio channel in this loop, try again
2726 * for more high prio data.
2727 */
2728 if (high_prio) {
2729 continue;
2730 }
2731
2732 /* Take care of low priority channels. */
2733 for (i = 0; i < nb_fd; i++) {
2734 health_code_update();
2735
2736 if (local_stream[i] == NULL) {
2737 continue;
2738 }
2739 if ((pollfd[i].revents & POLLIN) ||
2740 local_stream[i]->hangup_flush_done ||
2741 local_stream[i]->has_data) {
2742 DBG("Normal read on fd %d", pollfd[i].fd);
2743 len = ctx->on_buffer_ready(local_stream[i], ctx, false);
2744 /* it's ok to have an unavailable sub-buffer */
2745 if (len < 0 && len != -EAGAIN && len != -ENODATA) {
2746 /* Clean the stream and free it. */
2747 consumer_del_stream(local_stream[i], data_ht);
2748 local_stream[i] = NULL;
2749 } else if (len > 0) {
2750 local_stream[i]->data_read = 1;
2751 }
2752 }
2753 }
2754
2755 /* Handle hangup and errors */
2756 for (i = 0; i < nb_fd; i++) {
2757 health_code_update();
2758
2759 if (local_stream[i] == NULL) {
2760 continue;
2761 }
2762 if (!local_stream[i]->hangup_flush_done
2763 && (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
2764 && (the_consumer_data.type == LTTNG_CONSUMER32_UST
2765 || the_consumer_data.type == LTTNG_CONSUMER64_UST)) {
2766 DBG("fd %d is hup|err|nval. Attempting flush and read.",
2767 pollfd[i].fd);
2768 lttng_ustconsumer_on_stream_hangup(local_stream[i]);
2769 /* Attempt read again, for the data we just flushed. */
2770 local_stream[i]->data_read = 1;
2771 }
2772 /*
2773 * If the poll flag is HUP/ERR/NVAL and we have
2774 * read no data in this pass, we can remove the
2775 * stream from its hash table.
2776 */
2777 if ((pollfd[i].revents & POLLHUP)) {
2778 DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
2779 if (!local_stream[i]->data_read) {
2780 consumer_del_stream(local_stream[i], data_ht);
2781 local_stream[i] = NULL;
2782 num_hup++;
2783 }
2784 } else if (pollfd[i].revents & POLLERR) {
2785 ERR("Error returned in polling fd %d.", pollfd[i].fd);
2786 if (!local_stream[i]->data_read) {
2787 consumer_del_stream(local_stream[i], data_ht);
2788 local_stream[i] = NULL;
2789 num_hup++;
2790 }
2791 } else if (pollfd[i].revents & POLLNVAL) {
2792 ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
2793 if (!local_stream[i]->data_read) {
2794 consumer_del_stream(local_stream[i], data_ht);
2795 local_stream[i] = NULL;
2796 num_hup++;
2797 }
2798 }
2799 if (local_stream[i] != NULL) {
2800 local_stream[i]->data_read = 0;
2801 }
2802 }
2803 }
2804 /* All is OK */
2805 err = 0;
2806 end:
2807 DBG("polling thread exiting");
2808 free(pollfd);
2809 free(local_stream);
2810
2811 /*
2812 * Close the write side of the pipe so epoll_wait() in
2813 * consumer_thread_metadata_poll can catch it. The thread is monitoring the
2814 * read side of the pipe. If we close them both, epoll_wait strangely does
2815 * not return and could create a endless wait period if the pipe is the
2816 * only tracked fd in the poll set. The thread will take care of closing
2817 * the read side.
2818 */
2819 (void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
2820
2821 error_testpoint:
2822 if (err) {
2823 health_error();
2824 ERR("Health error occurred in %s", __func__);
2825 }
2826 health_unregister(health_consumerd);
2827
2828 rcu_unregister_thread();
2829 return NULL;
2830 }
2831
2832 /*
2833 * Close wake-up end of each stream belonging to the channel. This will
2834 * allow the poll() on the stream read-side to detect when the
2835 * write-side (application) finally closes them.
2836 */
2837 static
2838 void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
2839 {
2840 struct lttng_ht *ht;
2841 struct lttng_consumer_stream *stream;
2842 struct lttng_ht_iter iter;
2843
2844 ht = the_consumer_data.stream_per_chan_id_ht;
2845
2846 rcu_read_lock();
2847 cds_lfht_for_each_entry_duplicate(ht->ht,
2848 ht->hash_fct(&channel->key, lttng_ht_seed),
2849 ht->match_fct, &channel->key,
2850 &iter.iter, stream, node_channel_id.node) {
2851 /*
2852 * Protect against teardown with mutex.
2853 */
2854 pthread_mutex_lock(&stream->lock);
2855 if (cds_lfht_is_node_deleted(&stream->node.node)) {
2856 goto next;
2857 }
2858 switch (the_consumer_data.type) {
2859 case LTTNG_CONSUMER_KERNEL:
2860 break;
2861 case LTTNG_CONSUMER32_UST:
2862 case LTTNG_CONSUMER64_UST:
2863 if (stream->metadata_flag) {
2864 /* Safe and protected by the stream lock. */
2865 lttng_ustconsumer_close_metadata(stream->chan);
2866 } else {
2867 /*
2868 * Note: a mutex is taken internally within
2869 * liblttng-ust-ctl to protect timer wakeup_fd
2870 * use from concurrent close.
2871 */
2872 lttng_ustconsumer_close_stream_wakeup(stream);
2873 }
2874 break;
2875 default:
2876 ERR("Unknown consumer_data type");
2877 abort();
2878 }
2879 next:
2880 pthread_mutex_unlock(&stream->lock);
2881 }
2882 rcu_read_unlock();
2883 }
2884
2885 static void destroy_channel_ht(struct lttng_ht *ht)
2886 {
2887 struct lttng_ht_iter iter;
2888 struct lttng_consumer_channel *channel;
2889 int ret;
2890
2891 if (ht == NULL) {
2892 return;
2893 }
2894
2895 rcu_read_lock();
2896 cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
2897 ret = lttng_ht_del(ht, &iter);
2898 LTTNG_ASSERT(ret != 0);
2899 }
2900 rcu_read_unlock();
2901
2902 lttng_ht_destroy(ht);
2903 }
2904
2905 /*
2906 * This thread polls the channel fds to detect when they are being
2907 * closed. It closes all related streams if the channel is detected as
2908 * closed. It is currently only used as a shim layer for UST because the
2909 * consumerd needs to keep the per-stream wakeup end of pipes open for
2910 * periodical flush.
2911 */
2912 void *consumer_thread_channel_poll(void *data)
2913 {
2914 int ret, i, pollfd, err = -1;
2915 uint32_t revents, nb_fd;
2916 struct lttng_consumer_channel *chan = NULL;
2917 struct lttng_ht_iter iter;
2918 struct lttng_ht_node_u64 *node;
2919 struct lttng_poll_event events;
2920 struct lttng_consumer_local_data *ctx = (lttng_consumer_local_data *) data;
2921 struct lttng_ht *channel_ht;
2922
2923 rcu_register_thread();
2924
2925 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL);
2926
2927 if (testpoint(consumerd_thread_channel)) {
2928 goto error_testpoint;
2929 }
2930
2931 health_code_update();
2932
2933 channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
2934 if (!channel_ht) {
2935 /* ENOMEM at this point. Better to bail out. */
2936 goto end_ht;
2937 }
2938
2939 DBG("Thread channel poll started");
2940
2941 /* Size is set to 1 for the consumer_channel pipe */
2942 ret = lttng_poll_create(&events, 2, LTTNG_CLOEXEC);
2943 if (ret < 0) {
2944 ERR("Poll set creation failed");
2945 goto end_poll;
2946 }
2947
2948 ret = lttng_poll_add(&events, ctx->consumer_channel_pipe[0], LPOLLIN);
2949 if (ret < 0) {
2950 goto end;
2951 }
2952
2953 /* Main loop */
2954 DBG("Channel main loop started");
2955
2956 while (1) {
2957 restart:
2958 health_code_update();
2959 DBG("Channel poll wait");
2960 health_poll_entry();
2961 ret = lttng_poll_wait(&events, -1);
2962 DBG("Channel poll return from wait with %d fd(s)",
2963 LTTNG_POLL_GETNB(&events));
2964 health_poll_exit();
2965 DBG("Channel event caught in thread");
2966 if (ret < 0) {
2967 if (errno == EINTR) {
2968 ERR("Poll EINTR caught");
2969 goto restart;
2970 }
2971 if (LTTNG_POLL_GETNB(&events) == 0) {
2972 err = 0; /* All is OK */
2973 }
2974 goto end;
2975 }
2976
2977 nb_fd = ret;
2978
2979 /* From here, the event is a channel wait fd */
2980 for (i = 0; i < nb_fd; i++) {
2981 health_code_update();
2982
2983 revents = LTTNG_POLL_GETEV(&events, i);
2984 pollfd = LTTNG_POLL_GETFD(&events, i);
2985
2986 if (pollfd == ctx->consumer_channel_pipe[0]) {
2987 if (revents & LPOLLIN) {
2988 enum consumer_channel_action action;
2989 uint64_t key;
2990
2991 ret = read_channel_pipe(ctx, &chan, &key, &action);
2992 if (ret <= 0) {
2993 if (ret < 0) {
2994 ERR("Error reading channel pipe");
2995 }
2996 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
2997 continue;
2998 }
2999
3000 switch (action) {
3001 case CONSUMER_CHANNEL_ADD:
3002 DBG("Adding channel %d to poll set",
3003 chan->wait_fd);
3004
3005 lttng_ht_node_init_u64(&chan->wait_fd_node,
3006 chan->wait_fd);
3007 rcu_read_lock();
3008 lttng_ht_add_unique_u64(channel_ht,
3009 &chan->wait_fd_node);
3010 rcu_read_unlock();
3011 /* Add channel to the global poll events list */
3012 lttng_poll_add(&events, chan->wait_fd,
3013 LPOLLERR | LPOLLHUP);
3014 break;
3015 case CONSUMER_CHANNEL_DEL:
3016 {
3017 /*
3018 * This command should never be called if the channel
3019 * has streams monitored by either the data or metadata
3020 * thread. The consumer only notify this thread with a
3021 * channel del. command if it receives a destroy
3022 * channel command from the session daemon that send it
3023 * if a command prior to the GET_CHANNEL failed.
3024 */
3025
3026 rcu_read_lock();
3027 chan = consumer_find_channel(key);
3028 if (!chan) {
3029 rcu_read_unlock();
3030 ERR("UST consumer get channel key %" PRIu64 " not found for del channel", key);
3031 break;
3032 }
3033 lttng_poll_del(&events, chan->wait_fd);
3034 iter.iter.node = &chan->wait_fd_node.node;
3035 ret = lttng_ht_del(channel_ht, &iter);
3036 LTTNG_ASSERT(ret == 0);
3037
3038 switch (the_consumer_data.type) {
3039 case LTTNG_CONSUMER_KERNEL:
3040 break;
3041 case LTTNG_CONSUMER32_UST:
3042 case LTTNG_CONSUMER64_UST:
3043 health_code_update();
3044 /* Destroy streams that might have been left in the stream list. */
3045 clean_channel_stream_list(chan);
3046 break;
3047 default:
3048 ERR("Unknown consumer_data type");
3049 abort();
3050 }
3051
3052 /*
3053 * Release our own refcount. Force channel deletion even if
3054 * streams were not initialized.
3055 */
3056 if (!uatomic_sub_return(&chan->refcount, 1)) {
3057 consumer_del_channel(chan);
3058 }
3059 rcu_read_unlock();
3060 goto restart;
3061 }
3062 case CONSUMER_CHANNEL_QUIT:
3063 /*
3064 * Remove the pipe from the poll set and continue the loop
3065 * since their might be data to consume.
3066 */
3067 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3068 continue;
3069 default:
3070 ERR("Unknown action");
3071 break;
3072 }
3073 } else if (revents & (LPOLLERR | LPOLLHUP)) {
3074 DBG("Channel thread pipe hung up");
3075 /*
3076 * Remove the pipe from the poll set and continue the loop
3077 * since their might be data to consume.
3078 */
3079 lttng_poll_del(&events, ctx->consumer_channel_pipe[0]);
3080 continue;
3081 } else {
3082 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3083 goto end;
3084 }
3085
3086 /* Handle other stream */
3087 continue;
3088 }
3089
3090 rcu_read_lock();
3091 {
3092 uint64_t tmp_id = (uint64_t) pollfd;
3093
3094 lttng_ht_lookup(channel_ht, &tmp_id, &iter);
3095 }
3096 node = lttng_ht_iter_get_node_u64(&iter);
3097 LTTNG_ASSERT(node);
3098
3099 chan = caa_container_of(node, struct lttng_consumer_channel,
3100 wait_fd_node);
3101
3102 /* Check for error event */
3103 if (revents & (LPOLLERR | LPOLLHUP)) {
3104 DBG("Channel fd %d is hup|err.", pollfd);
3105
3106 lttng_poll_del(&events, chan->wait_fd);
3107 ret = lttng_ht_del(channel_ht, &iter);
3108 LTTNG_ASSERT(ret == 0);
3109
3110 /*
3111 * This will close the wait fd for each stream associated to
3112 * this channel AND monitored by the data/metadata thread thus
3113 * will be clean by the right thread.
3114 */
3115 consumer_close_channel_streams(chan);
3116
3117 /* Release our own refcount */
3118 if (!uatomic_sub_return(&chan->refcount, 1)
3119 && !uatomic_read(&chan->nb_init_stream_left)) {
3120 consumer_del_channel(chan);
3121 }
3122 } else {
3123 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
3124 rcu_read_unlock();
3125 goto end;
3126 }
3127
3128 /* Release RCU lock for the channel looked up */
3129 rcu_read_unlock();
3130 }
3131 }
3132
3133 /* All is OK */
3134 err = 0;
3135 end:
3136 lttng_poll_clean(&events);
3137 end_poll:
3138 destroy_channel_ht(channel_ht);
3139 end_ht:
3140 error_testpoint:
3141 DBG("Channel poll thread exiting");
3142 if (err) {
3143 health_error();
3144 ERR("Health error occurred in %s", __func__);
3145 }
3146 health_unregister(health_consumerd);
3147 rcu_unregister_thread();
3148 return NULL;
3149 }
3150
3151 static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
3152 struct pollfd *sockpoll, int client_socket)
3153 {
3154 int ret;
3155
3156 LTTNG_ASSERT(ctx);
3157 LTTNG_ASSERT(sockpoll);
3158
3159 ret = lttng_consumer_poll_socket(sockpoll);
3160 if (ret) {
3161 goto error;
3162 }
3163 DBG("Metadata connection on client_socket");
3164
3165 /* Blocking call, waiting for transmission */
3166 ctx->consumer_metadata_socket = lttcomm_accept_unix_sock(client_socket);
3167 if (ctx->consumer_metadata_socket < 0) {
3168 WARN("On accept metadata");
3169 ret = -1;
3170 goto error;
3171 }
3172 ret = 0;
3173
3174 error:
3175 return ret;
3176 }
3177
3178 /*
3179 * This thread listens on the consumerd socket and receives the file
3180 * descriptors from the session daemon.
3181 */
3182 void *consumer_thread_sessiond_poll(void *data)
3183 {
3184 int sock = -1, client_socket, ret, err = -1;
3185 /*
3186 * structure to poll for incoming data on communication socket avoids
3187 * making blocking sockets.
3188 */
3189 struct pollfd consumer_sockpoll[2];
3190 struct lttng_consumer_local_data *ctx = (lttng_consumer_local_data *) data;
3191
3192 rcu_register_thread();
3193
3194 health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_SESSIOND);
3195
3196 if (testpoint(consumerd_thread_sessiond)) {
3197 goto error_testpoint;
3198 }
3199
3200 health_code_update();
3201
3202 DBG("Creating command socket %s", ctx->consumer_command_sock_path);
3203 unlink(ctx->consumer_command_sock_path);
3204 client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path);
3205 if (client_socket < 0) {
3206 ERR("Cannot create command socket");
3207 goto end;
3208 }
3209
3210 ret = lttcomm_listen_unix_sock(client_socket);
3211 if (ret < 0) {
3212 goto end;
3213 }
3214
3215 DBG("Sending ready command to lttng-sessiond");
3216 ret = lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_COMMAND_SOCK_READY);
3217 /* return < 0 on error, but == 0 is not fatal */
3218 if (ret < 0) {
3219 ERR("Error sending ready command to lttng-sessiond");
3220 goto end;
3221 }
3222
3223 /* prepare the FDs to poll : to client socket and the should_quit pipe */
3224 consumer_sockpoll[0].fd = ctx->consumer_should_quit[0];
3225 consumer_sockpoll[0].events = POLLIN | POLLPRI;
3226 consumer_sockpoll[1].fd = client_socket;
3227 consumer_sockpoll[1].events = POLLIN | POLLPRI;
3228
3229 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3230 if (ret) {
3231 if (ret > 0) {
3232 /* should exit */
3233 err = 0;
3234 }
3235 goto end;
3236 }
3237 DBG("Connection on client_socket");
3238
3239 /* Blocking call, waiting for transmission */
3240 sock = lttcomm_accept_unix_sock(client_socket);
3241 if (sock < 0) {
3242 WARN("On accept");
3243 goto end;
3244 }
3245
3246 /*
3247 * Setup metadata socket which is the second socket connection on the
3248 * command unix socket.
3249 */
3250 ret = set_metadata_socket(ctx, consumer_sockpoll, client_socket);
3251 if (ret) {
3252 if (ret > 0) {
3253 /* should exit */
3254 err = 0;
3255 }
3256 goto end;
3257 }
3258
3259 /* This socket is not useful anymore. */
3260 ret = close(client_socket);
3261 if (ret < 0) {
3262 PERROR("close client_socket");
3263 }
3264 client_socket = -1;
3265
3266 /* update the polling structure to poll on the established socket */
3267 consumer_sockpoll[1].fd = sock;
3268 consumer_sockpoll[1].events = POLLIN | POLLPRI;
3269
3270 while (1) {
3271 health_code_update();
3272
3273 health_poll_entry();
3274 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3275 health_poll_exit();
3276 if (ret) {
3277 if (ret > 0) {
3278 /* should exit */
3279 err = 0;
3280 }
3281 goto end;
3282 }
3283 DBG("Incoming command on sock");
3284 ret = lttng_consumer_recv_cmd(ctx, sock, consumer_sockpoll);
3285 if (ret <= 0) {
3286 /*
3287 * This could simply be a session daemon quitting. Don't output
3288 * ERR() here.
3289 */
3290 DBG("Communication interrupted on command socket");
3291 err = 0;
3292 goto end;
3293 }
3294 if (CMM_LOAD_SHARED(consumer_quit)) {
3295 DBG("consumer_thread_receive_fds received quit from signal");
3296 err = 0; /* All is OK */
3297 goto end;
3298 }
3299 DBG("Received command on sock");
3300 }
3301 /* All is OK */
3302 err = 0;
3303
3304 end:
3305 DBG("Consumer thread sessiond poll exiting");
3306
3307 /*
3308 * Close metadata streams since the producer is the session daemon which
3309 * just died.
3310 *
3311 * NOTE: for now, this only applies to the UST tracer.
3312 */
3313 lttng_consumer_close_all_metadata();
3314
3315 /*
3316 * when all fds have hung up, the polling thread
3317 * can exit cleanly
3318 */
3319 CMM_STORE_SHARED(consumer_quit, 1);
3320
3321 /*
3322 * Notify the data poll thread to poll back again and test the
3323 * consumer_quit state that we just set so to quit gracefully.
3324 */
3325 notify_thread_lttng_pipe(ctx->consumer_data_pipe);
3326
3327 notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
3328
3329 notify_health_quit_pipe(health_quit_pipe);
3330
3331 /* Cleaning up possibly open sockets. */
3332 if (sock >= 0) {
3333 ret = close(sock);
3334 if (ret < 0) {
3335 PERROR("close sock sessiond poll");
3336 }
3337 }
3338 if (client_socket >= 0) {
3339 ret = close(client_socket);
3340 if (ret < 0) {
3341 PERROR("close client_socket sessiond poll");
3342 }
3343 }
3344
3345 error_testpoint:
3346 if (err) {
3347 health_error();
3348 ERR("Health error occurred in %s", __func__);
3349 }
3350 health_unregister(health_consumerd);
3351
3352 rcu_unregister_thread();
3353 return NULL;
3354 }
3355
3356 static int post_consume(struct lttng_consumer_stream *stream,
3357 const struct stream_subbuffer *subbuffer,
3358 struct lttng_consumer_local_data *ctx)
3359 {
3360 size_t i;
3361 int ret = 0;
3362 const size_t count = lttng_dynamic_array_get_count(
3363 &stream->read_subbuffer_ops.post_consume_cbs);
3364
3365 for (i = 0; i < count; i++) {
3366 const post_consume_cb op = *(post_consume_cb *) lttng_dynamic_array_get_element(
3367 &stream->read_subbuffer_ops.post_consume_cbs,
3368 i);
3369
3370 ret = op(stream, subbuffer, ctx);
3371 if (ret) {
3372 goto end;
3373 }
3374 }
3375 end:
3376 return ret;
3377 }
3378
3379 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
3380 struct lttng_consumer_local_data *ctx,
3381 bool locked_by_caller)
3382 {
3383 ssize_t ret, written_bytes = 0;
3384 int rotation_ret;
3385 struct stream_subbuffer subbuffer = {};
3386 enum get_next_subbuffer_status get_next_status;
3387
3388 if (!locked_by_caller) {
3389 stream->read_subbuffer_ops.lock(stream);
3390 } else {
3391 stream->read_subbuffer_ops.assert_locked(stream);
3392 }
3393
3394 if (stream->read_subbuffer_ops.on_wake_up) {
3395 ret = stream->read_subbuffer_ops.on_wake_up(stream);
3396 if (ret) {
3397 goto end;
3398 }
3399 }
3400
3401 /*
3402 * If the stream was flagged to be ready for rotation before we extract
3403 * the next packet, rotate it now.
3404 */
3405 if (stream->rotate_ready) {
3406 DBG("Rotate stream before consuming data");
3407 ret = lttng_consumer_rotate_stream(ctx, stream);
3408 if (ret < 0) {
3409 ERR("Stream rotation error before consuming data");
3410 goto end;
3411 }
3412 }
3413
3414 get_next_status = stream->read_subbuffer_ops.get_next_subbuffer(
3415 stream, &subbuffer);
3416 switch (get_next_status) {
3417 case GET_NEXT_SUBBUFFER_STATUS_OK:
3418 break;
3419 case GET_NEXT_SUBBUFFER_STATUS_NO_DATA:
3420 /* Not an error. */
3421 ret = 0;
3422 goto sleep_stream;
3423 case GET_NEXT_SUBBUFFER_STATUS_ERROR:
3424 ret = -1;
3425 goto end;
3426 default:
3427 abort();
3428 }
3429
3430 ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
3431 stream, &subbuffer);
3432 if (ret) {
3433 goto error_put_subbuf;
3434 }
3435
3436 written_bytes = stream->read_subbuffer_ops.consume_subbuffer(
3437 ctx, stream, &subbuffer);
3438 if (written_bytes <= 0) {
3439 ERR("Error consuming subbuffer: (%zd)", written_bytes);
3440 ret = (int) written_bytes;
3441 goto error_put_subbuf;
3442 }
3443
3444 ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
3445 if (ret) {
3446 goto end;
3447 }
3448
3449 ret = post_consume(stream, &subbuffer, ctx);
3450 if (ret) {
3451 goto end;
3452 }
3453
3454 /*
3455 * After extracting the packet, we check if the stream is now ready to
3456 * be rotated and perform the action immediately.
3457 *
3458 * Don't overwrite `ret` as callers expect the number of bytes
3459 * consumed to be returned on success.
3460 */
3461 rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
3462 if (rotation_ret == 1) {
3463 rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
3464 if (rotation_ret < 0) {
3465 ret = rotation_ret;
3466 ERR("Stream rotation error after consuming data");
3467 goto end;
3468 }
3469
3470 } else if (rotation_ret < 0) {
3471 ret = rotation_ret;
3472 ERR("Failed to check if stream was ready to rotate after consuming data");
3473 goto end;
3474 }
3475
3476 sleep_stream:
3477 if (stream->read_subbuffer_ops.on_sleep) {
3478 stream->read_subbuffer_ops.on_sleep(stream, ctx);
3479 }
3480
3481 ret = written_bytes;
3482 end:
3483 if (!locked_by_caller) {
3484 stream->read_subbuffer_ops.unlock(stream);
3485 }
3486
3487 return ret;
3488 error_put_subbuf:
3489 (void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
3490 goto end;
3491 }
3492
3493 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
3494 {
3495 switch (the_consumer_data.type) {
3496 case LTTNG_CONSUMER_KERNEL:
3497 return lttng_kconsumer_on_recv_stream(stream);
3498 case LTTNG_CONSUMER32_UST:
3499 case LTTNG_CONSUMER64_UST:
3500 return lttng_ustconsumer_on_recv_stream(stream);
3501 default:
3502 ERR("Unknown consumer_data type");
3503 abort();
3504 return -ENOSYS;
3505 }
3506 }
3507
3508 /*
3509 * Allocate and set consumer data hash tables.
3510 */
3511 int lttng_consumer_init(void)
3512 {
3513 the_consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3514 if (!the_consumer_data.channel_ht) {
3515 goto error;
3516 }
3517
3518 the_consumer_data.channels_by_session_id_ht =
3519 lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3520 if (!the_consumer_data.channels_by_session_id_ht) {
3521 goto error;
3522 }
3523
3524 the_consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3525 if (!the_consumer_data.relayd_ht) {
3526 goto error;
3527 }
3528
3529 the_consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3530 if (!the_consumer_data.stream_list_ht) {
3531 goto error;
3532 }
3533
3534 the_consumer_data.stream_per_chan_id_ht =
3535 lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3536 if (!the_consumer_data.stream_per_chan_id_ht) {
3537 goto error;
3538 }
3539
3540 data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3541 if (!data_ht) {
3542 goto error;
3543 }
3544
3545 metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
3546 if (!metadata_ht) {
3547 goto error;
3548 }
3549
3550 the_consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
3551 if (!the_consumer_data.chunk_registry) {
3552 goto error;
3553 }
3554
3555 return 0;
3556
3557 error:
3558 return -1;
3559 }
3560
3561 /*
3562 * Process the ADD_RELAYD command receive by a consumer.
3563 *
3564 * This will create a relayd socket pair and add it to the relayd hash table.
3565 * The caller MUST acquire a RCU read side lock before calling it.
3566 */
3567 void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
3568 struct lttng_consumer_local_data *ctx, int sock,
3569 struct pollfd *consumer_sockpoll,
3570 struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
3571 uint64_t relayd_session_id)
3572 {
3573 int fd = -1, ret = -1, relayd_created = 0;
3574 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
3575 struct consumer_relayd_sock_pair *relayd = NULL;
3576
3577 LTTNG_ASSERT(ctx);
3578 LTTNG_ASSERT(relayd_sock);
3579 ASSERT_RCU_READ_LOCKED();
3580
3581 DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
3582
3583 /* Get relayd reference if exists. */
3584 relayd = consumer_find_relayd(net_seq_idx);
3585 if (relayd == NULL) {
3586 LTTNG_ASSERT(sock_type == LTTNG_STREAM_CONTROL);
3587 /* Not found. Allocate one. */
3588 relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
3589 if (relayd == NULL) {
3590 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3591 goto error;
3592 } else {
3593 relayd->sessiond_session_id = sessiond_id;
3594 relayd_created = 1;
3595 }
3596
3597 /*
3598 * This code path MUST continue to the consumer send status message to
3599 * we can notify the session daemon and continue our work without
3600 * killing everything.
3601 */
3602 } else {
3603 /*
3604 * relayd key should never be found for control socket.
3605 */
3606 LTTNG_ASSERT(sock_type != LTTNG_STREAM_CONTROL);
3607 }
3608
3609 /* First send a status message before receiving the fds. */
3610 ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
3611 if (ret < 0) {
3612 /* Somehow, the session daemon is not responding anymore. */
3613 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3614 goto error_nosignal;
3615 }
3616
3617 /* Poll on consumer socket. */
3618 ret = lttng_consumer_poll_socket(consumer_sockpoll);
3619 if (ret) {
3620 /* Needing to exit in the middle of a command: error. */
3621 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
3622 goto error_nosignal;
3623 }
3624
3625 /* Get relayd socket from session daemon */
3626 ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
3627 if (ret != sizeof(fd)) {
3628 fd = -1; /* Just in case it gets set with an invalid value. */
3629
3630 /*
3631 * Failing to receive FDs might indicate a major problem such as
3632 * reaching a fd limit during the receive where the kernel returns a
3633 * MSG_CTRUNC and fails to cleanup the fd in the queue. Any case, we
3634 * don't take any chances and stop everything.
3635 *
3636 * XXX: Feature request #558 will fix that and avoid this possible
3637 * issue when reaching the fd limit.
3638 */
3639 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
3640 ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD;
3641 goto error;
3642 }
3643
3644 /* Copy socket information and received FD */
3645 switch (sock_type) {
3646 case LTTNG_STREAM_CONTROL:
3647 /* Copy received lttcomm socket */
3648 lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
3649 ret = lttcomm_create_sock(&relayd->control_sock.sock);
3650 /* Handle create_sock error. */
3651 if (ret < 0) {
3652 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3653 goto error;
3654 }
3655 /*
3656 * Close the socket created internally by
3657 * lttcomm_create_sock, so we can replace it by the one
3658 * received from sessiond.
3659 */
3660 if (close(relayd->control_sock.sock.fd)) {
3661 PERROR("close");
3662 }
3663
3664 /* Assign new file descriptor */
3665 relayd->control_sock.sock.fd = fd;
3666 /* Assign version values. */
3667 relayd->control_sock.major = relayd_sock->major;
3668 relayd->control_sock.minor = relayd_sock->minor;
3669
3670 relayd->relayd_session_id = relayd_session_id;
3671
3672 break;
3673 case LTTNG_STREAM_DATA:
3674 /* Copy received lttcomm socket */
3675 lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
3676 ret = lttcomm_create_sock(&relayd->data_sock.sock);
3677 /* Handle create_sock error. */
3678 if (ret < 0) {
3679 ret_code = LTTCOMM_CONSUMERD_ENOMEM;
3680 goto error;
3681 }
3682 /*
3683 * Close the socket created internally by
3684 * lttcomm_create_sock, so we can replace it by the one
3685 * received from sessiond.
3686 */
3687 if (close(relayd->data_sock.sock.fd)) {
3688 PERROR("close");
3689 }
3690
3691 /* Assign new file descriptor */
3692 relayd->data_sock.sock.fd = fd;
3693 /* Assign version values. */
3694 relayd->data_sock.major = relayd_sock->major;
3695 relayd->data_sock.minor = relayd_sock->minor;
3696 break;
3697 default:
3698 ERR("Unknown relayd socket type (%d)", sock_type);
3699 ret_code = LTTCOMM_CONSUMERD_FATAL;
3700 goto error;
3701 }
3702
3703 DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
3704 sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
3705 relayd->net_seq_idx, fd);
3706 /*
3707 * We gave the ownership of the fd to the relayd structure. Set the
3708 * fd to -1 so we don't call close() on it in the error path below.
3709 */
3710 fd = -1;
3711
3712 /* We successfully added the socket. Send status back. */
3713 ret = consumer_send_status_msg(sock, ret_code);
3714 if (ret < 0) {
3715 /* Somehow, the session daemon is not responding anymore. */
3716 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3717 goto error_nosignal;
3718 }
3719
3720 /*
3721 * Add relayd socket pair to consumer data hashtable. If object already
3722 * exists or on error, the function gracefully returns.
3723 */
3724 relayd->ctx = ctx;
3725 add_relayd(relayd);
3726
3727 /* All good! */
3728 return;
3729
3730 error:
3731 if (consumer_send_status_msg(sock, ret_code) < 0) {
3732 lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
3733 }
3734
3735 error_nosignal:
3736 /* Close received socket if valid. */
3737 if (fd >= 0) {
3738 if (close(fd)) {
3739 PERROR("close received socket");
3740 }
3741 }
3742
3743 if (relayd_created) {
3744 free(relayd);
3745 }
3746 }
3747
3748 /*
3749 * Search for a relayd associated to the session id and return the reference.
3750 *
3751 * A rcu read side lock MUST be acquire before calling this function and locked
3752 * until the relayd object is no longer necessary.
3753 */
3754 static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
3755 {
3756 struct lttng_ht_iter iter;
3757 struct consumer_relayd_sock_pair *relayd = NULL;
3758
3759 ASSERT_RCU_READ_LOCKED();
3760
3761 /* Iterate over all relayd since they are indexed by net_seq_idx. */
3762 cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
3763 relayd, node.node) {
3764 /*
3765 * Check by sessiond id which is unique here where the relayd session
3766 * id might not be when having multiple relayd.
3767 */
3768 if (relayd->sessiond_session_id == id) {
3769 /* Found the relayd. There can be only one per id. */
3770 goto found;
3771 }
3772 }
3773
3774 return NULL;
3775
3776 found:
3777 return relayd;
3778 }
3779
3780 /*
3781 * Check if for a given session id there is still data needed to be extract
3782 * from the buffers.
3783 *
3784 * Return 1 if data is pending or else 0 meaning ready to be read.
3785 */
3786 int consumer_data_pending(uint64_t id)
3787 {
3788 int ret;
3789 struct lttng_ht_iter iter;
3790 struct lttng_ht *ht;
3791 struct lttng_consumer_stream *stream;
3792 struct consumer_relayd_sock_pair *relayd = NULL;
3793 int (*data_pending)(struct lttng_consumer_stream *);
3794
3795 DBG("Consumer data pending command on session id %" PRIu64, id);
3796
3797 rcu_read_lock();
3798 pthread_mutex_lock(&the_consumer_data.lock);
3799
3800 switch (the_consumer_data.type) {
3801 case LTTNG_CONSUMER_KERNEL:
3802 data_pending = lttng_kconsumer_data_pending;
3803 break;
3804 case LTTNG_CONSUMER32_UST:
3805 case LTTNG_CONSUMER64_UST:
3806 data_pending = lttng_ustconsumer_data_pending;
3807 break;
3808 default:
3809 ERR("Unknown consumer data type");
3810 abort();
3811 }
3812
3813 /* Ease our life a bit */
3814 ht = the_consumer_data.stream_list_ht;
3815
3816 cds_lfht_for_each_entry_duplicate(ht->ht,
3817 ht->hash_fct(&id, lttng_ht_seed),
3818 ht->match_fct, &id,
3819 &iter.iter, stream, node_session_id.node) {
3820 pthread_mutex_lock(&stream->lock);
3821
3822 /*
3823 * A removed node from the hash table indicates that the stream has
3824 * been deleted thus having a guarantee that the buffers are closed
3825 * on the consumer side. However, data can still be transmitted
3826 * over the network so don't skip the relayd check.
3827 */
3828 ret = cds_lfht_is_node_deleted(&stream->node.node);
3829 if (!ret) {
3830 /* Check the stream if there is data in the buffers. */
3831 ret = data_pending(stream);
3832 if (ret == 1) {
3833 pthread_mutex_unlock(&stream->lock);
3834 goto data_pending;
3835 }
3836 }
3837
3838 pthread_mutex_unlock(&stream->lock);
3839 }
3840
3841 relayd = find_relayd_by_session_id(id);
3842 if (relayd) {
3843 unsigned int is_data_inflight = 0;
3844
3845 /* Send init command for data pending. */
3846 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
3847 ret = relayd_begin_data_pending(&relayd->control_sock,
3848 relayd->relayd_session_id);
3849 if (ret < 0) {
3850 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3851 /* Communication error thus the relayd so no data pending. */
3852 goto data_not_pending;
3853 }
3854
3855 cds_lfht_for_each_entry_duplicate(ht->ht,
3856 ht->hash_fct(&id, lttng_ht_seed),
3857 ht->match_fct, &id,
3858 &iter.iter, stream, node_session_id.node) {
3859 if (stream->metadata_flag) {
3860 ret = relayd_quiescent_control(&relayd->control_sock,
3861 stream->relayd_stream_id);
3862 } else {
3863 ret = relayd_data_pending(&relayd->control_sock,
3864 stream->relayd_stream_id,
3865 stream->next_net_seq_num - 1);
3866 }
3867
3868 if (ret == 1) {
3869 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3870 goto data_pending;
3871 } else if (ret < 0) {
3872 ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
3873 lttng_consumer_cleanup_relayd(relayd);
3874 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3875 goto data_not_pending;
3876 }
3877 }
3878
3879 /* Send end command for data pending. */
3880 ret = relayd_end_data_pending(&relayd->control_sock,
3881 relayd->relayd_session_id, &is_data_inflight);
3882 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
3883 if (ret < 0) {
3884 ERR("Relayd end data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
3885 lttng_consumer_cleanup_relayd(relayd);
3886 goto data_not_pending;
3887 }
3888 if (is_data_inflight) {
3889 goto data_pending;
3890 }
3891 }
3892
3893 /*
3894 * Finding _no_ node in the hash table and no inflight data means that the
3895 * stream(s) have been removed thus data is guaranteed to be available for
3896 * analysis from the trace files.
3897 */
3898
3899 data_not_pending:
3900 /* Data is available to be read by a viewer. */
3901 pthread_mutex_unlock(&the_consumer_data.lock);
3902 rcu_read_unlock();
3903 return 0;
3904
3905 data_pending:
3906 /* Data is still being extracted from buffers. */
3907 pthread_mutex_unlock(&the_consumer_data.lock);
3908 rcu_read_unlock();
3909 return 1;
3910 }
3911
3912 /*
3913 * Send a ret code status message to the sessiond daemon.
3914 *
3915 * Return the sendmsg() return value.
3916 */
3917 int consumer_send_status_msg(int sock, int ret_code)
3918 {
3919 struct lttcomm_consumer_status_msg msg;
3920
3921 memset(&msg, 0, sizeof(msg));
3922 msg.ret_code = (lttcomm_return_code) ret_code;
3923
3924 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3925 }
3926
3927 /*
3928 * Send a channel status message to the sessiond daemon.
3929 *
3930 * Return the sendmsg() return value.
3931 */
3932 int consumer_send_status_channel(int sock,
3933 struct lttng_consumer_channel *channel)
3934 {
3935 struct lttcomm_consumer_status_channel msg;
3936
3937 LTTNG_ASSERT(sock >= 0);
3938
3939 memset(&msg, 0, sizeof(msg));
3940 if (!channel) {
3941 msg.ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
3942 } else {
3943 msg.ret_code = LTTCOMM_CONSUMERD_SUCCESS;
3944 msg.key = channel->key;
3945 msg.stream_count = channel->streams.count;
3946 }
3947
3948 return lttcomm_send_unix_sock(sock, &msg, sizeof(msg));
3949 }
3950
3951 unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
3952 unsigned long produced_pos, uint64_t nb_packets_per_stream,
3953 uint64_t max_sb_size)
3954 {
3955 unsigned long start_pos;
3956
3957 if (!nb_packets_per_stream) {
3958 return consumed_pos; /* Grab everything */
3959 }
3960 start_pos = produced_pos - lttng_offset_align_floor(produced_pos, max_sb_size);
3961 start_pos -= max_sb_size * nb_packets_per_stream;
3962 if ((long) (start_pos - consumed_pos) < 0) {
3963 return consumed_pos; /* Grab everything */
3964 }
3965 return start_pos;
3966 }
3967
3968 /* Stream lock must be held by the caller. */
3969 static int sample_stream_positions(struct lttng_consumer_stream *stream,
3970 unsigned long *produced, unsigned long *consumed)
3971 {
3972 int ret;
3973
3974 ASSERT_LOCKED(stream->lock);
3975
3976 ret = lttng_consumer_sample_snapshot_positions(stream);
3977 if (ret < 0) {
3978 ERR("Failed to sample snapshot positions");
3979 goto end;
3980 }
3981
3982 ret = lttng_consumer_get_produced_snapshot(stream, produced);
3983 if (ret < 0) {
3984 ERR("Failed to sample produced position");
3985 goto end;
3986 }
3987
3988 ret = lttng_consumer_get_consumed_snapshot(stream, consumed);
3989 if (ret < 0) {
3990 ERR("Failed to sample consumed position");
3991 goto end;
3992 }
3993
3994 end:
3995 return ret;
3996 }
3997
3998 /*
3999 * Sample the rotate position for all the streams of a channel. If a stream
4000 * is already at the rotate position (produced == consumed), we flag it as
4001 * ready for rotation. The rotation of ready streams occurs after we have
4002 * replied to the session daemon that we have finished sampling the positions.
4003 * Must be called with RCU read-side lock held to ensure existence of channel.
4004 *
4005 * Returns 0 on success, < 0 on error
4006 */
4007 int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
4008 uint64_t key, uint64_t relayd_id, uint32_t metadata,
4009 struct lttng_consumer_local_data *ctx)
4010 {
4011 int ret;
4012 struct lttng_consumer_stream *stream;
4013 struct lttng_ht_iter iter;
4014 struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
4015 struct lttng_dynamic_array stream_rotation_positions;
4016 uint64_t next_chunk_id, stream_count = 0;
4017 enum lttng_trace_chunk_status chunk_status;
4018 const bool is_local_trace = relayd_id == -1ULL;
4019 struct consumer_relayd_sock_pair *relayd = NULL;
4020 bool rotating_to_new_chunk = true;
4021 /* Array of `struct lttng_consumer_stream *` */
4022 struct lttng_dynamic_pointer_array streams_packet_to_open;
4023 size_t stream_idx;
4024
4025 ASSERT_RCU_READ_LOCKED();
4026
4027 DBG("Consumer sample rotate position for channel %" PRIu64, key);
4028
4029 lttng_dynamic_array_init(&stream_rotation_positions,
4030 sizeof(struct relayd_stream_rotation_position), NULL);
4031 lttng_dynamic_pointer_array_init(&streams_packet_to_open, NULL);
4032
4033 rcu_read_lock();
4034
4035 pthread_mutex_lock(&channel->lock);
4036 LTTNG_ASSERT(channel->trace_chunk);
4037 chunk_status = lttng_trace_chunk_get_id(channel->trace_chunk,
4038 &next_chunk_id);
4039 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4040 ret = -1;
4041 goto end_unlock_channel;
4042 }
4043
4044 cds_lfht_for_each_entry_duplicate(ht->ht,
4045 ht->hash_fct(&channel->key, lttng_ht_seed),
4046 ht->match_fct, &channel->key, &iter.iter,
4047 stream, node_channel_id.node) {
4048 unsigned long produced_pos = 0, consumed_pos = 0;
4049
4050 health_code_update();
4051
4052 /*
4053 * Lock stream because we are about to change its state.
4054 */
4055 pthread_mutex_lock(&stream->lock);
4056
4057 if (stream->trace_chunk == stream->chan->trace_chunk) {
4058 rotating_to_new_chunk = false;
4059 }
4060
4061 /*
4062 * Do not flush a packet when rotating from a NULL trace
4063 * chunk. The stream has no means to output data, and the prior
4064 * rotation which rotated to NULL performed that side-effect
4065 * already. No new data can be produced when a stream has no
4066 * associated trace chunk (e.g. a stop followed by a rotate).
4067 */
4068 if (stream->trace_chunk) {
4069 bool flush_active;
4070
4071 if (stream->metadata_flag) {
4072 /*
4073 * Don't produce an empty metadata packet,
4074 * simply close the current one.
4075 *
4076 * Metadata is regenerated on every trace chunk
4077 * switch; there is no concern that no data was
4078 * produced.
4079 */
4080 flush_active = true;
4081 } else {
4082 /*
4083 * Only flush an empty packet if the "packet
4084 * open" could not be performed on transition
4085 * to a new trace chunk and no packets were
4086 * consumed within the chunk's lifetime.
4087 */
4088 if (stream->opened_packet_in_current_trace_chunk) {
4089 flush_active = true;
4090 } else {
4091 /*
4092 * Stream could have been full at the
4093 * time of rotation, but then have had
4094 * no activity at all.
4095 *
4096 * It is important to flush a packet
4097 * to prevent 0-length files from being
4098 * produced as most viewers choke on
4099 * them.
4100 *
4101 * Unfortunately viewers will not be
4102 * able to know that tracing was active
4103 * for this stream during this trace
4104 * chunk's lifetime.
4105 */
4106 ret = sample_stream_positions(stream, &produced_pos, &consumed_pos);
4107 if (ret) {
4108 goto end_unlock_stream;
4109 }
4110
4111 /*
4112 * Don't flush an empty packet if data
4113 * was produced; it will be consumed
4114 * before the rotation completes.
4115 */
4116 flush_active = produced_pos != consumed_pos;
4117 if (!flush_active) {
4118 const char *trace_chunk_name;
4119 uint64_t trace_chunk_id;
4120
4121 chunk_status = lttng_trace_chunk_get_name(
4122 stream->trace_chunk,
4123 &trace_chunk_name,
4124 NULL);
4125 if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NONE) {
4126 trace_chunk_name = "none";
4127 }
4128
4129 /*
4130 * Consumer trace chunks are
4131 * never anonymous.
4132 */
4133 chunk_status = lttng_trace_chunk_get_id(
4134 stream->trace_chunk,
4135 &trace_chunk_id);
4136 LTTNG_ASSERT(chunk_status ==
4137 LTTNG_TRACE_CHUNK_STATUS_OK);
4138
4139 DBG("Unable to open packet for stream during trace chunk's lifetime. "
4140 "Flushing an empty packet to prevent an empty file from being created: "
4141 "stream id = %" PRIu64 ", trace chunk name = `%s`, trace chunk id = %" PRIu64,
4142 stream->key, trace_chunk_name, trace_chunk_id);
4143 }
4144 }
4145 }
4146
4147 /*
4148 * Close the current packet before sampling the
4149 * ring buffer positions.
4150 */
4151 ret = consumer_stream_flush_buffer(stream, flush_active);
4152 if (ret < 0) {
4153 ERR("Failed to flush stream %" PRIu64 " during channel rotation",
4154 stream->key);
4155 goto end_unlock_stream;
4156 }
4157 }
4158
4159 ret = lttng_consumer_take_snapshot(stream);
4160 if (ret < 0 && ret != -ENODATA && ret != -EAGAIN) {
4161 ERR("Failed to sample snapshot position during channel rotation");
4162 goto end_unlock_stream;
4163 }
4164 if (!ret) {
4165 ret = lttng_consumer_get_produced_snapshot(stream,
4166 &produced_pos);
4167 if (ret < 0) {
4168 ERR("Failed to sample produced position during channel rotation");
4169 goto end_unlock_stream;
4170 }
4171
4172 ret = lttng_consumer_get_consumed_snapshot(stream,
4173 &consumed_pos);
4174 if (ret < 0) {
4175 ERR("Failed to sample consumed position during channel rotation");
4176 goto end_unlock_stream;
4177 }
4178 }
4179 /*
4180 * Align produced position on the start-of-packet boundary of the first
4181 * packet going into the next trace chunk.
4182 */
4183 produced_pos = lttng_align_floor(produced_pos, stream->max_sb_size);
4184 if (consumed_pos == produced_pos) {
4185 DBG("Set rotate ready for stream %" PRIu64 " produced = %lu consumed = %lu",
4186 stream->key, produced_pos, consumed_pos);
4187 stream->rotate_ready = true;
4188 } else {
4189 DBG("Different consumed and produced positions "
4190 "for stream %" PRIu64 " produced = %lu consumed = %lu",
4191 stream->key, produced_pos, consumed_pos);
4192 }
4193 /*
4194 * The rotation position is based on the packet_seq_num of the
4195 * packet following the last packet that was consumed for this
4196 * stream, incremented by the offset between produced and
4197 * consumed positions. This rotation position is a lower bound
4198 * (inclusive) at which the next trace chunk starts. Since it
4199 * is a lower bound, it is OK if the packet_seq_num does not
4200 * correspond exactly to the same packet identified by the
4201 * consumed_pos, which can happen in overwrite mode.
4202 */
4203 if (stream->sequence_number_unavailable) {
4204 /*
4205 * Rotation should never be performed on a session which
4206 * interacts with a pre-2.8 lttng-modules, which does
4207 * not implement packet sequence number.
4208 */
4209 ERR("Failure to rotate stream %" PRIu64 ": sequence number unavailable",
4210 stream->key);
4211 ret = -1;
4212 goto end_unlock_stream;
4213 }
4214 stream->rotate_position = stream->last_sequence_number + 1 +
4215 ((produced_pos - consumed_pos) / stream->max_sb_size);
4216 DBG("Set rotation position for stream %" PRIu64 " at position %" PRIu64,
4217 stream->key, stream->rotate_position);
4218
4219 if (!is_local_trace) {
4220 /*
4221 * The relay daemon control protocol expects a rotation
4222 * position as "the sequence number of the first packet
4223 * _after_ the current trace chunk".
4224 */
4225 const struct relayd_stream_rotation_position position = {
4226 .stream_id = stream->relayd_stream_id,
4227 .rotate_at_seq_num = stream->rotate_position,
4228 };
4229
4230 ret = lttng_dynamic_array_add_element(
4231 &stream_rotation_positions,
4232 &position);
4233 if (ret) {
4234 ERR("Failed to allocate stream rotation position");
4235 goto end_unlock_stream;
4236 }
4237 stream_count++;
4238 }
4239
4240 stream->opened_packet_in_current_trace_chunk = false;
4241
4242 if (rotating_to_new_chunk && !stream->metadata_flag) {
4243 /*
4244 * Attempt to flush an empty packet as close to the
4245 * rotation point as possible. In the event where a
4246 * stream remains inactive after the rotation point,
4247 * this ensures that the new trace chunk has a
4248 * beginning timestamp set at the begining of the
4249 * trace chunk instead of only creating an empty
4250 * packet when the trace chunk is stopped.
4251 *
4252 * This indicates to the viewers that the stream
4253 * was being recorded, but more importantly it
4254 * allows viewers to determine a useable trace
4255 * intersection.
4256 *
4257 * This presents a problem in the case where the
4258 * ring-buffer is completely full.
4259 *
4260 * Consider the following scenario:
4261 * - The consumption of data is slow (slow network,
4262 * for instance),
4263 * - The ring buffer is full,
4264 * - A rotation is initiated,
4265 * - The flush below does nothing (no space left to
4266 * open a new packet),
4267 * - The other streams rotate very soon, and new
4268 * data is produced in the new chunk,
4269 * - This stream completes its rotation long after the
4270 * rotation was initiated
4271 * - The session is stopped before any event can be
4272 * produced in this stream's buffers.
4273 *
4274 * The resulting trace chunk will have a single packet
4275 * temporaly at the end of the trace chunk for this
4276 * stream making the stream intersection more narrow
4277 * than it should be.
4278 *
4279 * To work-around this, an empty flush is performed
4280 * after the first consumption of a packet during a
4281 * rotation if open_packet fails. The idea is that
4282 * consuming a packet frees enough space to switch
4283 * packets in this scenario and allows the tracer to
4284 * "stamp" the beginning of the new trace chunk at the
4285 * earliest possible point.
4286 *
4287 * The packet open is performed after the channel
4288 * rotation to ensure that no attempt to open a packet
4289 * is performed in a stream that has no active trace
4290 * chunk.
4291 */
4292 ret = lttng_dynamic_pointer_array_add_pointer(
4293 &streams_packet_to_open, stream);
4294 if (ret) {
4295 PERROR("Failed to add a stream pointer to array of streams in which to open a packet");
4296 ret = -1;
4297 goto end_unlock_stream;
4298 }
4299 }
4300
4301 pthread_mutex_unlock(&stream->lock);
4302 }
4303 stream = NULL;
4304
4305 if (!is_local_trace) {
4306 relayd = consumer_find_relayd(relayd_id);
4307 if (!relayd) {
4308 ERR("Failed to find relayd %" PRIu64, relayd_id);
4309 ret = -1;
4310 goto end_unlock_channel;
4311 }
4312
4313 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4314 ret = relayd_rotate_streams(&relayd->control_sock, stream_count,
4315 rotating_to_new_chunk ? &next_chunk_id : NULL,
4316 (const struct relayd_stream_rotation_position *)
4317 stream_rotation_positions.buffer
4318 .data);
4319 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4320 if (ret < 0) {
4321 ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64,
4322 relayd->net_seq_idx);
4323 lttng_consumer_cleanup_relayd(relayd);
4324 goto end_unlock_channel;
4325 }
4326 }
4327
4328 for (stream_idx = 0;
4329 stream_idx < lttng_dynamic_pointer_array_get_count(
4330 &streams_packet_to_open);
4331 stream_idx++) {
4332 enum consumer_stream_open_packet_status status;
4333
4334 stream = (lttng_consumer_stream *) lttng_dynamic_pointer_array_get_pointer(
4335 &streams_packet_to_open, stream_idx);
4336
4337 pthread_mutex_lock(&stream->lock);
4338 status = consumer_stream_open_packet(stream);
4339 pthread_mutex_unlock(&stream->lock);
4340 switch (status) {
4341 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
4342 DBG("Opened a packet after a rotation: stream id = %" PRIu64
4343 ", channel name = %s, session id = %" PRIu64,
4344 stream->key, stream->chan->name,
4345 stream->chan->session_id);
4346 break;
4347 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
4348 /*
4349 * Can't open a packet as there is no space left
4350 * in the buffer. A new packet will be opened
4351 * once one has been consumed.
4352 */
4353 DBG("No space left to open a packet after a rotation: stream id = %" PRIu64
4354 ", channel name = %s, session id = %" PRIu64,
4355 stream->key, stream->chan->name,
4356 stream->chan->session_id);
4357 break;
4358 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
4359 /* Logged by callee. */
4360 ret = -1;
4361 goto end_unlock_channel;
4362 default:
4363 abort();
4364 }
4365 }
4366
4367 pthread_mutex_unlock(&channel->lock);
4368 ret = 0;
4369 goto end;
4370
4371 end_unlock_stream:
4372 pthread_mutex_unlock(&stream->lock);
4373 end_unlock_channel:
4374 pthread_mutex_unlock(&channel->lock);
4375 end:
4376 rcu_read_unlock();
4377 lttng_dynamic_array_reset(&stream_rotation_positions);
4378 lttng_dynamic_pointer_array_reset(&streams_packet_to_open);
4379 return ret;
4380 }
4381
4382 static
4383 int consumer_clear_buffer(struct lttng_consumer_stream *stream)
4384 {
4385 int ret = 0;
4386 unsigned long consumed_pos_before, consumed_pos_after;
4387
4388 ret = lttng_consumer_sample_snapshot_positions(stream);
4389 if (ret < 0) {
4390 ERR("Taking snapshot positions");
4391 goto end;
4392 }
4393
4394 ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_before);
4395 if (ret < 0) {
4396 ERR("Consumed snapshot position");
4397 goto end;
4398 }
4399
4400 switch (the_consumer_data.type) {
4401 case LTTNG_CONSUMER_KERNEL:
4402 ret = kernctl_buffer_clear(stream->wait_fd);
4403 if (ret < 0) {
4404 ERR("Failed to clear kernel stream (ret = %d)", ret);
4405 goto end;
4406 }
4407 break;
4408 case LTTNG_CONSUMER32_UST:
4409 case LTTNG_CONSUMER64_UST:
4410 ret = lttng_ustconsumer_clear_buffer(stream);
4411 if (ret < 0) {
4412 ERR("Failed to clear ust stream (ret = %d)", ret);
4413 goto end;
4414 }
4415 break;
4416 default:
4417 ERR("Unknown consumer_data type");
4418 abort();
4419 }
4420
4421 ret = lttng_consumer_sample_snapshot_positions(stream);
4422 if (ret < 0) {
4423 ERR("Taking snapshot positions");
4424 goto end;
4425 }
4426 ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_after);
4427 if (ret < 0) {
4428 ERR("Consumed snapshot position");
4429 goto end;
4430 }
4431 DBG("clear: before: %lu after: %lu", consumed_pos_before, consumed_pos_after);
4432 end:
4433 return ret;
4434 }
4435
4436 static
4437 int consumer_clear_stream(struct lttng_consumer_stream *stream)
4438 {
4439 int ret;
4440
4441 ret = consumer_stream_flush_buffer(stream, 1);
4442 if (ret < 0) {
4443 ERR("Failed to flush stream %" PRIu64 " during channel clear",
4444 stream->key);
4445 ret = LTTCOMM_CONSUMERD_FATAL;
4446 goto error;
4447 }
4448
4449 ret = consumer_clear_buffer(stream);
4450 if (ret < 0) {
4451 ERR("Failed to clear stream %" PRIu64 " during channel clear",
4452 stream->key);
4453 ret = LTTCOMM_CONSUMERD_FATAL;
4454 goto error;
4455 }
4456
4457 ret = LTTCOMM_CONSUMERD_SUCCESS;
4458 error:
4459 return ret;
4460 }
4461
4462 static
4463 int consumer_clear_unmonitored_channel(struct lttng_consumer_channel *channel)
4464 {
4465 int ret;
4466 struct lttng_consumer_stream *stream;
4467
4468 rcu_read_lock();
4469 pthread_mutex_lock(&channel->lock);
4470 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
4471 health_code_update();
4472 pthread_mutex_lock(&stream->lock);
4473 ret = consumer_clear_stream(stream);
4474 if (ret) {
4475 goto error_unlock;
4476 }
4477 pthread_mutex_unlock(&stream->lock);
4478 }
4479 pthread_mutex_unlock(&channel->lock);
4480 rcu_read_unlock();
4481 return 0;
4482
4483 error_unlock:
4484 pthread_mutex_unlock(&stream->lock);
4485 pthread_mutex_unlock(&channel->lock);
4486 rcu_read_unlock();
4487 return ret;
4488 }
4489
4490 /*
4491 * Check if a stream is ready to be rotated after extracting it.
4492 *
4493 * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
4494 * error. Stream lock must be held.
4495 */
4496 int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
4497 {
4498 DBG("Check is rotate ready for stream %" PRIu64
4499 " ready %u rotate_position %" PRIu64
4500 " last_sequence_number %" PRIu64,
4501 stream->key, stream->rotate_ready,
4502 stream->rotate_position, stream->last_sequence_number);
4503 if (stream->rotate_ready) {
4504 return 1;
4505 }
4506
4507 /*
4508 * If packet seq num is unavailable, it means we are interacting
4509 * with a pre-2.8 lttng-modules which does not implement the
4510 * sequence number. Rotation should never be used by sessiond in this
4511 * scenario.
4512 */
4513 if (stream->sequence_number_unavailable) {
4514 ERR("Internal error: rotation used on stream %" PRIu64
4515 " with unavailable sequence number",
4516 stream->key);
4517 return -1;
4518 }
4519
4520 if (stream->rotate_position == -1ULL ||
4521 stream->last_sequence_number == -1ULL) {
4522 return 0;
4523 }
4524
4525 /*
4526 * Rotate position not reached yet. The stream rotate position is
4527 * the position of the next packet belonging to the next trace chunk,
4528 * but consumerd considers rotation ready when reaching the last
4529 * packet of the current chunk, hence the "rotate_position - 1".
4530 */
4531
4532 DBG("Check is rotate ready for stream %" PRIu64
4533 " last_sequence_number %" PRIu64
4534 " rotate_position %" PRIu64,
4535 stream->key, stream->last_sequence_number,
4536 stream->rotate_position);
4537 if (stream->last_sequence_number >= stream->rotate_position - 1) {
4538 return 1;
4539 }
4540
4541 return 0;
4542 }
4543
4544 /*
4545 * Reset the state for a stream after a rotation occurred.
4546 */
4547 void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream)
4548 {
4549 DBG("lttng_consumer_reset_stream_rotate_state for stream %" PRIu64,
4550 stream->key);
4551 stream->rotate_position = -1ULL;
4552 stream->rotate_ready = false;
4553 }
4554
4555 /*
4556 * Perform the rotation a local stream file.
4557 */
4558 static
4559 int rotate_local_stream(struct lttng_consumer_local_data *ctx,
4560 struct lttng_consumer_stream *stream)
4561 {
4562 int ret = 0;
4563
4564 DBG("Rotate local stream: stream key %" PRIu64 ", channel key %" PRIu64,
4565 stream->key,
4566 stream->chan->key);
4567 stream->tracefile_size_current = 0;
4568 stream->tracefile_count_current = 0;
4569
4570 if (stream->out_fd >= 0) {
4571 ret = close(stream->out_fd);
4572 if (ret) {
4573 PERROR("Failed to close stream out_fd of channel \"%s\"",
4574 stream->chan->name);
4575 }
4576 stream->out_fd = -1;
4577 }
4578
4579 if (stream->index_file) {
4580 lttng_index_file_put(stream->index_file);
4581 stream->index_file = NULL;
4582 }
4583
4584 if (!stream->trace_chunk) {
4585 goto end;
4586 }
4587
4588 ret = consumer_stream_create_output_files(stream, true);
4589 end:
4590 return ret;
4591 }
4592
4593 /*
4594 * Performs the stream rotation for the rotate session feature if needed.
4595 * It must be called with the channel and stream locks held.
4596 *
4597 * Return 0 on success, a negative number of error.
4598 */
4599 int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
4600 struct lttng_consumer_stream *stream)
4601 {
4602 int ret;
4603
4604 DBG("Consumer rotate stream %" PRIu64, stream->key);
4605
4606 /*
4607 * Update the stream's 'current' chunk to the session's (channel)
4608 * now-current chunk.
4609 */
4610 lttng_trace_chunk_put(stream->trace_chunk);
4611 if (stream->chan->trace_chunk == stream->trace_chunk) {
4612 /*
4613 * A channel can be rotated and not have a "next" chunk
4614 * to transition to. In that case, the channel's "current chunk"
4615 * has not been closed yet, but it has not been updated to
4616 * a "next" trace chunk either. Hence, the stream, like its
4617 * parent channel, becomes part of no chunk and can't output
4618 * anything until a new trace chunk is created.
4619 */
4620 stream->trace_chunk = NULL;
4621 } else if (stream->chan->trace_chunk &&
4622 !lttng_trace_chunk_get(stream->chan->trace_chunk)) {
4623 ERR("Failed to acquire a reference to channel's trace chunk during stream rotation");
4624 ret = -1;
4625 goto error;
4626 } else {
4627 /*
4628 * Update the stream's trace chunk to its parent channel's
4629 * current trace chunk.
4630 */
4631 stream->trace_chunk = stream->chan->trace_chunk;
4632 }
4633
4634 if (stream->net_seq_idx == (uint64_t) -1ULL) {
4635 ret = rotate_local_stream(ctx, stream);
4636 if (ret < 0) {
4637 ERR("Failed to rotate stream, ret = %i", ret);
4638 goto error;
4639 }
4640 }
4641
4642 if (stream->metadata_flag && stream->trace_chunk) {
4643 /*
4644 * If the stream has transitioned to a new trace
4645 * chunk, the metadata should be re-dumped to the
4646 * newest chunk.
4647 *
4648 * However, it is possible for a stream to transition to
4649 * a "no-chunk" state. This can happen if a rotation
4650 * occurs on an inactive session. In such cases, the metadata
4651 * regeneration will happen when the next trace chunk is
4652 * created.
4653 */
4654 ret = consumer_metadata_stream_dump(stream);
4655 if (ret) {
4656 goto error;
4657 }
4658 }
4659 lttng_consumer_reset_stream_rotate_state(stream);
4660
4661 ret = 0;
4662
4663 error:
4664 return ret;
4665 }
4666
4667 /*
4668 * Rotate all the ready streams now.
4669 *
4670 * This is especially important for low throughput streams that have already
4671 * been consumed, we cannot wait for their next packet to perform the
4672 * rotation.
4673 * Need to be called with RCU read-side lock held to ensure existence of
4674 * channel.
4675 *
4676 * Returns 0 on success, < 0 on error
4677 */
4678 int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel,
4679 uint64_t key, struct lttng_consumer_local_data *ctx)
4680 {
4681 int ret;
4682 struct lttng_consumer_stream *stream;
4683 struct lttng_ht_iter iter;
4684 struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
4685
4686 ASSERT_RCU_READ_LOCKED();
4687
4688 rcu_read_lock();
4689
4690 DBG("Consumer rotate ready streams in channel %" PRIu64, key);
4691
4692 cds_lfht_for_each_entry_duplicate(ht->ht,
4693 ht->hash_fct(&channel->key, lttng_ht_seed),
4694 ht->match_fct, &channel->key, &iter.iter,
4695 stream, node_channel_id.node) {
4696 health_code_update();
4697
4698 pthread_mutex_lock(&stream->chan->lock);
4699 pthread_mutex_lock(&stream->lock);
4700
4701 if (!stream->rotate_ready) {
4702 pthread_mutex_unlock(&stream->lock);
4703 pthread_mutex_unlock(&stream->chan->lock);
4704 continue;
4705 }
4706 DBG("Consumer rotate ready stream %" PRIu64, stream->key);
4707
4708 ret = lttng_consumer_rotate_stream(ctx, stream);
4709 pthread_mutex_unlock(&stream->lock);
4710 pthread_mutex_unlock(&stream->chan->lock);
4711 if (ret) {
4712 goto end;
4713 }
4714 }
4715
4716 ret = 0;
4717
4718 end:
4719 rcu_read_unlock();
4720 return ret;
4721 }
4722
4723 enum lttcomm_return_code lttng_consumer_init_command(
4724 struct lttng_consumer_local_data *ctx,
4725 const lttng_uuid sessiond_uuid)
4726 {
4727 enum lttcomm_return_code ret;
4728 char uuid_str[LTTNG_UUID_STR_LEN];
4729
4730 if (ctx->sessiond_uuid.is_set) {
4731 ret = LTTCOMM_CONSUMERD_ALREADY_SET;
4732 goto end;
4733 }
4734
4735 ctx->sessiond_uuid.is_set = true;
4736 memcpy(ctx->sessiond_uuid.value, sessiond_uuid, sizeof(lttng_uuid));
4737 ret = LTTCOMM_CONSUMERD_SUCCESS;
4738 lttng_uuid_to_str(sessiond_uuid, uuid_str);
4739 DBG("Received session daemon UUID: %s", uuid_str);
4740 end:
4741 return ret;
4742 }
4743
4744 enum lttcomm_return_code lttng_consumer_create_trace_chunk(
4745 const uint64_t *relayd_id, uint64_t session_id,
4746 uint64_t chunk_id,
4747 time_t chunk_creation_timestamp,
4748 const char *chunk_override_name,
4749 const struct lttng_credentials *credentials,
4750 struct lttng_directory_handle *chunk_directory_handle)
4751 {
4752 int ret;
4753 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
4754 struct lttng_trace_chunk *created_chunk = NULL, *published_chunk = NULL;
4755 enum lttng_trace_chunk_status chunk_status;
4756 char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
4757 char creation_timestamp_buffer[ISO8601_STR_LEN];
4758 const char *relayd_id_str = "(none)";
4759 const char *creation_timestamp_str;
4760 struct lttng_ht_iter iter;
4761 struct lttng_consumer_channel *channel;
4762
4763 if (relayd_id) {
4764 /* Only used for logging purposes. */
4765 ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
4766 "%" PRIu64, *relayd_id);
4767 if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
4768 relayd_id_str = relayd_id_buffer;
4769 } else {
4770 relayd_id_str = "(formatting error)";
4771 }
4772 }
4773
4774 /* Local protocol error. */
4775 LTTNG_ASSERT(chunk_creation_timestamp);
4776 ret = time_to_iso8601_str(chunk_creation_timestamp,
4777 creation_timestamp_buffer,
4778 sizeof(creation_timestamp_buffer));
4779 creation_timestamp_str = !ret ? creation_timestamp_buffer :
4780 "(formatting error)";
4781
4782 DBG("Consumer create trace chunk command: relay_id = %s"
4783 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
4784 ", chunk_override_name = %s"
4785 ", chunk_creation_timestamp = %s",
4786 relayd_id_str, session_id, chunk_id,
4787 chunk_override_name ? : "(none)",
4788 creation_timestamp_str);
4789
4790 /*
4791 * The trace chunk registry, as used by the consumer daemon, implicitly
4792 * owns the trace chunks. This is only needed in the consumer since
4793 * the consumer has no notion of a session beyond session IDs being
4794 * used to identify other objects.
4795 *
4796 * The lttng_trace_chunk_registry_publish() call below provides a
4797 * reference which is not released; it implicitly becomes the session
4798 * daemon's reference to the chunk in the consumer daemon.
4799 *
4800 * The lifetime of trace chunks in the consumer daemon is managed by
4801 * the session daemon through the LTTNG_CONSUMER_CREATE_TRACE_CHUNK
4802 * and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands.
4803 */
4804 created_chunk = lttng_trace_chunk_create(chunk_id,
4805 chunk_creation_timestamp, NULL);
4806 if (!created_chunk) {
4807 ERR("Failed to create trace chunk");
4808 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4809 goto error;
4810 }
4811
4812 if (chunk_override_name) {
4813 chunk_status = lttng_trace_chunk_override_name(created_chunk,
4814 chunk_override_name);
4815 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4816 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4817 goto error;
4818 }
4819 }
4820
4821 if (chunk_directory_handle) {
4822 chunk_status = lttng_trace_chunk_set_credentials(created_chunk,
4823 credentials);
4824 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4825 ERR("Failed to set trace chunk credentials");
4826 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4827 goto error;
4828 }
4829 /*
4830 * The consumer daemon has no ownership of the chunk output
4831 * directory.
4832 */
4833 chunk_status = lttng_trace_chunk_set_as_user(created_chunk,
4834 chunk_directory_handle);
4835 chunk_directory_handle = NULL;
4836 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4837 ERR("Failed to set trace chunk's directory handle");
4838 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4839 goto error;
4840 }
4841 }
4842
4843 published_chunk = lttng_trace_chunk_registry_publish_chunk(
4844 the_consumer_data.chunk_registry, session_id,
4845 created_chunk);
4846 lttng_trace_chunk_put(created_chunk);
4847 created_chunk = NULL;
4848 if (!published_chunk) {
4849 ERR("Failed to publish trace chunk");
4850 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4851 goto error;
4852 }
4853
4854 rcu_read_lock();
4855 cds_lfht_for_each_entry_duplicate(
4856 the_consumer_data.channels_by_session_id_ht->ht,
4857 the_consumer_data.channels_by_session_id_ht->hash_fct(
4858 &session_id, lttng_ht_seed),
4859 the_consumer_data.channels_by_session_id_ht->match_fct,
4860 &session_id, &iter.iter, channel,
4861 channels_by_session_id_ht_node.node) {
4862 ret = lttng_consumer_channel_set_trace_chunk(channel,
4863 published_chunk);
4864 if (ret) {
4865 /*
4866 * Roll-back the creation of this chunk.
4867 *
4868 * This is important since the session daemon will
4869 * assume that the creation of this chunk failed and
4870 * will never ask for it to be closed, resulting
4871 * in a leak and an inconsistent state for some
4872 * channels.
4873 */
4874 enum lttcomm_return_code close_ret;
4875 char path[LTTNG_PATH_MAX];
4876
4877 DBG("Failed to set new trace chunk on existing channels, rolling back");
4878 close_ret = lttng_consumer_close_trace_chunk(relayd_id,
4879 session_id, chunk_id,
4880 chunk_creation_timestamp, NULL,
4881 path);
4882 if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
4883 ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
4884 session_id, chunk_id);
4885 }
4886
4887 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4888 break;
4889 }
4890 }
4891
4892 if (relayd_id) {
4893 struct consumer_relayd_sock_pair *relayd;
4894
4895 relayd = consumer_find_relayd(*relayd_id);
4896 if (relayd) {
4897 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
4898 ret = relayd_create_trace_chunk(
4899 &relayd->control_sock, published_chunk);
4900 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
4901 } else {
4902 ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64, *relayd_id);
4903 }
4904
4905 if (!relayd || ret) {
4906 enum lttcomm_return_code close_ret;
4907 char path[LTTNG_PATH_MAX];
4908
4909 close_ret = lttng_consumer_close_trace_chunk(relayd_id,
4910 session_id,
4911 chunk_id,
4912 chunk_creation_timestamp,
4913 NULL, path);
4914 if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
4915 ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
4916 session_id,
4917 chunk_id);
4918 }
4919
4920 ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
4921 goto error_unlock;
4922 }
4923 }
4924 error_unlock:
4925 rcu_read_unlock();
4926 error:
4927 /* Release the reference returned by the "publish" operation. */
4928 lttng_trace_chunk_put(published_chunk);
4929 lttng_trace_chunk_put(created_chunk);
4930 return ret_code;
4931 }
4932
4933 enum lttcomm_return_code lttng_consumer_close_trace_chunk(
4934 const uint64_t *relayd_id, uint64_t session_id,
4935 uint64_t chunk_id, time_t chunk_close_timestamp,
4936 const enum lttng_trace_chunk_command_type *close_command,
4937 char *path)
4938 {
4939 enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
4940 struct lttng_trace_chunk *chunk;
4941 char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
4942 const char *relayd_id_str = "(none)";
4943 const char *close_command_name = "none";
4944 struct lttng_ht_iter iter;
4945 struct lttng_consumer_channel *channel;
4946 enum lttng_trace_chunk_status chunk_status;
4947
4948 if (relayd_id) {
4949 int ret;
4950
4951 /* Only used for logging purposes. */
4952 ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
4953 "%" PRIu64, *relayd_id);
4954 if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
4955 relayd_id_str = relayd_id_buffer;
4956 } else {
4957 relayd_id_str = "(formatting error)";
4958 }
4959 }
4960 if (close_command) {
4961 close_command_name = lttng_trace_chunk_command_type_get_name(
4962 *close_command);
4963 }
4964
4965 DBG("Consumer close trace chunk command: relayd_id = %s"
4966 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
4967 ", close command = %s",
4968 relayd_id_str, session_id, chunk_id,
4969 close_command_name);
4970
4971 chunk = lttng_trace_chunk_registry_find_chunk(
4972 the_consumer_data.chunk_registry, session_id, chunk_id);
4973 if (!chunk) {
4974 ERR("Failed to find chunk: session_id = %" PRIu64
4975 ", chunk_id = %" PRIu64,
4976 session_id, chunk_id);
4977 ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
4978 goto end;
4979 }
4980
4981 chunk_status = lttng_trace_chunk_set_close_timestamp(chunk,
4982 chunk_close_timestamp);
4983 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4984 ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
4985 goto end;
4986 }
4987
4988 if (close_command) {
4989 chunk_status = lttng_trace_chunk_set_close_command(
4990 chunk, *close_command);
4991 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
4992 ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
4993 goto end;
4994 }
4995 }
4996
4997 /*
4998 * chunk is now invalid to access as we no longer hold a reference to
4999 * it; it is only kept around to compare it (by address) to the
5000 * current chunk found in the session's channels.
5001 */
5002 rcu_read_lock();
5003 cds_lfht_for_each_entry(the_consumer_data.channel_ht->ht, &iter.iter,
5004 channel, node.node) {
5005 int ret;
5006
5007 /*
5008 * Only change the channel's chunk to NULL if it still
5009 * references the chunk being closed. The channel may
5010 * reference a newer channel in the case of a session
5011 * rotation. When a session rotation occurs, the "next"
5012 * chunk is created before the "current" chunk is closed.
5013 */
5014 if (channel->trace_chunk != chunk) {
5015 continue;
5016 }
5017 ret = lttng_consumer_channel_set_trace_chunk(channel, NULL);
5018 if (ret) {
5019 /*
5020 * Attempt to close the chunk on as many channels as
5021 * possible.
5022 */
5023 ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
5024 }
5025 }
5026
5027 if (relayd_id) {
5028 int ret;
5029 struct consumer_relayd_sock_pair *relayd;
5030
5031 relayd = consumer_find_relayd(*relayd_id);
5032 if (relayd) {
5033 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
5034 ret = relayd_close_trace_chunk(
5035 &relayd->control_sock, chunk,
5036 path);
5037 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
5038 } else {
5039 ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64,
5040 *relayd_id);
5041 }
5042
5043 if (!relayd || ret) {
5044 ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
5045 goto error_unlock;
5046 }
5047 }
5048 error_unlock:
5049 rcu_read_unlock();
5050 end:
5051 /*
5052 * Release the reference returned by the "find" operation and
5053 * the session daemon's implicit reference to the chunk.
5054 */
5055 lttng_trace_chunk_put(chunk);
5056 lttng_trace_chunk_put(chunk);
5057
5058 return ret_code;
5059 }
5060
5061 enum lttcomm_return_code lttng_consumer_trace_chunk_exists(
5062 const uint64_t *relayd_id, uint64_t session_id,
5063 uint64_t chunk_id)
5064 {
5065 int ret;
5066 enum lttcomm_return_code ret_code;
5067 char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
5068 const char *relayd_id_str = "(none)";
5069 const bool is_local_trace = !relayd_id;
5070 struct consumer_relayd_sock_pair *relayd = NULL;
5071 bool chunk_exists_local, chunk_exists_remote;
5072
5073 if (relayd_id) {
5074 /* Only used for logging purposes. */
5075 ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
5076 "%" PRIu64, *relayd_id);
5077 if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
5078 relayd_id_str = relayd_id_buffer;
5079 } else {
5080 relayd_id_str = "(formatting error)";
5081 }
5082 }
5083
5084 DBG("Consumer trace chunk exists command: relayd_id = %s"
5085 ", chunk_id = %" PRIu64, relayd_id_str,
5086 chunk_id);
5087 ret = lttng_trace_chunk_registry_chunk_exists(
5088 the_consumer_data.chunk_registry, session_id, chunk_id,
5089 &chunk_exists_local);
5090 if (ret) {
5091 /* Internal error. */
5092 ERR("Failed to query the existence of a trace chunk");
5093 ret_code = LTTCOMM_CONSUMERD_FATAL;
5094 goto end;
5095 }
5096 DBG("Trace chunk %s locally",
5097 chunk_exists_local ? "exists" : "does not exist");
5098 if (chunk_exists_local) {
5099 ret_code = LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL;
5100 goto end;
5101 } else if (is_local_trace) {
5102 ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
5103 goto end;
5104 }
5105
5106 rcu_read_lock();
5107 relayd = consumer_find_relayd(*relayd_id);
5108 if (!relayd) {
5109 ERR("Failed to find relayd %" PRIu64, *relayd_id);
5110 ret_code = LTTCOMM_CONSUMERD_INVALID_PARAMETERS;
5111 goto end_rcu_unlock;
5112 }
5113 DBG("Looking up existence of trace chunk on relay daemon");
5114 pthread_mutex_lock(&relayd->ctrl_sock_mutex);
5115 ret = relayd_trace_chunk_exists(&relayd->control_sock, chunk_id,
5116 &chunk_exists_remote);
5117 pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
5118 if (ret < 0) {
5119 ERR("Failed to look-up the existence of trace chunk on relay daemon");
5120 ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
5121 goto end_rcu_unlock;
5122 }
5123
5124 ret_code = chunk_exists_remote ?
5125 LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE :
5126 LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
5127 DBG("Trace chunk %s on relay daemon",
5128 chunk_exists_remote ? "exists" : "does not exist");
5129
5130 end_rcu_unlock:
5131 rcu_read_unlock();
5132 end:
5133 return ret_code;
5134 }
5135
5136 static
5137 int consumer_clear_monitored_channel(struct lttng_consumer_channel *channel)
5138 {
5139 struct lttng_ht *ht;
5140 struct lttng_consumer_stream *stream;
5141 struct lttng_ht_iter iter;
5142 int ret;
5143
5144 ht = the_consumer_data.stream_per_chan_id_ht;
5145
5146 rcu_read_lock();
5147 cds_lfht_for_each_entry_duplicate(ht->ht,
5148 ht->hash_fct(&channel->key, lttng_ht_seed),
5149 ht->match_fct, &channel->key,
5150 &iter.iter, stream, node_channel_id.node) {
5151 /*
5152 * Protect against teardown with mutex.
5153 */
5154 pthread_mutex_lock(&stream->lock);
5155 if (cds_lfht_is_node_deleted(&stream->node.node)) {
5156 goto next;
5157 }
5158 ret = consumer_clear_stream(stream);
5159 if (ret) {
5160 goto error_unlock;
5161 }
5162 next:
5163 pthread_mutex_unlock(&stream->lock);
5164 }
5165 rcu_read_unlock();
5166 return LTTCOMM_CONSUMERD_SUCCESS;
5167
5168 error_unlock:
5169 pthread_mutex_unlock(&stream->lock);
5170 rcu_read_unlock();
5171 return ret;
5172 }
5173
5174 int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel)
5175 {
5176 int ret;
5177
5178 DBG("Consumer clear channel %" PRIu64, channel->key);
5179
5180 if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
5181 /*
5182 * Nothing to do for the metadata channel/stream.
5183 * Snapshot mechanism already take care of the metadata
5184 * handling/generation, and monitored channels only need to
5185 * have their data stream cleared..
5186 */
5187 ret = LTTCOMM_CONSUMERD_SUCCESS;
5188 goto end;
5189 }
5190
5191 if (!channel->monitor) {
5192 ret = consumer_clear_unmonitored_channel(channel);
5193 } else {
5194 ret = consumer_clear_monitored_channel(channel);
5195 }
5196 end:
5197 return ret;
5198 }
5199
5200 enum lttcomm_return_code lttng_consumer_open_channel_packets(
5201 struct lttng_consumer_channel *channel)
5202 {
5203 struct lttng_consumer_stream *stream;
5204 enum lttcomm_return_code ret = LTTCOMM_CONSUMERD_SUCCESS;
5205
5206 if (channel->metadata_stream) {
5207 ERR("Open channel packets command attempted on a metadata channel");
5208 ret = LTTCOMM_CONSUMERD_INVALID_PARAMETERS;
5209 goto end;
5210 }
5211
5212 rcu_read_lock();
5213 cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
5214 enum consumer_stream_open_packet_status status;
5215
5216 pthread_mutex_lock(&stream->lock);
5217 if (cds_lfht_is_node_deleted(&stream->node.node)) {
5218 goto next;
5219 }
5220
5221 status = consumer_stream_open_packet(stream);
5222 switch (status) {
5223 case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
5224 DBG("Opened a packet in \"open channel packets\" command: stream id = %" PRIu64
5225 ", channel name = %s, session id = %" PRIu64,
5226 stream->key, stream->chan->name,
5227 stream->chan->session_id);
5228 stream->opened_packet_in_current_trace_chunk = true;
5229 break;
5230 case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
5231 DBG("No space left to open a packet in \"open channel packets\" command: stream id = %" PRIu64
5232 ", channel name = %s, session id = %" PRIu64,
5233 stream->key, stream->chan->name,
5234 stream->chan->session_id);
5235 break;
5236 case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
5237 /*
5238 * Only unexpected internal errors can lead to this
5239 * failing. Report an unknown error.
5240 */
5241 ERR("Failed to flush empty buffer in \"open channel packets\" command: stream id = %" PRIu64
5242 ", channel id = %" PRIu64
5243 ", channel name = %s"
5244 ", session id = %" PRIu64,
5245 stream->key, channel->key,
5246 channel->name, channel->session_id);
5247 ret = LTTCOMM_CONSUMERD_UNKNOWN_ERROR;
5248 goto error_unlock;
5249 default:
5250 abort();
5251 }
5252
5253 next:
5254 pthread_mutex_unlock(&stream->lock);
5255 }
5256
5257 end_rcu_unlock:
5258 rcu_read_unlock();
5259 end:
5260 return ret;
5261
5262 error_unlock:
5263 pthread_mutex_unlock(&stream->lock);
5264 goto end_rcu_unlock;
5265 }
5266
5267 void lttng_consumer_sigbus_handle(void *addr)
5268 {
5269 lttng_ustconsumer_sigbus_handle(addr);
5270 }
This page took 0.154102 seconds and 4 git commands to generate.