Use lttng_trigger_is_equal when iterating over the trigger ht
[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 * SPDX-License-Identifier: GPL-2.0-only
5 *
6 */
7
8 #include "lttng/action/action.h"
9 #include "lttng/trigger/trigger-internal.h"
10 #define _LGPL_SOURCE
11 #include <urcu.h>
12 #include <urcu/rculfhash.h>
13
14 #include <common/defaults.h>
15 #include <common/error.h>
16 #include <common/futex.h>
17 #include <common/unix.h>
18 #include <common/dynamic-buffer.h>
19 #include <common/hashtable/utils.h>
20 #include <common/sessiond-comm/sessiond-comm.h>
21 #include <common/macros.h>
22 #include <lttng/condition/condition.h>
23 #include <lttng/action/action-internal.h>
24 #include <lttng/notification/notification-internal.h>
25 #include <lttng/condition/condition-internal.h>
26 #include <lttng/condition/buffer-usage-internal.h>
27 #include <lttng/condition/session-consumed-size-internal.h>
28 #include <lttng/condition/session-rotation-internal.h>
29 #include <lttng/notification/channel-internal.h>
30 #include <lttng/trigger/trigger-internal.h>
31
32 #include <time.h>
33 #include <unistd.h>
34 #include <assert.h>
35 #include <inttypes.h>
36 #include <fcntl.h>
37
38 #include "notification-thread.h"
39 #include "notification-thread-events.h"
40 #include "notification-thread-commands.h"
41 #include "lttng-sessiond.h"
42 #include "kernel.h"
43
44 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
45 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
46
47 enum lttng_object_type {
48 LTTNG_OBJECT_TYPE_UNKNOWN,
49 LTTNG_OBJECT_TYPE_NONE,
50 LTTNG_OBJECT_TYPE_CHANNEL,
51 LTTNG_OBJECT_TYPE_SESSION,
52 };
53
54 struct lttng_trigger_list_element {
55 /* No ownership of the trigger object is assumed. */
56 struct lttng_trigger *trigger;
57 struct cds_list_head node;
58 };
59
60 struct lttng_channel_trigger_list {
61 struct channel_key channel_key;
62 /* List of struct lttng_trigger_list_element. */
63 struct cds_list_head list;
64 /* Node in the channel_triggers_ht */
65 struct cds_lfht_node channel_triggers_ht_node;
66 /* call_rcu delayed reclaim. */
67 struct rcu_head rcu_node;
68 };
69
70 /*
71 * List of triggers applying to a given session.
72 *
73 * See:
74 * - lttng_session_trigger_list_create()
75 * - lttng_session_trigger_list_build()
76 * - lttng_session_trigger_list_destroy()
77 * - lttng_session_trigger_list_add()
78 */
79 struct lttng_session_trigger_list {
80 /*
81 * Not owned by this; points to the session_info structure's
82 * session name.
83 */
84 const char *session_name;
85 /* List of struct lttng_trigger_list_element. */
86 struct cds_list_head list;
87 /* Node in the session_triggers_ht */
88 struct cds_lfht_node session_triggers_ht_node;
89 /*
90 * Weak reference to the notification system's session triggers
91 * hashtable.
92 *
93 * The session trigger list structure structure is owned by
94 * the session's session_info.
95 *
96 * The session_info is kept alive the the channel_infos holding a
97 * reference to it (reference counting). When those channels are
98 * destroyed (at runtime or on teardown), the reference they hold
99 * to the session_info are released. On destruction of session_info,
100 * session_info_destroy() will remove the list of triggers applying
101 * to this session from the notification system's state.
102 *
103 * This implies that the session_triggers_ht must be destroyed
104 * after the channels.
105 */
106 struct cds_lfht *session_triggers_ht;
107 /* Used for delayed RCU reclaim. */
108 struct rcu_head rcu_node;
109 };
110
111 struct lttng_trigger_ht_element {
112 struct lttng_trigger *trigger;
113 struct cds_lfht_node node;
114 struct cds_lfht_node node_by_name_uid;
115 /* call_rcu delayed reclaim. */
116 struct rcu_head rcu_node;
117 };
118
119 struct lttng_condition_list_element {
120 struct lttng_condition *condition;
121 struct cds_list_head node;
122 };
123
124 struct channel_state_sample {
125 struct channel_key key;
126 struct cds_lfht_node channel_state_ht_node;
127 uint64_t highest_usage;
128 uint64_t lowest_usage;
129 uint64_t channel_total_consumed;
130 /* call_rcu delayed reclaim. */
131 struct rcu_head rcu_node;
132 };
133
134 static unsigned long hash_channel_key(struct channel_key *key);
135 static int evaluate_buffer_condition(const struct lttng_condition *condition,
136 struct lttng_evaluation **evaluation,
137 const struct notification_thread_state *state,
138 const struct channel_state_sample *previous_sample,
139 const struct channel_state_sample *latest_sample,
140 uint64_t previous_session_consumed_total,
141 uint64_t latest_session_consumed_total,
142 struct channel_info *channel_info);
143 static
144 int send_evaluation_to_clients(const struct lttng_trigger *trigger,
145 const struct lttng_evaluation *evaluation,
146 struct notification_client_list *client_list,
147 struct notification_thread_state *state,
148 uid_t channel_uid, gid_t channel_gid);
149
150
151 /* session_info API */
152 static
153 void session_info_destroy(void *_data);
154 static
155 void session_info_get(struct session_info *session_info);
156 static
157 void session_info_put(struct session_info *session_info);
158 static
159 struct session_info *session_info_create(const char *name,
160 uid_t uid, gid_t gid,
161 struct lttng_session_trigger_list *trigger_list,
162 struct cds_lfht *sessions_ht);
163 static
164 void session_info_add_channel(struct session_info *session_info,
165 struct channel_info *channel_info);
166 static
167 void session_info_remove_channel(struct session_info *session_info,
168 struct channel_info *channel_info);
169
170 /* lttng_session_trigger_list API */
171 static
172 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
173 const char *session_name,
174 struct cds_lfht *session_triggers_ht);
175 static
176 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
177 const struct notification_thread_state *state,
178 const char *session_name);
179 static
180 void lttng_session_trigger_list_destroy(
181 struct lttng_session_trigger_list *list);
182 static
183 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
184 struct lttng_trigger *trigger);
185
186 static
187 int client_handle_transmission_status(
188 struct notification_client *client,
189 enum client_transmission_status transmission_status,
190 struct notification_thread_state *state);
191
192 static
193 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node);
194
195 static
196 int match_client_socket(struct cds_lfht_node *node, const void *key)
197 {
198 /* This double-cast is intended to supress pointer-to-cast warning. */
199 const int socket = (int) (intptr_t) key;
200 const struct notification_client *client = caa_container_of(node,
201 struct notification_client, client_socket_ht_node);
202
203 return client->socket == socket;
204 }
205
206 static
207 int match_client_id(struct cds_lfht_node *node, const void *key)
208 {
209 /* This double-cast is intended to supress pointer-to-cast warning. */
210 const notification_client_id id = *((notification_client_id *) key);
211 const struct notification_client *client = caa_container_of(
212 node, struct notification_client, client_id_ht_node);
213
214 return client->id == id;
215 }
216
217 static
218 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
219 {
220 struct channel_key *channel_key = (struct channel_key *) key;
221 struct lttng_channel_trigger_list *trigger_list;
222
223 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
224 channel_triggers_ht_node);
225
226 return !!((channel_key->key == trigger_list->channel_key.key) &&
227 (channel_key->domain == trigger_list->channel_key.domain));
228 }
229
230 static
231 int match_session_trigger_list(struct cds_lfht_node *node, const void *key)
232 {
233 const char *session_name = (const char *) key;
234 struct lttng_session_trigger_list *trigger_list;
235
236 trigger_list = caa_container_of(node, struct lttng_session_trigger_list,
237 session_triggers_ht_node);
238
239 return !!(strcmp(trigger_list->session_name, session_name) == 0);
240 }
241
242 static
243 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
244 {
245 struct channel_key *channel_key = (struct channel_key *) key;
246 struct channel_state_sample *sample;
247
248 sample = caa_container_of(node, struct channel_state_sample,
249 channel_state_ht_node);
250
251 return !!((channel_key->key == sample->key.key) &&
252 (channel_key->domain == sample->key.domain));
253 }
254
255 static
256 int match_channel_info(struct cds_lfht_node *node, const void *key)
257 {
258 struct channel_key *channel_key = (struct channel_key *) key;
259 struct channel_info *channel_info;
260
261 channel_info = caa_container_of(node, struct channel_info,
262 channels_ht_node);
263
264 return !!((channel_key->key == channel_info->key.key) &&
265 (channel_key->domain == channel_info->key.domain));
266 }
267
268 static
269 int match_trigger(struct cds_lfht_node *node, const void *key)
270 {
271 struct lttng_trigger *trigger_key = (struct lttng_trigger *) key;
272 struct lttng_trigger_ht_element *trigger_ht_element;
273
274 trigger_ht_element = caa_container_of(node, struct lttng_trigger_ht_element,
275 node);
276
277 return !!lttng_trigger_is_equal(trigger_key, trigger_ht_element->trigger);
278 }
279
280 static
281 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
282 {
283 struct lttng_condition *condition_key = (struct lttng_condition *) key;
284 struct notification_client_list *client_list;
285 const struct lttng_condition *condition;
286
287 assert(condition_key);
288
289 client_list = caa_container_of(node, struct notification_client_list,
290 notification_trigger_clients_ht_node);
291 condition = lttng_trigger_get_const_condition(client_list->trigger);
292
293 return !!lttng_condition_is_equal(condition_key, condition);
294 }
295
296 static
297 int match_session(struct cds_lfht_node *node, const void *key)
298 {
299 const char *name = key;
300 struct session_info *session_info = caa_container_of(
301 node, struct session_info, sessions_ht_node);
302
303 return !strcmp(session_info->name, name);
304 }
305
306 /*
307 * Match trigger based on name and credentials only.
308 * Name duplication is NOT allowed for the same uid.
309 */
310 static
311 int match_trigger_by_name_uid(struct cds_lfht_node *node,
312 const void *key)
313 {
314 bool match = false;
315 const char *name;
316 const char *key_name;
317 enum lttng_trigger_status status;
318 const struct lttng_credentials *key_creds;
319 const struct lttng_credentials *node_creds;
320 const struct lttng_trigger *trigger_key =
321 (const struct lttng_trigger *) key;
322 const struct lttng_trigger_ht_element *trigger_ht_element =
323 caa_container_of(node,
324 struct lttng_trigger_ht_element,
325 node_by_name_uid);
326
327 status = lttng_trigger_get_name(trigger_ht_element->trigger, &name);
328 assert(status == LTTNG_TRIGGER_STATUS_OK);
329
330 status = lttng_trigger_get_name(trigger_key, &key_name);
331 assert(status == LTTNG_TRIGGER_STATUS_OK);
332
333 /* Compare the names. */
334 if (strcmp(name, key_name) != 0) {
335 goto end;
336 }
337
338 /* Compare the owners' UIDs. */
339 key_creds = lttng_trigger_get_credentials(trigger_key);
340 node_creds = lttng_trigger_get_credentials(trigger_ht_element->trigger);
341
342 match = lttng_credentials_is_equal_uid(key_creds, node_creds);
343
344 end:
345 return match;
346 }
347
348 /*
349 * Hash trigger based on name and credentials only.
350 */
351 static
352 unsigned long hash_trigger_by_name_uid(const struct lttng_trigger *trigger)
353 {
354 unsigned long hash = 0;
355 const struct lttng_credentials *trigger_creds;
356 const char *trigger_name;
357 enum lttng_trigger_status status;
358
359 status = lttng_trigger_get_name(trigger, &trigger_name);
360 if (status == LTTNG_TRIGGER_STATUS_OK) {
361 hash = hash_key_str(trigger_name, lttng_ht_seed);
362 }
363
364 trigger_creds = lttng_trigger_get_credentials(trigger);
365 hash ^= hash_key_ulong((void *) (unsigned long) LTTNG_OPTIONAL_GET(trigger_creds->uid),
366 lttng_ht_seed);
367
368 return hash;
369 }
370
371 static
372 unsigned long lttng_condition_buffer_usage_hash(
373 const struct lttng_condition *_condition)
374 {
375 unsigned long hash;
376 unsigned long condition_type;
377 struct lttng_condition_buffer_usage *condition;
378
379 condition = container_of(_condition,
380 struct lttng_condition_buffer_usage, parent);
381
382 condition_type = (unsigned long) condition->parent.type;
383 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
384 if (condition->session_name) {
385 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
386 }
387 if (condition->channel_name) {
388 hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
389 }
390 if (condition->domain.set) {
391 hash ^= hash_key_ulong(
392 (void *) condition->domain.type,
393 lttng_ht_seed);
394 }
395 if (condition->threshold_ratio.set) {
396 uint64_t val;
397
398 val = condition->threshold_ratio.value * (double) UINT32_MAX;
399 hash ^= hash_key_u64(&val, lttng_ht_seed);
400 } else if (condition->threshold_bytes.set) {
401 uint64_t val;
402
403 val = condition->threshold_bytes.value;
404 hash ^= hash_key_u64(&val, lttng_ht_seed);
405 }
406 return hash;
407 }
408
409 static
410 unsigned long lttng_condition_session_consumed_size_hash(
411 const struct lttng_condition *_condition)
412 {
413 unsigned long hash;
414 unsigned long condition_type =
415 (unsigned long) LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE;
416 struct lttng_condition_session_consumed_size *condition;
417 uint64_t val;
418
419 condition = container_of(_condition,
420 struct lttng_condition_session_consumed_size, parent);
421
422 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
423 if (condition->session_name) {
424 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
425 }
426 val = condition->consumed_threshold_bytes.value;
427 hash ^= hash_key_u64(&val, lttng_ht_seed);
428 return hash;
429 }
430
431 static
432 unsigned long lttng_condition_session_rotation_hash(
433 const struct lttng_condition *_condition)
434 {
435 unsigned long hash, condition_type;
436 struct lttng_condition_session_rotation *condition;
437
438 condition = container_of(_condition,
439 struct lttng_condition_session_rotation, parent);
440 condition_type = (unsigned long) condition->parent.type;
441 hash = hash_key_ulong((void *) condition_type, lttng_ht_seed);
442 assert(condition->session_name);
443 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
444 return hash;
445 }
446
447 /*
448 * The lttng_condition hashing code is kept in this file (rather than
449 * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
450 * don't want to link in liblttng-ctl.
451 */
452 static
453 unsigned long lttng_condition_hash(const struct lttng_condition *condition)
454 {
455 switch (condition->type) {
456 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
457 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
458 return lttng_condition_buffer_usage_hash(condition);
459 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
460 return lttng_condition_session_consumed_size_hash(condition);
461 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
462 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
463 return lttng_condition_session_rotation_hash(condition);
464 default:
465 ERR("[notification-thread] Unexpected condition type caught");
466 abort();
467 }
468 }
469
470 static
471 unsigned long hash_channel_key(struct channel_key *key)
472 {
473 unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed);
474 unsigned long domain_hash = hash_key_ulong(
475 (void *) (unsigned long) key->domain, lttng_ht_seed);
476
477 return key_hash ^ domain_hash;
478 }
479
480 static
481 unsigned long hash_client_socket(int socket)
482 {
483 return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
484 }
485
486 static
487 unsigned long hash_client_id(notification_client_id id)
488 {
489 return hash_key_u64(&id, lttng_ht_seed);
490 }
491
492 /*
493 * Get the type of object to which a given condition applies. Bindings let
494 * the notification system evaluate a trigger's condition when a given
495 * object's state is updated.
496 *
497 * For instance, a condition bound to a channel will be evaluated everytime
498 * the channel's state is changed by a channel monitoring sample.
499 */
500 static
501 enum lttng_object_type get_condition_binding_object(
502 const struct lttng_condition *condition)
503 {
504 switch (lttng_condition_get_type(condition)) {
505 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
506 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
507 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
508 return LTTNG_OBJECT_TYPE_CHANNEL;
509 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
510 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
511 return LTTNG_OBJECT_TYPE_SESSION;
512 default:
513 return LTTNG_OBJECT_TYPE_UNKNOWN;
514 }
515 }
516
517 static
518 void free_channel_info_rcu(struct rcu_head *node)
519 {
520 free(caa_container_of(node, struct channel_info, rcu_node));
521 }
522
523 static
524 void channel_info_destroy(struct channel_info *channel_info)
525 {
526 if (!channel_info) {
527 return;
528 }
529
530 if (channel_info->session_info) {
531 session_info_remove_channel(channel_info->session_info,
532 channel_info);
533 session_info_put(channel_info->session_info);
534 }
535 if (channel_info->name) {
536 free(channel_info->name);
537 }
538 call_rcu(&channel_info->rcu_node, free_channel_info_rcu);
539 }
540
541 static
542 void free_session_info_rcu(struct rcu_head *node)
543 {
544 free(caa_container_of(node, struct session_info, rcu_node));
545 }
546
547 /* Don't call directly, use the ref-counting mechanism. */
548 static
549 void session_info_destroy(void *_data)
550 {
551 struct session_info *session_info = _data;
552 int ret;
553
554 assert(session_info);
555 if (session_info->channel_infos_ht) {
556 ret = cds_lfht_destroy(session_info->channel_infos_ht, NULL);
557 if (ret) {
558 ERR("[notification-thread] Failed to destroy channel information hash table");
559 }
560 }
561 lttng_session_trigger_list_destroy(session_info->trigger_list);
562
563 rcu_read_lock();
564 cds_lfht_del(session_info->sessions_ht,
565 &session_info->sessions_ht_node);
566 rcu_read_unlock();
567 free(session_info->name);
568 call_rcu(&session_info->rcu_node, free_session_info_rcu);
569 }
570
571 static
572 void session_info_get(struct session_info *session_info)
573 {
574 if (!session_info) {
575 return;
576 }
577 lttng_ref_get(&session_info->ref);
578 }
579
580 static
581 void session_info_put(struct session_info *session_info)
582 {
583 if (!session_info) {
584 return;
585 }
586 lttng_ref_put(&session_info->ref);
587 }
588
589 static
590 struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid,
591 struct lttng_session_trigger_list *trigger_list,
592 struct cds_lfht *sessions_ht)
593 {
594 struct session_info *session_info;
595
596 assert(name);
597
598 session_info = zmalloc(sizeof(*session_info));
599 if (!session_info) {
600 goto end;
601 }
602 lttng_ref_init(&session_info->ref, session_info_destroy);
603
604 session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
605 1, 0, CDS_LFHT_AUTO_RESIZE | CDS_LFHT_ACCOUNTING, NULL);
606 if (!session_info->channel_infos_ht) {
607 goto error;
608 }
609
610 cds_lfht_node_init(&session_info->sessions_ht_node);
611 session_info->name = strdup(name);
612 if (!session_info->name) {
613 goto error;
614 }
615 session_info->uid = uid;
616 session_info->gid = gid;
617 session_info->trigger_list = trigger_list;
618 session_info->sessions_ht = sessions_ht;
619 end:
620 return session_info;
621 error:
622 session_info_put(session_info);
623 return NULL;
624 }
625
626 static
627 void session_info_add_channel(struct session_info *session_info,
628 struct channel_info *channel_info)
629 {
630 rcu_read_lock();
631 cds_lfht_add(session_info->channel_infos_ht,
632 hash_channel_key(&channel_info->key),
633 &channel_info->session_info_channels_ht_node);
634 rcu_read_unlock();
635 }
636
637 static
638 void session_info_remove_channel(struct session_info *session_info,
639 struct channel_info *channel_info)
640 {
641 rcu_read_lock();
642 cds_lfht_del(session_info->channel_infos_ht,
643 &channel_info->session_info_channels_ht_node);
644 rcu_read_unlock();
645 }
646
647 static
648 struct channel_info *channel_info_create(const char *channel_name,
649 struct channel_key *channel_key, uint64_t channel_capacity,
650 struct session_info *session_info)
651 {
652 struct channel_info *channel_info = zmalloc(sizeof(*channel_info));
653
654 if (!channel_info) {
655 goto end;
656 }
657
658 cds_lfht_node_init(&channel_info->channels_ht_node);
659 cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
660 memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
661 channel_info->capacity = channel_capacity;
662
663 channel_info->name = strdup(channel_name);
664 if (!channel_info->name) {
665 goto error;
666 }
667
668 /*
669 * Set the references between session and channel infos:
670 * - channel_info holds a strong reference to session_info
671 * - session_info holds a weak reference to channel_info
672 */
673 session_info_get(session_info);
674 session_info_add_channel(session_info, channel_info);
675 channel_info->session_info = session_info;
676 end:
677 return channel_info;
678 error:
679 channel_info_destroy(channel_info);
680 return NULL;
681 }
682
683 LTTNG_HIDDEN
684 bool notification_client_list_get(struct notification_client_list *list)
685 {
686 return urcu_ref_get_unless_zero(&list->ref);
687 }
688
689 static
690 void free_notification_client_list_rcu(struct rcu_head *node)
691 {
692 free(caa_container_of(node, struct notification_client_list,
693 rcu_node));
694 }
695
696 static
697 void notification_client_list_release(struct urcu_ref *list_ref)
698 {
699 struct notification_client_list *list =
700 container_of(list_ref, typeof(*list), ref);
701 struct notification_client_list_element *client_list_element, *tmp;
702
703 if (list->notification_trigger_clients_ht) {
704 rcu_read_lock();
705 cds_lfht_del(list->notification_trigger_clients_ht,
706 &list->notification_trigger_clients_ht_node);
707 rcu_read_unlock();
708 list->notification_trigger_clients_ht = NULL;
709 }
710 cds_list_for_each_entry_safe(client_list_element, tmp,
711 &list->list, node) {
712 free(client_list_element);
713 }
714 pthread_mutex_destroy(&list->lock);
715 call_rcu(&list->rcu_node, free_notification_client_list_rcu);
716 }
717
718 static
719 struct notification_client_list *notification_client_list_create(
720 const struct lttng_trigger *trigger)
721 {
722 struct notification_client_list *client_list =
723 zmalloc(sizeof(*client_list));
724
725 if (!client_list) {
726 goto error;
727 }
728 pthread_mutex_init(&client_list->lock, NULL);
729 urcu_ref_init(&client_list->ref);
730 cds_lfht_node_init(&client_list->notification_trigger_clients_ht_node);
731 CDS_INIT_LIST_HEAD(&client_list->list);
732 client_list->trigger = trigger;
733 error:
734 return client_list;
735 }
736
737 static
738 void publish_notification_client_list(
739 struct notification_thread_state *state,
740 struct notification_client_list *list)
741 {
742 const struct lttng_condition *condition =
743 lttng_trigger_get_const_condition(list->trigger);
744
745 assert(!list->notification_trigger_clients_ht);
746 notification_client_list_get(list);
747
748 list->notification_trigger_clients_ht =
749 state->notification_trigger_clients_ht;
750
751 rcu_read_lock();
752 cds_lfht_add(state->notification_trigger_clients_ht,
753 lttng_condition_hash(condition),
754 &list->notification_trigger_clients_ht_node);
755 rcu_read_unlock();
756 }
757
758 LTTNG_HIDDEN
759 void notification_client_list_put(struct notification_client_list *list)
760 {
761 if (!list) {
762 return;
763 }
764 return urcu_ref_put(&list->ref, notification_client_list_release);
765 }
766
767 /* Provides a reference to the returned list. */
768 static
769 struct notification_client_list *get_client_list_from_condition(
770 struct notification_thread_state *state,
771 const struct lttng_condition *condition)
772 {
773 struct cds_lfht_node *node;
774 struct cds_lfht_iter iter;
775 struct notification_client_list *list = NULL;
776
777 rcu_read_lock();
778 cds_lfht_lookup(state->notification_trigger_clients_ht,
779 lttng_condition_hash(condition),
780 match_client_list_condition,
781 condition,
782 &iter);
783 node = cds_lfht_iter_get_node(&iter);
784 if (node) {
785 list = container_of(node, struct notification_client_list,
786 notification_trigger_clients_ht_node);
787 list = notification_client_list_get(list) ? list : NULL;
788 }
789
790 rcu_read_unlock();
791 return list;
792 }
793
794 static
795 int evaluate_channel_condition_for_client(
796 const struct lttng_condition *condition,
797 struct notification_thread_state *state,
798 struct lttng_evaluation **evaluation,
799 uid_t *session_uid, gid_t *session_gid)
800 {
801 int ret;
802 struct cds_lfht_iter iter;
803 struct cds_lfht_node *node;
804 struct channel_info *channel_info = NULL;
805 struct channel_key *channel_key = NULL;
806 struct channel_state_sample *last_sample = NULL;
807 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
808
809 rcu_read_lock();
810
811 /* Find the channel associated with the condition. */
812 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
813 channel_trigger_list, channel_triggers_ht_node) {
814 struct lttng_trigger_list_element *element;
815
816 cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
817 const struct lttng_condition *current_condition =
818 lttng_trigger_get_const_condition(
819 element->trigger);
820
821 assert(current_condition);
822 if (!lttng_condition_is_equal(condition,
823 current_condition)) {
824 continue;
825 }
826
827 /* Found the trigger, save the channel key. */
828 channel_key = &channel_trigger_list->channel_key;
829 break;
830 }
831 if (channel_key) {
832 /* The channel key was found stop iteration. */
833 break;
834 }
835 }
836
837 if (!channel_key){
838 /* No channel found; normal exit. */
839 DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
840 ret = 0;
841 goto end;
842 }
843
844 /* Fetch channel info for the matching channel. */
845 cds_lfht_lookup(state->channels_ht,
846 hash_channel_key(channel_key),
847 match_channel_info,
848 channel_key,
849 &iter);
850 node = cds_lfht_iter_get_node(&iter);
851 assert(node);
852 channel_info = caa_container_of(node, struct channel_info,
853 channels_ht_node);
854
855 /* Retrieve the channel's last sample, if it exists. */
856 cds_lfht_lookup(state->channel_state_ht,
857 hash_channel_key(channel_key),
858 match_channel_state_sample,
859 channel_key,
860 &iter);
861 node = cds_lfht_iter_get_node(&iter);
862 if (node) {
863 last_sample = caa_container_of(node,
864 struct channel_state_sample,
865 channel_state_ht_node);
866 } else {
867 /* Nothing to evaluate, no sample was ever taken. Normal exit */
868 DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
869 ret = 0;
870 goto end;
871 }
872
873 ret = evaluate_buffer_condition(condition, evaluation, state,
874 NULL, last_sample,
875 0, channel_info->session_info->consumed_data_size,
876 channel_info);
877 if (ret) {
878 WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
879 goto end;
880 }
881
882 *session_uid = channel_info->session_info->uid;
883 *session_gid = channel_info->session_info->gid;
884 end:
885 rcu_read_unlock();
886 return ret;
887 }
888
889 static
890 const char *get_condition_session_name(const struct lttng_condition *condition)
891 {
892 const char *session_name = NULL;
893 enum lttng_condition_status status;
894
895 switch (lttng_condition_get_type(condition)) {
896 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
897 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
898 status = lttng_condition_buffer_usage_get_session_name(
899 condition, &session_name);
900 break;
901 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
902 status = lttng_condition_session_consumed_size_get_session_name(
903 condition, &session_name);
904 break;
905 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
906 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
907 status = lttng_condition_session_rotation_get_session_name(
908 condition, &session_name);
909 break;
910 default:
911 abort();
912 }
913 if (status != LTTNG_CONDITION_STATUS_OK) {
914 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
915 goto end;
916 }
917 end:
918 return session_name;
919 }
920
921 static
922 int evaluate_session_condition_for_client(
923 const struct lttng_condition *condition,
924 struct notification_thread_state *state,
925 struct lttng_evaluation **evaluation,
926 uid_t *session_uid, gid_t *session_gid)
927 {
928 int ret;
929 struct cds_lfht_iter iter;
930 struct cds_lfht_node *node;
931 const char *session_name;
932 struct session_info *session_info = NULL;
933
934 rcu_read_lock();
935 session_name = get_condition_session_name(condition);
936
937 /* Find the session associated with the trigger. */
938 cds_lfht_lookup(state->sessions_ht,
939 hash_key_str(session_name, lttng_ht_seed),
940 match_session,
941 session_name,
942 &iter);
943 node = cds_lfht_iter_get_node(&iter);
944 if (!node) {
945 DBG("[notification-thread] No known session matching name \"%s\"",
946 session_name);
947 ret = 0;
948 goto end;
949 }
950
951 session_info = caa_container_of(node, struct session_info,
952 sessions_ht_node);
953 session_info_get(session_info);
954
955 /*
956 * Evaluation is performed in-line here since only one type of
957 * session-bound condition is handled for the moment.
958 */
959 switch (lttng_condition_get_type(condition)) {
960 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
961 if (!session_info->rotation.ongoing) {
962 ret = 0;
963 goto end_session_put;
964 }
965
966 *evaluation = lttng_evaluation_session_rotation_ongoing_create(
967 session_info->rotation.id);
968 if (!*evaluation) {
969 /* Fatal error. */
970 ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
971 session_info->name);
972 ret = -1;
973 goto end_session_put;
974 }
975 ret = 0;
976 break;
977 default:
978 ret = 0;
979 goto end_session_put;
980 }
981
982 *session_uid = session_info->uid;
983 *session_gid = session_info->gid;
984
985 end_session_put:
986 session_info_put(session_info);
987 end:
988 rcu_read_unlock();
989 return ret;
990 }
991
992 static
993 int evaluate_condition_for_client(const struct lttng_trigger *trigger,
994 const struct lttng_condition *condition,
995 struct notification_client *client,
996 struct notification_thread_state *state)
997 {
998 int ret;
999 struct lttng_evaluation *evaluation = NULL;
1000 struct notification_client_list client_list = {
1001 .lock = PTHREAD_MUTEX_INITIALIZER,
1002 };
1003 struct notification_client_list_element client_list_element = { 0 };
1004 uid_t object_uid = 0;
1005 gid_t object_gid = 0;
1006
1007 assert(trigger);
1008 assert(condition);
1009 assert(client);
1010 assert(state);
1011
1012 switch (get_condition_binding_object(condition)) {
1013 case LTTNG_OBJECT_TYPE_SESSION:
1014 ret = evaluate_session_condition_for_client(condition, state,
1015 &evaluation, &object_uid, &object_gid);
1016 break;
1017 case LTTNG_OBJECT_TYPE_CHANNEL:
1018 ret = evaluate_channel_condition_for_client(condition, state,
1019 &evaluation, &object_uid, &object_gid);
1020 break;
1021 case LTTNG_OBJECT_TYPE_NONE:
1022 ret = 0;
1023 goto end;
1024 case LTTNG_OBJECT_TYPE_UNKNOWN:
1025 default:
1026 ret = -1;
1027 goto end;
1028 }
1029 if (ret) {
1030 /* Fatal error. */
1031 goto end;
1032 }
1033 if (!evaluation) {
1034 /* Evaluation yielded nothing. Normal exit. */
1035 DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
1036 ret = 0;
1037 goto end;
1038 }
1039
1040 /*
1041 * Create a temporary client list with the client currently
1042 * subscribing.
1043 */
1044 cds_lfht_node_init(&client_list.notification_trigger_clients_ht_node);
1045 CDS_INIT_LIST_HEAD(&client_list.list);
1046 client_list.trigger = trigger;
1047
1048 CDS_INIT_LIST_HEAD(&client_list_element.node);
1049 client_list_element.client = client;
1050 cds_list_add(&client_list_element.node, &client_list.list);
1051
1052 /* Send evaluation result to the newly-subscribed client. */
1053 DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
1054 ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
1055 state, object_uid, object_gid);
1056
1057 end:
1058 return ret;
1059 }
1060
1061 static
1062 int notification_thread_client_subscribe(struct notification_client *client,
1063 struct lttng_condition *condition,
1064 struct notification_thread_state *state,
1065 enum lttng_notification_channel_status *_status)
1066 {
1067 int ret = 0;
1068 struct notification_client_list *client_list = NULL;
1069 struct lttng_condition_list_element *condition_list_element = NULL;
1070 struct notification_client_list_element *client_list_element = NULL;
1071 enum lttng_notification_channel_status status =
1072 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1073
1074 /*
1075 * Ensure that the client has not already subscribed to this condition
1076 * before.
1077 */
1078 cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
1079 if (lttng_condition_is_equal(condition_list_element->condition,
1080 condition)) {
1081 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
1082 goto end;
1083 }
1084 }
1085
1086 condition_list_element = zmalloc(sizeof(*condition_list_element));
1087 if (!condition_list_element) {
1088 ret = -1;
1089 goto error;
1090 }
1091 client_list_element = zmalloc(sizeof(*client_list_element));
1092 if (!client_list_element) {
1093 ret = -1;
1094 goto error;
1095 }
1096
1097 /*
1098 * Add the newly-subscribed condition to the client's subscription list.
1099 */
1100 CDS_INIT_LIST_HEAD(&condition_list_element->node);
1101 condition_list_element->condition = condition;
1102 cds_list_add(&condition_list_element->node, &client->condition_list);
1103
1104 client_list = get_client_list_from_condition(state, condition);
1105 if (!client_list) {
1106 /*
1107 * No notification-emiting trigger registered with this
1108 * condition. We don't evaluate the condition right away
1109 * since this trigger is not registered yet.
1110 */
1111 free(client_list_element);
1112 goto end;
1113 }
1114
1115 /*
1116 * The condition to which the client just subscribed is evaluated
1117 * at this point so that conditions that are already TRUE result
1118 * in a notification being sent out.
1119 *
1120 * The client_list's trigger is used without locking the list itself.
1121 * This is correct since the list doesn't own the trigger and the
1122 * object is immutable.
1123 */
1124 if (evaluate_condition_for_client(client_list->trigger, condition,
1125 client, state)) {
1126 WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
1127 ret = -1;
1128 free(client_list_element);
1129 goto end;
1130 }
1131
1132 /*
1133 * Add the client to the list of clients interested in a given trigger
1134 * if a "notification" trigger with a corresponding condition was
1135 * added prior.
1136 */
1137 client_list_element->client = client;
1138 CDS_INIT_LIST_HEAD(&client_list_element->node);
1139
1140 pthread_mutex_lock(&client_list->lock);
1141 cds_list_add(&client_list_element->node, &client_list->list);
1142 pthread_mutex_unlock(&client_list->lock);
1143 end:
1144 if (_status) {
1145 *_status = status;
1146 }
1147 if (client_list) {
1148 notification_client_list_put(client_list);
1149 }
1150 return ret;
1151 error:
1152 free(condition_list_element);
1153 free(client_list_element);
1154 return ret;
1155 }
1156
1157 static
1158 int notification_thread_client_unsubscribe(
1159 struct notification_client *client,
1160 struct lttng_condition *condition,
1161 struct notification_thread_state *state,
1162 enum lttng_notification_channel_status *_status)
1163 {
1164 struct notification_client_list *client_list;
1165 struct lttng_condition_list_element *condition_list_element,
1166 *condition_tmp;
1167 struct notification_client_list_element *client_list_element,
1168 *client_tmp;
1169 bool condition_found = false;
1170 enum lttng_notification_channel_status status =
1171 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1172
1173 /* Remove the condition from the client's condition list. */
1174 cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
1175 &client->condition_list, node) {
1176 if (!lttng_condition_is_equal(condition_list_element->condition,
1177 condition)) {
1178 continue;
1179 }
1180
1181 cds_list_del(&condition_list_element->node);
1182 /*
1183 * The caller may be iterating on the client's conditions to
1184 * tear down a client's connection. In this case, the condition
1185 * will be destroyed at the end.
1186 */
1187 if (condition != condition_list_element->condition) {
1188 lttng_condition_destroy(
1189 condition_list_element->condition);
1190 }
1191 free(condition_list_element);
1192 condition_found = true;
1193 break;
1194 }
1195
1196 if (!condition_found) {
1197 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
1198 goto end;
1199 }
1200
1201 /*
1202 * Remove the client from the list of clients interested the trigger
1203 * matching the condition.
1204 */
1205 client_list = get_client_list_from_condition(state, condition);
1206 if (!client_list) {
1207 goto end;
1208 }
1209
1210 pthread_mutex_lock(&client_list->lock);
1211 cds_list_for_each_entry_safe(client_list_element, client_tmp,
1212 &client_list->list, node) {
1213 if (client_list_element->client->id != client->id) {
1214 continue;
1215 }
1216 cds_list_del(&client_list_element->node);
1217 free(client_list_element);
1218 break;
1219 }
1220 pthread_mutex_unlock(&client_list->lock);
1221 notification_client_list_put(client_list);
1222 client_list = NULL;
1223 end:
1224 lttng_condition_destroy(condition);
1225 if (_status) {
1226 *_status = status;
1227 }
1228 return 0;
1229 }
1230
1231 static
1232 void free_notification_client_rcu(struct rcu_head *node)
1233 {
1234 free(caa_container_of(node, struct notification_client, rcu_node));
1235 }
1236
1237 static
1238 void notification_client_destroy(struct notification_client *client,
1239 struct notification_thread_state *state)
1240 {
1241 if (!client) {
1242 return;
1243 }
1244
1245 /*
1246 * The client object is not reachable by other threads, no need to lock
1247 * the client here.
1248 */
1249 if (client->socket >= 0) {
1250 (void) lttcomm_close_unix_sock(client->socket);
1251 client->socket = -1;
1252 }
1253 client->communication.active = false;
1254 lttng_payload_reset(&client->communication.inbound.payload);
1255 lttng_payload_reset(&client->communication.outbound.payload);
1256 pthread_mutex_destroy(&client->lock);
1257 call_rcu(&client->rcu_node, free_notification_client_rcu);
1258 }
1259
1260 /*
1261 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1262 * client pointer).
1263 */
1264 static
1265 struct notification_client *get_client_from_socket(int socket,
1266 struct notification_thread_state *state)
1267 {
1268 struct cds_lfht_iter iter;
1269 struct cds_lfht_node *node;
1270 struct notification_client *client = NULL;
1271
1272 cds_lfht_lookup(state->client_socket_ht,
1273 hash_client_socket(socket),
1274 match_client_socket,
1275 (void *) (unsigned long) socket,
1276 &iter);
1277 node = cds_lfht_iter_get_node(&iter);
1278 if (!node) {
1279 goto end;
1280 }
1281
1282 client = caa_container_of(node, struct notification_client,
1283 client_socket_ht_node);
1284 end:
1285 return client;
1286 }
1287
1288 /*
1289 * Call with rcu_read_lock held (and hold for the lifetime of the returned
1290 * client pointer).
1291 */
1292 static
1293 struct notification_client *get_client_from_id(notification_client_id id,
1294 struct notification_thread_state *state)
1295 {
1296 struct cds_lfht_iter iter;
1297 struct cds_lfht_node *node;
1298 struct notification_client *client = NULL;
1299
1300 cds_lfht_lookup(state->client_id_ht,
1301 hash_client_id(id),
1302 match_client_id,
1303 &id,
1304 &iter);
1305 node = cds_lfht_iter_get_node(&iter);
1306 if (!node) {
1307 goto end;
1308 }
1309
1310 client = caa_container_of(node, struct notification_client,
1311 client_id_ht_node);
1312 end:
1313 return client;
1314 }
1315
1316 static
1317 bool buffer_usage_condition_applies_to_channel(
1318 const struct lttng_condition *condition,
1319 const struct channel_info *channel_info)
1320 {
1321 enum lttng_condition_status status;
1322 enum lttng_domain_type condition_domain;
1323 const char *condition_session_name = NULL;
1324 const char *condition_channel_name = NULL;
1325
1326 status = lttng_condition_buffer_usage_get_domain_type(condition,
1327 &condition_domain);
1328 assert(status == LTTNG_CONDITION_STATUS_OK);
1329 if (channel_info->key.domain != condition_domain) {
1330 goto fail;
1331 }
1332
1333 status = lttng_condition_buffer_usage_get_session_name(
1334 condition, &condition_session_name);
1335 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1336
1337 status = lttng_condition_buffer_usage_get_channel_name(
1338 condition, &condition_channel_name);
1339 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
1340
1341 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1342 goto fail;
1343 }
1344 if (strcmp(channel_info->name, condition_channel_name)) {
1345 goto fail;
1346 }
1347
1348 return true;
1349 fail:
1350 return false;
1351 }
1352
1353 static
1354 bool session_consumed_size_condition_applies_to_channel(
1355 const struct lttng_condition *condition,
1356 const struct channel_info *channel_info)
1357 {
1358 enum lttng_condition_status status;
1359 const char *condition_session_name = NULL;
1360
1361 status = lttng_condition_session_consumed_size_get_session_name(
1362 condition, &condition_session_name);
1363 assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
1364
1365 if (strcmp(channel_info->session_info->name, condition_session_name)) {
1366 goto fail;
1367 }
1368
1369 return true;
1370 fail:
1371 return false;
1372 }
1373
1374 static
1375 bool trigger_applies_to_channel(const struct lttng_trigger *trigger,
1376 const struct channel_info *channel_info)
1377 {
1378 const struct lttng_condition *condition;
1379 bool trigger_applies;
1380
1381 condition = lttng_trigger_get_const_condition(trigger);
1382 if (!condition) {
1383 goto fail;
1384 }
1385
1386 switch (lttng_condition_get_type(condition)) {
1387 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
1388 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
1389 trigger_applies = buffer_usage_condition_applies_to_channel(
1390 condition, channel_info);
1391 break;
1392 case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
1393 trigger_applies = session_consumed_size_condition_applies_to_channel(
1394 condition, channel_info);
1395 break;
1396 default:
1397 goto fail;
1398 }
1399
1400 return trigger_applies;
1401 fail:
1402 return false;
1403 }
1404
1405 static
1406 bool trigger_applies_to_client(struct lttng_trigger *trigger,
1407 struct notification_client *client)
1408 {
1409 bool applies = false;
1410 struct lttng_condition_list_element *condition_list_element;
1411
1412 cds_list_for_each_entry(condition_list_element, &client->condition_list,
1413 node) {
1414 applies = lttng_condition_is_equal(
1415 condition_list_element->condition,
1416 lttng_trigger_get_condition(trigger));
1417 if (applies) {
1418 break;
1419 }
1420 }
1421 return applies;
1422 }
1423
1424 /* Must be called with RCU read lock held. */
1425 static
1426 struct lttng_session_trigger_list *get_session_trigger_list(
1427 struct notification_thread_state *state,
1428 const char *session_name)
1429 {
1430 struct lttng_session_trigger_list *list = NULL;
1431 struct cds_lfht_node *node;
1432 struct cds_lfht_iter iter;
1433
1434 cds_lfht_lookup(state->session_triggers_ht,
1435 hash_key_str(session_name, lttng_ht_seed),
1436 match_session_trigger_list,
1437 session_name,
1438 &iter);
1439 node = cds_lfht_iter_get_node(&iter);
1440 if (!node) {
1441 /*
1442 * Not an error, the list of triggers applying to that session
1443 * will be initialized when the session is created.
1444 */
1445 DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
1446 session_name);
1447 goto end;
1448 }
1449
1450 list = caa_container_of(node,
1451 struct lttng_session_trigger_list,
1452 session_triggers_ht_node);
1453 end:
1454 return list;
1455 }
1456
1457 /*
1458 * Allocate an empty lttng_session_trigger_list for the session named
1459 * 'session_name'.
1460 *
1461 * No ownership of 'session_name' is assumed by the session trigger list.
1462 * It is the caller's responsability to ensure the session name is alive
1463 * for as long as this list is.
1464 */
1465 static
1466 struct lttng_session_trigger_list *lttng_session_trigger_list_create(
1467 const char *session_name,
1468 struct cds_lfht *session_triggers_ht)
1469 {
1470 struct lttng_session_trigger_list *list;
1471
1472 list = zmalloc(sizeof(*list));
1473 if (!list) {
1474 goto end;
1475 }
1476 list->session_name = session_name;
1477 CDS_INIT_LIST_HEAD(&list->list);
1478 cds_lfht_node_init(&list->session_triggers_ht_node);
1479 list->session_triggers_ht = session_triggers_ht;
1480
1481 rcu_read_lock();
1482 /* Publish the list through the session_triggers_ht. */
1483 cds_lfht_add(session_triggers_ht,
1484 hash_key_str(session_name, lttng_ht_seed),
1485 &list->session_triggers_ht_node);
1486 rcu_read_unlock();
1487 end:
1488 return list;
1489 }
1490
1491 static
1492 void free_session_trigger_list_rcu(struct rcu_head *node)
1493 {
1494 free(caa_container_of(node, struct lttng_session_trigger_list,
1495 rcu_node));
1496 }
1497
1498 static
1499 void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list *list)
1500 {
1501 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1502
1503 /* Empty the list element by element, and then free the list itself. */
1504 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1505 &list->list, node) {
1506 cds_list_del(&trigger_list_element->node);
1507 free(trigger_list_element);
1508 }
1509 rcu_read_lock();
1510 /* Unpublish the list from the session_triggers_ht. */
1511 cds_lfht_del(list->session_triggers_ht,
1512 &list->session_triggers_ht_node);
1513 rcu_read_unlock();
1514 call_rcu(&list->rcu_node, free_session_trigger_list_rcu);
1515 }
1516
1517 static
1518 int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
1519 struct lttng_trigger *trigger)
1520 {
1521 int ret = 0;
1522 struct lttng_trigger_list_element *new_element =
1523 zmalloc(sizeof(*new_element));
1524
1525 if (!new_element) {
1526 ret = -1;
1527 goto end;
1528 }
1529 CDS_INIT_LIST_HEAD(&new_element->node);
1530 new_element->trigger = trigger;
1531 cds_list_add(&new_element->node, &list->list);
1532 end:
1533 return ret;
1534 }
1535
1536 static
1537 bool trigger_applies_to_session(const struct lttng_trigger *trigger,
1538 const char *session_name)
1539 {
1540 bool applies = false;
1541 const struct lttng_condition *condition;
1542
1543 condition = lttng_trigger_get_const_condition(trigger);
1544 switch (lttng_condition_get_type(condition)) {
1545 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
1546 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
1547 {
1548 enum lttng_condition_status condition_status;
1549 const char *condition_session_name;
1550
1551 condition_status = lttng_condition_session_rotation_get_session_name(
1552 condition, &condition_session_name);
1553 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
1554 ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
1555 goto end;
1556 }
1557
1558 assert(condition_session_name);
1559 applies = !strcmp(condition_session_name, session_name);
1560 break;
1561 }
1562 default:
1563 goto end;
1564 }
1565 end:
1566 return applies;
1567 }
1568
1569 /*
1570 * Allocate and initialize an lttng_session_trigger_list which contains
1571 * all triggers that apply to the session named 'session_name'.
1572 *
1573 * No ownership of 'session_name' is assumed by the session trigger list.
1574 * It is the caller's responsability to ensure the session name is alive
1575 * for as long as this list is.
1576 */
1577 static
1578 struct lttng_session_trigger_list *lttng_session_trigger_list_build(
1579 const struct notification_thread_state *state,
1580 const char *session_name)
1581 {
1582 int trigger_count = 0;
1583 struct lttng_session_trigger_list *session_trigger_list = NULL;
1584 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1585 struct cds_lfht_iter iter;
1586
1587 session_trigger_list = lttng_session_trigger_list_create(session_name,
1588 state->session_triggers_ht);
1589
1590 /* Add all triggers applying to the session named 'session_name'. */
1591 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1592 node) {
1593 int ret;
1594
1595 if (!trigger_applies_to_session(trigger_ht_element->trigger,
1596 session_name)) {
1597 continue;
1598 }
1599
1600 ret = lttng_session_trigger_list_add(session_trigger_list,
1601 trigger_ht_element->trigger);
1602 if (ret) {
1603 goto error;
1604 }
1605
1606 trigger_count++;
1607 }
1608
1609 DBG("[notification-thread] Found %i triggers that apply to newly created session",
1610 trigger_count);
1611 return session_trigger_list;
1612 error:
1613 lttng_session_trigger_list_destroy(session_trigger_list);
1614 return NULL;
1615 }
1616
1617 static
1618 struct session_info *find_or_create_session_info(
1619 struct notification_thread_state *state,
1620 const char *name, uid_t uid, gid_t gid)
1621 {
1622 struct session_info *session = NULL;
1623 struct cds_lfht_node *node;
1624 struct cds_lfht_iter iter;
1625 struct lttng_session_trigger_list *trigger_list;
1626
1627 rcu_read_lock();
1628 cds_lfht_lookup(state->sessions_ht,
1629 hash_key_str(name, lttng_ht_seed),
1630 match_session,
1631 name,
1632 &iter);
1633 node = cds_lfht_iter_get_node(&iter);
1634 if (node) {
1635 DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
1636 name, uid, gid);
1637 session = caa_container_of(node, struct session_info,
1638 sessions_ht_node);
1639 assert(session->uid == uid);
1640 assert(session->gid == gid);
1641 session_info_get(session);
1642 goto end;
1643 }
1644
1645 trigger_list = lttng_session_trigger_list_build(state, name);
1646 if (!trigger_list) {
1647 goto error;
1648 }
1649
1650 session = session_info_create(name, uid, gid, trigger_list,
1651 state->sessions_ht);
1652 if (!session) {
1653 ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
1654 name, uid, gid);
1655 lttng_session_trigger_list_destroy(trigger_list);
1656 goto error;
1657 }
1658 trigger_list = NULL;
1659
1660 cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
1661 &session->sessions_ht_node);
1662 end:
1663 rcu_read_unlock();
1664 return session;
1665 error:
1666 rcu_read_unlock();
1667 session_info_put(session);
1668 return NULL;
1669 }
1670
1671 static
1672 int handle_notification_thread_command_add_channel(
1673 struct notification_thread_state *state,
1674 const char *session_name, uid_t session_uid, gid_t session_gid,
1675 const char *channel_name, enum lttng_domain_type channel_domain,
1676 uint64_t channel_key_int, uint64_t channel_capacity,
1677 enum lttng_error_code *cmd_result)
1678 {
1679 struct cds_list_head trigger_list;
1680 struct channel_info *new_channel_info = NULL;
1681 struct channel_key channel_key = {
1682 .key = channel_key_int,
1683 .domain = channel_domain,
1684 };
1685 struct lttng_channel_trigger_list *channel_trigger_list = NULL;
1686 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
1687 int trigger_count = 0;
1688 struct cds_lfht_iter iter;
1689 struct session_info *session_info = NULL;
1690
1691 DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
1692 channel_name, session_name, channel_key_int,
1693 channel_domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1694
1695 CDS_INIT_LIST_HEAD(&trigger_list);
1696
1697 session_info = find_or_create_session_info(state, session_name,
1698 session_uid, session_gid);
1699 if (!session_info) {
1700 /* Allocation error or an internal error occurred. */
1701 goto error;
1702 }
1703
1704 new_channel_info = channel_info_create(channel_name, &channel_key,
1705 channel_capacity, session_info);
1706 if (!new_channel_info) {
1707 goto error;
1708 }
1709
1710 rcu_read_lock();
1711 /* Build a list of all triggers applying to the new channel. */
1712 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1713 node) {
1714 struct lttng_trigger_list_element *new_element;
1715
1716 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
1717 new_channel_info)) {
1718 continue;
1719 }
1720
1721 new_element = zmalloc(sizeof(*new_element));
1722 if (!new_element) {
1723 rcu_read_unlock();
1724 goto error;
1725 }
1726 CDS_INIT_LIST_HEAD(&new_element->node);
1727 new_element->trigger = trigger_ht_element->trigger;
1728 cds_list_add(&new_element->node, &trigger_list);
1729 trigger_count++;
1730 }
1731 rcu_read_unlock();
1732
1733 DBG("[notification-thread] Found %i triggers that apply to newly added channel",
1734 trigger_count);
1735 channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
1736 if (!channel_trigger_list) {
1737 goto error;
1738 }
1739 channel_trigger_list->channel_key = new_channel_info->key;
1740 CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
1741 cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
1742 cds_list_splice(&trigger_list, &channel_trigger_list->list);
1743
1744 rcu_read_lock();
1745 /* Add channel to the channel_ht which owns the channel_infos. */
1746 cds_lfht_add(state->channels_ht,
1747 hash_channel_key(&new_channel_info->key),
1748 &new_channel_info->channels_ht_node);
1749 /*
1750 * Add the list of triggers associated with this channel to the
1751 * channel_triggers_ht.
1752 */
1753 cds_lfht_add(state->channel_triggers_ht,
1754 hash_channel_key(&new_channel_info->key),
1755 &channel_trigger_list->channel_triggers_ht_node);
1756 rcu_read_unlock();
1757 session_info_put(session_info);
1758 *cmd_result = LTTNG_OK;
1759 return 0;
1760 error:
1761 channel_info_destroy(new_channel_info);
1762 session_info_put(session_info);
1763 return 1;
1764 }
1765
1766 static
1767 void free_channel_trigger_list_rcu(struct rcu_head *node)
1768 {
1769 free(caa_container_of(node, struct lttng_channel_trigger_list,
1770 rcu_node));
1771 }
1772
1773 static
1774 void free_channel_state_sample_rcu(struct rcu_head *node)
1775 {
1776 free(caa_container_of(node, struct channel_state_sample,
1777 rcu_node));
1778 }
1779
1780 static
1781 int handle_notification_thread_command_remove_channel(
1782 struct notification_thread_state *state,
1783 uint64_t channel_key, enum lttng_domain_type domain,
1784 enum lttng_error_code *cmd_result)
1785 {
1786 struct cds_lfht_node *node;
1787 struct cds_lfht_iter iter;
1788 struct lttng_channel_trigger_list *trigger_list;
1789 struct lttng_trigger_list_element *trigger_list_element, *tmp;
1790 struct channel_key key = { .key = channel_key, .domain = domain };
1791 struct channel_info *channel_info;
1792
1793 DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
1794 channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
1795
1796 rcu_read_lock();
1797
1798 cds_lfht_lookup(state->channel_triggers_ht,
1799 hash_channel_key(&key),
1800 match_channel_trigger_list,
1801 &key,
1802 &iter);
1803 node = cds_lfht_iter_get_node(&iter);
1804 /*
1805 * There is a severe internal error if we are being asked to remove a
1806 * channel that doesn't exist.
1807 */
1808 if (!node) {
1809 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
1810 goto end;
1811 }
1812
1813 /* Free the list of triggers associated with this channel. */
1814 trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
1815 channel_triggers_ht_node);
1816 cds_list_for_each_entry_safe(trigger_list_element, tmp,
1817 &trigger_list->list, node) {
1818 cds_list_del(&trigger_list_element->node);
1819 free(trigger_list_element);
1820 }
1821 cds_lfht_del(state->channel_triggers_ht, node);
1822 call_rcu(&trigger_list->rcu_node, free_channel_trigger_list_rcu);
1823
1824 /* Free sampled channel state. */
1825 cds_lfht_lookup(state->channel_state_ht,
1826 hash_channel_key(&key),
1827 match_channel_state_sample,
1828 &key,
1829 &iter);
1830 node = cds_lfht_iter_get_node(&iter);
1831 /*
1832 * This is expected to be NULL if the channel is destroyed before we
1833 * received a sample.
1834 */
1835 if (node) {
1836 struct channel_state_sample *sample = caa_container_of(node,
1837 struct channel_state_sample,
1838 channel_state_ht_node);
1839
1840 cds_lfht_del(state->channel_state_ht, node);
1841 call_rcu(&sample->rcu_node, free_channel_state_sample_rcu);
1842 }
1843
1844 /* Remove the channel from the channels_ht and free it. */
1845 cds_lfht_lookup(state->channels_ht,
1846 hash_channel_key(&key),
1847 match_channel_info,
1848 &key,
1849 &iter);
1850 node = cds_lfht_iter_get_node(&iter);
1851 assert(node);
1852 channel_info = caa_container_of(node, struct channel_info,
1853 channels_ht_node);
1854 cds_lfht_del(state->channels_ht, node);
1855 channel_info_destroy(channel_info);
1856 end:
1857 rcu_read_unlock();
1858 *cmd_result = LTTNG_OK;
1859 return 0;
1860 }
1861
1862 static
1863 int handle_notification_thread_command_session_rotation(
1864 struct notification_thread_state *state,
1865 enum notification_thread_command_type cmd_type,
1866 const char *session_name, uid_t session_uid, gid_t session_gid,
1867 uint64_t trace_archive_chunk_id,
1868 struct lttng_trace_archive_location *location,
1869 enum lttng_error_code *_cmd_result)
1870 {
1871 int ret = 0;
1872 enum lttng_error_code cmd_result = LTTNG_OK;
1873 struct lttng_session_trigger_list *trigger_list;
1874 struct lttng_trigger_list_element *trigger_list_element;
1875 struct session_info *session_info;
1876 const struct lttng_credentials session_creds = {
1877 .uid = LTTNG_OPTIONAL_INIT_VALUE(session_uid),
1878 .gid = LTTNG_OPTIONAL_INIT_VALUE(session_gid),
1879 };
1880
1881 rcu_read_lock();
1882
1883 session_info = find_or_create_session_info(state, session_name,
1884 session_uid, session_gid);
1885 if (!session_info) {
1886 /* Allocation error or an internal error occurred. */
1887 ret = -1;
1888 cmd_result = LTTNG_ERR_NOMEM;
1889 goto end;
1890 }
1891
1892 session_info->rotation.ongoing =
1893 cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING;
1894 session_info->rotation.id = trace_archive_chunk_id;
1895 trigger_list = get_session_trigger_list(state, session_name);
1896 if (!trigger_list) {
1897 DBG("[notification-thread] No triggers applying to session \"%s\" found",
1898 session_name);
1899 goto end;
1900 }
1901
1902 cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
1903 node) {
1904 const struct lttng_condition *condition;
1905 struct lttng_trigger *trigger;
1906 struct notification_client_list *client_list;
1907 struct lttng_evaluation *evaluation = NULL;
1908 enum lttng_condition_type condition_type;
1909 enum action_executor_status executor_status;
1910
1911 trigger = trigger_list_element->trigger;
1912 condition = lttng_trigger_get_const_condition(trigger);
1913 assert(condition);
1914 condition_type = lttng_condition_get_type(condition);
1915
1916 if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING &&
1917 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1918 continue;
1919 } else if (condition_type == LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED &&
1920 cmd_type != NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED) {
1921 continue;
1922 }
1923
1924 client_list = get_client_list_from_condition(state, condition);
1925 if (cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING) {
1926 evaluation = lttng_evaluation_session_rotation_ongoing_create(
1927 trace_archive_chunk_id);
1928 } else {
1929 evaluation = lttng_evaluation_session_rotation_completed_create(
1930 trace_archive_chunk_id, location);
1931 }
1932
1933 if (!evaluation) {
1934 /* Internal error */
1935 ret = -1;
1936 cmd_result = LTTNG_ERR_UNK;
1937 goto put_list;
1938 }
1939
1940 /*
1941 * Ownership of `evaluation` transferred to the action executor
1942 * no matter the result.
1943 */
1944 executor_status = action_executor_enqueue(state->executor,
1945 trigger, evaluation, &session_creds,
1946 client_list);
1947 evaluation = NULL;
1948 switch (executor_status) {
1949 case ACTION_EXECUTOR_STATUS_OK:
1950 break;
1951 case ACTION_EXECUTOR_STATUS_ERROR:
1952 case ACTION_EXECUTOR_STATUS_INVALID:
1953 /*
1954 * TODO Add trigger identification (name/id) when
1955 * it is added to the API.
1956 */
1957 ERR("Fatal error occurred while enqueuing action associated with session rotation trigger");
1958 ret = -1;
1959 goto put_list;
1960 case ACTION_EXECUTOR_STATUS_OVERFLOW:
1961 /*
1962 * TODO Add trigger identification (name/id) when
1963 * it is added to the API.
1964 *
1965 * Not a fatal error.
1966 */
1967 WARN("No space left when enqueuing action associated with session rotation trigger");
1968 ret = 0;
1969 goto put_list;
1970 default:
1971 abort();
1972 }
1973
1974 put_list:
1975 notification_client_list_put(client_list);
1976 if (caa_unlikely(ret)) {
1977 break;
1978 }
1979 }
1980 end:
1981 session_info_put(session_info);
1982 *_cmd_result = cmd_result;
1983 rcu_read_unlock();
1984 return ret;
1985 }
1986
1987 static int handle_notification_thread_command_list_triggers(
1988 struct notification_thread_handle *handle,
1989 struct notification_thread_state *state,
1990 uid_t client_uid,
1991 struct lttng_triggers **triggers,
1992 enum lttng_error_code *_cmd_result)
1993 {
1994 int ret = 0;
1995 enum lttng_error_code cmd_result = LTTNG_OK;
1996 struct cds_lfht_iter iter;
1997 struct lttng_trigger_ht_element *trigger_ht_element;
1998 struct lttng_triggers *local_triggers = NULL;
1999 const struct lttng_credentials *creds;
2000
2001 rcu_read_lock();
2002
2003 local_triggers = lttng_triggers_create();
2004 if (!local_triggers) {
2005 /* Not a fatal error. */
2006 cmd_result = LTTNG_ERR_NOMEM;
2007 goto end;
2008 }
2009
2010 cds_lfht_for_each_entry(state->triggers_ht, &iter,
2011 trigger_ht_element, node) {
2012 /*
2013 * Only return the triggers to which the client has access.
2014 * The root user has visibility over all triggers.
2015 */
2016 creds = lttng_trigger_get_credentials(trigger_ht_element->trigger);
2017 if (client_uid != lttng_credentials_get_uid(creds) && client_uid != 0) {
2018 continue;
2019 }
2020
2021 ret = lttng_triggers_add(local_triggers,
2022 trigger_ht_element->trigger);
2023 if (ret < 0) {
2024 /* Not a fatal error. */
2025 ret = 0;
2026 cmd_result = LTTNG_ERR_NOMEM;
2027 goto end;
2028 }
2029 }
2030
2031 /* Transferring ownership to the caller. */
2032 *triggers = local_triggers;
2033 local_triggers = NULL;
2034
2035 end:
2036 rcu_read_unlock();
2037 lttng_triggers_destroy(local_triggers);
2038 *_cmd_result = cmd_result;
2039 return ret;
2040 }
2041
2042 static
2043 int condition_is_supported(struct lttng_condition *condition)
2044 {
2045 int ret;
2046
2047 switch (lttng_condition_get_type(condition)) {
2048 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
2049 case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
2050 {
2051 enum lttng_domain_type domain;
2052
2053 ret = lttng_condition_buffer_usage_get_domain_type(condition,
2054 &domain);
2055 if (ret) {
2056 ret = -1;
2057 goto end;
2058 }
2059
2060 if (domain != LTTNG_DOMAIN_KERNEL) {
2061 ret = 1;
2062 goto end;
2063 }
2064
2065 /*
2066 * Older kernel tracers don't expose the API to monitor their
2067 * buffers. Therefore, we reject triggers that require that
2068 * mechanism to be available to be evaluated.
2069 */
2070 ret = kernel_supports_ring_buffer_snapshot_sample_positions();
2071 break;
2072 }
2073 default:
2074 ret = 1;
2075 }
2076 end:
2077 return ret;
2078 }
2079
2080 /* Must be called with RCU read lock held. */
2081 static
2082 int bind_trigger_to_matching_session(struct lttng_trigger *trigger,
2083 struct notification_thread_state *state)
2084 {
2085 int ret = 0;
2086 const struct lttng_condition *condition;
2087 const char *session_name;
2088 struct lttng_session_trigger_list *trigger_list;
2089
2090 condition = lttng_trigger_get_const_condition(trigger);
2091 switch (lttng_condition_get_type(condition)) {
2092 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
2093 case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
2094 {
2095 enum lttng_condition_status status;
2096
2097 status = lttng_condition_session_rotation_get_session_name(
2098 condition, &session_name);
2099 if (status != LTTNG_CONDITION_STATUS_OK) {
2100 ERR("[notification-thread] Failed to bind trigger to session: unable to get 'session_rotation' condition's session name");
2101 ret = -1;
2102 goto end;
2103 }
2104 break;
2105 }
2106 default:
2107 ret = -1;
2108 goto end;
2109 }
2110
2111 trigger_list = get_session_trigger_list(state, session_name);
2112 if (!trigger_list) {
2113 DBG("[notification-thread] Unable to bind trigger applying to session \"%s\" as it is not yet known to the notification system",
2114 session_name);
2115 goto end;
2116
2117 }
2118
2119 DBG("[notification-thread] Newly registered trigger bound to session \"%s\"",
2120 session_name);
2121 ret = lttng_session_trigger_list_add(trigger_list, trigger);
2122 end:
2123 return ret;
2124 }
2125
2126 /* Must be called with RCU read lock held. */
2127 static
2128 int bind_trigger_to_matching_channels(struct lttng_trigger *trigger,
2129 struct notification_thread_state *state)
2130 {
2131 int ret = 0;
2132 struct cds_lfht_node *node;
2133 struct cds_lfht_iter iter;
2134 struct channel_info *channel;
2135
2136 cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
2137 channels_ht_node) {
2138 struct lttng_trigger_list_element *trigger_list_element;
2139 struct lttng_channel_trigger_list *trigger_list;
2140 struct cds_lfht_iter lookup_iter;
2141
2142 if (!trigger_applies_to_channel(trigger, channel)) {
2143 continue;
2144 }
2145
2146 cds_lfht_lookup(state->channel_triggers_ht,
2147 hash_channel_key(&channel->key),
2148 match_channel_trigger_list,
2149 &channel->key,
2150 &lookup_iter);
2151 node = cds_lfht_iter_get_node(&lookup_iter);
2152 assert(node);
2153 trigger_list = caa_container_of(node,
2154 struct lttng_channel_trigger_list,
2155 channel_triggers_ht_node);
2156
2157 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
2158 if (!trigger_list_element) {
2159 ret = -1;
2160 goto end;
2161 }
2162 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
2163 trigger_list_element->trigger = trigger;
2164 cds_list_add(&trigger_list_element->node, &trigger_list->list);
2165 DBG("[notification-thread] Newly registered trigger bound to channel \"%s\"",
2166 channel->name);
2167 }
2168 end:
2169 return ret;
2170 }
2171
2172 static
2173 bool is_trigger_action_notify(const struct lttng_trigger *trigger)
2174 {
2175 bool is_notify = false;
2176 unsigned int i, count;
2177 enum lttng_action_status action_status;
2178 const struct lttng_action *action =
2179 lttng_trigger_get_const_action(trigger);
2180 enum lttng_action_type action_type;
2181
2182 assert(action);
2183 action_type = lttng_action_get_type(action);
2184 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2185 is_notify = true;
2186 goto end;
2187 } else if (action_type != LTTNG_ACTION_TYPE_GROUP) {
2188 goto end;
2189 }
2190
2191 action_status = lttng_action_group_get_count(action, &count);
2192 assert(action_status == LTTNG_ACTION_STATUS_OK);
2193
2194 for (i = 0; i < count; i++) {
2195 const struct lttng_action *inner_action =
2196 lttng_action_group_get_at_index(
2197 action, i);
2198
2199 action_type = lttng_action_get_type(inner_action);
2200 if (action_type == LTTNG_ACTION_TYPE_NOTIFY) {
2201 is_notify = true;
2202 goto end;
2203 }
2204 }
2205
2206 end:
2207 return is_notify;
2208 }
2209
2210 static bool trigger_name_taken(struct notification_thread_state *state,
2211 const struct lttng_trigger *trigger)
2212 {
2213 struct cds_lfht_iter iter;
2214
2215 /*
2216 * No duplicata is allowed in the triggers_by_name_uid_ht.
2217 * The match is done against the trigger name and uid.
2218 */
2219 cds_lfht_lookup(state->triggers_by_name_uid_ht,
2220 hash_trigger_by_name_uid(trigger),
2221 match_trigger_by_name_uid,
2222 trigger,
2223 &iter);
2224 return !!cds_lfht_iter_get_node(&iter);
2225 }
2226
2227 static
2228 enum lttng_error_code generate_trigger_name(
2229 struct notification_thread_state *state,
2230 struct lttng_trigger *trigger, const char **name)
2231 {
2232 enum lttng_error_code ret_code = LTTNG_OK;
2233 bool taken = false;
2234 enum lttng_trigger_status status;
2235
2236 do {
2237 const int ret = lttng_trigger_generate_name(trigger,
2238 state->trigger_id.name_offset++);
2239 if (ret) {
2240 /* The only reason this can fail right now. */
2241 ret_code = LTTNG_ERR_NOMEM;
2242 break;
2243 }
2244
2245 status = lttng_trigger_get_name(trigger, name);
2246 assert(status == LTTNG_TRIGGER_STATUS_OK);
2247
2248 taken = trigger_name_taken(state, trigger);
2249 } while (taken || state->trigger_id.name_offset == UINT64_MAX);
2250
2251 return ret_code;
2252 }
2253
2254 /*
2255 * FIXME A client's credentials are not checked when registering a trigger.
2256 *
2257 * The effects of this are benign since:
2258 * - The client will succeed in registering the trigger, as it is valid,
2259 * - The trigger will, internally, be bound to the channel/session,
2260 * - The notifications will not be sent since the client's credentials
2261 * are checked against the channel at that moment.
2262 *
2263 * If this function returns a non-zero value, it means something is
2264 * fundamentally broken and the whole subsystem/thread will be torn down.
2265 *
2266 * If a non-fatal error occurs, just set the cmd_result to the appropriate
2267 * error code.
2268 */
2269 static
2270 int handle_notification_thread_command_register_trigger(
2271 struct notification_thread_state *state,
2272 struct lttng_trigger *trigger,
2273 enum lttng_error_code *cmd_result)
2274 {
2275 int ret = 0;
2276 struct lttng_condition *condition;
2277 struct notification_client *client;
2278 struct notification_client_list *client_list = NULL;
2279 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2280 struct notification_client_list_element *client_list_element;
2281 struct cds_lfht_node *node;
2282 struct cds_lfht_iter iter;
2283 const char* trigger_name;
2284 bool free_trigger = true;
2285 struct lttng_evaluation *evaluation = NULL;
2286 struct lttng_credentials object_creds;
2287 uid_t object_uid;
2288 gid_t object_gid;
2289 enum action_executor_status executor_status;
2290 const uint64_t trigger_tracer_token =
2291 state->trigger_id.next_tracer_token++;
2292
2293 rcu_read_lock();
2294
2295 /* Set the trigger's tracer token. */
2296 lttng_trigger_set_tracer_token(trigger, trigger_tracer_token);
2297
2298 if (lttng_trigger_get_name(trigger, &trigger_name) ==
2299 LTTNG_TRIGGER_STATUS_UNSET) {
2300 const enum lttng_error_code ret_code = generate_trigger_name(
2301 state, trigger, &trigger_name);
2302
2303 if (ret_code != LTTNG_OK) {
2304 /* Fatal error. */
2305 ret = -1;
2306 *cmd_result = ret_code;
2307 goto error;
2308 }
2309 } else if (trigger_name_taken(state, trigger)) {
2310 /* Not a fatal error. */
2311 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2312 ret = 0;
2313 goto error;
2314 }
2315
2316 condition = lttng_trigger_get_condition(trigger);
2317 assert(condition);
2318
2319 ret = condition_is_supported(condition);
2320 if (ret < 0) {
2321 goto error;
2322 } else if (ret == 0) {
2323 *cmd_result = LTTNG_ERR_NOT_SUPPORTED;
2324 goto error;
2325 } else {
2326 /* Feature is supported, continue. */
2327 ret = 0;
2328 }
2329
2330 trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
2331 if (!trigger_ht_element) {
2332 ret = -1;
2333 goto error;
2334 }
2335
2336 /* Add trigger to the trigger_ht. */
2337 cds_lfht_node_init(&trigger_ht_element->node);
2338 cds_lfht_node_init(&trigger_ht_element->node_by_name_uid);
2339 trigger_ht_element->trigger = trigger;
2340
2341 node = cds_lfht_add_unique(state->triggers_ht,
2342 lttng_condition_hash(condition),
2343 match_trigger,
2344 trigger,
2345 &trigger_ht_element->node);
2346 if (node != &trigger_ht_element->node) {
2347 /* Not a fatal error, simply report it to the client. */
2348 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2349 goto error_free_ht_element;
2350 }
2351
2352 node = cds_lfht_add_unique(state->triggers_by_name_uid_ht,
2353 hash_trigger_by_name_uid(trigger),
2354 match_trigger_by_name_uid,
2355 trigger,
2356 &trigger_ht_element->node_by_name_uid);
2357 if (node != &trigger_ht_element->node_by_name_uid) {
2358 /* Not a fatal error, simply report it to the client. */
2359 cds_lfht_del(state->triggers_ht, &trigger_ht_element->node);
2360 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
2361 goto error_free_ht_element;
2362 }
2363
2364 /*
2365 * Ownership of the trigger and of its wrapper was transfered to
2366 * the triggers_ht.
2367 */
2368 trigger_ht_element = NULL;
2369 free_trigger = false;
2370
2371 /*
2372 * The rest only applies to triggers that have a "notify" action.
2373 * It is not skipped as this is the only action type currently
2374 * supported.
2375 */
2376 if (is_trigger_action_notify(trigger)) {
2377 client_list = notification_client_list_create(trigger);
2378 if (!client_list) {
2379 ret = -1;
2380 goto error_free_ht_element;
2381 }
2382
2383 /* Build a list of clients to which this new trigger applies. */
2384 cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
2385 client_socket_ht_node) {
2386 if (!trigger_applies_to_client(trigger, client)) {
2387 continue;
2388 }
2389
2390 client_list_element =
2391 zmalloc(sizeof(*client_list_element));
2392 if (!client_list_element) {
2393 ret = -1;
2394 goto error_put_client_list;
2395 }
2396
2397 CDS_INIT_LIST_HEAD(&client_list_element->node);
2398 client_list_element->client = client;
2399 cds_list_add(&client_list_element->node,
2400 &client_list->list);
2401 }
2402
2403 /*
2404 * Client list ownership transferred to the
2405 * notification_trigger_clients_ht.
2406 */
2407 publish_notification_client_list(state, client_list);
2408 }
2409
2410 switch (get_condition_binding_object(condition)) {
2411 case LTTNG_OBJECT_TYPE_SESSION:
2412 /* Add the trigger to the list if it matches a known session. */
2413 ret = bind_trigger_to_matching_session(trigger, state);
2414 if (ret) {
2415 goto error_put_client_list;
2416 }
2417 break;
2418 case LTTNG_OBJECT_TYPE_CHANNEL:
2419 /*
2420 * Add the trigger to list of triggers bound to the channels
2421 * currently known.
2422 */
2423 ret = bind_trigger_to_matching_channels(trigger, state);
2424 if (ret) {
2425 goto error_put_client_list;
2426 }
2427 break;
2428 case LTTNG_OBJECT_TYPE_NONE:
2429 break;
2430 default:
2431 ERR("Unknown object type on which to bind a newly registered trigger was encountered");
2432 ret = -1;
2433 goto error_put_client_list;
2434 }
2435
2436 /*
2437 * The new trigger's condition must be evaluated against the current
2438 * state.
2439 *
2440 * In the case of `notify` action, nothing preventing clients from
2441 * subscribing to a condition before the corresponding trigger is
2442 * registered, we have to evaluate this new condition right away.
2443 *
2444 * At some point, we were waiting for the next "evaluation" (e.g. on
2445 * reception of a channel sample) to evaluate this new condition, but
2446 * that was broken.
2447 *
2448 * The reason it was broken is that waiting for the next sample
2449 * does not allow us to properly handle transitions for edge-triggered
2450 * conditions.
2451 *
2452 * Consider this example: when we handle a new channel sample, we
2453 * evaluate each conditions twice: once with the previous state, and
2454 * again with the newest state. We then use those two results to
2455 * determine whether a state change happened: a condition was false and
2456 * became true. If a state change happened, we have to notify clients.
2457 *
2458 * Now, if a client subscribes to a given notification and registers
2459 * a trigger *after* that subscription, we have to make sure the
2460 * condition is evaluated at this point while considering only the
2461 * current state. Otherwise, the next evaluation cycle may only see
2462 * that the evaluations remain the same (true for samples n-1 and n) and
2463 * the client will never know that the condition has been met.
2464 */
2465 switch (get_condition_binding_object(condition)) {
2466 case LTTNG_OBJECT_TYPE_SESSION:
2467 ret = evaluate_session_condition_for_client(condition, state,
2468 &evaluation, &object_uid,
2469 &object_gid);
2470 break;
2471 case LTTNG_OBJECT_TYPE_CHANNEL:
2472 ret = evaluate_channel_condition_for_client(condition, state,
2473 &evaluation, &object_uid,
2474 &object_gid);
2475 break;
2476 case LTTNG_OBJECT_TYPE_NONE:
2477 ret = 0;
2478 break;
2479 case LTTNG_OBJECT_TYPE_UNKNOWN:
2480 default:
2481 ret = -1;
2482 break;
2483 }
2484
2485 if (ret) {
2486 /* Fatal error. */
2487 goto error_put_client_list;
2488 }
2489
2490 LTTNG_OPTIONAL_SET(&object_creds.uid, object_uid);
2491 LTTNG_OPTIONAL_SET(&object_creds.gid, object_gid);
2492
2493 DBG("Newly registered trigger's condition evaluated to %s",
2494 evaluation ? "true" : "false");
2495 if (!evaluation) {
2496 /* Evaluation yielded nothing. Normal exit. */
2497 ret = 0;
2498 goto end;
2499 }
2500
2501 /*
2502 * Ownership of `evaluation` transferred to the action executor
2503 * no matter the result.
2504 */
2505 executor_status = action_executor_enqueue(state->executor, trigger,
2506 evaluation, &object_creds, client_list);
2507 evaluation = NULL;
2508 switch (executor_status) {
2509 case ACTION_EXECUTOR_STATUS_OK:
2510 break;
2511 case ACTION_EXECUTOR_STATUS_ERROR:
2512 case ACTION_EXECUTOR_STATUS_INVALID:
2513 /*
2514 * TODO Add trigger identification (name/id) when
2515 * it is added to the API.
2516 */
2517 ERR("Fatal error occurred while enqueuing action associated to newly registered trigger");
2518 ret = -1;
2519 goto error_put_client_list;
2520 case ACTION_EXECUTOR_STATUS_OVERFLOW:
2521 /*
2522 * TODO Add trigger identification (name/id) when
2523 * it is added to the API.
2524 *
2525 * Not a fatal error.
2526 */
2527 WARN("No space left when enqueuing action associated to newly registered trigger");
2528 ret = 0;
2529 goto end;
2530 default:
2531 abort();
2532 }
2533
2534 end:
2535 *cmd_result = LTTNG_OK;
2536 DBG("Registered trigger: name = `%s`, tracer token = %" PRIu64,
2537 trigger_name, trigger_tracer_token);
2538
2539 error_put_client_list:
2540 notification_client_list_put(client_list);
2541
2542 error_free_ht_element:
2543 if (trigger_ht_element) {
2544 /* Delayed removal due to RCU constraint on delete. */
2545 call_rcu(&trigger_ht_element->rcu_node,
2546 free_lttng_trigger_ht_element_rcu);
2547 }
2548
2549 error:
2550 if (free_trigger) {
2551 lttng_trigger_destroy(trigger);
2552 }
2553 rcu_read_unlock();
2554 return ret;
2555 }
2556
2557 static
2558 void free_lttng_trigger_ht_element_rcu(struct rcu_head *node)
2559 {
2560 free(caa_container_of(node, struct lttng_trigger_ht_element,
2561 rcu_node));
2562 }
2563
2564 static
2565 int handle_notification_thread_command_unregister_trigger(
2566 struct notification_thread_state *state,
2567 struct lttng_trigger *trigger,
2568 enum lttng_error_code *_cmd_reply)
2569 {
2570 struct cds_lfht_iter iter;
2571 struct cds_lfht_node *triggers_ht_node;
2572 struct lttng_channel_trigger_list *trigger_list;
2573 struct notification_client_list *client_list;
2574 struct lttng_trigger_ht_element *trigger_ht_element = NULL;
2575 struct lttng_condition *condition = lttng_trigger_get_condition(
2576 trigger);
2577 enum lttng_error_code cmd_reply;
2578
2579 rcu_read_lock();
2580
2581 cds_lfht_lookup(state->triggers_ht,
2582 lttng_condition_hash(condition),
2583 match_trigger,
2584 trigger,
2585 &iter);
2586 triggers_ht_node = cds_lfht_iter_get_node(&iter);
2587 if (!triggers_ht_node) {
2588 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
2589 goto end;
2590 } else {
2591 cmd_reply = LTTNG_OK;
2592 }
2593
2594 /* Remove trigger from channel_triggers_ht. */
2595 cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
2596 channel_triggers_ht_node) {
2597 struct lttng_trigger_list_element *trigger_element, *tmp;
2598
2599 cds_list_for_each_entry_safe(trigger_element, tmp,
2600 &trigger_list->list, node) {
2601 if (!lttng_trigger_is_equal(trigger, trigger_element->trigger)) {
2602 continue;
2603 }
2604
2605 DBG("[notification-thread] Removed trigger from channel_triggers_ht");
2606 cds_list_del(&trigger_element->node);
2607 /* A trigger can only appear once per channel */
2608 break;
2609 }
2610 }
2611
2612 /*
2613 * Remove and release the client list from
2614 * notification_trigger_clients_ht.
2615 */
2616 client_list = get_client_list_from_condition(state, condition);
2617 assert(client_list);
2618
2619 /* Put new reference and the hashtable's reference. */
2620 notification_client_list_put(client_list);
2621 notification_client_list_put(client_list);
2622 client_list = NULL;
2623
2624 /* Remove trigger from triggers_ht. */
2625 trigger_ht_element = caa_container_of(triggers_ht_node,
2626 struct lttng_trigger_ht_element, node);
2627 cds_lfht_del(state->triggers_by_name_uid_ht, &trigger_ht_element->node_by_name_uid);
2628 cds_lfht_del(state->triggers_ht, triggers_ht_node);
2629
2630 /* Release the ownership of the trigger. */
2631 lttng_trigger_destroy(trigger_ht_element->trigger);
2632 call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu);
2633 end:
2634 rcu_read_unlock();
2635 if (_cmd_reply) {
2636 *_cmd_reply = cmd_reply;
2637 }
2638 return 0;
2639 }
2640
2641 /* Returns 0 on success, 1 on exit requested, negative value on error. */
2642 int handle_notification_thread_command(
2643 struct notification_thread_handle *handle,
2644 struct notification_thread_state *state)
2645 {
2646 int ret;
2647 uint64_t counter;
2648 struct notification_thread_command *cmd;
2649
2650 /* Read the event pipe to put it back into a quiescent state. */
2651 ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
2652 sizeof(counter));
2653 if (ret != sizeof(counter)) {
2654 goto error;
2655 }
2656
2657 pthread_mutex_lock(&handle->cmd_queue.lock);
2658 cmd = cds_list_first_entry(&handle->cmd_queue.list,
2659 struct notification_thread_command, cmd_list_node);
2660 cds_list_del(&cmd->cmd_list_node);
2661 pthread_mutex_unlock(&handle->cmd_queue.lock);
2662 switch (cmd->type) {
2663 case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
2664 DBG("[notification-thread] Received register trigger command");
2665 ret = handle_notification_thread_command_register_trigger(
2666 state, cmd->parameters.trigger,
2667 &cmd->reply_code);
2668 break;
2669 case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
2670 DBG("[notification-thread] Received unregister trigger command");
2671 ret = handle_notification_thread_command_unregister_trigger(
2672 state, cmd->parameters.trigger,
2673 &cmd->reply_code);
2674 break;
2675 case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
2676 DBG("[notification-thread] Received add channel command");
2677 ret = handle_notification_thread_command_add_channel(
2678 state,
2679 cmd->parameters.add_channel.session.name,
2680 cmd->parameters.add_channel.session.uid,
2681 cmd->parameters.add_channel.session.gid,
2682 cmd->parameters.add_channel.channel.name,
2683 cmd->parameters.add_channel.channel.domain,
2684 cmd->parameters.add_channel.channel.key,
2685 cmd->parameters.add_channel.channel.capacity,
2686 &cmd->reply_code);
2687 break;
2688 case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
2689 DBG("[notification-thread] Received remove channel command");
2690 ret = handle_notification_thread_command_remove_channel(
2691 state, cmd->parameters.remove_channel.key,
2692 cmd->parameters.remove_channel.domain,
2693 &cmd->reply_code);
2694 break;
2695 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
2696 case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
2697 DBG("[notification-thread] Received session rotation %s command",
2698 cmd->type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING ?
2699 "ongoing" : "completed");
2700 ret = handle_notification_thread_command_session_rotation(
2701 state,
2702 cmd->type,
2703 cmd->parameters.session_rotation.session_name,
2704 cmd->parameters.session_rotation.uid,
2705 cmd->parameters.session_rotation.gid,
2706 cmd->parameters.session_rotation.trace_archive_chunk_id,
2707 cmd->parameters.session_rotation.location,
2708 &cmd->reply_code);
2709 break;
2710 case NOTIFICATION_COMMAND_TYPE_LIST_TRIGGERS:
2711 {
2712 struct lttng_triggers *triggers = NULL;
2713
2714 ret = handle_notification_thread_command_list_triggers(
2715 handle,
2716 state,
2717 cmd->parameters.list_triggers.uid,
2718 &triggers,
2719 &cmd->reply_code);
2720 cmd->reply.list_triggers.triggers = triggers;
2721 ret = 0;
2722 break;
2723 }
2724 case NOTIFICATION_COMMAND_TYPE_QUIT:
2725 DBG("[notification-thread] Received quit command");
2726 cmd->reply_code = LTTNG_OK;
2727 ret = 1;
2728 goto end;
2729 case NOTIFICATION_COMMAND_TYPE_CLIENT_COMMUNICATION_UPDATE:
2730 {
2731 const enum client_transmission_status client_status =
2732 cmd->parameters.client_communication_update
2733 .status;
2734 const notification_client_id client_id =
2735 cmd->parameters.client_communication_update.id;
2736 struct notification_client *client;
2737
2738 rcu_read_lock();
2739 client = get_client_from_id(client_id, state);
2740
2741 if (!client) {
2742 /*
2743 * Client error was probably already picked-up by the
2744 * notification thread or it has disconnected
2745 * gracefully while this command was queued.
2746 */
2747 DBG("Failed to find notification client to update communication status, client id = %" PRIu64,
2748 client_id);
2749 ret = 0;
2750 } else {
2751 ret = client_handle_transmission_status(
2752 client, client_status, state);
2753 }
2754 rcu_read_unlock();
2755 break;
2756 }
2757 default:
2758 ERR("[notification-thread] Unknown internal command received");
2759 goto error_unlock;
2760 }
2761
2762 if (ret) {
2763 goto error_unlock;
2764 }
2765 end:
2766 if (cmd->is_async) {
2767 free(cmd);
2768 cmd = NULL;
2769 } else {
2770 lttng_waiter_wake_up(&cmd->reply_waiter);
2771 }
2772 return ret;
2773 error_unlock:
2774 /* Wake-up and return a fatal error to the calling thread. */
2775 lttng_waiter_wake_up(&cmd->reply_waiter);
2776 cmd->reply_code = LTTNG_ERR_FATAL;
2777 error:
2778 /* Indicate a fatal error to the caller. */
2779 return -1;
2780 }
2781
2782 static
2783 int socket_set_non_blocking(int socket)
2784 {
2785 int ret, flags;
2786
2787 /* Set the pipe as non-blocking. */
2788 ret = fcntl(socket, F_GETFL, 0);
2789 if (ret == -1) {
2790 PERROR("fcntl get socket flags");
2791 goto end;
2792 }
2793 flags = ret;
2794
2795 ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
2796 if (ret == -1) {
2797 PERROR("fcntl set O_NONBLOCK socket flag");
2798 goto end;
2799 }
2800 DBG("Client socket (fd = %i) set as non-blocking", socket);
2801 end:
2802 return ret;
2803 }
2804
2805 static
2806 int client_reset_inbound_state(struct notification_client *client)
2807 {
2808 int ret;
2809
2810
2811 lttng_payload_clear(&client->communication.inbound.payload);
2812
2813 client->communication.inbound.bytes_to_receive =
2814 sizeof(struct lttng_notification_channel_message);
2815 client->communication.inbound.msg_type =
2816 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
2817 LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
2818 LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
2819 ret = lttng_dynamic_buffer_set_size(
2820 &client->communication.inbound.payload.buffer,
2821 client->communication.inbound.bytes_to_receive);
2822
2823 return ret;
2824 }
2825
2826 int handle_notification_thread_client_connect(
2827 struct notification_thread_state *state)
2828 {
2829 int ret;
2830 struct notification_client *client;
2831
2832 DBG("[notification-thread] Handling new notification channel client connection");
2833
2834 client = zmalloc(sizeof(*client));
2835 if (!client) {
2836 /* Fatal error. */
2837 ret = -1;
2838 goto error;
2839 }
2840
2841 pthread_mutex_init(&client->lock, NULL);
2842 client->id = state->next_notification_client_id++;
2843 CDS_INIT_LIST_HEAD(&client->condition_list);
2844 lttng_payload_init(&client->communication.inbound.payload);
2845 lttng_payload_init(&client->communication.outbound.payload);
2846 client->communication.inbound.expect_creds = true;
2847
2848 ret = client_reset_inbound_state(client);
2849 if (ret) {
2850 ERR("[notification-thread] Failed to reset client communication's inbound state");
2851 ret = 0;
2852 goto error;
2853 }
2854
2855 ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
2856 if (ret < 0) {
2857 ERR("[notification-thread] Failed to accept new notification channel client connection");
2858 ret = 0;
2859 goto error;
2860 }
2861
2862 client->socket = ret;
2863
2864 ret = socket_set_non_blocking(client->socket);
2865 if (ret) {
2866 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
2867 goto error;
2868 }
2869
2870 ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
2871 if (ret < 0) {
2872 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
2873 ret = 0;
2874 goto error;
2875 }
2876
2877 ret = lttng_poll_add(&state->events, client->socket,
2878 LPOLLIN | LPOLLERR |
2879 LPOLLHUP | LPOLLRDHUP);
2880 if (ret < 0) {
2881 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
2882 ret = 0;
2883 goto error;
2884 }
2885 DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
2886 client->socket);
2887
2888 rcu_read_lock();
2889 cds_lfht_add(state->client_socket_ht,
2890 hash_client_socket(client->socket),
2891 &client->client_socket_ht_node);
2892 cds_lfht_add(state->client_id_ht,
2893 hash_client_id(client->id),
2894 &client->client_id_ht_node);
2895 rcu_read_unlock();
2896
2897 return ret;
2898
2899 error:
2900 notification_client_destroy(client, state);
2901 return ret;
2902 }
2903
2904 /*
2905 * RCU read-lock must be held by the caller.
2906 * Client lock must _not_ be held by the caller.
2907 */
2908 static
2909 int notification_thread_client_disconnect(
2910 struct notification_client *client,
2911 struct notification_thread_state *state)
2912 {
2913 int ret;
2914 struct lttng_condition_list_element *condition_list_element, *tmp;
2915
2916 /* Acquire the client lock to disable its communication atomically. */
2917 pthread_mutex_lock(&client->lock);
2918 client->communication.active = false;
2919 cds_lfht_del(state->client_socket_ht, &client->client_socket_ht_node);
2920 cds_lfht_del(state->client_id_ht, &client->client_id_ht_node);
2921 pthread_mutex_unlock(&client->lock);
2922
2923 ret = lttng_poll_del(&state->events, client->socket);
2924 if (ret) {
2925 ERR("[notification-thread] Failed to remove client socket %d from poll set",
2926 client->socket);
2927 }
2928
2929 /* Release all conditions to which the client was subscribed. */
2930 cds_list_for_each_entry_safe(condition_list_element, tmp,
2931 &client->condition_list, node) {
2932 (void) notification_thread_client_unsubscribe(client,
2933 condition_list_element->condition, state, NULL);
2934 }
2935
2936 /*
2937 * Client no longer accessible to other threads (through the
2938 * client lists).
2939 */
2940 notification_client_destroy(client, state);
2941 return ret;
2942 }
2943
2944 int handle_notification_thread_client_disconnect(
2945 int client_socket, struct notification_thread_state *state)
2946 {
2947 int ret = 0;
2948 struct notification_client *client;
2949
2950 rcu_read_lock();
2951 DBG("[notification-thread] Closing client connection (socket fd = %i)",
2952 client_socket);
2953 client = get_client_from_socket(client_socket, state);
2954 if (!client) {
2955 /* Internal state corruption, fatal error. */
2956 ERR("[notification-thread] Unable to find client (socket fd = %i)",
2957 client_socket);
2958 ret = -1;
2959 goto end;
2960 }
2961
2962 ret = notification_thread_client_disconnect(client, state);
2963 end:
2964 rcu_read_unlock();
2965 return ret;
2966 }
2967
2968 int handle_notification_thread_client_disconnect_all(
2969 struct notification_thread_state *state)
2970 {
2971 struct cds_lfht_iter iter;
2972 struct notification_client *client;
2973 bool error_encoutered = false;
2974
2975 rcu_read_lock();
2976 DBG("[notification-thread] Closing all client connections");
2977 cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
2978 client_socket_ht_node) {
2979 int ret;
2980
2981 ret = notification_thread_client_disconnect(
2982 client, state);
2983 if (ret) {
2984 error_encoutered = true;
2985 }
2986 }
2987 rcu_read_unlock();
2988 return error_encoutered ? 1 : 0;
2989 }
2990
2991 int handle_notification_thread_trigger_unregister_all(
2992 struct notification_thread_state *state)
2993 {
2994 bool error_occurred = false;
2995 struct cds_lfht_iter iter;
2996 struct lttng_trigger_ht_element *trigger_ht_element;
2997
2998 rcu_read_lock();
2999 cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
3000 node) {
3001 int ret = handle_notification_thread_command_unregister_trigger(
3002 state, trigger_ht_element->trigger, NULL);
3003 if (ret) {
3004 error_occurred = true;
3005 }
3006 }
3007 rcu_read_unlock();
3008 return error_occurred ? -1 : 0;
3009 }
3010
3011 static
3012 int client_handle_transmission_status(
3013 struct notification_client *client,
3014 enum client_transmission_status transmission_status,
3015 struct notification_thread_state *state)
3016 {
3017 int ret = 0;
3018
3019 switch (transmission_status) {
3020 case CLIENT_TRANSMISSION_STATUS_COMPLETE:
3021 ret = lttng_poll_mod(&state->events, client->socket,
3022 CLIENT_POLL_MASK_IN);
3023 if (ret) {
3024 goto end;
3025 }
3026
3027 break;
3028 case CLIENT_TRANSMISSION_STATUS_QUEUED:
3029 /*
3030 * We want to be notified whenever there is buffer space
3031 * available to send the rest of the payload.
3032 */
3033 ret = lttng_poll_mod(&state->events, client->socket,
3034 CLIENT_POLL_MASK_IN_OUT);
3035 if (ret) {
3036 goto end;
3037 }
3038 break;
3039 case CLIENT_TRANSMISSION_STATUS_FAIL:
3040 ret = notification_thread_client_disconnect(client, state);
3041 if (ret) {
3042 goto end;
3043 }
3044 break;