From 0e4b45ace8b034dc4352a30d89b74f5e59e2e27d Mon Sep 17 00:00:00 2001 From: Pierre-Marc Fournier Date: Thu, 11 Mar 2010 18:50:43 -0500 Subject: [PATCH] Add multipoll and change tracectl to use it --- libust/Makefile.am | 1 + libust/tracectl.c | 493 ++++++++++++++++++++--------------------- libustcomm/Makefile.am | 4 +- libustcomm/multipoll.c | 95 ++++++++ libustcomm/multipoll.h | 44 ++++ libustcomm/ustcomm.c | 86 +++++++ libustcomm/ustcomm.h | 7 + 7 files changed, 475 insertions(+), 255 deletions(-) create mode 100644 libustcomm/multipoll.c create mode 100644 libustcomm/multipoll.h diff --git a/libust/Makefile.am b/libust/Makefile.am index cb5dc129..b4d05980 100644 --- 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 82f51c1a..2fccb1ca 100644 --- 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 fa24cfb1..baebd846 100644 --- 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 index 00000000..db9c2ec9 --- /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 +#include +#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; in_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; in_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 index 00000000..8a0124f5 --- /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 cb72db65..c7641386 100644 --- 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 721d1ab0..87ed322b 100644 --- a/libustcomm/ustcomm.h +++ b/libustcomm/ustcomm.h @@ -22,6 +22,7 @@ #include #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); -- 2.34.1