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