1 /* Copyright (C) 2009 Pierre-Marc Fournier
3 * This library is free software; you can redistribute it and/or
4 * modify it under the terms of the GNU Lesser General Public
5 * License as published by the Free Software Foundation; either
6 * version 2.1 of the License, or (at your option) any later version.
8 * This library is distributed in the hope that it will be useful,
9 * but WITHOUT ANY WARRANTY; without even the implied warranty of
10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
11 * Lesser General Public License for more details.
13 * You should have received a copy of the GNU Lesser General Public
14 * License along with this library; if not, write to the Free Software
15 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
22 #include <sys/types.h>
23 #include <sys/socket.h>
35 #include "relay.h" /* FIXME: remove */
36 #include "marker-control.h"
40 #define USTSIGNAL SIGIO
42 #define MAX_MSG_SIZE (100)
44 #define MSG_REGISTER_NOTIF 2
46 char consumer_stack
[10000];
48 struct list_head blocked_consumers
= LIST_HEAD_INIT(blocked_consumers
);
50 static struct ustcomm_app ustcomm_app
;
52 struct tracecmd
{ /* no padding */
57 //struct listener_arg {
62 /* size: the size of all the fields except size itself */
65 /* Only the necessary part of the payload is transferred. It
66 * may even be none of it.
71 struct consumer_channel
{
73 struct ltt_channel_struct
*chan
;
76 struct blocked_consumer
{
81 /* args to ustcomm_send_reply */
82 struct ustcomm_server server
;
83 struct ustcomm_source src
;
85 /* args to ltt_do_get_subbuf */
86 struct rchan_buf
*rbuf
;
87 struct ltt_channel_buf_struct
*lttbuf
;
89 struct list_head list
;
92 static void print_markers(FILE *fp
)
94 struct marker_iter iter
;
97 marker_iter_reset(&iter
);
98 marker_iter_start(&iter
);
101 fprintf(fp
, "marker: %s_%s %d \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, (int)imv_read(iter
.marker
->state
), iter
.marker
->format
);
102 marker_iter_next(&iter
);
107 static int init_socket(void);
109 /* This needs to be called whenever a new thread is created. It notifies
110 * liburcu of the new thread.
113 void ust_register_thread(void)
115 rcu_register_thread();
122 struct trctl_msg msg
;
124 /* FIXME: fd_notif should probably be protected by a spinlock */
129 msg
.type
= MSG_NOTIF
;
130 msg
.size
= sizeof(msg
.type
);
132 /* FIXME: don't block here */
133 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
140 static void inform_consumer_daemon(void)
142 ustcomm_request_consumer(getpid(), "metadata");
143 ustcomm_request_consumer(getpid(), "ust");
146 void process_blocked_consumers(void)
150 struct blocked_consumer
*bc
;
155 list_for_each_entry(bc
, &blocked_consumers
, list
) {
159 fds
= (struct pollfd
*) malloc(n_fds
* sizeof(struct pollfd
));
161 ERR("malloc returned NULL");
165 list_for_each_entry(bc
, &blocked_consumers
, list
) {
166 fds
[idx
].fd
= bc
->fd_producer
;
167 fds
[idx
].events
= POLLIN
;
168 bc
->tmp_poll_idx
= idx
;
172 while((result
= poll(fds
, n_fds
, 0)) == -1 && errno
== EINTR
)
179 list_for_each_entry(bc
, &blocked_consumers
, list
) {
180 if(fds
[bc
->tmp_poll_idx
].revents
) {
181 long consumed_old
= 0;
184 result
= read(bc
->fd_producer
, &inbuf
, 1);
192 close(bc
->fd_producer
);
196 result
= ustcomm_send_reply(&bc
->server
, "END", &bc
->src
);
198 ERR("ustcomm_send_reply failed");
205 result
= ltt_do_get_subbuf(bc
->rbuf
, bc
->lttbuf
, &consumed_old
);
206 if(result
== -EAGAIN
) {
207 WARN("missed buffer?");
210 else if(result
< 0) {
211 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
213 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
214 result
= ustcomm_send_reply(&bc
->server
, reply
, &bc
->src
);
216 ERR("ustcomm_send_reply failed");
228 void *listener_main(void *p
)
232 ust_register_thread();
237 char trace_name
[] = "auto";
238 char trace_type
[] = "ustrelay";
241 struct ustcomm_source src
;
243 process_blocked_consumers();
245 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, 5);
247 WARN("error in ustcomm_app_recv_message");
250 else if(result
== 0) {
255 DBG("received a message! it's: %s\n", recvbuf
);
256 len
= strlen(recvbuf
);
258 if(!strcmp(recvbuf
, "print_markers")) {
259 print_markers(stderr
);
261 else if(!strcmp(recvbuf
, "list_markers")) {
266 fp
= open_memstream(&ptr
, &size
);
270 result
= ustcomm_send_reply(&ustcomm_app
.server
, ptr
, &src
);
274 else if(!strcmp(recvbuf
, "start")) {
275 /* start is an operation that setups the trace, allocates it and starts it */
276 result
= ltt_trace_setup(trace_name
);
278 ERR("ltt_trace_setup failed");
282 result
= ltt_trace_set_type(trace_name
, trace_type
);
284 ERR("ltt_trace_set_type failed");
288 result
= ltt_trace_alloc(trace_name
);
290 ERR("ltt_trace_alloc failed");
294 inform_consumer_daemon();
296 result
= ltt_trace_start(trace_name
);
298 ERR("ltt_trace_start failed");
302 else if(!strcmp(recvbuf
, "trace_setup")) {
305 result
= ltt_trace_setup(trace_name
);
307 ERR("ltt_trace_setup failed");
311 result
= ltt_trace_set_type(trace_name
, trace_type
);
313 ERR("ltt_trace_set_type failed");
317 else if(!strcmp(recvbuf
, "trace_alloc")) {
320 result
= ltt_trace_alloc(trace_name
);
322 ERR("ltt_trace_alloc failed");
326 else if(!strcmp(recvbuf
, "trace_start")) {
329 result
= ltt_trace_start(trace_name
);
331 ERR("ltt_trace_start failed");
335 else if(!strcmp(recvbuf
, "trace_stop")) {
338 result
= ltt_trace_stop(trace_name
);
340 ERR("ltt_trace_stop failed");
344 else if(!strcmp(recvbuf
, "trace_destroy")) {
346 DBG("trace destroy");
348 result
= ltt_trace_destroy(trace_name
);
350 ERR("ltt_trace_destroy failed");
354 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
355 struct ltt_trace_struct
*trace
;
356 char trace_name
[] = "auto";
362 channel_name
= nth_token(recvbuf
, 1);
363 if(channel_name
== NULL
) {
364 ERR("get_shmid: cannot parse channel");
369 trace
= _ltt_trace_find(trace_name
);
373 ERR("cannot find trace!");
377 for(i
=0; i
<trace
->nr_channels
; i
++) {
378 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
379 struct rchan_buf
*rbuf
= rchan
->buf
;
380 struct ltt_channel_struct
*ltt_channel
= (struct ltt_channel_struct
*)rchan
->private_data
;
382 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
385 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
386 DBG("the shmid for its buffer structure is %d", ltt_channel
->buf_shmid
);
387 asprintf(&reply
, "%d %d", rbuf
->shmid
, ltt_channel
->buf_shmid
);
389 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
391 ERR("listener: get_shmid: ustcomm_send_reply failed");
401 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
402 struct ltt_trace_struct
*trace
;
403 char trace_name
[] = "auto";
407 DBG("get_n_subbufs");
409 channel_name
= nth_token(recvbuf
, 1);
410 if(channel_name
== NULL
) {
411 ERR("get_n_subbufs: cannot parse channel");
416 trace
= _ltt_trace_find(trace_name
);
420 ERR("cannot find trace!");
424 for(i
=0; i
<trace
->nr_channels
; i
++) {
425 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
427 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
430 DBG("the n_subbufs for the requested channel is %zd", rchan
->n_subbufs
);
431 asprintf(&reply
, "%zd", rchan
->n_subbufs
);
433 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
435 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
445 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
446 struct ltt_trace_struct
*trace
;
447 char trace_name
[] = "auto";
451 DBG("get_subbuf_size");
453 channel_name
= nth_token(recvbuf
, 1);
454 if(channel_name
== NULL
) {
455 ERR("get_subbuf_size: cannot parse channel");
460 trace
= _ltt_trace_find(trace_name
);
464 ERR("cannot find trace!");
468 for(i
=0; i
<trace
->nr_channels
; i
++) {
469 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
471 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
474 DBG("the subbuf_size for the requested channel is %zd", rchan
->subbuf_size
);
475 asprintf(&reply
, "%zd", rchan
->subbuf_size
);
477 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
479 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
489 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
492 libfile
= nth_token(recvbuf
, 1);
494 DBG("load_probe_lib loading %s", libfile
);
496 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
497 struct ltt_trace_struct
*trace
;
498 char trace_name
[] = "auto";
504 channel_name
= nth_token(recvbuf
, 1);
505 if(channel_name
== NULL
) {
506 ERR("get_subbuf: cannot parse channel");
511 trace
= _ltt_trace_find(trace_name
);
515 ERR("cannot find trace!");
519 for(i
=0; i
<trace
->nr_channels
; i
++) {
520 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
522 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
523 struct rchan_buf
*rbuf
= rchan
->buf
;
524 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
525 struct blocked_consumer
*bc
;
527 bc
= (struct blocked_consumer
*) malloc(sizeof(struct blocked_consumer
));
529 ERR("malloc returned NULL");
532 bc
->fd_consumer
= src
.fd
;
533 bc
->fd_producer
= lttbuf
->data_ready_fd_read
;
537 bc
->server
= ustcomm_app
.server
;
539 list_add(&bc
->list
, &blocked_consumers
);
545 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
546 struct ltt_trace_struct
*trace
;
547 char trace_name
[] = "auto";
551 char *consumed_old_str
;
556 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
557 if(channel_name
== NULL
) {
558 ERR("put_subbuf_size: cannot parse channel");
562 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
563 if(consumed_old_str
== NULL
) {
564 ERR("put_subbuf: cannot parse consumed_old");
567 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
568 if(*endptr
!= '\0') {
569 ERR("put_subbuf: invalid value for consumed_old");
574 trace
= _ltt_trace_find(trace_name
);
578 ERR("cannot find trace!");
582 for(i
=0; i
<trace
->nr_channels
; i
++) {
583 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
585 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
586 struct rchan_buf
*rbuf
= rchan
->buf
;
587 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
591 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
593 WARN("ltt_do_put_subbuf: error (subbuf=%s)", channel_name
);
594 asprintf(&reply
, "%s", "ERROR");
597 DBG("ltt_do_put_subbuf: success (subbuf=%s)", channel_name
);
598 asprintf(&reply
, "%s", "OK");
601 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
603 ERR("listener: put_subbuf: ustcomm_send_reply failed");
614 free(consumed_old_str
);
616 else if(nth_token_is(recvbuf
, "enable_marker", 0) == 1) {
617 char *channel_slash_name
= nth_token(recvbuf
, 1);
618 char channel_name
[256]="";
619 char marker_name
[256]="";
621 result
= sscanf(channel_slash_name
, "%255[^/]/%255s", channel_name
, marker_name
);
623 if(channel_name
== NULL
|| marker_name
== NULL
) {
624 WARN("invalid marker name");
627 printf("%s %s\n", channel_name
, marker_name
);
629 result
= ltt_marker_connect(channel_name
, marker_name
, "default");
631 WARN("could not enable marker; channel=%s, name=%s", channel_name
, marker_name
);
634 else if(nth_token_is(recvbuf
, "disable_marker", 0) == 1) {
635 char *channel_slash_name
= nth_token(recvbuf
, 1);
639 result
= sscanf(channel_slash_name
, "%a[^/]/%as", &channel_name
, &marker_name
);
641 if(marker_name
== NULL
) {
643 printf("%s %s\n", channel_name
, marker_name
);
645 result
= ltt_marker_disconnect(channel_name
, marker_name
, "default");
647 WARN("could not disable marker; channel=%s, name=%s", channel_name
, marker_name
);
650 // else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
651 // struct ltt_trace_struct *trace;
652 // char trace_name[] = "auto";
654 // char *channel_name;
656 // DBG("get_notifications");
658 // channel_name = strdup_malloc(nth_token(recvbuf, 1));
659 // if(channel_name == NULL) {
660 // ERR("put_subbuf_size: cannot parse channel");
664 // ltt_lock_traces();
665 // trace = _ltt_trace_find(trace_name);
666 // ltt_unlock_traces();
668 // if(trace == NULL) {
669 // ERR("cannot find trace!");
673 // for(i=0; i<trace->nr_channels; i++) {
674 // struct rchan *rchan = trace->channels[i].trans_channel_data;
677 // if(!strcmp(trace->channels[i].channel_name, channel_name)) {
678 // struct rchan_buf *rbuf = rchan->buf;
679 // struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
681 // result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
682 // if(result == -1) {
683 // ERR("ustcomm_app_detach_client failed");
687 // lttbuf->wake_consumer_arg = (void *) fd;
691 // lttbuf->call_wake_consumer = 1;
697 // free(channel_name);
700 ERR("unable to parse message: %s", recvbuf
);
708 int have_listener
= 0;
710 void create_listener(void)
713 static char listener_stack
[16384];
722 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
729 pthread_create(&thread
, NULL
, listener_main
, NULL
);
735 /* The signal handler itself. Signals must be setup so there cannot be
738 void sighandler(int sig
)
740 static char have_listener
= 0;
749 /* Called by the app signal handler to chain it to us. */
751 void chain_signal(void)
753 sighandler(USTSIGNAL
);
756 static int init_socket(void)
758 return ustcomm_init_app(getpid(), &ustcomm_app
);
761 /* FIXME: reenable this to delete socket file. */
764 static void destroy_socket(void)
768 if(mysocketfile
[0] == '\0')
771 result
= unlink(mysocketfile
);
778 static int init_signal_handler(void)
780 /* Attempt to handler SIGIO. If the main program wants to
781 * handle it, fine, it'll override us. They it'll have to
782 * use the chaining function.
786 struct sigaction act
;
788 result
= sigemptyset(&act
.sa_mask
);
790 PERROR("sigemptyset");
794 act
.sa_handler
= sighandler
;
795 act
.sa_flags
= SA_RESTART
;
797 /* Only defer ourselves. Also, try to restart interrupted
798 * syscalls to disturb the traced program as little as possible.
800 result
= sigaction(SIGIO
, &act
, NULL
);
809 static void auto_probe_connect(struct marker
*m
)
813 result
= ltt_marker_connect(m
->channel
, m
->name
, "default");
814 if(result
&& result
!= -EEXIST
)
815 ERR("ltt_marker_connect (marker = %s/%s, errno = %d)", m
->channel
, m
->name
, -result
);
817 DBG("just auto connected marker %s %s to probe default", m
->channel
, m
->name
);
820 static void __attribute__((constructor(101))) init0()
822 DBG("UST_AUTOPROBE constructor");
825 static void __attribute__((constructor(1000))) init()
829 /* Initialize RCU in case the constructor order is not good. */
832 /* It is important to do this before events start to be generated. */
833 ust_register_thread();
835 DBG("UST_TRACE constructor");
837 /* Must create socket before signal handler to prevent races.
839 result
= init_socket();
841 ERR("init_socket error");
844 result
= init_signal_handler();
846 ERR("init_signal_handler error");
850 if(getenv("UST_AUTOPROBE")) {
851 struct marker_iter iter
;
853 DBG("IN AUTOPROBE\n");
855 /* Ensure markers are initialized */
858 /* Ensure marker control is initialized, for the probe */
859 init_marker_control();
861 /* first, set the callback that will connect the
862 * probe on new markers
864 marker_set_new_marker_cb(auto_probe_connect
);
866 /* Now, connect the probes that were already registered. */
867 marker_iter_reset(&iter
);
868 marker_iter_start(&iter
);
870 DBG("now iterating on markers already registered\n");
872 DBG("now iterating on marker %s\n", iter
.marker
->name
);
873 auto_probe_connect(iter
.marker
);
874 marker_iter_next(&iter
);
878 if(getenv("UST_TRACE")) {
879 char trace_name
[] = "auto";
880 char trace_type
[] = "ustrelay";
882 DBG("starting early tracing");
884 /* Ensure marker control is initialized */
885 init_marker_control();
887 /* Ensure relay is initialized */
888 init_ustrelay_transport();
890 /* Ensure markers are initialized */
894 ltt_channels_register("ust");
896 result
= ltt_trace_setup(trace_name
);
898 ERR("ltt_trace_setup failed");
902 result
= ltt_trace_set_type(trace_name
, trace_type
);
904 ERR("ltt_trace_set_type failed");
908 result
= ltt_trace_alloc(trace_name
);
910 ERR("ltt_trace_alloc failed");
914 result
= ltt_trace_start(trace_name
);
916 ERR("ltt_trace_start failed");
919 inform_consumer_daemon();
925 /* should decrementally destroy stuff if error */
929 /* This is only called if we terminate normally, not with an unhandled signal,
930 * so we cannot rely on it. */
932 /* This destructor probably isn't needed, because ustd can do crash recovery. */
934 static void __attribute__((destructor
)) fini()
938 /* if trace running, finish it */
940 DBG("destructor stopping traces");
942 result
= ltt_trace_stop("auto");
944 ERR("ltt_trace_stop error");
947 result
= ltt_trace_destroy("auto");
949 ERR("ltt_trace_destroy error");
956 /* Notify ust that there was a fork. This needs to be called inside
957 * the new process, anytime a process whose memory is not shared with
958 * the parent is created. If this function is not called, the events
959 * of the new process will not be collected.
965 ltt_trace_stop("auto");
966 ltt_trace_destroy("auto");
967 ltt_trace_alloc("auto");
968 ltt_trace_start("auto");
972 ustcomm_request_consumer(getpid(), "metadata");
973 ustcomm_request_consumer(getpid(), "ust");