Add important DEBUG statement
[lttng-tools.git] / liblttng-ustconsumer / lttng-ustconsumer.c
CommitLineData
3bd1e081
MD
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/types.h>
30#include <unistd.h>
31
3bd1e081
MD
32#include <lttng-sessiond-comm.h>
33#include <lttng/lttng-ustconsumer.h>
9df8df5e 34#include <lttng/ust-ctl.h>
3bd1e081
MD
35#include <lttngerr.h>
36
37extern struct lttng_consumer_global_data consumer_data;
38extern int consumer_poll_timeout;
39extern volatile int consumer_quit;
40
41/*
42 * Mmap the ring buffer, read it and write the data to the tracefile.
43 *
44 * Returns the number of bytes written
45 */
46int lttng_ustconsumer_on_read_subbuffer_mmap(
47 struct lttng_consumer_local_data *ctx,
48 struct lttng_consumer_stream *stream, unsigned long len)
49{
50 unsigned long mmap_offset;
51 long ret = 0;
52 off_t orig_offset = stream->out_fd_offset;
53 int outfd = stream->out_fd;
54
55 /* get the offset inside the fd to mmap */
56 ret = ustctl_get_mmap_read_offset(stream->chan->handle,
57 stream->buf, &mmap_offset);
58 if (ret != 0) {
59 ret = -errno;
60 perror("ustctl_get_mmap_read_offset");
61 goto end;
62 }
63 while (len > 0) {
64 ret = write(outfd, stream->mmap_base + mmap_offset, len);
65 if (ret >= len) {
66 len = 0;
67 } else if (ret < 0) {
68 ret = -errno;
69 perror("Error in file write");
70 goto end;
71 }
72 /* This won't block, but will start writeout asynchronously */
73 sync_file_range(outfd, stream->out_fd_offset, ret,
74 SYNC_FILE_RANGE_WRITE);
75 stream->out_fd_offset += ret;
76 }
77
78 lttng_consumer_sync_trace_file(stream, orig_offset);
79
80 goto end;
81
82end:
83 return ret;
84}
85
86/*
87 * Splice the data from the ring buffer to the tracefile.
88 *
89 * Returns the number of bytes spliced.
90 */
91int lttng_ustconsumer_on_read_subbuffer_splice(
92 struct lttng_consumer_local_data *ctx,
93 struct lttng_consumer_stream *stream, unsigned long len)
94{
95 return -ENOSYS;
96}
97
98/*
99 * Take a snapshot for a specific fd
100 *
101 * Returns 0 on success, < 0 on error
102 */
103int lttng_ustconsumer_take_snapshot(struct lttng_consumer_local_data *ctx,
104 struct lttng_consumer_stream *stream)
105{
106 int ret = 0;
107
108 ret = ustctl_snapshot(stream->chan->handle, stream->buf);
109 if (ret != 0) {
110 ret = errno;
111 perror("Getting sub-buffer snapshot.");
112 }
113
114 return ret;
115}
116
117/*
118 * Get the produced position
119 *
120 * Returns 0 on success, < 0 on error
121 */
122int lttng_ustconsumer_get_produced_snapshot(
123 struct lttng_consumer_local_data *ctx,
124 struct lttng_consumer_stream *stream,
125 unsigned long *pos)
126{
127 int ret;
128
129 ret = ustctl_snapshot_get_produced(stream->chan->handle,
130 stream->buf, pos);
131 if (ret != 0) {
132 ret = errno;
133 perror("kernctl_snapshot_get_produced");
134 }
135
136 return ret;
137}
138
139int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
140 int sock, struct pollfd *consumer_sockpoll)
141{
142 ssize_t ret;
143 struct lttcomm_consumer_msg msg;
144
145 ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
146 if (ret != sizeof(msg)) {
147 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
148 return ret;
149 }
150 if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
151 return -ENOENT;
152 }
153
154 switch (msg.cmd_type) {
155 case LTTNG_CONSUMER_ADD_CHANNEL:
156 {
157 struct lttng_consumer_channel *new_channel;
158 int fds[1];
159 size_t nb_fd = 1;
160
161 /* block */
162 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
163 return -EINTR;
164 }
165 ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
166 if (ret != sizeof(fds)) {
167 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
168 return ret;
169 }
170
171 DBG("consumer_add_channel %d", msg.u.channel.channel_key);
172
173 new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
174 fds[0], -1,
175 msg.u.channel.mmap_len,
176 msg.u.channel.max_sb_size);
177 if (new_channel == NULL) {
178 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
179 goto end_nosignal;
180 }
181 if (ctx->on_recv_channel != NULL) {
182 ret = ctx->on_recv_channel(new_channel);
183 if (ret == 0) {
184 consumer_add_channel(new_channel);
185 } else if (ret < 0) {
186 goto end_nosignal;
187 }
188 } else {
189 consumer_add_channel(new_channel);
190 }
191 goto end_nosignal;
192 }
193 case LTTNG_CONSUMER_ADD_STREAM:
194 {
195 struct lttng_consumer_stream *new_stream;
196 int fds[2];
197 size_t nb_fd = 2;
198
199 /* block */
200 if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
201 return -EINTR;
202 }
203 ret = lttcomm_recv_fds_unix_sock(sock, fds, nb_fd);
204 if (ret != sizeof(fds)) {
205 lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
206 return ret;
207 }
208
209 DBG("consumer_add_stream %s (%d,%d)", msg.u.stream.path_name,
210 fds[0], fds[1]);
d41f73b7 211 assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
3bd1e081
MD
212 new_stream = consumer_allocate_stream(msg.u.stream.channel_key,
213 msg.u.stream.stream_key,
214 fds[0], fds[1],
215 msg.u.stream.state,
216 msg.u.stream.mmap_len,
217 msg.u.stream.output,
218 msg.u.stream.path_name);
219 if (new_stream == NULL) {
220 lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
221 goto end;
222 }
223 if (ctx->on_recv_stream != NULL) {
224 ret = ctx->on_recv_stream(new_stream);
225 if (ret == 0) {
226 consumer_add_stream(new_stream);
227 } else if (ret < 0) {
228 goto end;
229 }
230 } else {
231 consumer_add_stream(new_stream);
232 }
233 break;
234 }
235 case LTTNG_CONSUMER_UPDATE_STREAM:
236 {
237 if (ctx->on_update_stream != NULL) {
238 ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
239 if (ret == 0) {
240 consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
241 } else if (ret < 0) {
242 goto end;
243 }
244 } else {
245 consumer_change_stream_state(msg.u.stream.stream_key,
246 msg.u.stream.state);
247 }
248 break;
249 }
250 default:
251 break;
252 }
253end:
254 /* signal the poll thread */
255 ret = write(ctx->consumer_poll_pipe[1], "4", 1);
256 if (ret < 0) {
257 perror("write consumer poll");
258 }
259end_nosignal:
260 return 0;
261}
262
263int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
264{
13161846 265 struct lttng_ust_object_data obj;
3bd1e081
MD
266
267 obj.handle = -1;
268 obj.shm_fd = chan->shm_fd;
269 obj.wait_fd = chan->wait_fd;
270 obj.memory_map_size = chan->mmap_len;
271 chan->handle = ustctl_map_channel(&obj);
272 if (!chan->handle) {
273 return -ENOMEM;
274 }
ee77a7b0 275 /*
b5c5fc29 276 * The channel fds are passed to ustctl, we only keep a copy.
ee77a7b0 277 */
b5c5fc29
MD
278 chan->shm_fd_is_copy = 1;
279 chan->wait_fd_is_copy = 1;
280
3bd1e081
MD
281 return 0;
282}
283
d056b477
MD
284void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
285{
286 ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
287}
288
3bd1e081
MD
289void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
290{
291 ustctl_unmap_channel(chan->handle);
292}
293
294int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
295{
13161846 296 struct lttng_ust_object_data obj;
3bd1e081
MD
297 int ret;
298
299 obj.handle = -1;
300 obj.shm_fd = stream->shm_fd;
301 obj.wait_fd = stream->wait_fd;
302 obj.memory_map_size = stream->mmap_len;
303 ret = ustctl_add_stream(stream->chan->handle, &obj);
304 if (ret)
305 return ret;
306 stream->buf = ustctl_open_stream_read(stream->chan->handle, stream->cpu);
307 if (!stream->buf)
308 return -EBUSY;
309 stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
310 if (!stream->mmap_base) {
311 return -EINVAL;
312 }
ee77a7b0 313 /*
b5c5fc29 314 * The stream fds are passed to ustctl, we only keep a copy.
ee77a7b0 315 */
b5c5fc29
MD
316 stream->shm_fd_is_copy = 1;
317 stream->wait_fd_is_copy = 1;
ee77a7b0 318
3bd1e081
MD
319 return 0;
320}
321
322void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
323{
324 ustctl_close_stream_read(stream->chan->handle, stream->buf);
325}
d41f73b7
MD
326
327
328int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
329 struct lttng_consumer_local_data *ctx)
330{
331 unsigned long len;
332 int err;
333 long ret = 0;
334 struct lttng_ust_shm_handle *handle;
335 struct lttng_ust_lib_ring_buffer *buf;
336 char dummy;
337 ssize_t readlen;
338
339 DBG("In read_subbuffer (wait_fd: %d, stream key: %d)",
340 stream->wait_fd, stream->key);
341
342 /* We can consume the 1 byte written into the wait_fd by UST */
343 do {
344 readlen = read(stream->wait_fd, &dummy, 1);
345 } while (readlen == -1 && errno == -EINTR);
346 if (readlen == -1) {
347 ret = readlen;
348 goto end;
349 }
350
351 buf = stream->buf;
352 handle = stream->chan->handle;
353 /* Get the next subbuffer */
354 err = ustctl_get_next_subbuf(handle, buf);
355 if (err != 0) {
356 ret = errno;
357 /*
358 * This is a debug message even for single-threaded consumer,
359 * because poll() have more relaxed criterions than get subbuf,
360 * so get_subbuf may fail for short race windows where poll()
361 * would issue wakeups.
362 */
363 DBG("Reserving sub buffer failed (everything is normal, "
364 "it is due to concurrency)");
365 goto end;
366 }
367 assert(stream->output == LTTNG_EVENT_MMAP);
368 /* read the used subbuffer size */
369 err = ustctl_get_padded_subbuf_size(handle, buf, &len);
370 if (err != 0) {
371 ret = errno;
372 perror("Getting sub-buffer len failed.");
373 goto end;
374 }
375 /* write the subbuffer to the tracefile */
376 ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
377 if (ret < 0) {
378 /*
379 * display the error but continue processing to try
380 * to release the subbuffer
381 */
382 ERR("Error writing to tracefile");
383 }
384 err = ustctl_put_next_subbuf(handle, buf);
385 if (err != 0) {
386 ret = errno;
387 if (errno == EFAULT) {
388 perror("Error in unreserving sub buffer\n");
389 } else if (errno == EIO) {
390 /* Should never happen with newer LTTng versions */
391 perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
392 }
393 goto end;
394 }
395end:
396 return ret;
397}
398
399int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream)
400{
401 int ret;
402
403 /* Opening the tracefile in write mode */
404 if (stream->path_name != NULL) {
405 ret = open(stream->path_name,
406 O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
407 if (ret < 0) {
408 ERR("Opening %s", stream->path_name);
409 perror("open");
410 goto error;
411 }
412 stream->out_fd = ret;
413 }
414
415 /* we return 0 to let the library handle the FD internally */
416 return 0;
417
418error:
419 return ret;
420}
This page took 0.058247 seconds and 4 git commands to generate.