6 #include <sys/socket.h>
16 #include "relay.h" /* FIXME: remove */
17 #include "marker-control.h"
21 #define USTSIGNAL SIGIO
23 #define MAX_MSG_SIZE (100)
25 #define MSG_REGISTER_NOTIF 2
27 char consumer_stack
[10000];
29 struct list_head blocked_consumers
= LIST_HEAD_INIT(blocked_consumers
);
31 static struct ustcomm_app ustcomm_app
;
33 struct tracecmd
{ /* no padding */
38 //struct listener_arg {
43 /* size: the size of all the fields except size itself */
46 /* Only the necessary part of the payload is transferred. It
47 * may even be none of it.
52 struct consumer_channel
{
54 struct ltt_channel_struct
*chan
;
57 struct blocked_consumer
{
62 /* args to ustcomm_send_reply */
63 struct ustcomm_server server
;
64 struct ustcomm_source src
;
66 /* args to ltt_do_get_subbuf */
67 struct rchan_buf
*rbuf
;
68 struct ltt_channel_buf_struct
*lttbuf
;
70 struct list_head list
;
73 static void print_markers(FILE *fp
)
75 struct marker_iter iter
;
78 marker_iter_reset(&iter
);
79 marker_iter_start(&iter
);
82 fprintf(fp
, "marker: %s_%s %d \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, (int)imv_read(iter
.marker
->state
), iter
.marker
->format
);
83 marker_iter_next(&iter
);
88 void do_command(struct tracecmd
*cmd
)
92 void receive_commands()
100 struct trctl_msg msg
;
102 /* FIXME: fd_notif should probably be protected by a spinlock */
107 msg
.type
= MSG_NOTIF
;
108 msg
.size
= sizeof(msg
.type
);
110 /* FIXME: don't block here */
111 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
118 static void inform_consumer_daemon(void)
120 ustcomm_request_consumer(getpid(), "metadata");
121 ustcomm_request_consumer(getpid(), "ust");
124 void process_blocked_consumers(void)
128 struct blocked_consumer
*bc
;
133 list_for_each_entry(bc
, &blocked_consumers
, list
) {
137 fds
= (struct pollfd
*) malloc(n_fds
* sizeof(struct pollfd
));
139 ERR("malloc returned NULL");
143 list_for_each_entry(bc
, &blocked_consumers
, list
) {
144 fds
[idx
].fd
= bc
->fd_producer
;
145 fds
[idx
].events
= POLLIN
;
146 bc
->tmp_poll_idx
= idx
;
150 result
= poll(fds
, n_fds
, 0);
156 list_for_each_entry(bc
, &blocked_consumers
, list
) {
157 if(fds
[bc
->tmp_poll_idx
].revents
) {
158 long consumed_old
= 0;
161 result
= read(bc
->fd_producer
, &inbuf
, 1);
169 close(bc
->fd_producer
);
173 result
= ustcomm_send_reply(&bc
->server
, "END", &bc
->src
);
175 ERR("ustcomm_send_reply failed");
182 result
= ltt_do_get_subbuf(bc
->rbuf
, bc
->lttbuf
, &consumed_old
);
183 if(result
== -EAGAIN
) {
184 WARN("missed buffer?");
187 else if(result
< 0) {
188 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
190 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
191 result
= ustcomm_send_reply(&bc
->server
, reply
, &bc
->src
);
193 ERR("ustcomm_send_reply failed");
205 void *listener_main(void *p
)
212 char trace_name
[] = "auto";
213 char trace_type
[] = "ustrelay";
216 struct ustcomm_source src
;
218 process_blocked_consumers();
220 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, 5);
222 WARN("error in ustcomm_app_recv_message");
225 else if(result
== 0) {
230 DBG("received a message! it's: %s\n", recvbuf
);
231 len
= strlen(recvbuf
);
233 if(!strcmp(recvbuf
, "print_markers")) {
234 print_markers(stderr
);
236 else if(!strcmp(recvbuf
, "list_markers")) {
241 fp
= open_memstream(&ptr
, &size
);
245 result
= ustcomm_send_reply(&ustcomm_app
.server
, ptr
, &src
);
249 else if(!strcmp(recvbuf
, "start")) {
250 /* start is an operation that setups the trace, allocates it and starts it */
251 result
= ltt_trace_setup(trace_name
);
253 ERR("ltt_trace_setup failed");
257 result
= ltt_trace_set_type(trace_name
, trace_type
);
259 ERR("ltt_trace_set_type failed");
263 result
= ltt_trace_alloc(trace_name
);
265 ERR("ltt_trace_alloc failed");
269 inform_consumer_daemon();
271 result
= ltt_trace_start(trace_name
);
273 ERR("ltt_trace_start failed");
277 else if(!strcmp(recvbuf
, "trace_setup")) {
280 result
= ltt_trace_setup(trace_name
);
282 ERR("ltt_trace_setup failed");
286 result
= ltt_trace_set_type(trace_name
, trace_type
);
288 ERR("ltt_trace_set_type failed");
292 else if(!strcmp(recvbuf
, "trace_alloc")) {
295 result
= ltt_trace_alloc(trace_name
);
297 ERR("ltt_trace_alloc failed");
301 else if(!strcmp(recvbuf
, "trace_start")) {
304 result
= ltt_trace_start(trace_name
);
306 ERR("ltt_trace_start failed");
310 else if(!strcmp(recvbuf
, "trace_stop")) {
313 result
= ltt_trace_stop(trace_name
);
315 ERR("ltt_trace_stop failed");
319 else if(!strcmp(recvbuf
, "trace_destroy")) {
321 DBG("trace destroy");
323 result
= ltt_trace_destroy(trace_name
);
325 ERR("ltt_trace_destroy failed");
329 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
330 struct ltt_trace_struct
*trace
;
331 char trace_name
[] = "auto";
337 channel_name
= nth_token(recvbuf
, 1);
338 if(channel_name
== NULL
) {
339 ERR("get_shmid: cannot parse channel");
344 trace
= _ltt_trace_find(trace_name
);
348 ERR("cannot find trace!");
352 for(i
=0; i
<trace
->nr_channels
; i
++) {
353 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
354 struct rchan_buf
*rbuf
= rchan
->buf
;
355 struct ltt_channel_struct
*ltt_channel
= (struct ltt_channel_struct
*)rchan
->private_data
;
357 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
360 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
361 DBG("the shmid for its buffer structure is %d", ltt_channel
->buf_shmid
);
362 asprintf(&reply
, "%d %d", rbuf
->shmid
, ltt_channel
->buf_shmid
);
364 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
366 ERR("listener: get_shmid: ustcomm_send_reply failed");
376 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
377 struct ltt_trace_struct
*trace
;
378 char trace_name
[] = "auto";
382 DBG("get_n_subbufs");
384 channel_name
= nth_token(recvbuf
, 1);
385 if(channel_name
== NULL
) {
386 ERR("get_n_subbufs: cannot parse channel");
391 trace
= _ltt_trace_find(trace_name
);
395 ERR("cannot find trace!");
399 for(i
=0; i
<trace
->nr_channels
; i
++) {
400 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
402 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
405 DBG("the n_subbufs for the requested channel is %zd", rchan
->n_subbufs
);
406 asprintf(&reply
, "%zd", rchan
->n_subbufs
);
408 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
410 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
420 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
421 struct ltt_trace_struct
*trace
;
422 char trace_name
[] = "auto";
426 DBG("get_subbuf_size");
428 channel_name
= nth_token(recvbuf
, 1);
429 if(channel_name
== NULL
) {
430 ERR("get_subbuf_size: cannot parse channel");
435 trace
= _ltt_trace_find(trace_name
);
439 ERR("cannot find trace!");
443 for(i
=0; i
<trace
->nr_channels
; i
++) {
444 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
446 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
449 DBG("the subbuf_size for the requested channel is %zd", rchan
->subbuf_size
);
450 asprintf(&reply
, "%zd", rchan
->subbuf_size
);
452 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
454 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
464 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
467 libfile
= nth_token(recvbuf
, 1);
469 DBG("load_probe_lib loading %s", libfile
);
471 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
472 struct ltt_trace_struct
*trace
;
473 char trace_name
[] = "auto";
479 channel_name
= nth_token(recvbuf
, 1);
480 if(channel_name
== NULL
) {
481 ERR("get_subbuf: cannot parse channel");
486 trace
= _ltt_trace_find(trace_name
);
490 ERR("cannot find trace!");
494 for(i
=0; i
<trace
->nr_channels
; i
++) {
495 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
497 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
498 struct rchan_buf
*rbuf
= rchan
->buf
;
499 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
500 struct blocked_consumer
*bc
;
502 bc
= (struct blocked_consumer
*) malloc(sizeof(struct blocked_consumer
));
504 ERR("malloc returned NULL");
507 bc
->fd_consumer
= src
.fd
;
508 bc
->fd_producer
= lttbuf
->data_ready_fd_read
;
512 bc
->server
= ustcomm_app
.server
;
514 list_add(&bc
->list
, &blocked_consumers
);
520 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
521 struct ltt_trace_struct
*trace
;
522 char trace_name
[] = "auto";
526 char *consumed_old_str
;
531 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
532 if(channel_name
== NULL
) {
533 ERR("put_subbuf_size: cannot parse channel");
537 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
538 if(consumed_old_str
== NULL
) {
539 ERR("put_subbuf: cannot parse consumed_old");
542 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
543 if(*endptr
!= '\0') {
544 ERR("put_subbuf: invalid value for consumed_old");
549 trace
= _ltt_trace_find(trace_name
);
553 ERR("cannot find trace!");
557 for(i
=0; i
<trace
->nr_channels
; i
++) {
558 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
560 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
561 struct rchan_buf
*rbuf
= rchan
->buf
;
562 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
566 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
568 WARN("ltt_do_put_subbuf: error (subbuf=%s)", channel_name
);
569 asprintf(&reply
, "%s", "ERROR");
572 DBG("ltt_do_put_subbuf: success (subbuf=%s)", channel_name
);
573 asprintf(&reply
, "%s", "OK");
576 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
578 ERR("listener: put_subbuf: ustcomm_send_reply failed");
589 free(consumed_old_str
);
591 else if(nth_token_is(recvbuf
, "enable_marker", 0) == 1) {
592 char *channel_slash_name
= nth_token(recvbuf
, 1);
593 char channel_name
[256]="";
594 char marker_name
[256]="";
596 result
= sscanf(channel_slash_name
, "%255[^/]/%255s", channel_name
, marker_name
);
598 if(channel_name
== NULL
|| marker_name
== NULL
) {
599 WARN("invalid marker name");
602 printf("%s %s\n", channel_name
, marker_name
);
604 result
= ltt_marker_connect(channel_name
, marker_name
, "default");
606 WARN("could not enable marker; channel=%s, name=%s", channel_name
, marker_name
);
609 else if(nth_token_is(recvbuf
, "disable_marker", 0) == 1) {
610 char *channel_slash_name
= nth_token(recvbuf
, 1);
614 result
= sscanf(channel_slash_name
, "%a[^/]/%as", &channel_name
, &marker_name
);
616 if(marker_name
== NULL
) {
618 printf("%s %s\n", channel_name
, marker_name
);
620 result
= ltt_marker_disconnect(channel_name
, marker_name
, "default");
622 WARN("could not disable marker; channel=%s, name=%s", channel_name
, marker_name
);
625 // else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
626 // struct ltt_trace_struct *trace;
627 // char trace_name[] = "auto";
629 // char *channel_name;
631 // DBG("get_notifications");
633 // channel_name = strdup_malloc(nth_token(recvbuf, 1));
634 // if(channel_name == NULL) {
635 // ERR("put_subbuf_size: cannot parse channel");
639 // ltt_lock_traces();
640 // trace = _ltt_trace_find(trace_name);
641 // ltt_unlock_traces();
643 // if(trace == NULL) {
644 // ERR("cannot find trace!");
648 // for(i=0; i<trace->nr_channels; i++) {
649 // struct rchan *rchan = trace->channels[i].trans_channel_data;
652 // if(!strcmp(trace->channels[i].channel_name, channel_name)) {
653 // struct rchan_buf *rbuf = rchan->buf;
654 // struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
656 // result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
657 // if(result == -1) {
658 // ERR("ustcomm_app_detach_client failed");
662 // lttbuf->wake_consumer_arg = (void *) fd;
666 // lttbuf->call_wake_consumer = 1;
672 // free(channel_name);
675 ERR("unable to parse message: %s", recvbuf
);
683 void create_listener(void)
686 static char listener_stack
[16384];
690 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
697 pthread_create(&thread
, NULL
, listener_main
, NULL
);
701 /* The signal handler itself. Signals must be setup so there cannot be
704 void sighandler(int sig
)
706 static char have_listener
= 0;
715 /* Called by the app signal handler to chain it to us. */
717 void chain_signal(void)
719 sighandler(USTSIGNAL
);
722 static int init_socket(void)
724 return ustcomm_init_app(getpid(), &ustcomm_app
);
727 /* FIXME: reenable this to delete socket file. */
730 static void destroy_socket(void)
734 if(mysocketfile
[0] == '\0')
737 result
= unlink(mysocketfile
);
744 static int init_signal_handler(void)
746 /* Attempt to handler SIGIO. If the main program wants to
747 * handle it, fine, it'll override us. They it'll have to
748 * use the chaining function.
752 struct sigaction act
;
754 result
= sigemptyset(&act
.sa_mask
);
756 PERROR("sigemptyset");
760 act
.sa_handler
= sighandler
;
761 act
.sa_flags
= SA_RESTART
;
763 /* Only defer ourselves. Also, try to restart interrupted
764 * syscalls to disturb the traced program as little as possible.
766 result
= sigaction(SIGIO
, &act
, NULL
);
775 static void auto_probe_connect(struct marker
*m
)
779 result
= ltt_marker_connect(m
->channel
, m
->name
, "default");
781 ERR("ltt_marker_connect");
783 DBG("just auto connected marker %s %s to probe default", m
->channel
, m
->name
);
786 static void __attribute__((constructor(101))) init0()
788 DBG("UST_AUTOPROBE constructor");
789 if(getenv("UST_AUTOPROBE")) {
790 marker_set_new_marker_cb(auto_probe_connect
);
794 static void __attribute__((constructor(1000))) init()
798 DBG("UST_TRACE constructor");
800 /* Must create socket before signal handler to prevent races.
802 result
= init_socket();
804 ERR("init_socket error");
807 result
= init_signal_handler();
809 ERR("init_signal_handler error");
813 if(getenv("UST_TRACE")) {
814 char trace_name
[] = "auto";
815 char trace_type
[] = "ustrelay";
817 DBG("starting early tracing");
819 /* Ensure marker control is initialized */
820 init_marker_control();
822 /* Ensure relay is initialized */
823 init_ustrelay_transport();
825 /* Ensure markers are initialized */
829 ltt_channels_register("ust");
831 result
= ltt_trace_setup(trace_name
);
833 ERR("ltt_trace_setup failed");
837 result
= ltt_trace_set_type(trace_name
, trace_type
);
839 ERR("ltt_trace_set_type failed");
843 result
= ltt_trace_alloc(trace_name
);
845 ERR("ltt_trace_alloc failed");
849 result
= ltt_trace_start(trace_name
);
851 ERR("ltt_trace_start failed");
854 inform_consumer_daemon();
860 /* should decrementally destroy stuff if error */
864 /* This is only called if we terminate normally, not with an unhandled signal,
865 * so we cannot rely on it. */
867 /* This destructor probably isn't needed, because ustd can do crash recovery. */
869 static void __attribute__((destructor
)) fini()
873 /* if trace running, finish it */
875 DBG("destructor stopping traces");
877 result
= ltt_trace_stop("auto");
879 ERR("ltt_trace_stop error");
882 result
= ltt_trace_destroy("auto");
884 ERR("ltt_trace_destroy error");