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                 uint64_t session_id_per_pid)
 689 {
 690         assert(msg);
 691
 692         /* Zeroed structure */
 693         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 694
 695         msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
 696         msg->u.ask_channel.subbuf_size = subbuf_size;
 697         msg->u.ask_channel.num_subbuf = num_subbuf ;
 698         msg->u.ask_channel.overwrite = overwrite;
 699         msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
 700         msg->u.ask_channel.read_timer_interval = read_timer_interval;
 701         msg->u.ask_channel.output = output;
 702         msg->u.ask_channel.type = type;
 703         msg->u.ask_channel.session_id = session_id;
 704         msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
 705         msg->u.ask_channel.uid = uid;
 706         msg->u.ask_channel.gid = gid;
 707         msg->u.ask_channel.relayd_id = relayd_id;
 708         msg->u.ask_channel.key = key;
 709         msg->u.ask_channel.chan_id = chan_id;
 710         msg->u.ask_channel.tracefile_size = tracefile_size;
 711         msg->u.ask_channel.tracefile_count = tracefile_count;
 712
 713         memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
 714
 715         strncpy(msg->u.ask_channel.pathname, pathname,
 716                         sizeof(msg->u.ask_channel.pathname));
 717         msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
 718
 719         strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
 720         msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
 721 }
 722
 723 /*
 724  * Init channel communication message structure.
 725  */
 726 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
 727                 enum lttng_consumer_command cmd,
 728                 uint64_t channel_key,
 729                 uint64_t session_id,
 730                 const char *pathname,
 731                 uid_t uid,
 732                 gid_t gid,
 733                 uint64_t relayd_id,
 734                 const char *name,
 735                 unsigned int nb_init_streams,
 736                 enum lttng_event_output output,
 737                 int type,
 738                 uint64_t tracefile_size,
 739                 uint64_t tracefile_count)
 740 {
 741         assert(msg);
 742
 743         /* Zeroed structure */
 744         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 745
 746         /* Send channel */
 747         msg->cmd_type = cmd;
 748         msg->u.channel.channel_key = channel_key;
 749         msg->u.channel.session_id = session_id;
 750         msg->u.channel.uid = uid;
 751         msg->u.channel.gid = gid;
 752         msg->u.channel.relayd_id = relayd_id;
 753         msg->u.channel.nb_init_streams = nb_init_streams;
 754         msg->u.channel.output = output;
 755         msg->u.channel.type = type;
 756         msg->u.channel.tracefile_size = tracefile_size;
 757         msg->u.channel.tracefile_count = tracefile_count;
 758
 759         strncpy(msg->u.channel.pathname, pathname,
 760                         sizeof(msg->u.channel.pathname));
 761         msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
 762
 763         strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
 764         msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
 765 }
 766
 767 /*
 768  * Init stream communication message structure.
 769  */
 770 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
 771                 enum lttng_consumer_command cmd,
 772                 uint64_t channel_key,
 773                 uint64_t stream_key,
 774                 int cpu)
 775 {
 776         assert(msg);
 777
 778         memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
 779
 780         msg->cmd_type = cmd;
 781         msg->u.stream.channel_key = channel_key;
 782         msg->u.stream.stream_key = stream_key;
 783         msg->u.stream.cpu = cpu;
 784 }
 785
 786 /*
 787  * Send stream communication structure to the consumer.
 788  */
 789 int consumer_send_stream(struct consumer_socket *sock,
 790                 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
 791                 int *fds, size_t nb_fd)
 792 {
 793         int ret;
 794
 795         assert(msg);
 796         assert(dst);
 797         assert(sock);
 798         assert(fds);
 799
 800         /* Send on socket */
 801         ret = lttcomm_send_unix_sock(sock->fd, msg,
 802                         sizeof(struct lttcomm_consumer_msg));
 803         if (ret < 0) {
 804                 /* The above call will print a PERROR on error. */
 805                 DBG("Error when sending consumer stream on sock %d", sock->fd);
 806                 goto error;
 807         }
 808
 809         ret = consumer_recv_status_reply(sock);
 810         if (ret < 0) {
 811                 goto error;
 812         }
 813
 814         ret = consumer_send_fds(sock, fds, nb_fd);
 815         if (ret < 0) {
 816                 goto error;
 817         }
 818
 819 error:
 820         return ret;
 821 }
 822
 823 /*
 824  * Send relayd socket to consumer associated with a session name.
 825  *
 826  * On success return positive value. On error, negative value.
 827  */
 828 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 829                 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
 830                 enum lttng_stream_type type, uint64_t session_id)
 831 {
 832         int ret;
 833         struct lttcomm_consumer_msg msg;
 834
 835         /* Code flow error. Safety net. */
 836         assert(rsock);
 837         assert(consumer);
 838         assert(consumer_sock);
 839
 840         /* Bail out if consumer is disabled */
 841         if (!consumer->enabled) {
 842                 ret = LTTNG_OK;
 843                 goto error;
 844         }
 845
 846         msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
 847         /*
 848          * Assign network consumer output index using the temporary consumer since
 849          * this call should only be made from within a set_consumer_uri() function
 850          * call in the session daemon.
 851          */
 852         msg.u.relayd_sock.net_index = consumer->net_seq_index;
 853         msg.u.relayd_sock.type = type;
 854         msg.u.relayd_sock.session_id = session_id;
 855         memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
 856
 857         DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
 858         ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
 859         if (ret < 0) {
 860                 /* The above call will print a PERROR on error. */
 861                 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
 862                 goto error;
 863         }
 864
 865         ret = consumer_recv_status_reply(consumer_sock);
 866         if (ret < 0) {
 867                 goto error;
 868         }
 869
 870         DBG3("Sending relayd socket file descriptor to consumer");
 871         ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
 872         if (ret < 0) {
 873                 goto error;
 874         }
 875
 876         DBG2("Consumer relayd socket sent");
 877
 878 error:
 879         return ret;
 880 }
 881
 882 /*
 883  * Set consumer subdirectory using the session name and a generated datetime if
 884  * needed. This is appended to the current subdirectory.
 885  */
 886 int consumer_set_subdir(struct consumer_output *consumer,
 887                 const char *session_name)
 888 {
 889         int ret = 0;
 890         unsigned int have_default_name = 0;
 891         char datetime[16], tmp_path[PATH_MAX];
 892         time_t rawtime;
 893         struct tm *timeinfo;
 894
 895         assert(consumer);
 896         assert(session_name);
 897
 898         memset(tmp_path, 0, sizeof(tmp_path));
 899
 900         /* Flag if we have a default session. */
 901         if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
 902                                 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
 903                 have_default_name = 1;
 904         } else {
 905                 /* Get date and time for session path */
 906                 time(&rawtime);
 907                 timeinfo = localtime(&rawtime);
 908                 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
 909         }
 910
 911         if (have_default_name) {
 912                 ret = snprintf(tmp_path, sizeof(tmp_path),
 913                                 "%s/%s", consumer->subdir, session_name);
 914         } else {
 915                 ret = snprintf(tmp_path, sizeof(tmp_path),
 916                                 "%s/%s-%s/", consumer->subdir, session_name, datetime);
 917         }
 918         if (ret < 0) {
 919                 PERROR("snprintf session name date");
 920                 goto error;
 921         }
 922
 923         strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
 924         DBG2("Consumer subdir set to %s", consumer->subdir);
 925
 926 error:
 927         return ret;
 928 }
 929
 930 /*
 931  * Ask the consumer if the data is ready to read (NOT pending) for the specific
 932  * session id.
 933  *
 934  * This function has a different behavior with the consumer i.e. that it waits
 935  * for a reply from the consumer if yes or no the data is pending.
 936  */
 937 int consumer_is_data_pending(uint64_t session_id,
 938                 struct consumer_output *consumer)
 939 {
 940         int ret;
 941         int32_t ret_code = 0;  /* Default is that the data is NOT pending */
 942         struct consumer_socket *socket;
 943         struct lttng_ht_iter iter;
 944         struct lttcomm_consumer_msg msg;
 945
 946         assert(consumer);
 947
 948         msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
 949
 950         msg.u.data_pending.session_id = session_id;
 951
 952         DBG3("Consumer data pending for id %" PRIu64, session_id);
 953
 954         /* Send command for each consumer */
 955         rcu_read_lock();
 956         cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 957                         node.node) {
 958                 /* Code flow error */
 959                 assert(socket->fd >= 0);
 960
 961                 pthread_mutex_lock(socket->lock);
 962
 963                 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
 964                 if (ret < 0) {
 965                         /* The above call will print a PERROR on error. */
 966                         DBG("Error on consumer is data pending on sock %d", socket->fd);
 967                         pthread_mutex_unlock(socket->lock);
 968                         goto error_unlock;
 969                 }
 970
 971                 /*
 972                  * No need for a recv reply status because the answer to the command is
 973                  * the reply status message.
 974                  */
 975
 976                 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
 977                 if (ret <= 0) {
 978                         if (ret == 0) {
 979                                 /* Orderly shutdown. Don't return 0 which means success. */
 980                                 ret = -1;
 981                         }
 982                         /* The above call will print a PERROR on error. */
 983                         DBG("Error on recv consumer is data pending on sock %d", socket->fd);
 984                         pthread_mutex_unlock(socket->lock);
 985                         goto error_unlock;
 986                 }
 987
 988                 pthread_mutex_unlock(socket->lock);
 989
 990                 if (ret_code == 1) {
 991                         break;
 992                 }
 993         }
 994         rcu_read_unlock();
 995
 996         DBG("Consumer data is %s pending for session id %" PRIu64,
 997                         ret_code == 1 ? "" : "NOT", session_id);
 998         return ret_code;
 999
1000 error_unlock:
1001         rcu_read_unlock();
1002         return -1;
1003 }
1004
1005 /*
1006  * Send a flush command to consumer using the given channel key.
1007  *
1008  * Return 0 on success else a negative value.
1009  */
1010 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1011 {
1012         int ret;
1013         struct lttcomm_consumer_msg msg;
1014
1015         assert(socket);
1016         assert(socket->fd >= 0);
1017
1018         DBG2("Consumer flush channel key %" PRIu64, key);
1019
1020         msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1021         msg.u.flush_channel.key = key;
1022
1023         pthread_mutex_lock(socket->lock);
1024         health_code_update();
1025
1026         ret = consumer_send_msg(socket, &msg);
1027         if (ret < 0) {
1028                 goto end;
1029         }
1030
1031 end:
1032         health_code_update();
1033         pthread_mutex_unlock(socket->lock);
1034         return ret;
1035 }
1036
1037 /*
1038  * Send a close metdata command to consumer using the given channel key.
1039  *
1040  * Return 0 on success else a negative value.
1041  */
1042 int consumer_close_metadata(struct consumer_socket *socket,
1043                 uint64_t metadata_key)
1044 {
1045         int ret;
1046         struct lttcomm_consumer_msg msg;
1047
1048         assert(socket);
1049         assert(socket->fd >= 0);
1050
1051         DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1052
1053         msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1054         msg.u.close_metadata.key = metadata_key;
1055
1056         pthread_mutex_lock(socket->lock);
1057         health_code_update();
1058
1059         ret = consumer_send_msg(socket, &msg);
1060         if (ret < 0) {
1061                 goto end;
1062         }
1063
1064 end:
1065         health_code_update();
1066         pthread_mutex_unlock(socket->lock);
1067         return ret;
1068 }
1069
1070 /*
1071  * Send a setup metdata command to consumer using the given channel key.
1072  *
1073  * Return 0 on success else a negative value.
1074  */
1075 int consumer_setup_metadata(struct consumer_socket *socket,
1076                 uint64_t metadata_key)
1077 {
1078         int ret;
1079         struct lttcomm_consumer_msg msg;
1080
1081         assert(socket);
1082         assert(socket->fd >= 0);
1083
1084         DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1085
1086         msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1087         msg.u.setup_metadata.key = metadata_key;
1088
1089         pthread_mutex_lock(socket->lock);
1090         health_code_update();
1091
1092         ret = consumer_send_msg(socket, &msg);
1093         if (ret < 0) {
1094                 goto end;
1095         }
1096
1097 end:
1098         health_code_update();
1099         pthread_mutex_unlock(socket->lock);
1100         return ret;
1101 }
1102
1103 /*
1104  * Send metadata string to consumer. Socket lock MUST be acquired.
1105  *
1106  * Return 0 on success else a negative value.
1107  */
1108 int consumer_push_metadata(struct consumer_socket *socket,
1109                 uint64_t metadata_key, char *metadata_str, size_t len,
1110                 size_t target_offset)
1111 {
1112         int ret;
1113         struct lttcomm_consumer_msg msg;
1114
1115         assert(socket);
1116         assert(socket->fd >= 0);
1117
1118         DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1119
1120         msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1121         msg.u.push_metadata.key = metadata_key;
1122         msg.u.push_metadata.target_offset = target_offset;
1123         msg.u.push_metadata.len = len;
1124
1125         health_code_update();
1126         ret = consumer_send_msg(socket, &msg);
1127         if (ret < 0 || len == 0) {
1128                 goto end;
1129         }
1130
1131         DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
1132
1133         ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1134         if (ret < 0) {
1135                 goto end;
1136         }
1137
1138         health_code_update();
1139         ret = consumer_recv_status_reply(socket);
1140         if (ret < 0) {
1141                 goto end;
1142         }
1143
1144 end:
1145         health_code_update();
1146         return ret;
1147 }