5 #include <sys/socket.h>
15 #include "relay.h" /* FIXME: remove */
19 #define USTSIGNAL SIGIO
21 #define MAX_MSG_SIZE (100)
23 #define MSG_REGISTER_NOTIF 2
25 char consumer_stack
[10000];
27 struct list_head blocked_consumers
= LIST_HEAD_INIT(blocked_consumers
);
29 static struct ustcomm_app ustcomm_app
;
31 struct tracecmd
{ /* no padding */
36 //struct listener_arg {
41 /* size: the size of all the fields except size itself */
44 /* Only the necessary part of the payload is transferred. It
45 * may even be none of it.
50 struct consumer_channel
{
52 struct ltt_channel_struct
*chan
;
55 struct blocked_consumer
{
60 /* args to ustcomm_send_reply */
61 struct ustcomm_server server
;
62 struct ustcomm_source src
;
64 /* args to ltt_do_get_subbuf */
65 struct rchan_buf
*rbuf
;
66 struct ltt_channel_buf_struct
*lttbuf
;
68 struct list_head list
;
71 static void print_markers(FILE *fp
)
73 struct marker_iter iter
;
76 marker_iter_reset(&iter
);
77 marker_iter_start(&iter
);
80 fprintf(fp
, "marker: %s_%s %d \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, (int)imv_read(iter
.marker
->state
), iter
.marker
->format
);
81 marker_iter_next(&iter
);
86 void do_command(struct tracecmd
*cmd
)
90 void receive_commands()
100 /* FIXME: fd_notif should probably be protected by a spinlock */
105 msg
.type
= MSG_NOTIF
;
106 msg
.size
= sizeof(msg
.type
);
108 /* FIXME: don't block here */
109 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
116 static int inform_consumer_daemon(void)
118 ustcomm_request_consumer(getpid(), "metadata");
119 ustcomm_request_consumer(getpid(), "ust");
122 void process_blocked_consumers(void)
126 struct blocked_consumer
*bc
;
131 list_for_each_entry(bc
, &blocked_consumers
, list
) {
135 fds
= (struct pollfd
*) malloc(n_fds
* sizeof(struct pollfd
));
137 ERR("malloc returned NULL");
141 list_for_each_entry(bc
, &blocked_consumers
, list
) {
142 fds
[idx
].fd
= bc
->fd_producer
;
143 fds
[idx
].events
= POLLIN
;
144 bc
->tmp_poll_idx
= idx
;
148 result
= poll(fds
, n_fds
, 0);
154 list_for_each_entry(bc
, &blocked_consumers
, list
) {
155 if(fds
[bc
->tmp_poll_idx
].revents
) {
156 long consumed_old
= 0;
159 result
= read(bc
->fd_producer
, &inbuf
, 1);
167 close(bc
->fd_producer
);
169 __list_del(bc
->list
.prev
, bc
->list
.next
);
171 result
= ustcomm_send_reply(&bc
->server
, "END", &bc
->src
);
173 ERR("ustcomm_send_reply failed");
180 result
= ltt_do_get_subbuf(bc
->rbuf
, bc
->lttbuf
, &consumed_old
);
181 if(result
== -EAGAIN
) {
182 WARN("missed buffer?");
185 else if(result
< 0) {
186 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
188 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
189 result
= ustcomm_send_reply(&bc
->server
, reply
, &bc
->src
);
191 ERR("ustcomm_send_reply failed");
197 __list_del(bc
->list
.prev
, bc
->list
.next
);
203 int listener_main(void *p
)
211 struct sockaddr_un addr
;
212 socklen_t addrlen
= sizeof(addr
);
213 char trace_name
[] = "auto";
214 char trace_type
[] = "ustrelay";
217 struct ustcomm_source src
;
219 process_blocked_consumers();
221 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, 5);
223 WARN("error in ustcomm_app_recv_message");
226 else if(result
== 0) {
231 DBG("received a message! it's: %s\n", recvbuf
);
232 len
= strlen(recvbuf
);
234 if(!strcmp(recvbuf
, "print_markers")) {
235 print_markers(stderr
);
237 else if(!strcmp(recvbuf
, "list_markers")) {
242 fp
= open_memstream(&ptr
, &size
);
246 result
= ustcomm_send_reply(&ustcomm_app
.server
, ptr
, &src
);
250 else if(!strcmp(recvbuf
, "start")) {
251 /* start is an operation that setups the trace, allocates it and starts it */
252 result
= ltt_trace_setup(trace_name
);
254 ERR("ltt_trace_setup failed");
258 result
= ltt_trace_set_type(trace_name
, trace_type
);
260 ERR("ltt_trace_set_type failed");
264 result
= ltt_trace_alloc(trace_name
);
266 ERR("ltt_trace_alloc failed");
270 inform_consumer_daemon();
272 result
= ltt_trace_start(trace_name
);
274 ERR("ltt_trace_start failed");
278 else if(!strcmp(recvbuf
, "trace_setup")) {
281 result
= ltt_trace_setup(trace_name
);
283 ERR("ltt_trace_setup failed");
287 result
= ltt_trace_set_type(trace_name
, trace_type
);
289 ERR("ltt_trace_set_type failed");
293 else if(!strcmp(recvbuf
, "trace_alloc")) {
296 result
= ltt_trace_alloc(trace_name
);
298 ERR("ltt_trace_alloc failed");
302 else if(!strcmp(recvbuf
, "trace_start")) {
305 result
= ltt_trace_start(trace_name
);
307 ERR("ltt_trace_start failed");
311 else if(!strcmp(recvbuf
, "trace_stop")) {
314 result
= ltt_trace_stop(trace_name
);
316 ERR("ltt_trace_stop failed");
320 else if(!strcmp(recvbuf
, "trace_destroy")) {
322 DBG("trace destroy");
324 result
= ltt_trace_destroy(trace_name
);
326 ERR("ltt_trace_destroy failed");
330 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
331 struct ltt_trace_struct
*trace
;
332 char trace_name
[] = "auto";
338 channel_name
= nth_token(recvbuf
, 1);
339 if(channel_name
== NULL
) {
340 ERR("get_shmid: cannot parse channel");
345 trace
= _ltt_trace_find(trace_name
);
349 CPRINTF("cannot find trace!");
353 for(i
=0; i
<trace
->nr_channels
; i
++) {
354 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
355 struct rchan_buf
*rbuf
= rchan
->buf
;
356 struct ltt_channel_struct
*ltt_channel
= (struct ltt_channel_struct
*)rchan
->private_data
;
357 struct ltt_channel_buf_struct
*ltt_buf
= ltt_channel
->buf
;
359 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
362 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
363 DBG("the shmid for its buffer structure is %d", ltt_channel
->buf_shmid
);
364 asprintf(&reply
, "%d %d", rbuf
->shmid
, ltt_channel
->buf_shmid
);
366 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
368 ERR("listener: get_shmid: ustcomm_send_reply failed");
378 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
379 struct ltt_trace_struct
*trace
;
380 char trace_name
[] = "auto";
384 DBG("get_n_subbufs");
386 channel_name
= nth_token(recvbuf
, 1);
387 if(channel_name
== NULL
) {
388 ERR("get_n_subbufs: cannot parse channel");
393 trace
= _ltt_trace_find(trace_name
);
397 CPRINTF("cannot find trace!");
401 for(i
=0; i
<trace
->nr_channels
; i
++) {
402 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
404 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
407 DBG("the n_subbufs for the requested channel is %d", rchan
->n_subbufs
);
408 asprintf(&reply
, "%d", rchan
->n_subbufs
);
410 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
412 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
422 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
423 struct ltt_trace_struct
*trace
;
424 char trace_name
[] = "auto";
428 DBG("get_subbuf_size");
430 channel_name
= nth_token(recvbuf
, 1);
431 if(channel_name
== NULL
) {
432 ERR("get_subbuf_size: cannot parse channel");
437 trace
= _ltt_trace_find(trace_name
);
441 CPRINTF("cannot find trace!");
445 for(i
=0; i
<trace
->nr_channels
; i
++) {
446 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
448 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
451 DBG("the subbuf_size for the requested channel is %d", rchan
->subbuf_size
);
452 asprintf(&reply
, "%d", rchan
->subbuf_size
);
454 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
456 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
466 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
469 libfile
= nth_token(recvbuf
, 1);
471 DBG("load_probe_lib loading %s", libfile
);
473 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
474 struct ltt_trace_struct
*trace
;
475 char trace_name
[] = "auto";
481 channel_name
= nth_token(recvbuf
, 1);
482 if(channel_name
== NULL
) {
483 ERR("get_subbuf: cannot parse channel");
488 trace
= _ltt_trace_find(trace_name
);
492 CPRINTF("cannot find trace!");
496 for(i
=0; i
<trace
->nr_channels
; i
++) {
497 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
499 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
500 struct rchan_buf
*rbuf
= rchan
->buf
;
501 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
505 struct blocked_consumer
*bc
;
507 bc
= (struct blocked_consumer
*) malloc(sizeof(struct blocked_consumer
));
509 ERR("malloc returned NULL");
512 bc
->fd_consumer
= src
.fd
;
513 bc
->fd_producer
= lttbuf
->data_ready_fd_read
;
517 bc
->server
= ustcomm_app
.server
;
519 list_add(&bc
->list
, &blocked_consumers
);
525 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
526 struct ltt_trace_struct
*trace
;
527 char trace_name
[] = "auto";
531 char *consumed_old_str
;
536 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
537 if(channel_name
== NULL
) {
538 ERR("put_subbuf_size: cannot parse channel");
542 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
543 if(consumed_old_str
== NULL
) {
544 ERR("put_subbuf: cannot parse consumed_old");
547 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
548 if(*endptr
!= '\0') {
549 ERR("put_subbuf: invalid value for consumed_old");
554 trace
= _ltt_trace_find(trace_name
);
558 CPRINTF("cannot find trace!");
562 for(i
=0; i
<trace
->nr_channels
; i
++) {
563 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
565 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
566 struct rchan_buf
*rbuf
= rchan
->buf
;
567 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
571 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
573 WARN("ltt_do_put_subbuf: error");
576 DBG("ltt_do_put_subbuf: success");
578 asprintf(&reply
, "%s", "OK", consumed_old
);
580 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
582 ERR("listener: put_subbuf: ustcomm_send_reply failed");
593 free(consumed_old_str
);
595 else if(nth_token_is(recvbuf
, "enable_marker", 0) == 1) {
596 char *channel_slash_name
= nth_token(recvbuf
, 1);
597 char channel_name
[256]="";
598 char marker_name
[256]="";
599 struct marker_iter iter
;
601 result
= sscanf(channel_slash_name
, "%255[^/]/%255s", channel_name
, marker_name
);
603 if(channel_name
== NULL
|| marker_name
== NULL
) {
604 WARN("invalid marker name");
607 printf("%s %s\n", channel_name
, marker_name
);
609 result
= ltt_marker_connect(channel_name
, marker_name
, "default");
611 WARN("could not enable marker; channel=%s, name=%s", channel_name
, marker_name
);
614 else if(nth_token_is(recvbuf
, "disable_marker", 0) == 1) {
615 char *channel_slash_name
= nth_token(recvbuf
, 1);
618 struct marker_iter iter
;
620 result
= sscanf(channel_slash_name
, "%a[^/]/%as", &channel_name
, &marker_name
);
622 if(marker_name
== NULL
) {
624 printf("%s %s\n", channel_name
, marker_name
);
626 result
= ltt_marker_disconnect(channel_name
, marker_name
, "default");
628 WARN("could not disable marker; channel=%s, name=%s", channel_name
, marker_name
);
631 // else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
632 // struct ltt_trace_struct *trace;
633 // char trace_name[] = "auto";
635 // char *channel_name;
637 // DBG("get_notifications");
639 // channel_name = strdup_malloc(nth_token(recvbuf, 1));
640 // if(channel_name == NULL) {
641 // ERR("put_subbuf_size: cannot parse channel");
645 // ltt_lock_traces();
646 // trace = _ltt_trace_find(trace_name);
647 // ltt_unlock_traces();
649 // if(trace == NULL) {
650 // CPRINTF("cannot find trace!");
654 // for(i=0; i<trace->nr_channels; i++) {
655 // struct rchan *rchan = trace->channels[i].trans_channel_data;
658 // if(!strcmp(trace->channels[i].channel_name, channel_name)) {
659 // struct rchan_buf *rbuf = rchan->buf;
660 // struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
662 // result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
663 // if(result == -1) {
664 // ERR("ustcomm_app_detach_client failed");
668 // lttbuf->wake_consumer_arg = (void *) fd;
672 // lttbuf->call_wake_consumer = 1;
678 // free(channel_name);
681 ERR("unable to parse message: %s", recvbuf
);
689 static char listener_stack
[16384];
691 void create_listener(void)
694 static char listener_stack
[16384];
695 //char *listener_stack = malloc(16384);
698 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
705 pthread_create(&thread
, NULL
, listener_main
, NULL
);
709 /* The signal handler itself. Signals must be setup so there cannot be
712 void sighandler(int sig
)
714 static char have_listener
= 0;
723 /* Called by the app signal handler to chain it to us. */
725 void chain_signal(void)
727 sighandler(USTSIGNAL
);
730 static int init_socket(void)
732 return ustcomm_init_app(getpid(), &ustcomm_app
);
735 static void destroy_socket(void)
739 // if(mysocketfile[0] == '\0')
742 // result = unlink(mysocketfile);
743 // if(result == -1) {
748 static int init_signal_handler(void)
750 /* Attempt to handler SIGIO. If the main program wants to
751 * handle it, fine, it'll override us. They it'll have to
752 * use the chaining function.
756 struct sigaction act
;
758 result
= sigemptyset(&act
.sa_mask
);
760 PERROR("sigemptyset");
764 act
.sa_handler
= sighandler
;
765 act
.sa_flags
= SA_RESTART
;
767 /* Only defer ourselves. Also, try to restart interrupted
768 * syscalls to disturb the traced program as little as possible.
770 result
= sigaction(SIGIO
, &act
, NULL
);
779 static void auto_probe_connect(struct marker
*m
)
783 result
= ltt_marker_connect(m
->channel
, m
->name
, "default");
785 ERR("ltt_marker_connect");
787 DBG("just auto connected marker %s %s to probe default", m
->channel
, m
->name
);
790 static void __attribute__((constructor(101))) init0()
792 DBG("UST_AUTOPROBE constructor");
793 if(getenv("UST_AUTOPROBE")) {
794 marker_set_new_marker_cb(auto_probe_connect
);
798 static void fini(void);
800 static void __attribute__((constructor(1000))) init()
804 DBG("UST_TRACE constructor");
806 /* Must create socket before signal handler to prevent races.
808 result
= init_socket();
810 ERR("init_socket error");
813 result
= init_signal_handler();
815 ERR("init_signal_handler error");
819 if(getenv("UST_TRACE")) {
820 char trace_name
[] = "auto";
821 char trace_type
[] = "ustrelay";
823 DBG("starting early tracing");
825 /* Ensure marker control is initialized */
826 init_marker_control();
828 /* Ensure relay is initialized */
829 init_ustrelay_transport();
831 /* Ensure markers are initialized */
835 ltt_channels_register("ust");
837 result
= ltt_trace_setup(trace_name
);
839 ERR("ltt_trace_setup failed");
843 result
= ltt_trace_set_type(trace_name
, trace_type
);
845 ERR("ltt_trace_set_type failed");
849 result
= ltt_trace_alloc(trace_name
);
851 ERR("ltt_trace_alloc failed");
855 result
= ltt_trace_start(trace_name
);
857 ERR("ltt_trace_start failed");
860 inform_consumer_daemon();
866 /* should decrementally destroy stuff if error */
870 /* This is only called if we terminate normally, not with an unhandled signal,
871 * so we cannot rely on it. */
873 static void __attribute__((destructor
)) fini()
877 /* if trace running, finish it */
879 DBG("destructor stopping traces");
881 result
= ltt_trace_stop("auto");
883 ERR("ltt_trace_stop error");
886 result
= ltt_trace_destroy("auto");
888 ERR("ltt_trace_destroy error");
891 /* FIXME: wait for the consumer to be done */
892 //DBG("waiting 5 sec for consume");