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