6fd4e2e5ce5ae08160475b3d6ba02d4069ca95df
[lttng-tools.git] / src / common / consumer.h
1 /*
2 * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * 2012 - David Goulet <dgoulet@efficios.com>
5 *
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
9 *
10 * This program is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * more details.
14 *
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20 #ifndef LIB_CONSUMER_H
21 #define LIB_CONSUMER_H
22
23 #include <limits.h>
24 #include <poll.h>
25 #include <unistd.h>
26 #include <urcu/list.h>
27
28 #include <lttng/lttng.h>
29
30 #include <common/hashtable/hashtable.h>
31 #include <common/compat/fcntl.h>
32 #include <common/compat/uuid.h>
33 #include <common/sessiond-comm/sessiond-comm.h>
34
35 /* Commands for consumer */
36 enum lttng_consumer_command {
37 LTTNG_CONSUMER_ADD_CHANNEL,
38 LTTNG_CONSUMER_ADD_STREAM,
39 /* pause, delete, active depending on fd state */
40 LTTNG_CONSUMER_UPDATE_STREAM,
41 /* inform the consumer to quit when all fd has hang up */
42 LTTNG_CONSUMER_STOP,
43 LTTNG_CONSUMER_ADD_RELAYD_SOCKET,
44 /* Inform the consumer to kill a specific relayd connection */
45 LTTNG_CONSUMER_DESTROY_RELAYD,
46 /* Return to the sessiond if there is data pending for a session */
47 LTTNG_CONSUMER_DATA_PENDING,
48 /* Consumer creates a channel and returns it to sessiond. */
49 LTTNG_CONSUMER_ASK_CHANNEL_CREATION,
50 LTTNG_CONSUMER_GET_CHANNEL,
51 LTTNG_CONSUMER_DESTROY_CHANNEL,
52 LTTNG_CONSUMER_PUSH_METADATA,
53 LTTNG_CONSUMER_CLOSE_METADATA,
54 LTTNG_CONSUMER_SETUP_METADATA,
55 LTTNG_CONSUMER_FLUSH_CHANNEL,
56 };
57
58 /* State of each fd in consumer */
59 enum lttng_consumer_stream_state {
60 LTTNG_CONSUMER_ACTIVE_STREAM,
61 LTTNG_CONSUMER_PAUSE_STREAM,
62 LTTNG_CONSUMER_DELETE_STREAM,
63 };
64
65 enum lttng_consumer_type {
66 LTTNG_CONSUMER_UNKNOWN = 0,
67 LTTNG_CONSUMER_KERNEL,
68 LTTNG_CONSUMER64_UST,
69 LTTNG_CONSUMER32_UST,
70 };
71
72 enum consumer_endpoint_status {
73 CONSUMER_ENDPOINT_ACTIVE,
74 CONSUMER_ENDPOINT_INACTIVE,
75 };
76
77 enum consumer_channel_output {
78 CONSUMER_CHANNEL_MMAP = 0,
79 CONSUMER_CHANNEL_SPLICE = 1,
80 };
81
82 enum consumer_channel_type {
83 CONSUMER_CHANNEL_TYPE_METADATA = 0,
84 CONSUMER_CHANNEL_TYPE_DATA = 1,
85 };
86
87 struct stream_list {
88 struct cds_list_head head;
89 unsigned int count;
90 };
91
92 /* Stub. */
93 struct consumer_metadata_cache;
94
95 struct lttng_consumer_channel {
96 /* HT node used for consumer_data.channel_ht */
97 struct lttng_ht_node_u64 node;
98 /* Indexed key. Incremented value in the consumer. */
99 uint64_t key;
100 /* Number of streams referencing this channel */
101 int refcount;
102 /* Tracing session id on the session daemon side. */
103 uint64_t session_id;
104 /* Channel trace file path name. */
105 char pathname[PATH_MAX];
106 /* Channel name. */
107 char name[LTTNG_SYMBOL_NAME_LEN];
108 /* UID and GID of the channel. */
109 uid_t uid;
110 gid_t gid;
111 /* Relayd id of the channel. -1 if it does not apply. */
112 int64_t relayd_id;
113 /*
114 * Number of streams NOT initialized yet. This is used in order to not
115 * delete this channel if streams are getting initialized.
116 */
117 unsigned int nb_init_stream_left;
118 /* Output type (mmap or splice). */
119 enum consumer_channel_output output;
120 /* Channel type for stream */
121 enum consumer_channel_type type;
122
123 /* For UST */
124 struct ustctl_consumer_channel *uchan;
125 unsigned char uuid[UUID_STR_LEN];
126 /*
127 * Temporary stream list used to store the streams once created and waiting
128 * to be sent to the session daemon by receiving the
129 * LTTNG_CONSUMER_GET_CHANNEL.
130 */
131 struct stream_list streams;
132 /*
133 * Set if the channel is metadata. We keep a reference to the stream
134 * because we have to flush data once pushed by the session daemon. For a
135 * regular channel, this is always set to NULL.
136 */
137 struct lttng_consumer_stream *metadata_stream;
138
139 /* for UST */
140 int wait_fd;
141 /* Node within channel thread ht */
142 struct lttng_ht_node_u64 wait_fd_node;
143
144 /* Metadata cache is metadata channel */
145 struct consumer_metadata_cache *metadata_cache;
146 /* For metadata periodical flush */
147 int switch_timer_enabled;
148 timer_t switch_timer;
149 /* On-disk circular buffer */
150 uint64_t tracefile_size;
151 uint64_t tracefile_count;
152 };
153
154 /*
155 * Internal representation of the streams, sessiond_key is used to identify
156 * uniquely a stream.
157 */
158 struct lttng_consumer_stream {
159 /* HT node used by the data_ht and metadata_ht */
160 struct lttng_ht_node_u64 node;
161 /* stream indexed per channel key node */
162 struct lttng_ht_node_u64 node_channel_id;
163 /* HT node used in consumer_data.stream_list_ht */
164 struct lttng_ht_node_u64 node_session_id;
165 /* Pointer to associated channel. */
166 struct lttng_consumer_channel *chan;
167
168 /* Key by which the stream is indexed for 'node'. */
169 uint64_t key;
170 /*
171 * File descriptor of the data output file. This can be either a file or a
172 * socket fd for relayd streaming.
173 */
174 int out_fd; /* output file to write the data */
175 /* Write position in the output file descriptor */
176 off_t out_fd_offset;
177 enum lttng_consumer_stream_state state;
178 int shm_fd_is_copy;
179 int data_read;
180 int hangup_flush_done;
181 enum lttng_event_output output;
182 /* Maximum subbuffer size. */
183 unsigned long max_sb_size;
184
185 /*
186 * Still used by the kernel for MMAP output. For UST, the ustctl getter is
187 * used for the mmap base and offset.
188 */
189 void *mmap_base;
190 unsigned long mmap_len;
191
192 /* For UST */
193
194 int wait_fd;
195 /* UID/GID of the user owning the session to which stream belongs */
196 uid_t uid;
197 gid_t gid;
198 /* Network sequence number. Indicating on which relayd socket it goes. */
199 uint64_t net_seq_idx;
200 /* Identify if the stream is the metadata */
201 unsigned int metadata_flag;
202 /* Used when the stream is set for network streaming */
203 uint64_t relayd_stream_id;
204 /*
205 * When sending a stream packet to a relayd, this number is used to track
206 * the packet sent by the consumer and seen by the relayd. When sending the
207 * data header to the relayd, this number is sent and if the transmission
208 * was successful, it is incremented.
209 *
210 * Even if the full data is not fully transmitted it won't matter since
211 * only two possible error can happen after that where either the relayd
212 * died or a read error is detected on the stream making this value useless
213 * after that.
214 *
215 * This value SHOULD be read/updated atomically or with the lock acquired.
216 */
217 uint64_t next_net_seq_num;
218 /*
219 * Lock to use the stream FDs since they are used between threads.
220 *
221 * This is nested INSIDE the consumer_data lock.
222 * This is nested OUTSIDE consumer_relayd_sock_pair lock.
223 */
224 pthread_mutex_t lock;
225 /* Tracing session id */
226 uint64_t session_id;
227 /*
228 * Indicates if the stream end point is still active or not (network
229 * streaming or local file system). The thread "owning" the stream is
230 * handling this status and can be notified of a state change through the
231 * consumer data appropriate pipe.
232 */
233 enum consumer_endpoint_status endpoint_status;
234 /* Stream name. Format is: <channel_name>_<cpu_number> */
235 char name[LTTNG_SYMBOL_NAME_LEN];
236 /* Internal state of libustctl. */
237 struct ustctl_consumer_stream *ustream;
238 struct cds_list_head send_node;
239 /* On-disk circular buffer */
240 uint64_t tracefile_size_current;
241 uint64_t tracefile_count_current;
242 };
243
244 /*
245 * Internal representation of a relayd socket pair.
246 */
247 struct consumer_relayd_sock_pair {
248 /* Network sequence number. */
249 int64_t net_seq_idx;
250 /* Number of stream associated with this relayd */
251 unsigned int refcount;
252
253 /*
254 * This flag indicates whether or not we should destroy this object. The
255 * destruction should ONLY occurs when this flag is set and the refcount is
256 * set to zero.
257 */
258 unsigned int destroy_flag;
259
260 /*
261 * Mutex protecting the control socket to avoid out of order packets
262 * between threads sending data to the relayd. Since metadata data is sent
263 * over that socket, at least two sendmsg() are needed (header + data)
264 * creating a race for packets to overlap between threads using it.
265 *
266 * This is nested INSIDE the consumer_data lock.
267 * This is nested INSIDE the stream lock.
268 */
269 pthread_mutex_t ctrl_sock_mutex;
270
271 /* Control socket. Command and metadata are passed over it */
272 struct lttcomm_sock control_sock;
273
274 /*
275 * We don't need a mutex at this point since we only splice or write single
276 * large chunk of data with a header appended at the begining. Moreover,
277 * this socket is for now only used in a single thread.
278 */
279 struct lttcomm_sock data_sock;
280 struct lttng_ht_node_u64 node;
281
282 /* Session id on both sides for the sockets. */
283 uint64_t relayd_session_id;
284 uint64_t sessiond_session_id;
285 };
286
287 /*
288 * UST consumer local data to the program. One or more instance per
289 * process.
290 */
291 struct lttng_consumer_local_data {
292 /*
293 * Function to call when data is available on a buffer.
294 * Returns the number of bytes read, or negative error value.
295 */
296 ssize_t (*on_buffer_ready)(struct lttng_consumer_stream *stream,
297 struct lttng_consumer_local_data *ctx);
298 /*
299 * function to call when we receive a new channel, it receives a
300 * newly allocated channel, depending on the return code of this
301 * function, the new channel will be handled by the application
302 * or the library.
303 *
304 * Returns:
305 * > 0 (success, FD is kept by application)
306 * == 0 (success, FD is left to library)
307 * < 0 (error)
308 */
309 int (*on_recv_channel)(struct lttng_consumer_channel *channel);
310 /*
311 * function to call when we receive a new stream, it receives a
312 * newly allocated stream, depending on the return code of this
313 * function, the new stream will be handled by the application
314 * or the library.
315 *
316 * Returns:
317 * > 0 (success, FD is kept by application)
318 * == 0 (success, FD is left to library)
319 * < 0 (error)
320 */
321 int (*on_recv_stream)(struct lttng_consumer_stream *stream);
322 /*
323 * function to call when a stream is getting updated by the session
324 * daemon, this function receives the sessiond key and the new
325 * state, depending on the return code of this function the
326 * update of state for the stream is handled by the application
327 * or the library.
328 *
329 * Returns:
330 * > 0 (success, FD is kept by application)
331 * == 0 (success, FD is left to library)
332 * < 0 (error)
333 */
334 int (*on_update_stream)(int sessiond_key, uint32_t state);
335 enum lttng_consumer_type type;
336 /* socket to communicate errors with sessiond */
337 int consumer_error_socket;
338 /* socket to ask metadata to sessiond */
339 int consumer_metadata_socket;
340 /* socket to exchange commands with sessiond */
341 char *consumer_command_sock_path;
342 /* communication with splice */
343 int consumer_thread_pipe[2];
344 int consumer_channel_pipe[2];
345 int consumer_splice_metadata_pipe[2];
346 /* Data stream poll thread pipe. To transfer data stream to the thread */
347 int consumer_data_pipe[2];
348 /* to let the signal handler wake up the fd receiver thread */
349 int consumer_should_quit[2];
350 /* Metadata poll thread pipe. Transfer metadata stream to it */
351 int consumer_metadata_pipe[2];
352 };
353
354 /*
355 * Library-level data. One instance per process.
356 */
357 struct lttng_consumer_global_data {
358 /*
359 * At this time, this lock is used to ensure coherence between the count
360 * and number of element in the hash table. It's also a protection for
361 * concurrent read/write between threads.
362 *
363 * This is nested OUTSIDE the stream lock.
364 * This is nested OUTSIDE the consumer_relayd_sock_pair lock.
365 */
366 pthread_mutex_t lock;
367
368 /*
369 * Number of streams in the data stream hash table declared outside.
370 * Protected by consumer_data.lock.
371 */
372 int stream_count;
373
374 /* Channel hash table protected by consumer_data.lock. */
375 struct lttng_ht *channel_ht;
376 /*
377 * Flag specifying if the local array of FDs needs update in the
378 * poll function. Protected by consumer_data.lock.
379 */
380 unsigned int need_update;
381 enum lttng_consumer_type type;
382
383 /*
384 * Relayd socket(s) hashtable indexed by network sequence number. Each
385 * stream has an index which associate the right relayd socket to use.
386 */
387 struct lttng_ht *relayd_ht;
388
389 /*
390 * This hash table contains all streams (metadata and data) indexed by
391 * session id. In other words, the ht is indexed by session id and each
392 * bucket contains the list of associated streams.
393 *
394 * This HT uses the "node_session_id" of the consumer stream.
395 */
396 struct lttng_ht *stream_list_ht;
397
398 /*
399 * This HT uses the "node_channel_id" of the consumer stream.
400 */
401 struct lttng_ht *stream_per_chan_id_ht;
402 };
403
404 /*
405 * Init consumer data structures.
406 */
407 void lttng_consumer_init(void);
408
409 /*
410 * Set the error socket for communication with a session daemon.
411 */
412 void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
413 int sock);
414
415 /*
416 * Set the command socket path for communication with a session daemon.
417 */
418 void lttng_consumer_set_command_sock_path(
419 struct lttng_consumer_local_data *ctx, char *sock);
420
421 /*
422 * Send return code to session daemon.
423 *
424 * Returns the return code of sendmsg : the number of bytes transmitted or -1
425 * on error.
426 */
427 int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd);
428
429 /*
430 * Called from signal handler to ensure a clean exit.
431 */
432 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx);
433
434 /*
435 * Cleanup the daemon's socket on exit.
436 */
437 void lttng_consumer_cleanup(void);
438
439 /*
440 * Flush pending writes to trace output disk file.
441 */
442 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
443 off_t orig_offset);
444
445 /*
446 * Poll on the should_quit pipe and the command socket return -1 on error and
447 * should exit, 0 if data is available on the command socket
448 */
449 int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
450
451 struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
452 uint64_t stream_key,
453 enum lttng_consumer_stream_state state,
454 const char *channel_name,
455 uid_t uid,
456 gid_t gid,
457 int relayd_id,
458 uint64_t session_id,
459 int cpu,
460 int *alloc_ret,
461 enum consumer_channel_type type);
462 struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
463 uint64_t session_id,
464 const char *pathname,
465 const char *name,
466 uid_t uid,
467 gid_t gid,
468 int relayd_id,
469 enum lttng_event_output output,
470 uint64_t tracefile_size,
471 uint64_t tracefile_count);
472 void consumer_del_stream(struct lttng_consumer_stream *stream,
473 struct lttng_ht *ht);
474 void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
475 struct lttng_ht *ht);
476 int consumer_add_channel(struct lttng_consumer_channel *channel,
477 struct lttng_consumer_local_data *ctx);
478 void consumer_del_channel(struct lttng_consumer_channel *channel);
479
480 /* lttng-relayd consumer command */
481 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
482 int net_seq_idx);
483 struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key);
484 struct lttng_consumer_channel *consumer_find_channel(uint64_t key);
485 int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
486 size_t data_size);
487 void consumer_steal_stream_key(int key, struct lttng_ht *ht);
488
489 struct lttng_consumer_local_data *lttng_consumer_create(
490 enum lttng_consumer_type type,
491 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
492 struct lttng_consumer_local_data *ctx),
493 int (*recv_channel)(struct lttng_consumer_channel *channel),
494 int (*recv_stream)(struct lttng_consumer_stream *stream),
495 int (*update_stream)(int sessiond_key, uint32_t state));
496 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
497 int lttng_create_output_file(struct lttng_consumer_stream *stream);
498 ssize_t lttng_consumer_on_read_subbuffer_mmap(
499 struct lttng_consumer_local_data *ctx,
500 struct lttng_consumer_stream *stream, unsigned long len,
501 unsigned long padding);
502 ssize_t lttng_consumer_on_read_subbuffer_splice(
503 struct lttng_consumer_local_data *ctx,
504 struct lttng_consumer_stream *stream, unsigned long len,
505 unsigned long padding);
506 int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream);
507 int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
508 unsigned long *pos);
509 void *consumer_thread_metadata_poll(void *data);
510 void *consumer_thread_data_poll(void *data);
511 void *consumer_thread_sessiond_poll(void *data);
512 void *consumer_thread_channel_poll(void *data);
513 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
514 int sock, struct pollfd *consumer_sockpoll);
515
516 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
517 struct lttng_consumer_local_data *ctx);
518 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
519 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
520 struct lttng_consumer_local_data *ctx, int sock,
521 struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock,
522 unsigned int sessiond_id);
523 void consumer_flag_relayd_for_destroy(
524 struct consumer_relayd_sock_pair *relayd);
525 int consumer_data_pending(uint64_t id);
526 int consumer_send_status_msg(int sock, int ret_code);
527 int consumer_send_status_channel(int sock,
528 struct lttng_consumer_channel *channel);
529
530 #endif /* LIB_CONSUMER_H */
This page took 0.040829 seconds and 4 git commands to generate.