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