Use WaitForMultipleObjects for polling on windows. This works on all kinds of objects, e.g. sockets, files, especially ioctl calls to the kernel. One additional paramater is passed down in poll_fd_wait() to help WaitForMultipleObjects. latch is signaled with event, to be waited/polled by WaitForMultipleObjects() as well. Changed array of fds to hmap to check for duplicate fds.
Signed-off-by: Linda Sun <[email protected]> --- lib/dpif-linux.c | 2 +- lib/fatal-signal.c | 2 +- lib/latch.c | 37 ++++++++++++++- lib/latch.h | 3 ++ lib/netdev-linux.c | 2 +- lib/netlink-socket.c | 2 +- lib/poll-loop.c | 121 +++++++++++++++++++++++++++++++++++++++++--------- lib/poll-loop.h | 4 +- lib/process.c | 2 +- lib/signals.c | 2 +- lib/socket-util.c | 2 +- lib/stream-fd.c | 6 +-- lib/timeval.c | 15 +++++-- lib/timeval.h | 4 +- tests/test-netflow.c | 2 +- tests/test-sflow.c | 2 +- 16 files changed, 167 insertions(+), 41 deletions(-) diff --git a/lib/dpif-linux.c b/lib/dpif-linux.c index 25715f4..a8246f3 100644 --- a/lib/dpif-linux.c +++ b/lib/dpif-linux.c @@ -1482,7 +1482,7 @@ dpif_linux_recv_wait(struct dpif *dpif_) ovs_mutex_lock(&dpif->upcall_lock); if (dpif->epoll_fd >= 0) { - poll_fd_wait(dpif->epoll_fd, POLLIN); + poll_fd_wait(dpif->epoll_fd, 0, POLLIN); } ovs_mutex_unlock(&dpif->upcall_lock); } diff --git a/lib/fatal-signal.c b/lib/fatal-signal.c index e980f4b..052ca26 100644 --- a/lib/fatal-signal.c +++ b/lib/fatal-signal.c @@ -182,7 +182,7 @@ void fatal_signal_wait(void) { fatal_signal_init(); - poll_fd_wait(signal_fds[0], POLLIN); + poll_fd_wait(signal_fds[0], 0, POLLIN); } static void diff --git a/lib/latch.c b/lib/latch.c index bf518b9..5bc97b4 100644 --- a/lib/latch.c +++ b/lib/latch.c @@ -27,15 +27,28 @@ void latch_init(struct latch *latch) { +#ifndef WIN32 xpipe_nonblocking(latch->fds); + latch->wevent = 0; +#else + latch->fds[0] = 0; + latch->is_set = FALSE; + latch->wevent = CreateEvent(NULL, TRUE, FALSE, NULL); +#endif } /* Destroys 'latch'. */ void latch_destroy(struct latch *latch) { +#ifndef WIN32 close(latch->fds[0]); close(latch->fds[1]); +#else + latch->fds[0] = 0; + latch->is_set = FALSE; + CloseHandle(latch->wevent); +#endif } /* Resets 'latch' to the unset state. Returns true if 'latch' was previously @@ -43,9 +56,18 @@ latch_destroy(struct latch *latch) bool latch_poll(struct latch *latch) { +#ifndef WIN32 char buffer[_POSIX_PIPE_BUF]; return read(latch->fds[0], buffer, sizeof buffer) > 0; +#else + bool is_set; + + is_set = latch->is_set; + latch->is_set = FALSE; + ResetEvent(latch->wevent); + return is_set; +#endif } /* Sets 'latch'. @@ -55,7 +77,12 @@ latch_poll(struct latch *latch) void latch_set(struct latch *latch) { +#ifndef WIN32 ignore(write(latch->fds[1], "", 1)); +#else + latch->is_set = TRUE; + SetEvent(latch->wevent); +#endif } /* Returns true if 'latch' is set, false otherwise. Does not reset 'latch' @@ -63,6 +90,7 @@ latch_set(struct latch *latch) bool latch_is_set(const struct latch *latch) { +#ifndef WIN32 struct pollfd pfd; int retval; @@ -73,6 +101,9 @@ latch_is_set(const struct latch *latch) } while (retval < 0 && errno == EINTR); return pfd.revents & POLLIN; +#else + return latch->is_set; +#endif } /* Causes the next poll_block() to wake up when 'latch' is set. @@ -83,5 +114,9 @@ latch_is_set(const struct latch *latch) void latch_wait_at(const struct latch *latch, const char *where) { - poll_fd_wait_at(latch->fds[0], POLLIN, where); +#ifndef WIN32 + poll_fd_wait_at(latch->fds[0], latch->wevent, POLLIN, where); +#else + poll_fd_wait_at(latch->fds[0], latch->wevent, POLLIN, where); +#endif } diff --git a/lib/latch.h b/lib/latch.h index 0b6e8a3..3c27f86 100644 --- a/lib/latch.h +++ b/lib/latch.h @@ -24,9 +24,12 @@ #include <stdbool.h> #include "util.h" +#include "ovs-thread.h" struct latch { int fds[2]; + uint32_t wevent; + bool is_set; }; void latch_init(struct latch *); diff --git a/lib/netdev-linux.c b/lib/netdev-linux.c index 3e0da48..755e9da 100644 --- a/lib/netdev-linux.c +++ b/lib/netdev-linux.c @@ -878,7 +878,7 @@ static void netdev_linux_rx_wait(struct netdev_rx *rx_) { struct netdev_rx_linux *rx = netdev_rx_linux_cast(rx_); - poll_fd_wait(rx->fd, POLLIN); + poll_fd_wait(rx->fd, 0, POLLIN); } static int diff --git a/lib/netlink-socket.c b/lib/netlink-socket.c index 4bd6d36..c015d93 100644 --- a/lib/netlink-socket.c +++ b/lib/netlink-socket.c @@ -802,7 +802,7 @@ nl_dump_done(struct nl_dump *dump) void nl_sock_wait(const struct nl_sock *sock, short int events) { - poll_fd_wait(sock->fd, events); + poll_fd_wait(sock->fd, 0, events); } /* Returns the underlying fd for 'sock', for use in "poll()"-like operations diff --git a/lib/poll-loop.c b/lib/poll-loop.c index 5e3618b..ef5ee44 100644 --- a/lib/poll-loop.c +++ b/lib/poll-loop.c @@ -30,18 +30,29 @@ #include "socket-util.h" #include "timeval.h" #include "vlog.h" +#include "hmap.h" +#include "hash.h" VLOG_DEFINE_THIS_MODULE(poll_loop); COVERAGE_DEFINE(poll_fd_wait); COVERAGE_DEFINE(poll_zero_timeout); +struct poll_node { + struct hmap_node hmap_node; + struct pollfd poll_fd; /* Events to pass to time_poll() */ +#ifdef WIN32 + HANDLE wevent; /* events for waitformultipleobjects */ +#else + uint32_t wevent; +#endif + const char *where; /* where each pollfd was created */ +}; + struct poll_loop { /* All active poll waiters. */ - struct pollfd *pollfds; /* Events to pass to poll(). */ - const char **where; /* Where each pollfd was created. */ + struct hmap poll_nodes; size_t n_waiters; /* Number of elems in 'where' and 'pollfds'. */ - size_t allocated_waiters; /* Allocated elems in 'where' and 'pollfds'. */ /* Time at which to wake up the next call to poll_block(), LLONG_MIN to * wake up immediately, or LLONG_MAX to wait forever. */ @@ -51,6 +62,20 @@ struct poll_loop { static struct poll_loop *poll_loop(void); +/* Look up the node with same fd and wevent */ +static struct poll_node * +poll_fd_node_find(struct poll_loop *loop, int fd, uint32_t wevent) +{ + struct poll_node *node; + + HMAP_FOR_EACH_WITH_HASH(node, hmap_node, hash_2words(fd, wevent), &loop->poll_nodes) { + if (node->poll_fd.fd == fd && node->wevent == wevent) { + return node; + } + } + return NULL; +} + /* Registers 'fd' as waiting for the specified 'events' (which should be POLLIN * or POLLOUT or POLLIN | POLLOUT). The following call to poll_block() will * wake up when 'fd' becomes ready for one or more of the requested events. @@ -63,23 +88,38 @@ static struct poll_loop *poll_loop(void); * automatically provide the caller's source file and line number for * 'where'.) */ void -poll_fd_wait_at(int fd, short int events, const char *where) +poll_fd_wait_at(int fd, uint32_t wevent, short int events, const char *where) { struct poll_loop *loop = poll_loop(); + struct poll_node *node; COVERAGE_INC(poll_fd_wait); - if (loop->n_waiters >= loop->allocated_waiters) { - loop->where = x2nrealloc(loop->where, &loop->allocated_waiters, - sizeof *loop->where); - loop->pollfds = xrealloc(loop->pollfds, - (loop->allocated_waiters - * sizeof *loop->pollfds)); + +#ifdef WIN32 + /* null event cannot be polled */ + if (wevent == NULL) { + VLOG_ERR("No event to wait fd %d\n", fd); + return; } +#endif - loop->where[loop->n_waiters] = where; - loop->pollfds[loop->n_waiters].fd = fd; - loop->pollfds[loop->n_waiters].events = events; - loop->n_waiters++; + /* check for duplicate. If found, "or" the event */ + node = poll_fd_node_find(loop, fd, wevent); + if (node) { + node->poll_fd.events |= events; + } else { + node = xzalloc(sizeof *node); + if (node == NULL) { + return; + } + node->where = where; + node->poll_fd.fd = fd; + node->wevent = wevent; + node->poll_fd.events = events; + hmap_insert(&loop->poll_nodes, &node->hmap_node, + hash_2words(fd, wevent)); + loop->n_waiters++; + } } /* Causes the following call to poll_block() to block for no more than 'msec' @@ -215,8 +255,16 @@ void poll_block(void) { struct poll_loop *loop = poll_loop(); + struct poll_node *node, *next; + struct pollfd *pollfds; +#ifdef WIN32 + HANDLE *wevents; +#else + uint32_t *wevents; +#endif int elapsed; int retval; + int i = 0; /* Register fatal signal events before actually doing any real work for * poll_block. */ @@ -227,7 +275,25 @@ poll_block(void) } timewarp_wait(); - retval = time_poll(loop->pollfds, loop->n_waiters, + pollfds = xzalloc(loop->n_waiters * sizeof *pollfds); + if (pollfds == NULL) { + return; + } + + wevents = xzalloc(loop->n_waiters * sizeof *wevents); + if (wevents == NULL) { + free(pollfds); + return; + } + + /* populate with all the fds and events */ + HMAP_FOR_EACH(node, hmap_node, &loop->poll_nodes) { + memcpy(&pollfds[i], &node->poll_fd, sizeof node->poll_fd); + wevents[i] = node->wevent; + i++; + } + + retval = time_poll(pollfds, loop->n_waiters, wevents, loop->timeout_when, &elapsed); if (retval < 0) { static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5); @@ -237,16 +303,23 @@ poll_block(void) } else if (get_cpu_usage() > 50 || VLOG_IS_DBG_ENABLED()) { size_t i; - for (i = 0; i < loop->n_waiters; i++) { - if (loop->pollfds[i].revents) { - log_wakeup(loop->where[i], &loop->pollfds[i], 0); + HMAP_FOR_EACH(node, hmap_node, &loop->poll_nodes) { + if (node->poll_fd.revents) { + log_wakeup(node->where, &node->poll_fd, 0); } } } - loop->timeout_when = LLONG_MAX; + HMAP_FOR_EACH_SAFE(node, next, hmap_node, &loop->poll_nodes) { + hmap_remove(&loop->poll_nodes, &node->hmap_node); + free(node); + } + + loop->timeout_when = LLONG_MIN; loop->timeout_where = NULL; loop->n_waiters = 0; + free(pollfds); + free(wevents); /* Handle any pending signals before doing anything else. */ fatal_signal_run(); @@ -258,9 +331,14 @@ static void free_poll_loop(void *loop_) { struct poll_loop *loop = loop_; + struct poll_node *node, *next; - free(loop->pollfds); - free(loop->where); + HMAP_FOR_EACH_SAFE(node, next, hmap_node, &loop->poll_nodes) { + hmap_remove(&loop->poll_nodes, &node->hmap_node); + free(node); + } + + hmap_destroy(&loop->poll_nodes); free(loop); } @@ -279,6 +357,7 @@ poll_loop(void) loop = pthread_getspecific(key); if (!loop) { loop = xzalloc(sizeof *loop); + hmap_init(&loop->poll_nodes); xpthread_setspecific(key, loop); } return loop; diff --git a/lib/poll-loop.h b/lib/poll-loop.h index 0397853..f5c5665 100644 --- a/lib/poll-loop.h +++ b/lib/poll-loop.h @@ -50,8 +50,8 @@ extern "C" { * caller to supply a location explicitly, which is useful if the caller's own * caller would be more useful in log output. See timer_wait_at() for an * example. */ -void poll_fd_wait_at(int fd, short int events, const char *where); -#define poll_fd_wait(fd, events) poll_fd_wait_at(fd, events, SOURCE_LOCATOR) +void poll_fd_wait_at(int fd, uint32_t wevent, short int events, const char *where); +#define poll_fd_wait(fd, wevent, events) poll_fd_wait_at(fd, wevent, events, SOURCE_LOCATOR) void poll_timer_wait_at(long long int msec, const char *where); #define poll_timer_wait(msec) poll_timer_wait_at(msec, SOURCE_LOCATOR) diff --git a/lib/process.c b/lib/process.c index 5dd34b3..e0a2166 100644 --- a/lib/process.c +++ b/lib/process.c @@ -333,7 +333,7 @@ process_wait(struct process *p) if (p->exited) { poll_immediate_wake(); } else { - poll_fd_wait(fds[0], POLLIN); + poll_fd_wait(fds[0], 0, POLLIN); } } diff --git a/lib/signals.c b/lib/signals.c index 27da5d6..cc56375 100644 --- a/lib/signals.c +++ b/lib/signals.c @@ -88,7 +88,7 @@ signal_poll(struct signal *s) void signal_wait(struct signal *s) { - poll_fd_wait(s->fds[0], POLLIN); + poll_fd_wait(s->fds[0], 0, POLLIN); } static void diff --git a/lib/socket-util.c b/lib/socket-util.c index bb48ade..6c5d0ba 100644 --- a/lib/socket-util.c +++ b/lib/socket-util.c @@ -1354,7 +1354,7 @@ send_iovec_and_fds_fully_block(int sock, if (error != EAGAIN) { return error; } - poll_fd_wait(sock, POLLOUT); + poll_fd_wait(sock, 0, POLLOUT); poll_block(); } } diff --git a/lib/stream-fd.c b/lib/stream-fd.c index 1171f32..3520201 100644 --- a/lib/stream-fd.c +++ b/lib/stream-fd.c @@ -117,11 +117,11 @@ fd_wait(struct stream *stream, enum stream_wait_type wait) switch (wait) { case STREAM_CONNECT: case STREAM_SEND: - poll_fd_wait(s->fd, POLLOUT); + poll_fd_wait(s->fd, 0, POLLOUT); break; case STREAM_RECV: - poll_fd_wait(s->fd, POLLIN); + poll_fd_wait(s->fd, 0, POLLIN); break; default: @@ -235,7 +235,7 @@ static void pfd_wait(struct pstream *pstream) { struct fd_pstream *ps = fd_pstream_cast(pstream); - poll_fd_wait(ps->fd, POLLIN); + poll_fd_wait(ps->fd, 0, POLLIN); } static int diff --git a/lib/timeval.c b/lib/timeval.c index 2ce45fc..445232e 100644 --- a/lib/timeval.c +++ b/lib/timeval.c @@ -232,12 +232,12 @@ time_alarm(unsigned int secs) * * Stores the number of milliseconds elapsed during poll in '*elapsed'. */ int -time_poll(struct pollfd *pollfds, int n_pollfds, long long int timeout_when, - int *elapsed) +time_poll(struct pollfd *pollfds, int n_pollfds, void *handles, + long long int timeout_when, int *elapsed) { long long int *last_wakeup = last_wakeup_get(); long long int start; - int retval; + int retval = 0; time_init(); coverage_clear(); @@ -261,10 +261,19 @@ time_poll(struct pollfd *pollfds, int n_pollfds, long long int timeout_when, time_left = timeout_when - now; } +#ifndef WIN32 retval = poll(pollfds, n_pollfds, time_left); if (retval < 0) { retval = -errno; } +#else + if (n_pollfds != 0) { + retval = WaitForMultipleObjects(n_pollfds, handles, FALSE, time_left); + } + if (retval < 0) { + retval = -WSAGetLastError(); + } +#endif if (deadline <= time_msec()) { fatal_signal_handler(SIGALRM); diff --git a/lib/timeval.h b/lib/timeval.h index 1bbfd5c..3c700c5 100644 --- a/lib/timeval.h +++ b/lib/timeval.h @@ -52,8 +52,8 @@ long long int time_wall_msec(void); void time_timespec(struct timespec *); void time_wall_timespec(struct timespec *); void time_alarm(unsigned int secs); -int time_poll(struct pollfd *, int n_pollfds, long long int timeout_when, - int *elapsed); +int time_poll(struct pollfd *, int n_pollfds, void *handles, + long long int timeout_when, int *elapsed); long long int timespec_to_msec(const struct timespec *); long long int timeval_to_msec(const struct timeval *); diff --git a/tests/test-netflow.c b/tests/test-netflow.c index b6c3109..38486ac 100644 --- a/tests/test-netflow.c +++ b/tests/test-netflow.c @@ -228,7 +228,7 @@ main(int argc, char *argv[]) break; } - poll_fd_wait(sock, POLLIN); + poll_fd_wait(sock, 0, POLLIN); unixctl_server_wait(server); poll_block(); } diff --git a/tests/test-sflow.c b/tests/test-sflow.c index cba01b9..634ca4a 100644 --- a/tests/test-sflow.c +++ b/tests/test-sflow.c @@ -544,7 +544,7 @@ main(int argc, char *argv[]) break; } - poll_fd_wait(sock, POLLIN); + poll_fd_wait(sock, 0, POLLIN); unixctl_server_wait(server); poll_block(); } -- 1.7.9.5 _______________________________________________ dev mailing list [email protected] http://openvswitch.org/mailman/listinfo/dev
