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