plugins/ctf/fs-src/data-stream.c

   1 /*
   2  * Copyright 2016 - Philippe Proulx <pproulx@efficios.com>
   3  * Copyright 2016 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4  * Copyright 2010-2011 - EfficiOS Inc. and Linux Foundation
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a copy
   7  * of this software and associated documentation files (the "Software"), to deal
   8  * in the Software without restriction, including without limitation the rights
   9  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10  * copies of the Software, and to permit persons to whom the Software is
  11  * furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  22  * SOFTWARE.
  23  */
  24
  25 #include <stdio.h>
  26 #include <stdint.h>
  27 #include <stdlib.h>
  28 #include <stdbool.h>
  29 #include <glib.h>
  30 #include <inttypes.h>
  31 #include <sys/mman.h>
  32 #include <babeltrace/ctf-ir/stream.h>
  33 #include <babeltrace/graph/notification-iterator.h>
  34 #include <babeltrace/graph/notification-stream.h>
  35 #include <babeltrace/graph/notification-event.h>
  36 #include <babeltrace/graph/notification-packet.h>
  37 #include "file.h"
  38 #include "metadata.h"
  39 #include "../common/notif-iter/notif-iter.h"
  40 #include <assert.h>
  41 #include "data-stream.h"
  42
  43 #define PRINT_ERR_STREAM        ctf_fs->error_fp
  44 #define PRINT_PREFIX            "ctf-fs-data-stream"
  45 #include "print.h"
  46
  47 static inline
  48 size_t remaining_mmap_bytes(struct ctf_fs_stream *stream)
  49 {
  50         return stream->mmap_valid_len - stream->request_offset;
  51 }
  52
  53 static
  54 int stream_munmap(struct ctf_fs_stream *stream)
  55 {
  56         int ret = 0;
  57         struct ctf_fs_component *ctf_fs = stream->file->ctf_fs;
  58
  59         if (munmap(stream->mmap_addr, stream->mmap_len)) {
  60                 PERR("Cannot memory-unmap address %p (size %zu) of file \"%s\" (%p): %s\n",
  61                         stream->mmap_addr, stream->mmap_len,
  62                         stream->file->path->str, stream->file->fp,
  63                         strerror(errno));
  64                 ret = -1;
  65                 goto end;
  66         }
  67 end:
  68         return ret;
  69 }
  70
  71 static
  72 enum bt_ctf_notif_iter_medium_status mmap_next(struct ctf_fs_stream *stream)
  73 {
  74         enum bt_ctf_notif_iter_medium_status ret =
  75                         BT_CTF_NOTIF_ITER_MEDIUM_STATUS_OK;
  76         struct ctf_fs_component *ctf_fs = stream->file->ctf_fs;
  77
  78         /* Unmap old region */
  79         if (stream->mmap_addr) {
  80                 if (stream_munmap(stream)) {
  81                         goto error;
  82                 }
  83
  84                 stream->mmap_offset += stream->mmap_valid_len;
  85                 stream->request_offset = 0;
  86         }
  87
  88         stream->mmap_valid_len = MIN(stream->file->size - stream->mmap_offset,
  89                         stream->mmap_max_len);
  90         if (stream->mmap_valid_len == 0) {
  91                 ret = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_EOF;
  92                 goto end;
  93         }
  94         /* Round up to next page, assuming page size being a power of 2. */
  95         stream->mmap_len = (stream->mmap_valid_len + ctf_fs->page_size - 1)
  96                         & ~(ctf_fs->page_size - 1);
  97         /* Map new region */
  98         assert(stream->mmap_len);
  99         stream->mmap_addr = mmap((void *) 0, stream->mmap_len,
 100                         PROT_READ, MAP_PRIVATE, fileno(stream->file->fp),
 101                         stream->mmap_offset);
 102         if (stream->mmap_addr == MAP_FAILED) {
 103                 PERR("Cannot memory-map address (size %zu) of file \"%s\" (%p) at offset %zu: %s\n",
 104                                 stream->mmap_len, stream->file->path->str,
 105                                 stream->file->fp, stream->mmap_offset,
 106                                 strerror(errno));
 107                 goto error;
 108         }
 109
 110         goto end;
 111 error:
 112         stream_munmap(stream);
 113         ret = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_ERROR;
 114 end:
 115         return ret;
 116 }
 117
 118 static
 119 enum bt_ctf_notif_iter_medium_status medop_request_bytes(
 120                 size_t request_sz, uint8_t **buffer_addr,
 121                 size_t *buffer_sz, void *data)
 122 {
 123         enum bt_ctf_notif_iter_medium_status status =
 124                 BT_CTF_NOTIF_ITER_MEDIUM_STATUS_OK;
 125         struct ctf_fs_stream *stream = data;
 126         struct ctf_fs_component *ctf_fs = stream->file->ctf_fs;
 127
 128         if (request_sz == 0) {
 129                 goto end;
 130         }
 131
 132         /* Check if we have at least one memory-mapped byte left */
 133         if (remaining_mmap_bytes(stream) == 0) {
 134                 /* Are we at the end of the file? */
 135                 if (stream->mmap_offset >= stream->file->size) {
 136                         PDBG("Reached end of file \"%s\" (%p)\n",
 137                                 stream->file->path->str, stream->file->fp);
 138                         status = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_EOF;
 139                         goto end;
 140                 }
 141
 142                 status = mmap_next(stream);
 143                 switch (status) {
 144                 case BT_CTF_NOTIF_ITER_MEDIUM_STATUS_OK:
 145                         break;
 146                 case BT_CTF_NOTIF_ITER_MEDIUM_STATUS_EOF:
 147                         goto end;
 148                 default:
 149                         PERR("Cannot memory-map next region of file \"%s\" (%p)\n",
 150                                         stream->file->path->str,
 151                                         stream->file->fp);
 152                         goto error;
 153                 }
 154         }
 155
 156         *buffer_sz = MIN(remaining_mmap_bytes(stream), request_sz);
 157         *buffer_addr = ((uint8_t *) stream->mmap_addr) + stream->request_offset;
 158         stream->request_offset += *buffer_sz;
 159         goto end;
 160
 161 error:
 162         status = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_ERROR;
 163
 164 end:
 165         return status;
 166 }
 167
 168 static
 169 struct bt_ctf_stream *medop_get_stream(
 170                 struct bt_ctf_stream_class *stream_class, void *data)
 171 {
 172         struct ctf_fs_stream *fs_stream = data;
 173         struct ctf_fs_component *ctf_fs = fs_stream->file->ctf_fs;
 174
 175         if (!fs_stream->stream) {
 176                 int64_t id = bt_ctf_stream_class_get_id(stream_class);
 177
 178                 PDBG("Creating stream out of stream class %" PRId64 "\n", id);
 179                 fs_stream->stream = bt_ctf_stream_create(stream_class,
 180                                 fs_stream->file->path->str);
 181                 if (!fs_stream->stream) {
 182                         PERR("Cannot create stream (stream class %" PRId64 ")\n",
 183                                         id);
 184                 }
 185         }
 186
 187         return fs_stream->stream;
 188 }
 189
 190 static struct bt_ctf_notif_iter_medium_ops medops = {
 191         .request_bytes = medop_request_bytes,
 192         .get_stream = medop_get_stream,
 193 };
 194
 195 static
 196 int build_index_from_idx_file(struct ctf_fs_stream *stream)
 197 {
 198         int ret = 0;
 199         gchar *directory = NULL;
 200         gchar *basename = NULL;
 201         GString *index_basename = NULL;
 202         gchar *index_file_path = NULL;
 203         GMappedFile *mapped_file = NULL;
 204         gsize filesize;
 205         const struct ctf_packet_index_file_hdr *header;
 206         const char *mmap_begin, *file_pos;
 207         struct index_entry *index;
 208         uint64_t total_packets_size = 0;
 209         size_t file_index_entry_size;
 210         size_t file_entry_count;
 211         size_t i;
 212
 213         /* Look for index file in relative path index/name.idx. */
 214         basename = g_path_get_basename(stream->file->path->str);
 215         if (!basename) {
 216                 ret = -1;
 217                 goto end;
 218         }
 219
 220         directory = g_path_get_dirname(stream->file->path->str);
 221         if (!directory) {
 222                 ret = -1;
 223                 goto end;
 224         }
 225
 226         index_basename = g_string_new(basename);
 227         if (!index_basename) {
 228                 ret = -1;
 229                 goto end;
 230         }
 231
 232         g_string_append(index_basename, ".idx");
 233         index_file_path = g_build_filename(directory, "index",
 234                         index_basename->str, NULL);
 235         mapped_file = g_mapped_file_new(index_file_path, FALSE, NULL);
 236         if (!mapped_file) {
 237                 ret = -1;
 238                 goto end;
 239         }
 240         filesize = g_mapped_file_get_length(mapped_file);
 241         if (filesize < sizeof(*header)) {
 242                 printf_error("Invalid LTTng trace index: file size < header size");
 243                 ret = -1;
 244                 goto end;
 245         }
 246
 247         mmap_begin = g_mapped_file_get_contents(mapped_file);
 248         header = (struct ctf_packet_index_file_hdr *) mmap_begin;
 249
 250         file_pos = g_mapped_file_get_contents(mapped_file) + sizeof(*header);
 251         if (be32toh(header->magic) != CTF_INDEX_MAGIC) {
 252                 printf_error("Invalid LTTng trace index: \"magic\" validation failed");
 253                 ret = -1;
 254                 goto end;
 255         }
 256
 257         file_index_entry_size = be32toh(header->packet_index_len);
 258         file_entry_count = (filesize - sizeof(*header)) / file_index_entry_size;
 259         if ((filesize - sizeof(*header)) % (file_entry_count * file_index_entry_size)) {
 260                 printf_error("Invalid index file size; not a multiple of index entry size");
 261                 ret = -1;
 262                 goto end;
 263         }
 264
 265         stream->index.entries = g_array_sized_new(FALSE, TRUE,
 266                         sizeof(struct index_entry), file_entry_count);
 267         if (!stream->index.entries) {
 268                 ret = -1;
 269                 goto end;
 270         }
 271         index = (struct index_entry *) stream->index.entries->data;
 272         for (i = 0; i < file_entry_count; i++) {
 273                 struct ctf_packet_index *file_index =
 274                                 (struct ctf_packet_index *) file_pos;
 275                 uint64_t packet_size = be64toh(file_index->packet_size);
 276
 277                 if (packet_size % CHAR_BIT) {
 278                         ret = -1;
 279                         printf_error("Invalid packet size encountered in index file");
 280                         goto invalid_index;
 281                 }
 282
 283                 /* Convert size in bits to bytes. */
 284                 packet_size /= CHAR_BIT;
 285                 index->packet_size = packet_size;
 286
 287                 index->offset = be64toh(file_index->offset);
 288                 if (i != 0 && index->offset < (index - 1)->offset) {
 289                         printf_error("Invalid, non-monotonic, packet offset encountered in index file");
 290                         ret = -1;
 291                         goto invalid_index;
 292                 }
 293
 294                 index->timestamp_begin = be64toh(file_index->timestamp_begin);
 295                 index->timestamp_end = be64toh(file_index->timestamp_end);
 296                 if (index->timestamp_end < index->timestamp_begin) {
 297                         printf_error("Invalid packet time bounds encountered in index file");
 298                         ret = -1;
 299                         goto invalid_index;
 300                 }
 301
 302                 total_packets_size += packet_size;
 303                 file_pos += file_index_entry_size;
 304                 index++;
 305         }
 306
 307         /* Validate that the index addresses the complete stream. */
 308         if (stream->file->size != total_packets_size) {
 309                 printf_error("Invalid index; indexed size != stream file size");
 310                 ret = -1;
 311                 goto invalid_index;
 312         }
 313 end:
 314         g_free(directory);
 315         g_free(basename);
 316         g_free(index_file_path);
 317         if (index_basename) {
 318                 g_string_free(index_basename, TRUE);
 319         }
 320         if (mapped_file) {
 321                 g_mapped_file_unref(mapped_file);
 322         }
 323         return ret;
 324 invalid_index:
 325         g_array_free(stream->index.entries, TRUE);
 326         goto end;
 327 }
 328
 329 static
 330 int build_index_from_stream(struct ctf_fs_stream *stream)
 331 {
 332         return 0;
 333 }
 334
 335 static
 336 int init_stream_index(struct ctf_fs_stream *stream)
 337 {
 338         int ret;
 339
 340         ret = build_index_from_idx_file(stream);
 341         if (!ret) {
 342                 goto end;
 343         }
 344
 345         ret = build_index_from_stream(stream);
 346 end:
 347         return ret;
 348 }
 349
 350 BT_HIDDEN
 351 struct ctf_fs_stream *ctf_fs_stream_create(
 352                 struct ctf_fs_component *ctf_fs, const char *path)
 353 {
 354         int ret;
 355         struct ctf_fs_stream *stream = g_new0(struct ctf_fs_stream, 1);
 356
 357         if (!stream) {
 358                 goto error;
 359         }
 360
 361         stream->file = ctf_fs_file_create(ctf_fs);
 362         if (!stream->file) {
 363                 goto error;
 364         }
 365
 366         stream->cc_prio_map = bt_get(ctf_fs->cc_prio_map);
 367         g_string_assign(stream->file->path, path);
 368         ret = ctf_fs_file_open(ctf_fs, stream->file, "rb");
 369         if (ret) {
 370                 goto error;
 371         }
 372
 373         stream->notif_iter = bt_ctf_notif_iter_create(ctf_fs->metadata->trace,
 374                         ctf_fs->page_size, medops, stream,
 375                         ctf_fs->error_fp);
 376         if (!stream->notif_iter) {
 377                 goto error;
 378         }
 379
 380         stream->mmap_max_len = ctf_fs->page_size * 2048;
 381         ret = init_stream_index(stream);
 382         if (ret) {
 383                 goto error;
 384         }
 385         goto end;
 386 error:
 387         /* Do not touch "borrowed" file. */
 388         ctf_fs_stream_destroy(stream);
 389         stream = NULL;
 390 end:
 391         return stream;
 392 }
 393
 394 BT_HIDDEN
 395 void ctf_fs_stream_destroy(struct ctf_fs_stream *stream)
 396 {
 397         if (!stream) {
 398                 return;
 399         }
 400
 401         bt_put(stream->cc_prio_map);
 402
 403         if (stream->file) {
 404                 ctf_fs_file_destroy(stream->file);
 405         }
 406
 407         if (stream->stream) {
 408                 BT_PUT(stream->stream);
 409         }
 410
 411         if (stream->notif_iter) {
 412                 bt_ctf_notif_iter_destroy(stream->notif_iter);
 413         }
 414
 415         if (stream->index.entries) {
 416                 g_array_free(stream->index.entries, TRUE);
 417         }
 418
 419         g_free(stream);
 420 }
 421
 422 BT_HIDDEN
 423 struct bt_notification_iterator_next_return ctf_fs_stream_next(
 424                 struct ctf_fs_stream *stream)
 425 {
 426         enum bt_ctf_notif_iter_status notif_iter_status;
 427         struct bt_notification_iterator_next_return ret = {
 428                 .status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR,
 429                 .notification = NULL,
 430         };
 431
 432         if (stream->end_reached) {
 433                 notif_iter_status = BT_CTF_NOTIF_ITER_STATUS_EOF;
 434                 goto translate_status;
 435         }
 436
 437         notif_iter_status = bt_ctf_notif_iter_get_next_notification(
 438                 stream->notif_iter, stream->cc_prio_map, &ret.notification);
 439         if (notif_iter_status != BT_CTF_NOTIF_ITER_STATUS_OK &&
 440                         notif_iter_status != BT_CTF_NOTIF_ITER_STATUS_EOF) {
 441                 goto translate_status;
 442         }
 443
 444         /* Should be handled in bt_ctf_notif_iter_get_next_notification. */
 445         if (notif_iter_status == BT_CTF_NOTIF_ITER_STATUS_EOF) {
 446                 ret.notification = bt_notification_stream_end_create(
 447                         stream->stream);
 448                 if (!ret.notification) {
 449                         notif_iter_status = BT_CTF_NOTIF_ITER_STATUS_ERROR;
 450                         goto translate_status;
 451                 }
 452
 453                 notif_iter_status = BT_CTF_NOTIF_ITER_STATUS_OK;
 454                 stream->end_reached = true;
 455         }
 456
 457 translate_status:
 458         switch (notif_iter_status) {
 459         case BT_CTF_NOTIF_ITER_STATUS_EOF:
 460                 ret.status = BT_NOTIFICATION_ITERATOR_STATUS_END;
 461                 break;
 462         case BT_CTF_NOTIF_ITER_STATUS_OK:
 463                 ret.status = BT_NOTIFICATION_ITERATOR_STATUS_OK;
 464                 break;
 465         case BT_CTF_NOTIF_ITER_STATUS_AGAIN:
 466                 /*
 467                  * Should not make it this far as this is
 468                  * medium-specific; there is nothing for the user to do
 469                  * and it should have been handled upstream.
 470                  */
 471                 assert(false);
 472         case BT_CTF_NOTIF_ITER_STATUS_INVAL:
 473         case BT_CTF_NOTIF_ITER_STATUS_ERROR:
 474         default:
 475                 ret.status = BT_NOTIFICATION_ITERATOR_STATUS_ERROR;
 476                 break;
 477         }
 478
 479         return ret;
 480 }