src/bin/lttng-sessiond/ust-consumer.c

   1 /*
   2  * Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
   3  *
   4  * This program is free software; you can redistribute it and/or modify
   5  * it under the terms of the GNU General Public License, version 2 only,
   6  * as published by the Free Software Foundation.
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11  * GNU General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along
  14  * with this program; if not, write to the Free Software Foundation, Inc.,
  15  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  16  */
  17
  18 #define _GNU_SOURCE
  19 #include <errno.h>
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22 #include <string.h>
  23 #include <unistd.h>
  24
  25 #include <common/common.h>
  26 #include <common/consumer.h>
  27 #include <common/defaults.h>
  28
  29 #include "consumer.h"
  30 #include "ust-consumer.h"
  31
  32 /*
  33  * Send a single channel to the consumer using command ADD_CHANNEL.
  34  */
  35 static int send_channel(int sock, struct ust_app_channel *uchan)
  36 {
  37         int ret, fd;
  38         struct lttcomm_consumer_msg msg;
  39
  40         /* Safety net */
  41         assert(uchan);
  42
  43         if (sock < 0) {
  44                 ret = -EINVAL;
  45                 goto error;
  46         }
  47
  48         DBG2("Sending channel %s to UST consumer", uchan->name);
  49
  50         consumer_init_channel_comm_msg(&msg,
  51                         LTTNG_CONSUMER_ADD_CHANNEL,
  52                         uchan->obj->shm_fd,
  53                         uchan->attr.subbuf_size,
  54                         uchan->obj->memory_map_size,
  55                         uchan->name,
  56                         uchan->streams.count);
  57
  58         ret = consumer_send_channel(sock, &msg);
  59         if (ret < 0) {
  60                 goto error;
  61         }
  62
  63         fd = uchan->obj->shm_fd;
  64         ret = consumer_send_fds(sock, &fd, 1);
  65         if (ret < 0) {
  66                 goto error;
  67         }
  68
  69 error:
  70         return ret;
  71 }
  72
  73 /*
  74  * Send a single stream to the consumer using ADD_STREAM command.
  75  */
  76 static int send_channel_stream(int sock, struct ust_app_channel *uchan,
  77                 struct ust_app_session *usess, struct ltt_ust_stream *stream,
  78                 struct consumer_output *consumer, const char *pathname)
  79 {
  80         int ret, fds[2];
  81         struct lttcomm_consumer_msg msg;
  82
  83         /* Safety net */
  84         assert(uchan);
  85         assert(usess);
  86         assert(stream);
  87         assert(consumer);
  88
  89         DBG2("Sending stream %d of channel %s to kernel consumer",
  90                         stream->obj->shm_fd, uchan->name);
  91
  92         consumer_init_stream_comm_msg(&msg,
  93                         LTTNG_CONSUMER_ADD_STREAM,
  94                         uchan->obj->shm_fd,
  95                         stream->obj->shm_fd,
  96                         LTTNG_CONSUMER_ACTIVE_STREAM,
  97                         DEFAULT_UST_CHANNEL_OUTPUT,
  98                         stream->obj->memory_map_size,
  99                         usess->uid,
 100                         usess->gid,
 101                         consumer->net_seq_index,
 102                         0, /* Metadata flag unset */
 103                         stream->name,
 104                         pathname,
 105                         usess->id);
 106
 107         /* Send stream and file descriptor */
 108         fds[0] = stream->obj->shm_fd;
 109         fds[1] = stream->obj->wait_fd;
 110         ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
 111         if (ret < 0) {
 112                 goto error;
 113         }
 114
 115 error:
 116         return ret;
 117 }
 118
 119 /*
 120  * Send all stream fds of UST channel to the consumer.
 121  */
 122 static int send_channel_streams(int sock,
 123                 struct ust_app_channel *uchan, struct ust_app_session *usess,
 124                 struct consumer_output *consumer)
 125 {
 126         int ret;
 127         char tmp_path[PATH_MAX];
 128         const char *pathname;
 129         struct ltt_ust_stream *stream, *tmp;
 130
 131         DBG("Sending streams of channel %s to UST consumer", uchan->name);
 132
 133         ret = send_channel(sock, uchan);
 134         if (ret < 0) {
 135                 goto error;
 136         }
 137
 138         /* Get the right path name destination */
 139         if (consumer->type == CONSUMER_DST_LOCAL) {
 140                 /* Set application path to the destination path */
 141                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
 142                                 consumer->dst.trace_path, consumer->subdir, usess->path);
 143                 if (ret < 0) {
 144                         PERROR("snprintf stream path");
 145                         goto error;
 146                 }
 147                 pathname = tmp_path;
 148                 DBG3("UST local consumer tracefile path: %s", pathname);
 149         } else {
 150                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 151                                 consumer->subdir, usess->path);
 152                 if (ret < 0) {
 153                         PERROR("snprintf stream path");
 154                         goto error;
 155                 }
 156                 pathname = tmp_path;
 157                 DBG3("UST network consumer subdir path: %s", pathname);
 158         }
 159
 160         cds_list_for_each_entry_safe(stream, tmp, &uchan->streams.head, list) {
 161                 if (!stream->obj->shm_fd) {
 162                         continue;
 163                 }
 164
 165                 ret = send_channel_stream(sock, uchan, usess, stream, consumer,
 166                                 pathname);
 167                 if (ret < 0) {
 168                         goto error;
 169                 }
 170         }
 171
 172         DBG("UST consumer channel streams sent");
 173
 174         return 0;
 175
 176 error:
 177         return ret;
 178 }
 179
 180 /*
 181  * Sending metadata to the consumer with command ADD_CHANNEL and ADD_STREAM.
 182  */
 183 static int send_metadata(int sock, struct ust_app_session *usess,
 184                 struct consumer_output *consumer)
 185 {
 186         int ret, fd, fds[2];
 187         char tmp_path[PATH_MAX];
 188         const char *pathname;
 189         struct lttcomm_consumer_msg msg;
 190
 191         /* Safety net */
 192         assert(usess);
 193         assert(consumer);
 194
 195         if (sock < 0) {
 196                 ERR("Consumer socket is negative (%d)", sock);
 197                 return -EINVAL;
 198         }
 199
 200         if (usess->metadata->obj->shm_fd == 0) {
 201                 ERR("Metadata obj shm_fd is 0");
 202                 ret = -1;
 203                 goto error;
 204         }
 205
 206         DBG("UST consumer sending metadata stream fd");
 207
 208         consumer_init_channel_comm_msg(&msg,
 209                         LTTNG_CONSUMER_ADD_CHANNEL,
 210                         usess->metadata->obj->shm_fd,
 211                         usess->metadata->attr.subbuf_size,
 212                         usess->metadata->obj->memory_map_size,
 213                         "metadata",
 214                         1);
 215
 216         ret = consumer_send_channel(sock, &msg);
 217         if (ret < 0) {
 218                 goto error;
 219         }
 220
 221         /* Sending metadata shared memory fd */
 222         fd = usess->metadata->obj->shm_fd;
 223         ret = consumer_send_fds(sock, &fd, 1);
 224         if (ret < 0) {
 225                 goto error;
 226         }
 227
 228         /* Get correct path name destination */
 229         if (consumer->type == CONSUMER_DST_LOCAL) {
 230                 /* Set application path to the destination path */
 231                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s/%s",
 232                                 consumer->dst.trace_path, consumer->subdir, usess->path);
 233                 if (ret < 0) {
 234                         PERROR("snprintf stream path");
 235                         goto error;
 236                 }
 237                 pathname = tmp_path;
 238
 239                 /* Create directory */
 240                 ret = run_as_mkdir_recursive(pathname, S_IRWXU | S_IRWXG,
 241                                 usess->uid, usess->gid);
 242                 if (ret < 0) {
 243                         if (ret != -EEXIST) {
 244                                 ERR("Trace directory creation error");
 245                                 goto error;
 246                         }
 247                 }
 248         } else {
 249                 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
 250                                 consumer->subdir, usess->path);
 251                 if (ret < 0) {
 252                         PERROR("snprintf metadata path");
 253                         goto error;
 254                 }
 255                 pathname = tmp_path;
 256         }
 257
 258         consumer_init_stream_comm_msg(&msg,
 259                         LTTNG_CONSUMER_ADD_STREAM,
 260                         usess->metadata->obj->shm_fd,
 261                         usess->metadata->stream_obj->shm_fd,
 262                         LTTNG_CONSUMER_ACTIVE_STREAM,
 263                         DEFAULT_UST_CHANNEL_OUTPUT,
 264                         usess->metadata->stream_obj->memory_map_size,
 265                         usess->uid,
 266                         usess->gid,
 267                         consumer->net_seq_index,
 268                         1, /* Flag metadata set */
 269                         "metadata",
 270                         pathname,
 271                         usess->id);
 272
 273         /* Send stream and file descriptor */
 274         fds[0] = usess->metadata->stream_obj->shm_fd;
 275         fds[1] = usess->metadata->stream_obj->wait_fd;
 276         ret = consumer_send_stream(sock, consumer, &msg, fds, 2);
 277         if (ret < 0) {
 278                 goto error;
 279         }
 280
 281 error:
 282         return ret;
 283 }
 284
 285 /*
 286  * Send all stream fds of the UST session to the consumer.
 287  */
 288 int ust_consumer_send_session(struct ust_app_session *usess,
 289                 struct consumer_output *consumer, struct consumer_socket *sock)
 290 {
 291         int ret = 0;
 292         struct lttng_ht_iter iter;
 293         struct ust_app_channel *ua_chan;
 294
 295         assert(usess);
 296
 297         if (consumer == NULL || sock == NULL) {
 298                 /* There is no consumer so just ignoring the command. */
 299                 DBG("UST consumer does not exist. Not sending streams");
 300                 return 0;
 301         }
 302
 303         DBG("Sending metadata stream fd to consumer on %d", sock->fd);
 304
 305         pthread_mutex_lock(sock->lock);
 306
 307         /* Sending metadata information to the consumer */
 308         ret = send_metadata(sock->fd, usess, consumer);
 309         if (ret < 0) {
 310                 goto error;
 311         }
 312
 313         /* Send each channel fd streams of session */
 314         rcu_read_lock();
 315         cds_lfht_for_each_entry(usess->channels->ht, &iter.iter, ua_chan,
 316                         node.node) {
 317                 /*
 318                  * Indicate that the channel was not created on the tracer side so skip
 319                  * sending unexisting streams.
 320                  */
 321                 if (ua_chan->obj == NULL) {
 322                         continue;
 323                 }
 324
 325                 ret = send_channel_streams(sock->fd, ua_chan, usess, consumer);
 326                 if (ret < 0) {
 327                         rcu_read_unlock();
 328                         goto error;
 329                 }
 330         }
 331         rcu_read_unlock();
 332
 333         DBG("consumer fds (metadata and channel streams) sent");
 334
 335         /* All good! */
 336         ret = 0;
 337
 338 error:
 339         pthread_mutex_unlock(sock->lock);
 340         return ret;
 341 }