Fix: channel management thread should hold a refcount
[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 int switch_timer_error;
150
151 /* On-disk circular buffer */
152 uint64_t tracefile_size;
153 uint64_t tracefile_count;
154 };
155
156 /*
157 * Internal representation of the streams, sessiond_key is used to identify
158 * uniquely a stream.
159 */
160 struct lttng_consumer_stream {
161 /* HT node used by the data_ht and metadata_ht */
162 struct lttng_ht_node_u64 node;
163 /* stream indexed per channel key node */
164 struct lttng_ht_node_u64 node_channel_id;
165 /* HT node used in consumer_data.stream_list_ht */
166 struct lttng_ht_node_u64 node_session_id;
167 /* Pointer to associated channel. */
168 struct lttng_consumer_channel *chan;
169
170 /* Key by which the stream is indexed for 'node'. */
171 uint64_t key;
172 /*
173 * File descriptor of the data output file. This can be either a file or a
174 * socket fd for relayd streaming.
175 */
176 int out_fd; /* output file to write the data */
177 /* Write position in the output file descriptor */
178 off_t out_fd_offset;
179 enum lttng_consumer_stream_state state;
180 int shm_fd_is_copy;
181 int data_read;
182 int hangup_flush_done;
183 enum lttng_event_output output;
184 /* Maximum subbuffer size. */
185 unsigned long max_sb_size;
186
187 /*
188 * Still used by the kernel for MMAP output. For UST, the ustctl getter is
189 * used for the mmap base and offset.
190 */
191 void *mmap_base;
192 unsigned long mmap_len;
193
194 /* For UST */
195
196 int wait_fd;
197 /* UID/GID of the user owning the session to which stream belongs */
198 uid_t uid;
199 gid_t gid;
200 /* Network sequence number. Indicating on which relayd socket it goes. */
201 uint64_t net_seq_idx;
202 /* Identify if the stream is the metadata */
203 unsigned int metadata_flag;
204 /* Used when the stream is set for network streaming */
205 uint64_t relayd_stream_id;
206 /*
207 * When sending a stream packet to a relayd, this number is used to track
208 * the packet sent by the consumer and seen by the relayd. When sending the
209 * data header to the relayd, this number is sent and if the transmission
210 * was successful, it is incremented.
211 *
212 * Even if the full data is not fully transmitted it won't matter since
213 * only two possible error can happen after that where either the relayd
214 * died or a read error is detected on the stream making this value useless
215 * after that.
216 *
217 * This value SHOULD be read/updated atomically or with the lock acquired.
218 */
219 uint64_t next_net_seq_num;
220 /*
221 * Lock to use the stream FDs since they are used between threads.
222 *
223 * This is nested INSIDE the consumer_data lock.
224 * This is nested OUTSIDE consumer_relayd_sock_pair lock.
225 */
226 pthread_mutex_t lock;
227 /* Tracing session id */
228 uint64_t session_id;
229 /*
230 * Indicates if the stream end point is still active or not (network
231 * streaming or local file system). The thread "owning" the stream is
232 * handling this status and can be notified of a state change through the
233 * consumer data appropriate pipe.
234 */
235 enum consumer_endpoint_status endpoint_status;
236 /* Stream name. Format is: <channel_name>_<cpu_number> */
237 char name[LTTNG_SYMBOL_NAME_LEN];
238 /* Internal state of libustctl. */
239 struct ustctl_consumer_stream *ustream;
240 struct cds_list_head send_node;
241 /* On-disk circular buffer */
242 uint64_t tracefile_size_current;
243 uint64_t tracefile_count_current;
244 };
245
246 /*
247 * Internal representation of a relayd socket pair.
248 */
249 struct consumer_relayd_sock_pair {
250 /* Network sequence number. */
251 int64_t net_seq_idx;
252 /* Number of stream associated with this relayd */
253 unsigned int refcount;
254
255 /*
256 * This flag indicates whether or not we should destroy this object. The
257 * destruction should ONLY occurs when this flag is set and the refcount is
258 * set to zero.
259 */
260 unsigned int destroy_flag;
261
262 /*
263 * Mutex protecting the control socket to avoid out of order packets
264 * between threads sending data to the relayd. Since metadata data is sent
265 * over that socket, at least two sendmsg() are needed (header + data)
266 * creating a race for packets to overlap between threads using it.
267 *
268 * This is nested INSIDE the consumer_data lock.
269 * This is nested INSIDE the stream lock.
270 */
271 pthread_mutex_t ctrl_sock_mutex;
272
273 /* Control socket. Command and metadata are passed over it */
274 struct lttcomm_relayd_sock control_sock;
275
276 /*
277 * We don't need a mutex at this point since we only splice or write single
278 * large chunk of data with a header appended at the begining. Moreover,
279 * this socket is for now only used in a single thread.
280 */
281 struct lttcomm_relayd_sock data_sock;
282 struct lttng_ht_node_u64 node;
283
284 /* Session id on both sides for the sockets. */
285 uint64_t relayd_session_id;
286 uint64_t sessiond_session_id;
287 };
288
289 /*
290 * UST consumer local data to the program. One or more instance per
291 * process.
292 */
293 struct lttng_consumer_local_data {
294 /*
295 * Function to call when data is available on a buffer.
296 * Returns the number of bytes read, or negative error value.
297 */
298 ssize_t (*on_buffer_ready)(struct lttng_consumer_stream *stream,
299 struct lttng_consumer_local_data *ctx);
300 /*
301 * function to call when we receive a new channel, it receives a
302 * newly allocated channel, depending on the return code of this
303 * function, the new channel will be handled by the application
304 * or the library.
305 *
306 * Returns:
307 * > 0 (success, FD is kept by application)
308 * == 0 (success, FD is left to library)
309 * < 0 (error)
310 */
311 int (*on_recv_channel)(struct lttng_consumer_channel *channel);
312 /*
313 * function to call when we receive a new stream, it receives a
314 * newly allocated stream, depending on the return code of this
315 * function, the new stream will be handled by the application
316 * or the library.
317 *
318 * Returns:
319 * > 0 (success, FD is kept by application)
320 * == 0 (success, FD is left to library)
321 * < 0 (error)
322 */
323 int (*on_recv_stream)(struct lttng_consumer_stream *stream);
324 /*
325 * function to call when a stream is getting updated by the session
326 * daemon, this function receives the sessiond key and the new
327 * state, depending on the return code of this function the
328 * update of state for the stream is handled by the application
329 * or the library.
330 *
331 * Returns:
332 * > 0 (success, FD is kept by application)
333 * == 0 (success, FD is left to library)
334 * < 0 (error)
335 */
336 int (*on_update_stream)(int sessiond_key, uint32_t state);
337 enum lttng_consumer_type type;
338 /* socket to communicate errors with sessiond */
339 int consumer_error_socket;
340 /* socket to ask metadata to sessiond */
341 int consumer_metadata_socket;
342 /* socket to exchange commands with sessiond */
343 char *consumer_command_sock_path;
344 /* communication with splice */
345 int consumer_thread_pipe[2];
346 int consumer_channel_pipe[2];
347 int consumer_splice_metadata_pipe[2];
348 /* Data stream poll thread pipe. To transfer data stream to the thread */
349 int consumer_data_pipe[2];
350 /* to let the signal handler wake up the fd receiver thread */
351 int consumer_should_quit[2];
352 /* Metadata poll thread pipe. Transfer metadata stream to it */
353 int consumer_metadata_pipe[2];
354 };
355
356 /*
357 * Library-level data. One instance per process.
358 */
359 struct lttng_consumer_global_data {
360 /*
361 * At this time, this lock is used to ensure coherence between the count
362 * and number of element in the hash table. It's also a protection for
363 * concurrent read/write between threads.
364 *
365 * This is nested OUTSIDE the stream lock.
366 * This is nested OUTSIDE the consumer_relayd_sock_pair lock.
367 */
368 pthread_mutex_t lock;
369
370 /*
371 * Number of streams in the data stream hash table declared outside.
372 * Protected by consumer_data.lock.
373 */
374 int stream_count;
375
376 /* Channel hash table protected by consumer_data.lock. */
377 struct lttng_ht *channel_ht;
378 /*
379 * Flag specifying if the local array of FDs needs update in the
380 * poll function. Protected by consumer_data.lock.
381 */
382 unsigned int need_update;
383 enum lttng_consumer_type type;
384
385 /*
386 * Relayd socket(s) hashtable indexed by network sequence number. Each
387 * stream has an index which associate the right relayd socket to use.
388 */
389 struct lttng_ht *relayd_ht;
390
391 /*
392 * This hash table contains all streams (metadata and data) indexed by
393 * session id. In other words, the ht is indexed by session id and each
394 * bucket contains the list of associated streams.
395 *
396 * This HT uses the "node_session_id" of the consumer stream.
397 */
398 struct lttng_ht *stream_list_ht;
399
400 /*
401 * This HT uses the "node_channel_id" of the consumer stream.
402 */
403 struct lttng_ht *stream_per_chan_id_ht;
404 };
405
406 /*
407 * Init consumer data structures.
408 */
409 void lttng_consumer_init(void);
410
411 /*
412 * Set the error socket for communication with a session daemon.
413 */
414 void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx,
415 int sock);
416
417 /*
418 * Set the command socket path for communication with a session daemon.
419 */
420 void lttng_consumer_set_command_sock_path(
421 struct lttng_consumer_local_data *ctx, char *sock);
422
423 /*
424 * Send return code to session daemon.
425 *
426 * Returns the return code of sendmsg : the number of bytes transmitted or -1
427 * on error.
428 */
429 int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd);
430
431 /*
432 * Called from signal handler to ensure a clean exit.
433 */
434 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx);
435
436 /*
437 * Cleanup the daemon's socket on exit.
438 */
439 void lttng_consumer_cleanup(void);
440
441 /*
442 * Flush pending writes to trace output disk file.
443 */
444 void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
445 off_t orig_offset);
446
447 /*
448 * Poll on the should_quit pipe and the command socket return -1 on error and
449 * should exit, 0 if data is available on the command socket
450 */
451 int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
452
453 struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
454 uint64_t stream_key,
455 enum lttng_consumer_stream_state state,
456 const char *channel_name,
457 uid_t uid,
458 gid_t gid,
459 int relayd_id,
460 uint64_t session_id,
461 int cpu,
462 int *alloc_ret,
463 enum consumer_channel_type type);
464 struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
465 uint64_t session_id,
466 const char *pathname,
467 const char *name,
468 uid_t uid,
469 gid_t gid,
470 int relayd_id,
471 enum lttng_event_output output,
472 uint64_t tracefile_size,
473 uint64_t tracefile_count);
474 void consumer_del_stream(struct lttng_consumer_stream *stream,
475 struct lttng_ht *ht);
476 void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
477 struct lttng_ht *ht);
478 int consumer_add_channel(struct lttng_consumer_channel *channel,
479 struct lttng_consumer_local_data *ctx);
480 void consumer_del_channel(struct lttng_consumer_channel *channel);
481
482 /* lttng-relayd consumer command */
483 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
484 int net_seq_idx);
485 struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key);
486 struct lttng_consumer_channel *consumer_find_channel(uint64_t key);
487 int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
488 size_t data_size);
489 void consumer_steal_stream_key(int key, struct lttng_ht *ht);
490
491 struct lttng_consumer_local_data *lttng_consumer_create(
492 enum lttng_consumer_type type,
493 ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
494 struct lttng_consumer_local_data *ctx),
495 int (*recv_channel)(struct lttng_consumer_channel *channel),
496 int (*recv_stream)(struct lttng_consumer_stream *stream),
497 int (*update_stream)(int sessiond_key, uint32_t state));
498 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
499 ssize_t lttng_consumer_on_read_subbuffer_mmap(
500 struct lttng_consumer_local_data *ctx,
501 struct lttng_consumer_stream *stream, unsigned long len,
502 unsigned long padding);
503 ssize_t lttng_consumer_on_read_subbuffer_splice(
504 struct lttng_consumer_local_data *ctx,
505 struct lttng_consumer_stream *stream, unsigned long len,
506 unsigned long padding);
507 int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream);
508 int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
509 unsigned long *pos);
510 void *consumer_thread_metadata_poll(void *data);
511 void *consumer_thread_data_poll(void *data);
512 void *consumer_thread_sessiond_poll(void *data);
513 void *consumer_thread_channel_poll(void *data);
514 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
515 int sock, struct pollfd *consumer_sockpoll);
516
517 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
518 struct lttng_consumer_local_data *ctx);
519 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
520 int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
521 struct lttng_consumer_local_data *ctx, int sock,
522 struct pollfd *consumer_sockpoll, struct lttcomm_relayd_sock *relayd_sock,
523 unsigned int sessiond_id);
524 void consumer_flag_relayd_for_destroy(
525 struct consumer_relayd_sock_pair *relayd);
526 int consumer_data_pending(uint64_t id);
527 int consumer_send_status_msg(int sock, int ret_code);
528 int consumer_send_status_channel(int sock,
529 struct lttng_consumer_channel *channel);
530
531 #endif /* LIB_CONSUMER_H */
This page took 0.047115 seconds and 4 git commands to generate.