Clean-up: remove useless comment
[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
ab0ee2ca
JG
1225 rcu_read_lock();
1226 cds_lfht_add(state->client_socket_ht,
1227 hash_client_socket(client->socket),
1228 &client->client_socket_ht_node);
1229 rcu_read_unlock();
1230
1231 return ret;
1232error:
1233 notification_client_destroy(client, state);
1234 return ret;
1235}
1236
1237int handle_notification_thread_client_disconnect(
1238 int client_socket,
1239 struct notification_thread_state *state)
1240{
1241 int ret = 0;
1242 struct notification_client *client;
1243
1244 rcu_read_lock();
1245 DBG("[notification-thread] Closing client connection (socket fd = %i)",
1246 client_socket);
1247 client = get_client_from_socket(client_socket, state);
1248 if (!client) {
1249 /* Internal state corruption, fatal error. */
1250 ERR("[notification-thread] Unable to find client (socket fd = %i)",
1251 client_socket);
1252 ret = -1;
1253 goto end;
1254 }
1255
1256 ret = lttng_poll_del(&state->events, client_socket);
1257 if (ret) {
1258 ERR("[notification-thread] Failed to remove client socket from poll set");
1259 }
1260 cds_lfht_del(state->client_socket_ht,
1261 &client->client_socket_ht_node);
1262 notification_client_destroy(client, state);
1263end:
1264 rcu_read_unlock();
1265 return ret;
1266}
1267
1268int handle_notification_thread_client_disconnect_all(
1269 struct notification_thread_state *state)
1270{
1271 struct cds_lfht_iter iter;
1272 struct notification_client *client;
1273 bool error_encoutered = false;
1274
1275 rcu_read_lock();
1276 DBG("[notification-thread] Closing all client connections");
1277 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1278 client_socket_ht_node) {
1279 int ret;
1280
1281 ret = handle_notification_thread_client_disconnect(
1282 client->socket, state);
1283 if (ret) {
1284 error_encoutered = true;
1285 }
1286 }
1287 rcu_read_unlock();
1288 return error_encoutered ? 1 : 0;
1289}
1290
1291int handle_notification_thread_trigger_unregister_all(
1292 struct notification_thread_state *state)
1293{
1294 bool error_occured = false;
1295 struct cds_lfht_iter iter;
1296 struct lttng_trigger_ht_element *trigger_ht_element;
1297
1298 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1299 node) {
1300 int ret = handle_notification_thread_command_unregister_trigger(
1301 state, trigger_ht_element->trigger, NULL);
1302 if (ret) {
1303 error_occured = true;
1304 }
1305 }
1306 return error_occured ? -1 : 0;
1307}
1308
1309static
1310int client_flush_outgoing_queue(struct notification_client *client,
1311 struct notification_thread_state *state)
1312{
1313 ssize_t ret;
1314 size_t to_send_count;
1315
1316 assert(client->communication.outbound.buffer.size != 0);
1317 to_send_count = client->communication.outbound.buffer.size;
1318 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
1319 client->socket);
1320
1321 ret = lttcomm_send_unix_sock_non_block(client->socket,
1322 client->communication.outbound.buffer.data,
1323 to_send_count);
1324 if ((ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) ||
1325 (ret > 0 && ret < to_send_count)) {
1326 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
1327 client->socket);
1328 to_send_count -= max(ret, 0);
1329
1330 memcpy(client->communication.outbound.buffer.data,
1331 client->communication.outbound.buffer.data +
1332 client->communication.outbound.buffer.size - to_send_count,
1333 to_send_count);
1334 ret = lttng_dynamic_buffer_set_size(
1335 &client->communication.outbound.buffer,
1336 to_send_count);
1337 if (ret) {
1338 goto error;
1339 }
1340
1341 /*
1342 * We want to be notified whenever there is buffer space
1343 * available to send the rest of the payload.
1344 */
1345 ret = lttng_poll_mod(&state->events, client->socket,
1346 CLIENT_POLL_MASK_IN_OUT);
1347 if (ret) {
1348 goto error;
1349 }
1350 } else if (ret < 0) {
1351 /* Generic error, disconnect the client. */
1352 ERR("[notification-thread] Failed to send flush outgoing queue, disconnecting client (socket fd = %i)",
1353 client->socket);
1354 ret = handle_notification_thread_client_disconnect(
1355 client->socket, state);
1356 if (ret) {
1357 goto error;
1358 }
1359 } else {
1360 /* No error and flushed the queue completely. */
1361 ret = lttng_dynamic_buffer_set_size(
1362 &client->communication.outbound.buffer, 0);
1363 if (ret) {
1364 goto error;
1365 }
1366 ret = lttng_poll_mod(&state->events, client->socket,
1367 CLIENT_POLL_MASK_IN);
1368 if (ret) {
1369 goto error;
1370 }
1371
1372 client->communication.outbound.queued_command_reply = false;
1373 client->communication.outbound.dropped_notification = false;
1374 }
1375
1376 return 0;
1377error:
1378 return -1;
1379}
1380
1381static
1382int client_send_command_reply(struct notification_client *client,
1383 struct notification_thread_state *state,
1384 enum lttng_notification_channel_status status)
1385{
1386 int ret;
1387 struct lttng_notification_channel_command_reply reply = {
1388 .status = (int8_t) status,
1389 };
1390 struct lttng_notification_channel_message msg = {
1391 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
1392 .size = sizeof(reply),
1393 };
1394 char buffer[sizeof(msg) + sizeof(reply)];
1395
1396 if (client->communication.outbound.queued_command_reply) {
1397 /* Protocol error. */
1398 goto error;
1399 }
1400
1401 memcpy(buffer, &msg, sizeof(msg));
1402 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
1403 DBG("[notification-thread] Send command reply (%i)", (int) status);
1404
1405 /* Enqueue buffer to outgoing queue and flush it. */
1406 ret = lttng_dynamic_buffer_append(
1407 &client->communication.outbound.buffer,
1408 buffer, sizeof(buffer));
1409 if (ret) {
1410 goto error;
1411 }
1412
1413 ret = client_flush_outgoing_queue(client, state);
1414 if (ret) {
1415 goto error;
1416 }
1417
1418 if (client->communication.outbound.buffer.size != 0) {
1419 /* Queue could not be emptied. */
1420 client->communication.outbound.queued_command_reply = true;
1421 }
1422
1423 return 0;
1424error:
1425 return -1;
1426}
1427
1428static
1429int client_dispatch_message(struct notification_client *client,
1430 struct notification_thread_state *state)
1431{
1432 int ret = 0;
1433
1434 if (client->communication.inbound.msg_type !=
1435 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
1436 client->communication.inbound.msg_type !=
1437 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
1438 !client->validated) {
1439 WARN("[notification-thread] client attempted a command before handshake");
1440 ret = -1;
1441 goto end;
1442 }
1443
1444 switch (client->communication.inbound.msg_type) {
1445 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
1446 {
1447 /*
1448 * Receiving message header. The function will be called again
1449 * once the rest of the message as been received and can be
1450 * interpreted.
1451 */
1452 const struct lttng_notification_channel_message *msg;
1453
1454 assert(sizeof(*msg) ==
1455 client->communication.inbound.buffer.size);
1456 msg = (const struct lttng_notification_channel_message *)
1457 client->communication.inbound.buffer.data;
1458
1459 if (msg->size == 0 || msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
1460 ERR("[notification-thread] Invalid notification channel message: length = %u", msg->size);
1461 ret = -1;
1462 goto end;
1463 }
1464
1465 switch (msg->type) {
1466 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1467 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1468 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1469 break;
1470 default:
1471 ret = -1;
1472 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
1473 goto end;
1474 }
1475
1476 client->communication.inbound.bytes_to_receive = msg->size;
1477 client->communication.inbound.msg_type =
1478 (enum lttng_notification_channel_message_type) msg->type;
1479 if (client->communication.inbound.msg_type ==
1480 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE) {
1481 client->communication.inbound.receive_creds = true;
1482 }
1483 ret = lttng_dynamic_buffer_set_size(
1484 &client->communication.inbound.buffer, 0);
1485 if (ret) {
1486 goto end;
1487 }
1488 break;
1489 }
1490 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1491 {
1492 struct lttng_notification_channel_command_handshake *handshake_client;
1493 struct lttng_notification_channel_command_handshake handshake_reply = {
1494 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
1495 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
1496 };
1497 struct lttng_notification_channel_message msg_header = {
1498 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
1499 .size = sizeof(handshake_reply),
1500 };
1501 enum lttng_notification_channel_status status =
1502 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1503 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
1504
1505 memcpy(send_buffer, &msg_header, sizeof(msg_header));
1506 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
1507 sizeof(handshake_reply));
1508
1509 handshake_client =
1510 (struct lttng_notification_channel_command_handshake *)
1511 client->communication.inbound.buffer.data;
1512 client->major = handshake_client->major;
1513 client->minor = handshake_client->minor;
1514 if (!client->communication.inbound.creds_received) {
1515 ERR("[notification-thread] No credentials received from client");
1516 ret = -1;
1517 goto end;
1518 }
1519
1520 client->uid = LTTNG_SOCK_GET_UID_CRED(
1521 &client->communication.inbound.creds);
1522 client->gid = LTTNG_SOCK_GET_GID_CRED(
1523 &client->communication.inbound.creds);
1524 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
1525 client->uid, client->gid, (int) client->major,
1526 (int) client->minor);
1527
1528 if (handshake_client->major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
1529 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
1530 }
1531
1532 ret = lttng_dynamic_buffer_append(&client->communication.outbound.buffer,
1533 send_buffer, sizeof(send_buffer));
1534 if (ret) {
1535 ERR("[notification-thread] Failed to send protocol version to notification channel client");
1536 goto end;
1537 }
1538
1539 ret = client_flush_outgoing_queue(client, state);
1540 if (ret) {
1541 goto end;
1542 }
1543
1544 ret = client_send_command_reply(client, state, status);
1545 if (ret) {
1546 ERR("[notification-thread] Failed to send reply to notification channel client");
1547 goto end;
1548 }
1549
1550 /* Set reception state to receive the next message header. */
1551 client_reset_inbound_state(client);
1552 client->validated = true;
1553 break;
1554 }
1555 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1556 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1557 {
1558 struct lttng_condition *condition;
1559 enum lttng_notification_channel_status status =
1560 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1561 const struct lttng_buffer_view condition_view =
1562 lttng_buffer_view_from_dynamic_buffer(
1563 &client->communication.inbound.buffer,
1564 0, -1);
1565 size_t expected_condition_size =
1566 client->communication.inbound.buffer.size;
1567
1568 ret = lttng_condition_create_from_buffer(&condition_view,
1569 &condition);
1570 if (ret != expected_condition_size) {
1571 ERR("[notification-thread] Malformed condition received from client");
1572 goto end;
1573 }
1574
1575 if (client->communication.inbound.msg_type ==
1576 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
1577 /*
1578 * FIXME The current state should be evaluated on
1579 * subscription.
1580 */
1581 ret = notification_thread_client_subscribe(client,
1582 condition, state, &status);
1583 } else {
1584 ret = notification_thread_client_unsubscribe(client,
1585 condition, state, &status);
1586 }
1587 if (ret) {
1588 goto end;
1589 }
1590
1591 ret = client_send_command_reply(client, state, status);
1592 if (ret) {
1593 ERR("[notification-thread] Failed to send reply to notification channel client");
1594 goto end;
1595 }
1596
1597 /* Set reception state to receive the next message header. */
1598 client_reset_inbound_state(client);
1599 break;
1600 }
1601 default:
1602 abort();
1603 }
1604end:
1605 return ret;
1606}
1607
1608/* Incoming data from client. */
1609int handle_notification_thread_client_in(
1610 struct notification_thread_state *state, int socket)
1611{
1612 int ret;
1613 struct notification_client *client;
1614 ssize_t recv_ret;
1615 size_t offset;
1616
1617 client = get_client_from_socket(socket, state);
1618 if (!client) {
1619 /* Internal error, abort. */
1620 ret = -1;
1621 goto end;
1622 }
1623
1624 offset = client->communication.inbound.buffer.size;
1625 ret = lttng_dynamic_buffer_set_size(
1626 &client->communication.inbound.buffer,
1627 client->communication.inbound.bytes_to_receive);
1628 if (ret) {
1629 goto end;
1630 }
1631
1632 if (client->communication.inbound.receive_creds) {
1633 recv_ret = lttcomm_recv_creds_unix_sock(socket,
1634 client->communication.inbound.buffer.data + offset,
1635 client->communication.inbound.bytes_to_receive,
1636 &client->communication.inbound.creds);
1637 if (recv_ret > 0) {
1638 client->communication.inbound.receive_creds = false;
1639 client->communication.inbound.creds_received = true;
1640 }
1641 } else {
1642 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
1643 client->communication.inbound.buffer.data + offset,
1644 client->communication.inbound.bytes_to_receive);
1645 }
1646 if (recv_ret < 0) {
1647 goto error_disconnect_client;
1648 }
1649
1650 client->communication.inbound.bytes_to_receive -= recv_ret;
1651 ret = lttng_dynamic_buffer_set_size(
1652 &client->communication.inbound.buffer,
1653 client->communication.inbound.buffer.size -
1654 client->communication.inbound.bytes_to_receive);
1655 if (ret) {
1656 goto end;
1657 }
1658
1659 if (client->communication.inbound.bytes_to_receive == 0) {
1660 ret = client_dispatch_message(client, state);
1661 if (ret) {
1662 /*
1663 * Only returns an error if this client must be
1664 * disconnected.
1665 */
1666 goto error_disconnect_client;
1667 }
1668 } else {
1669 goto end;
1670 }
1671end:
1672 return ret;
1673error_disconnect_client:
1674 ret = handle_notification_thread_client_disconnect(socket, state);
1675 return ret;
1676}
1677
1678/* Client ready to receive outgoing data. */
1679int handle_notification_thread_client_out(
1680 struct notification_thread_state *state, int socket)
1681{
1682 int ret;
1683 struct notification_client *client;
1684
1685 client = get_client_from_socket(socket, state);
1686 if (!client) {
1687 /* Internal error, abort. */
1688 ret = -1;
1689 goto end;
1690 }
1691
1692 ret = client_flush_outgoing_queue(client, state);
1693 if (ret) {
1694 goto end;
1695 }
1696end:
1697 return ret;
1698}
1699
1700static
1701bool evaluate_buffer_usage_condition(struct lttng_condition *condition,
1702 struct channel_state_sample *sample, uint64_t buffer_capacity)
1703{
1704 bool result = false;
1705 uint64_t threshold;
1706 enum lttng_condition_type condition_type;
1707 struct lttng_condition_buffer_usage *use_condition = container_of(
1708 condition, struct lttng_condition_buffer_usage,
1709 parent);
1710
1711 if (!sample) {
1712 goto end;
1713 }
1714
1715 if (use_condition->threshold_bytes.set) {
1716 threshold = use_condition->threshold_bytes.value;
1717 } else {
1718 /*
1719 * Threshold was expressed as a ratio.
1720 *
1721 * TODO the threshold (in bytes) of conditions expressed
1722 * as a ratio of total buffer size could be cached to
1723 * forego this double-multiplication or it could be performed
1724 * as fixed-point math.
1725 *
1726 * Note that caching should accomodate the case where the
1727 * condition applies to multiple channels (i.e. don't assume
1728 * that all channels matching my_chann* have the same size...)
1729 */
1730 threshold = (uint64_t) (use_condition->threshold_ratio.value *
1731 (double) buffer_capacity);
1732 }
1733
1734 condition_type = lttng_condition_get_type(condition);
1735 if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
1736 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
1737 threshold, sample->highest_usage);
1738
1739 /*
1740 * The low condition should only be triggered once _all_ of the
1741 * streams in a channel have gone below the "low" threshold.
1742 */
1743 if (sample->highest_usage <= threshold) {
1744 result = true;
1745 }
1746 } else {
1747 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
1748 threshold, sample->highest_usage);
1749
1750 /*
1751 * For high buffer usage scenarios, we want to trigger whenever
1752 * _any_ of the streams has reached the "high" threshold.
1753 */
1754 if (sample->highest_usage >= threshold) {
1755 result = true;
1756 }
1757 }
1758end:
1759 return result;
1760}
1761
1762static
1763int evaluate_condition(struct lttng_condition *condition,
1764 struct lttng_evaluation **evaluation,
1765 struct notification_thread_state *state,
1766 struct channel_state_sample *previous_sample,
1767 struct channel_state_sample *latest_sample,
1768 uint64_t buffer_capacity)
1769{
1770 int ret = 0;
1771 enum lttng_condition_type condition_type;
1772 bool previous_sample_result;
1773 bool latest_sample_result;
1774
1775 condition_type = lttng_condition_get_type(condition);
1776 /* No other condition type supported for the moment. */
1777 assert(condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW ||
1778 condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH);
1779
1780 previous_sample_result = evaluate_buffer_usage_condition(condition,
1781 previous_sample, buffer_capacity);
1782 latest_sample_result = evaluate_buffer_usage_condition(condition,
1783 latest_sample, buffer_capacity);
1784
1785 if (!latest_sample_result ||
1786 (previous_sample_result == latest_sample_result)) {
1787 /*
1788 * Only trigger on a condition evaluation transition.
1789 *
1790 * NOTE: This edge-triggered logic may not be appropriate for
1791 * future condition types.
1792 */
1793 goto end;
1794 }
1795
1796 if (evaluation && latest_sample_result) {
1797 *evaluation = lttng_evaluation_buffer_usage_create(
1798 condition_type,
1799 latest_sample->highest_usage,
1800 buffer_capacity);
1801 if (!*evaluation) {
1802 ret = -1;
1803 goto end;
1804 }
1805 }
1806end:
1807 return ret;
1808}
1809
1810static
1811int client_enqueue_dropped_notification(struct notification_client *client,
1812 struct notification_thread_state *state)
1813{
1814 int ret;
1815 struct lttng_notification_channel_message msg = {
1816 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
1817 .size = 0,
1818 };
1819
1820 ret = lttng_dynamic_buffer_append(
1821 &client->communication.outbound.buffer, &msg,
1822 sizeof(msg));
1823 return ret;
1824}
1825
1826static
1827int send_evaluation_to_clients(struct lttng_trigger *trigger,
1828 struct lttng_evaluation *evaluation,
1829 struct notification_client_list* client_list,
1830 struct notification_thread_state *state,
1831 uid_t channel_uid, gid_t channel_gid)
1832{
1833 int ret = 0;
1834 struct lttng_dynamic_buffer msg_buffer;
1835 struct notification_client_list_element *client_list_element, *tmp;
1836 struct lttng_notification *notification;
1837 struct lttng_condition *condition;
1838 ssize_t expected_notification_size, notification_size;
1839 struct lttng_notification_channel_message msg;
1840
1841 lttng_dynamic_buffer_init(&msg_buffer);
1842
1843 condition = lttng_trigger_get_condition(trigger);
1844 assert(condition);
1845
1846 notification = lttng_notification_create(condition, evaluation);
1847 if (!notification) {
1848 ret = -1;
1849 goto end;
1850 }
1851
1852 expected_notification_size = lttng_notification_serialize(notification,
1853 NULL);
1854 if (expected_notification_size < 0) {
1855 ERR("[notification-thread] Failed to get size of serialized notification");
1856 ret = -1;
1857 goto end;
1858 }
1859
1860 msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION;
1861 msg.size = (uint32_t) expected_notification_size;
1862 ret = lttng_dynamic_buffer_append(&msg_buffer, &msg, sizeof(msg));
1863 if (ret) {
1864 goto end;
1865 }
1866
1867 ret = lttng_dynamic_buffer_set_size(&msg_buffer,
1868 msg_buffer.size + expected_notification_size);
1869 if (ret) {
1870 goto end;
1871 }
1872
1873 notification_size = lttng_notification_serialize(notification,
1874 msg_buffer.data + sizeof(msg));
1875 if (notification_size != expected_notification_size) {
1876 ERR("[notification-thread] Failed to serialize notification");
1877 ret = -1;
1878 goto end;
1879 }
1880
1881 cds_list_for_each_entry_safe(client_list_element, tmp,
1882 &client_list->list, node) {
1883 struct notification_client *client =
1884 client_list_element->client;
1885
1886 if (client->uid != channel_uid && client->gid != channel_gid &&
1887 client->uid != 0) {
1888 /* Client is not allowed to monitor this channel. */
1889 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this channel");
1890 continue;
1891 }
1892
1893 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
1894 client->socket, msg_buffer.size);
1895 if (client->communication.outbound.buffer.size) {
1896 /*
1897 * Outgoing data is already buffered for this client;
1898 * drop the notification and enqueue a "dropped
1899 * notification" message if this is the first dropped
1900 * notification since the socket spilled-over to the
1901 * queue.
1902 */
1903 DBG("[notification-thread] Dropping notification addressed to client (socket fd = %i)",
1904 client->socket);
1905 if (!client->communication.outbound.dropped_notification) {
1906 client->communication.outbound.dropped_notification = true;
1907 ret = client_enqueue_dropped_notification(
1908 client, state);
1909 if (ret) {
1910 goto end;
1911 }
1912 }
1913 continue;
1914 }
1915
1916 ret = lttng_dynamic_buffer_append_buffer(
1917 &client->communication.outbound.buffer,
1918 &msg_buffer);
1919 if (ret) {
1920 goto end;
1921 }
1922
1923 ret = client_flush_outgoing_queue(client, state);
1924 if (ret) {
1925 goto end;
1926 }
1927 }
1928 ret = 0;
1929end:
1930 lttng_notification_destroy(notification);
1931 lttng_dynamic_buffer_reset(&msg_buffer);
1932 return ret;
1933}
1934
1935int handle_notification_thread_channel_sample(
1936 struct notification_thread_state *state, int pipe,
1937 enum lttng_domain_type domain)
1938{
1939 int ret = 0;
1940 struct lttcomm_consumer_channel_monitor_msg sample_msg;
1941 struct channel_state_sample previous_sample, latest_sample;
1942 struct channel_info *channel_info;
1943 struct cds_lfht_node *node;
1944 struct cds_lfht_iter iter;
1945 struct lttng_channel_trigger_list *trigger_list;
1946 struct lttng_trigger_list_element *trigger_list_element;
1947 bool previous_sample_available = false;
1948
1949 /*
1950 * The monitoring pipe only holds messages smaller than PIPE_BUF,
1951 * ensuring that read/write of sampling messages are atomic.
1952 */
7c2551ef 1953 ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
ab0ee2ca
JG
1954 if (ret != sizeof(sample_msg)) {
1955 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
1956 pipe);
1957 ret = -1;
1958 goto end;
1959 }
1960
1961 ret = 0;
1962 latest_sample.key.key = sample_msg.key;
1963 latest_sample.key.domain = domain;
1964 latest_sample.highest_usage = sample_msg.highest;
1965 latest_sample.lowest_usage = sample_msg.lowest;
1966
1967 rcu_read_lock();
1968
1969 /* Retrieve the channel's informations */
1970 cds_lfht_lookup(state->channels_ht,
1971 hash_channel_key(&latest_sample.key),
1972 match_channel_info,
1973 &latest_sample.key,
1974 &iter);
1975 node = cds_lfht_iter_get_node(&iter);
1976 if (!node) {
1977 /*
1978 * Not an error since the consumer can push a sample to the pipe
1979 * and the rest of the session daemon could notify us of the
1980 * channel's destruction before we get a chance to process that
1981 * sample.
1982 */
1983 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
1984 latest_sample.key.key,
1985 domain == LTTNG_DOMAIN_KERNEL ? "kernel" :
1986 "user space");
1987 goto end_unlock;
1988 }
1989 channel_info = caa_container_of(node, struct channel_info,
1990 channels_ht_node);
1991 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64")",
1992 channel_info->channel_name,
1993 latest_sample.key.key,
1994 channel_info->session_name,
1995 latest_sample.highest_usage,
1996 latest_sample.lowest_usage);
1997
1998 /* Retrieve the channel's last sample, if it exists, and update it. */
1999 cds_lfht_lookup(state->channel_state_ht,
2000 hash_channel_key(&latest_sample.key),
2001 match_channel_state_sample,
2002 &latest_sample.key,
2003 &iter);
2004 node = cds_lfht_iter_get_node(&iter);
2005 if (node) {
2006 struct channel_state_sample *stored_sample;
2007
2008 /* Update the sample stored. */
2009 stored_sample = caa_container_of(node,
2010 struct channel_state_sample,
2011 channel_state_ht_node);
2012 memcpy(&previous_sample, stored_sample,
2013 sizeof(previous_sample));
2014 stored_sample->highest_usage = latest_sample.highest_usage;
2015 stored_sample->lowest_usage = latest_sample.lowest_usage;
2016 previous_sample_available = true;
2017 } else {
2018 /*
2019 * This is the channel's first sample, allocate space for and
2020 * store the new sample.
2021 */
2022 struct channel_state_sample *stored_sample;
2023
2024 stored_sample = zmalloc(sizeof(*stored_sample));
2025 if (!stored_sample) {
2026 ret = -1;
2027 goto end_unlock;
2028 }
2029
2030 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
2031 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
2032 cds_lfht_add(state->channel_state_ht,
2033 hash_channel_key(&stored_sample->key),
2034 &stored_sample->channel_state_ht_node);
2035 }
2036
2037 /* Find triggers associated with this channel. */
2038 cds_lfht_lookup(state->channel_triggers_ht,
2039 hash_channel_key(&latest_sample.key),
2040 match_channel_trigger_list,
2041 &latest_sample.key,
2042 &iter);
2043 node = cds_lfht_iter_get_node(&iter);
2044 if (!node) {
2045 goto end_unlock;
2046 }
2047
2048 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
2049 channel_triggers_ht_node);
2050 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
2051 node) {
2052 struct lttng_condition *condition;
2053 struct lttng_action *action;
2054 struct lttng_trigger *trigger;
2055 struct notification_client_list *client_list;
2056 struct lttng_evaluation *evaluation = NULL;
2057
2058 trigger = trigger_list_element->trigger;
2059 condition = lttng_trigger_get_condition(trigger);
2060 assert(condition);
2061 action = lttng_trigger_get_action(trigger);
2062
2063 /* Notify actions are the only type currently supported. */
2064 assert(lttng_action_get_type(action) ==
2065 LTTNG_ACTION_TYPE_NOTIFY);
2066
2067 /*
2068 * Check if any client is subscribed to the result of this
2069 * evaluation.
2070 */
2071 cds_lfht_lookup(state->notification_trigger_clients_ht,
2072 lttng_condition_hash(condition),
2073 match_client_list,
2074 trigger,
2075 &iter);
2076 node = cds_lfht_iter_get_node(&iter);
2077 assert(node);
2078
2079 client_list = caa_container_of(node,
2080 struct notification_client_list,
2081 notification_trigger_ht_node);
2082 if (cds_list_empty(&client_list->list)) {
2083 /*
2084 * No clients interested in the evaluation's result,
2085 * skip it.
2086 */
2087 continue;
2088 }
2089
2090 ret = evaluate_condition(condition, &evaluation, state,
2091 previous_sample_available ? &previous_sample : NULL,
2092 &latest_sample, channel_info->capacity);
2093 if (ret) {
2094 goto end_unlock;
2095 }
2096
2097 if (!evaluation) {
2098 continue;
2099 }
2100
2101 /* Dispatch evaluation result to all clients. */
2102 ret = send_evaluation_to_clients(trigger_list_element->trigger,
2103 evaluation, client_list, state,
2104 channel_info->uid, channel_info->gid);
2105 if (ret) {
2106 goto end_unlock;
2107 }
2108 }
2109end_unlock:
2110 rcu_read_unlock();
2111end:
2112 return ret;
2113}
This page took 0.106084 seconds and 5 git commands to generate.