#include <sys/socket.h>
#include <sys/un.h>
#include <sched.h>
+#include <fcntl.h>
+
+#include "marker.h"
+#include "tracer.h"
+#include "usterr.h"
+
#define UNIX_PATH_MAX 108
//#define SOCKETDIR "/var/run/ust/socks"
#define SOCKETDIRLEN sizeof(SOCKETDIR)
#define USTSIGNAL SIGIO
-#define DBG(fmt, args...) fprintf(stderr, fmt "\n", ## args)
-#define WARN(fmt, args...) fprintf(stderr, "usertrace: WARNING: " fmt "\n", ## args)
-#define ERR(fmt, args...) fprintf(stderr, "usertrace: ERROR: " fmt "\n", ## args)
-#define PERROR(call) perror("usertrace: ERROR: " call)
-
#define MAX_MSG_SIZE (100)
#define MSG_NOTIF 1
#define MSG_REGISTER_NOTIF 2
+char consumer_stack[10000];
+
struct tracecmd { /* no padding */
uint32_t size;
uint16_t command;
char mysocketfile[UNIX_PATH_MAX] = "";
int pfd = -1;
+struct consumer_channel {
+ int fd;
+ struct ltt_channel_struct *chan;
+};
+
+int consumer(void *arg)
+{
+ int result;
+ int fd;
+ char str[] = "Hello, this is the consumer.\n";
+ struct ltt_trace_struct *trace;
+ struct consumer_channel *consumer_channels;
+ int i;
+ char trace_name[] = "auto";
+
+ ltt_lock_traces();
+ trace = _ltt_trace_find(trace_name);
+ ltt_unlock_traces();
+
+ if(trace == NULL) {
+ CPRINTF("cannot find trace!");
+ return 1;
+ }
+
+ consumer_channels = (struct consumer_channel *) malloc(trace->nr_channels * sizeof(struct consumer_channel));
+ if(consumer_channels == NULL) {
+ ERR("malloc returned NULL");
+ return 1;
+ }
+
+ CPRINTF("opening trace files");
+ for(i=0; i<trace->nr_channels; i++) {
+ char tmp[100];
+ struct ltt_channel_struct *chan = &trace->channels[i];
+
+ consumer_channels[i].chan = chan;
+
+ snprintf(tmp, sizeof(tmp), "trace/%s_0", chan->channel_name);
+ result = consumer_channels[i].fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 00644);
+ if(result == -1) {
+ perror("open");
+ return -1;
+ }
+ CPRINTF("\topened trace file %s", tmp);
+
+ }
+ CPRINTF("done opening trace files");
+
+ for(;;) {
+ /*wait*/
+
+ for(i=0; i<trace->nr_channels; i++) {
+ struct rchan *rchan = consumer_channels[i].chan->trans_channel_data;
+ struct rchan_buf *rbuf = rchan->buf;
+ struct ltt_channel_buf_struct *lttbuf = consumer_channels[i].chan->buf;
+ long consumed_old;
+
+ result = ltt_do_get_subbuf(rbuf, lttbuf, &consumed_old);
+ if(result < 0) {
+ DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
+ }
+ else {
+ DBG("success!");
+
+ result = write(consumer_channels[i].fd, rbuf->buf_data + (consumed_old & (2 * 4096-1)), 4096);
+ ltt_do_put_subbuf(rbuf, lttbuf, consumed_old);
+ }
+ }
+
+ sleep(1);
+ }
+
+// CPRINTF("consumer: got a trace: %s with %d channels\n", trace_name, trace->nr_channels);
+//
+// struct ltt_channel_struct *chan = &trace->channels[0];
+//
+// CPRINTF("channel 1 (%s) active=%u", chan->channel_name, chan->active & 1);
+
+// struct rchan *rchan = chan->trans_channel_data;
+// struct rchan_buf *rbuf = rchan->buf;
+// struct ltt_channel_buf_struct *lttbuf = chan->buf;
+// long consumed_old;
+//
+// result = fd = open("trace.out", O_WRONLY | O_CREAT | O_TRUNC, 00644);
+// if(result == -1) {
+// perror("open");
+// return -1;
+// }
+
+// for(;;) {
+// write(STDOUT_FILENO, str, sizeof(str));
+//
+// result = ltt_do_get_subbuf(rbuf, lttbuf, &consumed_old);
+// if(result < 0) {
+// CPRINTF("ltt_do_get_subbuf: error: %s", strerror(-result));
+// }
+// else {
+// CPRINTF("success!");
+//
+// result = write(fd, rbuf->buf_data + (consumed_old & (2 * 4096-1)), 4096);
+// ltt_do_put_subbuf(rbuf, lttbuf, consumed_old);
+// }
+//
+// //CPRINTF("There seems to be %ld bytes available", SUBBUF_TRUNC(local_read(<tbuf->offset), rbuf->chan) - consumed_old);
+// CPRINTF("Commit count %ld", local_read(<tbuf->commit_count[0]));
+//
+//
+// sleep(1);
+// }
+}
+
+void start_consumer(void)
+{
+ int result;
+
+ result = clone(consumer, consumer_stack+sizeof(consumer_stack)-1, CLONE_FS | CLONE_FILES | CLONE_VM | CLONE_SIGHAND | CLONE_THREAD, NULL);
+ if(result == -1) {
+ perror("clone");
+ }
+}
+
+static void print_markers(void)
+{
+ struct marker_iter iter;
+
+ marker_iter_reset(&iter);
+ marker_iter_start(&iter);
+
+ while(iter.marker) {
+ fprintf(stderr, "marker: %s_%s \"%s\"\n", iter.marker->channel, iter.marker->name, iter.marker->format);
+ marker_iter_next(&iter);
+ }
+}
+
void do_command(struct tracecmd *cmd)
{
}
}
}
+char recvbuf[10000];
+
int listener_main(void *p)
{
int result;
- /* Allowing only 1 connection for now. */
- result = listen(pfd, 1);
- if(result == -1) {
- PERROR("listen");
- return 1;
- }
-
for(;;) {
-
uint32_t size;
- int fd;
struct sockaddr_un addr;
socklen_t addrlen = sizeof(addr);
-
- result = fd = accept(pfd, (struct sockaddr *)&addr, &addrlen);
- if(result == -1) {
- PERROR("accept");
- continue;
- }
+ char trace_name[] = "auto";
+ char trace_type[] = "ustrelay";
for(;;) {
struct trctl_msg msg;
- unsigned char dontclose=0;
+ int len;
- result = read(fd, &msg.size, sizeof(msg.size));
+ result = len = recvfrom(pfd, recvbuf, sizeof(recvbuf), 0, &addr, &addrlen);
if(result == -1) {
- PERROR("read");
+ PERROR("recvfrom");
continue;
}
- if(size > MAX_MSG_SIZE) {
- ERR("trctl message too big");
- break;
- }
+ if(recvbuf[len-1] == '\n')
+ recvbuf[len-1] = '\0';
- result = read(fd, &msg.type, sizeof(msg.type));
- if(result == -1) {
- PERROR("read");
- continue;
+ fprintf(stderr, "received a message! it's: %s\n", recvbuf);
+
+
+ if(!strcmp(recvbuf, "print_markers")) {
+ print_markers();
+ }
+ else if(!strcmp(recvbuf, "trace_setup")) {
+ DBG("trace setup");
+
+ result = ltt_trace_setup(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_setup failed");
+ return;
+ }
+
+ result = ltt_trace_set_type(trace_name, trace_type);
+ if(result < 0) {
+ ERR("ltt_trace_set_type failed");
+ return;
+ }
+ }
+ else if(!strcmp(recvbuf, "trace_alloc")) {
+ DBG("trace alloc");
+
+ result = ltt_trace_alloc(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_alloc failed");
+ return;
+ }
}
+ else if(!strcmp(recvbuf, "trace_start")) {
+ DBG("trace start");
+
+ result = ltt_trace_start(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_start failed");
+ return;
+ }
+ }
+ else if(!strcmp(recvbuf, "trace_stop")) {
+ DBG("trace stop");
+
+ result = ltt_trace_stop(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_stop failed");
+ return;
+ }
+ }
+ else if(!strcmp(recvbuf, "trace_destroy")) {
+
+ DBG("trace destroy");
- switch(msg.type) {
- case MSG_REGISTER_NOTIF:
- /* TODO: put it in notif mode */
- goto next_conn;
- };
+ result = ltt_trace_destroy(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_destroy failed");
+ return;
+ }
+ }
}
next_conn:;
}
int result;
static char listener_stack[16384];
- result = clone(listener_main, listener_stack+sizeof(listener_stack)-1, CLONE_FS | CLONE_FILES | CLONE_VM, NULL);
+ result = clone(listener_main, listener_stack+sizeof(listener_stack)-1, CLONE_FS | CLONE_FILES | CLONE_VM | CLONE_SIGHAND | CLONE_THREAD, NULL);
if(result == -1) {
perror("clone");
}
void sighandler(int sig)
{
DBG("sighandler");
- receive_commands();
+ create_listener();
}
/* Called by the app signal handler to chain it to us. */
struct sockaddr_un addr;
- result = fd = socket(PF_UNIX, SOCK_SEQPACKET, 0);
+ result = fd = socket(PF_UNIX, SOCK_DGRAM, 0);
if(result == -1) {
PERROR("socket");
return -1;
return 0;
}
-static void __attribute__((constructor)) init()
+static void auto_probe_connect(struct marker *m)
{
int result;
+ result = ltt_marker_connect(m->channel, m->name, "default");
+ if(result)
+ ERR("ltt_marker_connect");
+
+ DBG("just auto connected marker %s %s to probe default", m->channel, m->name);
+}
+
+static void __attribute__((constructor(101))) init0()
+{
+ DBG("UST_AUTOPROBE constructor");
+ if(getenv("UST_AUTOPROBE")) {
+ marker_set_new_marker_cb(auto_probe_connect);
+ }
+}
+
+static void fini(void);
+
+static void __attribute__((constructor(1000))) init()
+{
+ int result;
+
+ DBG("UST_TRACE constructor");
+
mypid = getpid();
+ if(getenv("UST_TRACE")) {
+ char trace_name[] = "auto";
+ char trace_type[] = "ustrelay";
+
+ DBG("starting early tracing");
+
+ /* Ensure marker control is initialized */
+ init_marker_control();
+
+ /* Ensure relay is initialized */
+ init_ustrelay_transport();
+
+ /* Ensure markers are initialized */
+ init_markers();
+
+ /* In case. */
+ ltt_channels_register("ust");
+
+ result = ltt_trace_setup(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_setup failed");
+ return;
+ }
+
+ result = ltt_trace_set_type(trace_name, trace_type);
+ if(result < 0) {
+ ERR("ltt_trace_set_type failed");
+ return;
+ }
+
+ result = ltt_trace_alloc(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_alloc failed");
+ return;
+ }
+
+ result = ltt_trace_start(trace_name);
+ if(result < 0) {
+ ERR("ltt_trace_start failed");
+ return;
+ }
+ start_consumer();
+ }
+
/* Must create socket before signal handler to prevent races
* on pfd variable.
*/
static void __attribute__((destructor)) fini()
{
+ int result;
+
+ /* if trace running, finish it */
+
+ DBG("destructor stopping traces");
+
+ result = ltt_trace_stop("auto");
+ if(result == -1) {
+ ERR("ltt_trace_stop error");
+ }
+
+ result = ltt_trace_destroy("auto");
+ if(result == -1) {
+ ERR("ltt_trace_destroy error");
+ }
+
+ /* FIXME: wait for the consumer to be done */
+ sleep(3);
+
destroy_socket();
}
projects / ust.git / blobdiff