Initial import of the new binary lttng-relayd
[lttng-tools.git] / src / common / consumer.h
CommitLineData
3bd1e081
MD
1/*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Copyright (C) 2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 *
d14d33bf
AM
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
3bd1e081 8 *
d14d33bf
AM
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
3bd1e081 13 *
d14d33bf
AM
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
3bd1e081
MD
17 */
18
19#ifndef _LTTNG_CONSUMER_H
20#define _LTTNG_CONSUMER_H
21
22#include <limits.h>
23#include <poll.h>
6df2e2c9 24#include <unistd.h>
e4421fec 25
3bd1e081 26#include <lttng/lttng.h>
10a8a223 27
b9182dd9
DG
28#include <common/hashtable/hashtable.h>
29#include <common/compat/fcntl.h>
3bd1e081
MD
30
31/*
32 * When the receiving thread dies, we need to have a way to make the polling
33 * thread exit eventually. If all FDs hang up (normal case when the
32258573
DG
34 * lttng-sessiond stops), we can exit cleanly, but if there is a problem and
35 * for whatever reason some FDs remain open, the consumer should still exit
3bd1e081
MD
36 * eventually.
37 *
38 * If the timeout is reached, it means that during this period no events
39 * occurred on the FDs so we need to force an exit. This case should not happen
40 * but it is a safety to ensure we won't block the consumer indefinitely.
41 *
42 * The value of 2 seconds is an arbitrary choice.
43 */
44#define LTTNG_CONSUMER_POLL_TIMEOUT 2000
45
46/* Commands for consumer */
47enum lttng_consumer_command {
48 LTTNG_CONSUMER_ADD_CHANNEL,
49 LTTNG_CONSUMER_ADD_STREAM,
50 /* pause, delete, active depending on fd state */
51 LTTNG_CONSUMER_UPDATE_STREAM,
52 /* inform the consumer to quit when all fd has hang up */
53 LTTNG_CONSUMER_STOP,
54};
55
56/* State of each fd in consumer */
57enum lttng_consumer_stream_state {
58 LTTNG_CONSUMER_ACTIVE_STREAM,
59 LTTNG_CONSUMER_PAUSE_STREAM,
60 LTTNG_CONSUMER_DELETE_STREAM,
61};
62
3bd1e081
MD
63enum lttng_consumer_type {
64 LTTNG_CONSUMER_UNKNOWN = 0,
65 LTTNG_CONSUMER_KERNEL,
7753dea8
MD
66 LTTNG_CONSUMER64_UST,
67 LTTNG_CONSUMER32_UST,
3bd1e081
MD
68};
69
70struct lttng_consumer_channel {
e4421fec 71 struct lttng_ht_node_ulong node;
3bd1e081
MD
72 int key;
73 uint64_t max_sb_size; /* the subbuffer size for this channel */
74 int refcount; /* Number of streams referencing this channel */
75 /* For UST */
76 int shm_fd;
77 int wait_fd;
78 void *mmap_base;
79 size_t mmap_len;
13161846 80 struct lttng_ust_shm_handle *handle;
3bd1e081 81 int nr_streams;
b5c5fc29 82 int wait_fd_is_copy;
5af2f756 83 int cpucount;
3bd1e081
MD
84};
85
86/* Forward declaration for UST. */
13161846 87struct lttng_ust_lib_ring_buffer;
3bd1e081
MD
88
89/*
90 * Internal representation of the streams, sessiond_key is used to identify
91 * uniquely a stream.
92 */
93struct lttng_consumer_stream {
e4421fec 94 struct lttng_ht_node_ulong node;
3bd1e081
MD
95 struct lttng_consumer_channel *chan; /* associated channel */
96 /*
97 * key is the key used by the session daemon to refer to the
98 * object in the consumer daemon.
99 */
100 int key;
101 int shm_fd;
102 int wait_fd;
103 int out_fd; /* output file to write the data */
104 off_t out_fd_offset; /* write position in the output file descriptor */
105 char path_name[PATH_MAX]; /* tracefile name */
106 enum lttng_consumer_stream_state state;
107 size_t shm_len;
108 void *mmap_base;
109 size_t mmap_len;
110 enum lttng_event_output output; /* splice or mmap */
b5c5fc29
MD
111 int shm_fd_is_copy;
112 int wait_fd_is_copy;
3bd1e081 113 /* For UST */
13161846 114 struct lttng_ust_lib_ring_buffer *buf;
3bd1e081 115 int cpu;
4078b776 116 int data_read;
d056b477 117 int hangup_flush_done;
6df2e2c9
MD
118 /* UID/GID of the user owning the session to which stream belongs */
119 uid_t uid;
120 gid_t gid;
3bd1e081
MD
121};
122
123/*
124 * UST consumer local data to the program. One or more instance per
125 * process.
126 */
127struct lttng_consumer_local_data {
4078b776
MD
128 /*
129 * Function to call when data is available on a buffer.
130 * Returns the number of bytes read, or negative error value.
131 */
132 ssize_t (*on_buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 133 struct lttng_consumer_local_data *ctx);
3bd1e081
MD
134 /*
135 * function to call when we receive a new channel, it receives a
136 * newly allocated channel, depending on the return code of this
137 * function, the new channel will be handled by the application
138 * or the library.
139 *
140 * Returns:
141 * > 0 (success, FD is kept by application)
142 * == 0 (success, FD is left to library)
143 * < 0 (error)
144 */
145 int (*on_recv_channel)(struct lttng_consumer_channel *channel);
146 /*
147 * function to call when we receive a new stream, it receives a
148 * newly allocated stream, depending on the return code of this
149 * function, the new stream will be handled by the application
150 * or the library.
151 *
152 * Returns:
153 * > 0 (success, FD is kept by application)
154 * == 0 (success, FD is left to library)
155 * < 0 (error)
156 */
157 int (*on_recv_stream)(struct lttng_consumer_stream *stream);
158 /*
159 * function to call when a stream is getting updated by the session
160 * daemon, this function receives the sessiond key and the new
161 * state, depending on the return code of this function the
162 * update of state for the stream is handled by the application
163 * or the library.
164 *
165 * Returns:
166 * > 0 (success, FD is kept by application)
167 * == 0 (success, FD is left to library)
168 * < 0 (error)
169 */
170 int (*on_update_stream)(int sessiond_key, uint32_t state);
171 /* socket to communicate errors with sessiond */
172 int consumer_error_socket;
173 /* socket to exchange commands with sessiond */
174 char *consumer_command_sock_path;
175 /* communication with splice */
176 int consumer_thread_pipe[2];
177 /* pipe to wake the poll thread when necessary */
178 int consumer_poll_pipe[2];
179 /* to let the signal handler wake up the fd receiver thread */
180 int consumer_should_quit[2];
181};
182
183/*
184 * Library-level data. One instance per process.
185 */
186struct lttng_consumer_global_data {
e4421fec 187
3bd1e081 188 /*
e4421fec
DG
189 * At this time, this lock is used to ensure coherence between the count
190 * and number of element in the hash table. It's also a protection for
6065ceec
DG
191 * concurrent read/write between threads.
192 *
193 * XXX: We need to see if this lock is still needed with the lockless RCU
194 * hash tables.
3bd1e081
MD
195 */
196 pthread_mutex_t lock;
e4421fec 197
3bd1e081 198 /*
e4421fec 199 * Number of streams in the hash table. Protected by consumer_data.lock.
3bd1e081
MD
200 */
201 int stream_count;
202 /*
e4421fec 203 * Hash tables of streams and channels. Protected by consumer_data.lock.
3bd1e081 204 */
e4421fec
DG
205 struct lttng_ht *stream_ht;
206 struct lttng_ht *channel_ht;
3bd1e081
MD
207 /*
208 * Flag specifying if the local array of FDs needs update in the
209 * poll function. Protected by consumer_data.lock.
210 */
211 unsigned int need_update;
212 enum lttng_consumer_type type;
213};
214
e4421fec
DG
215/*
216 * Init consumer data structures.
217 */
218extern void lttng_consumer_init(void);
219
3bd1e081
MD
220/*
221 * Set the error socket for communication with a session daemon.
222 */
223extern void lttng_consumer_set_error_sock(
224 struct lttng_consumer_local_data *ctx, int sock);
225
226/*
227 * Set the command socket path for communication with a session daemon.
228 */
229extern void lttng_consumer_set_command_sock_path(
230 struct lttng_consumer_local_data *ctx, char *sock);
231
232/*
233 * Send return code to session daemon.
234 *
235 * Returns the return code of sendmsg : the number of bytes transmitted or -1
236 * on error.
237 */
238extern int lttng_consumer_send_error(
239 struct lttng_consumer_local_data *ctx, int cmd);
240
241/*
242 * Called from signal handler to ensure a clean exit.
243 */
244extern void lttng_consumer_should_exit(
245 struct lttng_consumer_local_data *ctx);
246
247/*
248 * Cleanup the daemon's socket on exit.
249 */
250extern void lttng_consumer_cleanup(void);
251
252/*
253 * Flush pending writes to trace output disk file.
254 */
255extern void lttng_consumer_sync_trace_file(
256 struct lttng_consumer_stream *stream, off_t orig_offset);
257
258/*
259 * Poll on the should_quit pipe and the command socket return -1 on error and
260 * should exit, 0 if data is available on the command socket
261 */
262extern int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
263
264extern int consumer_update_poll_array(
265 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
266 struct lttng_consumer_stream **local_consumer_streams);
267
268extern struct lttng_consumer_stream *consumer_allocate_stream(
269 int channel_key, int stream_key,
270 int shm_fd, int wait_fd,
271 enum lttng_consumer_stream_state state,
272 uint64_t mmap_len,
273 enum lttng_event_output output,
6df2e2c9
MD
274 const char *path_name,
275 uid_t uid,
276 gid_t gid);
3bd1e081
MD
277extern int consumer_add_stream(struct lttng_consumer_stream *stream);
278extern void consumer_del_stream(struct lttng_consumer_stream *stream);
279extern void consumer_change_stream_state(int stream_key,
280 enum lttng_consumer_stream_state state);
281extern void consumer_del_channel(struct lttng_consumer_channel *channel);
282extern struct lttng_consumer_channel *consumer_allocate_channel(
283 int channel_key,
284 int shm_fd, int wait_fd,
285 uint64_t mmap_len,
286 uint64_t max_sb_size);
287int consumer_add_channel(struct lttng_consumer_channel *channel);
288
289extern struct lttng_consumer_local_data *lttng_consumer_create(
290 enum lttng_consumer_type type,
4078b776 291 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
d41f73b7 292 struct lttng_consumer_local_data *ctx),
3bd1e081
MD
293 int (*recv_channel)(struct lttng_consumer_channel *channel),
294 int (*recv_stream)(struct lttng_consumer_stream *stream),
295 int (*update_stream)(int sessiond_key, uint32_t state));
296extern void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
4078b776 297extern ssize_t lttng_consumer_on_read_subbuffer_mmap(
3bd1e081
MD
298 struct lttng_consumer_local_data *ctx,
299 struct lttng_consumer_stream *stream, unsigned long len);
4078b776 300extern ssize_t lttng_consumer_on_read_subbuffer_splice(
3bd1e081
MD
301 struct lttng_consumer_local_data *ctx,
302 struct lttng_consumer_stream *stream, unsigned long len);
303extern int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
304 struct lttng_consumer_stream *stream);
305extern int lttng_consumer_get_produced_snapshot(
306 struct lttng_consumer_local_data *ctx,
307 struct lttng_consumer_stream *stream,
308 unsigned long *pos);
309extern void *lttng_consumer_thread_poll_fds(void *data);
310extern void *lttng_consumer_thread_receive_fds(void *data);
311extern int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
312 int sock, struct pollfd *consumer_sockpoll);
313
4078b776 314ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
d41f73b7
MD
315 struct lttng_consumer_local_data *ctx);
316int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
317
3bd1e081 318#endif /* _LTTNG_CONSUMER_H */
This page took 0.064864 seconds and 4 git commands to generate.