#include <sys/un.h>
#include <sched.h>
#include <fcntl.h>
+#include <poll.h>
#include "marker.h"
#include "tracer.h"
#include "localerr.h"
#include "ustcomm.h"
+#include "relay.h" /* FIXME: remove */
//#define USE_CLONE
-#define UNIX_PATH_MAX 108
-
-#define SOCKETDIR "/tmp/socks"
-#define SOCKETDIRLEN sizeof(SOCKETDIR)
#define USTSIGNAL SIGIO
#define MAX_MSG_SIZE (100)
char consumer_stack[10000];
+struct list_head blocked_consumers = LIST_HEAD_INIT(blocked_consumers);
+
static struct ustcomm_app ustcomm_app;
struct tracecmd { /* no padding */
char payload[94];
};
-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;
- }
+struct blocked_consumer {
+ int fd_consumer;
+ int fd_producer;
+ int tmp_poll_idx;
- 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];
+ /* args to ustcomm_send_reply */
+ struct ustcomm_server server;
+ struct ustcomm_source src;
- 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, 00600);
- if(result == -1) {
- perror("open");
- return -1;
- }
- CPRINTF("\topened trace file %s", tmp);
-
- }
- CPRINTF("done opening trace files");
+ /* args to ltt_do_get_subbuf */
+ struct rchan_buf *rbuf;
+ struct ltt_channel_buf_struct *lttbuf;
- 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);
- }
-}
-
-void start_consumer(void)
-{
-#ifdef USE_CLONE
- 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");
- }
-#else
- pthread_t thread;
-
- pthread_create(&thread, NULL, consumer, NULL);
-#endif
-}
+ struct list_head list;
+};
static void print_markers(void)
{
}
}
-#define CONSUMER_DAEMON_SOCK SOCKETDIR "/ustd"
-
static int inform_consumer_daemon(void)
{
ustcomm_request_consumer(getpid(), "metadata");
ustcomm_request_consumer(getpid(), "ust");
}
+void process_blocked_consumers(void)
+{
+ int n_fds = 0;
+ struct pollfd *fds;
+ struct blocked_consumer *bc;
+ int idx = 0;
+ char inbuf;
+ int result;
+
+ list_for_each_entry(bc, &blocked_consumers, list) {
+ n_fds++;
+ }
+
+ fds = (struct pollfd *) malloc(n_fds * sizeof(struct pollfd));
+ if(fds == NULL) {
+ ERR("malloc returned NULL");
+ return;
+ }
+
+ list_for_each_entry(bc, &blocked_consumers, list) {
+ fds[idx].fd = bc->fd_producer;
+ fds[idx].events = POLLIN;
+ bc->tmp_poll_idx = idx;
+ idx++;
+ }
+
+ result = poll(fds, n_fds, 0);
+ if(result == -1) {
+ PERROR("poll");
+ return -1;
+ }
+
+ list_for_each_entry(bc, &blocked_consumers, list) {
+ if(fds[bc->tmp_poll_idx].revents) {
+ long consumed_old = 0;
+ char *reply;
+
+ result = read(bc->fd_producer, &inbuf, 1);
+ if(result == -1) {
+ PERROR("read");
+ continue;
+ }
+ if(result == 0) {
+ DBG("PRODUCER END");
+
+ close(bc->fd_producer);
+
+ __list_del(bc->list.prev, bc->list.next);
+
+ result = ustcomm_send_reply(&bc->server, "END", &bc->src);
+ if(result < 0) {
+ ERR("ustcomm_send_reply failed");
+ continue;
+ }
+
+ continue;
+ }
+
+ result = ltt_do_get_subbuf(bc->rbuf, bc->lttbuf, &consumed_old);
+ if(result == -EAGAIN) {
+ WARN("missed buffer?");
+ continue;
+ }
+ else if(result < 0) {
+ DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
+ }
+ asprintf(&reply, "%s %ld", "OK", consumed_old);
+ result = ustcomm_send_reply(&bc->server, reply, &bc->src);
+ if(result < 0) {
+ ERR("ustcomm_send_reply failed");
+ free(reply);
+ continue;
+ }
+ free(reply);
+
+ __list_del(bc->list.prev, bc->list.next);
+ }
+ }
+
+}
+
int listener_main(void *p)
{
int result;
int len;
struct ustcomm_source src;
- result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, -1);
- if(result <= 0) {
+ process_blocked_consumers();
+
+ result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, 5);
+ if(result < 0) {
WARN("error in ustcomm_app_recv_message");
continue;
}
+ else if(result == 0) {
+ /* no message */
+ continue;
+ }
DBG("received a message! it's: %s\n", recvbuf);
len = strlen(recvbuf);
for(i=0; i<trace->nr_channels; i++) {
struct rchan *rchan = trace->channels[i].trans_channel_data;
struct rchan_buf *rbuf = rchan->buf;
+ struct ltt_channel_struct *ltt_channel = (struct ltt_channel_struct *)rchan->private_data;
+ struct ltt_channel_buf_struct *ltt_buf = ltt_channel->buf;
if(!strcmp(trace->channels[i].channel_name, channel_name)) {
char *reply;
DBG("the shmid for the requested channel is %d", rbuf->shmid);
- asprintf(&reply, "%d", rbuf->shmid);
+ DBG("the shmid for its buffer structure is %d", ltt_channel->buf_shmid);
+ asprintf(&reply, "%d %d", rbuf->shmid, ltt_channel->buf_shmid);
result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
if(result) {
struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
char *reply;
long consumed_old=0;
+ int fd;
+ struct blocked_consumer *bc;
- result = ltt_do_get_subbuf(rbuf, lttbuf, &consumed_old);
- if(result < 0) {
- DBG("ltt_do_get_subbuf: error: %s", strerror(-result));
- asprintf(&reply, "%s %ld", "UNAVAIL", 0);
- }
- else {
- DBG("ltt_do_get_subbuf: success");
- asprintf(&reply, "%s %ld", "OK", consumed_old);
- }
-
- result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
- if(result) {
- ERR("listener: get_subbuf: ustcomm_send_reply failed");
+ bc = (struct blocked_consumer *) malloc(sizeof(struct blocked_consumer));
+ if(bc == NULL) {
+ ERR("malloc returned NULL");
goto next_cmd;
}
+ bc->fd_consumer = src.fd;
+ bc->fd_producer = lttbuf->data_ready_fd_read;
+ bc->rbuf = rbuf;
+ bc->lttbuf = lttbuf;
+ bc->src = src;
+ bc->server = ustcomm_app.server;
- free(reply);
+ list_add(&bc->list, &blocked_consumers);
break;
}
free(channel_name);
free(consumed_old_str);
}
+// else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
+// struct ltt_trace_struct *trace;
+// char trace_name[] = "auto";
+// int i;
+// char *channel_name;
+//
+// DBG("get_notifications");
+//
+// channel_name = strdup_malloc(nth_token(recvbuf, 1));
+// if(channel_name == NULL) {
+// ERR("put_subbuf_size: cannot parse channel");
+// goto next_cmd;
+// }
+//
+// ltt_lock_traces();
+// trace = _ltt_trace_find(trace_name);
+// ltt_unlock_traces();
+//
+// if(trace == NULL) {
+// CPRINTF("cannot find trace!");
+// return 1;
+// }
+//
+// for(i=0; i<trace->nr_channels; i++) {
+// struct rchan *rchan = trace->channels[i].trans_channel_data;
+// int fd;
+//
+// if(!strcmp(trace->channels[i].channel_name, channel_name)) {
+// struct rchan_buf *rbuf = rchan->buf;
+// struct ltt_channel_buf_struct *lttbuf = trace->channels[i].buf;
+//
+// result = fd = ustcomm_app_detach_client(&ustcomm_app, &src);
+// if(result == -1) {
+// ERR("ustcomm_app_detach_client failed");
+// goto next_cmd;
+// }
+//
+// lttbuf->wake_consumer_arg = (void *) fd;
+//
+// smp_wmb();
+//
+// lttbuf->call_wake_consumer = 1;
+//
+// break;
+// }
+// }
+//
+// free(channel_name);
+// }
else {
ERR("unable to parse message: %s", recvbuf);
}
static void destroy_socket(void)
{
- int result;
-
- if(mysocketfile[0] == '\0')
- return;
-
- result = unlink(mysocketfile);
- if(result == -1) {
- PERROR("unlink");
- }
+// int result;
+//
+// if(mysocketfile[0] == '\0')
+// return;
+//
+// result = unlink(mysocketfile);
+// if(result == -1) {
+// PERROR("unlink");
+// }
}
static int init_signal_handler(void)
}
/* FIXME: wait for the consumer to be done */
- DBG("waiting 5 sec for consume");
- sleep(5);
+ //DBG("waiting 5 sec for consume");
+ //sleep(5);
destroy_socket();
}
projects / ust.git / blobdiff