Clarify notification channel info ht destruction error log
[lttng-tools.git] / src / bin / lttng-sessiond / notification-thread-events.c
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 <common/defaults.h>
23 #include <common/error.h>
24 #include <common/futex.h>
25 #include <common/unix.h>
26 #include <common/dynamic-buffer.h>
27 #include <common/hashtable/utils.h>
28 #include <common/sessiond-comm/sessiond-comm.h>
29 #include <common/macros.h>
30 #include <lttng/condition/condition.h>
31 #include <lttng/action/action.h>
32 #include <lttng/notification/notification-internal.h>
33 #include <lttng/condition/condition-internal.h>
34 #include <lttng/condition/buffer-usage-internal.h>
35 #include <lttng/condition/session-consumed-size-internal.h>
36 #include <lttng/notification/channel-internal.h>
37
38 #include <time.h>
39 #include <unistd.h>
40 #include <assert.h>
41 #include <inttypes.h>
42 #include <fcntl.h>
43
44 #include "notification-thread.h"
45 #include "notification-thread-events.h"
46 #include "notification-thread-commands.h"
47 #include "lttng-sessiond.h"
48 #include "kernel.h"
49
50 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
51 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
52
53 struct lttng_trigger_list_element {
54 struct lttng_trigger *trigger;
55 struct cds_list_head node;
56 };
57
58 struct lttng_channel_trigger_list {
59 struct channel_key channel_key;
60 struct cds_list_head list;
61 struct cds_lfht_node channel_triggers_ht_node;
62 };
63
64 struct lttng_trigger_ht_element {
65 struct lttng_trigger *trigger;
66 struct cds_lfht_node node;
67 };
68
69 struct lttng_condition_list_element {
70 struct lttng_condition *condition;
71 struct cds_list_head node;
72 };
73
74 struct notification_client_list_element {
75 struct notification_client *client;
76 struct cds_list_head node;
77 };
78
79 struct notification_client_list {
80 struct lttng_trigger *trigger;
81 struct cds_list_head list;
82 struct cds_lfht_node notification_trigger_ht_node;
83 };
84
85 struct notification_client {
86 int socket;
87 /* Client protocol version. */
88 uint8_t major, minor;
89 uid_t uid;
90 gid_t gid;
91 /*
92 * Indicates if the credentials and versions of the client have been
93 * checked.
94 */
95 bool validated;
96 /*
97 * Conditions to which the client's notification channel is subscribed.
98 * List of struct lttng_condition_list_node. The condition member is
99 * owned by the client.
100 */
101 struct cds_list_head condition_list;
102 struct cds_lfht_node client_socket_ht_node;
103 struct {
104 struct {
105 /*
106 * During the reception of a message, the reception
107 * buffers' "size" is set to contain the current
108 * message's complete payload.
109 */
110 struct lttng_dynamic_buffer buffer;
111 /* Bytes left to receive for the current message. */
112 size_t bytes_to_receive;
113 /* Type of the message being received. */
114 enum lttng_notification_channel_message_type msg_type;
115 /*
116 * Indicates whether or not credentials are expected
117 * from the client.
118 */
119 bool expect_creds;
120 /*
121 * Indicates whether or not credentials were received
122 * from the client.
123 */
124 bool creds_received;
125 /* Only used during credentials reception. */
126 lttng_sock_cred creds;
127 } inbound;
128 struct {
129 /*
130 * Indicates whether or not a notification addressed to
131 * this client was dropped because a command reply was
132 * already buffered.
133 *
134 * A notification is dropped whenever the buffer is not
135 * empty.
136 */
137 bool dropped_notification;
138 /*
139 * Indicates whether or not a command reply is already
140 * buffered. In this case, it means that the client is
141 * not consuming command replies before emitting a new
142 * one. This could be caused by a protocol error or a
143 * misbehaving/malicious client.
144 */
145 bool queued_command_reply;
146 struct lttng_dynamic_buffer buffer;
147 } outbound;
148 } communication;
149 };
150
151 struct channel_state_sample {
152 struct channel_key key;
153 struct cds_lfht_node channel_state_ht_node;
154 uint64_t highest_usage;
155 uint64_t lowest_usage;
156 uint64_t channel_total_consumed;
157 };
158
159 static unsigned long hash_channel_key(struct channel_key *key);
160 static int evaluate_condition(const struct lttng_condition *condition,
161 struct lttng_evaluation **evaluation,
162 const struct notification_thread_state *state,
163 const struct channel_state_sample *previous_sample,
164 const struct channel_state_sample *latest_sample,
165 uint64_t previous_session_consumed_total,
166 uint64_t latest_session_consumed_total,
167 struct channel_info *channel_info);
168 static
169 int send_evaluation_to_clients(struct lttng_trigger *trigger,
170 struct lttng_evaluation *evaluation,
171 struct notification_client_list *client_list,
172 struct notification_thread_state *state,
173 uid_t channel_uid, gid_t channel_gid);
174
175
176 static
177 void session_info_destroy(void *_data);
178 static
179 void session_info_get(struct session_info *session_info);
180 static
181 void session_info_put(struct session_info *session_info);
182 static
183 struct session_info *session_info_create(const char *name,
184 uid_t uid, gid_t gid);
185 static
186 void session_info_add_channel(struct session_info *session_info,
187 struct channel_info *channel_info);
188 static
189 void session_info_remove_channel(struct session_info *session_info,
190 struct channel_info *channel_info);
191
192 static
193 int match_client(struct cds_lfht_node *node, const void *key)
194 {
195 /* This double-cast is intended to supress pointer-to-cast warning. */
196 int socket = (int) (intptr_t) key;
197 struct notification_client *client;
198
199 client = caa_container_of(node, struct notification_client,
200 client_socket_ht_node);
201
202 return !!(client->socket == socket);
203 }
204
205 static
206 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
207 {
208 struct channel_key *channel_key = (struct channel_key *) key;
209 struct lttng_channel_trigger_list *trigger_list;
210
211 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
212 channel_triggers_ht_node);
213
214 return !!((channel_key->key == trigger_list->channel_key.key) &&
215 (channel_key->domain == trigger_list->channel_key.domain));
216 }
217
218 static
219 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
220 {
221 struct channel_key *channel_key = (struct channel_key *) key;
222 struct channel_state_sample *sample;
223
224 sample = caa_container_of(node, struct channel_state_sample,
225 channel_state_ht_node);
226
227 return !!((channel_key->key == sample->key.key) &&
228 (channel_key->domain == sample->key.domain));
229 }
230
231 static
232 int match_channel_info(struct cds_lfht_node *node, const void *key)
233 {
234 struct channel_key *channel_key = (struct channel_key *) key;
235 struct channel_info *channel_info;
236
237 channel_info = caa_container_of(node, struct channel_info,
238 channels_ht_node);
239
240 return !!((channel_key->key == channel_info->key.key) &&
241 (channel_key->domain == channel_info->key.domain));
242 }
243
244 static
245 int match_condition(struct cds_lfht_node *node, const void *key)
246 {
247 struct lttng_condition *condition_key = (struct lttng_condition *) key;
248 struct lttng_trigger_ht_element *trigger;
249 struct lttng_condition *condition;
250
251 trigger = caa_container_of(node, struct lttng_trigger_ht_element,
252 node);
253 condition = lttng_trigger_get_condition(trigger->trigger);
254 assert(condition);
255
256 return !!lttng_condition_is_equal(condition_key, condition);
257 }
258
259 static
260 int match_client_list(struct cds_lfht_node *node, const void *key)
261 {
262 struct lttng_trigger *trigger_key = (struct lttng_trigger *) key;
263 struct notification_client_list *client_list;
264 struct lttng_condition *condition;
265 struct lttng_condition *condition_key = lttng_trigger_get_condition(
266 trigger_key);
267
268 assert(condition_key);
269
270 client_list = caa_container_of(node, struct notification_client_list,
271 notification_trigger_ht_node);
272 condition = lttng_trigger_get_condition(client_list->trigger);
273
274 return !!lttng_condition_is_equal(condition_key, condition);
275 }
276
277 static
278 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
279 {
280 struct lttng_condition *condition_key = (struct lttng_condition *) key;
281 struct notification_client_list *client_list;
282 struct lttng_condition *condition;
283
284 assert(condition_key);
285
286 client_list = caa_container_of(node, struct notification_client_list,
287 notification_trigger_ht_node);
288 condition = lttng_trigger_get_condition(client_list->trigger);
289
290 return !!lttng_condition_is_equal(condition_key, condition);
291 }
292
293 static
294 unsigned long lttng_condition_buffer_usage_hash(
295 struct lttng_condition *_condition)
296 {
297 unsigned long hash = 0;
298 struct lttng_condition_buffer_usage *condition;
299
300 condition = container_of(_condition,
301 struct lttng_condition_buffer_usage, parent);
302
303 if (condition->session_name) {
304 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
305 }
306 if (condition->channel_name) {
307 hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
308 }
309 if (condition->domain.set) {
310 hash ^= hash_key_ulong(
311 (void *) condition->domain.type,
312 lttng_ht_seed);
313 }
314 if (condition->threshold_ratio.set) {
315 uint64_t val;
316
317 val = condition->threshold_ratio.value * (double) UINT32_MAX;
318 hash ^= hash_key_u64(&val, lttng_ht_seed);
319 } else if (condition->threshold_bytes.set) {
320 uint64_t val;
321
322 val = condition->threshold_bytes.value;
323 hash ^= hash_key_u64(&val, lttng_ht_seed);
324 }
325 return hash;
326 }
327
328 static
329 unsigned long lttng_condition_session_consumed_size_hash(
330 struct lttng_condition *_condition)
331 {
332 unsigned long hash = 0;
333 struct lttng_condition_session_consumed_size *condition;
334 uint64_t val;
335
336 condition = container_of(_condition,
337 struct lttng_condition_session_consumed_size, parent);
338
339 if (condition->session_name) {
340 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
341 }
342 val = condition->consumed_threshold_bytes.value;
343 hash ^= hash_key_u64(&val, lttng_ht_seed);
344 return hash;
345 }
346
347 /*
348 * The lttng_condition hashing code is kept in this file (rather than
349 * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
350 * don't want to link in liblttng-ctl.
351 */
352 static
353 unsigned long lttng_condition_hash(struct lttng_condition *condition)
354 {
355 switch (condition->type) {
356 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
357 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
358 return lttng_condition_buffer_usage_hash(condition);
359 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
360 return lttng_condition_session_consumed_size_hash(condition);
361 default:
362 ERR("[notification-thread] Unexpected condition type caught");
363 abort();
364 }
365 }
366
367 static
368 unsigned long hash_channel_key(struct channel_key *key)
369 {
370 unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed);
371 unsigned long domain_hash = hash_key_ulong(
372 (void *) (unsigned long) key->domain, lttng_ht_seed);
373
374 return key_hash ^ domain_hash;
375 }
376
377 static
378 void channel_info_destroy(struct channel_info *channel_info)
379 {
380 if (!channel_info) {
381 return;
382 }
383
384 if (channel_info->session_info) {
385 session_info_remove_channel(channel_info->session_info,
386 channel_info);
387 session_info_put(channel_info->session_info);
388 }
389 if (channel_info->name) {
390 free(channel_info->name);
391 }
392 free(channel_info);
393 }
394
395 /* Don't call directly, use the ref-counting mechanism. */
396 static
397 void session_info_destroy(void *_data)
398 {
399 struct session_info *session_info = _data;
400 int ret;
401
402 assert(session_info);
403 if (session_info->channel_infos_ht) {
404 ret = cds_lfht_destroy(session_info->channel_infos_ht, NULL);
405 if (ret) {
406 ERR("[notification-thread] Failed to destroy channel information hash table");
407 }
408 }
409 free(session_info->name);
410 free(session_info);
411 }
412
413 static
414 void session_info_get(struct session_info *session_info)
415 {
416 if (!session_info) {
417 return;
418 }
419 lttng_ref_get(&session_info->ref);
420 }
421
422 static
423 void session_info_put(struct session_info *session_info)
424 {
425 if (!session_info) {
426 return;
427 }
428 lttng_ref_put(&session_info->ref);
429 }
430
431 static
432 struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid)
433 {
434 struct session_info *session_info;
435
436 assert(name);
437
438 session_info = zmalloc(sizeof(*session_info));
439 if (!session_info) {
440 goto end;
441 }
442 lttng_ref_init(&session_info->ref, session_info_destroy);
443
444 session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
445 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
446 if (!session_info->channel_infos_ht) {
447 goto error;
448 }
449
450 cds_lfht_node_init(&session_info->sessions_ht_node);
451 session_info->name = strdup(name);
452 if (!session_info->name) {
453 goto error;
454 }
455 session_info->uid = uid;
456 session_info->gid = gid;
457 end:
458 return session_info;
459 error:
460 session_info_put(session_info);
461 return NULL;
462 }
463
464 static
465 void session_info_add_channel(struct session_info *session_info,
466 struct channel_info *channel_info)
467 {
468 rcu_read_lock();
469 cds_lfht_add(session_info->channel_infos_ht,
470 hash_channel_key(&channel_info->key),
471 &channel_info->session_info_channels_ht_node);
472 rcu_read_unlock();
473 }
474
475 static
476 void session_info_remove_channel(struct session_info *session_info,
477 struct channel_info *channel_info)
478 {
479 rcu_read_lock();
480 cds_lfht_del(session_info->channel_infos_ht,
481 &channel_info->session_info_channels_ht_node);
482 rcu_read_unlock();
483 }
484
485 static
486 struct channel_info *channel_info_create(const char *channel_name,
487 struct channel_key *channel_key, uint64_t channel_capacity,
488 struct session_info *session_info)
489 {
490 struct channel_info *channel_info = zmalloc(sizeof(*channel_info));
491
492 if (!channel_info) {
493 goto end;
494 }
495
496 cds_lfht_node_init(&channel_info->channels_ht_node);
497 cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
498 memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
499 channel_info->capacity = channel_capacity;
500
501 channel_info->name = strdup(channel_name);
502 if (!channel_info->name) {
503 goto error;
504 }
505
506 /*
507 * Set the references between session and channel infos:
508 * - channel_info holds a strong reference to session_info
509 * - session_info holds a weak reference to channel_info
510 */
511 session_info_get(session_info);
512 session_info_add_channel(session_info, channel_info);
513 channel_info->session_info = session_info;
514 end:
515 return channel_info;
516 error:
517 channel_info_destroy(channel_info);
518 return NULL;
519 }
520
521 /* This function must be called with the RCU read lock held. */
522 static
523 int evaluate_condition_for_client(struct lttng_trigger *trigger,
524 struct lttng_condition *condition,
525 struct notification_client *client,
526 struct notification_thread_state *state)
527 {
528 int ret;
529 struct cds_lfht_iter iter;
530 struct cds_lfht_node *node;
531 struct channel_info *channel_info = NULL;
532 struct channel_key *channel_key = NULL;
533 struct channel_state_sample *last_sample = NULL;
534 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
535 struct lttng_evaluation *evaluation = NULL;
536 struct notification_client_list client_list = { 0 };
537 struct notification_client_list_element client_list_element = { 0 };
538
539 assert(trigger);
540 assert(condition);
541 assert(client);
542 assert(state);
543
544 /* Find the channel associated with the trigger. */
545 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
546 channel_trigger_list , channel_triggers_ht_node) {
547 struct lttng_trigger_list_element *element;
548
549 cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
550 struct lttng_condition *current_condition =
551 lttng_trigger_get_condition(
552 element->trigger);
553
554 assert(current_condition);
555 if (!lttng_condition_is_equal(condition,
556 current_condition)) {
557 continue;
558 }
559
560 /* Found the trigger, save the channel key. */
561 channel_key = &channel_trigger_list->channel_key;
562 break;
563 }
564 if (channel_key) {
565 /* The channel key was found stop iteration. */
566 break;
567 }
568 }
569
570 if (!channel_key){
571 /* No channel found; normal exit. */
572 DBG("[notification-thread] No channel associated with newly subscribed-to condition");
573 ret = 0;
574 goto end;
575 }
576
577 /* Fetch channel info for the matching channel. */
578 cds_lfht_lookup(state->channels_ht,
579 hash_channel_key(channel_key),
580 match_channel_info,
581 channel_key,
582 &iter);
583 node = cds_lfht_iter_get_node(&iter);
584 assert(node);
585 channel_info = caa_container_of(node, struct channel_info,
586 channels_ht_node);
587
588 /* Retrieve the channel's last sample, if it exists. */
589 cds_lfht_lookup(state->channel_state_ht,
590 hash_channel_key(channel_key),
591 match_channel_state_sample,
592 channel_key,
593 &iter);
594 node = cds_lfht_iter_get_node(&iter);
595 if (node) {
596 last_sample = caa_container_of(node,
597 struct channel_state_sample,
598 channel_state_ht_node);
599 } else {
600 /* Nothing to evaluate, no sample was ever taken. Normal exit */
601 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
602 ret = 0;
603 goto end;
604 }
605
606 ret = evaluate_condition(condition, &evaluation, state,
607 NULL, last_sample,
608 0, channel_info->session_info->consumed_data_size,
609 channel_info);
610 if (ret) {
611 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
612 goto end;
613 }
614
615 if (!evaluation) {
616 /* Evaluation yielded nothing. Normal exit. */
617 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
618 ret = 0;
619 goto end;
620 }
621
622 /*
623 * Create a temporary client list with the client currently
624 * subscribing.
625 */
626 cds_lfht_node_init(&client_list.notification_trigger_ht_node);
627 CDS_INIT_LIST_HEAD(&client_list.list);
628 client_list.trigger = trigger;
629
630 CDS_INIT_LIST_HEAD(&client_list_element.node);
631 client_list_element.client = client;
632 cds_list_add(&client_list_element.node, &client_list.list);
633
634 /* Send evaluation result to the newly-subscribed client. */
635 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
636 ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
637 state, channel_info->session_info->uid,
638 channel_info->session_info->gid);
639
640 end:
641 return ret;
642 }
643
644 static
645 int notification_thread_client_subscribe(struct notification_client *client,
646 struct lttng_condition *condition,
647 struct notification_thread_state *state,
648 enum lttng_notification_channel_status *_status)
649 {
650 int ret = 0;
651 struct cds_lfht_iter iter;
652 struct cds_lfht_node *node;
653 struct notification_client_list *client_list;
654 struct lttng_condition_list_element *condition_list_element = NULL;
655 struct notification_client_list_element *client_list_element = NULL;
656 enum lttng_notification_channel_status status =
657 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
658
659 /*
660 * Ensure that the client has not already subscribed to this condition
661 * before.
662 */
663 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
664 if (lttng_condition_is_equal(condition_list_element->condition,
665 condition)) {
666 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
667 goto end;
668 }
669 }
670
671 condition_list_element = zmalloc(sizeof(*condition_list_element));
672 if (!condition_list_element) {
673 ret = -1;
674 goto error;
675 }
676 client_list_element = zmalloc(sizeof(*client_list_element));
677 if (!client_list_element) {
678 ret = -1;
679 goto error;
680 }
681
682 rcu_read_lock();
683
684 /*
685 * Add the newly-subscribed condition to the client's subscription list.
686 */
687 CDS_INIT_LIST_HEAD(&condition_list_element->node);
688 condition_list_element->condition = condition;
689 cds_list_add(&condition_list_element->node, &client->condition_list);
690
691 cds_lfht_lookup(state->notification_trigger_clients_ht,
692 lttng_condition_hash(condition),
693 match_client_list_condition,
694 condition,
695 &iter);
696 node = cds_lfht_iter_get_node(&iter);
697 if (!node) {
698 /*
699 * No notification-emiting trigger registered with this
700 * condition. We don't evaluate the condition right away
701 * since this trigger is not registered yet.
702 */
703 free(client_list_element);
704 goto end_unlock;
705 }
706
707 client_list = caa_container_of(node, struct notification_client_list,
708 notification_trigger_ht_node);
709 /*
710 * The condition to which the client just subscribed is evaluated
711 * at this point so that conditions that are already TRUE result
712 * in a notification being sent out.
713 */
714 if (evaluate_condition_for_client(client_list->trigger, condition,
715 client, state)) {
716 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
717 ret = -1;
718 goto end_unlock;
719 }
720
721 /*
722 * Add the client to the list of clients interested in a given trigger
723 * if a "notification" trigger with a corresponding condition was
724 * added prior.
725 */
726 client_list_element->client = client;
727 CDS_INIT_LIST_HEAD(&client_list_element->node);
728 cds_list_add(&client_list_element->node, &client_list->list);
729 end_unlock:
730 rcu_read_unlock();
731 end:
732 if (_status) {
733 *_status = status;
734 }
735 return ret;
736 error:
737 free(condition_list_element);
738 free(client_list_element);
739 return ret;
740 }
741
742 static
743 int notification_thread_client_unsubscribe(
744 struct notification_client *client,
745 struct lttng_condition *condition,
746 struct notification_thread_state *state,
747 enum lttng_notification_channel_status *_status)
748 {
749 struct cds_lfht_iter iter;
750 struct cds_lfht_node *node;
751 struct notification_client_list *client_list;
752 struct lttng_condition_list_element *condition_list_element,
753 *condition_tmp;
754 struct notification_client_list_element *client_list_element,
755 *client_tmp;
756 bool condition_found = false;
757 enum lttng_notification_channel_status status =
758 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
759
760 /* Remove the condition from the client's condition list. */
761 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
762 &client->condition_list, node) {
763 if (!lttng_condition_is_equal(condition_list_element->condition,
764 condition)) {
765 continue;
766 }
767
768 cds_list_del(&condition_list_element->node);
769 /*
770 * The caller may be iterating on the client's conditions to
771 * tear down a client's connection. In this case, the condition
772 * will be destroyed at the end.
773 */
774 if (condition != condition_list_element->condition) {
775 lttng_condition_destroy(
776 condition_list_element->condition);
777 }
778 free(condition_list_element);
779 condition_found = true;
780 break;
781 }
782
783 if (!condition_found) {
784 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
785 goto end;
786 }
787
788 /*
789 * Remove the client from the list of clients interested the trigger
790 * matching the condition.
791 */
792 rcu_read_lock();
793 cds_lfht_lookup(state->notification_trigger_clients_ht,
794 lttng_condition_hash(condition),
795 match_client_list_condition,
796 condition,
797 &iter);
798 node = cds_lfht_iter_get_node(&iter);
799 if (!node) {
800 goto end_unlock;
801 }
802
803 client_list = caa_container_of(node, struct notification_client_list,
804 notification_trigger_ht_node);
805 cds_list_for_each_entry_safe(client_list_element, client_tmp,
806 &client_list->list, node) {
807 if (client_list_element->client->socket != client->socket) {
808 continue;
809 }
810 cds_list_del(&client_list_element->node);
811 free(client_list_element);
812 break;
813 }
814 end_unlock:
815 rcu_read_unlock();
816 end:
817 lttng_condition_destroy(condition);
818 if (_status) {
819 *_status = status;
820 }
821 return 0;
822 }
823
824 static
825 void notification_client_destroy(struct notification_client *client,
826 struct notification_thread_state *state)
827 {
828 struct lttng_condition_list_element *condition_list_element, *tmp;
829
830 if (!client) {
831 return;
832 }
833
834 /* Release all conditions to which the client was subscribed. */
835 cds_list_for_each_entry_safe(condition_list_element, tmp,
836 &client->condition_list, node) {
837 (void) notification_thread_client_unsubscribe(client,
838 condition_list_element->condition, state, NULL);
839 }
840
841 if (client->socket >= 0) {
842 (void) lttcomm_close_unix_sock(client->socket);
843 }
844 lttng_dynamic_buffer_reset(&client->communication.inbound.buffer);
845 lttng_dynamic_buffer_reset(&client->communication.outbound.buffer);
846 free(client);
847 }
848
849 /*
850 * Call with rcu_read_lock held (and hold for the lifetime of the returned
851 * client pointer).
852 */
853 static
854 struct notification_client *get_client_from_socket(int socket,
855 struct notification_thread_state *state)
856 {
857 struct cds_lfht_iter iter;
858 struct cds_lfht_node *node;
859 struct notification_client *client = NULL;
860
861 cds_lfht_lookup(state->client_socket_ht,
862 hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed),
863 match_client,
864 (void *) (unsigned long) socket,
865 &iter);
866 node = cds_lfht_iter_get_node(&iter);
867 if (!node) {
868 goto end;
869 }
870
871 client = caa_container_of(node, struct notification_client,
872 client_socket_ht_node);
873 end:
874 return client;
875 }
876
877 static
878 bool buffer_usage_condition_applies_to_channel(
879 struct lttng_condition *condition,
880 struct channel_info *channel_info)
881 {
882 enum lttng_condition_status status;
883 enum lttng_domain_type condition_domain;
884 const char *condition_session_name = NULL;
885 const char *condition_channel_name = NULL;
886
887 status = lttng_condition_buffer_usage_get_domain_type(condition,
888 &condition_domain);
889 assert(status == LTTNG_CONDITION_STATUS_OK);
890 if (channel_info->key.domain != condition_domain) {
891 goto fail;
892 }
893
894 status = lttng_condition_buffer_usage_get_session_name(
895 condition, &condition_session_name);
896 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
897
898 status = lttng_condition_buffer_usage_get_channel_name(
899 condition, &condition_channel_name);
900 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
901
902 if (strcmp(channel_info->session_info->name, condition_session_name)) {
903 goto fail;
904 }
905 if (strcmp(channel_info->name, condition_channel_name)) {
906 goto fail;
907 }
908
909 return true;
910 fail:
911 return false;
912 }
913
914 static
915 bool session_consumed_size_condition_applies_to_channel(
916 struct lttng_condition *condition,
917 struct channel_info *channel_info)
918 {
919 enum lttng_condition_status status;
920 const char *condition_session_name = NULL;
921
922 status = lttng_condition_session_consumed_size_get_session_name(
923 condition, &condition_session_name);
924 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
925
926 if (strcmp(channel_info->session_info->name, condition_session_name)) {
927 goto fail;
928 }
929
930 return true;
931 fail:
932 return false;
933 }
934
935 static
936 bool trigger_applies_to_channel(struct lttng_trigger *trigger,
937 struct channel_info *channel_info)
938 {
939 struct lttng_condition *condition;
940 bool trigger_applies;
941
942 condition = lttng_trigger_get_condition(trigger);
943 if (!condition) {
944 goto fail;
945 }
946
947 switch (lttng_condition_get_type(condition)) {
948 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
949 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
950 trigger_applies = buffer_usage_condition_applies_to_channel(
951 condition, channel_info);
952 break;
953 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
954 trigger_applies = session_consumed_size_condition_applies_to_channel(
955 condition, channel_info);
956 break;
957 default:
958 goto fail;
959 }
960
961 return trigger_applies;
962 fail:
963 return false;
964 }
965
966 static
967 bool trigger_applies_to_client(struct lttng_trigger *trigger,
968 struct notification_client *client)
969 {
970 bool applies = false;
971 struct lttng_condition_list_element *condition_list_element;
972
973 cds_list_for_each_entry(condition_list_element, &client->condition_list,
974 node) {
975 applies = lttng_condition_is_equal(
976 condition_list_element->condition,
977 lttng_trigger_get_condition(trigger));
978 if (applies) {
979 break;
980 }
981 }
982 return applies;
983 }
984
985 static
986 int match_session(struct cds_lfht_node *node, const void *key)
987 {
988 const char *name = key;
989 struct session_info *session_info = caa_container_of(
990 node, struct session_info, sessions_ht_node);
991
992 return !strcmp(session_info->name, name);
993 }
994
995 static
996 struct session_info *find_or_create_session_info(
997 struct notification_thread_state *state,
998 const char *name, uid_t uid, gid_t gid)
999 {
1000 struct session_info *session = NULL;
1001 struct cds_lfht_node *node;
1002 struct cds_lfht_iter iter;
1003
1004 rcu_read_lock();
1005 cds_lfht_lookup(state->sessions_ht,
1006 hash_key_str(name, lttng_ht_seed),
1007 match_session,
1008 name,
1009 &iter);
1010 node = cds_lfht_iter_get_node(&iter);
1011 if (node) {
1012 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1013 name, uid, gid);
1014 session = caa_container_of(node, struct session_info,
1015 sessions_ht_node);
1016 assert(session->uid == uid);
1017 assert(session->gid == gid);
1018 goto end;
1019 }
1020
1021 session = session_info_create(name, uid, gid);
1022 if (!session) {
1023 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1024 name, uid, gid);
1025 goto end;
1026 }
1027 end:
1028 rcu_read_unlock();
1029 return session;
1030 }
1031
1032 static
1033 int handle_notification_thread_command_add_channel(
1034 struct notification_thread_state *state,
1035 const char *session_name, uid_t session_uid, gid_t session_gid,
1036 const char *channel_name, enum lttng_domain_type channel_domain,
1037 uint64_t channel_key_int, uint64_t channel_capacity,
1038 enum lttng_error_code *cmd_result)
1039 {
1040 struct cds_list_head trigger_list;
1041 struct channel_info *new_channel_info = NULL;
1042 struct channel_key channel_key = {
1043 .key = channel_key_int,
1044 .domain = channel_domain,
1045 };
1046 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1047 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1048 int trigger_count = 0;
1049 struct cds_lfht_iter iter;
1050 struct session_info *session_info = NULL;
1051
1052 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1053 channel_name, session_name, channel_key_int,
1054 channel_domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1055
1056 CDS_INIT_LIST_HEAD(&trigger_list);
1057
1058 session_info = find_or_create_session_info(state, session_name,
1059 session_uid, session_gid);
1060 if (!session_info) {
1061 /* Allocation error or an internal error occured. */
1062 goto error;
1063 }
1064
1065 new_channel_info = channel_info_create(channel_name, &channel_key,
1066 channel_capacity, session_info);
1067 if (!new_channel_info) {
1068 goto error;
1069 }
1070
1071 /* Build a list of all triggers applying to the new channel. */
1072 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1073 node) {
1074 struct lttng_trigger_list_element *new_element;
1075
1076 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1077 new_channel_info)) {
1078 continue;
1079 }
1080
1081 new_element = zmalloc(sizeof(*new_element));
1082 if (!new_element) {
1083 goto error;
1084 }
1085 CDS_INIT_LIST_HEAD(&new_element->node);
1086 new_element->trigger = trigger_ht_element->trigger;
1087 cds_list_add(&new_element->node, &trigger_list);
1088 trigger_count++;
1089 }
1090
1091 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1092 trigger_count);
1093 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
1094 if (!channel_trigger_list) {
1095 goto error;
1096 }
1097 channel_trigger_list->channel_key = new_channel_info->key;
1098 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1099 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1100 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1101
1102 rcu_read_lock();
1103 /* Add channel to the channel_ht which owns the channel_infos. */
1104 cds_lfht_add(state->channels_ht,
1105 hash_channel_key(&new_channel_info->key),
1106 &new_channel_info->channels_ht_node);
1107 /*
1108 * Add the list of triggers associated with this channel to the
1109 * channel_triggers_ht.
1110 */
1111 cds_lfht_add(state->channel_triggers_ht,
1112 hash_channel_key(&new_channel_info->key),
1113 &channel_trigger_list->channel_triggers_ht_node);
1114 rcu_read_unlock();
1115 *cmd_result = LTTNG_OK;
1116 return 0;
1117 error:
1118 channel_info_destroy(new_channel_info);
1119 session_info_put(session_info);
1120 return 1;
1121 }
1122
1123 static
1124 int handle_notification_thread_command_remove_channel(
1125 struct notification_thread_state *state,
1126 uint64_t channel_key, enum lttng_domain_type domain,
1127 enum lttng_error_code *cmd_result)
1128 {
1129 struct cds_lfht_node *node;
1130 struct cds_lfht_iter iter;
1131 struct lttng_channel_trigger_list *trigger_list;
1132 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1133 struct channel_key key = { .key = channel_key, .domain = domain };
1134 struct channel_info *channel_info;
1135
1136 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
1137 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1138
1139 rcu_read_lock();
1140
1141 cds_lfht_lookup(state->channel_triggers_ht,
1142 hash_channel_key(&key),
1143 match_channel_trigger_list,
1144 &key,
1145 &iter);
1146 node = cds_lfht_iter_get_node(&iter);
1147 /*
1148 * There is a severe internal error if we are being asked to remove a
1149 * channel that doesn't exist.
1150 */
1151 if (!node) {
1152 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1153 goto end;
1154 }
1155
1156 /* Free the list of triggers associated with this channel. */
1157 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1158 channel_triggers_ht_node);
1159 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1160 &trigger_list->list, node) {
1161 cds_list_del(&trigger_list_element->node);
1162 free(trigger_list_element);
1163 }
1164 cds_lfht_del(state->channel_triggers_ht, node);
1165 free(trigger_list);
1166
1167 /* Free sampled channel state. */
1168 cds_lfht_lookup(state->channel_state_ht,
1169 hash_channel_key(&key),
1170 match_channel_state_sample,
1171 &key,
1172 &iter);
1173 node = cds_lfht_iter_get_node(&iter);
1174 /*
1175 * This is expected to be NULL if the channel is destroyed before we
1176 * received a sample.
1177 */
1178 if (node) {
1179 struct channel_state_sample *sample = caa_container_of(node,
1180 struct channel_state_sample,
1181 channel_state_ht_node);
1182
1183 cds_lfht_del(state->channel_state_ht, node);
1184 free(sample);
1185 }
1186
1187 /* Remove the channel from the channels_ht and free it. */
1188 cds_lfht_lookup(state->channels_ht,
1189 hash_channel_key(&key),
1190 match_channel_info,
1191 &key,
1192 &iter);
1193 node = cds_lfht_iter_get_node(&iter);
1194 assert(node);
1195 channel_info = caa_container_of(node, struct channel_info,
1196 channels_ht_node);
1197 cds_lfht_del(state->channels_ht, node);
1198 channel_info_destroy(channel_info);
1199 end:
1200 rcu_read_unlock();
1201 *cmd_result = LTTNG_OK;
1202 return 0;
1203 }
1204
1205 static
1206 int condition_is_supported(struct lttng_condition *condition)
1207 {
1208 int ret;
1209
1210 switch (lttng_condition_get_type(condition)) {
1211 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1212 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1213 {
1214 enum lttng_domain_type domain;
1215
1216 ret = lttng_condition_buffer_usage_get_domain_type(condition,
1217 &domain);
1218 if (ret) {
1219 ret = -1;
1220 goto end;
1221 }
1222
1223 if (domain != LTTNG_DOMAIN_KERNEL) {
1224 ret = 1;
1225 goto end;
1226 }
1227
1228 /*
1229 * Older kernel tracers don't expose the API to monitor their
1230 * buffers. Therefore, we reject triggers that require that
1231 * mechanism to be available to be evaluated.
1232 */
1233 ret = kernel_supports_ring_buffer_snapshot_sample_positions(
1234 kernel_tracer_fd);
1235 break;
1236 }
1237 default:
1238 ret = 1;
1239 }
1240 end:
1241 return ret;
1242 }
1243
1244 /*
1245 * FIXME A client's credentials are not checked when registering a trigger, nor
1246 * are they stored alongside with the trigger.
1247 *
1248 * The effects of this are benign since:
1249 * - The client will succeed in registering the trigger, as it is valid,
1250 * - The trigger will, internally, be bound to the channel,
1251 * - The notifications will not be sent since the client's credentials
1252 * are checked against the channel at that moment.
1253 *
1254 * If this function returns a non-zero value, it means something is
1255 * fundamentally broken and the whole subsystem/thread will be torn down.
1256 *
1257 * If a non-fatal error occurs, just set the cmd_result to the appropriate
1258 * error code.
1259 */
1260 static
1261 int handle_notification_thread_command_register_trigger(
1262 struct notification_thread_state *state,
1263 struct lttng_trigger *trigger,
1264 enum lttng_error_code *cmd_result)
1265 {
1266 int ret = 0;
1267 struct lttng_condition *condition;
1268 struct notification_client *client;
1269 struct notification_client_list *client_list = NULL;
1270 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1271 struct notification_client_list_element *client_list_element, *tmp;
1272 struct cds_lfht_node *node;
1273 struct cds_lfht_iter iter;
1274 struct channel_info *channel;
1275 bool free_trigger = true;
1276
1277 rcu_read_lock();
1278
1279 condition = lttng_trigger_get_condition(trigger);
1280 assert(condition);
1281
1282 ret = condition_is_supported(condition);
1283 if (ret < 0) {
1284 goto error;
1285 } else if (ret == 0) {
1286 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
1287 goto error;
1288 } else {
1289 /* Feature is supported, continue. */
1290 ret = 0;
1291 }
1292
1293 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
1294 if (!trigger_ht_element) {
1295 ret = -1;
1296 goto error;
1297 }
1298
1299 /* Add trigger to the trigger_ht. */
1300 cds_lfht_node_init(&trigger_ht_element->node);
1301 trigger_ht_element->trigger = trigger;
1302
1303 node = cds_lfht_add_unique(state->triggers_ht,
1304 lttng_condition_hash(condition),
1305 match_condition,
1306 condition,
1307 &trigger_ht_element->node);
1308 if (node != &trigger_ht_element->node) {
1309 /* Not a fatal error, simply report it to the client. */
1310 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
1311 goto error_free_ht_element;
1312 }
1313
1314 /*
1315 * Ownership of the trigger and of its wrapper was transfered to
1316 * the triggers_ht.
1317 */
1318 trigger_ht_element = NULL;
1319 free_trigger = false;
1320
1321 /*
1322 * The rest only applies to triggers that have a "notify" action.
1323 * It is not skipped as this is the only action type currently
1324 * supported.
1325 */
1326 client_list = zmalloc(sizeof(*client_list));
1327 if (!client_list) {
1328 ret = -1;
1329 goto error_free_ht_element;
1330 }
1331 cds_lfht_node_init(&client_list->notification_trigger_ht_node);
1332 CDS_INIT_LIST_HEAD(&client_list->list);
1333 client_list->trigger = trigger;
1334
1335 /* Build a list of clients to which this new trigger applies. */
1336 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1337 client_socket_ht_node) {
1338 if (!trigger_applies_to_client(trigger, client)) {
1339 continue;
1340 }
1341
1342 client_list_element = zmalloc(sizeof(*client_list_element));
1343 if (!client_list_element) {
1344 ret = -1;
1345 goto error_free_client_list;
1346 }
1347 CDS_INIT_LIST_HEAD(&client_list_element->node);
1348 client_list_element->client = client;
1349 cds_list_add(&client_list_element->node, &client_list->list);
1350 }
1351
1352 cds_lfht_add(state->notification_trigger_clients_ht,
1353 lttng_condition_hash(condition),
1354 &client_list->notification_trigger_ht_node);
1355
1356 /*
1357 * Add the trigger to list of triggers bound to the channels currently
1358 * known.
1359 */
1360 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
1361 channels_ht_node) {
1362 struct lttng_trigger_list_element *trigger_list_element;
1363 struct lttng_channel_trigger_list *trigger_list;
1364
1365 if (!trigger_applies_to_channel(trigger, channel)) {
1366 continue;
1367 }
1368
1369 cds_lfht_lookup(state->channel_triggers_ht,
1370 hash_channel_key(&channel->key),
1371 match_channel_trigger_list,
1372 &channel->key,
1373 &iter);
1374 node = cds_lfht_iter_get_node(&iter);
1375 assert(node);
1376 trigger_list = caa_container_of(node,
1377 struct lttng_channel_trigger_list,
1378 channel_triggers_ht_node);
1379
1380 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
1381 if (!trigger_list_element) {
1382 ret = -1;
1383 goto error_free_client_list;
1384 }
1385 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
1386 trigger_list_element->trigger = trigger;
1387 cds_list_add(&trigger_list_element->node, &trigger_list->list);
1388 }
1389
1390 /*
1391 * Since there is nothing preventing clients from subscribing to a
1392 * condition before the corresponding trigger is registered, we have
1393 * to evaluate this new condition right away.
1394 *
1395 * At some point, we were waiting for the next "evaluation" (e.g. on
1396 * reception of a channel sample) to evaluate this new condition, but
1397 * that was broken.
1398 *
1399 * The reason it was broken is that waiting for the next sample
1400 * does not allow us to properly handle transitions for edge-triggered
1401 * conditions.
1402 *
1403 * Consider this example: when we handle a new channel sample, we
1404 * evaluate each conditions twice: once with the previous state, and
1405 * again with the newest state. We then use those two results to
1406 * determine whether a state change happened: a condition was false and
1407 * became true. If a state change happened, we have to notify clients.
1408 *
1409 * Now, if a client subscribes to a given notification and registers
1410 * a trigger *after* that subscription, we have to make sure the
1411 * condition is evaluated at this point while considering only the
1412 * current state. Otherwise, the next evaluation cycle may only see
1413 * that the evaluations remain the same (true for samples n-1 and n) and
1414 * the client will never know that the condition has been met.
1415 */
1416 cds_list_for_each_entry_safe(client_list_element, tmp,
1417 &client_list->list, node) {
1418 ret = evaluate_condition_for_client(trigger, condition,
1419 client_list_element->client, state);
1420 if (ret) {
1421 goto error_free_client_list;
1422 }
1423 }
1424
1425 /*
1426 * Client list ownership transferred to the
1427 * notification_trigger_clients_ht.
1428 */
1429 client_list = NULL;
1430
1431 *cmd_result = LTTNG_OK;
1432 error_free_client_list:
1433 if (client_list) {
1434 cds_list_for_each_entry_safe(client_list_element, tmp,
1435 &client_list->list, node) {
1436 free(client_list_element);
1437 }
1438 free(client_list);
1439 }
1440 error_free_ht_element:
1441 free(trigger_ht_element);
1442 error:
1443 if (free_trigger) {
1444 struct lttng_action *action = lttng_trigger_get_action(trigger);
1445
1446 lttng_condition_destroy(condition);
1447 lttng_action_destroy(action);
1448 lttng_trigger_destroy(trigger);
1449 }
1450 rcu_read_unlock();
1451 return ret;
1452 }
1453
1454 static
1455 int handle_notification_thread_command_unregister_trigger(
1456 struct notification_thread_state *state,
1457 struct lttng_trigger *trigger,
1458 enum lttng_error_code *_cmd_reply)
1459 {
1460 struct cds_lfht_iter iter;
1461 struct cds_lfht_node *node, *triggers_ht_node;
1462 struct lttng_channel_trigger_list *trigger_list;
1463 struct notification_client_list *client_list;
1464 struct notification_client_list_element *client_list_element, *tmp;
1465 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1466 struct lttng_condition *condition = lttng_trigger_get_condition(
1467 trigger);
1468 struct lttng_action *action;
1469 enum lttng_error_code cmd_reply;
1470
1471 rcu_read_lock();
1472
1473 cds_lfht_lookup(state->triggers_ht,
1474 lttng_condition_hash(condition),
1475 match_condition,
1476 condition,
1477 &iter);
1478 triggers_ht_node = cds_lfht_iter_get_node(&iter);
1479 if (!triggers_ht_node) {
1480 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
1481 goto end;
1482 } else {
1483 cmd_reply = LTTNG_OK;
1484 }
1485
1486 /* Remove trigger from channel_triggers_ht. */
1487 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
1488 channel_triggers_ht_node) {
1489 struct lttng_trigger_list_element *trigger_element, *tmp;
1490
1491 cds_list_for_each_entry_safe(trigger_element, tmp,
1492 &trigger_list->list, node) {
1493 struct lttng_condition *current_condition =
1494 lttng_trigger_get_condition(
1495 trigger_element->trigger);
1496
1497 assert(current_condition);
1498 if (!lttng_condition_is_equal(condition,
1499 current_condition)) {
1500 continue;
1501 }
1502
1503 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
1504 cds_list_del(&trigger_element->node);
1505 /* A trigger can only appear once per channel */
1506 break;
1507 }
1508 }
1509
1510 /*
1511 * Remove and release the client list from
1512 * notification_trigger_clients_ht.
1513 */
1514 cds_lfht_lookup(state->notification_trigger_clients_ht,
1515 lttng_condition_hash(condition),
1516 match_client_list,
1517 trigger,
1518 &iter);
1519 node = cds_lfht_iter_get_node(&iter);
1520 assert(node);
1521 client_list = caa_container_of(node, struct notification_client_list,
1522 notification_trigger_ht_node);
1523 cds_list_for_each_entry_safe(client_list_element, tmp,
1524 &client_list->list, node) {
1525 free(client_list_element);
1526 }
1527 cds_lfht_del(state->notification_trigger_clients_ht, node);
1528 free(client_list);
1529
1530 /* Remove trigger from triggers_ht. */
1531 trigger_ht_element = caa_container_of(triggers_ht_node,
1532 struct lttng_trigger_ht_element, node);
1533 cds_lfht_del(state->triggers_ht, triggers_ht_node);
1534
1535 condition = lttng_trigger_get_condition(trigger_ht_element->trigger);
1536 lttng_condition_destroy(condition);
1537 action = lttng_trigger_get_action(trigger_ht_element->trigger);
1538 lttng_action_destroy(action);
1539 lttng_trigger_destroy(trigger_ht_element->trigger);
1540 free(trigger_ht_element);
1541 end:
1542 rcu_read_unlock();
1543 if (_cmd_reply) {
1544 *_cmd_reply = cmd_reply;
1545 }
1546 return 0;
1547 }
1548
1549 /* Returns 0 on success, 1 on exit requested, negative value on error. */
1550 int handle_notification_thread_command(
1551 struct notification_thread_handle *handle,
1552 struct notification_thread_state *state)
1553 {
1554 int ret;
1555 uint64_t counter;
1556 struct notification_thread_command *cmd;
1557
1558 /* Read the event pipe to put it back into a quiescent state. */
1559 ret = read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
1560 sizeof(counter));
1561 if (ret == -1) {
1562 goto error;
1563 }
1564
1565 pthread_mutex_lock(&handle->cmd_queue.lock);
1566 cmd = cds_list_first_entry(&handle->cmd_queue.list,
1567 struct notification_thread_command, cmd_list_node);
1568 switch (cmd->type) {
1569 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
1570 DBG("[notification-thread] Received register trigger command");
1571 ret = handle_notification_thread_command_register_trigger(
1572 state, cmd->parameters.trigger,
1573 &cmd->reply_code);
1574 break;
1575 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
1576 DBG("[notification-thread] Received unregister trigger command");
1577 ret = handle_notification_thread_command_unregister_trigger(
1578 state, cmd->parameters.trigger,
1579 &cmd->reply_code);
1580 break;
1581 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
1582 DBG("[notification-thread] Received add channel command");
1583 ret = handle_notification_thread_command_add_channel(
1584 state,
1585 cmd->parameters.add_channel.session.name,
1586 cmd->parameters.add_channel.session.uid,
1587 cmd->parameters.add_channel.session.gid,
1588 cmd->parameters.add_channel.channel.name,
1589 cmd->parameters.add_channel.channel.domain,
1590 cmd->parameters.add_channel.channel.key,
1591 cmd->parameters.add_channel.channel.capacity,
1592 &cmd->reply_code);
1593 break;
1594 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
1595 DBG("[notification-thread] Received remove channel command");
1596 ret = handle_notification_thread_command_remove_channel(
1597 state, cmd->parameters.remove_channel.key,
1598 cmd->parameters.remove_channel.domain,
1599 &cmd->reply_code);
1600 break;
1601 case NOTIFICATION_COMMAND_TYPE_QUIT:
1602 DBG("[notification-thread] Received quit command");
1603 cmd->reply_code = LTTNG_OK;
1604 ret = 1;
1605 goto end;
1606 default:
1607 ERR("[notification-thread] Unknown internal command received");
1608 goto error_unlock;
1609 }
1610
1611 if (ret) {
1612 goto error_unlock;
1613 }
1614 end:
1615 cds_list_del(&cmd->cmd_list_node);
1616 lttng_waiter_wake_up(&cmd->reply_waiter);
1617 pthread_mutex_unlock(&handle->cmd_queue.lock);
1618 return ret;
1619 error_unlock:
1620 /* Wake-up and return a fatal error to the calling thread. */
1621 lttng_waiter_wake_up(&cmd->reply_waiter);
1622 pthread_mutex_unlock(&handle->cmd_queue.lock);
1623 cmd->reply_code = LTTNG_ERR_FATAL;
1624 error:
1625 /* Indicate a fatal error to the caller. */
1626 return -1;
1627 }
1628
1629 static
1630 unsigned long hash_client_socket(int socket)
1631 {
1632 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
1633 }
1634
1635 static
1636 int socket_set_non_blocking(int socket)
1637 {
1638 int ret, flags;
1639
1640 /* Set the pipe as non-blocking. */
1641 ret = fcntl(socket, F_GETFL, 0);
1642 if (ret == -1) {
1643 PERROR("fcntl get socket flags");
1644 goto end;
1645 }
1646 flags = ret;
1647
1648 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
1649 if (ret == -1) {
1650 PERROR("fcntl set O_NONBLOCK socket flag");
1651 goto end;
1652 }
1653 DBG("Client socket (fd = %i) set as non-blocking", socket);
1654 end:
1655 return ret;
1656 }
1657
1658 static
1659 int client_reset_inbound_state(struct notification_client *client)
1660 {
1661 int ret;
1662
1663 ret = lttng_dynamic_buffer_set_size(
1664 &client->communication.inbound.buffer, 0);
1665 assert(!ret);
1666
1667 client->communication.inbound.bytes_to_receive =
1668 sizeof(struct lttng_notification_channel_message);
1669 client->communication.inbound.msg_type =
1670 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
1671 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
1672 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
1673 ret = lttng_dynamic_buffer_set_size(
1674 &client->communication.inbound.buffer,
1675 client->communication.inbound.bytes_to_receive);
1676 return ret;
1677 }
1678
1679 int handle_notification_thread_client_connect(
1680 struct notification_thread_state *state)
1681 {
1682 int ret;
1683 struct notification_client *client;
1684
1685 DBG("[notification-thread] Handling new notification channel client connection");
1686
1687 client = zmalloc(sizeof(*client));
1688 if (!client) {
1689 /* Fatal error. */
1690 ret = -1;
1691 goto error;
1692 }
1693 CDS_INIT_LIST_HEAD(&client->condition_list);
1694 lttng_dynamic_buffer_init(&client->communication.inbound.buffer);
1695 lttng_dynamic_buffer_init(&client->communication.outbound.buffer);
1696 client->communication.inbound.expect_creds = true;
1697 ret = client_reset_inbound_state(client);
1698 if (ret) {
1699 ERR("[notification-thread] Failed to reset client communication's inbound state");
1700 ret = 0;
1701 goto error;
1702 }
1703
1704 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
1705 if (ret < 0) {
1706 ERR("[notification-thread] Failed to accept new notification channel client connection");
1707 ret = 0;
1708 goto error;
1709 }
1710
1711 client->socket = ret;
1712
1713 ret = socket_set_non_blocking(client->socket);
1714 if (ret) {
1715 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
1716 goto error;
1717 }
1718
1719 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
1720 if (ret < 0) {
1721 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
1722 ret = 0;
1723 goto error;
1724 }
1725
1726 ret = lttng_poll_add(&state->events, client->socket,
1727 LPOLLIN | LPOLLERR |
1728 LPOLLHUP | LPOLLRDHUP);
1729 if (ret < 0) {
1730 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
1731 ret = 0;
1732 goto error;
1733 }
1734 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
1735 client->socket);
1736
1737 rcu_read_lock();
1738 cds_lfht_add(state->client_socket_ht,
1739 hash_client_socket(client->socket),
1740 &client->client_socket_ht_node);
1741 rcu_read_unlock();
1742
1743 return ret;
1744 error:
1745 notification_client_destroy(client, state);
1746 return ret;
1747 }
1748
1749 int handle_notification_thread_client_disconnect(
1750 int client_socket,
1751 struct notification_thread_state *state)
1752 {
1753 int ret = 0;
1754 struct notification_client *client;
1755
1756 rcu_read_lock();
1757 DBG("[notification-thread] Closing client connection (socket fd = %i)",
1758 client_socket);
1759 client = get_client_from_socket(client_socket, state);
1760 if (!client) {
1761 /* Internal state corruption, fatal error. */
1762 ERR("[notification-thread] Unable to find client (socket fd = %i)",
1763 client_socket);
1764 ret = -1;
1765 goto end;
1766 }
1767
1768 ret = lttng_poll_del(&state->events, client_socket);
1769 if (ret) {
1770 ERR("[notification-thread] Failed to remove client socket from poll set");
1771 }
1772 cds_lfht_del(state->client_socket_ht,
1773 &client->client_socket_ht_node);
1774 notification_client_destroy(client, state);
1775 end:
1776 rcu_read_unlock();
1777 return ret;
1778 }
1779
1780 int handle_notification_thread_client_disconnect_all(
1781 struct notification_thread_state *state)
1782 {
1783 struct cds_lfht_iter iter;
1784 struct notification_client *client;
1785 bool error_encoutered = false;
1786
1787 rcu_read_lock();
1788 DBG("[notification-thread] Closing all client connections");
1789 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1790 client_socket_ht_node) {
1791 int ret;
1792
1793 ret = handle_notification_thread_client_disconnect(
1794 client->socket, state);
1795 if (ret) {
1796 error_encoutered = true;
1797 }
1798 }
1799 rcu_read_unlock();
1800 return error_encoutered ? 1 : 0;
1801 }
1802
1803 int handle_notification_thread_trigger_unregister_all(
1804 struct notification_thread_state *state)
1805 {
1806 bool error_occurred = false;
1807 struct cds_lfht_iter iter;
1808 struct lttng_trigger_ht_element *trigger_ht_element;
1809
1810 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1811 node) {
1812 int ret = handle_notification_thread_command_unregister_trigger(
1813 state, trigger_ht_element->trigger, NULL);
1814 if (ret) {
1815 error_occurred = true;
1816 }
1817 }
1818 return error_occurred ? -1 : 0;
1819 }
1820
1821 static
1822 int client_flush_outgoing_queue(struct notification_client *client,
1823 struct notification_thread_state *state)
1824 {
1825 ssize_t ret;
1826 size_t to_send_count;
1827
1828 assert(client->communication.outbound.buffer.size != 0);
1829 to_send_count = client->communication.outbound.buffer.size;
1830 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
1831 client->socket);
1832
1833 ret = lttcomm_send_unix_sock_non_block(client->socket,
1834 client->communication.outbound.buffer.data,
1835 to_send_count);
1836 if ((ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) ||
1837 (ret > 0 && ret < to_send_count)) {
1838 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
1839 client->socket);
1840 to_send_count -= max(ret, 0);
1841
1842 memcpy(client->communication.outbound.buffer.data,
1843 client->communication.outbound.buffer.data +
1844 client->communication.outbound.buffer.size - to_send_count,
1845 to_send_count);
1846 ret = lttng_dynamic_buffer_set_size(
1847 &client->communication.outbound.buffer,
1848 to_send_count);
1849 if (ret) {
1850 goto error;
1851 }
1852
1853 /*
1854 * We want to be notified whenever there is buffer space
1855 * available to send the rest of the payload.
1856 */
1857 ret = lttng_poll_mod(&state->events, client->socket,
1858 CLIENT_POLL_MASK_IN_OUT);
1859 if (ret) {
1860 goto error;
1861 }
1862 } else if (ret < 0) {
1863 /* Generic error, disconnect the client. */
1864 ERR("[notification-thread] Failed to send flush outgoing queue, disconnecting client (socket fd = %i)",
1865 client->socket);
1866 ret = handle_notification_thread_client_disconnect(
1867 client->socket, state);
1868 if (ret) {
1869 goto error;
1870 }
1871 } else {
1872 /* No error and flushed the queue completely. */
1873 ret = lttng_dynamic_buffer_set_size(
1874 &client->communication.outbound.buffer, 0);
1875 if (ret) {
1876 goto error;
1877 }
1878 ret = lttng_poll_mod(&state->events, client->socket,
1879 CLIENT_POLL_MASK_IN);
1880 if (ret) {
1881 goto error;
1882 }
1883
1884 client->communication.outbound.queued_command_reply = false;
1885 client->communication.outbound.dropped_notification = false;
1886 }
1887
1888 return 0;
1889 error:
1890 return -1;
1891 }
1892
1893 static
1894 int client_send_command_reply(struct notification_client *client,
1895 struct notification_thread_state *state,
1896 enum lttng_notification_channel_status status)
1897 {
1898 int ret;
1899 struct lttng_notification_channel_command_reply reply = {
1900 .status = (int8_t) status,
1901 };
1902 struct lttng_notification_channel_message msg = {
1903 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
1904 .size = sizeof(reply),
1905 };
1906 char buffer[sizeof(msg) + sizeof(reply)];
1907
1908 if (client->communication.outbound.queued_command_reply) {
1909 /* Protocol error. */
1910 goto error;
1911 }
1912
1913 memcpy(buffer, &msg, sizeof(msg));
1914 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
1915 DBG("[notification-thread] Send command reply (%i)", (int) status);
1916
1917 /* Enqueue buffer to outgoing queue and flush it. */
1918 ret = lttng_dynamic_buffer_append(
1919 &client->communication.outbound.buffer,
1920 buffer, sizeof(buffer));
1921 if (ret) {
1922 goto error;
1923 }
1924
1925 ret = client_flush_outgoing_queue(client, state);
1926 if (ret) {
1927 goto error;
1928 }
1929
1930 if (client->communication.outbound.buffer.size != 0) {
1931 /* Queue could not be emptied. */
1932 client->communication.outbound.queued_command_reply = true;
1933 }
1934
1935 return 0;
1936 error:
1937 return -1;
1938 }
1939
1940 static
1941 int client_dispatch_message(struct notification_client *client,
1942 struct notification_thread_state *state)
1943 {
1944 int ret = 0;
1945
1946 if (client->communication.inbound.msg_type !=
1947 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
1948 client->communication.inbound.msg_type !=
1949 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
1950 !client->validated) {
1951 WARN("[notification-thread] client attempted a command before handshake");
1952 ret = -1;
1953 goto end;
1954 }
1955
1956 switch (client->communication.inbound.msg_type) {
1957 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
1958 {
1959 /*
1960 * Receiving message header. The function will be called again
1961 * once the rest of the message as been received and can be
1962 * interpreted.
1963 */
1964 const struct lttng_notification_channel_message *msg;
1965
1966 assert(sizeof(*msg) ==
1967 client->communication.inbound.buffer.size);
1968 msg = (const struct lttng_notification_channel_message *)
1969 client->communication.inbound.buffer.data;
1970
1971 if (msg->size == 0 || msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
1972 ERR("[notification-thread] Invalid notification channel message: length = %u", msg->size);
1973 ret = -1;
1974 goto end;
1975 }
1976
1977 switch (msg->type) {
1978 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1979 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1980 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1981 break;
1982 default:
1983 ret = -1;
1984 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
1985 goto end;
1986 }
1987
1988 client->communication.inbound.bytes_to_receive = msg->size;
1989 client->communication.inbound.msg_type =
1990 (enum lttng_notification_channel_message_type) msg->type;
1991 ret = lttng_dynamic_buffer_set_size(
1992 &client->communication.inbound.buffer, msg->size);
1993 if (ret) {
1994 goto end;
1995 }
1996 break;
1997 }
1998 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1999 {
2000 struct lttng_notification_channel_command_handshake *handshake_client;
2001 struct lttng_notification_channel_command_handshake handshake_reply = {
2002 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
2003 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
2004 };
2005 struct lttng_notification_channel_message msg_header = {
2006 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
2007 .size = sizeof(handshake_reply),
2008 };
2009 enum lttng_notification_channel_status status =
2010 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
2011 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
2012
2013 memcpy(send_buffer, &msg_header, sizeof(msg_header));
2014 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
2015 sizeof(handshake_reply));
2016
2017 handshake_client =
2018 (struct lttng_notification_channel_command_handshake *)
2019 client->communication.inbound.buffer.data;
2020 client->major = handshake_client->major;
2021 client->minor = handshake_client->minor;
2022 if (!client->communication.inbound.creds_received) {
2023 ERR("[notification-thread] No credentials received from client");
2024 ret = -1;
2025 goto end;
2026 }
2027
2028 client->uid = LTTNG_SOCK_GET_UID_CRED(
2029 &client->communication.inbound.creds);
2030 client->gid = LTTNG_SOCK_GET_GID_CRED(
2031 &client->communication.inbound.creds);
2032 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
2033 client->uid, client->gid, (int) client->major,
2034 (int) client->minor);
2035
2036 if (handshake_client->major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
2037 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
2038 }
2039
2040 ret = lttng_dynamic_buffer_append(&client->communication.outbound.buffer,
2041 send_buffer, sizeof(send_buffer));
2042 if (ret) {
2043 ERR("[notification-thread] Failed to send protocol version to notification channel client");
2044 goto end;
2045 }
2046
2047 ret = client_flush_outgoing_queue(client, state);
2048 if (ret) {
2049 goto end;
2050 }
2051
2052 ret = client_send_command_reply(client, state, status);
2053 if (ret) {
2054 ERR("[notification-thread] Failed to send reply to notification channel client");
2055 goto end;
2056 }
2057
2058 /* Set reception state to receive the next message header. */
2059 ret = client_reset_inbound_state(client);
2060 if (ret) {
2061 ERR("[notification-thread] Failed to reset client communication's inbound state");
2062 goto end;
2063 }
2064 client->validated = true;
2065 break;
2066 }
2067 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
2068 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
2069 {
2070 struct lttng_condition *condition;
2071 enum lttng_notification_channel_status status =
2072 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
2073 const struct lttng_buffer_view condition_view =
2074 lttng_buffer_view_from_dynamic_buffer(
2075 &client->communication.inbound.buffer,
2076 0, -1);
2077 size_t expected_condition_size =
2078 client->communication.inbound.buffer.size;
2079
2080 ret = lttng_condition_create_from_buffer(&condition_view,
2081 &condition);
2082 if (ret != expected_condition_size) {
2083 ERR("[notification-thread] Malformed condition received from client");
2084 goto end;
2085 }
2086
2087 if (client->communication.inbound.msg_type ==
2088 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
2089 ret = notification_thread_client_subscribe(client,
2090 condition, state, &status);
2091 } else {
2092 ret = notification_thread_client_unsubscribe(client,
2093 condition, state, &status);
2094 }
2095 if (ret) {
2096 goto end;
2097 }
2098
2099 ret = client_send_command_reply(client, state, status);
2100 if (ret) {
2101 ERR("[notification-thread] Failed to send reply to notification channel client");
2102 goto end;
2103 }
2104
2105 /* Set reception state to receive the next message header. */
2106 ret = client_reset_inbound_state(client);
2107 if (ret) {
2108 ERR("[notification-thread] Failed to reset client communication's inbound state");
2109 goto end;
2110 }
2111 break;
2112 }
2113 default:
2114 abort();
2115 }
2116 end:
2117 return ret;
2118 }
2119
2120 /* Incoming data from client. */
2121 int handle_notification_thread_client_in(
2122 struct notification_thread_state *state, int socket)
2123 {
2124 int ret = 0;
2125 struct notification_client *client;
2126 ssize_t recv_ret;
2127 size_t offset;
2128
2129 client = get_client_from_socket(socket, state);
2130 if (!client) {
2131 /* Internal error, abort. */
2132 ret = -1;
2133 goto end;
2134 }
2135
2136 offset = client->communication.inbound.buffer.size -
2137 client->communication.inbound.bytes_to_receive;
2138 if (client->communication.inbound.expect_creds) {
2139 recv_ret = lttcomm_recv_creds_unix_sock(socket,
2140 client->communication.inbound.buffer.data + offset,
2141 client->communication.inbound.bytes_to_receive,
2142 &client->communication.inbound.creds);
2143 if (recv_ret > 0) {
2144 client->communication.inbound.expect_creds = false;
2145 client->communication.inbound.creds_received = true;
2146 }
2147 } else {
2148 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
2149 client->communication.inbound.buffer.data + offset,
2150 client->communication.inbound.bytes_to_receive);
2151 }
2152 if (recv_ret < 0) {
2153 goto error_disconnect_client;
2154 }
2155
2156 client->communication.inbound.bytes_to_receive -= recv_ret;
2157 if (client->communication.inbound.bytes_to_receive == 0) {
2158 ret = client_dispatch_message(client, state);
2159 if (ret) {
2160 /*
2161 * Only returns an error if this client must be
2162 * disconnected.
2163 */
2164 goto error_disconnect_client;
2165 }
2166 } else {
2167 goto end;
2168 }
2169 end:
2170 return ret;
2171 error_disconnect_client:
2172 ret = handle_notification_thread_client_disconnect(socket, state);
2173 return ret;
2174 }
2175
2176 /* Client ready to receive outgoing data. */
2177 int handle_notification_thread_client_out(
2178 struct notification_thread_state *state, int socket)
2179 {
2180 int ret;
2181 struct notification_client *client;
2182
2183 client = get_client_from_socket(socket, state);
2184 if (!client) {
2185 /* Internal error, abort. */
2186 ret = -1;
2187 goto end;
2188 }
2189
2190 ret = client_flush_outgoing_queue(client, state);
2191 if (ret) {
2192 goto end;
2193 }
2194 end:
2195 return ret;
2196 }
2197
2198 static
2199 bool evaluate_buffer_usage_condition(const struct lttng_condition *condition,
2200 const struct channel_state_sample *sample,
2201 uint64_t buffer_capacity)
2202 {
2203 bool result = false;
2204 uint64_t threshold;
2205 enum lttng_condition_type condition_type;
2206 const struct lttng_condition_buffer_usage *use_condition = container_of(
2207 condition, struct lttng_condition_buffer_usage,
2208 parent);
2209
2210 if (use_condition->threshold_bytes.set) {
2211 threshold = use_condition->threshold_bytes.value;
2212 } else {
2213 /*
2214 * Threshold was expressed as a ratio.
2215 *
2216 * TODO the threshold (in bytes) of conditions expressed
2217 * as a ratio of total buffer size could be cached to
2218 * forego this double-multiplication or it could be performed
2219 * as fixed-point math.
2220 *
2221 * Note that caching should accomodate the case where the
2222 * condition applies to multiple channels (i.e. don't assume
2223 * that all channels matching my_chann* have the same size...)
2224 */
2225 threshold = (uint64_t) (use_condition->threshold_ratio.value *
2226 (double) buffer_capacity);
2227 }
2228
2229 condition_type = lttng_condition_get_type(condition);
2230 if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
2231 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
2232 threshold, sample->highest_usage);
2233
2234 /*
2235 * The low condition should only be triggered once _all_ of the
2236 * streams in a channel have gone below the "low" threshold.
2237 */
2238 if (sample->highest_usage <= threshold) {
2239 result = true;
2240 }
2241 } else {
2242 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
2243 threshold, sample->highest_usage);
2244
2245 /*
2246 * For high buffer usage scenarios, we want to trigger whenever
2247 * _any_ of the streams has reached the "high" threshold.
2248 */
2249 if (sample->highest_usage >= threshold) {
2250 result = true;
2251 }
2252 }
2253
2254 return result;
2255 }
2256
2257 static
2258 bool evaluate_session_consumed_size_condition(
2259 const struct lttng_condition *condition,
2260 uint64_t session_consumed_size)
2261 {
2262 uint64_t threshold;
2263 const struct lttng_condition_session_consumed_size *size_condition =
2264 container_of(condition,
2265 struct lttng_condition_session_consumed_size,
2266 parent);
2267
2268 threshold = size_condition->consumed_threshold_bytes.value;
2269 DBG("[notification-thread] Session consumed size condition being evaluated: threshold = %" PRIu64 ", current size = %" PRIu64,
2270 threshold, session_consumed_size);
2271 return session_consumed_size >= threshold;
2272 }
2273
2274 static
2275 int evaluate_condition(const struct lttng_condition *condition,
2276 struct lttng_evaluation **evaluation,
2277 const struct notification_thread_state *state,
2278 const struct channel_state_sample *previous_sample,
2279 const struct channel_state_sample *latest_sample,
2280 uint64_t previous_session_consumed_total,
2281 uint64_t latest_session_consumed_total,
2282 struct channel_info *channel_info)
2283 {
2284 int ret = 0;
2285 enum lttng_condition_type condition_type;
2286 const bool previous_sample_available = !!previous_sample;
2287 bool previous_sample_result = false;
2288 bool latest_sample_result;
2289
2290 condition_type = lttng_condition_get_type(condition);
2291
2292 switch (condition_type) {
2293 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2294 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2295 if (caa_likely(previous_sample_available)) {
2296 previous_sample_result =
2297 evaluate_buffer_usage_condition(condition,
2298 previous_sample, channel_info->capacity);
2299 }
2300 latest_sample_result = evaluate_buffer_usage_condition(
2301 condition, latest_sample,
2302 channel_info->capacity);
2303 break;
2304 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
2305 if (caa_likely(previous_sample_available)) {
2306 previous_sample_result =
2307 evaluate_session_consumed_size_condition(
2308 condition,
2309 previous_session_consumed_total);
2310 }
2311 latest_sample_result =
2312 evaluate_session_consumed_size_condition(
2313 condition,
2314 latest_session_consumed_total);
2315 break;
2316 default:
2317 /* Unknown condition type; internal error. */
2318 abort();
2319 }
2320
2321 if (!latest_sample_result ||
2322 (previous_sample_result == latest_sample_result)) {
2323 /*
2324 * Only trigger on a condition evaluation transition.
2325 *
2326 * NOTE: This edge-triggered logic may not be appropriate for
2327 * future condition types.
2328 */
2329 goto end;
2330 }
2331
2332 if (!evaluation || !latest_sample_result) {
2333 goto end;
2334 }
2335
2336 switch (condition_type) {
2337 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2338 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2339 *evaluation = lttng_evaluation_buffer_usage_create(
2340 condition_type,
2341 latest_sample->highest_usage,
2342 channel_info->capacity);
2343 break;
2344 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
2345 *evaluation = lttng_evaluation_session_consumed_size_create(
2346 condition_type,
2347 latest_session_consumed_total);
2348 break;
2349 default:
2350 abort();
2351 }
2352
2353 if (!*evaluation) {
2354 ret = -1;
2355 goto end;
2356 }
2357 end:
2358 return ret;
2359 }
2360
2361 static
2362 int client_enqueue_dropped_notification(struct notification_client *client,
2363 struct notification_thread_state *state)
2364 {
2365 int ret;
2366 struct lttng_notification_channel_message msg = {
2367 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
2368 .size = 0,
2369 };
2370
2371 ret = lttng_dynamic_buffer_append(
2372 &client->communication.outbound.buffer, &msg,
2373 sizeof(msg));
2374 return ret;
2375 }
2376
2377 static
2378 int send_evaluation_to_clients(struct lttng_trigger *trigger,
2379 struct lttng_evaluation *evaluation,
2380 struct notification_client_list* client_list,
2381 struct notification_thread_state *state,
2382 uid_t channel_uid, gid_t channel_gid)
2383 {
2384 int ret = 0;
2385 struct lttng_dynamic_buffer msg_buffer;
2386 struct notification_client_list_element *client_list_element, *tmp;
2387 struct lttng_notification *notification;
2388 struct lttng_condition *condition;
2389 ssize_t expected_notification_size, notification_size;
2390 struct lttng_notification_channel_message msg;
2391
2392 lttng_dynamic_buffer_init(&msg_buffer);
2393
2394 condition = lttng_trigger_get_condition(trigger);
2395 assert(condition);
2396
2397 notification = lttng_notification_create(condition, evaluation);
2398 if (!notification) {
2399 ret = -1;
2400 goto end;
2401 }
2402
2403 expected_notification_size = lttng_notification_serialize(notification,
2404 NULL);
2405 if (expected_notification_size < 0) {
2406 ERR("[notification-thread] Failed to get size of serialized notification");
2407 ret = -1;
2408 goto end;
2409 }
2410
2411 msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION;
2412 msg.size = (uint32_t) expected_notification_size;
2413 ret = lttng_dynamic_buffer_append(&msg_buffer, &msg, sizeof(msg));
2414 if (ret) {
2415 goto end;
2416 }
2417
2418 ret = lttng_dynamic_buffer_set_size(&msg_buffer,
2419 msg_buffer.size + expected_notification_size);
2420 if (ret) {
2421 goto end;
2422 }
2423
2424 notification_size = lttng_notification_serialize(notification,
2425 msg_buffer.data + sizeof(msg));
2426 if (notification_size != expected_notification_size) {
2427 ERR("[notification-thread] Failed to serialize notification");
2428 ret = -1;
2429 goto end;
2430 }
2431
2432 cds_list_for_each_entry_safe(client_list_element, tmp,
2433 &client_list->list, node) {
2434 struct notification_client *client =
2435 client_list_element->client;
2436
2437 if (client->uid != channel_uid && client->gid != channel_gid &&
2438 client->uid != 0) {
2439 /* Client is not allowed to monitor this channel. */
2440 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this channel");
2441 continue;
2442 }
2443
2444 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
2445 client->socket, msg_buffer.size);
2446 if (client->communication.outbound.buffer.size) {
2447 /*
2448 * Outgoing data is already buffered for this client;
2449 * drop the notification and enqueue a "dropped
2450 * notification" message if this is the first dropped
2451 * notification since the socket spilled-over to the
2452 * queue.
2453 */
2454 DBG("[notification-thread] Dropping notification addressed to client (socket fd = %i)",
2455 client->socket);
2456 if (!client->communication.outbound.dropped_notification) {
2457 client->communication.outbound.dropped_notification = true;
2458 ret = client_enqueue_dropped_notification(
2459 client, state);
2460 if (ret) {
2461 goto end;
2462 }
2463 }
2464 continue;
2465 }
2466
2467 ret = lttng_dynamic_buffer_append_buffer(
2468 &client->communication.outbound.buffer,
2469 &msg_buffer);
2470 if (ret) {
2471 goto end;
2472 }
2473
2474 ret = client_flush_outgoing_queue(client, state);
2475 if (ret) {
2476 goto end;
2477 }
2478 }
2479 ret = 0;
2480 end:
2481 lttng_notification_destroy(notification);
2482 lttng_dynamic_buffer_reset(&msg_buffer);
2483 return ret;
2484 }
2485
2486 int handle_notification_thread_channel_sample(
2487 struct notification_thread_state *state, int pipe,
2488 enum lttng_domain_type domain)
2489 {
2490 int ret = 0;
2491 struct lttcomm_consumer_channel_monitor_msg sample_msg;
2492 struct channel_info *channel_info;
2493 struct cds_lfht_node *node;
2494 struct cds_lfht_iter iter;
2495 struct lttng_channel_trigger_list *trigger_list;
2496 struct lttng_trigger_list_element *trigger_list_element;
2497 bool previous_sample_available = false;
2498 struct channel_state_sample previous_sample, latest_sample;
2499 uint64_t previous_session_consumed_total, latest_session_consumed_total;
2500
2501 /*
2502 * The monitoring pipe only holds messages smaller than PIPE_BUF,
2503 * ensuring that read/write of sampling messages are atomic.
2504 */
2505 ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
2506 if (ret != sizeof(sample_msg)) {
2507 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
2508 pipe);
2509 ret = -1;
2510 goto end;
2511 }
2512
2513 ret = 0;
2514 latest_sample.key.key = sample_msg.key;
2515 latest_sample.key.domain = domain;
2516 latest_sample.highest_usage = sample_msg.highest;
2517 latest_sample.lowest_usage = sample_msg.lowest;
2518 latest_sample.channel_total_consumed = sample_msg.total_consumed;
2519
2520 rcu_read_lock();
2521
2522 /* Retrieve the channel's informations */
2523 cds_lfht_lookup(state->channels_ht,
2524 hash_channel_key(&latest_sample.key),
2525 match_channel_info,
2526 &latest_sample.key,
2527 &iter);
2528 node = cds_lfht_iter_get_node(&iter);
2529 if (caa_unlikely(!node)) {
2530 /*
2531 * Not an error since the consumer can push a sample to the pipe
2532 * and the rest of the session daemon could notify us of the
2533 * channel's destruction before we get a chance to process that
2534 * sample.
2535 */
2536 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
2537 latest_sample.key.key,
2538 domain == LTTNG_DOMAIN_KERNEL ? "kernel" :
2539 "user space");
2540 goto end_unlock;
2541 }
2542 channel_info = caa_container_of(node, struct channel_info,
2543 channels_ht_node);
2544 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64", total consumed = %" PRIu64")",
2545 channel_info->name,
2546 latest_sample.key.key,
2547 channel_info->session_info->name,
2548 latest_sample.highest_usage,
2549 latest_sample.lowest_usage,
2550 latest_sample.channel_total_consumed);
2551
2552 previous_session_consumed_total =
2553 channel_info->session_info->consumed_data_size;
2554
2555 /* Retrieve the channel's last sample, if it exists, and update it. */
2556 cds_lfht_lookup(state->channel_state_ht,
2557 hash_channel_key(&latest_sample.key),
2558 match_channel_state_sample,
2559 &latest_sample.key,
2560 &iter);
2561 node = cds_lfht_iter_get_node(&iter);
2562 if (caa_likely(node)) {
2563 struct channel_state_sample *stored_sample;
2564
2565 /* Update the sample stored. */
2566 stored_sample = caa_container_of(node,
2567 struct channel_state_sample,
2568 channel_state_ht_node);
2569
2570 memcpy(&previous_sample, stored_sample,
2571 sizeof(previous_sample));
2572 stored_sample->highest_usage = latest_sample.highest_usage;
2573 stored_sample->lowest_usage = latest_sample.lowest_usage;
2574 stored_sample->channel_total_consumed = latest_sample.channel_total_consumed;
2575 previous_sample_available = true;
2576
2577 latest_session_consumed_total =
2578 previous_session_consumed_total +
2579 (latest_sample.channel_total_consumed - previous_sample.channel_total_consumed);
2580 } else {
2581 /*
2582 * This is the channel's first sample, allocate space for and
2583 * store the new sample.
2584 */
2585 struct channel_state_sample *stored_sample;
2586
2587 stored_sample = zmalloc(sizeof(*stored_sample));
2588 if (!stored_sample) {
2589 ret = -1;
2590 goto end_unlock;
2591 }
2592
2593 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
2594 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
2595 cds_lfht_add(state->channel_state_ht,
2596 hash_channel_key(&stored_sample->key),
2597 &stored_sample->channel_state_ht_node);
2598
2599 latest_session_consumed_total =
2600 previous_session_consumed_total +
2601 latest_sample.channel_total_consumed;
2602 }
2603
2604 channel_info->session_info->consumed_data_size =
2605 latest_session_consumed_total;
2606
2607 /* Find triggers associated with this channel. */
2608 cds_lfht_lookup(state->channel_triggers_ht,
2609 hash_channel_key(&latest_sample.key),
2610 match_channel_trigger_list,
2611 &latest_sample.key,
2612 &iter);
2613 node = cds_lfht_iter_get_node(&iter);
2614 if (caa_likely(!node)) {
2615 goto end_unlock;
2616 }
2617
2618 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
2619 channel_triggers_ht_node);
2620 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
2621 node) {
2622 struct lttng_condition *condition;
2623 struct lttng_action *action;
2624 struct lttng_trigger *trigger;
2625 struct notification_client_list *client_list;
2626 struct lttng_evaluation *evaluation = NULL;
2627
2628 trigger = trigger_list_element->trigger;
2629 condition = lttng_trigger_get_condition(trigger);
2630 assert(condition);
2631 action = lttng_trigger_get_action(trigger);
2632
2633 /* Notify actions are the only type currently supported. */
2634 assert(lttng_action_get_type(action) ==
2635 LTTNG_ACTION_TYPE_NOTIFY);
2636
2637 /*
2638 * Check if any client is subscribed to the result of this
2639 * evaluation.
2640 */
2641 cds_lfht_lookup(state->notification_trigger_clients_ht,
2642 lttng_condition_hash(condition),
2643 match_client_list,
2644 trigger,
2645 &iter);
2646 node = cds_lfht_iter_get_node(&iter);
2647 assert(node);
2648
2649 client_list = caa_container_of(node,
2650 struct notification_client_list,
2651 notification_trigger_ht_node);
2652 if (cds_list_empty(&client_list->list)) {
2653 /*
2654 * No clients interested in the evaluation's result,
2655 * skip it.
2656 */
2657 continue;
2658 }
2659
2660 ret = evaluate_condition(condition, &evaluation, state,
2661 previous_sample_available ? &previous_sample : NULL,
2662 &latest_sample,
2663 previous_session_consumed_total,
2664 latest_session_consumed_total,
2665 channel_info);
2666 if (caa_unlikely(ret)) {
2667 goto end_unlock;
2668 }
2669
2670 if (caa_likely(!evaluation)) {
2671 continue;
2672 }
2673
2674 /* Dispatch evaluation result to all clients. */
2675 ret = send_evaluation_to_clients(trigger_list_element->trigger,
2676 evaluation, client_list, state,
2677 channel_info->session_info->uid,
2678 channel_info->session_info->gid);
2679 lttng_evaluation_destroy(evaluation);
2680 if (caa_unlikely(ret)) {
2681 goto end_unlock;
2682 }
2683 }
2684 end_unlock:
2685 rcu_read_unlock();
2686 end:
2687 return ret;
2688 }
This page took 0.153424 seconds and 4 git commands to generate.