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>
36 #include "relay.h" /* FIXME: remove */
37 #include "marker-control.h"
41 #define USTSIGNAL SIGIO
43 #define MAX_MSG_SIZE (100)
45 #define MSG_REGISTER_NOTIF 2
47 char consumer_stack
[10000];
49 struct list_head blocked_consumers
= LIST_HEAD_INIT(blocked_consumers
);
51 static struct ustcomm_app ustcomm_app
;
53 struct tracecmd
{ /* no padding */
58 //struct listener_arg {
63 /* size: the size of all the fields except size itself */
66 /* Only the necessary part of the payload is transferred. It
67 * may even be none of it.
72 struct consumer_channel
{
74 struct ltt_channel_struct
*chan
;
77 struct blocked_consumer
{
82 /* args to ustcomm_send_reply */
83 struct ustcomm_server server
;
84 struct ustcomm_source src
;
86 /* args to ltt_do_get_subbuf */
87 struct rchan_buf
*rbuf
;
88 struct ltt_channel_buf_struct
*lttbuf
;
90 struct list_head list
;
93 static void print_markers(FILE *fp
)
95 struct marker_iter iter
;
98 marker_iter_reset(&iter
);
99 marker_iter_start(&iter
);
102 fprintf(fp
, "marker: %s_%s %d \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, (int)imv_read(iter
.marker
->state
), iter
.marker
->format
);
103 marker_iter_next(&iter
);
108 static int init_socket(void);
110 /* This needs to be called whenever a new thread is created. It notifies
111 * liburcu of the new thread.
114 void ust_register_thread(void)
116 rcu_register_thread();
123 struct trctl_msg msg
;
125 /* FIXME: fd_notif should probably be protected by a spinlock */
130 msg
.type
= MSG_NOTIF
;
131 msg
.size
= sizeof(msg
.type
);
133 /* FIXME: don't block here */
134 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
141 static void inform_consumer_daemon(void)
143 ustcomm_request_consumer(getpid(), "metadata");
144 ustcomm_request_consumer(getpid(), "ust");
147 void process_blocked_consumers(void)
151 struct blocked_consumer
*bc
;
156 list_for_each_entry(bc
, &blocked_consumers
, list
) {
160 fds
= (struct pollfd
*) malloc(n_fds
* sizeof(struct pollfd
));
162 ERR("malloc returned NULL");
166 list_for_each_entry(bc
, &blocked_consumers
, list
) {
167 fds
[idx
].fd
= bc
->fd_producer
;
168 fds
[idx
].events
= POLLIN
;
169 bc
->tmp_poll_idx
= idx
;
173 while((result
= poll(fds
, n_fds
, 0)) == -1 && errno
== EINTR
)
180 list_for_each_entry(bc
, &blocked_consumers
, list
) {
181 if(fds
[bc
->tmp_poll_idx
].revents
) {
182 long consumed_old
= 0;
185 result
= read(bc
->fd_producer
, &inbuf
, 1);
193 close(bc
->fd_producer
);
197 result
= ustcomm_send_reply(&bc
->server
, "END", &bc
->src
);
199 ERR("ustcomm_send_reply failed");
206 result
= ltt_do_get_subbuf(bc
->rbuf
, bc
->lttbuf
, &consumed_old
);
207 if(result
== -EAGAIN
) {
208 WARN("missed buffer?");
211 else if(result
< 0) {
212 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
214 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
215 result
= ustcomm_send_reply(&bc
->server
, reply
, &bc
->src
);
217 ERR("ustcomm_send_reply failed");
229 void *listener_main(void *p
)
233 ust_register_thread();
238 char trace_name
[] = "auto";
239 char trace_type
[] = "ustrelay";
242 struct ustcomm_source src
;
244 process_blocked_consumers();
246 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, 5);
248 WARN("error in ustcomm_app_recv_message");
251 else if(result
== 0) {
256 DBG("received a message! it's: %s", recvbuf
);
257 len
= strlen(recvbuf
);
259 if(!strcmp(recvbuf
, "print_markers")) {
260 print_markers(stderr
);
262 else if(!strcmp(recvbuf
, "list_markers")) {
267 fp
= open_memstream(&ptr
, &size
);
271 result
= ustcomm_send_reply(&ustcomm_app
.server
, ptr
, &src
);
275 else if(!strcmp(recvbuf
, "start")) {
276 /* start is an operation that setups the trace, allocates it and starts it */
277 result
= ltt_trace_setup(trace_name
);
279 ERR("ltt_trace_setup failed");
283 result
= ltt_trace_set_type(trace_name
, trace_type
);
285 ERR("ltt_trace_set_type failed");
289 result
= ltt_trace_alloc(trace_name
);
291 ERR("ltt_trace_alloc failed");
295 inform_consumer_daemon();
297 result
= ltt_trace_start(trace_name
);
299 ERR("ltt_trace_start failed");
303 else if(!strcmp(recvbuf
, "trace_setup")) {
306 result
= ltt_trace_setup(trace_name
);
308 ERR("ltt_trace_setup failed");
312 result
= ltt_trace_set_type(trace_name
, trace_type
);
314 ERR("ltt_trace_set_type failed");
318 else if(!strcmp(recvbuf
, "trace_alloc")) {
321 result
= ltt_trace_alloc(trace_name
);
323 ERR("ltt_trace_alloc failed");
327 else if(!strcmp(recvbuf
, "trace_start")) {
330 result
= ltt_trace_start(trace_name
);
332 ERR("ltt_trace_start failed");
336 else if(!strcmp(recvbuf
, "trace_stop")) {
339 result
= ltt_trace_stop(trace_name
);
341 ERR("ltt_trace_stop failed");
345 else if(!strcmp(recvbuf
, "trace_destroy")) {
347 DBG("trace destroy");
349 result
= ltt_trace_destroy(trace_name
);
351 ERR("ltt_trace_destroy failed");
355 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
356 struct ltt_trace_struct
*trace
;
357 char trace_name
[] = "auto";
363 channel_name
= nth_token(recvbuf
, 1);
364 if(channel_name
== NULL
) {
365 ERR("get_shmid: cannot parse channel");
370 trace
= _ltt_trace_find(trace_name
);
374 ERR("cannot find trace!");
378 for(i
=0; i
<trace
->nr_channels
; i
++) {
379 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
380 struct rchan_buf
*rbuf
= rchan
->buf
;
381 struct ltt_channel_struct
*ltt_channel
= (struct ltt_channel_struct
*)rchan
->private_data
;
383 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
386 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
387 DBG("the shmid for its buffer structure is %d", ltt_channel
->buf_shmid
);
388 asprintf(&reply
, "%d %d", rbuf
->shmid
, ltt_channel
->buf_shmid
);
390 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
392 ERR("listener: get_shmid: ustcomm_send_reply failed");
402 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
403 struct ltt_trace_struct
*trace
;
404 char trace_name
[] = "auto";
408 DBG("get_n_subbufs");
410 channel_name
= nth_token(recvbuf
, 1);
411 if(channel_name
== NULL
) {
412 ERR("get_n_subbufs: cannot parse channel");
417 trace
= _ltt_trace_find(trace_name
);
421 ERR("cannot find trace!");
425 for(i
=0; i
<trace
->nr_channels
; i
++) {
426 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
428 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
431 DBG("the n_subbufs for the requested channel is %zd", rchan
->n_subbufs
);
432 asprintf(&reply
, "%zd", rchan
->n_subbufs
);
434 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
436 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
446 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
447 struct ltt_trace_struct
*trace
;
448 char trace_name
[] = "auto";
452 DBG("get_subbuf_size");
454 channel_name
= nth_token(recvbuf
, 1);
455 if(channel_name
== NULL
) {
456 ERR("get_subbuf_size: cannot parse channel");
461 trace
= _ltt_trace_find(trace_name
);
465 ERR("cannot find trace!");
469 for(i
=0; i
<trace
->nr_channels
; i
++) {
470 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
472 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
475 DBG("the subbuf_size for the requested channel is %zd", rchan
->subbuf_size
);
476 asprintf(&reply
, "%zd", rchan
->subbuf_size
);
478 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
480 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
490 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
493 libfile
= nth_token(recvbuf
, 1);
495 DBG("load_probe_lib loading %s", libfile
);
497 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
498 struct ltt_trace_struct
*trace
;
499 char trace_name
[] = "auto";
505 channel_name
= nth_token(recvbuf
, 1);
506 if(channel_name
== NULL
) {
507 ERR("get_subbuf: cannot parse channel");
512 trace
= _ltt_trace_find(trace_name
);
516 ERR("cannot find trace!");
520 for(i
=0; i
<trace
->nr_channels
; i
++) {
521 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
523 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
524 struct rchan_buf
*rbuf
= rchan
->buf
;
525 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
526 struct blocked_consumer
*bc
;
528 bc
= (struct blocked_consumer
*) malloc(sizeof(struct blocked_consumer
));
530 ERR("malloc returned NULL");
533 bc
->fd_consumer
= src
.fd
;
534 bc
->fd_producer
= lttbuf
->data_ready_fd_read
;
538 bc
->server
= ustcomm_app
.server
;
540 list_add(&bc
->list
, &blocked_consumers
);
546 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
547 struct ltt_trace_struct
*trace
;
548 char trace_name
[] = "auto";
552 char *consumed_old_str
;
557 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
558 if(channel_name
== NULL
) {
559 ERR("put_subbuf_size: cannot parse channel");
563 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
564 if(consumed_old_str
== NULL
) {
565 ERR("put_subbuf: cannot parse consumed_old");
568 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
569 if(*endptr
!= '\0') {
570 ERR("put_subbuf: invalid value for consumed_old");
575 trace
= _ltt_trace_find(trace_name
);
579 ERR("cannot find trace!");
583 for(i
=0; i
<trace
->nr_channels
; i
++) {
584 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
586 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
587 struct rchan_buf
*rbuf
= rchan
->buf
;
588 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
592 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
594 WARN("ltt_do_put_subbuf: error (subbuf=%s)", channel_name
);
595 asprintf(&reply
, "%s", "ERROR");
598 DBG("ltt_do_put_subbuf: success (subbuf=%s)", channel_name
);
599 asprintf(&reply
, "%s", "OK");
602 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
604 ERR("listener: put_subbuf: ustcomm_send_reply failed");
615 free(consumed_old_str
);
617 else if(nth_token_is(recvbuf
, "enable_marker", 0) == 1) {
618 char *channel_slash_name
= nth_token(recvbuf
, 1);
619 char channel_name
[256]="";
620 char marker_name
[256]="";
622 result
= sscanf(channel_slash_name
, "%255[^/]/%255s", channel_name
, marker_name
);
624 if(channel_name
== NULL
|| marker_name
== NULL
) {
625 WARN("invalid marker name");
628 printf("%s %s\n", channel_name
, marker_name
);
630 result
= ltt_marker_connect(channel_name
, marker_name
, "default");
632 WARN("could not enable marker; channel=%s, name=%s", channel_name
, marker_name
);
635 else if(nth_token_is(recvbuf
, "disable_marker", 0) == 1) {
636 char *channel_slash_name
= nth_token(recvbuf
, 1);
640 result
= sscanf(channel_slash_name
, "%a[^/]/%as", &channel_name
, &marker_name
);
642 if(marker_name
== NULL
) {
644 printf("%s %s\n", channel_name
, marker_name
);
646 result
= ltt_marker_disconnect(channel_name
, marker_name
, "default");
648 WARN("could not disable marker; channel=%s, name=%s", channel_name
, marker_name
);
651 // else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
652 // struct ltt_trace_struct *trace;
653 // char trace_name[] = "auto";
655 // char *channel_name;
657 // DBG("get_notifications");
659 // channel_name = strdup_malloc(nth_token(recvbuf, 1));
660 // if(channel_name == NULL) {
661 // ERR("put_subbuf_size: cannot parse channel");
665 // ltt_lock_traces();
666 // trace = _ltt_trace_find(trace_name);
667 // ltt_unlock_traces();
669 // if(trace == NULL) {
670 // ERR("cannot find trace!");
674 // for(i=0; i<trace->nr_channels; i++) {
675 // struct rchan *rchan = trace->channels[i].trans_channel_data;
678 // if(!strcmp(trace->channels[i].channel_name, channel_name)) {
679 // struct rchan_buf *rbuf = rchan->buf;
680 // struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
682 // result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
683 // if(result == -1) {
684 // ERR("ustcomm_app_detach_client failed");
688 // lttbuf->wake_consumer_arg = (void *) fd;
692 // lttbuf->call_wake_consumer = 1;
698 // free(channel_name);
701 ERR("unable to parse message: %s", recvbuf
);
709 int have_listener
= 0;
711 void create_listener(void)
714 static char listener_stack
[16384];
723 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
730 pthread_create(&thread
, NULL
, listener_main
, NULL
);
736 /* The signal handler itself. Signals must be setup so there cannot be
739 void sighandler(int sig
)
741 static char have_listener
= 0;
750 /* Called by the app signal handler to chain it to us. */
752 void chain_signal(void)
754 sighandler(USTSIGNAL
);
757 static int init_socket(void)
759 return ustcomm_init_app(getpid(), &ustcomm_app
);
762 /* FIXME: reenable this to delete socket file. */
765 static void destroy_socket(void)
769 if(mysocketfile
[0] == '\0')
772 result
= unlink(mysocketfile
);
779 static int init_signal_handler(void)
781 /* Attempt to handler SIGIO. If the main program wants to
782 * handle it, fine, it'll override us. They it'll have to
783 * use the chaining function.
787 struct sigaction act
;
789 result
= sigemptyset(&act
.sa_mask
);
791 PERROR("sigemptyset");
795 act
.sa_handler
= sighandler
;
796 act
.sa_flags
= SA_RESTART
;
798 /* Only defer ourselves. Also, try to restart interrupted
799 * syscalls to disturb the traced program as little as possible.
801 result
= sigaction(SIGIO
, &act
, NULL
);
810 #define AUTOPROBE_DISABLED 0
811 #define AUTOPROBE_ENABLE_ALL 1
812 #define AUTOPROBE_ENABLE_REGEX 2
813 static int autoprobe_method
= AUTOPROBE_DISABLED
;
814 static regex_t autoprobe_regex
;
816 static void auto_probe_connect(struct marker
*m
)
820 char* concat_name
= NULL
;
821 const char *probe_name
= "default";
823 if(autoprobe_method
== AUTOPROBE_DISABLED
) {
826 else if(autoprobe_method
== AUTOPROBE_ENABLE_REGEX
) {
827 result
= asprintf(&concat_name
, "%s/%s", m
->channel
, m
->name
);
829 ERR("auto_probe_connect: asprintf failed (marker %s/%s)",
830 m
->channel
, m
->name
);
833 if (regexec(&autoprobe_regex
, concat_name
, 0, NULL
, 0)) {
840 result
= ltt_marker_connect(m
->channel
, m
->name
, probe_name
);
841 if(result
&& result
!= -EEXIST
)
842 ERR("ltt_marker_connect (marker = %s/%s, errno = %d)", m
->channel
, m
->name
, -result
);
844 DBG("auto connected marker %s %s to probe default", m
->channel
, m
->name
);
848 static void __attribute__((constructor(1000))) init()
851 char* autoprobe_val
= NULL
;
853 /* Initialize RCU in case the constructor order is not good. */
856 /* It is important to do this before events start to be generated. */
857 ust_register_thread();
859 DBG("Tracectl constructor");
861 /* Must create socket before signal handler to prevent races.
863 result
= init_socket();
865 ERR("init_socket error");
868 result
= init_signal_handler();
870 ERR("init_signal_handler error");
874 autoprobe_val
= getenv("UST_AUTOPROBE");
876 struct marker_iter iter
;
878 DBG("Autoprobe enabled.");
880 /* Ensure markers are initialized */
883 /* Ensure marker control is initialized, for the probe */
884 init_marker_control();
886 /* first, set the callback that will connect the
887 * probe on new markers
889 if(autoprobe_val
[0] == '/') {
890 result
= regcomp(&autoprobe_regex
, autoprobe_val
+1, 0);
894 regerror(result
, &autoprobe_regex
, regexerr
, sizeof(regexerr
));
895 ERR("cannot parse regex %s (%s), will ignore UST_AUTOPROBE", autoprobe_val
, regexerr
);
896 /* don't crash the application just for this */
899 autoprobe_method
= AUTOPROBE_ENABLE_REGEX
;
903 /* just enable all instrumentation */
904 autoprobe_method
= AUTOPROBE_ENABLE_ALL
;
907 marker_set_new_marker_cb(auto_probe_connect
);
909 /* Now, connect the probes that were already registered. */
910 marker_iter_reset(&iter
);
911 marker_iter_start(&iter
);
913 DBG("now iterating on markers already registered");
915 DBG("now iterating on marker %s", iter
.marker
->name
);
916 auto_probe_connect(iter
.marker
);
917 marker_iter_next(&iter
);
921 if(getenv("UST_TRACE")) {
922 char trace_name
[] = "auto";
923 char trace_type
[] = "ustrelay";
925 DBG("starting early tracing");
927 /* Ensure marker control is initialized */
928 init_marker_control();
930 /* Ensure relay is initialized */
931 init_ustrelay_transport();
933 /* Ensure markers are initialized */
937 ltt_channels_register("ust");
939 result
= ltt_trace_setup(trace_name
);
941 ERR("ltt_trace_setup failed");
945 result
= ltt_trace_set_type(trace_name
, trace_type
);
947 ERR("ltt_trace_set_type failed");
951 result
= ltt_trace_alloc(trace_name
);
953 ERR("ltt_trace_alloc failed");
957 result
= ltt_trace_start(trace_name
);
959 ERR("ltt_trace_start failed");
962 inform_consumer_daemon();
968 /* should decrementally destroy stuff if error */
972 /* This is only called if we terminate normally, not with an unhandled signal,
973 * so we cannot rely on it. */
975 /* This destructor probably isn't needed, because ustd can do crash recovery. */
977 static void __attribute__((destructor
)) fini()
981 /* if trace running, finish it */
983 DBG("destructor stopping traces");
985 result
= ltt_trace_stop("auto");
987 ERR("ltt_trace_stop error");
990 result
= ltt_trace_destroy("auto");
992 ERR("ltt_trace_destroy error");
1000 static int trace_recording(void)
1003 struct ltt_trace_struct
*trace
;
1007 list_for_each_entry(trace
, <t_traces
.head
, list
) {
1014 ltt_unlock_traces();
1019 static int have_consumer(void)
1021 return !list_empty(&blocked_consumers
);
1024 /* This destructor keeps the process alive for a few seconds in order
1025 * to leave time to ustd to consume its buffers.
1028 int restarting_sleep(int secs
)
1037 result
= nanosleep(&tv
, &tv
);
1038 } while(result
== -1 && errno
== EINTR
);
1043 static void __attribute__((destructor
)) keepalive()
1045 // struct ustcomm_ustd ustd;
1049 // result = sigemptyset(&sigset);
1050 // if(result == -1) {
1051 // perror("sigemptyset");
1054 // result = sigaddset(&sigset, SIGIO);
1055 // if(result == -1) {
1056 // perror("sigaddset");
1059 // result = sigprocmask(SIG_BLOCK, &sigset, NULL);
1060 // if(result == -1) {
1061 // perror("sigprocmask");
1065 // if(trace_recording()) {
1066 // if(!have_consumer()) {
1067 // /* Request listener creation. We've blocked SIGIO's in
1068 // * order to not interrupt sleep(), so we will miss the
1069 // * one sent by the daemon and therefore won't create
1070 // * the listener automatically.
1072 // create_listener();
1074 printf("Keeping process alive for consumer daemon...\n");
1075 restarting_sleep(3);
1076 printf("Finally dying...\n");
1080 // result = sigprocmask(SIG_UNBLOCK, &sigset, NULL);
1081 // if(result == -1) {
1082 // perror("sigprocmask");
1088 /* Notify ust that there was a fork. This needs to be called inside
1089 * the new process, anytime a process whose memory is not shared with
1090 * the parent is created. If this function is not called, the events
1091 * of the new process will not be collected.
1096 DBG("ust: forking");
1097 ltt_trace_stop("auto");
1098 ltt_trace_destroy("auto");
1099 ltt_trace_alloc("auto");
1100 ltt_trace_start("auto");
1104 ustcomm_request_consumer(getpid(), "metadata");
1105 ustcomm_request_consumer(getpid(), "ust");