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