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