src/common/ust-consumer/ust-consumer.c

   1 /*
   2  * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
   3  *                      Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
   4  *
   5  * This program is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU General Public License
   7  * as published by the Free Software Foundation; only version 2
   8  * of the License.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program; if not, write to the Free Software
  17  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  18  */
  19
  20 #define _GNU_SOURCE
  21 #include <assert.h>
  22 #include <fcntl.h>
  23 #include <poll.h>
  24 #include <pthread.h>
  25 #include <stdlib.h>
  26 #include <string.h>
  27 #include <sys/mman.h>
  28 #include <sys/socket.h>
  29 #include <sys/stat.h>
  30 #include <sys/types.h>
  31 #include <unistd.h>
  32 #include <lttng/ust-ctl.h>
  33
  34 #include <common/common.h>
  35 #include <common/sessiond-comm/sessiond-comm.h>
  36 #include <common/compat/fcntl.h>
  37
  38 #include "ust-consumer.h"
  39
  40 extern struct lttng_consumer_global_data consumer_data;
  41 extern int consumer_poll_timeout;
  42 extern volatile int consumer_quit;
  43
  44 /*
  45  * Mmap the ring buffer, read it and write the data to the tracefile.
  46  *
  47  * Returns the number of bytes written
  48  */
  49 int lttng_ustconsumer_on_read_subbuffer_mmap(
  50                 struct lttng_consumer_local_data *ctx,
  51                 struct lttng_consumer_stream *stream, unsigned long len)
  52 {
  53         unsigned long mmap_offset;
  54         long ret = 0;
  55         off_t orig_offset = stream->out_fd_offset;
  56         int outfd = stream->out_fd;
  57
  58         /* get the offset inside the fd to mmap */
  59         ret = ustctl_get_mmap_read_offset(stream->chan->handle,
  60                 stream->buf, &mmap_offset);
  61         if (ret != 0) {
  62                 ret = -errno;
  63                 perror("ustctl_get_mmap_read_offset");
  64                 goto end;
  65         }
  66         while (len > 0) {
  67                 ret = write(outfd, stream->mmap_base + mmap_offset, len);
  68                 if (ret >= len) {
  69                         len = 0;
  70                 } else if (ret < 0) {
  71                         ret = -errno;
  72                         perror("Error in file write");
  73                         goto end;
  74                 }
  75                 /* This won't block, but will start writeout asynchronously */
  76                 lttng_sync_file_range(outfd, stream->out_fd_offset, ret,
  77                                 SYNC_FILE_RANGE_WRITE);
  78                 stream->out_fd_offset += ret;
  79         }
  80
  81         lttng_consumer_sync_trace_file(stream, orig_offset);
  82
  83         goto end;
  84
  85 end:
  86         return ret;
  87 }
  88
  89 /*
  90  * Splice the data from the ring buffer to the tracefile.
  91  *
  92  * Returns the number of bytes spliced.
  93  */
  94 int lttng_ustconsumer_on_read_subbuffer_splice(
  95                 struct lttng_consumer_local_data *ctx,
  96                 struct lttng_consumer_stream *stream, unsigned long len)
  97 {
  98         return -ENOSYS;
  99 }
 100
 101 /*
 102  * Take a snapshot for a specific fd
 103  *
 104  * Returns 0 on success, < 0 on error
 105  */
 106 int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
 107                 struct lttng_consumer_stream *stream)
 108 {
 109         int ret = 0;
 110
 111         ret = ustctl_snapshot(stream->chan->handle, stream->buf);
 112         if (ret != 0) {
 113                 ret = errno;
 114                 perror("Getting sub-buffer snapshot.");
 115         }
 116
 117         return ret;
 118 }
 119
 120 /*
 121  * Get the produced position
 122  *
 123  * Returns 0 on success, < 0 on error
 124  */
 125 int lttng_ustconsumer_get_produced_snapshot(
 126                 struct lttng_consumer_local_data *ctx,
 127                 struct lttng_consumer_stream *stream,
 128                 unsigned long *pos)
 129 {
 130         int ret;
 131
 132         ret = ustctl_snapshot_get_produced(stream->chan->handle,
 133                         stream->buf, pos);
 134         if (ret != 0) {
 135                 ret = errno;
 136                 perror("kernctl_snapshot_get_produced");
 137         }
 138
 139         return ret;
 140 }
 141
 142 int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 143                 int sock, struct pollfd *consumer_sockpoll)
 144 {
 145         ssize_t ret;
 146         struct lttcomm_consumer_msg msg;
 147
 148         ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 149         if (ret != sizeof(msg)) {
 150                 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
 151                 return ret;
 152         }
 153         if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
 154                 return -ENOENT;
 155         }
 156
 157         switch (msg.cmd_type) {
 158         case LTTNG_CONSUMER_ADD_CHANNEL:
 159         {
 160                 struct lttng_consumer_channel *new_channel;
 161                 int fds[1];
 162                 size_t nb_fd = 1;
 163
 164                 /* block */
 165                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 166                         return -EINTR;
 167                 }
 168                 ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
 169                 if (ret != sizeof(fds)) {
 170                         lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
 171                         return ret;
 172                 }
 173
 174                 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
 175
 176                 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 177                                 fds[0], -1,
 178                                 msg.u.channel.mmap_len,
 179                                 msg.u.channel.max_sb_size);
 180                 if (new_channel == NULL) {
 181                         lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
 182                         goto end_nosignal;
 183                 }
 184                 if (ctx->on_recv_channel != NULL) {
 185                         ret = ctx->on_recv_channel(new_channel);
 186                         if (ret == 0) {
 187                                 consumer_add_channel(new_channel);
 188                         } else if (ret < 0) {
 189                                 goto end_nosignal;
 190                         }
 191                 } else {
 192                         consumer_add_channel(new_channel);
 193                 }
 194                 goto end_nosignal;
 195         }
 196         case LTTNG_CONSUMER_ADD_STREAM:
 197         {
 198                 struct lttng_consumer_stream *new_stream;
 199                 int fds[2];
 200                 size_t nb_fd = 2;
 201
 202                 /* block */
 203                 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 204                         return -EINTR;
 205                 }
 206                 ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
 207                 if (ret != sizeof(fds)) {
 208                         lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
 209                         return ret;
 210                 }
 211
 212                 DBG("consumer_add_stream %s (%d,%d)", msg.u.stream.path_name,
 213                         fds[0], fds[1]);
 214                 assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
 215                 new_stream = consumer_allocate_stream(msg.u.channel.channel_key,
 216                                 msg.u.stream.stream_key,
 217                                 fds[0], fds[1],
 218                                 msg.u.stream.state,
 219                                 msg.u.stream.mmap_len,
 220                                 msg.u.stream.output,
 221                                 msg.u.stream.path_name,
 222                                 msg.u.stream.uid,
 223                                 msg.u.stream.gid);
 224                 if (new_stream == NULL) {
 225                         lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
 226                         goto end;
 227                 }
 228                 if (ctx->on_recv_stream != NULL) {
 229                         ret = ctx->on_recv_stream(new_stream);
 230                         if (ret == 0) {
 231                                 consumer_add_stream(new_stream);
 232                         } else if (ret < 0) {
 233                                 goto end;
 234                         }
 235                 } else {
 236                         consumer_add_stream(new_stream);
 237                 }
 238                 break;
 239         }
 240         case LTTNG_CONSUMER_UPDATE_STREAM:
 241         {
 242                 return -ENOSYS;
 243 #if 0
 244                 if (ctx->on_update_stream != NULL) {
 245                         ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
 246                         if (ret == 0) {
 247                                 consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
 248                         } else if (ret < 0) {
 249                                 goto end;
 250                         }
 251                 } else {
 252                         consumer_change_stream_state(msg.u.stream.stream_key,
 253                                 msg.u.stream.state);
 254                 }
 255 #endif
 256                 break;
 257         }
 258         default:
 259                 break;
 260         }
 261 end:
 262         /* signal the poll thread */
 263         ret = write(ctx->consumer_poll_pipe[1], "4", 1);
 264         if (ret < 0) {
 265                 perror("write consumer poll");
 266         }
 267 end_nosignal:
 268         return 0;
 269 }
 270
 271 int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
 272 {
 273         struct lttng_ust_object_data obj;
 274
 275         obj.handle = -1;
 276         obj.shm_fd = chan->shm_fd;
 277         obj.wait_fd = chan->wait_fd;
 278         obj.memory_map_size = chan->mmap_len;
 279         chan->handle = ustctl_map_channel(&obj);
 280         if (!chan->handle) {
 281                 return -ENOMEM;
 282         }
 283         chan->wait_fd_is_copy = 1;
 284         chan->shm_fd = -1;
 285
 286         return 0;
 287 }
 288
 289 void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
 290 {
 291         ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
 292         stream->hangup_flush_done = 1;
 293 }
 294
 295 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
 296 {
 297         ustctl_unmap_channel(chan->handle);
 298 }
 299
 300 int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
 301 {
 302         struct lttng_ust_object_data obj;
 303         int ret;
 304
 305         obj.handle = -1;
 306         obj.shm_fd = stream->shm_fd;
 307         obj.wait_fd = stream->wait_fd;
 308         obj.memory_map_size = stream->mmap_len;
 309         ret = ustctl_add_stream(stream->chan->handle, &obj);
 310         if (ret)
 311                 return ret;
 312         stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
 313         if (!stream->buf)
 314                 return -EBUSY;
 315         /* ustctl_open_stream_read has closed the shm fd. */
 316         stream->wait_fd_is_copy = 1;
 317         stream->shm_fd = -1;
 318
 319         stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
 320         if (!stream->mmap_base) {
 321                 return -EINVAL;
 322         }
 323
 324         return 0;
 325 }
 326
 327 void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
 328 {
 329         ustctl_close_stream_read(stream->chan->handle, stream->buf);
 330 }
 331
 332
 333 int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 334                 struct lttng_consumer_local_data *ctx)
 335 {
 336         unsigned long len;
 337         int err;
 338         long ret = 0;
 339         struct lttng_ust_shm_handle *handle;
 340         struct lttng_ust_lib_ring_buffer *buf;
 341         char dummy;
 342         ssize_t readlen;
 343
 344         DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
 345                 stream->wait_fd, stream->key);
 346
 347         /* We can consume the 1 byte written into the wait_fd by UST */
 348         if (!stream->hangup_flush_done) {
 349                 do {
 350                         readlen = read(stream->wait_fd, &dummy, 1);
 351                 } while (readlen == -1 && errno == -EINTR);
 352                 if (readlen == -1) {
 353                         ret = readlen;
 354                         goto end;
 355                 }
 356         }
 357
 358         buf = stream->buf;
 359         handle = stream->chan->handle;
 360         /* Get the next subbuffer */
 361         err = ustctl_get_next_subbuf(handle, buf);
 362         if (err != 0) {
 363                 ret = -ret;     /* ustctl_get_next_subbuf returns negative, caller expect positive. */
 364                 /*
 365                  * This is a debug message even for single-threaded consumer,
 366                  * because poll() have more relaxed criterions than get subbuf,
 367                  * so get_subbuf may fail for short race windows where poll()
 368                  * would issue wakeups.
 369                  */
 370                 DBG("Reserving sub buffer failed (everything is normal, "
 371                                 "it is due to concurrency)");
 372                 goto end;
 373         }
 374         assert(stream->output == LTTNG_EVENT_MMAP);
 375         /* read the used subbuffer size */
 376         err = ustctl_get_padded_subbuf_size(handle, buf, &len);
 377         assert(err == 0);
 378         /* write the subbuffer to the tracefile */
 379         ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
 380         if (ret < 0) {
 381                 /*
 382                  * display the error but continue processing to try
 383                  * to release the subbuffer
 384                  */
 385                 ERR("Error writing to tracefile");
 386         }
 387         err = ustctl_put_next_subbuf(handle, buf);
 388         assert(err == 0);
 389 end:
 390         return ret;
 391 }
 392
 393 int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
 394 {
 395         int ret;
 396
 397         /* Opening the tracefile in write mode */
 398         if (stream->path_name != NULL) {
 399                 ret = run_as_open(stream->path_name,
 400                                 O_WRONLY|O_CREAT|O_TRUNC,
 401                                 S_IRWXU|S_IRWXG|S_IRWXO,
 402                                 stream->uid, stream->gid);
 403                 if (ret < 0) {
 404                         ERR("Opening %s", stream->path_name);
 405                         perror("open");
 406                         goto error;
 407                 }
 408                 stream->out_fd = ret;
 409         }
 410
 411         /* we return 0 to let the library handle the FD internally */
 412         return 0;
 413
 414 error:
 415         return ret;
 416 }