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