6ac781605c7284ecf7319e2b10fa0c3920b441b3
[lttng-tools.git] / src / common / consumer.h
1 /*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Copyright (C) 2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 *
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.
8 *
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.
13 *
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.
17 */
18
19 #ifndef _LTTNG_CONSUMER_H
20 #define _LTTNG_CONSUMER_H
21
22 #include <limits.h>
23 #include <poll.h>
24 #include <unistd.h>
25
26 #include <lttng/lttng.h>
27
28 #include <common/hashtable/hashtable.h>
29 #include <common/compat/fcntl.h>
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
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
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 */
47 enum 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 */
57 enum lttng_consumer_stream_state {
58 LTTNG_CONSUMER_ACTIVE_STREAM,
59 LTTNG_CONSUMER_PAUSE_STREAM,
60 LTTNG_CONSUMER_DELETE_STREAM,
61 };
62
63 enum lttng_consumer_type {
64 LTTNG_CONSUMER_UNKNOWN = 0,
65 LTTNG_CONSUMER_KERNEL,
66 LTTNG_CONSUMER64_UST,
67 LTTNG_CONSUMER32_UST,
68 };
69
70 struct lttng_consumer_channel {
71 struct lttng_ht_node_ulong node;
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;
80 struct lttng_ust_shm_handle *handle;
81 int nr_streams;
82 int wait_fd_is_copy;
83 int cpucount;
84 };
85
86 /* Forward declaration for UST. */
87 struct lttng_ust_lib_ring_buffer;
88
89 /*
90 * Internal representation of the streams, sessiond_key is used to identify
91 * uniquely a stream.
92 */
93 struct lttng_consumer_stream {
94 struct lttng_ht_node_ulong node;
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 */
111 int shm_fd_is_copy;
112 int wait_fd_is_copy;
113 /* For UST */
114 struct lttng_ust_lib_ring_buffer *buf;
115 int cpu;
116 int data_read;
117 int hangup_flush_done;
118 /* UID/GID of the user owning the session to which stream belongs */
119 uid_t uid;
120 gid_t gid;
121 };
122
123 /*
124 * UST consumer local data to the program. One or more instance per
125 * process.
126 */
127 struct lttng_consumer_local_data {
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,
133 struct lttng_consumer_local_data *ctx);
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 */
186 struct lttng_consumer_global_data {
187
188 /*
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
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.
195 */
196 pthread_mutex_t lock;
197
198 /*
199 * Number of streams in the hash table. Protected by consumer_data.lock.
200 */
201 int stream_count;
202 /*
203 * Hash tables of streams and channels. Protected by consumer_data.lock.
204 */
205 struct lttng_ht *stream_ht;
206 struct lttng_ht *channel_ht;
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
215 /*
216 * Init consumer data structures.
217 */
218 extern void lttng_consumer_init(void);
219
220 /*
221 * Set the error socket for communication with a session daemon.
222 */
223 extern 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 */
229 extern 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 */
238 extern 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 */
244 extern void lttng_consumer_should_exit(
245 struct lttng_consumer_local_data *ctx);
246
247 /*
248 * Cleanup the daemon's socket on exit.
249 */
250 extern void lttng_consumer_cleanup(void);
251
252 /*
253 * Flush pending writes to trace output disk file.
254 */
255 extern 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 */
262 extern int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
263
264 extern int consumer_update_poll_array(
265 struct lttng_consumer_local_data *ctx, struct pollfd **pollfd,
266 struct lttng_consumer_stream **local_consumer_streams);
267
268 extern 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,
274 const char *path_name,
275 uid_t uid,
276 gid_t gid);
277 extern int consumer_add_stream(struct lttng_consumer_stream *stream);
278 extern void consumer_del_stream(struct lttng_consumer_stream *stream);
279 extern void consumer_change_stream_state(int stream_key,
280 enum lttng_consumer_stream_state state);
281 extern void consumer_del_channel(struct lttng_consumer_channel *channel);
282 extern 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);
287 int consumer_add_channel(struct lttng_consumer_channel *channel);
288
289 extern struct lttng_consumer_local_data *lttng_consumer_create(
290 enum lttng_consumer_type type,
291 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
292 struct lttng_consumer_local_data *ctx),
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));
296 extern void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
297 extern ssize_t lttng_consumer_on_read_subbuffer_mmap(
298 struct lttng_consumer_local_data *ctx,
299 struct lttng_consumer_stream *stream, unsigned long len);
300 extern ssize_t lttng_consumer_on_read_subbuffer_splice(
301 struct lttng_consumer_local_data *ctx,
302 struct lttng_consumer_stream *stream, unsigned long len);
303 extern int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
304 struct lttng_consumer_stream *stream);
305 extern int lttng_consumer_get_produced_snapshot(
306 struct lttng_consumer_local_data *ctx,
307 struct lttng_consumer_stream *stream,
308 unsigned long *pos);
309 extern void *lttng_consumer_thread_poll_fds(void *data);
310 extern void *lttng_consumer_thread_receive_fds(void *data);
311 extern int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
312 int sock, struct pollfd *consumer_sockpoll);
313
314 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
315 struct lttng_consumer_local_data *ctx);
316 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
317
318 #endif /* _LTTNG_CONSUMER_H */
This page took 0.053025 seconds and 3 git commands to generate.