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