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