src/bin/lttng-sessiond/consumer.c

   1 /*
   2  * Copyright (C) 2012 - David Goulet <dgoulet@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 _GNU_SOURCE
  19 #include <assert.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <sys/stat.h>
  24 #include <sys/types.h>
  25 #include <unistd.h>
  26 #include <inttypes.h>
  27
  28 #include <common/common.h>
  29 #include <common/defaults.h>
  30 #include <common/uri.h>
  31
  32 #include "consumer.h"
  33 #include "health.h"
  34 #include "ust-app.h"
  35 #include "utils.h"
  36
  37 /*
  38  * Receive a reply command status message from the consumer. Consumer socket
  39  * lock MUST be acquired before calling this function.
  40  *
  41  * Return 0 on success, -1 on recv error or a negative lttng error code which
  42  * was possibly returned by the consumer.
  43  */
  44 int consumer_recv_status_reply(struct consumer_socket *sock)
  45 {
  46         int ret;
  47         struct lttcomm_consumer_status_msg reply;
  48
  49         assert(sock);
  50
  51         ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
  52         if (ret <= 0) {
  53                 if (ret == 0) {
  54                         /* Orderly shutdown. Don't return 0 which means success. */
  55                         ret = -1;
  56                 }
  57                 /* The above call will print a PERROR on error. */
  58                 DBG("Fail to receive status reply on sock %d", sock->fd);
  59                 goto end;
  60         }
  61
  62         if (reply.ret_code == LTTNG_OK) {
  63                 /* All good. */
  64                 ret = 0;
  65         } else {
  66                 ret = -reply.ret_code;
  67                 DBG("Consumer ret code %d", ret);
  68         }
  69
  70 end:
  71         return ret;
  72 }
  73
  74 /*
  75  * Once the ASK_CHANNEL command is sent to the consumer, the channel
  76  * information are sent back. This call receives that data and populates key
  77  * and stream_count.
  78  *
  79  * On success return 0 and both key and stream_count are set. On error, a
  80  * negative value is sent back and both parameters are untouched.
  81  */
  82 int consumer_recv_status_channel(struct consumer_socket *sock,
  83                 uint64_t *key, unsigned int *stream_count)
  84 {
  85         int ret;
  86         struct lttcomm_consumer_status_channel reply;
  87
  88         assert(sock);
  89         assert(stream_count);
  90         assert(key);
  91
  92         ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
  93         if (ret <= 0) {
  94                 if (ret == 0) {
  95                         /* Orderly shutdown. Don't return 0 which means success. */
  96                         ret = -1;
  97                 }
  98                 /* The above call will print a PERROR on error. */
  99                 DBG("Fail to receive status reply on sock %d", sock->fd);
 100                 goto end;
 101         }
 102
 103         /* An error is possible so don't touch the key and stream_count. */
 104         if (reply.ret_code != LTTNG_OK) {
 105                 ret = -1;
 106                 goto end;
 107         }
 108
 109         *key = reply.key;
 110         *stream_count = reply.stream_count;
 111
 112 end:
 113         return ret;
 114 }
 115
 116 /*
 117  * Send destroy relayd command to consumer.
 118  *
 119  * On success return positive value. On error, negative value.
 120  */
 121 int consumer_send_destroy_relayd(struct consumer_socket *sock,
 122                 struct consumer_output *consumer)
 123 {
 124         int ret;
 125         struct lttcomm_consumer_msg msg;
 126
 127         assert(consumer);
 128         assert(sock);
 129
 130         DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
 131
 132         /* Bail out if consumer is disabled */
 133         if (!consumer->enabled) {
 134                 ret = LTTNG_OK;
 135                 DBG3("Consumer is disabled");
 136                 goto error;
 137         }
 138
 139         msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
 140         msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
 141
 142         pthread_mutex_lock(sock->lock);
 143         ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
 144         if (ret < 0) {
 145                 /* Indicate that the consumer is probably closing at this point. */
 146                 DBG("send consumer destroy relayd command");
 147                 goto error_send;
 148         }
 149
 150         /* Don't check the return value. The caller will do it. */
 151         ret = consumer_recv_status_reply(sock);
 152
 153         DBG2("Consumer send destroy relayd command done");
 154
 155 error_send:
 156         pthread_mutex_unlock(sock->lock);
 157 error:
 158         return ret;
 159 }
 160
 161 /*
 162  * For each consumer socket in the consumer output object, send a destroy
 163  * relayd command.
 164  */
 165 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
 166 {
 167         struct lttng_ht_iter iter;
 168         struct consumer_socket *socket;
 169
 170         assert(consumer);
 171
 172         /* Destroy any relayd connection */
 173         if (consumer->type == CONSUMER_DST_NET) {
 174                 rcu_read_lock();
 175                 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 176                                 node.node) {
 177                         int ret;
 178
 179                         /* Send destroy relayd command */
 180                         ret = consumer_send_destroy_relayd(socket, consumer);
 181                         if (ret < 0) {
 182                                 DBG("Unable to send destroy relayd command to consumer");
 183                                 /* Continue since we MUST delete everything at this point. */
 184                         }
 185                 }
 186                 rcu_read_unlock();
 187         }
 188 }
 189
 190 /*
 191  * From a consumer_data structure, allocate and add a consumer socket to the
 192  * consumer output.
 193  *
 194  * Return 0 on success, else negative value on error
 195  */
 196 int consumer_create_socket(struct consumer_data *data,
 197                 struct consumer_output *output)
 198 {
 199         int ret = 0;
 200         struct consumer_socket *socket;
 201
 202         assert(data);
 203
 204         if (output == NULL || data->cmd_sock < 0) {
 205                 /*
 206                  * Not an error. Possible there is simply not spawned consumer or it's
 207                  * disabled for the tracing session asking the socket.
 208                  */
 209                 goto error;
 210         }
 211
 212         rcu_read_lock();
 213         socket = consumer_find_socket(data->cmd_sock, output);
 214         rcu_read_unlock();
 215         if (socket == NULL) {
 216                 socket = consumer_allocate_socket(data->cmd_sock);
 217                 if (socket == NULL) {
 218                         ret = -1;
 219                         goto error;
 220                 }
 221
 222                 socket->registered = 0;
 223                 socket->lock = &data->lock;
 224                 rcu_read_lock();
 225                 consumer_add_socket(socket, output);
 226                 rcu_read_unlock();
 227         }
 228
 229         socket->type = data->type;
 230
 231         DBG3("Consumer socket created (fd: %d) and added to output",
 232                         data->cmd_sock);
 233
 234 error:
 235         return ret;
 236 }
 237
 238 /*
 239  * Return the consumer socket from the given consumer output with the right
 240  * bitness. On error, returns NULL.
 241  *
 242  * The caller MUST acquire a rcu read side lock and keep it until the socket
 243  * object reference is not needed anymore.
 244  */
 245 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
 246                 struct consumer_output *consumer)
 247 {
 248         int consumer_fd;
 249         struct consumer_socket *socket = NULL;
 250
 251         switch (bits) {
 252         case 64:
 253                 consumer_fd = uatomic_read(&ust_consumerd64_fd);
 254                 break;
 255         case 32:
 256                 consumer_fd = uatomic_read(&ust_consumerd32_fd);
 257                 break;
 258         default:
 259                 assert(0);
 260                 goto end;
 261         }
 262
 263         socket = consumer_find_socket(consumer_fd, consumer);
 264         if (!socket) {
 265                 ERR("Consumer socket fd %d not found in consumer obj %p",
 266                                 consumer_fd, consumer);
 267         }
 268
 269 end:
 270         return socket;
 271 }
 272
 273 /*
 274  * Find a consumer_socket in a consumer_output hashtable. Read side lock must
 275  * be acquired before calling this function and across use of the
 276  * returned consumer_socket.
 277  */
 278 struct consumer_socket *consumer_find_socket(int key,
 279                 struct consumer_output *consumer)
 280 {
 281         struct lttng_ht_iter iter;
 282         struct lttng_ht_node_ulong *node;
 283         struct consumer_socket *socket = NULL;
 284
 285         /* Negative keys are lookup failures */
 286         if (key < 0 || consumer == NULL) {
 287                 return NULL;
 288         }
 289
 290         lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
 291                         &iter);
 292         node = lttng_ht_iter_get_node_ulong(&iter);
 293         if (node != NULL) {
 294                 socket = caa_container_of(node, struct consumer_socket, node);
 295         }
 296
 297         return socket;
 298 }
 299
 300 /*
 301  * Allocate a new consumer_socket and return the pointer.
 302  */
 303 struct consumer_socket *consumer_allocate_socket(int fd)
 304 {
 305         struct consumer_socket *socket = NULL;
 306
 307         socket = zmalloc(sizeof(struct consumer_socket));
 308         if (socket == NULL) {
 309                 PERROR("zmalloc consumer socket");
 310                 goto error;
 311         }
 312
 313         socket->fd = fd;
 314         lttng_ht_node_init_ulong(&socket->node, fd);
 315
 316 error:
 317         return socket;
 318 }
 319
 320 /*
 321  * Add consumer socket to consumer output object. Read side lock must be
 322  * acquired before calling this function.
 323  */
 324 void consumer_add_socket(struct consumer_socket *sock,
 325                 struct consumer_output *consumer)
 326 {
 327         assert(sock);
 328         assert(consumer);
 329
 330         lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
 331 }
 332
 333 /*
 334  * Delte consumer socket to consumer output object. Read side lock must be
 335  * acquired before calling this function.
 336  */
 337 void consumer_del_socket(struct consumer_socket *sock,
 338                 struct consumer_output *consumer)
 339 {
 340         int ret;
 341         struct lttng_ht_iter iter;
 342
 343         assert(sock);
 344         assert(consumer);
 345
 346         iter.iter.node = &sock->node.node;
 347         ret = lttng_ht_del(consumer->socks, &iter);
 348         assert(!ret);
 349 }
 350
 351 /*
 352  * RCU destroy call function.
 353  */
 354 static void destroy_socket_rcu(struct rcu_head *head)
 355 {
 356         struct lttng_ht_node_ulong *node =
 357                 caa_container_of(head, struct lttng_ht_node_ulong, head);
 358         struct consumer_socket *socket =
 359                 caa_container_of(node, struct consumer_socket, node);
 360
 361         free(socket);
 362 }
 363
 364 /*
 365  * Destroy and free socket pointer in a call RCU. Read side lock must be
 366  * acquired before calling this function.
 367  */
 368 void consumer_destroy_socket(struct consumer_socket *sock)
 369 {
 370         assert(sock);
 371
 372         /*
 373          * We DO NOT close the file descriptor here since it is global to the
 374          * session daemon and is closed only if the consumer dies or a custom
 375          * consumer was registered,
 376          */
 377         if (sock->registered) {
 378                 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
 379                 lttcomm_close_unix_sock(sock->fd);
 380         }
 381
 382         call_rcu(&sock->node.head, destroy_socket_rcu);
 383 }
 384
 385 /*
 386  * Allocate and assign data to a consumer_output object.
 387  *
 388  * Return pointer to structure.
 389  */
 390 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
 391 {
 392         struct consumer_output *output = NULL;
 393
 394         output = zmalloc(sizeof(struct consumer_output));
 395         if (output == NULL) {
 396                 PERROR("zmalloc consumer_output");
 397                 goto error;
 398         }
 399
 400         /* By default, consumer output is enabled */
 401         output->enabled = 1;
 402         output->type = type;
 403         output->net_seq_index = (uint64_t) -1ULL;
 404
 405         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 406
 407 error:
 408         return output;
 409 }
 410
 411 /*
 412  * Iterate over the consumer output socket hash table and destroy them. The
 413  * socket file descriptor are only closed if the consumer output was
 414  * registered meaning it's an external consumer.
 415  */
 416 void consumer_destroy_output_sockets(struct consumer_output *obj)
 417 {
 418         struct lttng_ht_iter iter;
 419         struct consumer_socket *socket;
 420
 421         if (!obj->socks) {
 422                 return;
 423         }
 424
 425         rcu_read_lock();
 426         cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 427                 consumer_del_socket(socket, obj);
 428                 consumer_destroy_socket(socket);
 429         }
 430         rcu_read_unlock();
 431 }
 432
 433 /*
 434  * Delete the consumer_output object from the list and free the ptr.
 435  *
 436  * Should *NOT* be called with RCU read-side lock held.
 437  */
 438 void consumer_destroy_output(struct consumer_output *obj)
 439 {
 440         if (obj == NULL) {
 441                 return;
 442         }
 443
 444         consumer_destroy_output_sockets(obj);
 445
 446         if (obj->socks) {
 447                 /* Finally destroy HT */
 448                 ht_cleanup_push(obj->socks);
 449         }
 450
 451         free(obj);
 452 }
 453
 454 /*
 455  * Copy consumer output and returned the newly allocated copy.
 456  *
 457  * Should *NOT* be called with RCU read-side lock held.
 458  */
 459 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
 460 {
 461         int ret;
 462         struct lttng_ht *tmp_ht_ptr;
 463         struct consumer_output *output;
 464
 465         assert(obj);
 466
 467         output = consumer_create_output(obj->type);
 468         if (output == NULL) {
 469                 goto error;
 470         }
 471         /* Avoid losing the HT reference after the memcpy() */
 472         tmp_ht_ptr = output->socks;
 473
 474         memcpy(output, obj, sizeof(struct consumer_output));
 475
 476         /* Putting back the HT pointer and start copying socket(s). */
 477         output->socks = tmp_ht_ptr;
 478
 479         ret = consumer_copy_sockets(output, obj);
 480         if (ret < 0) {
 481                 goto malloc_error;
 482         }
 483
 484 error:
 485         return output;
 486
 487 malloc_error:
 488         consumer_destroy_output(output);
 489         return NULL;
 490 }
 491
 492 /*
 493  * Copy consumer sockets from src to dst.
 494  *
 495  * Return 0 on success or else a negative value.
 496  */
 497 int consumer_copy_sockets(struct consumer_output *dst,
 498                 struct consumer_output *src)
 499 {
 500         int ret = 0;
 501         struct lttng_ht_iter iter;
 502         struct consumer_socket *socket, *copy_sock;
 503
 504         assert(dst);
 505         assert(src);
 506
 507         rcu_read_lock();
 508         cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
 509                 /* Ignore socket that are already there. */
 510                 copy_sock = consumer_find_socket(socket->fd, dst);
 511                 if (copy_sock) {
 512                         continue;
 513                 }
 514
 515                 /* Create new socket object. */
 516                 copy_sock = consumer_allocate_socket(socket->fd);
 517                 if (copy_sock == NULL) {
 518                         rcu_read_unlock();
 519                         ret = -ENOMEM;
 520                         goto error;
 521                 }
 522
 523                 copy_sock->registered = socket->registered;
 524                 /*
 525                  * This is valid because this lock is shared accross all consumer
 526                  * object being the global lock of the consumer data structure of the
 527                  * session daemon.
 528                  */
 529                 copy_sock->lock = socket->lock;
 530                 consumer_add_socket(copy_sock, dst);
 531         }
 532         rcu_read_unlock();
 533
 534 error:
 535         return ret;
 536 }
 537
 538 /*
 539  * Set network URI to the consumer output object.
 540  *
 541  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 542  * error.
 543  */
 544 int consumer_set_network_uri(struct consumer_output *obj,
 545                 struct lttng_uri *uri)
 546 {
 547         int ret;
 548         char tmp_path[PATH_MAX];
 549         char hostname[HOST_NAME_MAX];
 550         struct lttng_uri *dst_uri = NULL;
 551
 552         /* Code flow error safety net. */
 553         assert(obj);
 554         assert(uri);
 555
 556         switch (uri->stype) {
 557         case LTTNG_STREAM_CONTROL:
 558                 dst_uri = &obj->dst.net.control;
 559                 obj->dst.net.control_isset = 1;
 560                 if (uri->port == 0) {
 561                         /* Assign default port. */
 562                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 563                 } else {
 564                         if (obj->dst.net.data_isset && uri->port ==
 565                                         obj->dst.net.data.port) {
 566                                 ret = -LTTNG_ERR_INVALID;
 567                                 goto error;
 568                         }
 569                 }
 570                 DBG3("Consumer control URI set with port %d", uri->port);
 571                 break;
 572         case LTTNG_STREAM_DATA:
 573                 dst_uri = &obj->dst.net.data;
 574                 obj->dst.net.data_isset = 1;
 575                 if (uri->port == 0) {
 576                         /* Assign default port. */
 577                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 578                 } else {
 579                         if (obj->dst.net.control_isset && uri->port ==
 580                                         obj->dst.net.control.port) {
 581                                 ret = -LTTNG_ERR_INVALID;
 582                                 goto error;
 583                         }
 584                 }
 585                 DBG3("Consumer data URI set with port %d", uri->port);
 586                 break;
 587         default:
 588                 ERR("Set network uri type unknown %d", uri->stype);
 589                 ret = -LTTNG_ERR_INVALID;
 590                 goto error;
 591         }
 592
 593         ret = uri_compare(dst_uri, uri);
 594         if (!ret) {
 595                 /* Same URI, don't touch it and return success. */
 596                 DBG3("URI network compare are the same");
 597                 goto equal;
 598         }
 599
 600         /* URIs were not equal, replacing it. */
 601         memset(dst_uri, 0, sizeof(struct lttng_uri));
 602         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 603         obj->type = CONSUMER_DST_NET;
 604
 605         /* Handle subdir and add hostname in front. */
 606         if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
 607                 /* Get hostname to append it in the pathname */
 608                 ret = gethostname(hostname, sizeof(hostname));
 609                 if (ret < 0) {
 610                         PERROR("gethostname. Fallback on default localhost");
 611                         strncpy(hostname, "localhost", sizeof(hostname));
 612                 }
 613                 hostname[sizeof(hostname) - 1] = '\0';
 614
 615                 /* Setup consumer subdir if none present in the control URI */
 616                 if (strlen(dst_uri->subdir) == 0) {
 617                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 618                                         hostname, obj->subdir);
 619                 } else {
 620                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 621                                         hostname, dst_uri->subdir);
 622                 }
 623                 if (ret < 0) {
 624                         PERROR("snprintf set consumer uri subdir");
 625                         ret = -LTTNG_ERR_NOMEM;
 626                         goto error;
 627                 }
 628
 629                 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
 630                 DBG3("Consumer set network uri subdir path %s", tmp_path);
 631         }
 632
 633         return 0;
 634 equal:
 635         return 1;
 636 error:
 637         return ret;
 638 }
 639
 640 /*
 641  * Send file descriptor to consumer via sock.
 642  */
 643 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 644 {
 645         int ret;
 646
 647         assert(fds);
 648         assert(sock);
 649         assert(nb_fd > 0);
 650
 651         ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
 652         if (ret < 0) {
 653                 /* The above call will print a PERROR on error. */
 654                 DBG("Error when sending consumer fds on sock %d", sock->fd);
 655                 goto error;
 656         }
 657
 658         ret = consumer_recv_status_reply(sock);
 659
 660 error:
 661         return ret;
 662 }
 663
 664 /*
 665  * Consumer send communication message structure to consumer.
 666  */
 667 int consumer_send_msg(struct consumer_socket *sock,
 668                 struct lttcomm_consumer_msg *msg)
 669 {
 670         int ret;
 671
 672         assert(msg);
 673         assert(sock);
 674         assert(sock->fd >= 0);
 675
 676         ret = lttcomm_send_unix_sock(sock->fd, msg,
 677                         sizeof(struct lttcomm_consumer_msg));
 678         if (ret < 0) {
 679                 /* The above call will print a PERROR on error. */
 680                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 681                 goto error;
 682         }
 683
 684         ret = consumer_recv_status_reply(sock);
 685
 686 error:
 687         return ret;
 688 }
 689
 690 /*
 691  * Consumer send channel communication message structure to consumer.
 692  */
 693 int consumer_send_channel(struct consumer_socket *sock,
 694                 struct lttcomm_consumer_msg *msg)
 695 {
 696         int ret;
 697
 698         assert(msg);
 699         assert(sock);
 700         assert(sock->fd >= 0);
 701
 702         ret = lttcomm_send_unix_sock(sock->fd, msg,
 703                         sizeof(struct lttcomm_consumer_msg));
 704         if (ret < 0) {
 705                 /* The above call will print a PERROR on error. */
 706                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 707                 goto error;
 708         }
 709
 710         ret = consumer_recv_status_reply(sock);
 711
 712 error:
 713         return ret;
 714 }
 715
 716 /*
 717  * Populate the given consumer msg structure with the ask_channel command
 718  * information.
 719  */
 720 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 721                 uint64_t subbuf_size,
 722                 uint64_t num_subbuf,
 723                 int overwrite,
 724                 unsigned int switch_timer_interval,
 725                 unsigned int read_timer_interval,
 726                 int output,
 727                 int type,
 728                 uint64_t session_id,
 729                 const char *pathname,
 730                 const char *name,
 731                 uid_t uid,
 732                 gid_t gid,
 733                 uint64_t relayd_id,
 734                 uint64_t key,
 735                 unsigned char *uuid,
 736                 uint32_t chan_id,
 737                 uint64_t tracefile_size,
 738                 uint64_t tracefile_count,
 739                 uint64_t session_id_per_pid,
 740                 unsigned int monitor,
 741                 uint32_t ust_app_uid)
 742 {
 743         assert(msg);
 744
 745         /* Zeroed structure */
 746         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 747
 748         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 749         msg->u.ask_channel.subbuf_size = subbuf_size;
 750         msg->u.ask_channel.num_subbuf = num_subbuf ;
 751         msg->u.ask_channel.overwrite = overwrite;
 752         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 753         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 754         msg->u.ask_channel.output = output;
 755         msg->u.ask_channel.type = type;
 756         msg->u.ask_channel.session_id = session_id;
 757         msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
 758         msg->u.ask_channel.uid = uid;
 759         msg->u.ask_channel.gid = gid;
 760         msg->u.ask_channel.relayd_id = relayd_id;
 761         msg->u.ask_channel.key = key;
 762         msg->u.ask_channel.chan_id = chan_id;
 763         msg->u.ask_channel.tracefile_size = tracefile_size;
 764         msg->u.ask_channel.tracefile_count = tracefile_count;
 765         msg->u.ask_channel.monitor = monitor;
 766         msg->u.ask_channel.ust_app_uid = ust_app_uid;
 767
 768         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 769
 770         if (pathname) {
 771                 strncpy(msg->u.ask_channel.pathname, pathname,
 772                                 sizeof(msg->u.ask_channel.pathname));
 773                 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 774         }
 775
 776         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 777         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 778 }
 779
 780 /*
 781  * Init channel communication message structure.
 782  */
 783 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 784                 enum lttng_consumer_command cmd,
 785                 uint64_t channel_key,
 786                 uint64_t session_id,
 787                 const char *pathname,
 788                 uid_t uid,
 789                 gid_t gid,
 790                 uint64_t relayd_id,
 791                 const char *name,
 792                 unsigned int nb_init_streams,
 793                 enum lttng_event_output output,
 794                 int type,
 795                 uint64_t tracefile_size,
 796                 uint64_t tracefile_count,
 797                 unsigned int monitor)
 798 {
 799         assert(msg);
 800
 801         /* Zeroed structure */
 802         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 803
 804         /* Send channel */
 805         msg->cmd_type = cmd;
 806         msg->u.channel.channel_key = channel_key;
 807         msg->u.channel.session_id = session_id;
 808         msg->u.channel.uid = uid;
 809         msg->u.channel.gid = gid;
 810         msg->u.channel.relayd_id = relayd_id;
 811         msg->u.channel.nb_init_streams = nb_init_streams;
 812         msg->u.channel.output = output;
 813         msg->u.channel.type = type;
 814         msg->u.channel.tracefile_size = tracefile_size;
 815         msg->u.channel.tracefile_count = tracefile_count;
 816         msg->u.channel.monitor = monitor;
 817
 818         strncpy(msg->u.channel.pathname, pathname,
 819                         sizeof(msg->u.channel.pathname));
 820         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 821
 822         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 823         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 824 }
 825
 826 /*
 827  * Init stream communication message structure.
 828  */
 829 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 830                 enum lttng_consumer_command cmd,
 831                 uint64_t channel_key,
 832                 uint64_t stream_key,
 833                 int cpu)
 834 {
 835         assert(msg);
 836
 837         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 838
 839         msg->cmd_type = cmd;
 840         msg->u.stream.channel_key = channel_key;
 841         msg->u.stream.stream_key = stream_key;
 842         msg->u.stream.cpu = cpu;
 843 }
 844
 845 /*
 846  * Send stream communication structure to the consumer.
 847  */
 848 int consumer_send_stream(struct consumer_socket *sock,
 849                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 850                 int *fds, size_t nb_fd)
 851 {
 852         int ret;
 853
 854         assert(msg);
 855         assert(dst);
 856         assert(sock);
 857         assert(fds);
 858
 859         /* Send on socket */
 860         ret = lttcomm_send_unix_sock(sock->fd, msg,
 861                         sizeof(struct lttcomm_consumer_msg));
 862         if (ret < 0) {
 863                 /* The above call will print a PERROR on error. */
 864                 DBG("Error when sending consumer stream on sock %d", sock->fd);
 865                 goto error;
 866         }
 867
 868         ret = consumer_recv_status_reply(sock);
 869         if (ret < 0) {
 870                 goto error;
 871         }
 872
 873         ret = consumer_send_fds(sock, fds, nb_fd);
 874         if (ret < 0) {
 875                 goto error;
 876         }
 877
 878 error:
 879         return ret;
 880 }
 881
 882 /*
 883  * Send relayd socket to consumer associated with a session name.
 884  *
 885  * On success return positive value. On error, negative value.
 886  */
 887 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 888                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
 889                 enum lttng_stream_type type, uint64_t session_id)
 890 {
 891         int ret;
 892         struct lttcomm_consumer_msg msg;
 893
 894         /* Code flow error. Safety net. */
 895         assert(rsock);
 896         assert(consumer);
 897         assert(consumer_sock);
 898
 899         /* Bail out if consumer is disabled */
 900         if (!consumer->enabled) {
 901                 ret = LTTNG_OK;
 902                 goto error;
 903         }
 904
 905         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 906         /*
 907          * Assign network consumer output index using the temporary consumer since
 908          * this call should only be made from within a set_consumer_uri() function
 909          * call in the session daemon.
 910          */
 911         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 912         msg.u.relayd_sock.type = type;
 913         msg.u.relayd_sock.session_id = session_id;
 914         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 915
 916         DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
 917         ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
 918         if (ret < 0) {
 919                 /* The above call will print a PERROR on error. */
 920                 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
 921                 goto error;
 922         }
 923
 924         ret = consumer_recv_status_reply(consumer_sock);
 925         if (ret < 0) {
 926                 goto error;
 927         }
 928
 929         DBG3("Sending relayd socket file descriptor to consumer");
 930         ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
 931         if (ret < 0) {
 932                 goto error;
 933         }
 934
 935         DBG2("Consumer relayd socket sent");
 936
 937 error:
 938         return ret;
 939 }
 940
 941 /*
 942  * Set consumer subdirectory using the session name and a generated datetime if
 943  * needed. This is appended to the current subdirectory.
 944  */
 945 int consumer_set_subdir(struct consumer_output *consumer,
 946                 const char *session_name)
 947 {
 948         int ret = 0;
 949         unsigned int have_default_name = 0;
 950         char datetime[16], tmp_path[PATH_MAX];
 951         time_t rawtime;
 952         struct tm *timeinfo;
 953
 954         assert(consumer);
 955         assert(session_name);
 956
 957         memset(tmp_path, 0, sizeof(tmp_path));
 958
 959         /* Flag if we have a default session. */
 960         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 961                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 962                 have_default_name = 1;
 963         } else {
 964                 /* Get date and time for session path */
 965                 time(&rawtime);
 966                 timeinfo = localtime(&rawtime);
 967                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 968         }
 969
 970         if (have_default_name) {
 971                 ret = snprintf(tmp_path, sizeof(tmp_path),
 972                                 "%s/%s", consumer->subdir, session_name);
 973         } else {
 974                 ret = snprintf(tmp_path, sizeof(tmp_path),
 975                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 976         }
 977         if (ret < 0) {
 978                 PERROR("snprintf session name date");
 979                 goto error;
 980         }
 981
 982         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 983         DBG2("Consumer subdir set to %s", consumer->subdir);
 984
 985 error:
 986         return ret;
 987 }
 988
 989 /*
 990  * Ask the consumer if the data is ready to read (NOT pending) for the specific
 991  * session id.
 992  *
 993  * This function has a different behavior with the consumer i.e. that it waits
 994  * for a reply from the consumer if yes or no the data is pending.
 995  */
 996 int consumer_is_data_pending(uint64_t session_id,
 997                 struct consumer_output *consumer)
 998 {
 999         int ret;
1000         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
1001         struct consumer_socket *socket;
1002         struct lttng_ht_iter iter;
1003         struct lttcomm_consumer_msg msg;
1004
1005         assert(consumer);
1006
1007         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1008
1009         msg.u.data_pending.session_id = session_id;
1010
1011         DBG3("Consumer data pending for id %" PRIu64, session_id);
1012
1013         /* Send command for each consumer */
1014         rcu_read_lock();
1015         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1016                         node.node) {
1017                 /* Code flow error */
1018                 assert(socket->fd >= 0);
1019
1020                 pthread_mutex_lock(socket->lock);
1021
1022                 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
1023                 if (ret < 0) {
1024                         /* The above call will print a PERROR on error. */
1025                         DBG("Error on consumer is data pending on sock %d", socket->fd);
1026                         pthread_mutex_unlock(socket->lock);
1027                         goto error_unlock;
1028                 }
1029
1030                 /*
1031                  * No need for a recv reply status because the answer to the command is
1032                  * the reply status message.
1033                  */
1034
1035                 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
1036                 if (ret <= 0) {
1037                         if (ret == 0) {
1038                                 /* Orderly shutdown. Don't return 0 which means success. */
1039                                 ret = -1;
1040                         }
1041                         /* The above call will print a PERROR on error. */
1042                         DBG("Error on recv consumer is data pending on sock %d", socket->fd);
1043                         pthread_mutex_unlock(socket->lock);
1044                         goto error_unlock;
1045                 }
1046
1047                 pthread_mutex_unlock(socket->lock);
1048
1049                 if (ret_code == 1) {
1050                         break;
1051                 }
1052         }
1053         rcu_read_unlock();
1054
1055         DBG("Consumer data is %s pending for session id %" PRIu64,
1056                         ret_code == 1 ? "" : "NOT", session_id);
1057         return ret_code;
1058
1059 error_unlock:
1060         rcu_read_unlock();
1061         return -1;
1062 }
1063
1064 /*
1065  * Send a flush command to consumer using the given channel key.
1066  *
1067  * Return 0 on success else a negative value.
1068  */
1069 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1070 {
1071         int ret;
1072         struct lttcomm_consumer_msg msg;
1073
1074         assert(socket);
1075         assert(socket->fd >= 0);
1076
1077         DBG2("Consumer flush channel key %" PRIu64, key);
1078
1079         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1080         msg.u.flush_channel.key = key;
1081
1082         pthread_mutex_lock(socket->lock);
1083         health_code_update();
1084
1085         ret = consumer_send_msg(socket, &msg);
1086         if (ret < 0) {
1087                 goto end;
1088         }
1089
1090 end:
1091         health_code_update();
1092         pthread_mutex_unlock(socket->lock);
1093         return ret;
1094 }
1095
1096 /*
1097  * Send a close metdata command to consumer using the given channel key.
1098  *
1099  * Return 0 on success else a negative value.
1100  */
1101 int consumer_close_metadata(struct consumer_socket *socket,
1102                 uint64_t metadata_key)
1103 {
1104         int ret;
1105         struct lttcomm_consumer_msg msg;
1106
1107         assert(socket);
1108         assert(socket->fd >= 0);
1109
1110         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1111
1112         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1113         msg.u.close_metadata.key = metadata_key;
1114
1115         pthread_mutex_lock(socket->lock);
1116         health_code_update();
1117
1118         ret = consumer_send_msg(socket, &msg);
1119         if (ret < 0) {
1120                 goto end;
1121         }
1122
1123 end:
1124         health_code_update();
1125         pthread_mutex_unlock(socket->lock);
1126         return ret;
1127 }
1128
1129 /*
1130  * Send a setup metdata command to consumer using the given channel key.
1131  *
1132  * Return 0 on success else a negative value.
1133  */
1134 int consumer_setup_metadata(struct consumer_socket *socket,
1135                 uint64_t metadata_key)
1136 {
1137         int ret;
1138         struct lttcomm_consumer_msg msg;
1139
1140         assert(socket);
1141         assert(socket->fd >= 0);
1142
1143         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1144
1145         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1146         msg.u.setup_metadata.key = metadata_key;
1147
1148         pthread_mutex_lock(socket->lock);
1149         health_code_update();
1150
1151         ret = consumer_send_msg(socket, &msg);
1152         if (ret < 0) {
1153                 goto end;
1154         }
1155
1156 end:
1157         health_code_update();
1158         pthread_mutex_unlock(socket->lock);
1159         return ret;
1160 }
1161
1162 /*
1163  * Send metadata string to consumer. Socket lock MUST be acquired.
1164  *
1165  * Return 0 on success else a negative value.
1166  */
1167 int consumer_push_metadata(struct consumer_socket *socket,
1168                 uint64_t metadata_key, char *metadata_str, size_t len,
1169                 size_t target_offset)
1170 {
1171         int ret;
1172         struct lttcomm_consumer_msg msg;
1173
1174         assert(socket);
1175         assert(socket->fd >= 0);
1176
1177         DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1178
1179         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1180         msg.u.push_metadata.key = metadata_key;
1181         msg.u.push_metadata.target_offset = target_offset;
1182         msg.u.push_metadata.len = len;
1183
1184         health_code_update();
1185         ret = consumer_send_msg(socket, &msg);
1186         if (ret < 0 || len == 0) {
1187                 goto end;
1188         }
1189
1190         DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
1191
1192         ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1193         if (ret < 0) {
1194                 goto end;
1195         }
1196
1197         health_code_update();
1198         ret = consumer_recv_status_reply(socket);
1199         if (ret < 0) {
1200                 goto end;
1201         }
1202
1203 end:
1204         health_code_update();
1205         return ret;
1206 }
1207
1208 /*
1209  * Ask the consumer to snapshot a specific channel using the key.
1210  *
1211  * Return 0 on success or else a negative error.
1212  */
1213 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1214                 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1215                 const char *session_path, int wait, int max_stream_size)
1216 {
1217         int ret;
1218         struct lttcomm_consumer_msg msg;
1219
1220         assert(socket);
1221         assert(socket->fd >= 0);
1222         assert(output);
1223         assert(output->consumer);
1224
1225         DBG("Consumer snapshot channel key %" PRIu64, key);
1226
1227         memset(&msg, 0, sizeof(msg));
1228         msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1229         msg.u.snapshot_channel.key = key;
1230         msg.u.snapshot_channel.max_stream_size = max_stream_size;
1231         msg.u.snapshot_channel.metadata = metadata;
1232
1233         if (output->consumer->type == CONSUMER_DST_NET) {
1234                 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1235                 msg.u.snapshot_channel.use_relayd = 1;
1236                 ret = snprintf(msg.u.snapshot_channel.pathname,
1237                                 sizeof(msg.u.snapshot_channel.pathname),
1238                                 "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir,
1239                                 output->name, output->datetime, output->nb_snapshot,
1240                                 session_path);
1241                 if (ret < 0) {
1242                         ret = -LTTNG_ERR_NOMEM;
1243                         goto error;
1244                 }
1245         } else {
1246                 ret = snprintf(msg.u.snapshot_channel.pathname,
1247                                 sizeof(msg.u.snapshot_channel.pathname),
1248                                 "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path,
1249                                 output->name, output->datetime, output->nb_snapshot,
1250                                 session_path);
1251                 if (ret < 0) {
1252                         ret = -LTTNG_ERR_NOMEM;
1253                         goto error;
1254                 }
1255                 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1256
1257                 /* Create directory. Ignore if exist. */
1258                 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1259                                 S_IRWXU | S_IRWXG, uid, gid);
1260                 if (ret < 0) {
1261                         if (ret != -EEXIST) {
1262                                 ERR("Trace directory creation error");
1263                                 goto error;
1264                         }
1265                 }
1266         }
1267
1268         health_code_update();
1269         ret = consumer_send_msg(socket, &msg);
1270         if (ret < 0) {
1271                 goto error;
1272         }
1273
1274 error:
1275         health_code_update();
1276         return ret;
1277 }