5 #include <sys/socket.h>
14 #include "relay.h" /* FIXME: remove */
18 #define USTSIGNAL SIGIO
20 #define MAX_MSG_SIZE (100)
22 #define MSG_REGISTER_NOTIF 2
24 char consumer_stack
[10000];
26 static struct ustcomm_app ustcomm_app
;
28 struct tracecmd
{ /* no padding */
33 //struct listener_arg {
38 /* size: the size of all the fields except size itself */
41 /* Only the necessary part of the payload is transferred. It
42 * may even be none of it.
47 struct consumer_channel
{
49 struct ltt_channel_struct
*chan
;
52 int consumer(void *arg
)
56 char str
[] = "Hello, this is the consumer.\n";
57 struct ltt_trace_struct
*trace
;
58 struct consumer_channel
*consumer_channels
;
60 char trace_name
[] = "auto";
63 trace
= _ltt_trace_find(trace_name
);
67 CPRINTF("cannot find trace!");
71 consumer_channels
= (struct consumer_channel
*) malloc(trace
->nr_channels
* sizeof(struct consumer_channel
));
72 if(consumer_channels
== NULL
) {
73 ERR("malloc returned NULL");
77 CPRINTF("opening trace files");
78 for(i
=0; i
<trace
->nr_channels
; i
++) {
80 struct ltt_channel_struct
*chan
= &trace
->channels
[i
];
82 consumer_channels
[i
].chan
= chan
;
84 snprintf(tmp
, sizeof(tmp
), "trace/%s_0", chan
->channel_name
);
85 result
= consumer_channels
[i
].fd
= open(tmp
, O_WRONLY
| O_CREAT
| O_TRUNC
, 00600);
90 CPRINTF("\topened trace file %s", tmp
);
93 CPRINTF("done opening trace files");
98 for(i
=0; i
<trace
->nr_channels
; i
++) {
99 struct rchan
*rchan
= consumer_channels
[i
].chan
->trans_channel_data
;
100 struct rchan_buf
*rbuf
= rchan
->buf
;
101 struct ltt_channel_buf_struct
*lttbuf
= consumer_channels
[i
].chan
->buf
;
104 result
= ltt_do_get_subbuf(rbuf
, lttbuf
, &consumed_old
);
106 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
111 result
= write(consumer_channels
[i
].fd
, rbuf
->buf_data
+ (consumed_old
& (2 * 4096-1)), 4096);
112 ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
120 void start_consumer(void)
125 result
= clone(consumer
, consumer_stack
+sizeof(consumer_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
132 pthread_create(&thread
, NULL
, consumer
, NULL
);
136 static void print_markers(void)
138 struct marker_iter iter
;
141 marker_iter_reset(&iter
);
142 marker_iter_start(&iter
);
145 fprintf(stderr
, "marker: %s_%s \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, iter
.marker
->format
);
146 marker_iter_next(&iter
);
151 void do_command(struct tracecmd
*cmd
)
155 void receive_commands()
163 struct trctl_msg msg
;
165 /* FIXME: fd_notif should probably be protected by a spinlock */
170 msg
.type
= MSG_NOTIF
;
171 msg
.size
= sizeof(msg
.type
);
173 /* FIXME: don't block here */
174 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
181 static int inform_consumer_daemon(void)
183 ustcomm_request_consumer(getpid(), "metadata");
184 ustcomm_request_consumer(getpid(), "ust");
187 int listener_main(void *p
)
195 struct sockaddr_un addr
;
196 socklen_t addrlen
= sizeof(addr
);
197 char trace_name
[] = "auto";
198 char trace_type
[] = "ustrelay";
201 struct ustcomm_source src
;
203 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, -1);
205 WARN("error in ustcomm_app_recv_message");
209 DBG("received a message! it's: %s\n", recvbuf
);
210 len
= strlen(recvbuf
);
212 if(!strcmp(recvbuf
, "print_markers")) {
215 else if(!strcmp(recvbuf
, "trace_setup")) {
218 result
= ltt_trace_setup(trace_name
);
220 ERR("ltt_trace_setup failed");
224 result
= ltt_trace_set_type(trace_name
, trace_type
);
226 ERR("ltt_trace_set_type failed");
230 else if(!strcmp(recvbuf
, "trace_alloc")) {
233 result
= ltt_trace_alloc(trace_name
);
235 ERR("ltt_trace_alloc failed");
239 else if(!strcmp(recvbuf
, "trace_start")) {
242 result
= ltt_trace_start(trace_name
);
244 ERR("ltt_trace_start failed");
248 else if(!strcmp(recvbuf
, "trace_stop")) {
251 result
= ltt_trace_stop(trace_name
);
253 ERR("ltt_trace_stop failed");
257 else if(!strcmp(recvbuf
, "trace_destroy")) {
259 DBG("trace destroy");
261 result
= ltt_trace_destroy(trace_name
);
263 ERR("ltt_trace_destroy failed");
267 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
268 struct ltt_trace_struct
*trace
;
269 char trace_name
[] = "auto";
275 channel_name
= nth_token(recvbuf
, 1);
276 if(channel_name
== NULL
) {
277 ERR("get_shmid: cannot parse channel");
282 trace
= _ltt_trace_find(trace_name
);
286 CPRINTF("cannot find trace!");
290 for(i
=0; i
<trace
->nr_channels
; i
++) {
291 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
292 struct rchan_buf
*rbuf
= rchan
->buf
;
294 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
297 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
298 asprintf(&reply
, "%d", rbuf
->shmid
);
300 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
302 ERR("listener: get_shmid: ustcomm_send_reply failed");
312 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
313 struct ltt_trace_struct
*trace
;
314 char trace_name
[] = "auto";
318 DBG("get_n_subbufs");
320 channel_name
= nth_token(recvbuf
, 1);
321 if(channel_name
== NULL
) {
322 ERR("get_n_subbufs: cannot parse channel");
327 trace
= _ltt_trace_find(trace_name
);
331 CPRINTF("cannot find trace!");
335 for(i
=0; i
<trace
->nr_channels
; i
++) {
336 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
338 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
341 DBG("the n_subbufs for the requested channel is %d", rchan
->n_subbufs
);
342 asprintf(&reply
, "%d", rchan
->n_subbufs
);
344 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
346 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
356 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
357 struct ltt_trace_struct
*trace
;
358 char trace_name
[] = "auto";
362 DBG("get_subbuf_size");
364 channel_name
= nth_token(recvbuf
, 1);
365 if(channel_name
== NULL
) {
366 ERR("get_subbuf_size: cannot parse channel");
371 trace
= _ltt_trace_find(trace_name
);
375 CPRINTF("cannot find trace!");
379 for(i
=0; i
<trace
->nr_channels
; i
++) {
380 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
382 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
385 DBG("the subbuf_size for the requested channel is %d", rchan
->subbuf_size
);
386 asprintf(&reply
, "%d", rchan
->subbuf_size
);
388 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
390 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
400 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
403 libfile
= nth_token(recvbuf
, 1);
405 DBG("load_probe_lib loading %s", libfile
);
407 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
408 struct ltt_trace_struct
*trace
;
409 char trace_name
[] = "auto";
415 channel_name
= nth_token(recvbuf
, 1);
416 if(channel_name
== NULL
) {
417 ERR("get_subbuf: cannot parse channel");
422 trace
= _ltt_trace_find(trace_name
);
426 CPRINTF("cannot find trace!");
430 for(i
=0; i
<trace
->nr_channels
; i
++) {
431 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
433 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
434 struct rchan_buf
*rbuf
= rchan
->buf
;
435 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
439 result
= ltt_do_get_subbuf(rbuf
, lttbuf
, &consumed_old
);
441 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
442 asprintf(&reply
, "%s %ld", "UNAVAIL", 0);
445 DBG("ltt_do_get_subbuf: success");
446 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
449 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
451 ERR("listener: get_subbuf: ustcomm_send_reply failed");
461 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
462 struct ltt_trace_struct
*trace
;
463 char trace_name
[] = "auto";
467 char *consumed_old_str
;
472 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
473 if(channel_name
== NULL
) {
474 ERR("put_subbuf_size: cannot parse channel");
478 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
479 if(consumed_old_str
== NULL
) {
480 ERR("put_subbuf: cannot parse consumed_old");
483 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
484 if(*endptr
!= '\0') {
485 ERR("put_subbuf: invalid value for consumed_old");
490 trace
= _ltt_trace_find(trace_name
);
494 CPRINTF("cannot find trace!");
498 for(i
=0; i
<trace
->nr_channels
; i
++) {
499 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
501 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
502 struct rchan_buf
*rbuf
= rchan
->buf
;
503 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
507 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
509 WARN("ltt_do_put_subbuf: error");
512 DBG("ltt_do_put_subbuf: success");
514 asprintf(&reply
, "%s", "OK", consumed_old
);
516 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
518 ERR("listener: put_subbuf: ustcomm_send_reply failed");
529 free(consumed_old_str
);
531 else if(nth_token_is(recvbuf
, "get_notifications", 0) == 1) {
532 struct ltt_trace_struct
*trace
;
533 char trace_name
[] = "auto";
537 DBG("get_notifications");
539 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
540 if(channel_name
== NULL
) {
541 ERR("put_subbuf_size: cannot parse channel");
546 trace
= _ltt_trace_find(trace_name
);
550 CPRINTF("cannot find trace!");
554 for(i
=0; i
<trace
->nr_channels
; i
++) {
555 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
558 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
559 struct rchan_buf
*rbuf
= rchan
->buf
;
560 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
562 result
= fd
= ustcomm_app_detach_client(&ustcomm_app
, &src
);
564 ERR("ustcomm_app_detach_client failed");
568 lttbuf
->wake_consumer_arg
= (void *) fd
;
572 lttbuf
->call_wake_consumer
= 1;
581 ERR("unable to parse message: %s", recvbuf
);
589 static char listener_stack
[16384];
591 void create_listener(void)
594 static char listener_stack
[16384];
595 //char *listener_stack = malloc(16384);
598 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
605 pthread_create(&thread
, NULL
, listener_main
, NULL
);
609 /* The signal handler itself. Signals must be setup so there cannot be
612 void sighandler(int sig
)
614 static char have_listener
= 0;
623 /* Called by the app signal handler to chain it to us. */
625 void chain_signal(void)
627 sighandler(USTSIGNAL
);
630 static int init_socket(void)
632 return ustcomm_init_app(getpid(), &ustcomm_app
);
635 static void destroy_socket(void)
639 // if(mysocketfile[0] == '\0')
642 // result = unlink(mysocketfile);
643 // if(result == -1) {
648 static int init_signal_handler(void)
650 /* Attempt to handler SIGIO. If the main program wants to
651 * handle it, fine, it'll override us. They it'll have to
652 * use the chaining function.
656 struct sigaction act
;
658 result
= sigemptyset(&act
.sa_mask
);
660 PERROR("sigemptyset");
664 act
.sa_handler
= sighandler
;
665 act
.sa_flags
= SA_RESTART
;
667 /* Only defer ourselves. Also, try to restart interrupted
668 * syscalls to disturb the traced program as little as possible.
670 result
= sigaction(SIGIO
, &act
, NULL
);
679 static void auto_probe_connect(struct marker
*m
)
683 result
= ltt_marker_connect(m
->channel
, m
->name
, "default");
685 ERR("ltt_marker_connect");
687 DBG("just auto connected marker %s %s to probe default", m
->channel
, m
->name
);
690 /* Wake the consumer of a buffer
692 * wake_consumer_cb is called in tracing context so it must haste.
694 * FIXME: don't do a system call here; maybe schedule work to be done
695 * in listener context? Once this is done in listener context, we can
696 * check for the return value of send_message_fd and remove the fd if necessary
699 * @finished: 0: subbuffer full; 1: buffer being destroyed
702 static void wake_consumer_cb(void *arg
, int finished
)
704 struct ltt_channel_buf_struct
*ltt_buf
= (struct ltt_channel_buf_struct
*) arg
;
705 int fd
= (int)ACCESS_ONCE(arg
);
708 send_message_fd(fd
, "consume", NULL
);
711 send_message_fd(fd
, "destroyed", NULL
);
715 static void __attribute__((constructor(101))) init0()
717 DBG("UST_AUTOPROBE constructor");
718 if(getenv("UST_AUTOPROBE")) {
719 marker_set_new_marker_cb(auto_probe_connect
);
722 relay_set_wake_consumer(wake_consumer_cb
);
725 static void fini(void);
727 static void __attribute__((constructor(1000))) init()
731 DBG("UST_TRACE constructor");
733 /* Must create socket before signal handler to prevent races.
735 result
= init_socket();
737 ERR("init_socket error");
740 result
= init_signal_handler();
742 ERR("init_signal_handler error");
746 if(getenv("UST_TRACE")) {
747 char trace_name
[] = "auto";
748 char trace_type
[] = "ustrelay";
750 DBG("starting early tracing");
752 /* Ensure marker control is initialized */
753 init_marker_control();
755 /* Ensure relay is initialized */
756 init_ustrelay_transport();
758 /* Ensure markers are initialized */
762 ltt_channels_register("ust");
764 result
= ltt_trace_setup(trace_name
);
766 ERR("ltt_trace_setup failed");
770 result
= ltt_trace_set_type(trace_name
, trace_type
);
772 ERR("ltt_trace_set_type failed");
776 result
= ltt_trace_alloc(trace_name
);
778 ERR("ltt_trace_alloc failed");
782 result
= ltt_trace_start(trace_name
);
784 ERR("ltt_trace_start failed");
788 inform_consumer_daemon();
794 /* should decrementally destroy stuff if error */
798 /* This is only called if we terminate normally, not with an unhandled signal,
799 * so we cannot rely on it. */
801 static void __attribute__((destructor
)) fini()
805 /* if trace running, finish it */
807 DBG("destructor stopping traces");
809 result
= ltt_trace_stop("auto");
811 ERR("ltt_trace_stop error");
814 result
= ltt_trace_destroy("auto");
816 ERR("ltt_trace_destroy error");
819 /* FIXME: wait for the consumer to be done */
820 DBG("waiting 5 sec for consume");
This page took 0.056856 seconds and 4 git commands to generate.