plugins/ctf/lttng-live/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 "../common/notif-iter/notif-iter.h"
  34 #include <assert.h>
  35
  36 #define BT_LOG_TAG "PLUGIN-CTF-LTTNG-LIVE-DS"
  37
  38 #include "data-stream.h"
  39
  40 static
  41 enum bt_ctf_notif_iter_medium_status medop_request_bytes(
  42                 size_t request_sz, uint8_t **buffer_addr,
  43                 size_t *buffer_sz, void *data)
  44 {
  45         enum bt_ctf_notif_iter_medium_status status =
  46                 BT_CTF_NOTIF_ITER_MEDIUM_STATUS_OK;
  47         struct lttng_live_stream_iterator *stream = data;
  48         struct lttng_live_trace *trace = stream->trace;
  49         struct lttng_live_session *session = trace->session;
  50         struct lttng_live_component *lttng_live = session->lttng_live;
  51         uint64_t recv_len = 0;
  52         uint64_t len_left;
  53         uint64_t read_len;
  54         //int i;
  55
  56         len_left = stream->base_offset + stream->len - stream->offset;
  57         if (!len_left) {
  58                 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
  59                 status = BT_CTF_NOTIF_ITER_MEDIUM_STATUS_AGAIN;
  60                 return status;
  61         }
  62         read_len = MIN(request_sz, stream->buflen);
  63         read_len = MIN(read_len, len_left);
  64         status = lttng_live_get_stream_bytes(lttng_live,
  65                         stream, stream->buf, stream->offset,
  66                         read_len, &recv_len);
  67         BT_LOGV_MEM(stream->buf, recv_len, "Live receive payload");
  68         *buffer_addr = stream->buf;
  69         *buffer_sz = recv_len;
  70         stream->offset += recv_len;
  71         return status;
  72 }
  73
  74 static
  75 struct bt_ctf_stream *medop_get_stream(
  76                 struct bt_ctf_stream_class *stream_class, void *data)
  77 {
  78         struct lttng_live_stream_iterator *lttng_live_stream = data;
  79
  80         if (!lttng_live_stream->stream) {
  81                 int64_t id = bt_ctf_stream_class_get_id(stream_class);
  82
  83                 BT_LOGD("Creating stream %s out of stream class %" PRId64,
  84                         lttng_live_stream->name, id);
  85                 lttng_live_stream->stream = bt_ctf_stream_create(stream_class,
  86                                 lttng_live_stream->name);
  87                 if (!lttng_live_stream->stream) {
  88                         BT_LOGE("Cannot create stream %s (stream class %" PRId64 ")",
  89                                         lttng_live_stream->name, id);
  90                 }
  91         }
  92
  93         return lttng_live_stream->stream;
  94 }
  95
  96 static struct bt_ctf_notif_iter_medium_ops medops = {
  97         .request_bytes = medop_request_bytes,
  98         .get_stream = medop_get_stream,
  99 };
 100
 101 BT_HIDDEN
 102 enum bt_ctf_lttng_live_iterator_status lttng_live_lazy_notif_init(
 103                 struct lttng_live_session *session)
 104 {
 105         struct lttng_live_component *lttng_live = session->lttng_live;
 106         struct lttng_live_trace *trace;
 107
 108         if (!session->lazy_stream_notif_init) {
 109                 return BT_CTF_LTTNG_LIVE_ITERATOR_STATUS_OK;
 110         }
 111
 112         bt_list_for_each_entry(trace, &session->traces, node) {
 113                 struct lttng_live_stream_iterator *stream;
 114
 115                 bt_list_for_each_entry(stream, &trace->streams, node) {
 116                         if (stream->notif_iter) {
 117                                 continue;
 118                         }
 119                         stream->notif_iter = bt_ctf_notif_iter_create(trace->trace,
 120                                         lttng_live->max_query_size, medops,
 121                                         stream, stderr);
 122                         if (!stream->notif_iter) {
 123                                 goto error;
 124                         }
 125                 }
 126         }
 127
 128         session->lazy_stream_notif_init = false;
 129
 130         return BT_CTF_LTTNG_LIVE_ITERATOR_STATUS_OK;
 131
 132 error:
 133         return BT_CTF_LTTNG_LIVE_ITERATOR_STATUS_ERROR;
 134 }
 135
 136 BT_HIDDEN
 137 struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
 138                 struct lttng_live_session *session,
 139                 uint64_t ctf_trace_id,
 140                 uint64_t stream_id)
 141 {
 142         struct lttng_live_component *lttng_live = session->lttng_live;
 143         struct lttng_live_stream_iterator *stream =
 144                         g_new0(struct lttng_live_stream_iterator, 1);
 145         struct lttng_live_trace *trace;
 146         int ret;
 147
 148         trace = lttng_live_ref_trace(session, ctf_trace_id);
 149         if (!trace) {
 150                 goto error;
 151         }
 152
 153         stream->p.type = LIVE_STREAM_TYPE_STREAM;
 154         stream->trace = trace;
 155         stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
 156         stream->viewer_stream_id = stream_id;
 157         stream->ctf_stream_class_id = -1ULL;
 158         stream->last_returned_inactivity_timestamp = INT64_MIN;
 159
 160         if (trace->trace) {
 161                 stream->notif_iter = bt_ctf_notif_iter_create(trace->trace,
 162                                 lttng_live->max_query_size, medops,
 163                                 stream, stderr);
 164                 if (!stream->notif_iter) {
 165                         goto error;
 166                 }
 167         }
 168         stream->buf = g_new0(uint8_t, session->lttng_live->max_query_size);
 169         stream->buflen = session->lttng_live->max_query_size;
 170
 171         ret = lttng_live_add_port(lttng_live, stream);
 172         assert(!ret);
 173
 174         bt_list_add(&stream->node, &trace->streams);
 175
 176         goto end;
 177 error:
 178         /* Do not touch "borrowed" file. */
 179         lttng_live_stream_iterator_destroy(stream);
 180         stream = NULL;
 181 end:
 182         return stream;
 183 }
 184
 185 BT_HIDDEN
 186 void lttng_live_stream_iterator_destroy(struct lttng_live_stream_iterator *stream)
 187 {
 188         struct lttng_live_component *lttng_live;
 189         int ret;
 190
 191         if (!stream) {
 192                 return;
 193         }
 194
 195         lttng_live = stream->trace->session->lttng_live;
 196         ret = lttng_live_remove_port(lttng_live, stream->port);
 197         assert(!ret);
 198
 199         if (stream->stream) {
 200                 BT_PUT(stream->stream);
 201         }
 202
 203         if (stream->notif_iter) {
 204                 bt_ctf_notif_iter_destroy(stream->notif_iter);
 205         }
 206         g_free(stream->buf);
 207         BT_PUT(stream->packet_end_notif_queue);
 208         bt_list_del(&stream->node);
 209         /*
 210          * Ensure we poke the trace metadata in the future, which is
 211          * required to release the metadata reference on the trace.
 212          */
 213         stream->trace->new_metadata_needed = true;
 214         lttng_live_unref_trace(stream->trace);
 215         g_free(stream);
 216 }