Fix: use lttng_waiter instead of futex in notification thread
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread-events.c
CommitLineData
ab0ee2ca
JG
1/*
2 * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
3 *
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
11 * more details.
12 *
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16 */
17
18#define _LGPL_SOURCE
19#include <urcu.h>
20#include <urcu/rculfhash.h>
21
22#include "notification-thread.h"
23#include "notification-thread-events.h"
24#include "notification-thread-commands.h"
25#include <common/defaults.h>
26#include <common/error.h>
27#include <common/futex.h>
28#include <common/unix.h>
29#include <common/dynamic-buffer.h>
30#include <common/hashtable/utils.h>
31#include <common/sessiond-comm/sessiond-comm.h>
32#include <common/macros.h>
33#include <lttng/condition/condition.h>
34#include <lttng/action/action.h>
35#include <lttng/notification/notification-internal.h>
36#include <lttng/condition/condition-internal.h>
37#include <lttng/condition/buffer-usage-internal.h>
38#include <lttng/notification/channel-internal.h>
39#include <time.h>
40#include <unistd.h>
41#include <assert.h>
42#include <inttypes.h>
d53ea4e4 43#include <fcntl.h>
ab0ee2ca
JG
44
45#define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
46#define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
47
48struct lttng_trigger_list_element {
49 struct lttng_trigger *trigger;
50 struct cds_list_head node;
51};
52
53struct lttng_channel_trigger_list {
54 struct channel_key channel_key;
55 struct cds_list_head list;
56 struct cds_lfht_node channel_triggers_ht_node;
57};
58
59struct lttng_trigger_ht_element {
60 struct lttng_trigger *trigger;
61 struct cds_lfht_node node;
62};
63
64struct lttng_condition_list_element {
65 struct lttng_condition *condition;
66 struct cds_list_head node;
67};
68
69struct notification_client_list_element {
70 struct notification_client *client;
71 struct cds_list_head node;
72};
73
74struct notification_client_list {
75 struct lttng_trigger *trigger;
76 struct cds_list_head list;
77 struct cds_lfht_node notification_trigger_ht_node;
78};
79
80struct notification_client {
81 int socket;
82 /* Client protocol version. */
83 uint8_t major, minor;
84 uid_t uid;
85 gid_t gid;
86 /*
87 * Indicates if the credentials and versions of the client has been
88 * checked.
89 */
90 bool validated;
91 /*
92 * Conditions to which the client's notification channel is subscribed.
93 * List of struct lttng_condition_list_node. The condition member is
94 * owned by the client.
95 */
96 struct cds_list_head condition_list;
97 struct cds_lfht_node client_socket_ht_node;
98 struct {
99 struct {
100 struct lttng_dynamic_buffer buffer;
101 /* Bytes left to receive for the current message. */
102 size_t bytes_to_receive;
103 /* Type of the message being received. */
104 enum lttng_notification_channel_message_type msg_type;
105 /*
106 * Indicates whether or not credentials are expected
107 * from the client.
108 */
109 bool receive_creds;
110 /*
111 * Indicates whether or not credentials were received
112 * from the client.
113 */
114 bool creds_received;
115 lttng_sock_cred creds;
116 } inbound;
117 struct {
118 /*
119 * Indicates whether or not a notification addressed to
120 * this client was dropped because a command reply was
121 * already buffered.
122 *
123 * A notification is dropped whenever the buffer is not
124 * empty.
125 */
126 bool dropped_notification;
127 /*
128 * Indicates whether or not a command reply is already
129 * buffered. In this case, it means that the client is
130 * not consuming command replies before emitting a new
131 * one. This could be caused by a protocol error or a
132 * misbehaving/malicious client.
133 */
134 bool queued_command_reply;
135 struct lttng_dynamic_buffer buffer;
136 } outbound;
137 } communication;
138};
139
140struct channel_state_sample {
141 struct channel_key key;
142 struct cds_lfht_node channel_state_ht_node;
143 uint64_t highest_usage;
144 uint64_t lowest_usage;
145};
146
147static
148int match_client(struct cds_lfht_node *node, const void *key)
149{
150 /* This double-cast is intended to supress pointer-to-cast warning. */
151 int socket = (int) (intptr_t) key;
152 struct notification_client *client;
153
154 client = caa_container_of(node, struct notification_client,
155 client_socket_ht_node);
156
157 return !!(client->socket == socket);
158}
159
160static
161int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
162{
163 struct channel_key *channel_key = (struct channel_key *) key;
164 struct lttng_channel_trigger_list *trigger_list;
165
166 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
167 channel_triggers_ht_node);
168
169 return !!((channel_key->key == trigger_list->channel_key.key) &&
170 (channel_key->domain == trigger_list->channel_key.domain));
171}
172
173static
174int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
175{
176 struct channel_key *channel_key = (struct channel_key *) key;
177 struct channel_state_sample *sample;
178
179 sample = caa_container_of(node, struct channel_state_sample,
180 channel_state_ht_node);
181
182 return !!((channel_key->key == sample->key.key) &&
183 (channel_key->domain == sample->key.domain));
184}
185
186static
187int match_channel_info(struct cds_lfht_node *node, const void *key)
188{
189 struct channel_key *channel_key = (struct channel_key *) key;
190 struct channel_info *channel_info;
191
192 channel_info = caa_container_of(node, struct channel_info,
193 channels_ht_node);
194
195 return !!((channel_key->key == channel_info->key.key) &&
196 (channel_key->domain == channel_info->key.domain));
197}
198
199static
200int match_condition(struct cds_lfht_node *node, const void *key)
201{
202 struct lttng_condition *condition_key = (struct lttng_condition *) key;
203 struct lttng_trigger_ht_element *trigger;
204 struct lttng_condition *condition;
205
206 trigger = caa_container_of(node, struct lttng_trigger_ht_element,
207 node);
208 condition = lttng_trigger_get_condition(trigger->trigger);
209 assert(condition);
210
211 return !!lttng_condition_is_equal(condition_key, condition);
212}
213
214static
215int match_client_list(struct cds_lfht_node *node, const void *key)
216{
217 struct lttng_trigger *trigger_key = (struct lttng_trigger *) key;
218 struct notification_client_list *client_list;
219 struct lttng_condition *condition;
220 struct lttng_condition *condition_key = lttng_trigger_get_condition(
221 trigger_key);
222
223 assert(condition_key);
224
225 client_list = caa_container_of(node, struct notification_client_list,
226 notification_trigger_ht_node);
227 condition = lttng_trigger_get_condition(client_list->trigger);
228
229 return !!lttng_condition_is_equal(condition_key, condition);
230}
231
232static
233int match_client_list_condition(struct cds_lfht_node *node, const void *key)
234{
235 struct lttng_condition *condition_key = (struct lttng_condition *) key;
236 struct notification_client_list *client_list;
237 struct lttng_condition *condition;
238
239 assert(condition_key);
240
241 client_list = caa_container_of(node, struct notification_client_list,
242 notification_trigger_ht_node);
243 condition = lttng_trigger_get_condition(client_list->trigger);
244
245 return !!lttng_condition_is_equal(condition_key, condition);
246}
247
248static
249unsigned long lttng_condition_buffer_usage_hash(
250 struct lttng_condition *_condition)
251{
252 unsigned long hash = 0;
253 struct lttng_condition_buffer_usage *condition;
254
255 condition = container_of(_condition,
256 struct lttng_condition_buffer_usage, parent);
257
258 if (condition->session_name) {
259 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
260 }
261 if (condition->channel_name) {
8f56701f 262 hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
ab0ee2ca
JG
263 }
264 if (condition->domain.set) {
265 hash ^= hash_key_ulong(
266 (void *) condition->domain.type,
267 lttng_ht_seed);
268 }
269 if (condition->threshold_ratio.set) {
270 uint64_t val;
271
272 val = condition->threshold_ratio.value * (double) UINT32_MAX;
273 hash ^= hash_key_u64(&val, lttng_ht_seed);
274 } else if (condition->threshold_ratio.set) {
275 uint64_t val;
276
277 val = condition->threshold_bytes.value;
278 hash ^= hash_key_u64(&val, lttng_ht_seed);
279 }
280 return hash;
281}
282
283/*
284 * The lttng_condition hashing code is kept in this file (rather than
285 * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
286 * don't want to link in liblttng-ctl.
287 */
288static
289unsigned long lttng_condition_hash(struct lttng_condition *condition)
290{
291 switch (condition->type) {
292 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
293 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
294 return lttng_condition_buffer_usage_hash(condition);
295 default:
296 ERR("[notification-thread] Unexpected condition type caught");
297 abort();
298 }
299}
300
301static
302void channel_info_destroy(struct channel_info *channel_info)
303{
304 if (!channel_info) {
305 return;
306 }
307
308 if (channel_info->session_name) {
309 free(channel_info->session_name);
310 }
311 if (channel_info->channel_name) {
312 free(channel_info->channel_name);
313 }
314 free(channel_info);
315}
316
317static
318struct channel_info *channel_info_copy(struct channel_info *channel_info)
319{
320 struct channel_info *copy = zmalloc(sizeof(*channel_info));
321
322 assert(channel_info);
323 assert(channel_info->session_name);
324 assert(channel_info->channel_name);
325
326 if (!copy) {
327 goto end;
328 }
329
330 memcpy(copy, channel_info, sizeof(*channel_info));
331 copy->session_name = NULL;
332 copy->channel_name = NULL;
333
334 copy->session_name = strdup(channel_info->session_name);
335 if (!copy->session_name) {
336 goto error;
337 }
338 copy->channel_name = strdup(channel_info->channel_name);
339 if (!copy->channel_name) {
340 goto error;
341 }
342 cds_lfht_node_init(&channel_info->channels_ht_node);
343end:
344 return copy;
345error:
346 channel_info_destroy(copy);
347 return NULL;
348}
349
350static
351int notification_thread_client_subscribe(struct notification_client *client,
352 struct lttng_condition *condition,
353 struct notification_thread_state *state,
354 enum lttng_notification_channel_status *_status)
355{
356 int ret = 0;
357 struct cds_lfht_iter iter;
358 struct cds_lfht_node *node;
359 struct notification_client_list *client_list;
360 struct lttng_condition_list_element *condition_list_element = NULL;
361 struct notification_client_list_element *client_list_element = NULL;
362 enum lttng_notification_channel_status status =
363 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
364
365 /*
366 * Ensure that the client has not already subscribed to this condition
367 * before.
368 */
369 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
370 if (lttng_condition_is_equal(condition_list_element->condition,
371 condition)) {
372 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
373 goto end;
374 }
375 }
376
377 condition_list_element = zmalloc(sizeof(*condition_list_element));
378 if (!condition_list_element) {
379 ret = -1;
380 goto error;
381 }
382 client_list_element = zmalloc(sizeof(*client_list_element));
383 if (!client_list_element) {
384 ret = -1;
385 goto error;
386 }
387
388 rcu_read_lock();
389
390 /*
391 * Add the newly-subscribed condition to the client's subscription list.
392 */
393 CDS_INIT_LIST_HEAD(&condition_list_element->node);
394 condition_list_element->condition = condition;
395 cds_list_add(&condition_list_element->node, &client->condition_list);
396
397 /*
398 * Add the client to the list of clients interested in a given trigger
399 * if a "notification" trigger with a corresponding condition was
400 * added prior.
401 */
402 cds_lfht_lookup(state->notification_trigger_clients_ht,
403 lttng_condition_hash(condition),
404 match_client_list_condition,
405 condition,
406 &iter);
407 node = cds_lfht_iter_get_node(&iter);
408 if (!node) {
4fb43b68 409 free(client_list_element);
ab0ee2ca
JG
410 goto end_unlock;
411 }
412
413 client_list = caa_container_of(node, struct notification_client_list,
414 notification_trigger_ht_node);
415 client_list_element->client = client;
416 CDS_INIT_LIST_HEAD(&client_list_element->node);
417 cds_list_add(&client_list_element->node, &client_list->list);
418end_unlock:
419 rcu_read_unlock();
420end:
421 if (_status) {
422 *_status = status;
423 }
424 return ret;
425error:
426 free(condition_list_element);
427 free(client_list_element);
428 return ret;
429}
430
431static
432int notification_thread_client_unsubscribe(
433 struct notification_client *client,
434 struct lttng_condition *condition,
435 struct notification_thread_state *state,
436 enum lttng_notification_channel_status *_status)
437{
438 struct cds_lfht_iter iter;
439 struct cds_lfht_node *node;
440 struct notification_client_list *client_list;
441 struct lttng_condition_list_element *condition_list_element,
442 *condition_tmp;
443 struct notification_client_list_element *client_list_element,
444 *client_tmp;
445 bool condition_found = false;
446 enum lttng_notification_channel_status status =
447 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
448
449 /* Remove the condition from the client's condition list. */
450 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
451 &client->condition_list, node) {
452 if (!lttng_condition_is_equal(condition_list_element->condition,
453 condition)) {
454 continue;
455 }
456
457 cds_list_del(&condition_list_element->node);
458 /*
459 * The caller may be iterating on the client's conditions to
460 * tear down a client's connection. In this case, the condition
461 * will be destroyed at the end.
462 */
463 if (condition != condition_list_element->condition) {
464 lttng_condition_destroy(
465 condition_list_element->condition);
466 }
467 free(condition_list_element);
468 condition_found = true;
469 break;
470 }
471
472 if (!condition_found) {
473 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
474 goto end;
475 }
476
477 /*
478 * Remove the client from the list of clients interested the trigger
479 * matching the condition.
480 */
481 rcu_read_lock();
482 cds_lfht_lookup(state->notification_trigger_clients_ht,
483 lttng_condition_hash(condition),
484 match_client_list_condition,
485 condition,
486 &iter);
487 node = cds_lfht_iter_get_node(&iter);
488 if (!node) {
489 goto end_unlock;
490 }
491
492 client_list = caa_container_of(node, struct notification_client_list,
493 notification_trigger_ht_node);
494 cds_list_for_each_entry_safe(client_list_element, client_tmp,
495 &client_list->list, node) {
496 if (client_list_element->client->socket != client->socket) {
497 continue;
498 }
499 cds_list_del(&client_list_element->node);
500 free(client_list_element);
501 break;
502 }
503end_unlock:
504 rcu_read_unlock();
505end:
506 lttng_condition_destroy(condition);
507 if (_status) {
508 *_status = status;
509 }
510 return 0;
511}
512
513static
514void notification_client_destroy(struct notification_client *client,
515 struct notification_thread_state *state)
516{
517 struct lttng_condition_list_element *condition_list_element, *tmp;
518
519 if (!client) {
520 return;
521 }
522
523 /* Release all conditions to which the client was subscribed. */
524 cds_list_for_each_entry_safe(condition_list_element, tmp,
525 &client->condition_list, node) {
526 (void) notification_thread_client_unsubscribe(client,
527 condition_list_element->condition, state, NULL);
528 }
529
530 if (client->socket >= 0) {
531 (void) lttcomm_close_unix_sock(client->socket);
532 }
533 lttng_dynamic_buffer_reset(&client->communication.inbound.buffer);
534 lttng_dynamic_buffer_reset(&client->communication.outbound.buffer);
535 free(client);
536}
537
538/*
539 * Call with rcu_read_lock held (and hold for the lifetime of the returned
540 * client pointer).
541 */
542static
543struct notification_client *get_client_from_socket(int socket,
544 struct notification_thread_state *state)
545{
546 struct cds_lfht_iter iter;
547 struct cds_lfht_node *node;
548 struct notification_client *client = NULL;
549
550 cds_lfht_lookup(state->client_socket_ht,
551 hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed),
552 match_client,
553 (void *) (unsigned long) socket,
554 &iter);
555 node = cds_lfht_iter_get_node(&iter);
556 if (!node) {
557 goto end;
558 }
559
560 client = caa_container_of(node, struct notification_client,
561 client_socket_ht_node);
562end:
563 return client;
564}
565
566static
567bool trigger_applies_to_channel(struct lttng_trigger *trigger,
568 struct channel_info *info)
569{
570 enum lttng_condition_status status;
571 struct lttng_condition *condition;
572 const char *trigger_session_name = NULL;
573 const char *trigger_channel_name = NULL;
574 enum lttng_domain_type trigger_domain;
575
576 condition = lttng_trigger_get_condition(trigger);
577 if (!condition) {
578 goto fail;
579 }
580
581 switch (lttng_condition_get_type(condition)) {
582 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
583 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
584 break;
585 default:
586 goto fail;
587 }
588
589 status = lttng_condition_buffer_usage_get_domain_type(condition,
590 &trigger_domain);
591 assert(status == LTTNG_CONDITION_STATUS_OK);
592 if (info->key.domain != trigger_domain) {
593 goto fail;
594 }
595
596 status = lttng_condition_buffer_usage_get_session_name(
597 condition, &trigger_session_name);
598 assert((status == LTTNG_CONDITION_STATUS_OK) && trigger_session_name);
599
600 status = lttng_condition_buffer_usage_get_channel_name(
601 condition, &trigger_channel_name);
602 assert((status == LTTNG_CONDITION_STATUS_OK) && trigger_channel_name);
603
604 if (strcmp(info->session_name, trigger_session_name)) {
605 goto fail;
606 }
607 if (strcmp(info->channel_name, trigger_channel_name)) {
608 goto fail;
609 }
610
611 return true;
612fail:
613 return false;
614}
615
616static
617bool trigger_applies_to_client(struct lttng_trigger *trigger,
618 struct notification_client *client)
619{
620 bool applies = false;
621 struct lttng_condition_list_element *condition_list_element;
622
623 cds_list_for_each_entry(condition_list_element, &client->condition_list,
624 node) {
625 applies = lttng_condition_is_equal(
626 condition_list_element->condition,
627 lttng_trigger_get_condition(trigger));
628 if (applies) {
629 break;
630 }
631 }
632 return applies;
633}
634
635static
636unsigned long hash_channel_key(struct channel_key *key)
637{
638 return hash_key_u64(&key->key, lttng_ht_seed) ^ hash_key_ulong(
639 (void *) (unsigned long) key->domain, lttng_ht_seed);
640}
641
642static
643int handle_notification_thread_command_add_channel(
644 struct notification_thread_state *state,
645 struct channel_info *channel_info,
646 enum lttng_error_code *cmd_result)
647{
648 struct cds_list_head trigger_list;
649 struct channel_info *new_channel_info;
650 struct channel_key *channel_key;
651 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
652 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
653 int trigger_count = 0;
654 struct cds_lfht_iter iter;
655
656 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
657 channel_info->channel_name, channel_info->session_name,
658 channel_info->key.key, channel_info->key.domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
659
660 CDS_INIT_LIST_HEAD(&trigger_list);
661
662 new_channel_info = channel_info_copy(channel_info);
663 if (!new_channel_info) {
664 goto error;
665 }
666
667 channel_key = &new_channel_info->key;
668
669 /* Build a list of all triggers applying to the new channel. */
670 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
671 node) {
672 struct lttng_trigger_list_element *new_element;
673
674 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
675 channel_info)) {
676 continue;
677 }
678
679 new_element = zmalloc(sizeof(*new_element));
680 if (!new_element) {
681 goto error;
682 }
683 CDS_INIT_LIST_HEAD(&new_element->node);
684 new_element->trigger = trigger_ht_element->trigger;
685 cds_list_add(&new_element->node, &trigger_list);
686 trigger_count++;
687 }
688
689 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
690 trigger_count);
691 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
692 if (!channel_trigger_list) {
693 goto error;
694 }
695 channel_trigger_list->channel_key = *channel_key;
696 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
697 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
698 cds_list_splice(&trigger_list, &channel_trigger_list->list);
699
700 rcu_read_lock();
701 /* Add channel to the channel_ht which owns the channel_infos. */
702 cds_lfht_add(state->channels_ht,
703 hash_channel_key(channel_key),
704 &new_channel_info->channels_ht_node);
705 /*
706 * Add the list of triggers associated with this channel to the
707 * channel_triggers_ht.
708 */
709 cds_lfht_add(state->channel_triggers_ht,
710 hash_channel_key(channel_key),
711 &channel_trigger_list->channel_triggers_ht_node);
712 rcu_read_unlock();
713 *cmd_result = LTTNG_OK;
714 return 0;
715error:
716 /* Empty trigger list */
717 channel_info_destroy(new_channel_info);
718 return 1;
719}
720
721static
722int handle_notification_thread_command_remove_channel(
723 struct notification_thread_state *state,
724 uint64_t channel_key, enum lttng_domain_type domain,
725 enum lttng_error_code *cmd_result)
726{
727 struct cds_lfht_node *node;
728 struct cds_lfht_iter iter;
729 struct lttng_channel_trigger_list *trigger_list;
730 struct lttng_trigger_list_element *trigger_list_element, *tmp;
731 struct channel_key key = { .key = channel_key, .domain = domain };
732 struct channel_info *channel_info;
733
734 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
735 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
736
737 rcu_read_lock();
738
739 cds_lfht_lookup(state->channel_triggers_ht,
740 hash_channel_key(&key),
741 match_channel_trigger_list,
742 &key,
743 &iter);
744 node = cds_lfht_iter_get_node(&iter);
745 /*
746 * There is a severe internal error if we are being asked to remove a
747 * channel that doesn't exist.
748 */
749 if (!node) {
750 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
751 goto end;
752 }
753
754 /* Free the list of triggers associated with this channel. */
755 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
756 channel_triggers_ht_node);
757 cds_list_for_each_entry_safe(trigger_list_element, tmp,
758 &trigger_list->list, node) {
759 cds_list_del(&trigger_list_element->node);
760 free(trigger_list_element);
761 }
762 cds_lfht_del(state->channel_triggers_ht, node);
763 free(trigger_list);
764
765 /* Free sampled channel state. */
766 cds_lfht_lookup(state->channel_state_ht,
767 hash_channel_key(&key),
768 match_channel_state_sample,
769 &key,
770 &iter);
771 node = cds_lfht_iter_get_node(&iter);
772 /*
773 * This is expected to be NULL if the channel is destroyed before we
774 * received a sample.
775 */
776 if (node) {
777 struct channel_state_sample *sample = caa_container_of(node,
778 struct channel_state_sample,
779 channel_state_ht_node);
780
781 cds_lfht_del(state->channel_state_ht, node);
782 free(sample);
783 }
784
785 /* Remove the channel from the channels_ht and free it. */
786 cds_lfht_lookup(state->channels_ht,
787 hash_channel_key(&key),
788 match_channel_info,
789 &key,
790 &iter);
791 node = cds_lfht_iter_get_node(&iter);
792 assert(node);
793 channel_info = caa_container_of(node, struct channel_info,
794 channels_ht_node);
795 cds_lfht_del(state->channels_ht, node);
796 channel_info_destroy(channel_info);
797end:
798 rcu_read_unlock();
799 *cmd_result = LTTNG_OK;
800 return 0;
801}
802
803/*
804 * FIXME A client's credentials are not checked when registering a trigger, nor
805 * are they stored alongside with the trigger.
806 *
807 * The effects of this are benign:
808 * - The client will succeed in registering the trigger, as it is valid,
809 * - The trigger will, internally, be bound to the channel,
810 * - The notifications will not be sent since the client's credentials
811 * are checked against the channel at that moment.
812 */
813static
814int handle_notification_thread_command_register_trigger(
815 struct notification_thread_state *state,
816 struct lttng_trigger *trigger,
817 enum lttng_error_code *cmd_result)
818{
819 int ret = 0;
820 struct lttng_condition *condition;
821 struct notification_client *client;
822 struct notification_client_list *client_list = NULL;
823 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
824 struct notification_client_list_element *client_list_element, *tmp;
825 struct cds_lfht_node *node;
826 struct cds_lfht_iter iter;
827 struct channel_info *channel;
828 bool free_trigger = true;
829
830 rcu_read_lock();
831
832 condition = lttng_trigger_get_condition(trigger);
833 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
834 if (!trigger_ht_element) {
835 ret = -1;
836 goto error;
837 }
838
839 /* Add trigger to the trigger_ht. */
840 cds_lfht_node_init(&trigger_ht_element->node);
841 trigger_ht_element->trigger = trigger;
842
843 node = cds_lfht_add_unique(state->triggers_ht,
844 lttng_condition_hash(condition),
845 match_condition,
846 condition,
847 &trigger_ht_element->node);
848 if (node != &trigger_ht_element->node) {
849 /* Not a fatal error, simply report it to the client. */
850 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
851 goto error_free_ht_element;
852 }
853
854 /*
855 * Ownership of the trigger and of its wrapper was transfered to
856 * the triggers_ht.
857 */
858 trigger_ht_element = NULL;
859 free_trigger = false;
860
861 /*
862 * The rest only applies to triggers that have a "notify" action.
863 * It is not skipped as this is the only action type currently
864 * supported.
865 */
866 client_list = zmalloc(sizeof(*client_list));
867 if (!client_list) {
868 ret = -1;
869 goto error_free_ht_element;
870 }
871 cds_lfht_node_init(&client_list->notification_trigger_ht_node);
872 CDS_INIT_LIST_HEAD(&client_list->list);
873 client_list->trigger = trigger;
874
875 /* Build a list of clients to which this new trigger applies. */
876 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
877 client_socket_ht_node) {
878 if (!trigger_applies_to_client(trigger, client)) {
879 continue;
880 }
881
882 client_list_element = zmalloc(sizeof(*client_list_element));
883 if (!client_list_element) {
884 ret = -1;
885 goto error_free_client_list;
886 }
887 CDS_INIT_LIST_HEAD(&client_list_element->node);
888 client_list_element->client = client;
889 cds_list_add(&client_list_element->node, &client_list->list);
890 }
891
892 cds_lfht_add(state->notification_trigger_clients_ht,
893 lttng_condition_hash(condition),
894 &client_list->notification_trigger_ht_node);
895 /*
896 * Client list ownership transferred to the
897 * notification_trigger_clients_ht.
898 */
899 client_list = NULL;
900
901 /*
902 * Add the trigger to list of triggers bound to the channels currently
903 * known.
904 */
905 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
906 channels_ht_node) {
907 struct lttng_trigger_list_element *trigger_list_element;
908 struct lttng_channel_trigger_list *trigger_list;
909
910 if (!trigger_applies_to_channel(trigger, channel)) {
911 continue;
912 }
913
914 cds_lfht_lookup(state->channel_triggers_ht,
915 hash_channel_key(&channel->key),
916 match_channel_trigger_list,
917 &channel->key,
918 &iter);
919 node = cds_lfht_iter_get_node(&iter);
920 assert(node);
921 /* Free the list of triggers associated with this channel. */
922 trigger_list = caa_container_of(node,
923 struct lttng_channel_trigger_list,
924 channel_triggers_ht_node);
925
926 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
927 if (!trigger_list_element) {
928 ret = -1;
929 goto error_free_client_list;
930 }
931 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
932 trigger_list_element->trigger = trigger;
933 cds_list_add(&trigger_list_element->node, &trigger_list->list);
934 /* A trigger can only apply to one channel. */
935 break;
936 }
937
938 *cmd_result = LTTNG_OK;
939error_free_client_list:
940 if (client_list) {
941 cds_list_for_each_entry_safe(client_list_element, tmp,
942 &client_list->list, node) {
943 free(client_list_element);
944 }
945 free(client_list);
946 }
947error_free_ht_element:
948 free(trigger_ht_element);
949error:
950 if (free_trigger) {
951 struct lttng_action *action = lttng_trigger_get_action(trigger);
952
953 lttng_condition_destroy(condition);
954 lttng_action_destroy(action);
955 lttng_trigger_destroy(trigger);
956 }
957 rcu_read_unlock();
958 return ret;
959}
960
cc2295b5 961static
ab0ee2ca
JG
962int handle_notification_thread_command_unregister_trigger(
963 struct notification_thread_state *state,
964 struct lttng_trigger *trigger,
965 enum lttng_error_code *_cmd_reply)
966{
967 struct cds_lfht_iter iter;
968 struct cds_lfht_node *node, *triggers_ht_node;
969 struct lttng_channel_trigger_list *trigger_list;
970 struct notification_client_list *client_list;
971 struct notification_client_list_element *client_list_element, *tmp;
972 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
973 struct lttng_condition *condition = lttng_trigger_get_condition(
974 trigger);
975 struct lttng_action *action;
976 enum lttng_error_code cmd_reply;
977
978 rcu_read_lock();
979
980 cds_lfht_lookup(state->triggers_ht,
981 lttng_condition_hash(condition),
982 match_condition,
983 condition,
984 &iter);
985 triggers_ht_node = cds_lfht_iter_get_node(&iter);
986 if (!triggers_ht_node) {
987 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
988 goto end;
989 } else {
990 cmd_reply = LTTNG_OK;
991 }
992
993 /* Remove trigger from channel_triggers_ht. */
994 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
995 channel_triggers_ht_node) {
996 struct lttng_trigger_list_element *trigger_element, *tmp;
997
998 cds_list_for_each_entry_safe(trigger_element, tmp,
999 &trigger_list->list, node) {
1000 struct lttng_condition *current_condition =
1001 lttng_trigger_get_condition(
1002 trigger_element->trigger);
1003
1004 assert(current_condition);
1005 if (!lttng_condition_is_equal(condition,
1006 current_condition)) {
1007 continue;
1008 }
1009
1010 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
1011 cds_list_del(&trigger_element->node);
1012 }
1013 }
1014
1015 /*
1016 * Remove and release the client list from
1017 * notification_trigger_clients_ht.
1018 */
1019 cds_lfht_lookup(state->notification_trigger_clients_ht,
1020 lttng_condition_hash(condition),
1021 match_client_list,
1022 trigger,
1023 &iter);
1024 node = cds_lfht_iter_get_node(&iter);
1025 assert(node);
1026 client_list = caa_container_of(node, struct notification_client_list,
1027 notification_trigger_ht_node);
1028 cds_list_for_each_entry_safe(client_list_element, tmp,
1029 &client_list->list, node) {
1030 free(client_list_element);
1031 }
1032 cds_lfht_del(state->notification_trigger_clients_ht, node);
1033 free(client_list);
1034
1035 /* Remove trigger from triggers_ht. */
1036 trigger_ht_element = caa_container_of(triggers_ht_node,
1037 struct lttng_trigger_ht_element, node);
1038 cds_lfht_del(state->triggers_ht, triggers_ht_node);
1039
1040 condition = lttng_trigger_get_condition(trigger_ht_element->trigger);
1041 lttng_condition_destroy(condition);
1042 action = lttng_trigger_get_action(trigger_ht_element->trigger);
1043 lttng_action_destroy(action);
1044 lttng_trigger_destroy(trigger_ht_element->trigger);
1045 free(trigger_ht_element);
1046end:
1047 rcu_read_unlock();
1048 if (_cmd_reply) {
1049 *_cmd_reply = cmd_reply;
1050 }
1051 return 0;
1052}
1053
1054/* Returns 0 on success, 1 on exit requested, negative value on error. */
1055int handle_notification_thread_command(
1056 struct notification_thread_handle *handle,
1057 struct notification_thread_state *state)
1058{
1059 int ret;
1060 uint64_t counter;
1061 struct notification_thread_command *cmd;
1062
1063 /* Read event_fd to put it back into a quiescent state. */
1064 ret = read(handle->cmd_queue.event_fd, &counter, sizeof(counter));
1065 if (ret == -1) {
1066 goto error;
1067 }
1068
1069 pthread_mutex_lock(&handle->cmd_queue.lock);
1070 cmd = cds_list_first_entry(&handle->cmd_queue.list,
1071 struct notification_thread_command, cmd_list_node);
1072 switch (cmd->type) {
1073 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
1074 DBG("[notification-thread] Received register trigger command");
1075 ret = handle_notification_thread_command_register_trigger(
1076 state, cmd->parameters.trigger,
1077 &cmd->reply_code);
1078 break;
1079 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
1080 DBG("[notification-thread] Received unregister trigger command");
1081 ret = handle_notification_thread_command_unregister_trigger(
1082 state, cmd->parameters.trigger,
1083 &cmd->reply_code);
1084 break;
1085 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
1086 DBG("[notification-thread] Received add channel command");
1087 ret = handle_notification_thread_command_add_channel(
1088 state, &cmd->parameters.add_channel,
1089 &cmd->reply_code);
1090 break;
1091 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
1092 DBG("[notification-thread] Received remove channel command");
1093 ret = handle_notification_thread_command_remove_channel(
1094 state, cmd->parameters.remove_channel.key,
1095 cmd->parameters.remove_channel.domain,
1096 &cmd->reply_code);
1097 break;
1098 case NOTIFICATION_COMMAND_TYPE_QUIT:
1099 DBG("[notification-thread] Received quit command");
1100 cmd->reply_code = LTTNG_OK;
1101 ret = 1;
1102 goto end;
1103 default:
1104 ERR("[notification-thread] Unknown internal command received");
1105 goto error_unlock;
1106 }
1107
1108 if (ret) {
1109 goto error_unlock;
1110 }
1111end:
1112 cds_list_del(&cmd->cmd_list_node);
8ada111f 1113 lttng_waiter_wake_up(&cmd->reply_waiter);
ab0ee2ca
JG
1114 pthread_mutex_unlock(&handle->cmd_queue.lock);
1115 return ret;
1116error_unlock:
1117 /* Wake-up and return a fatal error to the calling thread. */
8ada111f 1118 lttng_waiter_wake_up(&cmd->reply_waiter);
ab0ee2ca
JG
1119 pthread_mutex_unlock(&handle->cmd_queue.lock);
1120 cmd->reply_code = LTTNG_ERR_FATAL;
1121error:
1122 /* Indicate a fatal error to the caller. */
1123 return -1;
1124}
1125
1126static
1127unsigned long hash_client_socket(int socket)
1128{
1129 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
1130}
1131
1132static
1133int socket_set_non_blocking(int socket)
1134{
1135 int ret, flags;
1136
1137 /* Set the pipe as non-blocking. */
1138 ret = fcntl(socket, F_GETFL, 0);
1139 if (ret == -1) {
1140 PERROR("fcntl get socket flags");
1141 goto end;
1142 }
1143 flags = ret;
1144
1145 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
1146 if (ret == -1) {
1147 PERROR("fcntl set O_NONBLOCK socket flag");
1148 goto end;
1149 }
1150 DBG("Client socket (fd = %i) set as non-blocking", socket);
1151end:
1152 return ret;
1153}
1154
1155static
1156void client_reset_inbound_state(struct notification_client *client)
1157{
1158 int ret;
1159
1160 ret = lttng_dynamic_buffer_set_size(
1161 &client->communication.inbound.buffer, 0);
1162 assert(!ret);
1163
1164 client->communication.inbound.bytes_to_receive =
1165 sizeof(struct lttng_notification_channel_message);
1166 client->communication.inbound.msg_type =
1167 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
1168 client->communication.inbound.receive_creds = false;
1169 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
1170 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
1171}
1172
1173int handle_notification_thread_client_connect(
1174 struct notification_thread_state *state)
1175{
1176 int ret;
1177 struct notification_client *client;
1178
1179 DBG("[notification-thread] Handling new notification channel client connection");
1180
1181 client = zmalloc(sizeof(*client));
1182 if (!client) {
1183 /* Fatal error. */
1184 ret = -1;
1185 goto error;
1186 }
1187 CDS_INIT_LIST_HEAD(&client->condition_list);
1188 lttng_dynamic_buffer_init(&client->communication.inbound.buffer);
1189 lttng_dynamic_buffer_init(&client->communication.outbound.buffer);
1190 client_reset_inbound_state(client);
1191
1192 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
1193 if (ret < 0) {
1194 ERR("[notification-thread] Failed to accept new notification channel client connection");
1195 ret = 0;
1196 goto error;
1197 }
1198
1199 client->socket = ret;
1200
1201 ret = socket_set_non_blocking(client->socket);
1202 if (ret) {
1203 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
1204 goto error;
1205 }
1206
1207 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
1208 if (ret < 0) {
1209 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
1210 ret = 0;
1211 goto error;
1212 }
1213
1214 ret = lttng_poll_add(&state->events, client->socket,
1215 LPOLLIN | LPOLLERR |
1216 LPOLLHUP | LPOLLRDHUP);
1217 if (ret < 0) {
1218 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
1219 ret = 0;
1220 goto error;
1221 }
1222 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
1223 client->socket);
1224
1225 /* Add to ht. */
1226 rcu_read_lock();
1227 cds_lfht_add(state->client_socket_ht,
1228 hash_client_socket(client->socket),
1229 &client->client_socket_ht_node);
1230 rcu_read_unlock();
1231
1232 return ret;
1233error:
1234 notification_client_destroy(client, state);
1235 return ret;
1236}
1237
1238int handle_notification_thread_client_disconnect(
1239 int client_socket,
1240 struct notification_thread_state *state)
1241{
1242 int ret = 0;
1243 struct notification_client *client;
1244
1245 rcu_read_lock();
1246 DBG("[notification-thread] Closing client connection (socket fd = %i)",
1247 client_socket);
1248 client = get_client_from_socket(client_socket, state);
1249 if (!client) {
1250 /* Internal state corruption, fatal error. */
1251 ERR("[notification-thread] Unable to find client (socket fd = %i)",
1252 client_socket);
1253 ret = -1;
1254 goto end;
1255 }
1256
1257 ret = lttng_poll_del(&state->events, client_socket);
1258 if (ret) {
1259 ERR("[notification-thread] Failed to remove client socket from poll set");
1260 }
1261 cds_lfht_del(state->client_socket_ht,
1262 &client->client_socket_ht_node);
1263 notification_client_destroy(client, state);
1264end:
1265 rcu_read_unlock();
1266 return ret;
1267}
1268
1269int handle_notification_thread_client_disconnect_all(
1270 struct notification_thread_state *state)
1271{
1272 struct cds_lfht_iter iter;
1273 struct notification_client *client;
1274 bool error_encoutered = false;
1275
1276 rcu_read_lock();
1277 DBG("[notification-thread] Closing all client connections");
1278 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1279 client_socket_ht_node) {
1280 int ret;
1281
1282 ret = handle_notification_thread_client_disconnect(
1283 client->socket, state);
1284 if (ret) {
1285 error_encoutered = true;
1286 }
1287 }
1288 rcu_read_unlock();
1289 return error_encoutered ? 1 : 0;
1290}
1291
1292int handle_notification_thread_trigger_unregister_all(
1293 struct notification_thread_state *state)
1294{
1295 bool error_occured = false;
1296 struct cds_lfht_iter iter;
1297 struct lttng_trigger_ht_element *trigger_ht_element;
1298
1299 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1300 node) {
1301 int ret = handle_notification_thread_command_unregister_trigger(
1302 state, trigger_ht_element->trigger, NULL);
1303 if (ret) {
1304 error_occured = true;
1305 }
1306 }
1307 return error_occured ? -1 : 0;
1308}
1309
1310static
1311int client_flush_outgoing_queue(struct notification_client *client,
1312 struct notification_thread_state *state)
1313{
1314 ssize_t ret;
1315 size_t to_send_count;
1316
1317 assert(client->communication.outbound.buffer.size != 0);
1318 to_send_count = client->communication.outbound.buffer.size;
1319 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
1320 client->socket);
1321
1322 ret = lttcomm_send_unix_sock_non_block(client->socket,
1323 client->communication.outbound.buffer.data,
1324 to_send_count);
1325 if ((ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) ||
1326 (ret > 0 && ret < to_send_count)) {
1327 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
1328 client->socket);
1329 to_send_count -= max(ret, 0);
1330
1331 memcpy(client->communication.outbound.buffer.data,
1332 client->communication.outbound.buffer.data +
1333 client->communication.outbound.buffer.size - to_send_count,
1334 to_send_count);
1335 ret = lttng_dynamic_buffer_set_size(
1336 &client->communication.outbound.buffer,
1337 to_send_count);
1338 if (ret) {
1339 goto error;
1340 }
1341
1342 /*
1343 * We want to be notified whenever there is buffer space
1344 * available to send the rest of the payload.
1345 */
1346 ret = lttng_poll_mod(&state->events, client->socket,
1347 CLIENT_POLL_MASK_IN_OUT);
1348 if (ret) {
1349 goto error;
1350 }
1351 } else if (ret < 0) {
1352 /* Generic error, disconnect the client. */
1353 ERR("[notification-thread] Failed to send flush outgoing queue, disconnecting client (socket fd = %i)",
1354 client->socket);
1355 ret = handle_notification_thread_client_disconnect(
1356 client->socket, state);
1357 if (ret) {
1358 goto error;
1359 }
1360 } else {
1361 /* No error and flushed the queue completely. */
1362 ret = lttng_dynamic_buffer_set_size(
1363 &client->communication.outbound.buffer, 0);
1364 if (ret) {
1365 goto error;
1366 }
1367 ret = lttng_poll_mod(&state->events, client->socket,
1368 CLIENT_POLL_MASK_IN);
1369 if (ret) {
1370 goto error;
1371 }
1372
1373 client->communication.outbound.queued_command_reply = false;
1374 client->communication.outbound.dropped_notification = false;
1375 }
1376
1377 return 0;
1378error:
1379 return -1;
1380}
1381
1382static
1383int client_send_command_reply(struct notification_client *client,
1384 struct notification_thread_state *state,
1385 enum lttng_notification_channel_status status)
1386{
1387 int ret;
1388 struct lttng_notification_channel_command_reply reply = {
1389 .status = (int8_t) status,
1390 };
1391 struct lttng_notification_channel_message msg = {
1392 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
1393 .size = sizeof(reply),
1394 };
1395 char buffer[sizeof(msg) + sizeof(reply)];
1396
1397 if (client->communication.outbound.queued_command_reply) {
1398 /* Protocol error. */
1399 goto error;
1400 }
1401
1402 memcpy(buffer, &msg, sizeof(msg));
1403 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
1404 DBG("[notification-thread] Send command reply (%i)", (int) status);
1405
1406 /* Enqueue buffer to outgoing queue and flush it. */
1407 ret = lttng_dynamic_buffer_append(
1408 &client->communication.outbound.buffer,
1409 buffer, sizeof(buffer));
1410 if (ret) {
1411 goto error;
1412 }
1413
1414 ret = client_flush_outgoing_queue(client, state);
1415 if (ret) {
1416 goto error;
1417 }
1418
1419 if (client->communication.outbound.buffer.size != 0) {
1420 /* Queue could not be emptied. */
1421 client->communication.outbound.queued_command_reply = true;
1422 }
1423
1424 return 0;
1425error:
1426 return -1;
1427}
1428
1429static
1430int client_dispatch_message(struct notification_client *client,
1431 struct notification_thread_state *state)
1432{
1433 int ret = 0;
1434
1435 if (client->communication.inbound.msg_type !=
1436 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
1437 client->communication.inbound.msg_type !=
1438 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
1439 !client->validated) {
1440 WARN("[notification-thread] client attempted a command before handshake");
1441 ret = -1;
1442 goto end;
1443 }
1444
1445 switch (client->communication.inbound.msg_type) {
1446 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
1447 {
1448 /*
1449 * Receiving message header. The function will be called again
1450 * once the rest of the message as been received and can be
1451 * interpreted.
1452 */
1453 const struct lttng_notification_channel_message *msg;
1454
1455 assert(sizeof(*msg) ==
1456 client->communication.inbound.buffer.size);
1457 msg = (const struct lttng_notification_channel_message *)
1458 client->communication.inbound.buffer.data;
1459
1460 if (msg->size == 0 || msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
1461 ERR("[notification-thread] Invalid notification channel message: length = %u", msg->size);
1462 ret = -1;
1463 goto end;
1464 }
1465
1466 switch (msg->type) {
1467 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1468 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1469 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1470 break;
1471 default:
1472 ret = -1;
1473 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
1474 goto end;
1475 }
1476
1477 client->communication.inbound.bytes_to_receive = msg->size;
1478 client->communication.inbound.msg_type =
1479 (enum lttng_notification_channel_message_type) msg->type;
1480 if (client->communication.inbound.msg_type ==
1481 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE) {
1482 client->communication.inbound.receive_creds = true;
1483 }
1484 ret = lttng_dynamic_buffer_set_size(
1485 &client->communication.inbound.buffer, 0);
1486 if (ret) {
1487 goto end;
1488 }
1489 break;
1490 }
1491 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1492 {
1493 struct lttng_notification_channel_command_handshake *handshake_client;
1494 struct lttng_notification_channel_command_handshake handshake_reply = {
1495 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
1496 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
1497 };
1498 struct lttng_notification_channel_message msg_header = {
1499 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
1500 .size = sizeof(handshake_reply),
1501 };
1502 enum lttng_notification_channel_status status =
1503 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1504 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
1505
1506 memcpy(send_buffer, &msg_header, sizeof(msg_header));
1507 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
1508 sizeof(handshake_reply));
1509
1510 handshake_client =
1511 (struct lttng_notification_channel_command_handshake *)
1512 client->communication.inbound.buffer.data;
1513 client->major = handshake_client->major;
1514 client->minor = handshake_client->minor;
1515 if (!client->communication.inbound.creds_received) {
1516 ERR("[notification-thread] No credentials received from client");
1517 ret = -1;
1518 goto end;
1519 }
1520
1521 client->uid = LTTNG_SOCK_GET_UID_CRED(
1522 &client->communication.inbound.creds);
1523 client->gid = LTTNG_SOCK_GET_GID_CRED(
1524 &client->communication.inbound.creds);
1525 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
1526 client->uid, client->gid, (int) client->major,
1527 (int) client->minor);
1528
1529 if (handshake_client->major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
1530 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
1531 }
1532
1533 ret = lttng_dynamic_buffer_append(&client->communication.outbound.buffer,
1534 send_buffer, sizeof(send_buffer));
1535 if (ret) {
1536 ERR("[notification-thread] Failed to send protocol version to notification channel client");
1537 goto end;
1538 }
1539
1540 ret = client_flush_outgoing_queue(client, state);
1541 if (ret) {
1542 goto end;
1543 }
1544
1545 ret = client_send_command_reply(client, state, status);
1546 if (ret) {
1547 ERR("[notification-thread] Failed to send reply to notification channel client");
1548 goto end;
1549 }
1550
1551 /* Set reception state to receive the next message header. */
1552 client_reset_inbound_state(client);
1553 client->validated = true;
1554 break;
1555 }
1556 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1557 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1558 {
1559 struct lttng_condition *condition;
1560 enum lttng_notification_channel_status status =
1561 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1562 const struct lttng_buffer_view condition_view =
1563 lttng_buffer_view_from_dynamic_buffer(
1564 &client->communication.inbound.buffer,
1565 0, -1);
1566 size_t expected_condition_size =
1567 client->communication.inbound.buffer.size;
1568
1569 ret = lttng_condition_create_from_buffer(&condition_view,
1570 &condition);
1571 if (ret != expected_condition_size) {
1572 ERR("[notification-thread] Malformed condition received from client");
1573 goto end;
1574 }
1575
1576 if (client->communication.inbound.msg_type ==
1577 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
1578 /*
1579 * FIXME The current state should be evaluated on
1580 * subscription.
1581 */
1582 ret = notification_thread_client_subscribe(client,
1583 condition, state, &status);
1584 } else {
1585 ret = notification_thread_client_unsubscribe(client,
1586 condition, state, &status);
1587 }
1588 if (ret) {
1589 goto end;
1590 }
1591
1592 ret = client_send_command_reply(client, state, status);
1593 if (ret) {
1594 ERR("[notification-thread] Failed to send reply to notification channel client");
1595 goto end;
1596 }
1597
1598 /* Set reception state to receive the next message header. */
1599 client_reset_inbound_state(client);
1600 break;
1601 }
1602 default:
1603 abort();
1604 }
1605end:
1606 return ret;
1607}
1608
1609/* Incoming data from client. */
1610int handle_notification_thread_client_in(
1611 struct notification_thread_state *state, int socket)
1612{
1613 int ret;
1614 struct notification_client *client;
1615 ssize_t recv_ret;
1616 size_t offset;
1617
1618 client = get_client_from_socket(socket, state);
1619 if (!client) {
1620 /* Internal error, abort. */
1621 ret = -1;
1622 goto end;
1623 }
1624
1625 offset = client->communication.inbound.buffer.size;
1626 ret = lttng_dynamic_buffer_set_size(
1627 &client->communication.inbound.buffer,
1628 client->communication.inbound.bytes_to_receive);
1629 if (ret) {
1630 goto end;
1631 }
1632
1633 if (client->communication.inbound.receive_creds) {
1634 recv_ret = lttcomm_recv_creds_unix_sock(socket,
1635 client->communication.inbound.buffer.data + offset,
1636 client->communication.inbound.bytes_to_receive,
1637 &client->communication.inbound.creds);
1638 if (recv_ret > 0) {
1639 client->communication.inbound.receive_creds = false;
1640 client->communication.inbound.creds_received = true;
1641 }
1642 } else {
1643 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
1644 client->communication.inbound.buffer.data + offset,
1645 client->communication.inbound.bytes_to_receive);
1646 }
1647 if (recv_ret < 0) {
1648 goto error_disconnect_client;
1649 }
1650
1651 client->communication.inbound.bytes_to_receive -= recv_ret;
1652 ret = lttng_dynamic_buffer_set_size(
1653 &client->communication.inbound.buffer,
1654 client->communication.inbound.buffer.size -
1655 client->communication.inbound.bytes_to_receive);
1656 if (ret) {
1657 goto end;
1658 }
1659
1660 if (client->communication.inbound.bytes_to_receive == 0) {
1661 ret = client_dispatch_message(client, state);
1662 if (ret) {
1663 /*
1664 * Only returns an error if this client must be
1665 * disconnected.
1666 */
1667 goto error_disconnect_client;
1668 }
1669 } else {
1670 goto end;
1671 }
1672end:
1673 return ret;
1674error_disconnect_client:
1675 ret = handle_notification_thread_client_disconnect(socket, state);
1676 return ret;
1677}
1678
1679/* Client ready to receive outgoing data. */
1680int handle_notification_thread_client_out(
1681 struct notification_thread_state *state, int socket)
1682{
1683 int ret;
1684 struct notification_client *client;
1685
1686 client = get_client_from_socket(socket, state);
1687 if (!client) {
1688 /* Internal error, abort. */
1689 ret = -1;
1690 goto end;
1691 }
1692
1693 ret = client_flush_outgoing_queue(client, state);
1694 if (ret) {
1695 goto end;
1696 }
1697end:
1698 return ret;
1699}
1700
1701static
1702bool evaluate_buffer_usage_condition(struct lttng_condition *condition,
1703 struct channel_state_sample *sample, uint64_t buffer_capacity)
1704{
1705 bool result = false;
1706 uint64_t threshold;
1707 enum lttng_condition_type condition_type;
1708 struct lttng_condition_buffer_usage *use_condition = container_of(
1709 condition, struct lttng_condition_buffer_usage,
1710 parent);
1711
1712 if (!sample) {
1713 goto end;
1714 }
1715
1716 if (use_condition->threshold_bytes.set) {
1717 threshold = use_condition->threshold_bytes.value;
1718 } else {
1719 /*
1720 * Threshold was expressed as a ratio.
1721 *
1722 * TODO the threshold (in bytes) of conditions expressed
1723 * as a ratio of total buffer size could be cached to
1724 * forego this double-multiplication or it could be performed
1725 * as fixed-point math.
1726 *
1727 * Note that caching should accomodate the case where the
1728 * condition applies to multiple channels (i.e. don't assume
1729 * that all channels matching my_chann* have the same size...)
1730 */
1731 threshold = (uint64_t) (use_condition->threshold_ratio.value *
1732 (double) buffer_capacity);
1733 }
1734
1735 condition_type = lttng_condition_get_type(condition);
1736 if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
1737 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
1738 threshold, sample->highest_usage);
1739
1740 /*
1741 * The low condition should only be triggered once _all_ of the
1742 * streams in a channel have gone below the "low" threshold.
1743 */
1744 if (sample->highest_usage <= threshold) {
1745 result = true;
1746 }
1747 } else {
1748 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
1749 threshold, sample->highest_usage);
1750
1751 /*
1752 * For high buffer usage scenarios, we want to trigger whenever
1753 * _any_ of the streams has reached the "high" threshold.
1754 */
1755 if (sample->highest_usage >= threshold) {
1756 result = true;
1757 }
1758 }
1759end:
1760 return result;
1761}
1762
1763static
1764int evaluate_condition(struct lttng_condition *condition,
1765 struct lttng_evaluation **evaluation,
1766 struct notification_thread_state *state,
1767 struct channel_state_sample *previous_sample,
1768 struct channel_state_sample *latest_sample,
1769 uint64_t buffer_capacity)
1770{
1771 int ret = 0;
1772 enum lttng_condition_type condition_type;
1773 bool previous_sample_result;
1774 bool latest_sample_result;
1775
1776 condition_type = lttng_condition_get_type(condition);
1777 /* No other condition type supported for the moment. */
1778 assert(condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW ||
1779 condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH);
1780
1781 previous_sample_result = evaluate_buffer_usage_condition(condition,
1782 previous_sample, buffer_capacity);
1783 latest_sample_result = evaluate_buffer_usage_condition(condition,
1784 latest_sample, buffer_capacity);
1785
1786 if (!latest_sample_result ||
1787 (previous_sample_result == latest_sample_result)) {
1788 /*
1789 * Only trigger on a condition evaluation transition.
1790 *
1791 * NOTE: This edge-triggered logic may not be appropriate for
1792 * future condition types.
1793 */
1794 goto end;
1795 }
1796
1797 if (evaluation && latest_sample_result) {
1798 *evaluation = lttng_evaluation_buffer_usage_create(
1799 condition_type,
1800 latest_sample->highest_usage,
1801 buffer_capacity);
1802 if (!*evaluation) {
1803 ret = -1;
1804 goto end;
1805 }
1806 }
1807end:
1808 return ret;
1809}
1810
1811static
1812int client_enqueue_dropped_notification(struct notification_client *client,
1813 struct notification_thread_state *state)
1814{
1815 int ret;
1816 struct lttng_notification_channel_message msg = {
1817 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
1818 .size = 0,
1819 };
1820
1821 ret = lttng_dynamic_buffer_append(
1822 &client->communication.outbound.buffer, &msg,
1823 sizeof(msg));
1824 return ret;
1825}
1826
1827static
1828int send_evaluation_to_clients(struct lttng_trigger *trigger,
1829 struct lttng_evaluation *evaluation,
1830 struct notification_client_list* client_list,
1831 struct notification_thread_state *state,
1832 uid_t channel_uid, gid_t channel_gid)
1833{
1834 int ret = 0;
1835 struct lttng_dynamic_buffer msg_buffer;
1836 struct notification_client_list_element *client_list_element, *tmp;
1837 struct lttng_notification *notification;
1838 struct lttng_condition *condition;
1839 ssize_t expected_notification_size, notification_size;
1840 struct lttng_notification_channel_message msg;
1841
1842 lttng_dynamic_buffer_init(&msg_buffer);
1843
1844 condition = lttng_trigger_get_condition(trigger);
1845 assert(condition);
1846
1847 notification = lttng_notification_create(condition, evaluation);
1848 if (!notification) {
1849 ret = -1;
1850 goto end;
1851 }
1852
1853 expected_notification_size = lttng_notification_serialize(notification,
1854 NULL);
1855 if (expected_notification_size < 0) {
1856 ERR("[notification-thread] Failed to get size of serialized notification");
1857 ret = -1;
1858 goto end;
1859 }
1860
1861 msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION;
1862 msg.size = (uint32_t) expected_notification_size;
1863 ret = lttng_dynamic_buffer_append(&msg_buffer, &msg, sizeof(msg));
1864 if (ret) {
1865 goto end;
1866 }
1867
1868 ret = lttng_dynamic_buffer_set_size(&msg_buffer,
1869 msg_buffer.size + expected_notification_size);
1870 if (ret) {
1871 goto end;
1872 }
1873
1874 notification_size = lttng_notification_serialize(notification,
1875 msg_buffer.data + sizeof(msg));
1876 if (notification_size != expected_notification_size) {
1877 ERR("[notification-thread] Failed to serialize notification");
1878 ret = -1;
1879 goto end;
1880 }
1881
1882 cds_list_for_each_entry_safe(client_list_element, tmp,
1883 &client_list->list, node) {
1884 struct notification_client *client =
1885 client_list_element->client;
1886
1887 if (client->uid != channel_uid && client->gid != channel_gid &&
1888 client->uid != 0) {
1889 /* Client is not allowed to monitor this channel. */
1890 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this channel");
1891 continue;
1892 }
1893
1894 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
1895 client->socket, msg_buffer.size);
1896 if (client->communication.outbound.buffer.size) {
1897 /*
1898 * Outgoing data is already buffered for this client;
1899 * drop the notification and enqueue a "dropped
1900 * notification" message if this is the first dropped
1901 * notification since the socket spilled-over to the
1902 * queue.
1903 */
1904 DBG("[notification-thread] Dropping notification addressed to client (socket fd = %i)",
1905 client->socket);
1906 if (!client->communication.outbound.dropped_notification) {
1907 client->communication.outbound.dropped_notification = true;
1908 ret = client_enqueue_dropped_notification(
1909 client, state);
1910 if (ret) {
1911 goto end;
1912 }
1913 }
1914 continue;
1915 }
1916
1917 ret = lttng_dynamic_buffer_append_buffer(
1918 &client->communication.outbound.buffer,
1919 &msg_buffer);
1920 if (ret) {
1921 goto end;
1922 }
1923
1924 ret = client_flush_outgoing_queue(client, state);
1925 if (ret) {
1926 goto end;
1927 }
1928 }
1929 ret = 0;
1930end:
1931 lttng_notification_destroy(notification);
1932 lttng_dynamic_buffer_reset(&msg_buffer);
1933 return ret;
1934}
1935
1936int handle_notification_thread_channel_sample(
1937 struct notification_thread_state *state, int pipe,
1938 enum lttng_domain_type domain)
1939{
1940 int ret = 0;
1941 struct lttcomm_consumer_channel_monitor_msg sample_msg;
1942 struct channel_state_sample previous_sample, latest_sample;
1943 struct channel_info *channel_info;
1944 struct cds_lfht_node *node;
1945 struct cds_lfht_iter iter;
1946 struct lttng_channel_trigger_list *trigger_list;
1947 struct lttng_trigger_list_element *trigger_list_element;
1948 bool previous_sample_available = false;
1949
1950 /*
1951 * The monitoring pipe only holds messages smaller than PIPE_BUF,
1952 * ensuring that read/write of sampling messages are atomic.
1953 */
7c2551ef 1954 ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
ab0ee2ca
JG
1955 if (ret != sizeof(sample_msg)) {
1956 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
1957 pipe);
1958 ret = -1;
1959 goto end;
1960 }
1961
1962 ret = 0;
1963 latest_sample.key.key = sample_msg.key;
1964 latest_sample.key.domain = domain;
1965 latest_sample.highest_usage = sample_msg.highest;
1966 latest_sample.lowest_usage = sample_msg.lowest;
1967
1968 rcu_read_lock();
1969
1970 /* Retrieve the channel's informations */
1971 cds_lfht_lookup(state->channels_ht,
1972 hash_channel_key(&latest_sample.key),
1973 match_channel_info,
1974 &latest_sample.key,
1975 &iter);
1976 node = cds_lfht_iter_get_node(&iter);
1977 if (!node) {
1978 /*
1979 * Not an error since the consumer can push a sample to the pipe
1980 * and the rest of the session daemon could notify us of the
1981 * channel's destruction before we get a chance to process that
1982 * sample.
1983 */
1984 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
1985 latest_sample.key.key,
1986 domain == LTTNG_DOMAIN_KERNEL ? "kernel" :
1987 "user space");
1988 goto end_unlock;
1989 }
1990 channel_info = caa_container_of(node, struct channel_info,
1991 channels_ht_node);
1992 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64")",
1993 channel_info->channel_name,
1994 latest_sample.key.key,
1995 channel_info->session_name,
1996 latest_sample.highest_usage,
1997 latest_sample.lowest_usage);
1998
1999 /* Retrieve the channel's last sample, if it exists, and update it. */
2000 cds_lfht_lookup(state->channel_state_ht,
2001 hash_channel_key(&latest_sample.key),
2002 match_channel_state_sample,
2003 &latest_sample.key,
2004 &iter);
2005 node = cds_lfht_iter_get_node(&iter);
2006 if (node) {
2007 struct channel_state_sample *stored_sample;
2008
2009 /* Update the sample stored. */
2010 stored_sample = caa_container_of(node,
2011 struct channel_state_sample,
2012 channel_state_ht_node);
2013 memcpy(&previous_sample, stored_sample,
2014 sizeof(previous_sample));
2015 stored_sample->highest_usage = latest_sample.highest_usage;
2016 stored_sample->lowest_usage = latest_sample.lowest_usage;
2017 previous_sample_available = true;
2018 } else {
2019 /*
2020 * This is the channel's first sample, allocate space for and
2021 * store the new sample.
2022 */
2023 struct channel_state_sample *stored_sample;
2024
2025 stored_sample = zmalloc(sizeof(*stored_sample));
2026 if (!stored_sample) {
2027 ret = -1;
2028 goto end_unlock;
2029 }
2030
2031 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
2032 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
2033 cds_lfht_add(state->channel_state_ht,
2034 hash_channel_key(&stored_sample->key),
2035 &stored_sample->channel_state_ht_node);
2036 }
2037
2038 /* Find triggers associated with this channel. */
2039 cds_lfht_lookup(state->channel_triggers_ht,
2040 hash_channel_key(&latest_sample.key),
2041 match_channel_trigger_list,
2042 &latest_sample.key,
2043 &iter);
2044 node = cds_lfht_iter_get_node(&iter);
2045 if (!node) {
2046 goto end_unlock;
2047 }
2048
2049 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
2050 channel_triggers_ht_node);
2051 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
2052 node) {
2053 struct lttng_condition *condition;
2054 struct lttng_action *action;
2055 struct lttng_trigger *trigger;
2056 struct notification_client_list *client_list;
2057 struct lttng_evaluation *evaluation = NULL;
2058
2059 trigger = trigger_list_element->trigger;
2060 condition = lttng_trigger_get_condition(trigger);
2061 assert(condition);
2062 action = lttng_trigger_get_action(trigger);
2063
2064 /* Notify actions are the only type currently supported. */
2065 assert(lttng_action_get_type(action) ==
2066 LTTNG_ACTION_TYPE_NOTIFY);
2067
2068 /*
2069 * Check if any client is subscribed to the result of this
2070 * evaluation.
2071 */
2072 cds_lfht_lookup(state->notification_trigger_clients_ht,
2073 lttng_condition_hash(condition),
2074 match_client_list,
2075 trigger,
2076 &iter);
2077 node = cds_lfht_iter_get_node(&iter);
2078 assert(node);
2079
2080 client_list = caa_container_of(node,
2081 struct notification_client_list,
2082 notification_trigger_ht_node);
2083 if (cds_list_empty(&client_list->list)) {
2084 /*
2085 * No clients interested in the evaluation's result,
2086 * skip it.
2087 */
2088 continue;
2089 }
2090
2091 ret = evaluate_condition(condition, &evaluation, state,
2092 previous_sample_available ? &previous_sample : NULL,
2093 &latest_sample, channel_info->capacity);
2094 if (ret) {
2095 goto end_unlock;
2096 }
2097
2098 if (!evaluation) {
2099 continue;
2100 }
2101
2102 /* Dispatch evaluation result to all clients. */
2103 ret = send_evaluation_to_clients(trigger_list_element->trigger,
2104 evaluation, client_list, state,
2105 channel_info->uid, channel_info->gid);
2106 if (ret) {
2107 goto end_unlock;
2108 }
2109 }
2110end_unlock:
2111 rcu_read_unlock();
2112end:
2113 return ret;
2114}
This page took 0.170731 seconds and 4 git commands to generate.