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