9d006845fb93a362c5289065e8398fd3412bdaf1
[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 free(client_list_element);
719 goto end_unlock;
720 }
721
722 /*
723 * Add the client to the list of clients interested in a given trigger
724 * if a "notification" trigger with a corresponding condition was
725 * added prior.
726 */
727 client_list_element->client = client;
728 CDS_INIT_LIST_HEAD(&client_list_element->node);
729 cds_list_add(&client_list_element->node, &client_list->list);
730 end_unlock:
731 rcu_read_unlock();
732 end:
733 if (_status) {
734 *_status = status;
735 }
736 return ret;
737 error:
738 free(condition_list_element);
739 free(client_list_element);
740 return ret;
741 }
742
743 static
744 int notification_thread_client_unsubscribe(
745 struct notification_client *client,
746 struct lttng_condition *condition,
747 struct notification_thread_state *state,
748 enum lttng_notification_channel_status *_status)
749 {
750 struct cds_lfht_iter iter;
751 struct cds_lfht_node *node;
752 struct notification_client_list *client_list;
753 struct lttng_condition_list_element *condition_list_element,
754 *condition_tmp;
755 struct notification_client_list_element *client_list_element,
756 *client_tmp;
757 bool condition_found = false;
758 enum lttng_notification_channel_status status =
759 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
760
761 /* Remove the condition from the client's condition list. */
762 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
763 &client->condition_list, node) {
764 if (!lttng_condition_is_equal(condition_list_element->condition,
765 condition)) {
766 continue;
767 }
768
769 cds_list_del(&condition_list_element->node);
770 /*
771 * The caller may be iterating on the client's conditions to
772 * tear down a client's connection. In this case, the condition
773 * will be destroyed at the end.
774 */
775 if (condition != condition_list_element->condition) {
776 lttng_condition_destroy(
777 condition_list_element->condition);
778 }
779 free(condition_list_element);
780 condition_found = true;
781 break;
782 }
783
784 if (!condition_found) {
785 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
786 goto end;
787 }
788
789 /*
790 * Remove the client from the list of clients interested the trigger
791 * matching the condition.
792 */
793 rcu_read_lock();
794 cds_lfht_lookup(state->notification_trigger_clients_ht,
795 lttng_condition_hash(condition),
796 match_client_list_condition,
797 condition,
798 &iter);
799 node = cds_lfht_iter_get_node(&iter);
800 if (!node) {
801 goto end_unlock;
802 }
803
804 client_list = caa_container_of(node, struct notification_client_list,
805 notification_trigger_ht_node);
806 cds_list_for_each_entry_safe(client_list_element, client_tmp,
807 &client_list->list, node) {
808 if (client_list_element->client->socket != client->socket) {
809 continue;
810 }
811 cds_list_del(&client_list_element->node);
812 free(client_list_element);
813 break;
814 }
815 end_unlock:
816 rcu_read_unlock();
817 end:
818 lttng_condition_destroy(condition);
819 if (_status) {
820 *_status = status;
821 }
822 return 0;
823 }
824
825 static
826 void notification_client_destroy(struct notification_client *client,
827 struct notification_thread_state *state)
828 {
829 struct lttng_condition_list_element *condition_list_element, *tmp;
830
831 if (!client) {
832 return;
833 }
834
835 /* Release all conditions to which the client was subscribed. */
836 cds_list_for_each_entry_safe(condition_list_element, tmp,
837 &client->condition_list, node) {
838 (void) notification_thread_client_unsubscribe(client,
839 condition_list_element->condition, state, NULL);
840 }
841
842 if (client->socket >= 0) {
843 (void) lttcomm_close_unix_sock(client->socket);
844 }
845 lttng_dynamic_buffer_reset(&client->communication.inbound.buffer);
846 lttng_dynamic_buffer_reset(&client->communication.outbound.buffer);
847 free(client);
848 }
849
850 /*
851 * Call with rcu_read_lock held (and hold for the lifetime of the returned
852 * client pointer).
853 */
854 static
855 struct notification_client *get_client_from_socket(int socket,
856 struct notification_thread_state *state)
857 {
858 struct cds_lfht_iter iter;
859 struct cds_lfht_node *node;
860 struct notification_client *client = NULL;
861
862 cds_lfht_lookup(state->client_socket_ht,
863 hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed),
864 match_client,
865 (void *) (unsigned long) socket,
866 &iter);
867 node = cds_lfht_iter_get_node(&iter);
868 if (!node) {
869 goto end;
870 }
871
872 client = caa_container_of(node, struct notification_client,
873 client_socket_ht_node);
874 end:
875 return client;
876 }
877
878 static
879 bool buffer_usage_condition_applies_to_channel(
880 struct lttng_condition *condition,
881 struct channel_info *channel_info)
882 {
883 enum lttng_condition_status status;
884 enum lttng_domain_type condition_domain;
885 const char *condition_session_name = NULL;
886 const char *condition_channel_name = NULL;
887
888 status = lttng_condition_buffer_usage_get_domain_type(condition,
889 &condition_domain);
890 assert(status == LTTNG_CONDITION_STATUS_OK);
891 if (channel_info->key.domain != condition_domain) {
892 goto fail;
893 }
894
895 status = lttng_condition_buffer_usage_get_session_name(
896 condition, &condition_session_name);
897 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
898
899 status = lttng_condition_buffer_usage_get_channel_name(
900 condition, &condition_channel_name);
901 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
902
903 if (strcmp(channel_info->session_info->name, condition_session_name)) {
904 goto fail;
905 }
906 if (strcmp(channel_info->name, condition_channel_name)) {
907 goto fail;
908 }
909
910 return true;
911 fail:
912 return false;
913 }
914
915 static
916 bool session_consumed_size_condition_applies_to_channel(
917 struct lttng_condition *condition,
918 struct channel_info *channel_info)
919 {
920 enum lttng_condition_status status;
921 const char *condition_session_name = NULL;
922
923 status = lttng_condition_session_consumed_size_get_session_name(
924 condition, &condition_session_name);
925 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
926
927 if (strcmp(channel_info->session_info->name, condition_session_name)) {
928 goto fail;
929 }
930
931 return true;
932 fail:
933 return false;
934 }
935
936 static
937 bool trigger_applies_to_channel(struct lttng_trigger *trigger,
938 struct channel_info *channel_info)
939 {
940 struct lttng_condition *condition;
941 bool trigger_applies;
942
943 condition = lttng_trigger_get_condition(trigger);
944 if (!condition) {
945 goto fail;
946 }
947
948 switch (lttng_condition_get_type(condition)) {
949 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
950 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
951 trigger_applies = buffer_usage_condition_applies_to_channel(
952 condition, channel_info);
953 break;
954 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
955 trigger_applies = session_consumed_size_condition_applies_to_channel(
956 condition, channel_info);
957 break;
958 default:
959 goto fail;
960 }
961
962 return trigger_applies;
963 fail:
964 return false;
965 }
966
967 static
968 bool trigger_applies_to_client(struct lttng_trigger *trigger,
969 struct notification_client *client)
970 {
971 bool applies = false;
972 struct lttng_condition_list_element *condition_list_element;
973
974 cds_list_for_each_entry(condition_list_element, &client->condition_list,
975 node) {
976 applies = lttng_condition_is_equal(
977 condition_list_element->condition,
978 lttng_trigger_get_condition(trigger));
979 if (applies) {
980 break;
981 }
982 }
983 return applies;
984 }
985
986 static
987 int match_session(struct cds_lfht_node *node, const void *key)
988 {
989 const char *name = key;
990 struct session_info *session_info = caa_container_of(
991 node, struct session_info, sessions_ht_node);
992
993 return !strcmp(session_info->name, name);
994 }
995
996 static
997 struct session_info *find_or_create_session_info(
998 struct notification_thread_state *state,
999 const char *name, uid_t uid, gid_t gid)
1000 {
1001 struct session_info *session = NULL;
1002 struct cds_lfht_node *node;
1003 struct cds_lfht_iter iter;
1004
1005 rcu_read_lock();
1006 cds_lfht_lookup(state->sessions_ht,
1007 hash_key_str(name, lttng_ht_seed),
1008 match_session,
1009 name,
1010 &iter);
1011 node = cds_lfht_iter_get_node(&iter);
1012 if (node) {
1013 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1014 name, uid, gid);
1015 session = caa_container_of(node, struct session_info,
1016 sessions_ht_node);
1017 assert(session->uid == uid);
1018 assert(session->gid == gid);
1019 goto end;
1020 }
1021
1022 session = session_info_create(name, uid, gid);
1023 if (!session) {
1024 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1025 name, uid, gid);
1026 goto end;
1027 }
1028 end:
1029 rcu_read_unlock();
1030 return session;
1031 }
1032
1033 static
1034 int handle_notification_thread_command_add_channel(
1035 struct notification_thread_state *state,
1036 const char *session_name, uid_t session_uid, gid_t session_gid,
1037 const char *channel_name, enum lttng_domain_type channel_domain,
1038 uint64_t channel_key_int, uint64_t channel_capacity,
1039 enum lttng_error_code *cmd_result)
1040 {
1041 struct cds_list_head trigger_list;
1042 struct channel_info *new_channel_info = NULL;
1043 struct channel_key channel_key = {
1044 .key = channel_key_int,
1045 .domain = channel_domain,
1046 };
1047 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1048 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1049 int trigger_count = 0;
1050 struct cds_lfht_iter iter;
1051 struct session_info *session_info = NULL;
1052
1053 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1054 channel_name, session_name, channel_key_int,
1055 channel_domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1056
1057 CDS_INIT_LIST_HEAD(&trigger_list);
1058
1059 session_info = find_or_create_session_info(state, session_name,
1060 session_uid, session_gid);
1061 if (!session_info) {
1062 /* Allocation error or an internal error occured. */
1063 goto error;
1064 }
1065
1066 new_channel_info = channel_info_create(channel_name, &channel_key,
1067 channel_capacity, session_info);
1068 if (!new_channel_info) {
1069 goto error;
1070 }
1071
1072 /* Build a list of all triggers applying to the new channel. */
1073 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1074 node) {
1075 struct lttng_trigger_list_element *new_element;
1076
1077 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1078 new_channel_info)) {
1079 continue;
1080 }
1081
1082 new_element = zmalloc(sizeof(*new_element));
1083 if (!new_element) {
1084 goto error;
1085 }
1086 CDS_INIT_LIST_HEAD(&new_element->node);
1087 new_element->trigger = trigger_ht_element->trigger;
1088 cds_list_add(&new_element->node, &trigger_list);
1089 trigger_count++;
1090 }
1091
1092 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1093 trigger_count);
1094 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
1095 if (!channel_trigger_list) {
1096 goto error;
1097 }
1098 channel_trigger_list->channel_key = new_channel_info->key;
1099 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1100 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1101 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1102
1103 rcu_read_lock();
1104 /* Add channel to the channel_ht which owns the channel_infos. */
1105 cds_lfht_add(state->channels_ht,
1106 hash_channel_key(&new_channel_info->key),
1107 &new_channel_info->channels_ht_node);
1108 /*
1109 * Add the list of triggers associated with this channel to the
1110 * channel_triggers_ht.
1111 */
1112 cds_lfht_add(state->channel_triggers_ht,
1113 hash_channel_key(&new_channel_info->key),
1114 &channel_trigger_list->channel_triggers_ht_node);
1115 rcu_read_unlock();
1116 *cmd_result = LTTNG_OK;
1117 return 0;
1118 error:
1119 channel_info_destroy(new_channel_info);
1120 session_info_put(session_info);
1121 return 1;
1122 }
1123
1124 static
1125 int handle_notification_thread_command_remove_channel(
1126 struct notification_thread_state *state,
1127 uint64_t channel_key, enum lttng_domain_type domain,
1128 enum lttng_error_code *cmd_result)
1129 {
1130 struct cds_lfht_node *node;
1131 struct cds_lfht_iter iter;
1132 struct lttng_channel_trigger_list *trigger_list;
1133 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1134 struct channel_key key = { .key = channel_key, .domain = domain };
1135 struct channel_info *channel_info;
1136
1137 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
1138 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1139
1140 rcu_read_lock();
1141
1142 cds_lfht_lookup(state->channel_triggers_ht,
1143 hash_channel_key(&key),
1144 match_channel_trigger_list,
1145 &key,
1146 &iter);
1147 node = cds_lfht_iter_get_node(&iter);
1148 /*
1149 * There is a severe internal error if we are being asked to remove a
1150 * channel that doesn't exist.
1151 */
1152 if (!node) {
1153 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1154 goto end;
1155 }
1156
1157 /* Free the list of triggers associated with this channel. */
1158 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1159 channel_triggers_ht_node);
1160 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1161 &trigger_list->list, node) {
1162 cds_list_del(&trigger_list_element->node);
1163 free(trigger_list_element);
1164 }
1165 cds_lfht_del(state->channel_triggers_ht, node);
1166 free(trigger_list);
1167
1168 /* Free sampled channel state. */
1169 cds_lfht_lookup(state->channel_state_ht,
1170 hash_channel_key(&key),
1171 match_channel_state_sample,
1172 &key,
1173 &iter);
1174 node = cds_lfht_iter_get_node(&iter);
1175 /*
1176 * This is expected to be NULL if the channel is destroyed before we
1177 * received a sample.
1178 */
1179 if (node) {
1180 struct channel_state_sample *sample = caa_container_of(node,
1181 struct channel_state_sample,
1182 channel_state_ht_node);
1183
1184 cds_lfht_del(state->channel_state_ht, node);
1185 free(sample);
1186 }
1187
1188 /* Remove the channel from the channels_ht and free it. */
1189 cds_lfht_lookup(state->channels_ht,
1190 hash_channel_key(&key),
1191 match_channel_info,
1192 &key,
1193 &iter);
1194 node = cds_lfht_iter_get_node(&iter);
1195 assert(node);
1196 channel_info = caa_container_of(node, struct channel_info,
1197 channels_ht_node);
1198 cds_lfht_del(state->channels_ht, node);
1199 channel_info_destroy(channel_info);
1200 end:
1201 rcu_read_unlock();
1202 *cmd_result = LTTNG_OK;
1203 return 0;
1204 }
1205
1206 static
1207 int condition_is_supported(struct lttng_condition *condition)
1208 {
1209 int ret;
1210
1211 switch (lttng_condition_get_type(condition)) {
1212 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1213 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1214 {
1215 enum lttng_domain_type domain;
1216
1217 ret = lttng_condition_buffer_usage_get_domain_type(condition,
1218 &domain);
1219 if (ret) {
1220 ret = -1;
1221 goto end;
1222 }
1223
1224 if (domain != LTTNG_DOMAIN_KERNEL) {
1225 ret = 1;
1226 goto end;
1227 }
1228
1229 /*
1230 * Older kernel tracers don't expose the API to monitor their
1231 * buffers. Therefore, we reject triggers that require that
1232 * mechanism to be available to be evaluated.
1233 */
1234 ret = kernel_supports_ring_buffer_snapshot_sample_positions(
1235 kernel_tracer_fd);
1236 break;
1237 }
1238 default:
1239 ret = 1;
1240 }
1241 end:
1242 return ret;
1243 }
1244
1245 /*
1246 * FIXME A client's credentials are not checked when registering a trigger, nor
1247 * are they stored alongside with the trigger.
1248 *
1249 * The effects of this are benign since:
1250 * - The client will succeed in registering the trigger, as it is valid,
1251 * - The trigger will, internally, be bound to the channel,
1252 * - The notifications will not be sent since the client's credentials
1253 * are checked against the channel at that moment.
1254 *
1255 * If this function returns a non-zero value, it means something is
1256 * fundamentally broken and the whole subsystem/thread will be torn down.
1257 *
1258 * If a non-fatal error occurs, just set the cmd_result to the appropriate
1259 * error code.
1260 */
1261 static
1262 int handle_notification_thread_command_register_trigger(
1263 struct notification_thread_state *state,
1264 struct lttng_trigger *trigger,
1265 enum lttng_error_code *cmd_result)
1266 {
1267 int ret = 0;
1268 struct lttng_condition *condition;
1269 struct notification_client *client;
1270 struct notification_client_list *client_list = NULL;
1271 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1272 struct notification_client_list_element *client_list_element, *tmp;
1273 struct cds_lfht_node *node;
1274 struct cds_lfht_iter iter;
1275 struct channel_info *channel;
1276 bool free_trigger = true;
1277
1278 rcu_read_lock();
1279
1280 condition = lttng_trigger_get_condition(trigger);
1281 assert(condition);
1282
1283 ret = condition_is_supported(condition);
1284 if (ret < 0) {
1285 goto error;
1286 } else if (ret == 0) {
1287 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
1288 goto error;
1289 } else {
1290 /* Feature is supported, continue. */
1291 ret = 0;
1292 }
1293
1294 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
1295 if (!trigger_ht_element) {
1296 ret = -1;
1297 goto error;
1298 }
1299
1300 /* Add trigger to the trigger_ht. */
1301 cds_lfht_node_init(&trigger_ht_element->node);
1302 trigger_ht_element->trigger = trigger;
1303
1304 node = cds_lfht_add_unique(state->triggers_ht,
1305 lttng_condition_hash(condition),
1306 match_condition,
1307 condition,
1308 &trigger_ht_element->node);
1309 if (node != &trigger_ht_element->node) {
1310 /* Not a fatal error, simply report it to the client. */
1311 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
1312 goto error_free_ht_element;
1313 }
1314
1315 /*
1316 * Ownership of the trigger and of its wrapper was transfered to
1317 * the triggers_ht.
1318 */
1319 trigger_ht_element = NULL;
1320 free_trigger = false;
1321
1322 /*
1323 * The rest only applies to triggers that have a "notify" action.
1324 * It is not skipped as this is the only action type currently
1325 * supported.
1326 */
1327 client_list = zmalloc(sizeof(*client_list));
1328 if (!client_list) {
1329 ret = -1;
1330 goto error_free_ht_element;
1331 }
1332 cds_lfht_node_init(&client_list->notification_trigger_ht_node);
1333 CDS_INIT_LIST_HEAD(&client_list->list);
1334 client_list->trigger = trigger;
1335
1336 /* Build a list of clients to which this new trigger applies. */
1337 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1338 client_socket_ht_node) {
1339 if (!trigger_applies_to_client(trigger, client)) {
1340 continue;
1341 }
1342
1343 client_list_element = zmalloc(sizeof(*client_list_element));
1344 if (!client_list_element) {
1345 ret = -1;
1346 goto error_free_client_list;
1347 }
1348 CDS_INIT_LIST_HEAD(&client_list_element->node);
1349 client_list_element->client = client;
1350 cds_list_add(&client_list_element->node, &client_list->list);
1351 }
1352
1353 cds_lfht_add(state->notification_trigger_clients_ht,
1354 lttng_condition_hash(condition),
1355 &client_list->notification_trigger_ht_node);
1356
1357 /*
1358 * Add the trigger to list of triggers bound to the channels currently
1359 * known.
1360 */
1361 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
1362 channels_ht_node) {
1363 struct lttng_trigger_list_element *trigger_list_element;
1364 struct lttng_channel_trigger_list *trigger_list;
1365
1366 if (!trigger_applies_to_channel(trigger, channel)) {
1367 continue;
1368 }
1369
1370 cds_lfht_lookup(state->channel_triggers_ht,
1371 hash_channel_key(&channel->key),
1372 match_channel_trigger_list,
1373 &channel->key,
1374 &iter);
1375 node = cds_lfht_iter_get_node(&iter);
1376 assert(node);
1377 trigger_list = caa_container_of(node,
1378 struct lttng_channel_trigger_list,
1379 channel_triggers_ht_node);
1380
1381 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
1382 if (!trigger_list_element) {
1383 ret = -1;
1384 goto error_free_client_list;
1385 }
1386 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
1387 trigger_list_element->trigger = trigger;
1388 cds_list_add(&trigger_list_element->node, &trigger_list->list);
1389 }
1390
1391 /*
1392 * Since there is nothing preventing clients from subscribing to a
1393 * condition before the corresponding trigger is registered, we have
1394 * to evaluate this new condition right away.
1395 *
1396 * At some point, we were waiting for the next "evaluation" (e.g. on
1397 * reception of a channel sample) to evaluate this new condition, but
1398 * that was broken.
1399 *
1400 * The reason it was broken is that waiting for the next sample
1401 * does not allow us to properly handle transitions for edge-triggered
1402 * conditions.
1403 *
1404 * Consider this example: when we handle a new channel sample, we
1405 * evaluate each conditions twice: once with the previous state, and
1406 * again with the newest state. We then use those two results to
1407 * determine whether a state change happened: a condition was false and
1408 * became true. If a state change happened, we have to notify clients.
1409 *
1410 * Now, if a client subscribes to a given notification and registers
1411 * a trigger *after* that subscription, we have to make sure the
1412 * condition is evaluated at this point while considering only the
1413 * current state. Otherwise, the next evaluation cycle may only see
1414 * that the evaluations remain the same (true for samples n-1 and n) and
1415 * the client will never know that the condition has been met.
1416 */
1417 cds_list_for_each_entry_safe(client_list_element, tmp,
1418 &client_list->list, node) {
1419 ret = evaluate_condition_for_client(trigger, condition,
1420 client_list_element->client, state);
1421 if (ret) {
1422 goto error_free_client_list;
1423 }
1424 }
1425
1426 /*
1427 * Client list ownership transferred to the
1428 * notification_trigger_clients_ht.
1429 */
1430 client_list = NULL;
1431
1432 *cmd_result = LTTNG_OK;
1433 error_free_client_list:
1434 if (client_list) {
1435 cds_list_for_each_entry_safe(client_list_element, tmp,
1436 &client_list->list, node) {
1437 free(client_list_element);
1438 }
1439 free(client_list);
1440 }
1441 error_free_ht_element:
1442 free(trigger_ht_element);
1443 error:
1444 if (free_trigger) {
1445 struct lttng_action *action = lttng_trigger_get_action(trigger);
1446
1447 lttng_condition_destroy(condition);
1448 lttng_action_destroy(action);
1449 lttng_trigger_destroy(trigger);
1450 }
1451 rcu_read_unlock();
1452 return ret;
1453 }
1454
1455 static
1456 int handle_notification_thread_command_unregister_trigger(
1457 struct notification_thread_state *state,
1458 struct lttng_trigger *trigger,
1459 enum lttng_error_code *_cmd_reply)
1460 {
1461 struct cds_lfht_iter iter;
1462 struct cds_lfht_node *node, *triggers_ht_node;
1463 struct lttng_channel_trigger_list *trigger_list;
1464 struct notification_client_list *client_list;
1465 struct notification_client_list_element *client_list_element, *tmp;
1466 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1467 struct lttng_condition *condition = lttng_trigger_get_condition(
1468 trigger);
1469 struct lttng_action *action;
1470 enum lttng_error_code cmd_reply;
1471
1472 rcu_read_lock();
1473
1474 cds_lfht_lookup(state->triggers_ht,
1475 lttng_condition_hash(condition),
1476 match_condition,
1477 condition,
1478 &iter);
1479 triggers_ht_node = cds_lfht_iter_get_node(&iter);
1480 if (!triggers_ht_node) {
1481 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
1482 goto end;
1483 } else {
1484 cmd_reply = LTTNG_OK;
1485 }
1486
1487 /* Remove trigger from channel_triggers_ht. */
1488 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
1489 channel_triggers_ht_node) {
1490 struct lttng_trigger_list_element *trigger_element, *tmp;
1491
1492 cds_list_for_each_entry_safe(trigger_element, tmp,
1493 &trigger_list->list, node) {
1494 struct lttng_condition *current_condition =
1495 lttng_trigger_get_condition(
1496 trigger_element->trigger);
1497
1498 assert(current_condition);
1499 if (!lttng_condition_is_equal(condition,
1500 current_condition)) {
1501 continue;
1502 }
1503
1504 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
1505 cds_list_del(&trigger_element->node);
1506 /* A trigger can only appear once per channel */
1507 break;
1508 }
1509 }
1510
1511 /*
1512 * Remove and release the client list from
1513 * notification_trigger_clients_ht.
1514 */
1515 cds_lfht_lookup(state->notification_trigger_clients_ht,
1516 lttng_condition_hash(condition),
1517 match_client_list,
1518 trigger,
1519 &iter);
1520 node = cds_lfht_iter_get_node(&iter);
1521 assert(node);
1522 client_list = caa_container_of(node, struct notification_client_list,
1523 notification_trigger_ht_node);
1524 cds_list_for_each_entry_safe(client_list_element, tmp,
1525 &client_list->list, node) {
1526 free(client_list_element);
1527 }
1528 cds_lfht_del(state->notification_trigger_clients_ht, node);
1529 free(client_list);
1530
1531 /* Remove trigger from triggers_ht. */
1532 trigger_ht_element = caa_container_of(triggers_ht_node,
1533 struct lttng_trigger_ht_element, node);
1534 cds_lfht_del(state->triggers_ht, triggers_ht_node);
1535
1536 condition = lttng_trigger_get_condition(trigger_ht_element->trigger);
1537 lttng_condition_destroy(condition);
1538 action = lttng_trigger_get_action(trigger_ht_element->trigger);
1539 lttng_action_destroy(action);
1540 lttng_trigger_destroy(trigger_ht_element->trigger);
1541 free(trigger_ht_element);
1542 end:
1543 rcu_read_unlock();
1544 if (_cmd_reply) {
1545 *_cmd_reply = cmd_reply;
1546 }
1547 return 0;
1548 }
1549
1550 /* Returns 0 on success, 1 on exit requested, negative value on error. */
1551 int handle_notification_thread_command(
1552 struct notification_thread_handle *handle,
1553 struct notification_thread_state *state)
1554 {
1555 int ret;
1556 uint64_t counter;
1557 struct notification_thread_command *cmd;
1558
1559 /* Read the event pipe to put it back into a quiescent state. */
1560 ret = read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
1561 sizeof(counter));
1562 if (ret == -1) {
1563 goto error;
1564 }
1565
1566 pthread_mutex_lock(&handle->cmd_queue.lock);
1567 cmd = cds_list_first_entry(&handle->cmd_queue.list,
1568 struct notification_thread_command, cmd_list_node);
1569 switch (cmd->type) {
1570 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
1571 DBG("[notification-thread] Received register trigger command");
1572 ret = handle_notification_thread_command_register_trigger(
1573 state, cmd->parameters.trigger,
1574 &cmd->reply_code);
1575 break;
1576 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
1577 DBG("[notification-thread] Received unregister trigger command");
1578 ret = handle_notification_thread_command_unregister_trigger(
1579 state, cmd->parameters.trigger,
1580 &cmd->reply_code);
1581 break;
1582 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
1583 DBG("[notification-thread] Received add channel command");
1584 ret = handle_notification_thread_command_add_channel(
1585 state,
1586 cmd->parameters.add_channel.session.name,
1587 cmd->parameters.add_channel.session.uid,
1588 cmd->parameters.add_channel.session.gid,
1589 cmd->parameters.add_channel.channel.name,
1590 cmd->parameters.add_channel.channel.domain,
1591 cmd->parameters.add_channel.channel.key,
1592 cmd->parameters.add_channel.channel.capacity,
1593 &cmd->reply_code);
1594 break;
1595 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
1596 DBG("[notification-thread] Received remove channel command");
1597 ret = handle_notification_thread_command_remove_channel(
1598 state, cmd->parameters.remove_channel.key,
1599 cmd->parameters.remove_channel.domain,
1600 &cmd->reply_code);
1601 break;
1602 case NOTIFICATION_COMMAND_TYPE_QUIT:
1603 DBG("[notification-thread] Received quit command");
1604 cmd->reply_code = LTTNG_OK;
1605 ret = 1;
1606 goto end;
1607 default:
1608 ERR("[notification-thread] Unknown internal command received");
1609 goto error_unlock;
1610 }
1611
1612 if (ret) {
1613 goto error_unlock;
1614 }
1615 end:
1616 cds_list_del(&cmd->cmd_list_node);
1617 lttng_waiter_wake_up(&cmd->reply_waiter);
1618 pthread_mutex_unlock(&handle->cmd_queue.lock);
1619 return ret;
1620 error_unlock:
1621 /* Wake-up and return a fatal error to the calling thread. */
1622 lttng_waiter_wake_up(&cmd->reply_waiter);
1623 pthread_mutex_unlock(&handle->cmd_queue.lock);
1624 cmd->reply_code = LTTNG_ERR_FATAL;
1625 error:
1626 /* Indicate a fatal error to the caller. */
1627 return -1;
1628 }
1629
1630 static
1631 unsigned long hash_client_socket(int socket)
1632 {
1633 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
1634 }
1635
1636 static
1637 int socket_set_non_blocking(int socket)
1638 {
1639 int ret, flags;
1640
1641 /* Set the pipe as non-blocking. */
1642 ret = fcntl(socket, F_GETFL, 0);
1643 if (ret == -1) {
1644 PERROR("fcntl get socket flags");
1645 goto end;
1646 }
1647 flags = ret;
1648
1649 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
1650 if (ret == -1) {
1651 PERROR("fcntl set O_NONBLOCK socket flag");
1652 goto end;
1653 }
1654 DBG("Client socket (fd = %i) set as non-blocking", socket);
1655 end:
1656 return ret;
1657 }
1658
1659 static
1660 int client_reset_inbound_state(struct notification_client *client)
1661 {
1662 int ret;
1663
1664 ret = lttng_dynamic_buffer_set_size(
1665 &client->communication.inbound.buffer, 0);
1666 assert(!ret);
1667
1668 client->communication.inbound.bytes_to_receive =
1669 sizeof(struct lttng_notification_channel_message);
1670 client->communication.inbound.msg_type =
1671 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
1672 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
1673 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
1674 ret = lttng_dynamic_buffer_set_size(
1675 &client->communication.inbound.buffer,
1676 client->communication.inbound.bytes_to_receive);
1677 return ret;
1678 }
1679
1680 int handle_notification_thread_client_connect(
1681 struct notification_thread_state *state)
1682 {
1683 int ret;
1684 struct notification_client *client;
1685
1686 DBG("[notification-thread] Handling new notification channel client connection");
1687
1688 client = zmalloc(sizeof(*client));
1689 if (!client) {
1690 /* Fatal error. */
1691 ret = -1;
1692 goto error;
1693 }
1694 CDS_INIT_LIST_HEAD(&client->condition_list);
1695 lttng_dynamic_buffer_init(&client->communication.inbound.buffer);
1696 lttng_dynamic_buffer_init(&client->communication.outbound.buffer);
1697 client->communication.inbound.expect_creds = true;
1698 ret = client_reset_inbound_state(client);
1699 if (ret) {
1700 ERR("[notification-thread] Failed to reset client communication's inbound state");
1701 ret = 0;
1702 goto error;
1703 }
1704
1705 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
1706 if (ret < 0) {
1707 ERR("[notification-thread] Failed to accept new notification channel client connection");
1708 ret = 0;
1709 goto error;
1710 }
1711
1712 client->socket = ret;
1713
1714 ret = socket_set_non_blocking(client->socket);
1715 if (ret) {
1716 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
1717 goto error;
1718 }
1719
1720 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
1721 if (ret < 0) {
1722 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
1723 ret = 0;
1724 goto error;
1725 }
1726
1727 ret = lttng_poll_add(&state->events, client->socket,
1728 LPOLLIN | LPOLLERR |
1729 LPOLLHUP | LPOLLRDHUP);
1730 if (ret < 0) {
1731 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
1732 ret = 0;
1733 goto error;
1734 }
1735 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
1736 client->socket);
1737
1738 rcu_read_lock();
1739 cds_lfht_add(state->client_socket_ht,
1740 hash_client_socket(client->socket),
1741 &client->client_socket_ht_node);
1742 rcu_read_unlock();
1743
1744 return ret;
1745 error:
1746 notification_client_destroy(client, state);
1747 return ret;
1748 }
1749
1750 int handle_notification_thread_client_disconnect(
1751 int client_socket,
1752 struct notification_thread_state *state)
1753 {
1754 int ret = 0;
1755 struct notification_client *client;
1756
1757 rcu_read_lock();
1758 DBG("[notification-thread] Closing client connection (socket fd = %i)",
1759 client_socket);
1760 client = get_client_from_socket(client_socket, state);
1761 if (!client) {
1762 /* Internal state corruption, fatal error. */
1763 ERR("[notification-thread] Unable to find client (socket fd = %i)",
1764 client_socket);
1765 ret = -1;
1766 goto end;
1767 }
1768
1769 ret = lttng_poll_del(&state->events, client_socket);
1770 if (ret) {
1771 ERR("[notification-thread] Failed to remove client socket from poll set");
1772 }
1773 cds_lfht_del(state->client_socket_ht,
1774 &client->client_socket_ht_node);
1775 notification_client_destroy(client, state);
1776 end:
1777 rcu_read_unlock();
1778 return ret;
1779 }
1780
1781 int handle_notification_thread_client_disconnect_all(
1782 struct notification_thread_state *state)
1783 {
1784 struct cds_lfht_iter iter;
1785 struct notification_client *client;
1786 bool error_encoutered = false;
1787
1788 rcu_read_lock();
1789 DBG("[notification-thread] Closing all client connections");
1790 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1791 client_socket_ht_node) {
1792 int ret;
1793
1794 ret = handle_notification_thread_client_disconnect(
1795 client->socket, state);
1796 if (ret) {
1797 error_encoutered = true;
1798 }
1799 }
1800 rcu_read_unlock();
1801 return error_encoutered ? 1 : 0;
1802 }
1803
1804 int handle_notification_thread_trigger_unregister_all(
1805 struct notification_thread_state *state)
1806 {
1807 bool error_occurred = false;
1808 struct cds_lfht_iter iter;
1809 struct lttng_trigger_ht_element *trigger_ht_element;
1810
1811 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1812 node) {
1813 int ret = handle_notification_thread_command_unregister_trigger(
1814 state, trigger_ht_element->trigger, NULL);
1815 if (ret) {
1816 error_occurred = true;
1817 }
1818 }
1819 return error_occurred ? -1 : 0;
1820 }
1821
1822 static
1823 int client_flush_outgoing_queue(struct notification_client *client,
1824 struct notification_thread_state *state)
1825 {
1826 ssize_t ret;
1827 size_t to_send_count;
1828
1829 assert(client->communication.outbound.buffer.size != 0);
1830 to_send_count = client->communication.outbound.buffer.size;
1831 DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
1832 client->socket);
1833
1834 ret = lttcomm_send_unix_sock_non_block(client->socket,
1835 client->communication.outbound.buffer.data,
1836 to_send_count);
1837 if ((ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) ||
1838 (ret > 0 && ret < to_send_count)) {
1839 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
1840 client->socket);
1841 to_send_count -= max(ret, 0);
1842
1843 memcpy(client->communication.outbound.buffer.data,
1844 client->communication.outbound.buffer.data +
1845 client->communication.outbound.buffer.size - to_send_count,
1846 to_send_count);
1847 ret = lttng_dynamic_buffer_set_size(
1848 &client->communication.outbound.buffer,
1849 to_send_count);
1850 if (ret) {
1851 goto error;
1852 }
1853
1854 /*
1855 * We want to be notified whenever there is buffer space
1856 * available to send the rest of the payload.
1857 */
1858 ret = lttng_poll_mod(&state->events, client->socket,
1859 CLIENT_POLL_MASK_IN_OUT);
1860 if (ret) {
1861 goto error;
1862 }
1863 } else if (ret < 0) {
1864 /* Generic error, disconnect the client. */
1865 ERR("[notification-thread] Failed to send flush outgoing queue, disconnecting client (socket fd = %i)",
1866 client->socket);
1867 ret = handle_notification_thread_client_disconnect(
1868 client->socket, state);
1869 if (ret) {
1870 goto error;
1871 }
1872 } else {
1873 /* No error and flushed the queue completely. */
1874 ret = lttng_dynamic_buffer_set_size(
1875 &client->communication.outbound.buffer, 0);
1876 if (ret) {
1877 goto error;
1878 }
1879 ret = lttng_poll_mod(&state->events, client->socket,
1880 CLIENT_POLL_MASK_IN);
1881 if (ret) {
1882 goto error;
1883 }
1884
1885 client->communication.outbound.queued_command_reply = false;
1886 client->communication.outbound.dropped_notification = false;
1887 }
1888
1889 return 0;
1890 error:
1891 return -1;
1892 }
1893
1894 static
1895 int client_send_command_reply(struct notification_client *client,
1896 struct notification_thread_state *state,
1897 enum lttng_notification_channel_status status)
1898 {
1899 int ret;
1900 struct lttng_notification_channel_command_reply reply = {
1901 .status = (int8_t) status,
1902 };
1903 struct lttng_notification_channel_message msg = {
1904 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
1905 .size = sizeof(reply),
1906 };
1907 char buffer[sizeof(msg) + sizeof(reply)];
1908
1909 if (client->communication.outbound.queued_command_reply) {
1910 /* Protocol error. */
1911 goto error;
1912 }
1913
1914 memcpy(buffer, &msg, sizeof(msg));
1915 memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
1916 DBG("[notification-thread] Send command reply (%i)", (int) status);
1917
1918 /* Enqueue buffer to outgoing queue and flush it. */
1919 ret = lttng_dynamic_buffer_append(
1920 &client->communication.outbound.buffer,
1921 buffer, sizeof(buffer));
1922 if (ret) {
1923 goto error;
1924 }
1925
1926 ret = client_flush_outgoing_queue(client, state);
1927 if (ret) {
1928 goto error;
1929 }
1930
1931 if (client->communication.outbound.buffer.size != 0) {
1932 /* Queue could not be emptied. */
1933 client->communication.outbound.queued_command_reply = true;
1934 }
1935
1936 return 0;
1937 error:
1938 return -1;
1939 }
1940
1941 static
1942 int client_dispatch_message(struct notification_client *client,
1943 struct notification_thread_state *state)
1944 {
1945 int ret = 0;
1946
1947 if (client->communication.inbound.msg_type !=
1948 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
1949 client->communication.inbound.msg_type !=
1950 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
1951 !client->validated) {
1952 WARN("[notification-thread] client attempted a command before handshake");
1953 ret = -1;
1954 goto end;
1955 }
1956
1957 switch (client->communication.inbound.msg_type) {
1958 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
1959 {
1960 /*
1961 * Receiving message header. The function will be called again
1962 * once the rest of the message as been received and can be
1963 * interpreted.
1964 */
1965 const struct lttng_notification_channel_message *msg;
1966
1967 assert(sizeof(*msg) ==
1968 client->communication.inbound.buffer.size);
1969 msg = (const struct lttng_notification_channel_message *)
1970 client->communication.inbound.buffer.data;
1971
1972 if (msg->size == 0 || msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
1973 ERR("[notification-thread] Invalid notification channel message: length = %u", msg->size);
1974 ret = -1;
1975 goto end;
1976 }
1977
1978 switch (msg->type) {
1979 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1980 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1981 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1982 break;
1983 default:
1984 ret = -1;
1985 ERR("[notification-thread] Invalid notification channel message: unexpected message type");
1986 goto end;
1987 }
1988
1989 client->communication.inbound.bytes_to_receive = msg->size;
1990 client->communication.inbound.msg_type =
1991 (enum lttng_notification_channel_message_type) msg->type;
1992 ret = lttng_dynamic_buffer_set_size(
1993 &client->communication.inbound.buffer, msg->size);
1994 if (ret) {
1995 goto end;
1996 }
1997 break;
1998 }
1999 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
2000 {
2001 struct lttng_notification_channel_command_handshake *handshake_client;
2002 struct lttng_notification_channel_command_handshake handshake_reply = {
2003 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
2004 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
2005 };
2006 struct lttng_notification_channel_message msg_header = {
2007 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
2008 .size = sizeof(handshake_reply),
2009 };
2010 enum lttng_notification_channel_status status =
2011 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
2012 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
2013
2014 memcpy(send_buffer, &msg_header, sizeof(msg_header));
2015 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
2016 sizeof(handshake_reply));
2017
2018 handshake_client =
2019 (struct lttng_notification_channel_command_handshake *)
2020 client->communication.inbound.buffer.data;
2021 client->major = handshake_client->major;
2022 client->minor = handshake_client->minor;
2023 if (!client->communication.inbound.creds_received) {
2024 ERR("[notification-thread] No credentials received from client");
2025 ret = -1;
2026 goto end;
2027 }
2028
2029 client->uid = LTTNG_SOCK_GET_UID_CRED(
2030 &client->communication.inbound.creds);
2031 client->gid = LTTNG_SOCK_GET_GID_CRED(
2032 &client->communication.inbound.creds);
2033 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
2034 client->uid, client->gid, (int) client->major,
2035 (int) client->minor);
2036
2037 if (handshake_client->major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
2038 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
2039 }
2040
2041 ret = lttng_dynamic_buffer_append(&client->communication.outbound.buffer,
2042 send_buffer, sizeof(send_buffer));
2043 if (ret) {
2044 ERR("[notification-thread] Failed to send protocol version to notification channel client");
2045 goto end;
2046 }
2047
2048 ret = client_flush_outgoing_queue(client, state);
2049 if (ret) {
2050 goto end;
2051 }
2052
2053 ret = client_send_command_reply(client, state, status);
2054 if (ret) {
2055 ERR("[notification-thread] Failed to send reply to notification channel client");
2056 goto end;
2057 }
2058
2059 /* Set reception state to receive the next message header. */
2060 ret = client_reset_inbound_state(client);
2061 if (ret) {
2062 ERR("[notification-thread] Failed to reset client communication's inbound state");
2063 goto end;
2064 }
2065 client->validated = true;
2066 break;
2067 }
2068 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
2069 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
2070 {
2071 struct lttng_condition *condition;
2072 enum lttng_notification_channel_status status =
2073 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
2074 const struct lttng_buffer_view condition_view =
2075 lttng_buffer_view_from_dynamic_buffer(
2076 &client->communication.inbound.buffer,
2077 0, -1);
2078 size_t expected_condition_size =
2079 client->communication.inbound.buffer.size;
2080
2081 ret = lttng_condition_create_from_buffer(&condition_view,
2082 &condition);
2083 if (ret != expected_condition_size) {
2084 ERR("[notification-thread] Malformed condition received from client");
2085 goto end;
2086 }
2087
2088 if (client->communication.inbound.msg_type ==
2089 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
2090 ret = notification_thread_client_subscribe(client,
2091 condition, state, &status);
2092 } else {
2093 ret = notification_thread_client_unsubscribe(client,
2094 condition, state, &status);
2095 }
2096 if (ret) {
2097 goto end;
2098 }
2099
2100 ret = client_send_command_reply(client, state, status);
2101 if (ret) {
2102 ERR("[notification-thread] Failed to send reply to notification channel client");
2103 goto end;
2104 }
2105
2106 /* Set reception state to receive the next message header. */
2107 ret = client_reset_inbound_state(client);
2108 if (ret) {
2109 ERR("[notification-thread] Failed to reset client communication's inbound state");
2110 goto end;
2111 }
2112 break;
2113 }
2114 default:
2115 abort();
2116 }
2117 end:
2118 return ret;
2119 }
2120
2121 /* Incoming data from client. */
2122 int handle_notification_thread_client_in(
2123 struct notification_thread_state *state, int socket)
2124 {
2125 int ret = 0;
2126 struct notification_client *client;
2127 ssize_t recv_ret;
2128 size_t offset;
2129
2130 client = get_client_from_socket(socket, state);
2131 if (!client) {
2132 /* Internal error, abort. */
2133 ret = -1;
2134 goto end;
2135 }
2136
2137 offset = client->communication.inbound.buffer.size -
2138 client->communication.inbound.bytes_to_receive;
2139 if (client->communication.inbound.expect_creds) {
2140 recv_ret = lttcomm_recv_creds_unix_sock(socket,
2141 client->communication.inbound.buffer.data + offset,
2142 client->communication.inbound.bytes_to_receive,
2143 &client->communication.inbound.creds);
2144 if (recv_ret > 0) {
2145 client->communication.inbound.expect_creds = false;
2146 client->communication.inbound.creds_received = true;
2147 }
2148 } else {
2149 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
2150 client->communication.inbound.buffer.data + offset,
2151 client->communication.inbound.bytes_to_receive);
2152 }
2153 if (recv_ret < 0) {
2154 goto error_disconnect_client;
2155 }
2156
2157 client->communication.inbound.bytes_to_receive -= recv_ret;
2158 if (client->communication.inbound.bytes_to_receive == 0) {
2159 ret = client_dispatch_message(client, state);
2160 if (ret) {
2161 /*
2162 * Only returns an error if this client must be
2163 * disconnected.
2164 */
2165 goto error_disconnect_client;
2166 }
2167 } else {
2168 goto end;
2169 }
2170 end:
2171 return ret;
2172 error_disconnect_client:
2173 ret = handle_notification_thread_client_disconnect(socket, state);
2174 return ret;
2175 }
2176
2177 /* Client ready to receive outgoing data. */
2178 int handle_notification_thread_client_out(
2179 struct notification_thread_state *state, int socket)
2180 {
2181 int ret;
2182 struct notification_client *client;
2183
2184 client = get_client_from_socket(socket, state);
2185 if (!client) {
2186 /* Internal error, abort. */
2187 ret = -1;
2188 goto end;
2189 }
2190
2191 ret = client_flush_outgoing_queue(client, state);
2192 if (ret) {
2193 goto end;
2194 }
2195 end:
2196 return ret;
2197 }
2198
2199 static
2200 bool evaluate_buffer_usage_condition(const struct lttng_condition *condition,
2201 const struct channel_state_sample *sample,
2202 uint64_t buffer_capacity)
2203 {
2204 bool result = false;
2205 uint64_t threshold;
2206 enum lttng_condition_type condition_type;
2207 const struct lttng_condition_buffer_usage *use_condition = container_of(
2208 condition, struct lttng_condition_buffer_usage,
2209 parent);
2210
2211 if (use_condition->threshold_bytes.set) {
2212 threshold = use_condition->threshold_bytes.value;
2213 } else {
2214 /*
2215 * Threshold was expressed as a ratio.
2216 *
2217 * TODO the threshold (in bytes) of conditions expressed
2218 * as a ratio of total buffer size could be cached to
2219 * forego this double-multiplication or it could be performed
2220 * as fixed-point math.
2221 *
2222 * Note that caching should accomodate the case where the
2223 * condition applies to multiple channels (i.e. don't assume
2224 * that all channels matching my_chann* have the same size...)
2225 */
2226 threshold = (uint64_t) (use_condition->threshold_ratio.value *
2227 (double) buffer_capacity);
2228 }
2229
2230 condition_type = lttng_condition_get_type(condition);
2231 if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
2232 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
2233 threshold, sample->highest_usage);
2234
2235 /*
2236 * The low condition should only be triggered once _all_ of the
2237 * streams in a channel have gone below the "low" threshold.
2238 */
2239 if (sample->highest_usage <= threshold) {
2240 result = true;
2241 }
2242 } else {
2243 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
2244 threshold, sample->highest_usage);
2245
2246 /*
2247 * For high buffer usage scenarios, we want to trigger whenever
2248 * _any_ of the streams has reached the "high" threshold.
2249 */
2250 if (sample->highest_usage >= threshold) {
2251 result = true;
2252 }
2253 }
2254
2255 return result;
2256 }
2257
2258 static
2259 bool evaluate_session_consumed_size_condition(
2260 const struct lttng_condition *condition,
2261 uint64_t session_consumed_size)
2262 {
2263 uint64_t threshold;
2264 const struct lttng_condition_session_consumed_size *size_condition =
2265 container_of(condition,
2266 struct lttng_condition_session_consumed_size,
2267 parent);
2268
2269 threshold = size_condition->consumed_threshold_bytes.value;
2270 DBG("[notification-thread] Session consumed size condition being evaluated: threshold = %" PRIu64 ", current size = %" PRIu64,
2271 threshold, session_consumed_size);
2272 return session_consumed_size >= threshold;
2273 }
2274
2275 static
2276 int evaluate_condition(const struct lttng_condition *condition,
2277 struct lttng_evaluation **evaluation,
2278 const struct notification_thread_state *state,
2279 const struct channel_state_sample *previous_sample,
2280 const struct channel_state_sample *latest_sample,
2281 uint64_t previous_session_consumed_total,
2282 uint64_t latest_session_consumed_total,
2283 struct channel_info *channel_info)
2284 {
2285 int ret = 0;
2286 enum lttng_condition_type condition_type;
2287 const bool previous_sample_available = !!previous_sample;
2288 bool previous_sample_result = false;
2289 bool latest_sample_result;
2290
2291 condition_type = lttng_condition_get_type(condition);
2292
2293 switch (condition_type) {
2294 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2295 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2296 if (caa_likely(previous_sample_available)) {
2297 previous_sample_result =
2298 evaluate_buffer_usage_condition(condition,
2299 previous_sample, channel_info->capacity);
2300 }
2301 latest_sample_result = evaluate_buffer_usage_condition(
2302 condition, latest_sample,
2303 channel_info->capacity);
2304 break;
2305 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
2306 if (caa_likely(previous_sample_available)) {
2307 previous_sample_result =
2308 evaluate_session_consumed_size_condition(
2309 condition,
2310 previous_session_consumed_total);
2311 }
2312 latest_sample_result =
2313 evaluate_session_consumed_size_condition(
2314 condition,
2315 latest_session_consumed_total);
2316 break;
2317 default:
2318 /* Unknown condition type; internal error. */
2319 abort();
2320 }
2321
2322 if (!latest_sample_result ||
2323 (previous_sample_result == latest_sample_result)) {
2324 /*
2325 * Only trigger on a condition evaluation transition.
2326 *
2327 * NOTE: This edge-triggered logic may not be appropriate for
2328 * future condition types.
2329 */
2330 goto end;
2331 }
2332
2333 if (!evaluation || !latest_sample_result) {
2334 goto end;
2335 }
2336
2337 switch (condition_type) {
2338 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2339 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2340 *evaluation = lttng_evaluation_buffer_usage_create(
2341 condition_type,
2342 latest_sample->highest_usage,
2343 channel_info->capacity);
2344 break;
2345 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
2346 *evaluation = lttng_evaluation_session_consumed_size_create(
2347 condition_type,
2348 latest_session_consumed_total);
2349 break;
2350 default:
2351 abort();
2352 }
2353
2354 if (!*evaluation) {
2355 ret = -1;
2356 goto end;
2357 }
2358 end:
2359 return ret;
2360 }
2361
2362 static
2363 int client_enqueue_dropped_notification(struct notification_client *client,
2364 struct notification_thread_state *state)
2365 {
2366 int ret;
2367 struct lttng_notification_channel_message msg = {
2368 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
2369 .size = 0,
2370 };
2371
2372 ret = lttng_dynamic_buffer_append(
2373 &client->communication.outbound.buffer, &msg,
2374 sizeof(msg));
2375 return ret;
2376 }
2377
2378 static
2379 int send_evaluation_to_clients(struct lttng_trigger *trigger,
2380 struct lttng_evaluation *evaluation,
2381 struct notification_client_list* client_list,
2382 struct notification_thread_state *state,
2383 uid_t channel_uid, gid_t channel_gid)
2384 {
2385 int ret = 0;
2386 struct lttng_dynamic_buffer msg_buffer;
2387 struct notification_client_list_element *client_list_element, *tmp;
2388 struct lttng_notification *notification;
2389 struct lttng_condition *condition;
2390 ssize_t expected_notification_size, notification_size;
2391 struct lttng_notification_channel_message msg;
2392
2393 lttng_dynamic_buffer_init(&msg_buffer);
2394
2395 condition = lttng_trigger_get_condition(trigger);
2396 assert(condition);
2397
2398 notification = lttng_notification_create(condition, evaluation);
2399 if (!notification) {
2400 ret = -1;
2401 goto end;
2402 }
2403
2404 expected_notification_size = lttng_notification_serialize(notification,
2405 NULL);
2406 if (expected_notification_size < 0) {
2407 ERR("[notification-thread] Failed to get size of serialized notification");
2408 ret = -1;
2409 goto end;
2410 }
2411
2412 msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION;
2413 msg.size = (uint32_t) expected_notification_size;
2414 ret = lttng_dynamic_buffer_append(&msg_buffer, &msg, sizeof(msg));
2415 if (ret) {
2416 goto end;
2417 }
2418
2419 ret = lttng_dynamic_buffer_set_size(&msg_buffer,
2420 msg_buffer.size + expected_notification_size);
2421 if (ret) {
2422 goto end;
2423 }
2424
2425 notification_size = lttng_notification_serialize(notification,
2426 msg_buffer.data + sizeof(msg));
2427 if (notification_size != expected_notification_size) {
2428 ERR("[notification-thread] Failed to serialize notification");
2429 ret = -1;
2430 goto end;
2431 }
2432
2433 cds_list_for_each_entry_safe(client_list_element, tmp,
2434 &client_list->list, node) {
2435 struct notification_client *client =
2436 client_list_element->client;
2437
2438 if (client->uid != channel_uid && client->gid != channel_gid &&
2439 client->uid != 0) {
2440 /* Client is not allowed to monitor this channel. */
2441 DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this channel");
2442 continue;
2443 }
2444
2445 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
2446 client->socket, msg_buffer.size);
2447 if (client->communication.outbound.buffer.size) {
2448 /*
2449 * Outgoing data is already buffered for this client;
2450 * drop the notification and enqueue a "dropped
2451 * notification" message if this is the first dropped
2452 * notification since the socket spilled-over to the
2453 * queue.
2454 */
2455 DBG("[notification-thread] Dropping notification addressed to client (socket fd = %i)",
2456 client->socket);
2457 if (!client->communication.outbound.dropped_notification) {
2458 client->communication.outbound.dropped_notification = true;
2459 ret = client_enqueue_dropped_notification(
2460 client, state);
2461 if (ret) {
2462 goto end;
2463 }
2464 }
2465 continue;
2466 }
2467
2468 ret = lttng_dynamic_buffer_append_buffer(
2469 &client->communication.outbound.buffer,
2470 &msg_buffer);
2471 if (ret) {
2472 goto end;
2473 }
2474
2475 ret = client_flush_outgoing_queue(client, state);
2476 if (ret) {
2477 goto end;
2478 }
2479 }
2480 ret = 0;
2481 end:
2482 lttng_notification_destroy(notification);
2483 lttng_dynamic_buffer_reset(&msg_buffer);
2484 return ret;
2485 }
2486
2487 int handle_notification_thread_channel_sample(
2488 struct notification_thread_state *state, int pipe,
2489 enum lttng_domain_type domain)
2490 {
2491 int ret = 0;
2492 struct lttcomm_consumer_channel_monitor_msg sample_msg;
2493 struct channel_info *channel_info;
2494 struct cds_lfht_node *node;
2495 struct cds_lfht_iter iter;
2496 struct lttng_channel_trigger_list *trigger_list;
2497 struct lttng_trigger_list_element *trigger_list_element;
2498 bool previous_sample_available = false;
2499 struct channel_state_sample previous_sample, latest_sample;
2500 uint64_t previous_session_consumed_total, latest_session_consumed_total;
2501
2502 /*
2503 * The monitoring pipe only holds messages smaller than PIPE_BUF,
2504 * ensuring that read/write of sampling messages are atomic.
2505 */
2506 ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
2507 if (ret != sizeof(sample_msg)) {
2508 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
2509 pipe);
2510 ret = -1;
2511 goto end;
2512 }
2513
2514 ret = 0;
2515 latest_sample.key.key = sample_msg.key;
2516 latest_sample.key.domain = domain;
2517 latest_sample.highest_usage = sample_msg.highest;
2518 latest_sample.lowest_usage = sample_msg.lowest;
2519 latest_sample.channel_total_consumed = sample_msg.total_consumed;
2520
2521 rcu_read_lock();
2522
2523 /* Retrieve the channel's informations */
2524 cds_lfht_lookup(state->channels_ht,
2525 hash_channel_key(&latest_sample.key),
2526 match_channel_info,
2527 &latest_sample.key,
2528 &iter);
2529 node = cds_lfht_iter_get_node(&iter);
2530 if (caa_unlikely(!node)) {
2531 /*
2532 * Not an error since the consumer can push a sample to the pipe
2533 * and the rest of the session daemon could notify us of the
2534 * channel's destruction before we get a chance to process that
2535 * sample.
2536 */
2537 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
2538 latest_sample.key.key,
2539 domain == LTTNG_DOMAIN_KERNEL ? "kernel" :
2540 "user space");
2541 goto end_unlock;
2542 }
2543 channel_info = caa_container_of(node, struct channel_info,
2544 channels_ht_node);
2545 DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64", total consumed = %" PRIu64")",
2546 channel_info->name,
2547 latest_sample.key.key,
2548 channel_info->session_info->name,
2549 latest_sample.highest_usage,
2550 latest_sample.lowest_usage,
2551 latest_sample.channel_total_consumed);
2552
2553 previous_session_consumed_total =
2554 channel_info->session_info->consumed_data_size;
2555
2556 /* Retrieve the channel's last sample, if it exists, and update it. */
2557 cds_lfht_lookup(state->channel_state_ht,
2558 hash_channel_key(&latest_sample.key),
2559 match_channel_state_sample,
2560 &latest_sample.key,
2561 &iter);
2562 node = cds_lfht_iter_get_node(&iter);
2563 if (caa_likely(node)) {
2564 struct channel_state_sample *stored_sample;
2565
2566 /* Update the sample stored. */
2567 stored_sample = caa_container_of(node,
2568 struct channel_state_sample,
2569 channel_state_ht_node);
2570
2571 memcpy(&previous_sample, stored_sample,
2572 sizeof(previous_sample));
2573 stored_sample->highest_usage = latest_sample.highest_usage;
2574 stored_sample->lowest_usage = latest_sample.lowest_usage;
2575 stored_sample->channel_total_consumed = latest_sample.channel_total_consumed;
2576 previous_sample_available = true;
2577
2578 latest_session_consumed_total =
2579 previous_session_consumed_total +
2580 (latest_sample.channel_total_consumed - previous_sample.channel_total_consumed);
2581 } else {
2582 /*
2583 * This is the channel's first sample, allocate space for and
2584 * store the new sample.
2585 */
2586 struct channel_state_sample *stored_sample;
2587
2588 stored_sample = zmalloc(sizeof(*stored_sample));
2589 if (!stored_sample) {
2590 ret = -1;
2591 goto end_unlock;
2592 }
2593
2594 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
2595 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
2596 cds_lfht_add(state->channel_state_ht,
2597 hash_channel_key(&stored_sample->key),
2598 &stored_sample->channel_state_ht_node);
2599
2600 latest_session_consumed_total =
2601 previous_session_consumed_total +
2602 latest_sample.channel_total_consumed;
2603 }
2604
2605 channel_info->session_info->consumed_data_size =
2606 latest_session_consumed_total;
2607
2608 /* Find triggers associated with this channel. */
2609 cds_lfht_lookup(state->channel_triggers_ht,
2610 hash_channel_key(&latest_sample.key),
2611 match_channel_trigger_list,
2612 &latest_sample.key,
2613 &iter);
2614 node = cds_lfht_iter_get_node(&iter);
2615 if (caa_likely(!node)) {
2616 goto end_unlock;
2617 }
2618
2619 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
2620 channel_triggers_ht_node);
2621 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
2622 node) {
2623 struct lttng_condition *condition;
2624 struct lttng_action *action;
2625 struct lttng_trigger *trigger;
2626 struct notification_client_list *client_list;
2627 struct lttng_evaluation *evaluation = NULL;
2628
2629 trigger = trigger_list_element->trigger;
2630 condition = lttng_trigger_get_condition(trigger);
2631 assert(condition);
2632 action = lttng_trigger_get_action(trigger);
2633
2634 /* Notify actions are the only type currently supported. */
2635 assert(lttng_action_get_type(action) ==
2636 LTTNG_ACTION_TYPE_NOTIFY);
2637
2638 /*
2639 * Check if any client is subscribed to the result of this
2640 * evaluation.
2641 */
2642 cds_lfht_lookup(state->notification_trigger_clients_ht,
2643 lttng_condition_hash(condition),
2644 match_client_list,
2645 trigger,
2646 &iter);
2647 node = cds_lfht_iter_get_node(&iter);
2648 assert(node);
2649
2650 client_list = caa_container_of(node,
2651 struct notification_client_list,
2652 notification_trigger_ht_node);
2653 if (cds_list_empty(&client_list->list)) {
2654 /*
2655 * No clients interested in the evaluation's result,
2656 * skip it.
2657 */
2658 continue;
2659 }
2660
2661 ret = evaluate_condition(condition, &evaluation, state,
2662 previous_sample_available ? &previous_sample : NULL,
2663 &latest_sample,
2664 previous_session_consumed_total,
2665 latest_session_consumed_total,
2666 channel_info);
2667 if (caa_unlikely(ret)) {
2668 goto end_unlock;
2669 }
2670
2671 if (caa_likely(!evaluation)) {
2672 continue;
2673 }
2674
2675 /* Dispatch evaluation result to all clients. */
2676 ret = send_evaluation_to_clients(trigger_list_element->trigger,
2677 evaluation, client_list, state,
2678 channel_info->session_info->uid,
2679 channel_info->session_info->gid);
2680 lttng_evaluation_destroy(evaluation);
2681 if (caa_unlikely(ret)) {
2682 goto end_unlock;
2683 }
2684 }
2685 end_unlock:
2686 rcu_read_unlock();
2687 end:
2688 return ret;
2689 }
This page took 0.150116 seconds and 4 git commands to generate.