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  * Should *NOT* be called with RCU read-side lock held.
 412  */
 413 void consumer_destroy_output(struct consumer_output *obj)
 414 {
 415         if (obj == NULL) {
 416                 return;
 417         }
 418
 419         if (obj->socks) {
 420                 struct lttng_ht_iter iter;
 421                 struct consumer_socket *socket;
 422
 423                 rcu_read_lock();
 424                 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 425                         consumer_del_socket(socket, obj);
 426                         consumer_destroy_socket(socket);
 427                 }
 428                 rcu_read_unlock();
 429
 430                 /* Finally destroy HT */
 431                 lttng_ht_destroy(obj->socks);
 432         }
 433
 434         free(obj);
 435 }
 436
 437 /*
 438  * Copy consumer output and returned the newly allocated copy.
 439  *
 440  * Should *NOT* be called with RCU read-side lock held.
 441  */
 442 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
 443 {
 444         struct lttng_ht *tmp_ht_ptr;
 445         struct lttng_ht_iter iter;
 446         struct consumer_socket *socket, *copy_sock;
 447         struct consumer_output *output;
 448
 449         assert(obj);
 450
 451         output = consumer_create_output(obj->type);
 452         if (output == NULL) {
 453                 goto error;
 454         }
 455         /* Avoid losing the HT reference after the memcpy() */
 456         tmp_ht_ptr = output->socks;
 457
 458         memcpy(output, obj, sizeof(struct consumer_output));
 459
 460         /* Putting back the HT pointer and start copying socket(s). */
 461         output->socks = tmp_ht_ptr;
 462
 463         rcu_read_lock();
 464         cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
 465                 /* Create new socket object. */
 466                 copy_sock = consumer_allocate_socket(socket->fd);
 467                 if (copy_sock == NULL) {
 468                         rcu_read_unlock();
 469                         goto malloc_error;
 470                 }
 471
 472                 copy_sock->registered = socket->registered;
 473                 copy_sock->lock = socket->lock;
 474                 consumer_add_socket(copy_sock, output);
 475         }
 476         rcu_read_unlock();
 477
 478 error:
 479         return output;
 480
 481 malloc_error:
 482         consumer_destroy_output(output);
 483         return NULL;
 484 }
 485
 486 /*
 487  * Set network URI to the consumer output object.
 488  *
 489  * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
 490  * error.
 491  */
 492 int consumer_set_network_uri(struct consumer_output *obj,
 493                 struct lttng_uri *uri)
 494 {
 495         int ret;
 496         char tmp_path[PATH_MAX];
 497         char hostname[HOST_NAME_MAX];
 498         struct lttng_uri *dst_uri = NULL;
 499
 500         /* Code flow error safety net. */
 501         assert(obj);
 502         assert(uri);
 503
 504         switch (uri->stype) {
 505         case LTTNG_STREAM_CONTROL:
 506                 dst_uri = &obj->dst.net.control;
 507                 obj->dst.net.control_isset = 1;
 508                 if (uri->port == 0) {
 509                         /* Assign default port. */
 510                         uri->port = DEFAULT_NETWORK_CONTROL_PORT;
 511                 } else {
 512                         if (obj->dst.net.data_isset && uri->port ==
 513                                         obj->dst.net.data.port) {
 514                                 ret = -LTTNG_ERR_INVALID;
 515                                 goto error;
 516                         }
 517                 }
 518                 DBG3("Consumer control URI set with port %d", uri->port);
 519                 break;
 520         case LTTNG_STREAM_DATA:
 521                 dst_uri = &obj->dst.net.data;
 522                 obj->dst.net.data_isset = 1;
 523                 if (uri->port == 0) {
 524                         /* Assign default port. */
 525                         uri->port = DEFAULT_NETWORK_DATA_PORT;
 526                 } else {
 527                         if (obj->dst.net.control_isset && uri->port ==
 528                                         obj->dst.net.control.port) {
 529                                 ret = -LTTNG_ERR_INVALID;
 530                                 goto error;
 531                         }
 532                 }
 533                 DBG3("Consumer data URI set with port %d", uri->port);
 534                 break;
 535         default:
 536                 ERR("Set network uri type unknown %d", uri->stype);
 537                 ret = -LTTNG_ERR_INVALID;
 538                 goto error;
 539         }
 540
 541         ret = uri_compare(dst_uri, uri);
 542         if (!ret) {
 543                 /* Same URI, don't touch it and return success. */
 544                 DBG3("URI network compare are the same");
 545                 goto equal;
 546         }
 547
 548         /* URIs were not equal, replacing it. */
 549         memset(dst_uri, 0, sizeof(struct lttng_uri));
 550         memcpy(dst_uri, uri, sizeof(struct lttng_uri));
 551         obj->type = CONSUMER_DST_NET;
 552
 553         /* Handle subdir and add hostname in front. */
 554         if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
 555                 /* Get hostname to append it in the pathname */
 556                 ret = gethostname(hostname, sizeof(hostname));
 557                 if (ret < 0) {
 558                         PERROR("gethostname. Fallback on default localhost");
 559                         strncpy(hostname, "localhost", sizeof(hostname));
 560                 }
 561                 hostname[sizeof(hostname) - 1] = '\0';
 562
 563                 /* Setup consumer subdir if none present in the control URI */
 564                 if (strlen(dst_uri->subdir) == 0) {
 565                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 566                                         hostname, obj->subdir);
 567                 } else {
 568                         ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 569                                         hostname, dst_uri->subdir);
 570                 }
 571                 if (ret < 0) {
 572                         PERROR("snprintf set consumer uri subdir");
 573                         ret = -LTTNG_ERR_NOMEM;
 574                         goto error;
 575                 }
 576
 577                 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
 578                 DBG3("Consumer set network uri subdir path %s", tmp_path);
 579         }
 580
 581         return 0;
 582 equal:
 583         return 1;
 584 error:
 585         return ret;
 586 }
 587
 588 /*
 589  * Send file descriptor to consumer via sock.
 590  */
 591 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 592 {
 593         int ret;
 594
 595         assert(fds);
 596         assert(sock);
 597         assert(nb_fd > 0);
 598
 599         ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
 600         if (ret < 0) {
 601                 /* The above call will print a PERROR on error. */
 602                 DBG("Error when sending consumer fds on sock %d", sock->fd);
 603                 goto error;
 604         }
 605
 606         ret = consumer_recv_status_reply(sock);
 607
 608 error:
 609         return ret;
 610 }
 611
 612 /*
 613  * Consumer send communication message structure to consumer.
 614  */
 615 int consumer_send_msg(struct consumer_socket *sock,
 616                 struct lttcomm_consumer_msg *msg)
 617 {
 618         int ret;
 619
 620         assert(msg);
 621         assert(sock);
 622         assert(sock->fd >= 0);
 623
 624         ret = lttcomm_send_unix_sock(sock->fd, msg,
 625                         sizeof(struct lttcomm_consumer_msg));
 626         if (ret < 0) {
 627                 /* The above call will print a PERROR on error. */
 628                 DBG("Error when sending consumer channel on sock %d", sock->fd);
 629                 goto error;
 630         }
 631
 632         ret = consumer_recv_status_reply(sock);
 633
 634 error:
 635         return ret;
 636 }
 637
 638 /*
 639  * Consumer send channel communication message structure to consumer.
 640  */
 641 int consumer_send_channel(struct consumer_socket *sock,
 642                 struct lttcomm_consumer_msg *msg)
 643 {
 644         int ret;
 645
 646         assert(msg);
 647         assert(sock);
 648         assert(sock->fd >= 0);
 649
 650         ret = lttcomm_send_unix_sock(sock->fd, msg,
 651                         sizeof(struct lttcomm_consumer_msg));
 652         if (ret < 0) {
 653                 /* The above call will print a PERROR on error. */
 654                 DBG("Error when sending consumer channel 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  * Populate the given consumer msg structure with the ask_channel command
 666  * information.
 667  */
 668 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 669                 uint64_t subbuf_size,
 670                 uint64_t num_subbuf,
 671                 int overwrite,
 672                 unsigned int switch_timer_interval,
 673                 unsigned int read_timer_interval,
 674                 int output,
 675                 int type,
 676                 uint64_t session_id,
 677                 const char *pathname,
 678                 const char *name,
 679                 uid_t uid,
 680                 gid_t gid,
 681                 uint64_t relayd_id,
 682                 uint64_t key,
 683                 unsigned char *uuid,
 684                 uint32_t chan_id,
 685                 uint64_t tracefile_size,
 686                 uint64_t tracefile_count)
 687 {
 688         assert(msg);
 689
 690         /* Zeroed structure */
 691         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 692
 693         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 694         msg->u.ask_channel.subbuf_size = subbuf_size;
 695         msg->u.ask_channel.num_subbuf = num_subbuf ;
 696         msg->u.ask_channel.overwrite = overwrite;
 697         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 698         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 699         msg->u.ask_channel.output = output;
 700         msg->u.ask_channel.type = type;
 701         msg->u.ask_channel.session_id = session_id;
 702         msg->u.ask_channel.uid = uid;
 703         msg->u.ask_channel.gid = gid;
 704         msg->u.ask_channel.relayd_id = relayd_id;
 705         msg->u.ask_channel.key = key;
 706         msg->u.ask_channel.chan_id = chan_id;
 707         msg->u.ask_channel.tracefile_size = tracefile_size;
 708         msg->u.ask_channel.tracefile_count = tracefile_count;
 709
 710         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 711
 712         strncpy(msg->u.ask_channel.pathname, pathname,
 713                         sizeof(msg->u.ask_channel.pathname));
 714         msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 715
 716         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 717         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 718 }
 719
 720 /*
 721  * Init channel communication message structure.
 722  */
 723 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 724                 enum lttng_consumer_command cmd,
 725                 uint64_t channel_key,
 726                 uint64_t session_id,
 727                 const char *pathname,
 728                 uid_t uid,
 729                 gid_t gid,
 730                 uint64_t relayd_id,
 731                 const char *name,
 732                 unsigned int nb_init_streams,
 733                 enum lttng_event_output output,
 734                 int type,
 735                 uint64_t tracefile_size,
 736                 uint64_t tracefile_count)
 737 {
 738         assert(msg);
 739
 740         /* Zeroed structure */
 741         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 742
 743         /* Send channel */
 744         msg->cmd_type = cmd;
 745         msg->u.channel.channel_key = channel_key;
 746         msg->u.channel.session_id = session_id;
 747         msg->u.channel.uid = uid;
 748         msg->u.channel.gid = gid;
 749         msg->u.channel.relayd_id = relayd_id;
 750         msg->u.channel.nb_init_streams = nb_init_streams;
 751         msg->u.channel.output = output;
 752         msg->u.channel.type = type;
 753         msg->u.channel.tracefile_size = tracefile_size;
 754         msg->u.channel.tracefile_count = tracefile_count;
 755
 756         strncpy(msg->u.channel.pathname, pathname,
 757                         sizeof(msg->u.channel.pathname));
 758         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 759
 760         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 761         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 762 }
 763
 764 /*
 765  * Init stream communication message structure.
 766  */
 767 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 768                 enum lttng_consumer_command cmd,
 769                 uint64_t channel_key,
 770                 uint64_t stream_key,
 771                 int cpu)
 772 {
 773         assert(msg);
 774
 775         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 776
 777         msg->cmd_type = cmd;
 778         msg->u.stream.channel_key = channel_key;
 779         msg->u.stream.stream_key = stream_key;
 780         msg->u.stream.cpu = cpu;
 781 }
 782
 783 /*
 784  * Send stream communication structure to the consumer.
 785  */
 786 int consumer_send_stream(struct consumer_socket *sock,
 787                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 788                 int *fds, size_t nb_fd)
 789 {
 790         int ret;
 791
 792         assert(msg);
 793         assert(dst);
 794         assert(sock);
 795         assert(fds);
 796
 797         /* Send on socket */
 798         ret = lttcomm_send_unix_sock(sock->fd, msg,
 799                         sizeof(struct lttcomm_consumer_msg));
 800         if (ret < 0) {
 801                 /* The above call will print a PERROR on error. */
 802                 DBG("Error when sending consumer stream on sock %d", sock->fd);
 803                 goto error;
 804         }
 805
 806         ret = consumer_recv_status_reply(sock);
 807         if (ret < 0) {
 808                 goto error;
 809         }
 810
 811         ret = consumer_send_fds(sock, fds, nb_fd);
 812         if (ret < 0) {
 813                 goto error;
 814         }
 815
 816 error:
 817         return ret;
 818 }
 819
 820 /*
 821  * Send relayd socket to consumer associated with a session name.
 822  *
 823  * On success return positive value. On error, negative value.
 824  */
 825 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 826                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
 827                 enum lttng_stream_type type, uint64_t session_id)
 828 {
 829         int ret;
 830         struct lttcomm_consumer_msg msg;
 831
 832         /* Code flow error. Safety net. */
 833         assert(rsock);
 834         assert(consumer);
 835         assert(consumer_sock);
 836
 837         /* Bail out if consumer is disabled */
 838         if (!consumer->enabled) {
 839                 ret = LTTNG_OK;
 840                 goto error;
 841         }
 842
 843         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 844         /*
 845          * Assign network consumer output index using the temporary consumer since
 846          * this call should only be made from within a set_consumer_uri() function
 847          * call in the session daemon.
 848          */
 849         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 850         msg.u.relayd_sock.type = type;
 851         msg.u.relayd_sock.session_id = session_id;
 852         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 853
 854         DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
 855         ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
 856         if (ret < 0) {
 857                 /* The above call will print a PERROR on error. */
 858                 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
 859                 goto error;
 860         }
 861
 862         ret = consumer_recv_status_reply(consumer_sock);
 863         if (ret < 0) {
 864                 goto error;
 865         }
 866
 867         DBG3("Sending relayd socket file descriptor to consumer");
 868         ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
 869         if (ret < 0) {
 870                 goto error;
 871         }
 872
 873         DBG2("Consumer relayd socket sent");
 874
 875 error:
 876         return ret;
 877 }
 878
 879 /*
 880  * Set consumer subdirectory using the session name and a generated datetime if
 881  * needed. This is appended to the current subdirectory.
 882  */
 883 int consumer_set_subdir(struct consumer_output *consumer,
 884                 const char *session_name)
 885 {
 886         int ret = 0;
 887         unsigned int have_default_name = 0;
 888         char datetime[16], tmp_path[PATH_MAX];
 889         time_t rawtime;
 890         struct tm *timeinfo;
 891
 892         assert(consumer);
 893         assert(session_name);
 894
 895         memset(tmp_path, 0, sizeof(tmp_path));
 896
 897         /* Flag if we have a default session. */
 898         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 899                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 900                 have_default_name = 1;
 901         } else {
 902                 /* Get date and time for session path */
 903                 time(&rawtime);
 904                 timeinfo = localtime(&rawtime);
 905                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 906         }
 907
 908         if (have_default_name) {
 909                 ret = snprintf(tmp_path, sizeof(tmp_path),
 910                                 "%s/%s", consumer->subdir, session_name);
 911         } else {
 912                 ret = snprintf(tmp_path, sizeof(tmp_path),
 913                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 914         }
 915         if (ret < 0) {
 916                 PERROR("snprintf session name date");
 917                 goto error;
 918         }
 919
 920         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 921         DBG2("Consumer subdir set to %s", consumer->subdir);
 922
 923 error:
 924         return ret;
 925 }
 926
 927 /*
 928  * Ask the consumer if the data is ready to read (NOT pending) for the specific
 929  * session id.
 930  *
 931  * This function has a different behavior with the consumer i.e. that it waits
 932  * for a reply from the consumer if yes or no the data is pending.
 933  */
 934 int consumer_is_data_pending(uint64_t session_id,
 935                 struct consumer_output *consumer)
 936 {
 937         int ret;
 938         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
 939         struct consumer_socket *socket;
 940         struct lttng_ht_iter iter;
 941         struct lttcomm_consumer_msg msg;
 942
 943         assert(consumer);
 944
 945         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
 946
 947         msg.u.data_pending.session_id = session_id;
 948
 949         DBG3("Consumer data pending for id %" PRIu64, session_id);
 950
 951         /* Send command for each consumer */
 952         rcu_read_lock();
 953         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 954                         node.node) {
 955                 /* Code flow error */
 956                 assert(socket->fd >= 0);
 957
 958                 pthread_mutex_lock(socket->lock);
 959
 960                 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
 961                 if (ret < 0) {
 962                         /* The above call will print a PERROR on error. */
 963                         DBG("Error on consumer is data pending on sock %d", socket->fd);
 964                         pthread_mutex_unlock(socket->lock);
 965                         goto error_unlock;
 966                 }
 967
 968                 /*
 969                  * No need for a recv reply status because the answer to the command is
 970                  * the reply status message.
 971                  */
 972
 973                 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
 974                 if (ret <= 0) {
 975                         if (ret == 0) {
 976                                 /* Orderly shutdown. Don't return 0 which means success. */
 977                                 ret = -1;
 978                         }
 979                         /* The above call will print a PERROR on error. */
 980                         DBG("Error on recv consumer is data pending on sock %d", socket->fd);
 981                         pthread_mutex_unlock(socket->lock);
 982                         goto error_unlock;
 983                 }
 984
 985                 pthread_mutex_unlock(socket->lock);
 986
 987                 if (ret_code == 1) {
 988                         break;
 989                 }
 990         }
 991         rcu_read_unlock();
 992
 993         DBG("Consumer data is %s pending for session id %" PRIu64,
 994                         ret_code == 1 ? "" : "NOT", session_id);
 995         return ret_code;
 996
 997 error_unlock:
 998         rcu_read_unlock();
 999         return -1;
1000 }
1001
1002 /*
1003  * Send a flush command to consumer using the given channel key.
1004  *
1005  * Return 0 on success else a negative value.
1006  */
1007 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1008 {
1009         int ret;
1010         struct lttcomm_consumer_msg msg;
1011
1012         assert(socket);
1013         assert(socket->fd >= 0);
1014
1015         DBG2("Consumer flush channel key %" PRIu64, key);
1016
1017         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1018         msg.u.flush_channel.key = key;
1019
1020         pthread_mutex_lock(socket->lock);
1021         health_code_update();
1022
1023         ret = consumer_send_msg(socket, &msg);
1024         if (ret < 0) {
1025                 goto end;
1026         }
1027
1028 end:
1029         health_code_update();
1030         pthread_mutex_unlock(socket->lock);
1031         return ret;
1032 }
1033
1034 /*
1035  * Send a close metdata command to consumer using the given channel key.
1036  *
1037  * Return 0 on success else a negative value.
1038  */
1039 int consumer_close_metadata(struct consumer_socket *socket,
1040                 uint64_t metadata_key)
1041 {
1042         int ret;
1043         struct lttcomm_consumer_msg msg;
1044
1045         assert(socket);
1046         assert(socket->fd >= 0);
1047
1048         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1049
1050         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1051         msg.u.close_metadata.key = metadata_key;
1052
1053         pthread_mutex_lock(socket->lock);
1054         health_code_update();
1055
1056         ret = consumer_send_msg(socket, &msg);
1057         if (ret < 0) {
1058                 goto end;
1059         }
1060
1061 end:
1062         health_code_update();
1063         pthread_mutex_unlock(socket->lock);
1064         return ret;
1065 }
1066
1067 /*
1068  * Send a setup metdata command to consumer using the given channel key.
1069  *
1070  * Return 0 on success else a negative value.
1071  */
1072 int consumer_setup_metadata(struct consumer_socket *socket,
1073                 uint64_t metadata_key)
1074 {
1075         int ret;
1076         struct lttcomm_consumer_msg msg;
1077
1078         assert(socket);
1079         assert(socket->fd >= 0);
1080
1081         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1082
1083         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1084         msg.u.setup_metadata.key = metadata_key;
1085
1086         pthread_mutex_lock(socket->lock);
1087         health_code_update();
1088
1089         ret = consumer_send_msg(socket, &msg);
1090         if (ret < 0) {
1091                 goto end;
1092         }
1093
1094 end:
1095         health_code_update();
1096         pthread_mutex_unlock(socket->lock);
1097         return ret;
1098 }
1099
1100 /*
1101  * Send metadata string to consumer. Socket lock MUST be acquired.
1102  *
1103  * Return 0 on success else a negative value.
1104  */
1105 int consumer_push_metadata(struct consumer_socket *socket,
1106                 uint64_t metadata_key, char *metadata_str, size_t len,
1107                 size_t target_offset)
1108 {
1109         int ret;
1110         struct lttcomm_consumer_msg msg;
1111
1112         assert(socket);
1113         assert(socket->fd >= 0);
1114
1115         DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1116
1117         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1118         msg.u.push_metadata.key = metadata_key;
1119         msg.u.push_metadata.target_offset = target_offset;
1120         msg.u.push_metadata.len = len;
1121
1122         health_code_update();
1123         ret = consumer_send_msg(socket, &msg);
1124         if (ret < 0 || len == 0) {
1125                 goto end;
1126         }
1127
1128         DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
1129
1130         ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1131         if (ret < 0) {
1132                 goto end;
1133         }
1134
1135         health_code_update();
1136         ret = consumer_recv_status_reply(socket);
1137         if (ret < 0) {
1138                 goto end;
1139         }
1140
1141 end:
1142         health_code_update();
1143         return ret;
1144 }