Replace assert() -> BT_ASSERT() and some preconditions with BT_ASSERT_PRE()
[babeltrace.git] / plugins / ctf / lttng-live / data-stream.c
CommitLineData
7cdc2bab
MD
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
020bc26f
PP
25#define BT_LOG_TAG "PLUGIN-CTF-LTTNG-LIVE-SRC-DS"
26#include "logging.h"
27
7cdc2bab
MD
28#include <stdio.h>
29#include <stdint.h>
30#include <stdlib.h>
31#include <stdbool.h>
32#include <glib.h>
33#include <inttypes.h>
8f76831a 34#include <babeltrace/compat/mman-internal.h>
9d408fca 35#include <babeltrace/babeltrace.h>
7cdc2bab 36#include "../common/notif-iter/notif-iter.h"
f6ccaed9 37#include <babeltrace/assert-internal.h>
7cdc2bab 38
087bc060 39#include "data-stream.h"
7cdc2bab
MD
40
41static
50842bdc 42enum bt_notif_iter_medium_status medop_request_bytes(
7cdc2bab
MD
43 size_t request_sz, uint8_t **buffer_addr,
44 size_t *buffer_sz, void *data)
45{
50842bdc
PP
46 enum bt_notif_iter_medium_status status =
47 BT_NOTIF_ITER_MEDIUM_STATUS_OK;
7cdc2bab
MD
48 struct lttng_live_stream_iterator *stream = data;
49 struct lttng_live_trace *trace = stream->trace;
50 struct lttng_live_session *session = trace->session;
51 struct lttng_live_component *lttng_live = session->lttng_live;
52 uint64_t recv_len = 0;
53 uint64_t len_left;
54 uint64_t read_len;
55 //int i;
56
57 len_left = stream->base_offset + stream->len - stream->offset;
58 if (!len_left) {
59 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
50842bdc 60 status = BT_NOTIF_ITER_MEDIUM_STATUS_AGAIN;
7cdc2bab
MD
61 return status;
62 }
63 read_len = MIN(request_sz, stream->buflen);
64 read_len = MIN(read_len, len_left);
65 status = lttng_live_get_stream_bytes(lttng_live,
66 stream, stream->buf, stream->offset,
67 read_len, &recv_len);
7cdc2bab
MD
68 *buffer_addr = stream->buf;
69 *buffer_sz = recv_len;
70 stream->offset += recv_len;
71 return status;
72}
73
74static
50842bdc
PP
75struct bt_stream *medop_get_stream(
76 struct bt_stream_class *stream_class,
b92735af 77 uint64_t stream_id, void *data)
7cdc2bab
MD
78{
79 struct lttng_live_stream_iterator *lttng_live_stream = data;
7cdc2bab
MD
80
81 if (!lttng_live_stream->stream) {
b92735af 82 int64_t stream_class_id =
50842bdc 83 bt_stream_class_get_id(stream_class);
b92735af
PP
84
85 BT_LOGD("Creating stream %s (ID: %" PRIu64 ") out of stream class %" PRId64,
86 lttng_live_stream->name, stream_id, stream_class_id);
87
88 if (stream_id == -1ULL) {
89 /* No stream ID */
50842bdc 90 lttng_live_stream->stream = bt_stream_create(
b92735af
PP
91 stream_class, lttng_live_stream->name);
92 } else {
93 lttng_live_stream->stream =
50842bdc 94 bt_stream_create_with_id(stream_class,
b92735af
PP
95 lttng_live_stream->name, stream_id);
96 }
7cdc2bab 97
7cdc2bab 98 if (!lttng_live_stream->stream) {
b92735af
PP
99 BT_LOGE("Cannot create stream %s (stream class %" PRId64 ", stream ID %" PRIu64 ")",
100 lttng_live_stream->name,
101 stream_class_id, stream_id);
7cdc2bab
MD
102 }
103 }
104
105 return lttng_live_stream->stream;
106}
107
50842bdc 108static struct bt_notif_iter_medium_ops medops = {
7cdc2bab
MD
109 .request_bytes = medop_request_bytes,
110 .get_stream = medop_get_stream,
111};
112
113BT_HIDDEN
50842bdc 114enum bt_lttng_live_iterator_status lttng_live_lazy_notif_init(
7cdc2bab
MD
115 struct lttng_live_session *session)
116{
117 struct lttng_live_component *lttng_live = session->lttng_live;
118 struct lttng_live_trace *trace;
119
120 if (!session->lazy_stream_notif_init) {
50842bdc 121 return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab
MD
122 }
123
124 bt_list_for_each_entry(trace, &session->traces, node) {
125 struct lttng_live_stream_iterator *stream;
126
127 bt_list_for_each_entry(stream, &trace->streams, node) {
128 if (stream->notif_iter) {
129 continue;
130 }
50842bdc 131 stream->notif_iter = bt_notif_iter_create(trace->trace,
7cdc2bab 132 lttng_live->max_query_size, medops,
55314f2a 133 stream);
7cdc2bab
MD
134 if (!stream->notif_iter) {
135 goto error;
136 }
137 }
138 }
139
140 session->lazy_stream_notif_init = false;
141
50842bdc 142 return BT_LTTNG_LIVE_ITERATOR_STATUS_OK;
7cdc2bab
MD
143
144error:
50842bdc 145 return BT_LTTNG_LIVE_ITERATOR_STATUS_ERROR;
7cdc2bab
MD
146}
147
148BT_HIDDEN
149struct lttng_live_stream_iterator *lttng_live_stream_iterator_create(
150 struct lttng_live_session *session,
151 uint64_t ctf_trace_id,
152 uint64_t stream_id)
153{
154 struct lttng_live_component *lttng_live = session->lttng_live;
155 struct lttng_live_stream_iterator *stream =
156 g_new0(struct lttng_live_stream_iterator, 1);
157 struct lttng_live_trace *trace;
158 int ret;
159
160 trace = lttng_live_ref_trace(session, ctf_trace_id);
161 if (!trace) {
162 goto error;
163 }
164
165 stream->p.type = LIVE_STREAM_TYPE_STREAM;
166 stream->trace = trace;
167 stream->state = LTTNG_LIVE_STREAM_ACTIVE_NO_DATA;
168 stream->viewer_stream_id = stream_id;
169 stream->ctf_stream_class_id = -1ULL;
170 stream->last_returned_inactivity_timestamp = INT64_MIN;
171
172 if (trace->trace) {
50842bdc 173 stream->notif_iter = bt_notif_iter_create(trace->trace,
7cdc2bab 174 lttng_live->max_query_size, medops,
55314f2a 175 stream);
7cdc2bab
MD
176 if (!stream->notif_iter) {
177 goto error;
178 }
179 }
180 stream->buf = g_new0(uint8_t, session->lttng_live->max_query_size);
181 stream->buflen = session->lttng_live->max_query_size;
182
183 ret = lttng_live_add_port(lttng_live, stream);
f6ccaed9 184 BT_ASSERT(!ret);
7cdc2bab
MD
185
186 bt_list_add(&stream->node, &trace->streams);
187
188 goto end;
189error:
190 /* Do not touch "borrowed" file. */
191 lttng_live_stream_iterator_destroy(stream);
192 stream = NULL;
193end:
194 return stream;
195}
196
197BT_HIDDEN
198void lttng_live_stream_iterator_destroy(struct lttng_live_stream_iterator *stream)
199{
200 struct lttng_live_component *lttng_live;
201 int ret;
202
203 if (!stream) {
204 return;
205 }
206
207 lttng_live = stream->trace->session->lttng_live;
208 ret = lttng_live_remove_port(lttng_live, stream->port);
f6ccaed9 209 BT_ASSERT(!ret);
7cdc2bab
MD
210
211 if (stream->stream) {
212 BT_PUT(stream->stream);
213 }
214
215 if (stream->notif_iter) {
50842bdc 216 bt_notif_iter_destroy(stream->notif_iter);
7cdc2bab
MD
217 }
218 g_free(stream->buf);
219 BT_PUT(stream->packet_end_notif_queue);
220 bt_list_del(&stream->node);
221 /*
222 * Ensure we poke the trace metadata in the future, which is
223 * required to release the metadata reference on the trace.
224 */
225 stream->trace->new_metadata_needed = true;
226 lttng_live_unref_trace(stream->trace);
227 g_free(stream);
228}
This page took 0.041616 seconds and 4 git commands to generate.