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 int consumer(void *arg
)
75 char str
[] = "Hello, this is the consumer.\n";
76 struct ltt_trace_struct
*trace
;
77 struct consumer_channel
*consumer_channels
;
79 char trace_name
[] = "auto";
82 trace
= _ltt_trace_find(trace_name
);
86 CPRINTF("cannot find trace!");
90 consumer_channels
= (struct consumer_channel
*) malloc(trace
->nr_channels
* sizeof(struct consumer_channel
));
91 if(consumer_channels
== NULL
) {
92 ERR("malloc returned NULL");
96 CPRINTF("opening trace files");
97 for(i
=0; i
<trace
->nr_channels
; i
++) {
99 struct ltt_channel_struct
*chan
= &trace
->channels
[i
];
101 consumer_channels
[i
].chan
= chan
;
103 snprintf(tmp
, sizeof(tmp
), "trace/%s_0", chan
->channel_name
);
104 result
= consumer_channels
[i
].fd
= open(tmp
, O_WRONLY
| O_CREAT
| O_TRUNC
, 00600);
109 CPRINTF("\topened trace file %s", tmp
);
112 CPRINTF("done opening trace files");
117 for(i
=0; i
<trace
->nr_channels
; i
++) {
118 struct rchan
*rchan
= consumer_channels
[i
].chan
->trans_channel_data
;
119 struct rchan_buf
*rbuf
= rchan
->buf
;
120 struct ltt_channel_buf_struct
*lttbuf
= consumer_channels
[i
].chan
->buf
;
123 result
= ltt_do_get_subbuf(rbuf
, lttbuf
, &consumed_old
);
125 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
130 result
= write(consumer_channels
[i
].fd
, rbuf
->buf_data
+ (consumed_old
& (2 * 4096-1)), 4096);
131 ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
139 void start_consumer(void)
144 result
= clone(consumer
, consumer_stack
+sizeof(consumer_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
151 pthread_create(&thread
, NULL
, consumer
, NULL
);
155 static void print_markers(void)
157 struct marker_iter iter
;
160 marker_iter_reset(&iter
);
161 marker_iter_start(&iter
);
164 fprintf(stderr
, "marker: %s_%s \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, iter
.marker
->format
);
165 marker_iter_next(&iter
);
170 void do_command(struct tracecmd
*cmd
)
174 void receive_commands()
182 struct trctl_msg msg
;
184 /* FIXME: fd_notif should probably be protected by a spinlock */
189 msg
.type
= MSG_NOTIF
;
190 msg
.size
= sizeof(msg
.type
);
192 /* FIXME: don't block here */
193 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
200 static int inform_consumer_daemon(void)
202 ustcomm_request_consumer(getpid(), "metadata");
203 ustcomm_request_consumer(getpid(), "ust");
206 void process_blocked_consumers(void)
210 struct blocked_consumer
*bc
;
215 list_for_each_entry(bc
, &blocked_consumers
, list
) {
219 fds
= (struct pollfd
*) malloc(n_fds
* sizeof(struct pollfd
));
221 ERR("malloc returned NULL");
225 list_for_each_entry(bc
, &blocked_consumers
, list
) {
226 fds
[idx
].fd
= bc
->fd_producer
;
227 fds
[idx
].events
= POLLIN
;
228 bc
->tmp_poll_idx
= idx
;
232 result
= poll(fds
, n_fds
, 0);
238 list_for_each_entry(bc
, &blocked_consumers
, list
) {
239 if(fds
[bc
->tmp_poll_idx
].revents
) {
240 long consumed_old
= 0;
243 result
= read(bc
->fd_producer
, &inbuf
, 1);
251 close(bc
->fd_producer
);
253 __list_del(bc
->list
.prev
, bc
->list
.next
);
255 result
= ustcomm_send_reply(&bc
->server
, "END", &bc
->src
);
257 ERR("ustcomm_send_reply failed");
264 result
= ltt_do_get_subbuf(bc
->rbuf
, bc
->lttbuf
, &consumed_old
);
265 if(result
== -EAGAIN
) {
266 WARN("missed buffer?");
269 else if(result
< 0) {
270 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
272 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
273 result
= ustcomm_send_reply(&bc
->server
, reply
, &bc
->src
);
275 ERR("ustcomm_send_reply failed");
281 __list_del(bc
->list
.prev
, bc
->list
.next
);
287 int listener_main(void *p
)
295 struct sockaddr_un addr
;
296 socklen_t addrlen
= sizeof(addr
);
297 char trace_name
[] = "auto";
298 char trace_type
[] = "ustrelay";
301 struct ustcomm_source src
;
303 process_blocked_consumers();
305 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, 5);
307 WARN("error in ustcomm_app_recv_message");
310 else if(result
== 0) {
315 DBG("received a message! it's: %s\n", recvbuf
);
316 len
= strlen(recvbuf
);
318 if(!strcmp(recvbuf
, "print_markers")) {
321 else if(!strcmp(recvbuf
, "trace_setup")) {
324 result
= ltt_trace_setup(trace_name
);
326 ERR("ltt_trace_setup failed");
330 result
= ltt_trace_set_type(trace_name
, trace_type
);
332 ERR("ltt_trace_set_type failed");
336 else if(!strcmp(recvbuf
, "trace_alloc")) {
339 result
= ltt_trace_alloc(trace_name
);
341 ERR("ltt_trace_alloc failed");
345 else if(!strcmp(recvbuf
, "trace_start")) {
348 result
= ltt_trace_start(trace_name
);
350 ERR("ltt_trace_start failed");
354 else if(!strcmp(recvbuf
, "trace_stop")) {
357 result
= ltt_trace_stop(trace_name
);
359 ERR("ltt_trace_stop failed");
363 else if(!strcmp(recvbuf
, "trace_destroy")) {
365 DBG("trace destroy");
367 result
= ltt_trace_destroy(trace_name
);
369 ERR("ltt_trace_destroy failed");
373 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
374 struct ltt_trace_struct
*trace
;
375 char trace_name
[] = "auto";
381 channel_name
= nth_token(recvbuf
, 1);
382 if(channel_name
== NULL
) {
383 ERR("get_shmid: cannot parse channel");
388 trace
= _ltt_trace_find(trace_name
);
392 CPRINTF("cannot find trace!");
396 for(i
=0; i
<trace
->nr_channels
; i
++) {
397 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
398 struct rchan_buf
*rbuf
= rchan
->buf
;
400 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
403 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
404 asprintf(&reply
, "%d", rbuf
->shmid
);
406 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
408 ERR("listener: get_shmid: ustcomm_send_reply failed");
418 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
419 struct ltt_trace_struct
*trace
;
420 char trace_name
[] = "auto";
424 DBG("get_n_subbufs");
426 channel_name
= nth_token(recvbuf
, 1);
427 if(channel_name
== NULL
) {
428 ERR("get_n_subbufs: cannot parse channel");
433 trace
= _ltt_trace_find(trace_name
);
437 CPRINTF("cannot find trace!");
441 for(i
=0; i
<trace
->nr_channels
; i
++) {
442 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
444 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
447 DBG("the n_subbufs for the requested channel is %d", rchan
->n_subbufs
);
448 asprintf(&reply
, "%d", rchan
->n_subbufs
);
450 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
452 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
462 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
463 struct ltt_trace_struct
*trace
;
464 char trace_name
[] = "auto";
468 DBG("get_subbuf_size");
470 channel_name
= nth_token(recvbuf
, 1);
471 if(channel_name
== NULL
) {
472 ERR("get_subbuf_size: cannot parse channel");
477 trace
= _ltt_trace_find(trace_name
);
481 CPRINTF("cannot find trace!");
485 for(i
=0; i
<trace
->nr_channels
; i
++) {
486 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
488 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
491 DBG("the subbuf_size for the requested channel is %d", rchan
->subbuf_size
);
492 asprintf(&reply
, "%d", rchan
->subbuf_size
);
494 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
496 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
506 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
509 libfile
= nth_token(recvbuf
, 1);
511 DBG("load_probe_lib loading %s", libfile
);
513 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
514 struct ltt_trace_struct
*trace
;
515 char trace_name
[] = "auto";
521 channel_name
= nth_token(recvbuf
, 1);
522 if(channel_name
== NULL
) {
523 ERR("get_subbuf: cannot parse channel");
528 trace
= _ltt_trace_find(trace_name
);
532 CPRINTF("cannot find trace!");
536 for(i
=0; i
<trace
->nr_channels
; i
++) {
537 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
539 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
540 struct rchan_buf
*rbuf
= rchan
->buf
;
541 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
545 struct blocked_consumer
*bc
;
547 bc
= (struct blocked_consumer
*) malloc(sizeof(struct blocked_consumer
));
549 ERR("malloc returned NULL");
552 bc
->fd_consumer
= src
.fd
;
553 bc
->fd_producer
= lttbuf
->data_ready_fd_read
;
557 bc
->server
= ustcomm_app
.server
;
559 list_add(&bc
->list
, &blocked_consumers
);
565 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
566 struct ltt_trace_struct
*trace
;
567 char trace_name
[] = "auto";
571 char *consumed_old_str
;
576 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
577 if(channel_name
== NULL
) {
578 ERR("put_subbuf_size: cannot parse channel");
582 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
583 if(consumed_old_str
== NULL
) {
584 ERR("put_subbuf: cannot parse consumed_old");
587 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
588 if(*endptr
!= '\0') {
589 ERR("put_subbuf: invalid value for consumed_old");
594 trace
= _ltt_trace_find(trace_name
);
598 CPRINTF("cannot find trace!");
602 for(i
=0; i
<trace
->nr_channels
; i
++) {
603 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
605 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
606 struct rchan_buf
*rbuf
= rchan
->buf
;
607 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
611 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
613 WARN("ltt_do_put_subbuf: error");
616 DBG("ltt_do_put_subbuf: success");
618 asprintf(&reply
, "%s", "OK", consumed_old
);
620 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
622 ERR("listener: put_subbuf: ustcomm_send_reply failed");
633 free(consumed_old_str
);
635 // else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
636 // struct ltt_trace_struct *trace;
637 // char trace_name[] = "auto";
639 // char *channel_name;
641 // DBG("get_notifications");
643 // channel_name = strdup_malloc(nth_token(recvbuf, 1));
644 // if(channel_name == NULL) {
645 // ERR("put_subbuf_size: cannot parse channel");
649 // ltt_lock_traces();
650 // trace = _ltt_trace_find(trace_name);
651 // ltt_unlock_traces();
653 // if(trace == NULL) {
654 // CPRINTF("cannot find trace!");
658 // for(i=0; i<trace->nr_channels; i++) {
659 // struct rchan *rchan = trace->channels[i].trans_channel_data;
662 // if(!strcmp(trace->channels[i].channel_name, channel_name)) {
663 // struct rchan_buf *rbuf = rchan->buf;
664 // struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
666 // result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
667 // if(result == -1) {
668 // ERR("ustcomm_app_detach_client failed");
672 // lttbuf->wake_consumer_arg = (void *) fd;
676 // lttbuf->call_wake_consumer = 1;
682 // free(channel_name);
685 ERR("unable to parse message: %s", recvbuf
);
693 static char listener_stack
[16384];
695 void create_listener(void)
698 static char listener_stack
[16384];
699 //char *listener_stack = malloc(16384);
702 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
709 pthread_create(&thread
, NULL
, listener_main
, NULL
);
713 /* The signal handler itself. Signals must be setup so there cannot be
716 void sighandler(int sig
)
718 static char have_listener
= 0;
727 /* Called by the app signal handler to chain it to us. */
729 void chain_signal(void)
731 sighandler(USTSIGNAL
);
734 static int init_socket(void)
736 return ustcomm_init_app(getpid(), &ustcomm_app
);
739 static void destroy_socket(void)
743 // if(mysocketfile[0] == '\0')
746 // result = unlink(mysocketfile);
747 // if(result == -1) {
752 static int init_signal_handler(void)
754 /* Attempt to handler SIGIO. If the main program wants to
755 * handle it, fine, it'll override us. They it'll have to
756 * use the chaining function.
760 struct sigaction act
;
762 result
= sigemptyset(&act
.sa_mask
);
764 PERROR("sigemptyset");
768 act
.sa_handler
= sighandler
;
769 act
.sa_flags
= SA_RESTART
;
771 /* Only defer ourselves. Also, try to restart interrupted
772 * syscalls to disturb the traced program as little as possible.
774 result
= sigaction(SIGIO
, &act
, NULL
);
783 static void auto_probe_connect(struct marker
*m
)
787 result
= ltt_marker_connect(m
->channel
, m
->name
, "default");
789 ERR("ltt_marker_connect");
791 DBG("just auto connected marker %s %s to probe default", m
->channel
, m
->name
);
794 static void __attribute__((constructor(101))) init0()
796 DBG("UST_AUTOPROBE constructor");
797 if(getenv("UST_AUTOPROBE")) {
798 marker_set_new_marker_cb(auto_probe_connect
);
802 static void fini(void);
804 static void __attribute__((constructor(1000))) init()
808 DBG("UST_TRACE constructor");
810 /* Must create socket before signal handler to prevent races.
812 result
= init_socket();
814 ERR("init_socket error");
817 result
= init_signal_handler();
819 ERR("init_signal_handler error");
823 if(getenv("UST_TRACE")) {
824 char trace_name
[] = "auto";
825 char trace_type
[] = "ustrelay";
827 DBG("starting early tracing");
829 /* Ensure marker control is initialized */
830 init_marker_control();
832 /* Ensure relay is initialized */
833 init_ustrelay_transport();
835 /* Ensure markers are initialized */
839 ltt_channels_register("ust");
841 result
= ltt_trace_setup(trace_name
);
843 ERR("ltt_trace_setup failed");
847 result
= ltt_trace_set_type(trace_name
, trace_type
);
849 ERR("ltt_trace_set_type failed");
853 result
= ltt_trace_alloc(trace_name
);
855 ERR("ltt_trace_alloc failed");
859 result
= ltt_trace_start(trace_name
);
861 ERR("ltt_trace_start failed");
865 inform_consumer_daemon();
871 /* should decrementally destroy stuff if error */
875 /* This is only called if we terminate normally, not with an unhandled signal,
876 * so we cannot rely on it. */
878 static void __attribute__((destructor
)) fini()
882 /* if trace running, finish it */
884 DBG("destructor stopping traces");
886 result
= ltt_trace_stop("auto");
888 ERR("ltt_trace_stop error");
891 result
= ltt_trace_destroy("auto");
893 ERR("ltt_trace_destroy error");
896 /* FIXME: wait for the consumer to be done */
897 DBG("waiting 5 sec for consume");