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