Add multipoll and change tracectl to use it

diff --git a/libust/Makefile.am b/libust/Makefile.am
index cb5dc1291511d4a34a80439111dc466a58e19fa3..b4d059803fa5deb1078ac845068bf9ad5360ca75 100644 (file)
--- a/libust/Makefile.am
+++ b/libust/Makefile.am
@@ -17,6 +17,7 @@ libust_la_SOURCES = \
        tracercore.h \
        serialize.c \
        tracectl.c \
+       $(top_builddir)/libustcomm/multipoll.c \
        tracerconst.h \
        header-inline.h
 

diff --git a/libust/tracectl.c b/libust/tracectl.c
index 82f51c1acf776fcee4ee29378dd2f151e784aa14..2fccb1caab9d73deef930469a29cd842ccb50a50 100644 (file)
--- a/libust/tracectl.c
+++ b/libust/tracectl.c
@@ -35,6 +35,7 @@
 #include "ustcomm.h"
 #include "buffers.h"
 #include "marker-control.h"
+#include "multipoll.h"
 
 #define USTSIGNAL SIGIO
 
@@ -185,88 +186,67 @@ static void inform_consumer_daemon(const char *trace_name)
        ltt_unlock_traces();
 }
 
-void process_blocked_consumers(void)
+int process_blkd_consumer_act(void *priv, int fd, short events)
 {
-       int n_fds = 0;
-       struct pollfd *fds;
-       struct blocked_consumer *bc;
-       int idx = 0;
-       char inbuf;
        int result;
+       long consumed_old = 0;
+       char *reply;
+       struct blocked_consumer *bc = (struct blocked_consumer *) priv;
+       char inbuf;
 
-       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++;
-       }
-
-       while((result = poll(fds, n_fds, 0)) == -1 && errno == EINTR)
-               /* nothing */;
+       result = read(bc->fd_producer, &inbuf, 1);
        if(result == -1) {
-               PERROR("poll");
-               return;
+               PERROR("read");
+               return -1;
        }
+       if(result == 0) {
+               int res;
+               DBG("listener: got messsage that a buffer ended");
 
-       list_for_each_entry(bc, &blocked_consumers, list) {
-               if(fds[bc->tmp_poll_idx].revents) {
-                       long consumed_old = 0;
-                       char *reply;
+               res = close(bc->fd_producer);
+               if(res == -1) {
+                       PERROR("close");
+               }
 
-                       result = read(bc->fd_producer, &inbuf, 1);
-                       if(result == -1) {
-                               PERROR("read");
-                               continue;
-                       }
-                       if(result == 0) {
-                               int res;
-                               DBG("PRODUCER END");
+               list_del(&bc->list);
+
+               result = ustcomm_send_reply(&bc->server, "END", &bc->src);
+               if(result < 0) {
+                       ERR("ustcomm_send_reply failed");
+                       return -1;
+               }
 
-                               res = close(bc->fd_producer);
-                               if(res == -1) {
-                                       PERROR("close");
-                               }
+               return 0;
+       }
 
-                               list_del(&bc->list);
+       result = ust_buffers_get_subbuf(bc->buf, &consumed_old);
+       if(result == -EAGAIN) {
+               WARN("missed buffer?");
+               return 0;
+       }
+       else if(result < 0) {
+               ERR("ust_buffers_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);
+               return -1;
+       }
+       free(reply);
 
-                               result = ustcomm_send_reply(&bc->server, "END", &bc->src);
-                               if(result < 0) {
-                                       ERR("ustcomm_send_reply failed");
-                                       continue;
-                               }
+       list_del(&bc->list);
 
-                               continue;
-                       }
+       return 0;
+}
 
-                       result = ust_buffers_get_subbuf(bc->buf, &consumed_old);
-                       if(result == -EAGAIN) {
-                               WARN("missed buffer?");
-                               continue;
-                       }
-                       else if(result < 0) {
-                               DBG("ust_buffers_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);
+void blocked_consumers_add_to_mp(struct mpentries *ent)
+{
+       struct blocked_consumer *bc;
 
-                       list_del(&bc->list);
-               }
+       list_for_each_entry(bc, &blocked_consumers, list) {
+               multipoll_add(ent, bc->fd_producer, POLLIN, process_blkd_consumer_act, bc, NULL);
        }
 
 }
@@ -821,232 +801,212 @@ static void listener_cleanup(void *ptr)
        ustcomm_fini_app(&ustcomm_app, 0);
 }
 
-void *listener_main(void *p)
+int process_client_cmd(char *recvbuf, struct ustcomm_source *src)
 {
        int result;
+       char trace_name[] = "auto";
+       char trace_type[] = "ustrelay";
+       int len;
 
-       DBG("LISTENER");
+       DBG("received a message! it's: %s", recvbuf);
+       len = strlen(recvbuf);
 
-       pthread_cleanup_push(listener_cleanup, NULL);
+       if(!strcmp(recvbuf, "print_markers")) {
+               print_markers(stderr);
+       }
+       else if(!strcmp(recvbuf, "list_markers")) {
+               char *ptr;
+               size_t size;
+               FILE *fp;
 
-       for(;;) {
-               char trace_name[] = "auto";
-               char trace_type[] = "ustrelay";
-               char *recvbuf;
-               int len;
-               struct ustcomm_source src;
+               fp = open_memstream(&ptr, &size);
+               print_markers(fp);
+               fclose(fp);
 
-               process_blocked_consumers();
+               result = ustcomm_send_reply(&ustcomm_app.server, ptr, src);
 
-               result = ustcomm_app_recv_message(&ustcomm_app, &recvbuf, &src, 5);
+               free(ptr);
+       }
+       else if(!strcmp(recvbuf, "start")) {
+               /* start is an operation that setups the trace, allocates it and starts it */
+               result = ltt_trace_setup(trace_name);
                if(result < 0) {
-                       WARN("error in ustcomm_app_recv_message");
-                       continue;
-               }
-               else if(result == 0) {
-                       /* no message */
-                       continue;
+                       ERR("ltt_trace_setup failed");
+                       return -1;
                }
 
-               DBG("received a message! it's: %s", recvbuf);
-               len = strlen(recvbuf);
-
-               if(!strcmp(recvbuf, "print_markers")) {
-                       print_markers(stderr);
+               result = ltt_trace_set_type(trace_name, trace_type);
+               if(result < 0) {
+                       ERR("ltt_trace_set_type failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "list_markers")) {
-                       char *ptr;
-                       size_t size;
-                       FILE *fp;
-
-                       fp = open_memstream(&ptr, &size);
-                       print_markers(fp);
-                       fclose(fp);
-
-                       result = ustcomm_send_reply(&ustcomm_app.server, ptr, &src);
 
-                       free(ptr);
+               result = ltt_trace_alloc(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_alloc failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "start")) {
-                       /* start is an operation that setups the trace, allocates it and starts it */
-                       result = ltt_trace_setup(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_setup failed");
-                               return (void *)1;
-                       }
-
-                       result = ltt_trace_set_type(trace_name, trace_type);
-                       if(result < 0) {
-                               ERR("ltt_trace_set_type failed");
-                               return (void *)1;
-                       }
-
-                       result = ltt_trace_alloc(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_alloc failed");
-                               return (void *)1;
-                       }
 
-                       inform_consumer_daemon(trace_name);
+               inform_consumer_daemon(trace_name);
 
-                       result = ltt_trace_start(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_start failed");
-                               continue;
-                       }
+               result = ltt_trace_start(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_start failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "trace_setup")) {
-                       DBG("trace setup");
-
-                       result = ltt_trace_setup(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_setup failed");
-                               return (void *)1;
-                       }
+       }
+       else if(!strcmp(recvbuf, "trace_setup")) {
+               DBG("trace setup");
 
-                       result = ltt_trace_set_type(trace_name, trace_type);
-                       if(result < 0) {
-                               ERR("ltt_trace_set_type failed");
-                               return (void *)1;
-                       }
+               result = ltt_trace_setup(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_setup failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "trace_alloc")) {
-                       DBG("trace alloc");
 
-                       result = ltt_trace_alloc(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_alloc failed");
-                               return (void *)1;
-                       }
-                       inform_consumer_daemon(trace_name);
+               result = ltt_trace_set_type(trace_name, trace_type);
+               if(result < 0) {
+                       ERR("ltt_trace_set_type failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "trace_create")) {
-                       DBG("trace create");
-
-                       result = ltt_trace_setup(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_setup failed");
-                               return (void *)1;
-                       }
+       }
+       else if(!strcmp(recvbuf, "trace_alloc")) {
+               DBG("trace alloc");
 
-                       result = ltt_trace_set_type(trace_name, trace_type);
-                       if(result < 0) {
-                               ERR("ltt_trace_set_type failed");
-                               return (void *)1;
-                       }
+               result = ltt_trace_alloc(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_alloc failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "trace_start")) {
-                       DBG("trace start");
-
-                       result = ltt_trace_alloc(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_alloc failed");
-                               return (void *)1;
-                       }
-                       if(!result) {
-                               inform_consumer_daemon(trace_name);
-                       }
+               inform_consumer_daemon(trace_name);
+       }
+       else if(!strcmp(recvbuf, "trace_create")) {
+               DBG("trace create");
 
-                       result = ltt_trace_start(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_start failed");
-                               continue;
-                       }
+               result = ltt_trace_setup(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_setup failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "trace_stop")) {
-                       DBG("trace stop");
 
-                       result = ltt_trace_stop(trace_name);
-                       if(result < 0) {
-                               ERR("ltt_trace_stop failed");
-                               return (void *)1;
-                       }
+               result = ltt_trace_set_type(trace_name, trace_type);
+               if(result < 0) {
+                       ERR("ltt_trace_set_type failed");
+                       return -1;
                }
-               else if(!strcmp(recvbuf, "trace_destroy")) {
-
-                       DBG("trace destroy");
+       }
+       else if(!strcmp(recvbuf, "trace_start")) {
+               DBG("trace start");
 
-                       result = ltt_trace_destroy(trace_name, 0);
-                       if(result < 0) {
-                               ERR("ltt_trace_destroy failed");
-                               return (void *)1;
-                       }
-               }
-               else if(nth_token_is(recvbuf, "get_shmid", 0) == 1) {
-                       do_cmd_get_shmid(recvbuf, &src);
+               result = ltt_trace_alloc(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_alloc failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "get_n_subbufs", 0) == 1) {
-                       do_cmd_get_n_subbufs(recvbuf, &src);
+               if(!result) {
+                       inform_consumer_daemon(trace_name);
                }
-               else if(nth_token_is(recvbuf, "get_subbuf_size", 0) == 1) {
-                       do_cmd_get_subbuf_size(recvbuf, &src);
+
+               result = ltt_trace_start(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_start failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
-                       char *libfile;
+       }
+       else if(!strcmp(recvbuf, "trace_stop")) {
+               DBG("trace stop");
 
-                       libfile = nth_token(recvbuf, 1);
+               result = ltt_trace_stop(trace_name);
+               if(result < 0) {
+                       ERR("ltt_trace_stop failed");
+                       return -1;
+               }
+       }
+       else if(!strcmp(recvbuf, "trace_destroy")) {
 
-                       DBG("load_probe_lib loading %s", libfile);
+               DBG("trace destroy");
 
-                       free(libfile);
-               }
-               else if(nth_token_is(recvbuf, "get_subbuffer", 0) == 1) {
-                       do_cmd_get_subbuffer(recvbuf, &src);
-               }
-               else if(nth_token_is(recvbuf, "put_subbuffer", 0) == 1) {
-                       do_cmd_put_subbuffer(recvbuf, &src);
-               }
-               else if(nth_token_is(recvbuf, "set_subbuf_size", 0) == 1) {
-                       do_cmd_set_subbuf_size(recvbuf, &src);
-               }
-               else if(nth_token_is(recvbuf, "set_subbuf_num", 0) == 1) {
-                       do_cmd_set_subbuf_num(recvbuf, &src);
+               result = ltt_trace_destroy(trace_name, 0);
+               if(result < 0) {
+                       ERR("ltt_trace_destroy failed");
+                       return -1;
                }
-               else if(nth_token_is(recvbuf, "enable_marker", 0) == 1) {
-                       char *channel_slash_name = nth_token(recvbuf, 1);
-                       char channel_name[256]="";
-                       char marker_name[256]="";
+       }
+       else if(nth_token_is(recvbuf, "get_shmid", 0) == 1) {
+               do_cmd_get_shmid(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "get_n_subbufs", 0) == 1) {
+               do_cmd_get_n_subbufs(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "get_subbuf_size", 0) == 1) {
+               do_cmd_get_subbuf_size(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "load_probe_lib", 0) == 1) {
+               char *libfile;
 
-                       result = sscanf(channel_slash_name, "%255[^/]/%255s", channel_name, marker_name);
+               libfile = nth_token(recvbuf, 1);
 
-                       if(channel_name == NULL || marker_name == NULL) {
-                               WARN("invalid marker name");
-                               goto next_cmd;
-                       }
+               DBG("load_probe_lib loading %s", libfile);
 
-                       result = ltt_marker_connect(channel_name, marker_name, "default");
-                       if(result < 0) {
-                               WARN("could not enable marker; channel=%s, name=%s", channel_name, marker_name);
-                       }
+               free(libfile);
+       }
+       else if(nth_token_is(recvbuf, "get_subbuffer", 0) == 1) {
+               do_cmd_get_subbuffer(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "put_subbuffer", 0) == 1) {
+               do_cmd_put_subbuffer(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "set_subbuf_size", 0) == 1) {
+               do_cmd_set_subbuf_size(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "set_subbuf_num", 0) == 1) {
+               do_cmd_set_subbuf_num(recvbuf, src);
+       }
+       else if(nth_token_is(recvbuf, "enable_marker", 0) == 1) {
+               char *channel_slash_name = nth_token(recvbuf, 1);
+               char channel_name[256]="";
+               char marker_name[256]="";
+
+               result = sscanf(channel_slash_name, "%255[^/]/%255s", channel_name, marker_name);
+
+               if(channel_name == NULL || marker_name == NULL) {
+                       WARN("invalid marker name");
+                       goto next_cmd;
                }
-               else if(nth_token_is(recvbuf, "disable_marker", 0) == 1) {
-                       char *channel_slash_name = nth_token(recvbuf, 1);
-                       char *marker_name;
-                       char *channel_name;
 
-                       result = sscanf(channel_slash_name, "%a[^/]/%as", &channel_name, &marker_name);
+               result = ltt_marker_connect(channel_name, marker_name, "default");
+               if(result < 0) {
+                       WARN("could not enable marker; channel=%s, name=%s", channel_name, marker_name);
+               }
+       }
+       else if(nth_token_is(recvbuf, "disable_marker", 0) == 1) {
+               char *channel_slash_name = nth_token(recvbuf, 1);
+               char *marker_name;
+               char *channel_name;
 
-                       if(marker_name == NULL) {
-                       }
+               result = sscanf(channel_slash_name, "%a[^/]/%as", &channel_name, &marker_name);
 
-                       result = ltt_marker_disconnect(channel_name, marker_name, "default");
-                       if(result < 0) {
-                               WARN("could not disable marker; channel=%s, name=%s", channel_name, marker_name);
-                       }
+               if(marker_name == NULL) {
                }
-               else if(nth_token_is(recvbuf, "get_pidunique", 0) == 1) {
-                       char *reply;
 
-                       asprintf(&reply, "%lld", pidunique);
+               result = ltt_marker_disconnect(channel_name, marker_name, "default");
+               if(result < 0) {
+                       WARN("could not disable marker; channel=%s, name=%s", channel_name, marker_name);
+               }
+       }
+       else if(nth_token_is(recvbuf, "get_pidunique", 0) == 1) {
+               char *reply;
 
-                       result = ustcomm_send_reply(&ustcomm_app.server, reply, &src);
-                       if(result) {
-                               ERR("listener: get_pidunique: ustcomm_send_reply failed");
-                               goto next_cmd;
-                       }
+               asprintf(&reply, "%lld", pidunique);
 
-                       free(reply);
+               result = ustcomm_send_reply(&ustcomm_app.server, reply, src);
+               if(result) {
+                       ERR("listener: get_pidunique: ustcomm_send_reply failed");
+                       goto next_cmd;
                }
+
+               free(reply);
+       }
 //             else if(nth_token_is(recvbuf, "get_notifications", 0) == 1) {
 //                     struct ust_trace *trace;
 //                     char trace_name[] = "auto";
@@ -1078,7 +1038,7 @@ void *listener_main(void *p)
 //                                     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);
+//                                     result = fd = ustcomm_app_detach_client(&ustcomm_app, src);
 //                                     if(result == -1) {
 //                                             ERR("ustcomm_app_detach_client failed");
 //                                             goto next_cmd;
@@ -1096,12 +1056,37 @@ void *listener_main(void *p)
 //
 //                     free(channel_name);
 //             }
-               else {
-                       ERR("unable to parse message: %s", recvbuf);
+       else {
+               ERR("unable to parse message: %s", recvbuf);
+       }
+
+next_cmd:
+
+       return 0;
+}
+
+void *listener_main(void *p)
+{
+       int result;
+
+       DBG("LISTENER");
+
+       pthread_cleanup_push(listener_cleanup, NULL);
+
+       for(;;) {
+               struct mpentries mpent;
+
+               multipoll_init(&mpent);
+
+               blocked_consumers_add_to_mp(&mpent);
+               ustcomm_mp_add_app_clients(&mpent, &ustcomm_app, process_client_cmd);
+
+               result = multipoll_poll(&mpent, -1);
+               if(result == -1) {
+                       ERR("error in multipoll_poll");
                }
 
-       next_cmd:
-               free(recvbuf);
+               multipoll_destroy(&mpent);
        }
 
        pthread_cleanup_pop(1);

diff --git a/libustcomm/Makefile.am b/libustcomm/Makefile.am
index fa24cfb133bd94f514b410617634ef3e1b25cae7..baebd8461f82c7a226ef85415515759a35a31572 100644 (file)
--- a/libustcomm/Makefile.am
+++ b/libustcomm/Makefile.am
@@ -3,6 +3,8 @@ AM_CPPFLAGS = -I$(top_srcdir)/include
 lib_LTLIBRARIES = libustcomm.la
 libustcomm_la_SOURCES = \
        ustcomm.h \
-       ustcomm.c
+       ustcomm.c \
+       multipoll.h \
+       multipoll.c
 libustcomm_la_LDFLAGS = -no-undefined -version-info 0:0:0 -static
 libustcomm_la_CFLAGS = -DUST_COMPONENT="libustcomm" -fPIC

diff --git a/libustcomm/multipoll.c b/libustcomm/multipoll.c
new file mode 100644 (file)
index 0000000..db9c2ec
--- /dev/null
+++ b/libustcomm/multipoll.c
@@ -0,0 +1,95 @@
+/*
+ * multipoll.c
+ *
+ * Copyright (C) 2010 - Pierre-Marc Fournier (pierre-marc dot fournier at polymtl dot ca)
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+ */
+
+#include <poll.h>
+#include <stdlib.h>
+#include "multipoll.h"
+#include "usterr.h"
+
+#define INITIAL_N_AVAIL 16
+
+int multipoll_init(struct mpentries *ent)
+{
+       ent->n_used = 0;
+       ent->n_avail = INITIAL_N_AVAIL;
+
+       ent->pollfds = (struct pollfd *) malloc(sizeof(struct pollfd) * 16);
+       ent->extras = (struct pollfd_extra *) malloc(sizeof(struct pollfd_extra) * 16);
+
+       return 0;
+}
+
+int multipoll_destroy(struct mpentries *ent)
+{
+       int i;
+
+       for(i=0; i<ent->n_used; i++) {
+               if(ent->extras[i].destroy_priv) {
+                       ent->extras[i].destroy_priv(ent->extras[i].priv);
+               }
+       }
+
+       free(ent->pollfds);
+       free(ent->extras);
+
+       return 0;
+}
+
+int multipoll_add(struct mpentries *ent, int fd, short events, int (*func)(void *priv, int fd, short events), void *priv, int (*destroy_priv)(void *))
+{
+       int cur;
+
+       if(ent->n_used == ent->n_avail) {
+               ent->n_avail *= 2;
+               ent->pollfds = (struct pollfd *) realloc(ent->pollfds, sizeof(struct pollfd) * ent->n_avail);
+               ent->extras = (struct pollfd_extra *) realloc(ent->extras, sizeof(struct pollfd_extra) * ent->n_avail);
+       }
+
+       cur = ent->n_used;
+       ent->n_used++;
+
+       ent->pollfds[cur].fd = fd;
+       ent->pollfds[cur].events = events;
+       ent->extras[cur].func = func;
+       ent->extras[cur].priv = priv;
+       ent->extras[cur].destroy_priv = destroy_priv;
+
+       return 0;
+}
+
+int multipoll_poll(struct mpentries *ent, int timeout)
+{
+       int result;
+       int i;
+
+       result = poll(ent->pollfds, ent->n_used, timeout);
+       if(result == -1) {
+               PERROR("poll");
+               return -1;
+       }
+
+       for(i=0; i<ent->n_used; i++) {
+               if(ent->pollfds[i].revents) {
+                       ent->extras[i].func(ent->extras[i].priv, ent->pollfds[i].fd, ent->pollfds[i].revents);
+               }
+       }
+
+       return 0;
+}

diff --git a/libustcomm/multipoll.h b/libustcomm/multipoll.h
new file mode 100644 (file)
index 0000000..8a0124f
--- /dev/null
+++ b/libustcomm/multipoll.h
@@ -0,0 +1,44 @@
+/*
+ * multipoll.h
+ *
+ * Copyright (C) 2010 - Pierre-Marc Fournier (pierre-marc dot fournier at polymtl dot ca)
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA
+ */
+
+#ifndef UST_MULTIPOLL_H
+#define UST_MULTIPOLL_H
+
+struct pollfd_extra {
+       int (*func)(void *priv, int fd, short events);
+       void *priv;
+
+       int (*destroy_priv)(void *priv);
+};
+
+struct mpentries {
+       struct pollfd *pollfds;
+       struct pollfd_extra *extras;
+
+       int n_used;
+       int n_avail;
+};
+
+extern int multipoll_init(struct mpentries *ent);
+extern int multipoll_add(struct mpentries *ent, int fd, short events, int (*func)(void *priv, int fd, short events), void *priv, int (*destroy_priv)(void *));
+extern int multipoll_destroy(struct mpentries *ent);
+extern int multipoll_poll(struct mpentries *ent, int timeout);
+
+#endif /* UST_MULTIPOLL_H */

diff --git a/libustcomm/ustcomm.c b/libustcomm/ustcomm.c
index cb72db65733eb5353c7d10a74018ff856874c648..c76413861cb418cf142c75b14192d85e1216b5a5 100644 (file)
--- a/libustcomm/ustcomm.c
+++ b/libustcomm/ustcomm.c
@@ -33,6 +33,7 @@
 #include "ustcomm.h"
 #include "usterr.h"
 #include "share.h"
+#include "multipoll.h"
 
 #define UNIX_PATH_MAX 108
 
@@ -885,3 +886,88 @@ char *nth_token(const char *str, int tok_no)
        return retval;
 }
 
+/* Callback from multipoll.
+ * Receive a new connection on the listening socket.
+ */
+
+static int process_mp_incoming_conn(void *priv, int fd, short events)
+{
+       struct ustcomm_connection *newconn;
+       struct ustcomm_server *server = (struct ustcomm_server *) priv;
+       int newfd;
+       int result;
+
+       result = newfd = accept(server->listen_fd, NULL, NULL);
+       if(result == -1) {
+               PERROR("accept");
+               return -1;
+       }
+
+       newconn = (struct ustcomm_connection *) malloc(sizeof(struct ustcomm_connection));
+       if(newconn == NULL) {
+               ERR("malloc returned NULL");
+               return -1;
+       }
+
+       ustcomm_init_connection(newconn);
+       newconn->fd = newfd;
+
+       list_add(&newconn->list, &server->connections);
+
+       return 0;
+}
+
+/* Callback from multipoll.
+ * Receive a message on an existing connection.
+ */
+
+static int process_mp_conn_msg(void *priv, int fd, short revents)
+{
+       struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) priv;
+       int result;
+       char *msg;
+       struct ustcomm_source src;
+
+       if(revents) {
+               src.fd = fd;
+
+               result = recv_message_conn(mpinfo->conn, &msg);
+               if(result == -1) {
+                       ERR("error in recv_message_conn");
+               }
+
+               else if(result == 0) {
+                       /* connection finished */
+                       ustcomm_close_app(mpinfo->conn);
+                       list_del(&mpinfo->conn->list);
+                       free(mpinfo->conn);
+               }
+               else {
+                       mpinfo->cb(msg, &src);
+                       free(msg);
+               }
+       }
+
+       return 0;
+}
+
+int free_ustcomm_client_poll(void *data)
+{
+       free(data);
+       return 0;
+}
+
+void ustcomm_mp_add_app_clients(struct mpentries *ent, struct ustcomm_app *app, int (*cb)(struct ustcomm_connection *conn, char *msg))
+{
+       struct ustcomm_connection *conn;
+
+       /* add listener socket */
+       multipoll_add(ent, app->server.listen_fd, POLLIN, process_mp_incoming_conn, &app->server, NULL);
+
+       list_for_each_entry(conn, &app->server.connections, list) {
+               struct ustcomm_multipoll_conn_info *mpinfo = (struct ustcomm_multipoll_conn_info *) malloc(sizeof(struct ustcomm_multipoll_conn_info));
+               mpinfo->conn = conn;
+               mpinfo->cb = cb;
+               multipoll_add(ent, conn->fd, POLLIN, process_mp_conn_msg, mpinfo, free_ustcomm_client_poll);
+       }
+}

diff --git a/libustcomm/ustcomm.h b/libustcomm/ustcomm.h
index 721d1ab08ce66cf7ebcdd5ef6021f4e5fc9f0f07..87ed322be43bec8f4e0ccdd3d28c8c430c820b85 100644 (file)
--- a/libustcomm/ustcomm.h
+++ b/libustcomm/ustcomm.h
@@ -22,6 +22,7 @@
 #include <sys/un.h>
 
 #include "kcompat.h"
+#include "multipoll.h"
 
 #define SOCK_DIR "/tmp/ust-app-socks"
 #define UST_SIGNAL SIGIO
@@ -58,6 +59,11 @@ struct ustcomm_source {
        void *priv;
 };
 
+struct ustcomm_multipoll_conn_info {
+       struct ustcomm_connection *conn;
+       int (*cb)(char *msg, struct ustcomm_source *src);
+};
+
 extern char *strdup_malloc(const char *s);
 
 //int send_message_pid(pid_t pid, const char *msg, char **reply);
@@ -79,6 +85,7 @@ extern int ustcomm_send_request(struct ustcomm_connection *conn, const char *req
 extern int ustcomm_send_reply(struct ustcomm_server *server, char *msg, struct ustcomm_source *src);
 extern int ustcomm_disconnect(struct ustcomm_connection *conn);
 extern int ustcomm_close_all_connections(struct ustcomm_server *server);
+extern void ustcomm_mp_add_app_clients(struct mpentries *ent, struct ustcomm_app *app, int (*cb)(struct ustcomm_connection *conn, char *msg));
 
 extern int nth_token_is(const char *str, const char *token, int tok_no);