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