5 #include <sys/socket.h>
17 #define UNIX_PATH_MAX 108
19 #define SOCKETDIR "/tmp/socks"
20 #define SOCKETDIRLEN sizeof(SOCKETDIR)
21 #define USTSIGNAL SIGIO
23 #define MAX_MSG_SIZE (100)
25 #define MSG_REGISTER_NOTIF 2
27 char consumer_stack
[10000];
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 char mysocketfile
[UNIX_PATH_MAX
] = "";
53 struct consumer_channel
{
55 struct ltt_channel_struct
*chan
;
58 int consumer(void *arg
)
62 char str
[] = "Hello, this is the consumer.\n";
63 struct ltt_trace_struct
*trace
;
64 struct consumer_channel
*consumer_channels
;
66 char trace_name
[] = "auto";
69 trace
= _ltt_trace_find(trace_name
);
73 CPRINTF("cannot find trace!");
77 consumer_channels
= (struct consumer_channel
*) malloc(trace
->nr_channels
* sizeof(struct consumer_channel
));
78 if(consumer_channels
== NULL
) {
79 ERR("malloc returned NULL");
83 CPRINTF("opening trace files");
84 for(i
=0; i
<trace
->nr_channels
; i
++) {
86 struct ltt_channel_struct
*chan
= &trace
->channels
[i
];
88 consumer_channels
[i
].chan
= chan
;
90 snprintf(tmp
, sizeof(tmp
), "trace/%s_0", chan
->channel_name
);
91 result
= consumer_channels
[i
].fd
= open(tmp
, O_WRONLY
| O_CREAT
| O_TRUNC
, 00600);
96 CPRINTF("\topened trace file %s", tmp
);
99 CPRINTF("done opening trace files");
104 for(i
=0; i
<trace
->nr_channels
; i
++) {
105 struct rchan
*rchan
= consumer_channels
[i
].chan
->trans_channel_data
;
106 struct rchan_buf
*rbuf
= rchan
->buf
;
107 struct ltt_channel_buf_struct
*lttbuf
= consumer_channels
[i
].chan
->buf
;
110 result
= ltt_do_get_subbuf(rbuf
, lttbuf
, &consumed_old
);
112 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
117 result
= write(consumer_channels
[i
].fd
, rbuf
->buf_data
+ (consumed_old
& (2 * 4096-1)), 4096);
118 ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
126 void start_consumer(void)
131 result
= clone(consumer
, consumer_stack
+sizeof(consumer_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
138 pthread_create(&thread
, NULL
, consumer
, NULL
);
142 static void print_markers(void)
144 struct marker_iter iter
;
147 marker_iter_reset(&iter
);
148 marker_iter_start(&iter
);
151 fprintf(stderr
, "marker: %s_%s \"%s\"\n", iter
.marker
->channel
, iter
.marker
->name
, iter
.marker
->format
);
152 marker_iter_next(&iter
);
157 void do_command(struct tracecmd
*cmd
)
161 void receive_commands()
169 struct trctl_msg msg
;
171 /* FIXME: fd_notif should probably be protected by a spinlock */
176 msg
.type
= MSG_NOTIF
;
177 msg
.size
= sizeof(msg
.type
);
179 /* FIXME: don't block here */
180 result
= write(fd_notif
, &msg
, msg
.size
+sizeof(msg
.size
));
187 #define CONSUMER_DAEMON_SOCK SOCKETDIR "/ustd"
189 static int inform_consumer_daemon(void)
191 ustcomm_request_consumer(getpid(), "metadata");
192 ustcomm_request_consumer(getpid(), "ust");
195 int listener_main(void *p
)
203 struct sockaddr_un addr
;
204 socklen_t addrlen
= sizeof(addr
);
205 char trace_name
[] = "auto";
206 char trace_type
[] = "ustrelay";
209 struct ustcomm_source src
;
211 result
= ustcomm_app_recv_message(&ustcomm_app
, &recvbuf
, &src
, -1);
213 WARN("error in ustcomm_app_recv_message");
217 DBG("received a message! it's: %s\n", recvbuf
);
218 len
= strlen(recvbuf
);
220 if(!strcmp(recvbuf
, "print_markers")) {
223 else if(!strcmp(recvbuf
, "trace_setup")) {
226 result
= ltt_trace_setup(trace_name
);
228 ERR("ltt_trace_setup failed");
232 result
= ltt_trace_set_type(trace_name
, trace_type
);
234 ERR("ltt_trace_set_type failed");
238 else if(!strcmp(recvbuf
, "trace_alloc")) {
241 result
= ltt_trace_alloc(trace_name
);
243 ERR("ltt_trace_alloc failed");
247 else if(!strcmp(recvbuf
, "trace_start")) {
250 result
= ltt_trace_start(trace_name
);
252 ERR("ltt_trace_start failed");
256 else if(!strcmp(recvbuf
, "trace_stop")) {
259 result
= ltt_trace_stop(trace_name
);
261 ERR("ltt_trace_stop failed");
265 else if(!strcmp(recvbuf
, "trace_destroy")) {
267 DBG("trace destroy");
269 result
= ltt_trace_destroy(trace_name
);
271 ERR("ltt_trace_destroy failed");
275 else if(nth_token_is(recvbuf
, "get_shmid", 0) == 1) {
276 struct ltt_trace_struct
*trace
;
277 char trace_name
[] = "auto";
283 channel_name
= nth_token(recvbuf
, 1);
284 if(channel_name
== NULL
) {
285 ERR("get_shmid: cannot parse channel");
290 trace
= _ltt_trace_find(trace_name
);
294 CPRINTF("cannot find trace!");
298 for(i
=0; i
<trace
->nr_channels
; i
++) {
299 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
300 struct rchan_buf
*rbuf
= rchan
->buf
;
302 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
305 DBG("the shmid for the requested channel is %d", rbuf
->shmid
);
306 asprintf(&reply
, "%d", rbuf
->shmid
);
308 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
310 ERR("listener: get_shmid: ustcomm_send_reply failed");
320 else if(nth_token_is(recvbuf
, "get_n_subbufs", 0) == 1) {
321 struct ltt_trace_struct
*trace
;
322 char trace_name
[] = "auto";
326 DBG("get_n_subbufs");
328 channel_name
= nth_token(recvbuf
, 1);
329 if(channel_name
== NULL
) {
330 ERR("get_n_subbufs: cannot parse channel");
335 trace
= _ltt_trace_find(trace_name
);
339 CPRINTF("cannot find trace!");
343 for(i
=0; i
<trace
->nr_channels
; i
++) {
344 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
346 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
349 DBG("the n_subbufs for the requested channel is %d", rchan
->n_subbufs
);
350 asprintf(&reply
, "%d", rchan
->n_subbufs
);
352 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
354 ERR("listener: get_n_subbufs: ustcomm_send_reply failed");
364 else if(nth_token_is(recvbuf
, "get_subbuf_size", 0) == 1) {
365 struct ltt_trace_struct
*trace
;
366 char trace_name
[] = "auto";
370 DBG("get_subbuf_size");
372 channel_name
= nth_token(recvbuf
, 1);
373 if(channel_name
== NULL
) {
374 ERR("get_subbuf_size: cannot parse channel");
379 trace
= _ltt_trace_find(trace_name
);
383 CPRINTF("cannot find trace!");
387 for(i
=0; i
<trace
->nr_channels
; i
++) {
388 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
390 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
393 DBG("the subbuf_size for the requested channel is %d", rchan
->subbuf_size
);
394 asprintf(&reply
, "%d", rchan
->subbuf_size
);
396 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
398 ERR("listener: get_subbuf_size: ustcomm_send_reply failed");
408 else if(nth_token_is(recvbuf
, "load_probe_lib", 0) == 1) {
411 libfile
= nth_token(recvbuf
, 1);
413 DBG("load_probe_lib loading %s", libfile
);
415 else if(nth_token_is(recvbuf
, "get_subbuffer", 0) == 1) {
416 struct ltt_trace_struct
*trace
;
417 char trace_name
[] = "auto";
423 channel_name
= nth_token(recvbuf
, 1);
424 if(channel_name
== NULL
) {
425 ERR("get_subbuf: cannot parse channel");
430 trace
= _ltt_trace_find(trace_name
);
434 CPRINTF("cannot find trace!");
438 for(i
=0; i
<trace
->nr_channels
; i
++) {
439 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
441 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
442 struct rchan_buf
*rbuf
= rchan
->buf
;
443 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
447 result
= ltt_do_get_subbuf(rbuf
, lttbuf
, &consumed_old
);
449 DBG("ltt_do_get_subbuf: error: %s", strerror(-result
));
450 asprintf(&reply
, "%s %ld", "UNAVAIL", 0);
453 DBG("ltt_do_get_subbuf: success");
454 asprintf(&reply
, "%s %ld", "OK", consumed_old
);
457 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
459 ERR("listener: get_subbuf: ustcomm_send_reply failed");
469 else if(nth_token_is(recvbuf
, "put_subbuffer", 0) == 1) {
470 struct ltt_trace_struct
*trace
;
471 char trace_name
[] = "auto";
475 char *consumed_old_str
;
480 channel_name
= strdup_malloc(nth_token(recvbuf
, 1));
481 if(channel_name
== NULL
) {
482 ERR("put_subbuf_size: cannot parse channel");
486 consumed_old_str
= strdup_malloc(nth_token(recvbuf
, 2));
487 if(consumed_old_str
== NULL
) {
488 ERR("put_subbuf: cannot parse consumed_old");
491 consumed_old
= strtol(consumed_old_str
, &endptr
, 10);
492 if(*endptr
!= '\0') {
493 ERR("put_subbuf: invalid value for consumed_old");
498 trace
= _ltt_trace_find(trace_name
);
502 CPRINTF("cannot find trace!");
506 for(i
=0; i
<trace
->nr_channels
; i
++) {
507 struct rchan
*rchan
= trace
->channels
[i
].trans_channel_data
;
509 if(!strcmp(trace
->channels
[i
].channel_name
, channel_name
)) {
510 struct rchan_buf
*rbuf
= rchan
->buf
;
511 struct ltt_channel_buf_struct
*lttbuf
= trace
->channels
[i
].buf
;
515 result
= ltt_do_put_subbuf(rbuf
, lttbuf
, consumed_old
);
517 WARN("ltt_do_put_subbuf: error");
520 DBG("ltt_do_put_subbuf: success");
522 asprintf(&reply
, "%s", "OK", consumed_old
);
524 result
= ustcomm_send_reply(&ustcomm_app
.server
, reply
, &src
);
526 ERR("listener: put_subbuf: ustcomm_send_reply failed");
537 free(consumed_old_str
);
540 ERR("unable to parse message: %s", recvbuf
);
548 static char listener_stack
[16384];
550 void create_listener(void)
553 static char listener_stack
[16384];
554 //char *listener_stack = malloc(16384);
557 result
= clone(listener_main
, listener_stack
+sizeof(listener_stack
)-1, CLONE_FS
| CLONE_FILES
| CLONE_VM
| CLONE_SIGHAND
| CLONE_THREAD
, NULL
);
564 pthread_create(&thread
, NULL
, listener_main
, NULL
);
568 /* The signal handler itself. Signals must be setup so there cannot be
571 void sighandler(int sig
)
573 static char have_listener
= 0;
582 /* Called by the app signal handler to chain it to us. */
584 void chain_signal(void)
586 sighandler(USTSIGNAL
);
589 static int init_socket(void)
591 return ustcomm_init_app(getpid(), &ustcomm_app
);
594 static void destroy_socket(void)
598 if(mysocketfile
[0] == '\0')
601 result
= unlink(mysocketfile
);
607 static int init_signal_handler(void)
609 /* Attempt to handler SIGIO. If the main program wants to
610 * handle it, fine, it'll override us. They it'll have to
611 * use the chaining function.
615 struct sigaction act
;
617 result
= sigemptyset(&act
.sa_mask
);
619 PERROR("sigemptyset");
623 act
.sa_handler
= sighandler
;
624 act
.sa_flags
= SA_RESTART
;
626 /* Only defer ourselves. Also, try to restart interrupted
627 * syscalls to disturb the traced program as little as possible.
629 result
= sigaction(SIGIO
, &act
, NULL
);
638 static void auto_probe_connect(struct marker
*m
)
642 result
= ltt_marker_connect(m
->channel
, m
->name
, "default");
644 ERR("ltt_marker_connect");
646 DBG("just auto connected marker %s %s to probe default", m
->channel
, m
->name
);
649 static void __attribute__((constructor(101))) init0()
651 DBG("UST_AUTOPROBE constructor");
652 if(getenv("UST_AUTOPROBE")) {
653 marker_set_new_marker_cb(auto_probe_connect
);
657 static void fini(void);
659 static void __attribute__((constructor(1000))) init()
663 DBG("UST_TRACE constructor");
665 /* Must create socket before signal handler to prevent races.
667 result
= init_socket();
669 ERR("init_socket error");
672 result
= init_signal_handler();
674 ERR("init_signal_handler error");
678 if(getenv("UST_TRACE")) {
679 char trace_name
[] = "auto";
680 char trace_type
[] = "ustrelay";
682 DBG("starting early tracing");
684 /* Ensure marker control is initialized */
685 init_marker_control();
687 /* Ensure relay is initialized */
688 init_ustrelay_transport();
690 /* Ensure markers are initialized */
694 ltt_channels_register("ust");
696 result
= ltt_trace_setup(trace_name
);
698 ERR("ltt_trace_setup failed");
702 result
= ltt_trace_set_type(trace_name
, trace_type
);
704 ERR("ltt_trace_set_type failed");
708 result
= ltt_trace_alloc(trace_name
);
710 ERR("ltt_trace_alloc failed");
714 result
= ltt_trace_start(trace_name
);
716 ERR("ltt_trace_start failed");
720 inform_consumer_daemon();
726 /* should decrementally destroy stuff if error */
730 /* This is only called if we terminate normally, not with an unhandled signal,
731 * so we cannot rely on it. */
733 static void __attribute__((destructor
)) fini()
737 /* if trace running, finish it */
739 DBG("destructor stopping traces");
741 result
= ltt_trace_stop("auto");
743 ERR("ltt_trace_stop error");
746 result
= ltt_trace_destroy("auto");
748 ERR("ltt_trace_destroy error");
751 /* FIXME: wait for the consumer to be done */
752 DBG("waiting 5 sec for consume");
This page took 0.048747 seconds and 5 git commands to generate.