src/lib/lttng-ctl/channel.c

   1 /*
   2  * Copyright (C) 2017 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   3  *
   4  * This library is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU Lesser General Public License, version 2.1 only,
   6  * as published by the Free Software Foundation.
   7  *
   8  * This library 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 Lesser General Public License
  11  * for more details.
  12  *
  13  * You should have received a copy of the GNU Lesser General Public License
  14  * along with this library; if not, write to the Free Software Foundation,
  15  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  16  */
  17
  18 #include <lttng/notification/notification-internal.h>
  19 #include <lttng/notification/channel-internal.h>
  20 #include <lttng/condition/condition-internal.h>
  21 #include <lttng/endpoint.h>
  22 #include <common/defaults.h>
  23 #include <common/error.h>
  24 #include <common/dynamic-buffer.h>
  25 #include <common/utils.h>
  26 #include <common/defaults.h>
  27 #include <assert.h>
  28 #include "lttng-ctl-helper.h"
  29 #include <common/compat/poll.h>
  30
  31 static
  32 int handshake(struct lttng_notification_channel *channel);
  33
  34 /*
  35  * Populates the reception buffer with the next complete message.
  36  * The caller must acquire the channel's lock.
  37  */
  38 static
  39 int receive_message(struct lttng_notification_channel *channel)
  40 {
  41         ssize_t ret;
  42         struct lttng_notification_channel_message msg;
  43
  44         if (lttng_dynamic_buffer_set_size(&channel->reception_buffer, 0)) {
  45                 ret = -1;
  46                 goto end;
  47         }
  48
  49         ret = lttcomm_recv_unix_sock(channel->socket, &msg, sizeof(msg));
  50         if (ret <= 0) {
  51                 ret = -1;
  52                 goto error;
  53         }
  54
  55         if (msg.size > DEFAULT_MAX_NOTIFICATION_CLIENT_MESSAGE_PAYLOAD_SIZE) {
  56                 ret = -1;
  57                 goto error;
  58         }
  59
  60         /* Add message header at buffer's start. */
  61         ret = lttng_dynamic_buffer_append(&channel->reception_buffer, &msg,
  62                         sizeof(msg));
  63         if (ret) {
  64                 goto error;
  65         }
  66
  67         /* Reserve space for the payload. */
  68         ret = lttng_dynamic_buffer_set_size(&channel->reception_buffer,
  69                         channel->reception_buffer.size + msg.size);
  70         if (ret) {
  71                 goto error;
  72         }
  73
  74         /* Receive message payload. */
  75         ret = lttcomm_recv_unix_sock(channel->socket,
  76                         channel->reception_buffer.data + sizeof(msg), msg.size);
  77         if (ret < (ssize_t) msg.size) {
  78                 ret = -1;
  79                 goto error;
  80         }
  81         ret = 0;
  82 end:
  83         return ret;
  84 error:
  85         if (lttng_dynamic_buffer_set_size(&channel->reception_buffer, 0)) {
  86                 ret = -1;
  87         }
  88         goto end;
  89 }
  90
  91 static
  92 enum lttng_notification_channel_message_type get_current_message_type(
  93                 struct lttng_notification_channel *channel)
  94 {
  95         struct lttng_notification_channel_message *msg;
  96
  97         assert(channel->reception_buffer.size >= sizeof(*msg));
  98
  99         msg = (struct lttng_notification_channel_message *)
 100                         channel->reception_buffer.data;
 101         return (enum lttng_notification_channel_message_type) msg->type;
 102 }
 103
 104 static
 105 struct lttng_notification *create_notification_from_current_message(
 106                 struct lttng_notification_channel *channel)
 107 {
 108         ssize_t ret;
 109         struct lttng_notification *notification = NULL;
 110         struct lttng_buffer_view view;
 111
 112         if (channel->reception_buffer.size <=
 113                         sizeof(struct lttng_notification_channel_message)) {
 114                 goto end;
 115         }
 116
 117         view = lttng_buffer_view_from_dynamic_buffer(&channel->reception_buffer,
 118                         sizeof(struct lttng_notification_channel_message), -1);
 119
 120         ret = lttng_notification_create_from_buffer(&view, &notification);
 121         if (ret != channel->reception_buffer.size -
 122                         sizeof(struct lttng_notification_channel_message)) {
 123                 lttng_notification_destroy(notification);
 124                 notification = NULL;
 125                 goto end;
 126         }
 127 end:
 128         return notification;
 129 }
 130
 131 struct lttng_notification_channel *lttng_notification_channel_create(
 132                 struct lttng_endpoint *endpoint)
 133 {
 134         int fd, ret;
 135         bool is_in_tracing_group = false, is_root = false;
 136         char *sock_path = NULL;
 137         struct lttng_notification_channel *channel = NULL;
 138
 139         if (!endpoint ||
 140                         endpoint != lttng_session_daemon_notification_endpoint) {
 141                 goto end;
 142         }
 143
 144         sock_path = zmalloc(LTTNG_PATH_MAX);
 145         if (!sock_path) {
 146                 goto end;
 147         }
 148
 149         channel = zmalloc(sizeof(struct lttng_notification_channel));
 150         if (!channel) {
 151                 goto end;
 152         }
 153         channel->socket = -1;
 154         pthread_mutex_init(&channel->lock, NULL);
 155         lttng_dynamic_buffer_init(&channel->reception_buffer);
 156         CDS_INIT_LIST_HEAD(&channel->pending_notifications.list);
 157
 158         is_root = (getuid() == 0);
 159         if (!is_root) {
 160                 is_in_tracing_group = lttng_check_tracing_group();
 161         }
 162
 163         if (is_root || is_in_tracing_group) {
 164                 ret = lttng_strncpy(sock_path,
 165                                 DEFAULT_GLOBAL_NOTIFICATION_CHANNEL_UNIX_SOCK,
 166                                 LTTNG_PATH_MAX);
 167                 if (ret) {
 168                         ret = -LTTNG_ERR_INVALID;
 169                         goto error;
 170                 }
 171
 172                 ret = lttcomm_connect_unix_sock(sock_path);
 173                 if (ret >= 0) {
 174                         fd = ret;
 175                         goto set_fd;
 176                 }
 177         }
 178
 179         /* Fallback to local session daemon. */
 180         ret = snprintf(sock_path, LTTNG_PATH_MAX,
 181                         DEFAULT_HOME_NOTIFICATION_CHANNEL_UNIX_SOCK,
 182                         utils_get_home_dir());
 183         if (ret < 0 || ret >= LTTNG_PATH_MAX) {
 184                 goto error;
 185         }
 186
 187         ret = lttcomm_connect_unix_sock(sock_path);
 188         if (ret < 0) {
 189                 goto error;
 190         }
 191         fd = ret;
 192
 193 set_fd:
 194         channel->socket = fd;
 195
 196         ret = handshake(channel);
 197         if (ret) {
 198                 goto error;
 199         }
 200 end:
 201         free(sock_path);
 202         return channel;
 203 error:
 204         lttng_notification_channel_destroy(channel);
 205         channel = NULL;
 206         goto end;
 207 }
 208
 209 enum lttng_notification_channel_status
 210 lttng_notification_channel_get_next_notification(
 211                 struct lttng_notification_channel *channel,
 212                 struct lttng_notification **_notification)
 213 {
 214         int ret;
 215         struct lttng_notification *notification = NULL;
 216         enum lttng_notification_channel_status status =
 217                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 218         struct lttng_poll_event events;
 219
 220         if (!channel || !_notification) {
 221                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 222                 goto end;
 223         }
 224
 225         pthread_mutex_lock(&channel->lock);
 226
 227         if (channel->pending_notifications.count) {
 228                 struct pending_notification *pending_notification;
 229
 230                 assert(!cds_list_empty(&channel->pending_notifications.list));
 231
 232                 /* Deliver one of the pending notifications. */
 233                 pending_notification = cds_list_first_entry(
 234                                 &channel->pending_notifications.list,
 235                                 struct pending_notification,
 236                                 node);
 237                 notification = pending_notification->notification;
 238                 if (!notification) {
 239                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
 240                 }
 241                 cds_list_del(&pending_notification->node);
 242                 channel->pending_notifications.count--;
 243                 free(pending_notification);
 244                 goto end_unlock;
 245         }
 246
 247         /*
 248          * Block on interruptible epoll/poll() instead of the message reception
 249          * itself as the recvmsg() wrappers always restart on EINTR. We choose
 250          * to wait using interruptible epoll/poll() in order to:
 251          *   1) Return if a signal occurs,
 252          *   2) Not deal with partially received messages.
 253          *
 254          * The drawback to this approach is that we assume that messages
 255          * are complete/well formed. If a message is shorter than its
 256          * announced length, receive_message() will block on recvmsg()
 257          * and never return (even if a signal is received).
 258          */
 259         ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
 260         if (ret < 0) {
 261                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 262                 goto end_unlock;
 263         }
 264         ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
 265         if (ret < 0) {
 266                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 267                 goto end_clean_poll;
 268         }
 269         ret = lttng_poll_wait_interruptible(&events, -1);
 270         if (ret <= 0) {
 271                 status = (ret == -1 && errno == EINTR) ?
 272                         LTTNG_NOTIFICATION_CHANNEL_STATUS_INTERRUPTED :
 273                         LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 274                 goto end_clean_poll;
 275         }
 276
 277         ret = receive_message(channel);
 278         if (ret) {
 279                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 280                 goto end_clean_poll;
 281         }
 282
 283         switch (get_current_message_type(channel)) {
 284         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 285                 notification = create_notification_from_current_message(
 286                                 channel);
 287                 if (!notification) {
 288                         status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 289                         goto end_clean_poll;
 290                 }
 291                 break;
 292         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 293                 /* No payload to consume. */
 294                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED;
 295                 break;
 296         default:
 297                 /* Protocol error. */
 298                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 299                 goto end_clean_poll;
 300         }
 301
 302 end_clean_poll:
 303         lttng_poll_clean(&events);
 304 end_unlock:
 305         pthread_mutex_unlock(&channel->lock);
 306         *_notification = notification;
 307 end:
 308         return status;
 309 }
 310
 311 static
 312 int enqueue_dropped_notification(
 313                 struct lttng_notification_channel *channel)
 314 {
 315         int ret = 0;
 316         struct pending_notification *pending_notification;
 317         struct cds_list_head *last_element =
 318                         channel->pending_notifications.list.prev;
 319
 320         pending_notification = caa_container_of(last_element,
 321                         struct pending_notification, node);
 322         if (!pending_notification->notification) {
 323                 /*
 324                  * The last enqueued notification indicates dropped
 325                  * notifications; there is nothing to do as we group
 326                  * dropped notifications together.
 327                  */
 328                 goto end;
 329         }
 330
 331         if (channel->pending_notifications.count >=
 332                         DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT &&
 333                         pending_notification->notification) {
 334                 /*
 335                  * Discard the last enqueued notification to indicate
 336                  * that notifications were dropped at this point.
 337                  */
 338                 lttng_notification_destroy(
 339                                 pending_notification->notification);
 340                 pending_notification->notification = NULL;
 341                 goto end;
 342         }
 343
 344         pending_notification = zmalloc(sizeof(*pending_notification));
 345         if (!pending_notification) {
 346                 ret = -1;
 347                 goto end;
 348         }
 349         CDS_INIT_LIST_HEAD(&pending_notification->node);
 350         cds_list_add(&pending_notification->node,
 351                         &channel->pending_notifications.list);
 352         channel->pending_notifications.count++;
 353 end:
 354         return ret;
 355 }
 356
 357 static
 358 int enqueue_notification_from_current_message(
 359                 struct lttng_notification_channel *channel)
 360 {
 361         int ret = 0;
 362         struct lttng_notification *notification;
 363         struct pending_notification *pending_notification;
 364
 365         if (channel->pending_notifications.count >=
 366                         DEFAULT_CLIENT_MAX_QUEUED_NOTIFICATIONS_COUNT) {
 367                 /* Drop the notification. */
 368                 ret = enqueue_dropped_notification(channel);
 369                 goto end;
 370         }
 371
 372         pending_notification = zmalloc(sizeof(*pending_notification));
 373         if (!pending_notification) {
 374                 ret = -1;
 375                 goto error;
 376         }
 377         CDS_INIT_LIST_HEAD(&pending_notification->node);
 378
 379         notification = create_notification_from_current_message(channel);
 380         if (!notification) {
 381                 ret = -1;
 382                 goto error;
 383         }
 384
 385         pending_notification->notification = notification;
 386         cds_list_add(&pending_notification->node,
 387                         &channel->pending_notifications.list);
 388         channel->pending_notifications.count++;
 389 end:
 390         return ret;
 391 error:
 392         free(pending_notification);
 393         goto end;
 394 }
 395
 396 enum lttng_notification_channel_status
 397 lttng_notification_channel_has_pending_notification(
 398                 struct lttng_notification_channel *channel,
 399                 bool *_notification_pending)
 400 {
 401         int ret;
 402         enum lttng_notification_channel_status status =
 403                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 404         struct lttng_poll_event events;
 405
 406         if (!channel || !_notification_pending) {
 407                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 408                 goto end;
 409         }
 410
 411         pthread_mutex_lock(&channel->lock);
 412
 413         if (channel->pending_notifications.count) {
 414                 *_notification_pending = true;
 415                 goto end_unlock;
 416         }
 417
 418         if (channel->socket < 0) {
 419                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED;
 420                 goto end_unlock;
 421         }
 422
 423         /*
 424          * Check, without blocking, if data is available on the channel's
 425          * socket. If there is data available, it is safe to read (blocking)
 426          * on the socket for a message from the session daemon.
 427          *
 428          * Since all commands wait for the session daemon's reply before
 429          * releasing the channel's lock, the protocol only allows for
 430          * notifications and "notification dropped" messages to come
 431          * through. If we receive a different message type, it is
 432          * considered a protocol error.
 433          *
 434          * Note that this function is not guaranteed not to block. This
 435          * will block until our peer (the session daemon) has sent a complete
 436          * message if we see data available on the socket. If the peer does
 437          * not respect the protocol, this may block indefinitely.
 438          */
 439         ret = lttng_poll_create(&events, 1, LTTNG_CLOEXEC);
 440         if (ret < 0) {
 441                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 442                 goto end_unlock;
 443         }
 444         ret = lttng_poll_add(&events, channel->socket, LPOLLIN | LPOLLERR);
 445         if (ret < 0) {
 446                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 447                 goto end_clean_poll;
 448         }
 449         /* timeout = 0: return immediately. */
 450         ret = lttng_poll_wait_interruptible(&events, 0);
 451         if (ret == 0) {
 452                 /* No data available. */
 453                 *_notification_pending = false;
 454                 goto end_clean_poll;
 455         } else if (ret < 0) {
 456                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 457                 goto end_clean_poll;
 458         }
 459
 460         /* Data available on socket. */
 461         ret = receive_message(channel);
 462         if (ret) {
 463                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 464                 goto end_clean_poll;
 465         }
 466
 467         switch (get_current_message_type(channel)) {
 468         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 469                 ret = enqueue_notification_from_current_message(channel);
 470                 if (ret) {
 471                         goto end_clean_poll;
 472                 }
 473                 *_notification_pending = true;
 474                 break;
 475         case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 476                 ret = enqueue_dropped_notification(channel);
 477                 if (ret) {
 478                         goto end_clean_poll;
 479                 }
 480                 *_notification_pending = true;
 481                 break;
 482         default:
 483                 /* Protocol error. */
 484                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 485                 goto end_clean_poll;
 486         }
 487
 488 end_clean_poll:
 489         lttng_poll_clean(&events);
 490 end_unlock:
 491         pthread_mutex_unlock(&channel->lock);
 492 end:
 493         return status;
 494 }
 495
 496 static
 497 int receive_command_reply(struct lttng_notification_channel *channel,
 498                 enum lttng_notification_channel_status *status)
 499 {
 500         int ret;
 501         struct lttng_notification_channel_command_reply *reply;
 502
 503         while (true) {
 504                 enum lttng_notification_channel_message_type msg_type;
 505
 506                 ret = receive_message(channel);
 507                 if (ret) {
 508                         goto end;
 509                 }
 510
 511                 msg_type = get_current_message_type(channel);
 512                 switch (msg_type) {
 513                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY:
 514                         goto exit_loop;
 515                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION:
 516                         ret = enqueue_notification_from_current_message(
 517                                         channel);
 518                         if (ret) {
 519                                 goto end;
 520                         }
 521                         break;
 522                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED:
 523                         ret = enqueue_dropped_notification(channel);
 524                         if (ret) {
 525                                 goto end;
 526                         }
 527                         break;
 528                 case LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE:
 529                 {
 530                         struct lttng_notification_channel_command_handshake *handshake;
 531
 532                         handshake = (struct lttng_notification_channel_command_handshake *)
 533                                         (channel->reception_buffer.data +
 534                                         sizeof(struct lttng_notification_channel_message));
 535                         channel->version.major = handshake->major;
 536                         channel->version.minor = handshake->minor;
 537                         channel->version.set = true;
 538                         break;
 539                 }
 540                 default:
 541                         ret = -1;
 542                         goto end;
 543                 }
 544         }
 545
 546 exit_loop:
 547         if (channel->reception_buffer.size <
 548                         (sizeof(struct lttng_notification_channel_message) +
 549                         sizeof(*reply))) {
 550                 /* Invalid message received. */
 551                 ret = -1;
 552                 goto end;
 553         }
 554
 555         reply = (struct lttng_notification_channel_command_reply *)
 556                         (channel->reception_buffer.data +
 557                         sizeof(struct lttng_notification_channel_message));
 558         *status = (enum lttng_notification_channel_status) reply->status;
 559 end:
 560         return ret;
 561 }
 562
 563 static
 564 int handshake(struct lttng_notification_channel *channel)
 565 {
 566         ssize_t ret;
 567         enum lttng_notification_channel_status status =
 568                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 569         struct lttng_notification_channel_command_handshake handshake = {
 570                 .major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
 571                 .minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
 572         };
 573         struct lttng_notification_channel_message msg_header = {
 574                 .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
 575                 .size = sizeof(handshake),
 576         };
 577         char send_buffer[sizeof(msg_header) + sizeof(handshake)];
 578
 579         memcpy(send_buffer, &msg_header, sizeof(msg_header));
 580         memcpy(send_buffer + sizeof(msg_header), &handshake, sizeof(handshake));
 581
 582         pthread_mutex_lock(&channel->lock);
 583
 584         ret = lttcomm_send_creds_unix_sock(channel->socket, send_buffer,
 585                         sizeof(send_buffer));
 586         if (ret < 0) {
 587                 goto end_unlock;
 588         }
 589
 590         /* Receive handshake info from the sessiond. */
 591         ret = receive_command_reply(channel, &status);
 592         if (ret < 0) {
 593                 goto end_unlock;
 594         }
 595
 596         if (!channel->version.set) {
 597                 ret = -1;
 598                 goto end_unlock;
 599         }
 600
 601         if (channel->version.major != LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR) {
 602                 ret = -1;
 603                 goto end_unlock;
 604         }
 605
 606 end_unlock:
 607         pthread_mutex_unlock(&channel->lock);
 608         return ret;
 609 }
 610
 611 static
 612 enum lttng_notification_channel_status send_condition_command(
 613                 struct lttng_notification_channel *channel,
 614                 enum lttng_notification_channel_message_type type,
 615                 const struct lttng_condition *condition)
 616 {
 617         int socket;
 618         ssize_t ret;
 619         enum lttng_notification_channel_status status =
 620                         LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
 621         struct lttng_dynamic_buffer buffer;
 622         struct lttng_notification_channel_message cmd_header = {
 623                 .type = (int8_t) type,
 624         };
 625
 626         lttng_dynamic_buffer_init(&buffer);
 627
 628         if (!channel) {
 629                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 630                 goto end;
 631         }
 632
 633         assert(type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE ||
 634                 type == LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE);
 635
 636         pthread_mutex_lock(&channel->lock);
 637         socket = channel->socket;
 638         if (!lttng_condition_validate(condition)) {
 639                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 640                 goto end_unlock;
 641         }
 642
 643         ret = lttng_dynamic_buffer_append(&buffer, &cmd_header,
 644                         sizeof(cmd_header));
 645         if (ret) {
 646                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 647                 goto end_unlock;
 648         }
 649
 650         ret = lttng_condition_serialize(condition, &buffer);
 651         if (ret) {
 652                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_INVALID;
 653                 goto end_unlock;
 654         }
 655
 656         /* Update payload length. */
 657         ((struct lttng_notification_channel_message *) buffer.data)->size =
 658                         (uint32_t) (buffer.size - sizeof(cmd_header));
 659
 660         ret = lttcomm_send_unix_sock(socket, buffer.data, buffer.size);
 661         if (ret < 0) {
 662                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 663                 goto end_unlock;
 664         }
 665
 666         ret = receive_command_reply(channel, &status);
 667         if (ret < 0) {
 668                 status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ERROR;
 669                 goto end_unlock;
 670         }
 671 end_unlock:
 672         pthread_mutex_unlock(&channel->lock);
 673 end:
 674         lttng_dynamic_buffer_reset(&buffer);
 675         return status;
 676 }
 677
 678 enum lttng_notification_channel_status lttng_notification_channel_subscribe(
 679                 struct lttng_notification_channel *channel,
 680                 const struct lttng_condition *condition)
 681 {
 682         return send_condition_command(channel,
 683                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_SUBSCRIBE,
 684                         condition);
 685 }
 686
 687 enum lttng_notification_channel_status lttng_notification_channel_unsubscribe(
 688                 struct lttng_notification_channel *channel,
 689                 const struct lttng_condition *condition)
 690 {
 691         return send_condition_command(channel,
 692                         LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_UNSUBSCRIBE,
 693                         condition);
 694 }
 695
 696 void lttng_notification_channel_destroy(
 697                 struct lttng_notification_channel *channel)
 698 {
 699         if (!channel) {
 700                 return;
 701         }
 702
 703         if (channel->socket >= 0) {
 704                 (void) lttcomm_close_unix_sock(channel->socket);
 705         }
 706         pthread_mutex_destroy(&channel->lock);
 707         lttng_dynamic_buffer_reset(&channel->reception_buffer);
 708         free(channel);
 709 }