src/bin/lttng-sessiond/notification-thread-events.c

   1 /*
   2  * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License, version 2 only, as
   6  * published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful, but WITHOUT
   9  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  10  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  11  * more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along with
  14  * this program; if not, write to the Free Software Foundation, Inc., 51
  15  * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _LGPL_SOURCE
  19 #include <urcu.h>
  20 #include <urcu/rculfhash.h>
  21
  22 #include "notification-thread.h"
  23 #include "notification-thread-events.h"
  24 #include "notification-thread-commands.h"
  25 #include <common/defaults.h>
  26 #include <common/error.h>
  27 #include <common/futex.h>
  28 #include <common/unix.h>
  29 #include <common/dynamic-buffer.h>
  30 #include <common/hashtable/utils.h>
  31 #include <common/sessiond-comm/sessiond-comm.h>
  32 #include <common/macros.h>
  33 #include <lttng/condition/condition.h>
  34 #include <lttng/action/action.h>
  35 #include <lttng/notification/notification-internal.h>
  36 #include <lttng/condition/condition-internal.h>
  37 #include <lttng/condition/buffer-usage-internal.h>
  38 #include <lttng/notification/channel-internal.h>
  39 #include <time.h>
  40 #include <unistd.h>
  41 #include <assert.h>
  42 #include <inttypes.h>
  43 #include <fcntl.h>
  44
  45 #define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
  46 #define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
  47
  48 struct lttng_trigger_list_element {
  49         struct lttng_trigger *trigger;
  50         struct cds_list_head node;
  51 };
  52
  53 struct lttng_channel_trigger_list {
  54         struct channel_key channel_key;
  55         struct cds_list_head list;
  56         struct cds_lfht_node channel_triggers_ht_node;
  57 };
  58
  59 struct lttng_trigger_ht_element {
  60         struct lttng_trigger *trigger;
  61         struct cds_lfht_node node;
  62 };
  63
  64 struct lttng_condition_list_element {
  65         struct lttng_condition *condition;
  66         struct cds_list_head node;
  67 };
  68
  69 struct notification_client_list_element {
  70         struct notification_client *client;
  71         struct cds_list_head node;
  72 };
  73
  74 struct notification_client_list {
  75         struct lttng_trigger *trigger;
  76         struct cds_list_head list;
  77         struct cds_lfht_node notification_trigger_ht_node;
  78 };
  79
  80 struct notification_client {
  81         int socket;
  82         /* Client protocol version. */
  83         uint8_t major, minor;
  84         uid_t uid;
  85         gid_t gid;
  86         /*
  87          * Indicates if the credentials and versions of the client has been
  88          * checked.
  89          */
  90         bool validated;
  91         /*
  92          * Conditions to which the client's notification channel is subscribed.
  93          * List of struct lttng_condition_list_node. The condition member is
  94          * owned by the client.
  95          */
  96         struct cds_list_head condition_list;
  97         struct cds_lfht_node client_socket_ht_node;
  98         struct {
  99                 struct {
 100                         struct lttng_dynamic_buffer buffer;
 101                         /* Bytes left to receive for the current message. */
 102                         size_t bytes_to_receive;
 103                         /* Type of the message being received. */
 104                         enum lttng_notification_channel_message_type msg_type;
 105                         /*
 106                          * Indicates whether or not credentials are expected
 107                          * from the client.
 108                          */
 109                         bool receive_creds;
 110                         /*
 111                          * Indicates whether or not credentials were received
 112                          * from the client.
 113                          */
 114                         bool creds_received;
 115                         lttng_sock_cred creds;
 116                 } inbound;
 117                 struct {
 118                         /*
 119                          * Indicates whether or not a notification addressed to
 120                          * this client was dropped because a command reply was
 121                          * already buffered.
 122                          *
 123                          * A notification is dropped whenever the buffer is not
 124                          * empty.
 125                          */
 126                         bool dropped_notification;
 127                         /*
 128                          * Indicates whether or not a command reply is already
 129                          * buffered. In this case, it means that the client is
 130                          * not consuming command replies before emitting a new
 131                          * one. This could be caused by a protocol error or a
 132                          * misbehaving/malicious client.
 133                          */
 134                         bool queued_command_reply;
 135                         struct lttng_dynamic_buffer buffer;
 136                 } outbound;
 137         } communication;
 138 };
 139
 140 struct channel_state_sample {
 141         struct channel_key key;
 142         struct cds_lfht_node channel_state_ht_node;
 143         uint64_t highest_usage;
 144         uint64_t lowest_usage;
 145 };
 146
 147 static
 148 int match_client(struct cds_lfht_node *node, const void *key)
 149 {
 150         /* This double-cast is intended to supress pointer-to-cast warning. */
 151         int socket = (int) (intptr_t) key;
 152         struct notification_client *client;
 153
 154         client = caa_container_of(node, struct notification_client,
 155                         client_socket_ht_node);
 156
 157         return !!(client->socket == socket);
 158 }
 159
 160 static
 161 int match_channel_trigger_list(struct cds_lfht_node *node, const void *key)
 162 {
 163         struct channel_key *channel_key = (struct channel_key *) key;
 164         struct lttng_channel_trigger_list *trigger_list;
 165
 166         trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
 167                         channel_triggers_ht_node);
 168
 169         return !!((channel_key->key == trigger_list->channel_key.key) &&
 170                         (channel_key->domain == trigger_list->channel_key.domain));
 171 }
 172
 173 static
 174 int match_channel_state_sample(struct cds_lfht_node *node, const void *key)
 175 {
 176         struct channel_key *channel_key = (struct channel_key *) key;
 177         struct channel_state_sample *sample;
 178
 179         sample = caa_container_of(node, struct channel_state_sample,
 180                         channel_state_ht_node);
 181
 182         return !!((channel_key->key == sample->key.key) &&
 183                         (channel_key->domain == sample->key.domain));
 184 }
 185
 186 static
 187 int match_channel_info(struct cds_lfht_node *node, const void *key)
 188 {
 189         struct channel_key *channel_key = (struct channel_key *) key;
 190         struct channel_info *channel_info;
 191
 192         channel_info = caa_container_of(node, struct channel_info,
 193                         channels_ht_node);
 194
 195         return !!((channel_key->key == channel_info->key.key) &&
 196                         (channel_key->domain == channel_info->key.domain));
 197 }
 198
 199 static
 200 int match_condition(struct cds_lfht_node *node, const void *key)
 201 {
 202         struct lttng_condition *condition_key = (struct lttng_condition *) key;
 203         struct lttng_trigger_ht_element *trigger;
 204         struct lttng_condition *condition;
 205
 206         trigger = caa_container_of(node, struct lttng_trigger_ht_element,
 207                         node);
 208         condition = lttng_trigger_get_condition(trigger->trigger);
 209         assert(condition);
 210
 211         return !!lttng_condition_is_equal(condition_key, condition);
 212 }
 213
 214 static
 215 int match_client_list(struct cds_lfht_node *node, const void *key)
 216 {
 217         struct lttng_trigger *trigger_key = (struct lttng_trigger *) key;
 218         struct notification_client_list *client_list;
 219         struct lttng_condition *condition;
 220         struct lttng_condition *condition_key = lttng_trigger_get_condition(
 221                         trigger_key);
 222
 223         assert(condition_key);
 224
 225         client_list = caa_container_of(node, struct notification_client_list,
 226                         notification_trigger_ht_node);
 227         condition = lttng_trigger_get_condition(client_list->trigger);
 228
 229         return !!lttng_condition_is_equal(condition_key, condition);
 230 }
 231
 232 static
 233 int match_client_list_condition(struct cds_lfht_node *node, const void *key)
 234 {
 235         struct lttng_condition *condition_key = (struct lttng_condition *) key;
 236         struct notification_client_list *client_list;
 237         struct lttng_condition *condition;
 238
 239         assert(condition_key);
 240
 241         client_list = caa_container_of(node, struct notification_client_list,
 242                         notification_trigger_ht_node);
 243         condition = lttng_trigger_get_condition(client_list->trigger);
 244
 245         return !!lttng_condition_is_equal(condition_key, condition);
 246 }
 247
 248 static
 249 unsigned long lttng_condition_buffer_usage_hash(
 250         struct lttng_condition *_condition)
 251 {
 252         unsigned long hash = 0;
 253         struct lttng_condition_buffer_usage *condition;
 254
 255         condition = container_of(_condition,
 256                         struct lttng_condition_buffer_usage, parent);
 257
 258         if (condition->session_name) {
 259                 hash ^= hash_key_str(condition->session_name, lttng_ht_seed);
 260         }
 261         if (condition->channel_name) {
 262                 hash ^= hash_key_str(condition->channel_name, lttng_ht_seed);
 263         }
 264         if (condition->domain.set) {
 265                 hash ^= hash_key_ulong(
 266                                 (void *) condition->domain.type,
 267                                 lttng_ht_seed);
 268         }
 269         if (condition->threshold_ratio.set) {
 270                 uint64_t val;
 271
 272                 val = condition->threshold_ratio.value * (double) UINT32_MAX;
 273                 hash ^= hash_key_u64(&val, lttng_ht_seed);
 274         } else if (condition->threshold_ratio.set) {
 275                 uint64_t val;
 276
 277                 val = condition->threshold_bytes.value;
 278                 hash ^= hash_key_u64(&val, lttng_ht_seed);
 279         }
 280         return hash;
 281 }
 282
 283 /*
 284  * The lttng_condition hashing code is kept in this file (rather than
 285  * condition.c) since it makes use of GPLv2 code (hashtable utils), which we
 286  * don't want to link in liblttng-ctl.
 287  */
 288 static
 289 unsigned long lttng_condition_hash(struct lttng_condition *condition)
 290 {
 291         switch (condition->type) {
 292         case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
 293         case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
 294                 return lttng_condition_buffer_usage_hash(condition);
 295         default:
 296                 ERR("[notification-thread] Unexpected condition type caught");
 297                 abort();
 298         }
 299 }
 300
 301 static
 302 void channel_info_destroy(struct channel_info *channel_info)
 303 {
 304         if (!channel_info) {
 305                 return;
 306         }
 307
 308         if (channel_info->session_name) {
 309                 free(channel_info->session_name);
 310         }
 311         if (channel_info->channel_name) {
 312                 free(channel_info->channel_name);
 313         }
 314         free(channel_info);
 315 }
 316
 317 static
 318 struct channel_info *channel_info_copy(struct channel_info *channel_info)
 319 {
 320         struct channel_info *copy = zmalloc(sizeof(*channel_info));
 321
 322         assert(channel_info);
 323         assert(channel_info->session_name);
 324         assert(channel_info->channel_name);
 325
 326         if (!copy) {
 327                 goto end;
 328         }
 329
 330         memcpy(copy, channel_info, sizeof(*channel_info));
 331         copy->session_name = NULL;
 332         copy->channel_name = NULL;
 333
 334         copy->session_name = strdup(channel_info->session_name);
 335         if (!copy->session_name) {
 336                 goto error;
 337         }
 338         copy->channel_name = strdup(channel_info->channel_name);
 339         if (!copy->channel_name) {
 340                 goto error;
 341         }
 342         cds_lfht_node_init(&channel_info->channels_ht_node);
 343 end:
 344         return copy;
 345 error:
 346         channel_info_destroy(copy);
 347         return NULL;
 348 }
 349
 350 static
 351 int notification_thread_client_subscribe(struct notification_client *client,
 352                 struct lttng_condition *condition,
 353                 struct notification_thread_state *state,
 354                 enum lttng_notification_channel_status *_status)
 355 {
 356         int ret = 0;
 357         struct cds_lfht_iter iter;
 358         struct cds_lfht_node *node;
 359         struct notification_client_list *client_list;
 360         struct lttng_condition_list_element *condition_list_element = NULL;
 361         struct notification_client_list_element *client_list_element = NULL;
 362         enum lttng_notification_channel_status status =
 363                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 364
 365         /*
 366          * Ensure that the client has not already subscribed to this condition
 367          * before.
 368          */
 369         cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
 370                 if (lttng_condition_is_equal(condition_list_element->condition,
 371                                 condition)) {
 372                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
 373                         goto end;
 374                 }
 375         }
 376
 377         condition_list_element = zmalloc(sizeof(*condition_list_element));
 378         if (!condition_list_element) {
 379                 ret = -1;
 380                 goto error;
 381         }
 382         client_list_element = zmalloc(sizeof(*client_list_element));
 383         if (!client_list_element) {
 384                 ret = -1;
 385                 goto error;
 386         }
 387
 388         rcu_read_lock();
 389
 390         /*
 391          * Add the newly-subscribed condition to the client's subscription list.
 392          */
 393         CDS_INIT_LIST_HEAD(&condition_list_element->node);
 394         condition_list_element->condition = condition;
 395         cds_list_add(&condition_list_element->node, &client->condition_list);
 396
 397         /*
 398          * Add the client to the list of clients interested in a given trigger
 399          * if a "notification" trigger with a corresponding condition was
 400          * added prior.
 401          */
 402         cds_lfht_lookup(state->notification_trigger_clients_ht,
 403                         lttng_condition_hash(condition),
 404                         match_client_list_condition,
 405                         condition,
 406                         &iter);
 407         node = cds_lfht_iter_get_node(&iter);
 408         if (!node) {
 409                 free(client_list_element);
 410                 goto end_unlock;
 411         }
 412
 413         client_list = caa_container_of(node, struct notification_client_list,
 414                         notification_trigger_ht_node);
 415         client_list_element->client = client;
 416         CDS_INIT_LIST_HEAD(&client_list_element->node);
 417         cds_list_add(&client_list_element->node, &client_list->list);
 418 end_unlock:
 419         rcu_read_unlock();
 420 end:
 421         if (_status) {
 422                 *_status = status;
 423         }
 424         return ret;
 425 error:
 426         free(condition_list_element);
 427         free(client_list_element);
 428         return ret;
 429 }
 430
 431 static
 432 int notification_thread_client_unsubscribe(
 433                 struct notification_client *client,
 434                 struct lttng_condition *condition,
 435                 struct notification_thread_state *state,
 436                 enum lttng_notification_channel_status *_status)
 437 {
 438         struct cds_lfht_iter iter;
 439         struct cds_lfht_node *node;
 440         struct notification_client_list *client_list;
 441         struct lttng_condition_list_element *condition_list_element,
 442                         *condition_tmp;
 443         struct notification_client_list_element *client_list_element,
 444                         *client_tmp;
 445         bool condition_found = false;
 446         enum lttng_notification_channel_status status =
 447                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 448
 449         /* Remove the condition from the client's condition list. */
 450         cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
 451                         &client->condition_list, node) {
 452                 if (!lttng_condition_is_equal(condition_list_element->condition,
 453                                 condition)) {
 454                         continue;
 455                 }
 456
 457                 cds_list_del(&condition_list_element->node);
 458                 /*
 459                  * The caller may be iterating on the client's conditions to
 460                  * tear down a client's connection. In this case, the condition
 461                  * will be destroyed at the end.
 462                  */
 463                 if (condition != condition_list_element->condition) {
 464                         lttng_condition_destroy(
 465                                         condition_list_element->condition);
 466                 }
 467                 free(condition_list_element);
 468                 condition_found = true;
 469                 break;
 470         }
 471
 472         if (!condition_found) {
 473                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
 474                 goto end;
 475         }
 476
 477         /*
 478          * Remove the client from the list of clients interested the trigger
 479          * matching the condition.
 480          */
 481         rcu_read_lock();
 482         cds_lfht_lookup(state->notification_trigger_clients_ht,
 483                         lttng_condition_hash(condition),
 484                         match_client_list_condition,
 485                         condition,
 486                         &iter);
 487         node = cds_lfht_iter_get_node(&iter);
 488         if (!node) {
 489                 goto end_unlock;
 490         }
 491
 492         client_list = caa_container_of(node, struct notification_client_list,
 493                         notification_trigger_ht_node);
 494         cds_list_for_each_entry_safe(client_list_element, client_tmp,
 495                         &client_list->list, node) {
 496                 if (client_list_element->client->socket != client->socket) {
 497                         continue;
 498                 }
 499                 cds_list_del(&client_list_element->node);
 500                 free(client_list_element);
 501                 break;
 502         }
 503 end_unlock:
 504         rcu_read_unlock();
 505 end:
 506         lttng_condition_destroy(condition);
 507         if (_status) {
 508                 *_status = status;
 509         }
 510         return 0;
 511 }
 512
 513 static
 514 void notification_client_destroy(struct notification_client *client,
 515                 struct notification_thread_state *state)
 516 {
 517         struct lttng_condition_list_element *condition_list_element, *tmp;
 518
 519         if (!client) {
 520                 return;
 521         }
 522
 523         /* Release all conditions to which the client was subscribed. */
 524         cds_list_for_each_entry_safe(condition_list_element, tmp,
 525                         &client->condition_list, node) {
 526                 (void) notification_thread_client_unsubscribe(client,
 527                                 condition_list_element->condition, state, NULL);
 528         }
 529
 530         if (client->socket >= 0) {
 531                 (void) lttcomm_close_unix_sock(client->socket);
 532         }
 533         lttng_dynamic_buffer_reset(&client->communication.inbound.buffer);
 534         lttng_dynamic_buffer_reset(&client->communication.outbound.buffer);
 535         free(client);
 536 }
 537
 538 /*
 539  * Call with rcu_read_lock held (and hold for the lifetime of the returned
 540  * client pointer).
 541  */
 542 static
 543 struct notification_client *get_client_from_socket(int socket,
 544                 struct notification_thread_state *state)
 545 {
 546         struct cds_lfht_iter iter;
 547         struct cds_lfht_node *node;
 548         struct notification_client *client = NULL;
 549
 550         cds_lfht_lookup(state->client_socket_ht,
 551                         hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed),
 552                         match_client,
 553                         (void *) (unsigned long) socket,
 554                         &iter);
 555         node = cds_lfht_iter_get_node(&iter);
 556         if (!node) {
 557                 goto end;
 558         }
 559
 560         client = caa_container_of(node, struct notification_client,
 561                         client_socket_ht_node);
 562 end:
 563         return client;
 564 }
 565
 566 static
 567 bool trigger_applies_to_channel(struct lttng_trigger *trigger,
 568                 struct channel_info *info)
 569 {
 570         enum lttng_condition_status status;
 571         struct lttng_condition *condition;
 572         const char *trigger_session_name = NULL;
 573         const char *trigger_channel_name = NULL;
 574         enum lttng_domain_type trigger_domain;
 575
 576         condition = lttng_trigger_get_condition(trigger);
 577         if (!condition) {
 578                 goto fail;
 579         }
 580
 581         switch (lttng_condition_get_type(condition)) {
 582         case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
 583         case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
 584                 break;
 585         default:
 586                 goto fail;
 587         }
 588
 589         status = lttng_condition_buffer_usage_get_domain_type(condition,
 590                         &trigger_domain);
 591         assert(status == LTTNG_CONDITION_STATUS_OK);
 592         if (info->key.domain != trigger_domain) {
 593                 goto fail;
 594         }
 595
 596         status = lttng_condition_buffer_usage_get_session_name(
 597                         condition, &trigger_session_name);
 598         assert((status == LTTNG_CONDITION_STATUS_OK) && trigger_session_name);
 599
 600         status = lttng_condition_buffer_usage_get_channel_name(
 601                         condition, &trigger_channel_name);
 602         assert((status == LTTNG_CONDITION_STATUS_OK) && trigger_channel_name);
 603
 604         if (strcmp(info->session_name, trigger_session_name)) {
 605                 goto fail;
 606         }
 607         if (strcmp(info->channel_name, trigger_channel_name)) {
 608                 goto fail;
 609         }
 610
 611         return true;
 612 fail:
 613         return false;
 614 }
 615
 616 static
 617 bool trigger_applies_to_client(struct lttng_trigger *trigger,
 618                 struct notification_client *client)
 619 {
 620         bool applies = false;
 621         struct lttng_condition_list_element *condition_list_element;
 622
 623         cds_list_for_each_entry(condition_list_element, &client->condition_list,
 624                         node) {
 625                 applies = lttng_condition_is_equal(
 626                                 condition_list_element->condition,
 627                                 lttng_trigger_get_condition(trigger));
 628                 if (applies) {
 629                         break;
 630                 }
 631         }
 632         return applies;
 633 }
 634
 635 static
 636 unsigned long hash_channel_key(struct channel_key *key)
 637 {
 638         return hash_key_u64(&key->key, lttng_ht_seed) ^ hash_key_ulong(
 639                 (void *) (unsigned long) key->domain, lttng_ht_seed);
 640 }
 641
 642 static
 643 int handle_notification_thread_command_add_channel(
 644         struct notification_thread_state *state,
 645         struct channel_info *channel_info,
 646         enum lttng_error_code *cmd_result)
 647 {
 648         struct cds_list_head trigger_list;
 649         struct channel_info *new_channel_info;
 650         struct channel_key *channel_key;
 651         struct lttng_channel_trigger_list *channel_trigger_list = NULL;
 652         struct lttng_trigger_ht_element *trigger_ht_element = NULL;
 653         int trigger_count = 0;
 654         struct cds_lfht_iter iter;
 655
 656         DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
 657                         channel_info->channel_name, channel_info->session_name,
 658                         channel_info->key.key, channel_info->key.domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
 659
 660         CDS_INIT_LIST_HEAD(&trigger_list);
 661
 662         new_channel_info = channel_info_copy(channel_info);
 663         if (!new_channel_info) {
 664                 goto error;
 665         }
 666
 667         channel_key = &new_channel_info->key;
 668
 669         /* Build a list of all triggers applying to the new channel. */
 670         cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
 671                         node) {
 672                 struct lttng_trigger_list_element *new_element;
 673
 674                 if (!trigger_applies_to_channel(trigger_ht_element->trigger,
 675                                 channel_info)) {
 676                         continue;
 677                 }
 678
 679                 new_element = zmalloc(sizeof(*new_element));
 680                 if (!new_element) {
 681                         goto error;
 682                 }
 683                 CDS_INIT_LIST_HEAD(&new_element->node);
 684                 new_element->trigger = trigger_ht_element->trigger;
 685                 cds_list_add(&new_element->node, &trigger_list);
 686                 trigger_count++;
 687         }
 688
 689         DBG("[notification-thread] Found %i triggers that apply to newly added channel",
 690                         trigger_count);
 691         channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
 692         if (!channel_trigger_list) {
 693                 goto error;
 694         }
 695         channel_trigger_list->channel_key = *channel_key;
 696         CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
 697         cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
 698         cds_list_splice(&trigger_list, &channel_trigger_list->list);
 699
 700         rcu_read_lock();
 701         /* Add channel to the channel_ht which owns the channel_infos. */
 702         cds_lfht_add(state->channels_ht,
 703                         hash_channel_key(channel_key),
 704                         &new_channel_info->channels_ht_node);
 705         /*
 706          * Add the list of triggers associated with this channel to the
 707          * channel_triggers_ht.
 708          */
 709         cds_lfht_add(state->channel_triggers_ht,
 710                         hash_channel_key(channel_key),
 711                         &channel_trigger_list->channel_triggers_ht_node);
 712         rcu_read_unlock();
 713         *cmd_result = LTTNG_OK;
 714         return 0;
 715 error:
 716         /* Empty trigger list */
 717         channel_info_destroy(new_channel_info);
 718         return 1;
 719 }
 720
 721 static
 722 int handle_notification_thread_command_remove_channel(
 723         struct notification_thread_state *state,
 724         uint64_t channel_key, enum lttng_domain_type domain,
 725         enum lttng_error_code *cmd_result)
 726 {
 727         struct cds_lfht_node *node;
 728         struct cds_lfht_iter iter;
 729         struct lttng_channel_trigger_list *trigger_list;
 730         struct lttng_trigger_list_element *trigger_list_element, *tmp;
 731         struct channel_key key = { .key = channel_key, .domain = domain };
 732         struct channel_info *channel_info;
 733
 734         DBG("[notification-thread] Removing channel key = %" PRIu64 " in %s domain",
 735                         channel_key, domain == LTTNG_DOMAIN_KERNEL ? "kernel" : "user space");
 736
 737         rcu_read_lock();
 738
 739         cds_lfht_lookup(state->channel_triggers_ht,
 740                         hash_channel_key(&key),
 741                         match_channel_trigger_list,
 742                         &key,
 743                         &iter);
 744         node = cds_lfht_iter_get_node(&iter);
 745         /*
 746          * There is a severe internal error if we are being asked to remove a
 747          * channel that doesn't exist.
 748          */
 749         if (!node) {
 750                 ERR("[notification-thread] Channel being removed is unknown to the notification thread");
 751                 goto end;
 752         }
 753
 754         /* Free the list of triggers associated with this channel. */
 755         trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
 756                         channel_triggers_ht_node);
 757         cds_list_for_each_entry_safe(trigger_list_element, tmp,
 758                         &trigger_list->list, node) {
 759                 cds_list_del(&trigger_list_element->node);
 760                 free(trigger_list_element);
 761         }
 762         cds_lfht_del(state->channel_triggers_ht, node);
 763         free(trigger_list);
 764
 765         /* Free sampled channel state. */
 766         cds_lfht_lookup(state->channel_state_ht,
 767                         hash_channel_key(&key),
 768                         match_channel_state_sample,
 769                         &key,
 770                         &iter);
 771         node = cds_lfht_iter_get_node(&iter);
 772         /*
 773          * This is expected to be NULL if the channel is destroyed before we
 774          * received a sample.
 775          */
 776         if (node) {
 777                 struct channel_state_sample *sample = caa_container_of(node,
 778                                 struct channel_state_sample,
 779                                 channel_state_ht_node);
 780
 781                 cds_lfht_del(state->channel_state_ht, node);
 782                 free(sample);
 783         }
 784
 785         /* Remove the channel from the channels_ht and free it. */
 786         cds_lfht_lookup(state->channels_ht,
 787                         hash_channel_key(&key),
 788                         match_channel_info,
 789                         &key,
 790                         &iter);
 791         node = cds_lfht_iter_get_node(&iter);
 792         assert(node);
 793         channel_info = caa_container_of(node, struct channel_info,
 794                         channels_ht_node);
 795         cds_lfht_del(state->channels_ht, node);
 796         channel_info_destroy(channel_info);
 797 end:
 798         rcu_read_unlock();
 799         *cmd_result = LTTNG_OK;
 800         return 0;
 801 }
 802
 803 /*
 804  * FIXME A client's credentials are not checked when registering a trigger, nor
 805  *       are they stored alongside with the trigger.
 806  *
 807  * The effects of this are benign:
 808  *     - The client will succeed in registering the trigger, as it is valid,
 809  *     - The trigger will, internally, be bound to the channel,
 810  *     - The notifications will not be sent since the client's credentials
 811  *       are checked against the channel at that moment.
 812  */
 813 static
 814 int handle_notification_thread_command_register_trigger(
 815         struct notification_thread_state *state,
 816         struct lttng_trigger *trigger,
 817         enum lttng_error_code *cmd_result)
 818 {
 819         int ret = 0;
 820         struct lttng_condition *condition;
 821         struct notification_client *client;
 822         struct notification_client_list *client_list = NULL;
 823         struct lttng_trigger_ht_element *trigger_ht_element = NULL;
 824         struct notification_client_list_element *client_list_element, *tmp;
 825         struct cds_lfht_node *node;
 826         struct cds_lfht_iter iter;
 827         struct channel_info *channel;
 828         bool free_trigger = true;
 829
 830         rcu_read_lock();
 831
 832         condition = lttng_trigger_get_condition(trigger);
 833         trigger_ht_element = zmalloc(sizeof(*trigger_ht_element));
 834         if (!trigger_ht_element) {
 835                 ret = -1;
 836                 goto error;
 837         }
 838
 839         /* Add trigger to the trigger_ht. */
 840         cds_lfht_node_init(&trigger_ht_element->node);
 841         trigger_ht_element->trigger = trigger;
 842
 843         node = cds_lfht_add_unique(state->triggers_ht,
 844                         lttng_condition_hash(condition),
 845                         match_condition,
 846                         condition,
 847                         &trigger_ht_element->node);
 848         if (node != &trigger_ht_element->node) {
 849                 /* Not a fatal error, simply report it to the client. */
 850                 *cmd_result = LTTNG_ERR_TRIGGER_EXISTS;
 851                 goto error_free_ht_element;
 852         }
 853
 854         /*
 855          * Ownership of the trigger and of its wrapper was transfered to
 856          * the triggers_ht.
 857          */
 858         trigger_ht_element = NULL;
 859         free_trigger = false;
 860
 861         /*
 862          * The rest only applies to triggers that have a "notify" action.
 863          * It is not skipped as this is the only action type currently
 864          * supported.
 865          */
 866         client_list = zmalloc(sizeof(*client_list));
 867         if (!client_list) {
 868                 ret = -1;
 869                 goto error_free_ht_element;
 870         }
 871         cds_lfht_node_init(&client_list->notification_trigger_ht_node);
 872         CDS_INIT_LIST_HEAD(&client_list->list);
 873         client_list->trigger = trigger;
 874
 875         /* Build a list of clients to which this new trigger applies. */
 876         cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
 877                         client_socket_ht_node) {
 878                 if (!trigger_applies_to_client(trigger, client)) {
 879                         continue;
 880                 }
 881
 882                 client_list_element = zmalloc(sizeof(*client_list_element));
 883                 if (!client_list_element) {
 884                         ret = -1;
 885                         goto error_free_client_list;
 886                 }
 887                 CDS_INIT_LIST_HEAD(&client_list_element->node);
 888                 client_list_element->client = client;
 889                 cds_list_add(&client_list_element->node, &client_list->list);
 890         }
 891
 892         cds_lfht_add(state->notification_trigger_clients_ht,
 893                         lttng_condition_hash(condition),
 894                         &client_list->notification_trigger_ht_node);
 895         /*
 896          * Client list ownership transferred to the
 897          * notification_trigger_clients_ht.
 898          */
 899         client_list = NULL;
 900
 901         /*
 902          * Add the trigger to list of triggers bound to the channels currently
 903          * known.
 904          */
 905         cds_lfht_for_each_entry(state->channels_ht, &iter, channel,
 906                         channels_ht_node) {
 907                 struct lttng_trigger_list_element *trigger_list_element;
 908                 struct lttng_channel_trigger_list *trigger_list;
 909
 910                 if (!trigger_applies_to_channel(trigger, channel)) {
 911                         continue;
 912                 }
 913
 914                 cds_lfht_lookup(state->channel_triggers_ht,
 915                                 hash_channel_key(&channel->key),
 916                                 match_channel_trigger_list,
 917                                 &channel->key,
 918                                 &iter);
 919                 node = cds_lfht_iter_get_node(&iter);
 920                 assert(node);
 921                 /* Free the list of triggers associated with this channel. */
 922                 trigger_list = caa_container_of(node,
 923                                 struct lttng_channel_trigger_list,
 924                                 channel_triggers_ht_node);
 925
 926                 trigger_list_element = zmalloc(sizeof(*trigger_list_element));
 927                 if (!trigger_list_element) {
 928                         ret = -1;
 929                         goto error_free_client_list;
 930                 }
 931                 CDS_INIT_LIST_HEAD(&trigger_list_element->node);
 932                 trigger_list_element->trigger = trigger;
 933                 cds_list_add(&trigger_list_element->node, &trigger_list->list);
 934                 /* A trigger can only apply to one channel. */
 935                 break;
 936         }
 937
 938         *cmd_result = LTTNG_OK;
 939 error_free_client_list:
 940         if (client_list) {
 941                 cds_list_for_each_entry_safe(client_list_element, tmp,
 942                                 &client_list->list, node) {
 943                         free(client_list_element);
 944                 }
 945                 free(client_list);
 946         }
 947 error_free_ht_element:
 948         free(trigger_ht_element);
 949 error:
 950         if (free_trigger) {
 951                 struct lttng_action *action = lttng_trigger_get_action(trigger);
 952
 953                 lttng_condition_destroy(condition);
 954                 lttng_action_destroy(action);
 955                 lttng_trigger_destroy(trigger);
 956         }
 957         rcu_read_unlock();
 958         return ret;
 959 }
 960
 961 static
 962 int handle_notification_thread_command_unregister_trigger(
 963                 struct notification_thread_state *state,
 964                 struct lttng_trigger *trigger,
 965                 enum lttng_error_code *_cmd_reply)
 966 {
 967         struct cds_lfht_iter iter;
 968         struct cds_lfht_node *node, *triggers_ht_node;
 969         struct lttng_channel_trigger_list *trigger_list;
 970         struct notification_client_list *client_list;
 971         struct notification_client_list_element *client_list_element, *tmp;
 972         struct lttng_trigger_ht_element *trigger_ht_element = NULL;
 973         struct lttng_condition *condition = lttng_trigger_get_condition(
 974                         trigger);
 975         struct lttng_action *action;
 976         enum lttng_error_code cmd_reply;
 977
 978         rcu_read_lock();
 979
 980         cds_lfht_lookup(state->triggers_ht,
 981                         lttng_condition_hash(condition),
 982                         match_condition,
 983                         condition,
 984                         &iter);
 985         triggers_ht_node = cds_lfht_iter_get_node(&iter);
 986         if (!triggers_ht_node) {
 987                 cmd_reply = LTTNG_ERR_TRIGGER_NOT_FOUND;
 988                 goto end;
 989         } else {
 990                 cmd_reply = LTTNG_OK;
 991         }
 992
 993         /* Remove trigger from channel_triggers_ht. */
 994         cds_lfht_for_each_entry(state->channel_triggers_ht, &iter, trigger_list,
 995                         channel_triggers_ht_node) {
 996                 struct lttng_trigger_list_element *trigger_element, *tmp;
 997
 998                 cds_list_for_each_entry_safe(trigger_element, tmp,
 999                                 &trigger_list->list, node) {
1000                         struct lttng_condition *current_condition =
1001                                         lttng_trigger_get_condition(
1002                                                 trigger_element->trigger);
1003
1004                         assert(current_condition);
1005                         if (!lttng_condition_is_equal(condition,
1006                                         current_condition)) {
1007                                 continue;
1008                         }
1009
1010                         DBG("[notification-thread] Removed trigger from channel_triggers_ht");
1011                         cds_list_del(&trigger_element->node);
1012                 }
1013         }
1014
1015         /*
1016          * Remove and release the client list from
1017          * notification_trigger_clients_ht.
1018          */
1019         cds_lfht_lookup(state->notification_trigger_clients_ht,
1020                         lttng_condition_hash(condition),
1021                         match_client_list,
1022                         trigger,
1023                         &iter);
1024         node = cds_lfht_iter_get_node(&iter);
1025         assert(node);
1026         client_list = caa_container_of(node, struct notification_client_list,
1027                         notification_trigger_ht_node);
1028         cds_list_for_each_entry_safe(client_list_element, tmp,
1029                         &client_list->list, node) {
1030                 free(client_list_element);
1031         }
1032         cds_lfht_del(state->notification_trigger_clients_ht, node);
1033         free(client_list);
1034
1035         /* Remove trigger from triggers_ht. */
1036         trigger_ht_element = caa_container_of(triggers_ht_node,
1037                         struct lttng_trigger_ht_element, node);
1038         cds_lfht_del(state->triggers_ht, triggers_ht_node);
1039
1040         condition = lttng_trigger_get_condition(trigger_ht_element->trigger);
1041         lttng_condition_destroy(condition);
1042         action = lttng_trigger_get_action(trigger_ht_element->trigger);
1043         lttng_action_destroy(action);
1044         lttng_trigger_destroy(trigger_ht_element->trigger);
1045         free(trigger_ht_element);
1046 end:
1047         rcu_read_unlock();
1048         if (_cmd_reply) {
1049                 *_cmd_reply = cmd_reply;
1050         }
1051         return 0;
1052 }
1053
1054 /* Returns 0 on success, 1 on exit requested, negative value on error. */
1055 int handle_notification_thread_command(
1056                 struct notification_thread_handle *handle,
1057                 struct notification_thread_state *state)
1058 {
1059         int ret;
1060         uint64_t counter;
1061         struct notification_thread_command *cmd;
1062
1063         /* Read event_fd to put it back into a quiescent state. */
1064         ret = read(handle->cmd_queue.event_fd, &counter, sizeof(counter));
1065         if (ret == -1) {
1066                 goto error;
1067         }
1068
1069         pthread_mutex_lock(&handle->cmd_queue.lock);
1070         cmd = cds_list_first_entry(&handle->cmd_queue.list,
1071                         struct notification_thread_command, cmd_list_node);
1072         switch (cmd->type) {
1073         case NOTIFICATION_COMMAND_TYPE_REGISTER_TRIGGER:
1074                 DBG("[notification-thread] Received register trigger command");
1075                 ret = handle_notification_thread_command_register_trigger(
1076                                 state, cmd->parameters.trigger,
1077                                 &cmd->reply_code);
1078                 break;
1079         case NOTIFICATION_COMMAND_TYPE_UNREGISTER_TRIGGER:
1080                 DBG("[notification-thread] Received unregister trigger command");
1081                 ret = handle_notification_thread_command_unregister_trigger(
1082                                 state, cmd->parameters.trigger,
1083                                 &cmd->reply_code);
1084                 break;
1085         case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
1086                 DBG("[notification-thread] Received add channel command");
1087                 ret = handle_notification_thread_command_add_channel(
1088                                 state, &cmd->parameters.add_channel,
1089                                 &cmd->reply_code);
1090                 break;
1091         case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
1092                 DBG("[notification-thread] Received remove channel command");
1093                 ret = handle_notification_thread_command_remove_channel(
1094                                 state, cmd->parameters.remove_channel.key,
1095                                 cmd->parameters.remove_channel.domain,
1096                                 &cmd->reply_code);
1097                 break;
1098         case NOTIFICATION_COMMAND_TYPE_QUIT:
1099                 DBG("[notification-thread] Received quit command");
1100                 cmd->reply_code = LTTNG_OK;
1101                 ret = 1;
1102                 goto end;
1103         default:
1104                 ERR("[notification-thread] Unknown internal command received");
1105                 goto error_unlock;
1106         }
1107
1108         if (ret) {
1109                 goto error_unlock;
1110         }
1111 end:
1112         cds_list_del(&cmd->cmd_list_node);
1113         futex_nto1_wake(&cmd->reply_futex);
1114         pthread_mutex_unlock(&handle->cmd_queue.lock);
1115         return ret;
1116 error_unlock:
1117         /* Wake-up and return a fatal error to the calling thread. */
1118         futex_nto1_wake(&cmd->reply_futex);
1119         pthread_mutex_unlock(&handle->cmd_queue.lock);
1120         cmd->reply_code = LTTNG_ERR_FATAL;
1121 error:
1122         /* Indicate a fatal error to the caller. */
1123         return -1;
1124 }
1125
1126 static
1127 unsigned long hash_client_socket(int socket)
1128 {
1129         return hash_key_ulong((void *) (unsigned long) socket, lttng_ht_seed);
1130 }
1131
1132 static
1133 int socket_set_non_blocking(int socket)
1134 {
1135         int ret, flags;
1136
1137         /* Set the pipe as non-blocking. */
1138         ret = fcntl(socket, F_GETFL, 0);
1139         if (ret == -1) {
1140                 PERROR("fcntl get socket flags");
1141                 goto end;
1142         }
1143         flags = ret;
1144
1145         ret = fcntl(socket, F_SETFL, flags | O_NONBLOCK);
1146         if (ret == -1) {
1147                 PERROR("fcntl set O_NONBLOCK socket flag");
1148                 goto end;
1149         }
1150         DBG("Client socket (fd = %i) set as non-blocking", socket);
1151 end:
1152         return ret;
1153 }
1154
1155 static
1156 void client_reset_inbound_state(struct notification_client *client)
1157 {
1158         int ret;
1159
1160         ret = lttng_dynamic_buffer_set_size(
1161                         &client->communication.inbound.buffer, 0);
1162         assert(!ret);
1163
1164         client->communication.inbound.bytes_to_receive =
1165                         sizeof(struct lttng_notification_channel_message);
1166         client->communication.inbound.msg_type =
1167                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN;
1168         client->communication.inbound.receive_creds = false;
1169         LTTNG_SOCK_SET_UID_CRED(&client->communication.inbound.creds, -1);
1170         LTTNG_SOCK_SET_GID_CRED(&client->communication.inbound.creds, -1);
1171 }
1172
1173 int handle_notification_thread_client_connect(
1174                 struct notification_thread_state *state)
1175 {
1176         int ret;
1177         struct notification_client *client;
1178
1179         DBG("[notification-thread] Handling new notification channel client connection");
1180
1181         client = zmalloc(sizeof(*client));
1182         if (!client) {
1183                 /* Fatal error. */
1184                 ret = -1;
1185                 goto error;
1186         }
1187         CDS_INIT_LIST_HEAD(&client->condition_list);
1188         lttng_dynamic_buffer_init(&client->communication.inbound.buffer);
1189         lttng_dynamic_buffer_init(&client->communication.outbound.buffer);
1190         client_reset_inbound_state(client);
1191
1192         ret = lttcomm_accept_unix_sock(state->notification_channel_socket);
1193         if (ret < 0) {
1194                 ERR("[notification-thread] Failed to accept new notification channel client connection");
1195                 ret = 0;
1196                 goto error;
1197         }
1198
1199         client->socket = ret;
1200
1201         ret = socket_set_non_blocking(client->socket);
1202         if (ret) {
1203                 ERR("[notification-thread] Failed to set new notification channel client connection socket as non-blocking");
1204                 goto error;
1205         }
1206
1207         ret = lttcomm_setsockopt_creds_unix_sock(client->socket);
1208         if (ret < 0) {
1209                 ERR("[notification-thread] Failed to set socket options on new notification channel client socket");
1210                 ret = 0;
1211                 goto error;
1212         }
1213
1214         ret = lttng_poll_add(&state->events, client->socket,
1215                         LPOLLIN | LPOLLERR |
1216                         LPOLLHUP | LPOLLRDHUP);
1217         if (ret < 0) {
1218                 ERR("[notification-thread] Failed to add notification channel client socket to poll set");
1219                 ret = 0;
1220                 goto error;
1221         }
1222         DBG("[notification-thread] Added new notification channel client socket (%i) to poll set",
1223                         client->socket);
1224
1225         /* Add to ht. */
1226         rcu_read_lock();
1227         cds_lfht_add(state->client_socket_ht,
1228                         hash_client_socket(client->socket),
1229                         &client->client_socket_ht_node);
1230         rcu_read_unlock();
1231
1232         return ret;
1233 error:
1234         notification_client_destroy(client, state);
1235         return ret;
1236 }
1237
1238 int handle_notification_thread_client_disconnect(
1239                 int client_socket,
1240                 struct notification_thread_state *state)
1241 {
1242         int ret = 0;
1243         struct notification_client *client;
1244
1245         rcu_read_lock();
1246         DBG("[notification-thread] Closing client connection (socket fd = %i)",
1247                         client_socket);
1248         client = get_client_from_socket(client_socket, state);
1249         if (!client) {
1250                 /* Internal state corruption, fatal error. */
1251                 ERR("[notification-thread] Unable to find client (socket fd = %i)",
1252                                 client_socket);
1253                 ret = -1;
1254                 goto end;
1255         }
1256
1257         ret = lttng_poll_del(&state->events, client_socket);
1258         if (ret) {
1259                 ERR("[notification-thread] Failed to remove client socket from poll set");
1260         }
1261         cds_lfht_del(state->client_socket_ht,
1262                         &client->client_socket_ht_node);
1263         notification_client_destroy(client, state);
1264 end:
1265         rcu_read_unlock();
1266         return ret;
1267 }
1268
1269 int handle_notification_thread_client_disconnect_all(
1270                 struct notification_thread_state *state)
1271 {
1272         struct cds_lfht_iter iter;
1273         struct notification_client *client;
1274         bool error_encoutered = false;
1275
1276         rcu_read_lock();
1277         DBG("[notification-thread] Closing all client connections");
1278         cds_lfht_for_each_entry(state->client_socket_ht, &iter, client,
1279                 client_socket_ht_node) {
1280                 int ret;
1281
1282                 ret = handle_notification_thread_client_disconnect(
1283                                 client->socket, state);
1284                 if (ret) {
1285                         error_encoutered = true;
1286                 }
1287         }
1288         rcu_read_unlock();
1289         return error_encoutered ? 1 : 0;
1290 }
1291
1292 int handle_notification_thread_trigger_unregister_all(
1293                 struct notification_thread_state *state)
1294 {
1295         bool error_occured = false;
1296         struct cds_lfht_iter iter;
1297         struct lttng_trigger_ht_element *trigger_ht_element;
1298
1299         cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
1300                         node) {
1301                 int ret = handle_notification_thread_command_unregister_trigger(
1302                                 state, trigger_ht_element->trigger, NULL);
1303                 if (ret) {
1304                         error_occured = true;
1305                 }
1306         }
1307         return error_occured ? -1 : 0;
1308 }
1309
1310 static
1311 int client_flush_outgoing_queue(struct notification_client *client,
1312                 struct notification_thread_state *state)
1313 {
1314         ssize_t ret;
1315         size_t to_send_count;
1316
1317         assert(client->communication.outbound.buffer.size != 0);
1318         to_send_count = client->communication.outbound.buffer.size;
1319         DBG("[notification-thread] Flushing client (socket fd = %i) outgoing queue",
1320                         client->socket);
1321
1322         ret = lttcomm_send_unix_sock_non_block(client->socket,
1323                         client->communication.outbound.buffer.data,
1324                         to_send_count);
1325         if ((ret < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) ||
1326                         (ret > 0 && ret < to_send_count)) {
1327                 DBG("[notification-thread] Client (socket fd = %i) outgoing queue could not be completely flushed",
1328                                 client->socket);
1329                 to_send_count -= max(ret, 0);
1330
1331                 memcpy(client->communication.outbound.buffer.data,
1332                                 client->communication.outbound.buffer.data +
1333                                 client->communication.outbound.buffer.size - to_send_count,
1334                                 to_send_count);
1335                 ret = lttng_dynamic_buffer_set_size(
1336                                 &client->communication.outbound.buffer,
1337                                 to_send_count);
1338                 if (ret) {
1339                         goto error;
1340                 }
1341
1342                 /*
1343                  * We want to be notified whenever there is buffer space
1344                  * available to send the rest of the payload.
1345                  */
1346                 ret = lttng_poll_mod(&state->events, client->socket,
1347                                 CLIENT_POLL_MASK_IN_OUT);
1348                 if (ret) {
1349                         goto error;
1350                 }
1351         } else if (ret < 0) {
1352                 /* Generic error, disconnect the client. */
1353                 ERR("[notification-thread] Failed to send flush outgoing queue, disconnecting client (socket fd = %i)",
1354                                 client->socket);
1355                 ret = handle_notification_thread_client_disconnect(
1356                                 client->socket, state);
1357                 if (ret) {
1358                         goto error;
1359                 }
1360         } else {
1361                 /* No error and flushed the queue completely. */
1362                 ret = lttng_dynamic_buffer_set_size(
1363                                 &client->communication.outbound.buffer, 0);
1364                 if (ret) {
1365                         goto error;
1366                 }
1367                 ret = lttng_poll_mod(&state->events, client->socket,
1368                                 CLIENT_POLL_MASK_IN);
1369                 if (ret) {
1370                         goto error;
1371                 }
1372
1373                 client->communication.outbound.queued_command_reply = false;
1374                 client->communication.outbound.dropped_notification = false;
1375         }
1376
1377         return 0;
1378 error:
1379         return -1;
1380 }
1381
1382 static
1383 int client_send_command_reply(struct notification_client *client,
1384                 struct notification_thread_state *state,
1385                 enum lttng_notification_channel_status status)
1386 {
1387         int ret;
1388         struct lttng_notification_channel_command_reply reply = {
1389                 .status = (int8_t) status,
1390         };
1391         struct lttng_notification_channel_message msg = {
1392                 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
1393                 .size = sizeof(reply),
1394         };
1395         char buffer[sizeof(msg) + sizeof(reply)];
1396
1397         if (client->communication.outbound.queued_command_reply) {
1398                 /* Protocol error. */
1399                 goto error;
1400         }
1401
1402         memcpy(buffer, &msg, sizeof(msg));
1403         memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
1404         DBG("[notification-thread] Send command reply (%i)", (int) status);
1405
1406         /* Enqueue buffer to outgoing queue and flush it. */
1407         ret = lttng_dynamic_buffer_append(
1408                         &client->communication.outbound.buffer,
1409                         buffer, sizeof(buffer));
1410         if (ret) {
1411                 goto error;
1412         }
1413
1414         ret = client_flush_outgoing_queue(client, state);
1415         if (ret) {
1416                 goto error;
1417         }
1418
1419         if (client->communication.outbound.buffer.size != 0) {
1420                 /* Queue could not be emptied. */
1421                 client->communication.outbound.queued_command_reply = true;
1422         }
1423
1424         return 0;
1425 error:
1426         return -1;
1427 }
1428
1429 static
1430 int client_dispatch_message(struct notification_client *client,
1431                 struct notification_thread_state *state)
1432 {
1433         int ret = 0;
1434
1435         if (client->communication.inbound.msg_type !=
1436                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE &&
1437                         client->communication.inbound.msg_type !=
1438                                 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN &&
1439                         !client->validated) {
1440                 WARN("[notification-thread] client attempted a command before handshake");
1441                 ret = -1;
1442                 goto end;
1443         }
1444
1445         switch (client->communication.inbound.msg_type) {
1446         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNKNOWN:
1447         {
1448                 /*
1449                  * Receiving message header. The function will be called again
1450                  * once the rest of the message as been received and can be
1451                  * interpreted.
1452                  */
1453                 const struct lttng_notification_channel_message *msg;
1454
1455                 assert(sizeof(*msg) ==
1456                                 client->communication.inbound.buffer.size);
1457                 msg = (const struct lttng_notification_channel_message *)
1458                                 client->communication.inbound.buffer.data;
1459
1460                 if (msg->size == 0 || msg->size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
1461                         ERR("[notification-thread] Invalid notification channel message: length = %u", msg->size);
1462                         ret = -1;
1463                         goto end;
1464                 }
1465
1466                 switch (msg->type) {
1467                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1468                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1469                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1470                         break;
1471                 default:
1472                         ret = -1;
1473                         ERR("[notification-thread] Invalid notification channel message: unexpected message type");
1474                         goto end;
1475                 }
1476
1477                 client->communication.inbound.bytes_to_receive = msg->size;
1478                 client->communication.inbound.msg_type =
1479                                 (enum lttng_notification_channel_message_type) msg->type;
1480                 if (client->communication.inbound.msg_type ==
1481                                 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE) {
1482                         client->communication.inbound.receive_creds = true;
1483                 }
1484                 ret = lttng_dynamic_buffer_set_size(
1485                                 &client->communication.inbound.buffer, 0);
1486                 if (ret) {
1487                         goto end;
1488                 }
1489                 break;
1490         }
1491         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
1492         {
1493                 struct lttng_notification_channel_command_handshake *handshake_client;
1494                 struct lttng_notification_channel_command_handshake handshake_reply = {
1495                         .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
1496                         .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
1497                 };
1498                 struct lttng_notification_channel_message msg_header = {
1499                         .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
1500                         .size = sizeof(handshake_reply),
1501                 };
1502                 enum lttng_notification_channel_status status =
1503                                 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1504                 char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
1505
1506                 memcpy(send_buffer, &msg_header, sizeof(msg_header));
1507                 memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
1508                                 sizeof(handshake_reply));
1509
1510                 handshake_client =
1511                                 (struct lttng_notification_channel_command_handshake *)
1512                                         client->communication.inbound.buffer.data;
1513                                 client->major = handshake_client->major;
1514                 client->minor = handshake_client->minor;
1515                 if (!client->communication.inbound.creds_received) {
1516                         ERR("[notification-thread] No credentials received from client");
1517                         ret = -1;
1518                         goto end;
1519                 }
1520
1521                 client->uid = LTTNG_SOCK_GET_UID_CRED(
1522                                 &client->communication.inbound.creds);
1523                 client->gid = LTTNG_SOCK_GET_GID_CRED(
1524                                 &client->communication.inbound.creds);
1525                 DBG("[notification-thread] Received handshake from client (uid = %u, gid = %u) with version %i.%i",
1526                                 client->uid, client->gid, (int) client->major,
1527                                 (int) client->minor);
1528
1529                 if (handshake_client->major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
1530                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNSUPPORTED_VERSION;
1531                 }
1532
1533                 ret = lttng_dynamic_buffer_append(&client->communication.outbound.buffer,
1534                                 send_buffer, sizeof(send_buffer));
1535                 if (ret) {
1536                         ERR("[notification-thread] Failed to send protocol version to notification channel client");
1537                         goto end;
1538                 }
1539
1540                 ret = client_flush_outgoing_queue(client, state);
1541                 if (ret) {
1542                         goto end;
1543                 }
1544
1545                 ret = client_send_command_reply(client, state, status);
1546                 if (ret) {
1547                         ERR("[notification-thread] Failed to send reply to notification channel client");
1548                         goto end;
1549                 }
1550
1551                 /* Set reception state to receive the next message header. */
1552                 client_reset_inbound_state(client);
1553                 client->validated = true;
1554                 break;
1555         }
1556         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE:
1557         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE:
1558         {
1559                 struct lttng_condition *condition;
1560                 enum lttng_notification_channel_status status =
1561                                 LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
1562                 const struct lttng_buffer_view condition_view =
1563                                 lttng_buffer_view_from_dynamic_buffer(
1564                                         &client->communication.inbound.buffer,
1565                                         0, -1);
1566                 size_t expected_condition_size =
1567                                 client->communication.inbound.buffer.size;
1568
1569                 ret = lttng_condition_create_from_buffer(&condition_view,
1570                                 &condition);
1571                 if (ret != expected_condition_size) {
1572                         ERR("[notification-thread] Malformed condition received from client");
1573                         goto end;
1574                 }
1575
1576                 if (client->communication.inbound.msg_type ==
1577                                 LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE) {
1578                         /*
1579                          * FIXME The current state should be evaluated on
1580                          * subscription.
1581                          */
1582                         ret = notification_thread_client_subscribe(client,
1583                                         condition, state, &status);
1584                 } else {
1585                         ret = notification_thread_client_unsubscribe(client,
1586                                         condition, state, &status);
1587                 }
1588                 if (ret) {
1589                         goto end;
1590                 }
1591
1592                 ret = client_send_command_reply(client, state, status);
1593                 if (ret) {
1594                         ERR("[notification-thread] Failed to send reply to notification channel client");
1595                         goto end;
1596                 }
1597
1598                 /* Set reception state to receive the next message header. */
1599                 client_reset_inbound_state(client);
1600                 break;
1601         }
1602         default:
1603                 abort();
1604         }
1605 end:
1606         return ret;
1607 }
1608
1609 /* Incoming data from client. */
1610 int handle_notification_thread_client_in(
1611                 struct notification_thread_state *state, int socket)
1612 {
1613         int ret;
1614         struct notification_client *client;
1615         ssize_t recv_ret;
1616         size_t offset;
1617
1618         client = get_client_from_socket(socket, state);
1619         if (!client) {
1620                 /* Internal error, abort. */
1621                 ret = -1;
1622                 goto end;
1623         }
1624
1625         offset = client->communication.inbound.buffer.size;
1626         ret = lttng_dynamic_buffer_set_size(
1627                         &client->communication.inbound.buffer,
1628                         client->communication.inbound.bytes_to_receive);
1629         if (ret) {
1630                 goto end;
1631         }
1632
1633         if (client->communication.inbound.receive_creds) {
1634                 recv_ret = lttcomm_recv_creds_unix_sock(socket,
1635                                 client->communication.inbound.buffer.data + offset,
1636                                 client->communication.inbound.bytes_to_receive,
1637                                 &client->communication.inbound.creds);
1638                 if (recv_ret > 0) {
1639                         client->communication.inbound.receive_creds = false;
1640                         client->communication.inbound.creds_received = true;
1641                 }
1642         } else {
1643                 recv_ret = lttcomm_recv_unix_sock_non_block(socket,
1644                                 client->communication.inbound.buffer.data + offset,
1645                                 client->communication.inbound.bytes_to_receive);
1646         }
1647         if (recv_ret < 0) {
1648                 goto error_disconnect_client;
1649         }
1650
1651         client->communication.inbound.bytes_to_receive -= recv_ret;
1652         ret = lttng_dynamic_buffer_set_size(
1653                         &client->communication.inbound.buffer,
1654                         client->communication.inbound.buffer.size -
1655                         client->communication.inbound.bytes_to_receive);
1656         if (ret) {
1657                 goto end;
1658         }
1659
1660         if (client->communication.inbound.bytes_to_receive == 0) {
1661                 ret = client_dispatch_message(client, state);
1662                 if (ret) {
1663                         /*
1664                          * Only returns an error if this client must be
1665                          * disconnected.
1666                          */
1667                         goto error_disconnect_client;
1668                 }
1669         } else {
1670                 goto end;
1671         }
1672 end:
1673         return ret;
1674 error_disconnect_client:
1675         ret = handle_notification_thread_client_disconnect(socket, state);
1676         return ret;
1677 }
1678
1679 /* Client ready to receive outgoing data. */
1680 int handle_notification_thread_client_out(
1681                 struct notification_thread_state *state, int socket)
1682 {
1683         int ret;
1684         struct notification_client *client;
1685
1686         client = get_client_from_socket(socket, state);
1687         if (!client) {
1688                 /* Internal error, abort. */
1689                 ret = -1;
1690                 goto end;
1691         }
1692
1693         ret = client_flush_outgoing_queue(client, state);
1694         if (ret) {
1695                 goto end;
1696         }
1697 end:
1698         return ret;
1699 }
1700
1701 static
1702 bool evaluate_buffer_usage_condition(struct lttng_condition *condition,
1703                 struct channel_state_sample *sample, uint64_t buffer_capacity)
1704 {
1705         bool result = false;
1706         uint64_t threshold;
1707         enum lttng_condition_type condition_type;
1708         struct lttng_condition_buffer_usage *use_condition = container_of(
1709                         condition, struct lttng_condition_buffer_usage,
1710                         parent);
1711
1712         if (!sample) {
1713                 goto end;
1714         }
1715
1716         if (use_condition->threshold_bytes.set) {
1717                 threshold = use_condition->threshold_bytes.value;
1718         } else {
1719                 /*
1720                  * Threshold was expressed as a ratio.
1721                  *
1722                  * TODO the threshold (in bytes) of conditions expressed
1723                  * as a ratio of total buffer size could be cached to
1724                  * forego this double-multiplication or it could be performed
1725                  * as fixed-point math.
1726                  *
1727                  * Note that caching should accomodate the case where the
1728                  * condition applies to multiple channels (i.e. don't assume
1729                  * that all channels matching my_chann* have the same size...)
1730                  */
1731                 threshold = (uint64_t) (use_condition->threshold_ratio.value *
1732                                 (double) buffer_capacity);
1733         }
1734
1735         condition_type = lttng_condition_get_type(condition);
1736         if (condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW) {
1737                 DBG("[notification-thread] Low buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
1738                                 threshold, sample->highest_usage);
1739
1740                 /*
1741                  * The low condition should only be triggered once _all_ of the
1742                  * streams in a channel have gone below the "low" threshold.
1743                  */
1744                 if (sample->highest_usage <= threshold) {
1745                         result = true;
1746                 }
1747         } else {
1748                 DBG("[notification-thread] High buffer usage condition being evaluated: threshold = %" PRIu64 ", highest usage = %" PRIu64,
1749                                 threshold, sample->highest_usage);
1750
1751                 /*
1752                  * For high buffer usage scenarios, we want to trigger whenever
1753                  * _any_ of the streams has reached the "high" threshold.
1754                  */
1755                 if (sample->highest_usage >= threshold) {
1756                         result = true;
1757                 }
1758         }
1759 end:
1760         return result;
1761 }
1762
1763 static
1764 int evaluate_condition(struct lttng_condition *condition,
1765                 struct lttng_evaluation **evaluation,
1766                 struct notification_thread_state *state,
1767                 struct channel_state_sample *previous_sample,
1768                 struct channel_state_sample *latest_sample,
1769                 uint64_t buffer_capacity)
1770 {
1771         int ret = 0;
1772         enum lttng_condition_type condition_type;
1773         bool previous_sample_result;
1774         bool latest_sample_result;
1775
1776         condition_type = lttng_condition_get_type(condition);
1777         /* No other condition type supported for the moment. */
1778         assert(condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW ||
1779                         condition_type == LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH);
1780
1781         previous_sample_result = evaluate_buffer_usage_condition(condition,
1782                         previous_sample, buffer_capacity);
1783         latest_sample_result = evaluate_buffer_usage_condition(condition,
1784                         latest_sample, buffer_capacity);
1785
1786         if (!latest_sample_result ||
1787                         (previous_sample_result == latest_sample_result)) {
1788                 /*
1789                  * Only trigger on a condition evaluation transition.
1790                  *
1791                  * NOTE: This edge-triggered logic may not be appropriate for
1792                  * future condition types.
1793                  */
1794                 goto end;
1795         }
1796
1797         if (evaluation && latest_sample_result) {
1798                 *evaluation = lttng_evaluation_buffer_usage_create(
1799                                 condition_type,
1800                                 latest_sample->highest_usage,
1801                                 buffer_capacity);
1802                 if (!*evaluation) {
1803                         ret = -1;
1804                         goto end;
1805                 }
1806         }
1807 end:
1808         return ret;
1809 }
1810
1811 static
1812 int client_enqueue_dropped_notification(struct notification_client *client,
1813                 struct notification_thread_state *state)
1814 {
1815         int ret;
1816         struct lttng_notification_channel_message msg = {
1817                 .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
1818                 .size = 0,
1819         };
1820
1821         ret = lttng_dynamic_buffer_append(
1822                         &client->communication.outbound.buffer, &msg,
1823                         sizeof(msg));
1824         return ret;
1825 }
1826
1827 static
1828 int send_evaluation_to_clients(struct lttng_trigger *trigger,
1829                 struct lttng_evaluation *evaluation,
1830                 struct notification_client_list* client_list,
1831                 struct notification_thread_state *state,
1832                 uid_t channel_uid, gid_t channel_gid)
1833 {
1834         int ret = 0;
1835         struct lttng_dynamic_buffer msg_buffer;
1836         struct notification_client_list_element *client_list_element, *tmp;
1837         struct lttng_notification *notification;
1838         struct lttng_condition *condition;
1839         ssize_t expected_notification_size, notification_size;
1840         struct lttng_notification_channel_message msg;
1841
1842         lttng_dynamic_buffer_init(&msg_buffer);
1843
1844         condition = lttng_trigger_get_condition(trigger);
1845         assert(condition);
1846
1847         notification = lttng_notification_create(condition, evaluation);
1848         if (!notification) {
1849                 ret = -1;
1850                 goto end;
1851         }
1852
1853         expected_notification_size = lttng_notification_serialize(notification,
1854                         NULL);
1855         if (expected_notification_size < 0) {
1856                 ERR("[notification-thread] Failed to get size of serialized notification");
1857                 ret = -1;
1858                 goto end;
1859         }
1860
1861         msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION;
1862         msg.size = (uint32_t) expected_notification_size;
1863         ret = lttng_dynamic_buffer_append(&msg_buffer, &msg, sizeof(msg));
1864         if (ret) {
1865                 goto end;
1866         }
1867
1868         ret = lttng_dynamic_buffer_set_size(&msg_buffer,
1869                         msg_buffer.size + expected_notification_size);
1870         if (ret) {
1871                 goto end;
1872         }
1873
1874         notification_size = lttng_notification_serialize(notification,
1875                         msg_buffer.data + sizeof(msg));
1876         if (notification_size != expected_notification_size) {
1877                 ERR("[notification-thread] Failed to serialize notification");
1878                 ret = -1;
1879                 goto end;
1880         }
1881
1882         cds_list_for_each_entry_safe(client_list_element, tmp,
1883                         &client_list->list, node) {
1884                 struct notification_client *client =
1885                                 client_list_element->client;
1886
1887                 if (client->uid != channel_uid && client->gid != channel_gid &&
1888                                 client->uid != 0) {
1889                         /* Client is not allowed to monitor this channel. */
1890                         DBG("[notification-thread] Skipping client at it does not have the permission to receive notification for this channel");
1891                         continue;
1892                 }
1893
1894                 DBG("[notification-thread] Sending notification to client (fd = %i, %zu bytes)",
1895                                 client->socket, msg_buffer.size);
1896                 if (client->communication.outbound.buffer.size) {
1897                         /*
1898                          * Outgoing data is already buffered for this client;
1899                          * drop the notification and enqueue a "dropped
1900                          * notification" message if this is the first dropped
1901                          * notification since the socket spilled-over to the
1902                          * queue.
1903                          */
1904                         DBG("[notification-thread] Dropping notification addressed to client (socket fd = %i)",
1905                                         client->socket);
1906                         if (!client->communication.outbound.dropped_notification) {
1907                                 client->communication.outbound.dropped_notification = true;
1908                                 ret = client_enqueue_dropped_notification(
1909                                                 client, state);
1910                                 if (ret) {
1911                                         goto end;
1912                                 }
1913                         }
1914                         continue;
1915                 }
1916
1917                 ret = lttng_dynamic_buffer_append_buffer(
1918                                 &client->communication.outbound.buffer,
1919                                 &msg_buffer);
1920                 if (ret) {
1921                         goto end;
1922                 }
1923
1924                 ret = client_flush_outgoing_queue(client, state);
1925                 if (ret) {
1926                         goto end;
1927                 }
1928         }
1929         ret = 0;
1930 end:
1931         lttng_notification_destroy(notification);
1932         lttng_dynamic_buffer_reset(&msg_buffer);
1933         return ret;
1934 }
1935
1936 int handle_notification_thread_channel_sample(
1937                 struct notification_thread_state *state, int pipe,
1938                 enum lttng_domain_type domain)
1939 {
1940         int ret = 0;
1941         struct lttcomm_consumer_channel_monitor_msg sample_msg;
1942         struct channel_state_sample previous_sample, latest_sample;
1943         struct channel_info *channel_info;
1944         struct cds_lfht_node *node;
1945         struct cds_lfht_iter iter;
1946         struct lttng_channel_trigger_list *trigger_list;
1947         struct lttng_trigger_list_element *trigger_list_element;
1948         bool previous_sample_available = false;
1949
1950         /*
1951          * The monitoring pipe only holds messages smaller than PIPE_BUF,
1952          * ensuring that read/write of sampling messages are atomic.
1953          */
1954         ret = lttng_read(pipe, &sample_msg, sizeof(sample_msg));
1955         if (ret != sizeof(sample_msg)) {
1956                 ERR("[notification-thread] Failed to read from monitoring pipe (fd = %i)",
1957                                 pipe);
1958                 ret = -1;
1959                 goto end;
1960         }
1961
1962         ret = 0;
1963         latest_sample.key.key = sample_msg.key;
1964         latest_sample.key.domain = domain;
1965         latest_sample.highest_usage = sample_msg.highest;
1966         latest_sample.lowest_usage = sample_msg.lowest;
1967
1968         rcu_read_lock();
1969
1970         /* Retrieve the channel's informations */
1971         cds_lfht_lookup(state->channels_ht,
1972                         hash_channel_key(&latest_sample.key),
1973                         match_channel_info,
1974                         &latest_sample.key,
1975                         &iter);
1976         node = cds_lfht_iter_get_node(&iter);
1977         if (!node) {
1978                 /*
1979                  * Not an error since the consumer can push a sample to the pipe
1980                  * and the rest of the session daemon could notify us of the
1981                  * channel's destruction before we get a chance to process that
1982                  * sample.
1983                  */
1984                 DBG("[notification-thread] Received a sample for an unknown channel from consumerd, key = %" PRIu64 " in %s domain",
1985                                 latest_sample.key.key,
1986                                 domain == LTTNG_DOMAIN_KERNEL ? "kernel" :
1987                                         "user space");
1988                 goto end_unlock;
1989         }
1990         channel_info = caa_container_of(node, struct channel_info,
1991                         channels_ht_node);
1992         DBG("[notification-thread] Handling channel sample for channel %s (key = %" PRIu64 ") in session %s (highest usage = %" PRIu64 ", lowest usage = %" PRIu64")",
1993                         channel_info->channel_name,
1994                         latest_sample.key.key,
1995                         channel_info->session_name,
1996                         latest_sample.highest_usage,
1997                         latest_sample.lowest_usage);
1998
1999         /* Retrieve the channel's last sample, if it exists, and update it. */
2000         cds_lfht_lookup(state->channel_state_ht,
2001                         hash_channel_key(&latest_sample.key),
2002                         match_channel_state_sample,
2003                         &latest_sample.key,
2004                         &iter);
2005         node = cds_lfht_iter_get_node(&iter);
2006         if (node) {
2007                 struct channel_state_sample *stored_sample;
2008
2009                 /* Update the sample stored. */
2010                 stored_sample = caa_container_of(node,
2011                                 struct channel_state_sample,
2012                                 channel_state_ht_node);
2013                 memcpy(&previous_sample, stored_sample,
2014                                 sizeof(previous_sample));
2015                 stored_sample->highest_usage = latest_sample.highest_usage;
2016                 stored_sample->lowest_usage = latest_sample.lowest_usage;
2017                 previous_sample_available = true;
2018         } else {
2019                 /*
2020                  * This is the channel's first sample, allocate space for and
2021                  * store the new sample.
2022                  */
2023                 struct channel_state_sample *stored_sample;
2024
2025                 stored_sample = zmalloc(sizeof(*stored_sample));
2026                 if (!stored_sample) {
2027                         ret = -1;
2028                         goto end_unlock;
2029                 }
2030
2031                 memcpy(stored_sample, &latest_sample, sizeof(*stored_sample));
2032                 cds_lfht_node_init(&stored_sample->channel_state_ht_node);
2033                 cds_lfht_add(state->channel_state_ht,
2034                                 hash_channel_key(&stored_sample->key),
2035                                 &stored_sample->channel_state_ht_node);
2036         }
2037
2038         /* Find triggers associated with this channel. */
2039         cds_lfht_lookup(state->channel_triggers_ht,
2040                         hash_channel_key(&latest_sample.key),
2041                         match_channel_trigger_list,
2042                         &latest_sample.key,
2043                         &iter);
2044         node = cds_lfht_iter_get_node(&iter);
2045         if (!node) {
2046                 goto end_unlock;
2047         }
2048
2049         trigger_list = caa_container_of(node, struct lttng_channel_trigger_list,
2050                         channel_triggers_ht_node);
2051         cds_list_for_each_entry(trigger_list_element, &trigger_list->list,
2052                         node) {
2053                 struct lttng_condition *condition;
2054                 struct lttng_action *action;
2055                 struct lttng_trigger *trigger;
2056                 struct notification_client_list *client_list;
2057                 struct lttng_evaluation *evaluation = NULL;
2058
2059                 trigger = trigger_list_element->trigger;
2060                 condition = lttng_trigger_get_condition(trigger);
2061                 assert(condition);
2062                 action = lttng_trigger_get_action(trigger);
2063
2064                 /* Notify actions are the only type currently supported. */
2065                 assert(lttng_action_get_type(action) ==
2066                                 LTTNG_ACTION_TYPE_NOTIFY);
2067
2068                 /*
2069                  * Check if any client is subscribed to the result of this
2070                  * evaluation.
2071                  */
2072                 cds_lfht_lookup(state->notification_trigger_clients_ht,
2073                                 lttng_condition_hash(condition),
2074                                 match_client_list,
2075                                 trigger,
2076                                 &iter);
2077                 node = cds_lfht_iter_get_node(&iter);
2078                 assert(node);
2079
2080                 client_list = caa_container_of(node,
2081                                 struct notification_client_list,
2082                                 notification_trigger_ht_node);
2083                 if (cds_list_empty(&client_list->list)) {
2084                         /*
2085                          * No clients interested in the evaluation's result,
2086                          * skip it.
2087                          */
2088                         continue;
2089                 }
2090
2091                 ret = evaluate_condition(condition, &evaluation, state,
2092                                 previous_sample_available ? &previous_sample : NULL,
2093                                 &latest_sample, channel_info->capacity);
2094                 if (ret) {
2095                         goto end_unlock;
2096                 }
2097
2098                 if (!evaluation) {
2099                         continue;
2100                 }
2101
2102                 /* Dispatch evaluation result to all clients. */
2103                 ret = send_evaluation_to_clients(trigger_list_element->trigger,
2104                                 evaluation, client_list, state,
2105                                 channel_info->uid, channel_info->gid);
2106                 if (ret) {
2107                         goto end_unlock;
2108                 }
2109         }
2110 end_unlock:
2111         rcu_read_unlock();
2112 end:
2113         return ret;
2114 }