Fix runas namespace
[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 35#include <lttngerr.h>
0857097f
DG
36
37#include "common/runas.h"
3bd1e081
MD
38
39extern struct lttng_consumer_global_data consumer_data;
40extern int consumer_poll_timeout;
41extern 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 */
48int 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
84end:
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 */
93int 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 */
105int 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 */
124int 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
141int 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]);
d41f73b7 213 assert(msg.u.stream.output == LTTNG_EVENT_MMAP);
7ad0a0cb 214 new_stream = consumer_allocate_stream(msg.u.channel.channel_key,
3bd1e081
MD
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,
6df2e2c9
MD
220 msg.u.stream.path_name,
221 msg.u.stream.uid,
222 msg.u.stream.gid);
3bd1e081
MD
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 {
7ad0a0cb
MD
241 return -ENOSYS;
242#if 0
3bd1e081
MD
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 }
7ad0a0cb 254#endif
3bd1e081
MD
255 break;
256 }
257 default:
258 break;
259 }
260end:
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 }
266end_nosignal:
267 return 0;
268}
269
270int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
271{
13161846 272 struct lttng_ust_object_data obj;
3bd1e081
MD
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 }
b5c5fc29 282 chan->wait_fd_is_copy = 1;
2c1dd183 283 chan->shm_fd = -1;
b5c5fc29 284
3bd1e081
MD
285 return 0;
286}
287
d056b477
MD
288void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
289{
290 ustctl_flush_buffer(stream->chan->handle, stream->buf, 0);
effcf122 291 stream->hangup_flush_done = 1;
d056b477
MD
292}
293
3bd1e081
MD
294void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
295{
296 ustctl_unmap_channel(chan->handle);
297}
298
299int lttng_ustconsumer_allocate_stream(struct lttng_consumer_stream *stream)
300{
13161846 301 struct lttng_ust_object_data obj;
3bd1e081
MD
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;
2c1dd183
MD
314 /* ustctl_open_stream_read has closed the shm fd. */
315 stream->wait_fd_is_copy = 1;
316 stream->shm_fd = -1;
317
3bd1e081
MD
318 stream->mmap_base = ustctl_get_mmap_base(stream->chan->handle, stream->buf);
319 if (!stream->mmap_base) {
320 return -EINVAL;
321 }
ee77a7b0 322
3bd1e081
MD
323 return 0;
324}
325
326void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream)
327{
328 ustctl_close_stream_read(stream->chan->handle, stream->buf);
329}
d41f73b7
MD
330
331
332int 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 */
effcf122
MD
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 }
d41f73b7
MD
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) {
effcf122 362 ret = -ret; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
d41f73b7
MD
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);
effcf122 376 assert(err == 0);
d41f73b7
MD
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);
effcf122 387 assert(err == 0);
d41f73b7
MD
388end:
389 return ret;
390}
391
392int 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) {
e11d277b 398 ret = run_as_open(stream->path_name,
60b6c79c
MD
399 O_WRONLY|O_CREAT|O_TRUNC,
400 S_IRWXU|S_IRWXG|S_IRWXO,
401 stream->uid, stream->gid);
d41f73b7
MD
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
413error:
414 return ret;
415}
This page took 0.059491 seconds and 4 git commands to generate.