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