src/bin/lttng-sessiond/consumer.c

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