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
  34 /*
  35  * Receive a reply command status message from the consumer. Consumer socket
  36  * lock MUST be acquired before calling this function.
  37  *
  38  * Return 0 on success, -1 on recv error or a negative lttng error code which
  39  * was possibly returned by the consumer.
  40  */
  41 int consumer_recv_status_reply(struct consumer_socket *sock)
  42 {
  43         int ret;
  44         struct lttcomm_consumer_status_msg reply;
  45
  46         assert(sock);
  47
  48         ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
  49         if (ret <= 0) {
  50                 if (ret == 0) {
  51                         /* Orderly shutdown. Don't return 0 which means success. */
  52                         ret = -1;
  53                 }
  54                 /* The above call will print a PERROR on error. */
  55                 DBG("Fail to receive status reply on sock %d", sock->fd);
  56                 goto end;
  57         }
  58
  59         if (reply.ret_code == LTTNG_OK) {
  60                 /* All good. */
  61                 ret = 0;
  62         } else {
  63                 ret = -reply.ret_code;
  64                 DBG("Consumer ret code %d", ret);
  65         }
  66
  67 end:
  68         return ret;
  69 }
  70
  71 /*
  72  * Once the ASK_CHANNEL command is sent to the consumer, the channel
  73  * information are sent back. This call receives that data and populates key
  74  * and stream_count.
  75  *
  76  * On success return 0 and both key and stream_count are set. On error, a
  77  * negative value is sent back and both parameters are untouched.
  78  */
  79 int consumer_recv_status_channel(struct consumer_socket *sock,
  80                 uint64_t *key, unsigned int *stream_count)
  81 {
  82         int ret;
  83         struct lttcomm_consumer_status_channel reply;
  84
  85         assert(sock);
  86         assert(stream_count);
  87         assert(key);
  88
  89         ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
  90         if (ret <= 0) {
  91                 if (ret == 0) {
  92                         /* Orderly shutdown. Don't return 0 which means success. */
  93                         ret = -1;
  94                 }
  95                 /* The above call will print a PERROR on error. */
  96                 DBG("Fail to receive status reply on sock %d", sock->fd);
  97                 goto end;
  98         }
  99
 100         /* An error is possible so don't touch the key and stream_count. */
 101         if (reply.ret_code != LTTNG_OK) {
 102                 ret = -1;
 103                 goto end;
 104         }
 105
 106         *key = reply.key;
 107         *stream_count = reply.stream_count;
 108
 109 end:
 110         return ret;
 111 }
 112
 113 /*
 114  * Send destroy relayd command to consumer.
 115  *
 116  * On success return positive value. On error, negative value.
 117  */
 118 int consumer_send_destroy_relayd(struct consumer_socket *sock,
 119                 struct consumer_output *consumer)
 120 {
 121         int ret;
 122         struct lttcomm_consumer_msg msg;
 123
 124         assert(consumer);
 125         assert(sock);
 126
 127         DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
 128
 129         /* Bail out if consumer is disabled */
 130         if (!consumer->enabled) {
 131                 ret = LTTNG_OK;
 132                 DBG3("Consumer is disabled");
 133                 goto error;
 134         }
 135
 136         msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
 137         msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
 138
 139         pthread_mutex_lock(sock->lock);
 140         ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
 141         if (ret < 0) {
 142                 /* Indicate that the consumer is probably closing at this point. */
 143                 DBG("send consumer destroy relayd command");
 144                 goto error_send;
 145         }
 146
 147         /* Don't check the return value. The caller will do it. */
 148         ret = consumer_recv_status_reply(sock);
 149
 150         DBG2("Consumer send destroy relayd command done");
 151
 152 error_send:
 153         pthread_mutex_unlock(sock->lock);
 154 error:
 155         return ret;
 156 }
 157
 158 /*
 159  * For each consumer socket in the consumer output object, send a destroy
 160  * relayd command.
 161  */
 162 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
 163 {
 164         struct lttng_ht_iter iter;
 165         struct consumer_socket *socket;
 166
 167         assert(consumer);
 168
 169         /* Destroy any relayd connection */
 170         if (consumer && consumer->type == CONSUMER_DST_NET) {
 171                 rcu_read_lock();
 172                 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 173                                 node.node) {
 174                         int ret;
 175
 176                         /* Send destroy relayd command */
 177                         ret = consumer_send_destroy_relayd(socket, consumer);
 178                         if (ret < 0) {
 179                                 DBG("Unable to send destroy relayd command to consumer");
 180                                 /* Continue since we MUST delete everything at this point. */
 181                         }
 182                 }
 183                 rcu_read_unlock();
 184         }
 185 }
 186
 187 /*
 188  * From a consumer_data structure, allocate and add a consumer socket to the
 189  * consumer output.
 190  *
 191  * Return 0 on success, else negative value on error
 192  */
 193 int consumer_create_socket(struct consumer_data *data,
 194                 struct consumer_output *output)
 195 {
 196         int ret = 0;
 197         struct consumer_socket *socket;
 198
 199         assert(data);
 200
 201         if (output == NULL || data->cmd_sock < 0) {
 202                 /*
 203                  * Not an error. Possible there is simply not spawned consumer or it's
 204                  * disabled for the tracing session asking the socket.
 205                  */
 206                 goto error;
 207         }
 208
 209         rcu_read_lock();
 210         socket = consumer_find_socket(data->cmd_sock, output);
 211         rcu_read_unlock();
 212         if (socket == NULL) {
 213                 socket = consumer_allocate_socket(data->cmd_sock);
 214                 if (socket == NULL) {
 215                         ret = -1;
 216                         goto error;
 217                 }
 218
 219                 socket->registered = 0;
 220                 socket->lock = &data->lock;
 221                 rcu_read_lock();
 222                 consumer_add_socket(socket, output);
 223                 rcu_read_unlock();
 224         }
 225
 226         DBG3("Consumer socket created (fd: %d) and added to output",
 227                         data->cmd_sock);
 228
 229 error:
 230         return ret;
 231 }
 232
 233 /*
 234  * Find a consumer_socket in a consumer_output hashtable. Read side lock must
 235  * be acquired before calling this function and across use of the
 236  * returned consumer_socket.
 237  */
 238 struct consumer_socket *consumer_find_socket(int key,
 239                 struct consumer_output *consumer)
 240 {
 241         struct lttng_ht_iter iter;
 242         struct lttng_ht_node_ulong *node;
 243         struct consumer_socket *socket = NULL;
 244
 245         /* Negative keys are lookup failures */
 246         if (key < 0 || consumer == NULL) {
 247                 return NULL;
 248         }
 249
 250         lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
 251                         &iter);
 252         node = lttng_ht_iter_get_node_ulong(&iter);
 253         if (node != NULL) {
 254                 socket = caa_container_of(node, struct consumer_socket, node);
 255         }
 256
 257         return socket;
 258 }
 259
 260 /*
 261  * Allocate a new consumer_socket and return the pointer.
 262  */
 263 struct consumer_socket *consumer_allocate_socket(int fd)
 264 {
 265         struct consumer_socket *socket = NULL;
 266
 267         socket = zmalloc(sizeof(struct consumer_socket));
 268         if (socket == NULL) {
 269                 PERROR("zmalloc consumer socket");
 270                 goto error;
 271         }
 272
 273         socket->fd = fd;
 274         lttng_ht_node_init_ulong(&socket->node, fd);
 275
 276 error:
 277         return socket;
 278 }
 279
 280 /*
 281  * Add consumer socket to consumer output object. Read side lock must be
 282  * acquired before calling this function.
 283  */
 284 void consumer_add_socket(struct consumer_socket *sock,
 285                 struct consumer_output *consumer)
 286 {
 287         assert(sock);
 288         assert(consumer);
 289
 290         lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
 291 }
 292
 293 /*
 294  * Delte consumer socket to consumer output object. Read side lock must be
 295  * acquired before calling this function.
 296  */
 297 void consumer_del_socket(struct consumer_socket *sock,
 298                 struct consumer_output *consumer)
 299 {
 300         int ret;
 301         struct lttng_ht_iter iter;
 302
 303         assert(sock);
 304         assert(consumer);
 305
 306         iter.iter.node = &sock->node.node;
 307         ret = lttng_ht_del(consumer->socks, &iter);
 308         assert(!ret);
 309 }
 310
 311 /*
 312  * RCU destroy call function.
 313  */
 314 static void destroy_socket_rcu(struct rcu_head *head)
 315 {
 316         struct lttng_ht_node_ulong *node =
 317                 caa_container_of(head, struct lttng_ht_node_ulong, head);
 318         struct consumer_socket *socket =
 319                 caa_container_of(node, struct consumer_socket, node);
 320
 321         free(socket);
 322 }
 323
 324 /*
 325  * Destroy and free socket pointer in a call RCU. Read side lock must be
 326  * acquired before calling this function.
 327  */
 328 void consumer_destroy_socket(struct consumer_socket *sock)
 329 {
 330         assert(sock);
 331
 332         /*
 333          * We DO NOT close the file descriptor here since it is global to the
 334          * session daemon and is closed only if the consumer dies or a custom
 335          * consumer was registered,
 336          */
 337         if (sock->registered) {
 338                 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
 339                 lttcomm_close_unix_sock(sock->fd);
 340         }
 341
 342         call_rcu(&sock->node.head, destroy_socket_rcu);
 343 }
 344
 345 /*
 346  * Allocate and assign data to a consumer_output object.
 347  *
 348  * Return pointer to structure.
 349  */
 350 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
 351 {
 352         struct consumer_output *output = NULL;
 353
 354         output = zmalloc(sizeof(struct consumer_output));
 355         if (output == NULL) {
 356                 PERROR("zmalloc consumer_output");
 357                 goto error;
 358         }
 359
 360         /* By default, consumer output is enabled */
 361         output->enabled = 1;
 362         output->type = type;
 363         output->net_seq_index = (uint64_t) -1ULL;
 364
 365         output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
 366
 367 error:
 368         return output;
 369 }
 370
 371 /*
 372  * Delete the consumer_output object from the list and free the ptr.
 373  */
 374 void consumer_destroy_output(struct consumer_output *obj)
 375 {
 376         if (obj == NULL) {
 377                 return;
 378         }
 379
 380         if (obj->socks) {
 381                 struct lttng_ht_iter iter;
 382                 struct consumer_socket *socket;
 383
 384                 rcu_read_lock();
 385                 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 386                         consumer_del_socket(socket, obj);
 387                         consumer_destroy_socket(socket);
 388                 }
 389                 rcu_read_unlock();
 390
 391                 /* Finally destroy HT */
 392                 lttng_ht_destroy(obj->socks);
 393         }
 394
 395         free(obj);
 396 }
 397
 398 /*
 399  * Copy consumer output and returned the newly allocated copy.
 400  */
 401 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
 402 {
 403         struct lttng_ht *tmp_ht_ptr;
 404         struct lttng_ht_iter iter;
 405         struct consumer_socket *socket, *copy_sock;
 406         struct consumer_output *output;
 407
 408         assert(obj);
 409
 410         output = consumer_create_output(obj->type);
 411         if (output == NULL) {
 412                 goto error;
 413         }
 414         /* Avoid losing the HT reference after the memcpy() */
 415         tmp_ht_ptr = output->socks;
 416
 417         memcpy(output, obj, sizeof(struct consumer_output));
 418
 419         /* Putting back the HT pointer and start copying socket(s). */
 420         output->socks = tmp_ht_ptr;
 421
 422         rcu_read_lock();
 423         cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 424                 /* Create new socket object. */
 425                 copy_sock = consumer_allocate_socket(socket->fd);
 426                 if (copy_sock == NULL) {
 427                         rcu_read_unlock();
 428                         goto malloc_error;
 429                 }
 430
 431                 copy_sock->registered = socket->registered;
 432                 copy_sock->lock = socket->lock;
 433                 consumer_add_socket(copy_sock, output);
 434         }
 435         rcu_read_unlock();
 436
 437 error:
 438         return output;
 439
 440 malloc_error:
 441         consumer_destroy_output(output);
 442         return NULL;
 443 }
 444
 445 /*
 446  * Set network URI to the consumer output object.
 447  *
 448  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 449  * error.
 450  */
 451 int consumer_set_network_uri(struct consumer_output *obj,
 452                 struct lttng_uri *uri)
 453 {
 454         int ret;
 455         char tmp_path[PATH_MAX];
 456         char hostname[HOST_NAME_MAX];
 457         struct lttng_uri *dst_uri = NULL;
 458
 459         /* Code flow error safety net. */
 460         assert(obj);
 461         assert(uri);
 462
 463         switch (uri->stype) {
 464         case LTTNG_STREAM_CONTROL:
 465                 dst_uri = &obj->dst.net.control;
 466                 obj->dst.net.control_isset = 1;
 467                 if (uri->port == 0) {
 468                         /* Assign default port. */
 469                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 470                 }
 471                 DBG3("Consumer control URI set with port %d", uri->port);
 472                 break;
 473         case LTTNG_STREAM_DATA:
 474                 dst_uri = &obj->dst.net.data;
 475                 obj->dst.net.data_isset = 1;
 476                 if (uri->port == 0) {
 477                         /* Assign default port. */
 478                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 479                 }
 480                 DBG3("Consumer data URI set with port %d", uri->port);
 481                 break;
 482         default:
 483                 ERR("Set network uri type unknown %d", uri->stype);
 484                 goto error;
 485         }
 486
 487         ret = uri_compare(dst_uri, uri);
 488         if (!ret) {
 489                 /* Same URI, don't touch it and return success. */
 490                 DBG3("URI network compare are the same");
 491                 goto equal;
 492         }
 493
 494         /* URIs were not equal, replacing it. */
 495         memset(dst_uri, 0, sizeof(struct lttng_uri));
 496         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 497         obj->type = CONSUMER_DST_NET;
 498
 499         /* Handle subdir and add hostname in front. */
 500         if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
 501                 /* Get hostname to append it in the pathname */
 502                 ret = gethostname(hostname, sizeof(hostname));
 503                 if (ret < 0) {
 504                         PERROR("gethostname. Fallback on default localhost");
 505                         strncpy(hostname, "localhost", sizeof(hostname));
 506                 }
 507                 hostname[sizeof(hostname) - 1] = '\0';
 508
 509                 /* Setup consumer subdir if none present in the control URI */
 510                 if (strlen(dst_uri->subdir) == 0) {
 511                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 512                                         hostname, obj->subdir);
 513                 } else {
 514                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 515                                         hostname, dst_uri->subdir);
 516                 }
 517                 if (ret < 0) {
 518                         PERROR("snprintf set consumer uri subdir");
 519                         goto error;
 520                 }
 521
 522                 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
 523                 DBG3("Consumer set network uri subdir path %s", tmp_path);
 524         }
 525
 526         return 0;
 527 equal:
 528         return 1;
 529 error:
 530         return -1;
 531 }
 532
 533 /*
 534  * Send file descriptor to consumer via sock.
 535  */
 536 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 537 {
 538         int ret;
 539
 540         assert(fds);
 541         assert(sock);
 542         assert(nb_fd > 0);
 543
 544         ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
 545         if (ret < 0) {
 546                 /* The above call will print a PERROR on error. */
 547                 DBG("Error when sending consumer fds on sock %d", sock->fd);
 548                 goto error;
 549         }
 550
 551         ret = consumer_recv_status_reply(sock);
 552
 553 error:
 554         return ret;
 555 }
 556
 557 /*
 558  * Consumer send communication message structure to consumer.
 559  */
 560 int consumer_send_msg(struct consumer_socket *sock,
 561                 struct lttcomm_consumer_msg *msg)
 562 {
 563         int ret;
 564
 565         assert(msg);
 566         assert(sock);
 567         assert(sock->fd >= 0);
 568
 569         ret = lttcomm_send_unix_sock(sock->fd, msg,
 570                         sizeof(struct lttcomm_consumer_msg));
 571         if (ret < 0) {
 572                 /* The above call will print a PERROR on error. */
 573                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 574                 goto error;
 575         }
 576
 577         ret = consumer_recv_status_reply(sock);
 578
 579 error:
 580         return ret;
 581 }
 582
 583 /*
 584  * Consumer send channel communication message structure to consumer.
 585  */
 586 int consumer_send_channel(struct consumer_socket *sock,
 587                 struct lttcomm_consumer_msg *msg)
 588 {
 589         int ret;
 590
 591         assert(msg);
 592         assert(sock);
 593         assert(sock->fd >= 0);
 594
 595         ret = lttcomm_send_unix_sock(sock->fd, msg,
 596                         sizeof(struct lttcomm_consumer_msg));
 597         if (ret < 0) {
 598                 /* The above call will print a PERROR on error. */
 599                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 600                 goto error;
 601         }
 602
 603         ret = consumer_recv_status_reply(sock);
 604
 605 error:
 606         return ret;
 607 }
 608
 609 /*
 610  * Populate the given consumer msg structure with the ask_channel command
 611  * information.
 612  */
 613 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 614                 uint64_t subbuf_size,
 615                 uint64_t num_subbuf,
 616                 int overwrite,
 617                 unsigned int switch_timer_interval,
 618                 unsigned int read_timer_interval,
 619                 int output,
 620                 int type,
 621                 uint64_t session_id,
 622                 const char *pathname,
 623                 const char *name,
 624                 uid_t uid,
 625                 gid_t gid,
 626                 uint64_t relayd_id,
 627                 uint64_t key,
 628                 unsigned char *uuid)
 629 {
 630         assert(msg);
 631
 632         /* Zeroed structure */
 633         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 634
 635         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 636         msg->u.ask_channel.subbuf_size = subbuf_size;
 637         msg->u.ask_channel.num_subbuf = num_subbuf ;
 638         msg->u.ask_channel.overwrite = overwrite;
 639         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 640         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 641         msg->u.ask_channel.output = output;
 642         msg->u.ask_channel.type = type;
 643         msg->u.ask_channel.session_id = session_id;
 644         msg->u.ask_channel.uid = uid;
 645         msg->u.ask_channel.gid = gid;
 646         msg->u.ask_channel.relayd_id = relayd_id;
 647         msg->u.ask_channel.key = key;
 648
 649         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 650
 651         strncpy(msg->u.ask_channel.pathname, pathname,
 652                         sizeof(msg->u.ask_channel.pathname));
 653         msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 654
 655         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 656         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 657 }
 658
 659 /*
 660  * Init channel communication message structure.
 661  */
 662 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 663                 enum lttng_consumer_command cmd,
 664                 uint64_t channel_key,
 665                 uint64_t session_id,
 666                 const char *pathname,
 667                 uid_t uid,
 668                 gid_t gid,
 669                 uint64_t relayd_id,
 670                 const char *name,
 671                 unsigned int nb_init_streams,
 672                 enum lttng_event_output output,
 673                 int type)
 674 {
 675         assert(msg);
 676
 677         /* Zeroed structure */
 678         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 679
 680         /* Send channel */
 681         msg->cmd_type = cmd;
 682         msg->u.channel.channel_key = channel_key;
 683         msg->u.channel.session_id = session_id;
 684         msg->u.channel.uid = uid;
 685         msg->u.channel.gid = gid;
 686         msg->u.channel.relayd_id = relayd_id;
 687         msg->u.channel.nb_init_streams = nb_init_streams;
 688         msg->u.channel.output = output;
 689         msg->u.channel.type = type;
 690
 691         strncpy(msg->u.channel.pathname, pathname,
 692                         sizeof(msg->u.channel.pathname));
 693         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 694
 695         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 696         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 697 }
 698
 699 /*
 700  * Init stream communication message structure.
 701  */
 702 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 703                 enum lttng_consumer_command cmd,
 704                 uint64_t channel_key,
 705                 uint64_t stream_key,
 706                 int cpu)
 707 {
 708         assert(msg);
 709
 710         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 711
 712         msg->cmd_type = cmd;
 713         msg->u.stream.channel_key = channel_key;
 714         msg->u.stream.stream_key = stream_key;
 715         msg->u.stream.cpu = cpu;
 716 }
 717
 718 /*
 719  * Send stream communication structure to the consumer.
 720  */
 721 int consumer_send_stream(struct consumer_socket *sock,
 722                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 723                 int *fds, size_t nb_fd)
 724 {
 725         int ret;
 726
 727         assert(msg);
 728         assert(dst);
 729         assert(sock);
 730         assert(fds);
 731
 732         /* Send on socket */
 733         ret = lttcomm_send_unix_sock(sock->fd, msg,
 734                         sizeof(struct lttcomm_consumer_msg));
 735         if (ret < 0) {
 736                 /* The above call will print a PERROR on error. */
 737                 DBG("Error when sending consumer stream on sock %d", sock->fd);
 738                 goto error;
 739         }
 740
 741         ret = consumer_recv_status_reply(sock);
 742         if (ret < 0) {
 743                 goto error;
 744         }
 745
 746         ret = consumer_send_fds(sock, fds, nb_fd);
 747         if (ret < 0) {
 748                 goto error;
 749         }
 750
 751 error:
 752         return ret;
 753 }
 754
 755 /*
 756  * Send relayd socket to consumer associated with a session name.
 757  *
 758  * On success return positive value. On error, negative value.
 759  */
 760 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 761                 struct lttcomm_sock *sock, struct consumer_output *consumer,
 762                 enum lttng_stream_type type, uint64_t session_id)
 763 {
 764         int ret;
 765         struct lttcomm_consumer_msg msg;
 766
 767         /* Code flow error. Safety net. */
 768         assert(sock);
 769         assert(consumer);
 770         assert(consumer_sock);
 771
 772         /* Bail out if consumer is disabled */
 773         if (!consumer->enabled) {
 774                 ret = LTTNG_OK;
 775                 goto error;
 776         }
 777
 778         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 779         /*
 780          * Assign network consumer output index using the temporary consumer since
 781          * this call should only be made from within a set_consumer_uri() function
 782          * call in the session daemon.
 783          */
 784         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 785         msg.u.relayd_sock.type = type;
 786         msg.u.relayd_sock.session_id = session_id;
 787         memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
 788
 789         DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
 790         ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
 791         if (ret < 0) {
 792                 /* The above call will print a PERROR on error. */
 793                 DBG("Error when sending relayd sockets on sock %d", sock->fd);
 794                 goto error;
 795         }
 796
 797         ret = consumer_recv_status_reply(consumer_sock);
 798         if (ret < 0) {
 799                 goto error;
 800         }
 801
 802         DBG3("Sending relayd socket file descriptor to consumer");
 803         ret = consumer_send_fds(consumer_sock, &sock->fd, 1);
 804         if (ret < 0) {
 805                 goto error;
 806         }
 807
 808         DBG2("Consumer relayd socket sent");
 809
 810 error:
 811         return ret;
 812 }
 813
 814 /*
 815  * Set consumer subdirectory using the session name and a generated datetime if
 816  * needed. This is appended to the current subdirectory.
 817  */
 818 int consumer_set_subdir(struct consumer_output *consumer,
 819                 const char *session_name)
 820 {
 821         int ret = 0;
 822         unsigned int have_default_name = 0;
 823         char datetime[16], tmp_path[PATH_MAX];
 824         time_t rawtime;
 825         struct tm *timeinfo;
 826
 827         assert(consumer);
 828         assert(session_name);
 829
 830         memset(tmp_path, 0, sizeof(tmp_path));
 831
 832         /* Flag if we have a default session. */
 833         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 834                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 835                 have_default_name = 1;
 836         } else {
 837                 /* Get date and time for session path */
 838                 time(&rawtime);
 839                 timeinfo = localtime(&rawtime);
 840                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 841         }
 842
 843         if (have_default_name) {
 844                 ret = snprintf(tmp_path, sizeof(tmp_path),
 845                                 "%s/%s", consumer->subdir, session_name);
 846         } else {
 847                 ret = snprintf(tmp_path, sizeof(tmp_path),
 848                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 849         }
 850         if (ret < 0) {
 851                 PERROR("snprintf session name date");
 852                 goto error;
 853         }
 854
 855         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 856         DBG2("Consumer subdir set to %s", consumer->subdir);
 857
 858 error:
 859         return ret;
 860 }
 861
 862 /*
 863  * Ask the consumer if the data is ready to read (NOT pending) for the specific
 864  * session id.
 865  *
 866  * This function has a different behavior with the consumer i.e. that it waits
 867  * for a reply from the consumer if yes or no the data is pending.
 868  */
 869 int consumer_is_data_pending(uint64_t session_id,
 870                 struct consumer_output *consumer)
 871 {
 872         int ret;
 873         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
 874         struct consumer_socket *socket;
 875         struct lttng_ht_iter iter;
 876         struct lttcomm_consumer_msg msg;
 877
 878         assert(consumer);
 879
 880         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
 881
 882         msg.u.data_pending.session_id = session_id;
 883
 884         DBG3("Consumer data pending for id %" PRIu64, session_id);
 885
 886         /* Send command for each consumer */
 887         rcu_read_lock();
 888         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 889                         node.node) {
 890                 /* Code flow error */
 891                 assert(socket->fd >= 0);
 892
 893                 pthread_mutex_lock(socket->lock);
 894
 895                 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
 896                 if (ret < 0) {
 897                         /* The above call will print a PERROR on error. */
 898                         DBG("Error on consumer is data pending on sock %d", socket->fd);
 899                         pthread_mutex_unlock(socket->lock);
 900                         goto error_unlock;
 901                 }
 902
 903                 /*
 904                  * No need for a recv reply status because the answer to the command is
 905                  * the reply status message.
 906                  */
 907
 908                 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
 909                 if (ret <= 0) {
 910                         if (ret == 0) {
 911                                 /* Orderly shutdown. Don't return 0 which means success. */
 912                                 ret = -1;
 913                         }
 914                         /* The above call will print a PERROR on error. */
 915                         DBG("Error on recv consumer is data pending on sock %d", socket->fd);
 916                         pthread_mutex_unlock(socket->lock);
 917                         goto error_unlock;
 918                 }
 919
 920                 pthread_mutex_unlock(socket->lock);
 921
 922                 if (ret_code == 1) {
 923                         break;
 924                 }
 925         }
 926         rcu_read_unlock();
 927
 928         DBG("Consumer data is %s pending for session id %" PRIu64,
 929                         ret_code == 1 ? "" : "NOT", session_id);
 930         return ret_code;
 931
 932 error_unlock:
 933         rcu_read_unlock();
 934         return -1;
 935 }