1 /*
2  * Copyright (c) 2014,2015 DeNA Co., Ltd., Kazuho Oku
3  *
4  * Permission is hereby granted, free of charge, to any person obtaining a copy
5  * of this software and associated documentation files (the "Software"), to
6  * deal in the Software without restriction, including without limitation the
7  * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
8  * sell copies of the Software, and to permit persons to whom the Software is
9  * furnished to do so, subject to the following conditions:
10  *
11  * The above copyright notice and this permission notice shall be included in
12  * all copies or substantial portions of the Software.
13  *
14  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
19  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
20  * IN THE SOFTWARE.
21  */
22 #include <stdio.h>
23 #include <poll.h>
24 
25 #if 0
26 #define DEBUG_LOG(...) h2o_error_printf(__VA_ARGS__)
27 #else
28 #define DEBUG_LOG(...)
29 #endif
30 
31 struct st_h2o_evloop_poll_t {
32     h2o_evloop_t super;
33     H2O_VECTOR(struct st_h2o_evloop_socket_t *) socks;
34 };
35 
update_socks(struct st_h2o_evloop_poll_t * loop)36 static void update_socks(struct st_h2o_evloop_poll_t *loop)
37 {
38     /* update loop->socks */
39     while (loop->super._statechanged.head != NULL) {
40         /* detach the top */
41         struct st_h2o_evloop_socket_t *sock = loop->super._statechanged.head;
42         loop->super._statechanged.head = sock->_next_statechanged;
43         sock->_next_statechanged = sock;
44         /* update the state */
45         if ((sock->_flags & H2O_SOCKET_FLAG_IS_DISPOSED) != 0) {
46             assert(sock->fd == -1);
47             free(sock);
48         } else {
49             assert(sock->fd < loop->socks.size);
50             if (loop->socks.entries[sock->fd] == NULL) {
51                 loop->socks.entries[sock->fd] = sock;
52             } else {
53                 assert(loop->socks.entries[sock->fd] == sock);
54             }
55             if (h2o_socket_is_reading(&sock->super)) {
56                 DEBUG_LOG("setting READ for fd: %d\n", sock->fd);
57                 sock->_flags |= H2O_SOCKET_FLAG_IS_POLLED_FOR_READ;
58             } else {
59                 DEBUG_LOG("clearing READ for fd: %d\n", sock->fd);
60                 sock->_flags &= ~H2O_SOCKET_FLAG_IS_POLLED_FOR_READ;
61             }
62             if (h2o_socket_is_writing(&sock->super)) {
63                 DEBUG_LOG("setting WRITE for fd: %d\n", sock->fd);
64                 sock->_flags |= H2O_SOCKET_FLAG_IS_POLLED_FOR_WRITE;
65             } else {
66                 DEBUG_LOG("clearing WRITE for fd: %d\n", sock->fd);
67                 sock->_flags &= ~H2O_SOCKET_FLAG_IS_POLLED_FOR_WRITE;
68             }
69         }
70     }
71     loop->super._statechanged.tail_ref = &loop->super._statechanged.head;
72 }
73 
evloop_do_proceed(h2o_evloop_t * _loop,int32_t max_wait)74 int evloop_do_proceed(h2o_evloop_t *_loop, int32_t max_wait)
75 {
76     struct st_h2o_evloop_poll_t *loop = (struct st_h2o_evloop_poll_t *)_loop;
77     H2O_VECTOR(struct pollfd) pollfds = {NULL};
78     int fd, ret;
79 
80     /* update status */
81     update_socks(loop);
82 
83     /* build list of fds to be polled */
84     for (fd = 0; fd != loop->socks.size; ++fd) {
85         struct st_h2o_evloop_socket_t *sock = loop->socks.entries[fd];
86         if (sock == NULL)
87             continue;
88         assert(fd == sock->fd);
89         if ((sock->_flags & (H2O_SOCKET_FLAG_IS_POLLED_FOR_READ | H2O_SOCKET_FLAG_IS_POLLED_FOR_WRITE)) != 0) {
90             h2o_vector_reserve(NULL, &pollfds, pollfds.size + 1);
91             struct pollfd *slot = pollfds.entries + pollfds.size++;
92             slot->fd = fd;
93             slot->events = 0;
94             slot->revents = 0;
95             if ((sock->_flags & H2O_SOCKET_FLAG_IS_POLLED_FOR_READ) != 0)
96                 slot->events |= POLLIN;
97             if ((sock->_flags & H2O_SOCKET_FLAG_IS_POLLED_FOR_WRITE) != 0)
98                 slot->events |= POLLOUT;
99         }
100     }
101 
102     /* call */
103     max_wait = adjust_max_wait(&loop->super, max_wait);
104     ret = poll(pollfds.entries, (nfds_t)pollfds.size, max_wait);
105     update_now(&loop->super);
106     if (ret == -1)
107         goto Exit;
108     DEBUG_LOG("poll returned: %d\n", ret);
109 
110     /* update readable flags, perform writes */
111     if (ret > 0) {
112         size_t i;
113         h2o_sliding_counter_start(&loop->super.exec_time_nanosec_counter, loop->super._now_nanosec);
114         for (i = 0; i != pollfds.size; ++i) {
115             /* set read_ready flag before calling the write cb, since app. code invoked by the latter may close the socket, clearing
116              * the former flag */
117             if ((pollfds.entries[i].revents & POLLIN) != 0) {
118                 struct st_h2o_evloop_socket_t *sock = loop->socks.entries[pollfds.entries[i].fd];
119                 assert(sock != NULL);
120                 assert(sock->fd == pollfds.entries[i].fd);
121                 if (sock->_flags != H2O_SOCKET_FLAG_IS_DISPOSED) {
122                     sock->_flags |= H2O_SOCKET_FLAG_IS_READ_READY;
123                     link_to_pending(sock);
124                     DEBUG_LOG("added fd %d as read_ready\n", sock->fd);
125                 }
126             }
127             if ((pollfds.entries[i].revents & POLLOUT) != 0) {
128                 struct st_h2o_evloop_socket_t *sock = loop->socks.entries[pollfds.entries[i].fd];
129                 assert(sock != NULL);
130                 assert(sock->fd == pollfds.entries[i].fd);
131                 if (sock->_flags != H2O_SOCKET_FLAG_IS_DISPOSED) {
132                     DEBUG_LOG("handling pending writes on fd %d\n", fd);
133                     write_pending(sock);
134                 }
135             }
136         }
137         ret = 0;
138     }
139 
140 Exit:
141     free(pollfds.entries);
142     return ret;
143 }
144 
evloop_do_on_socket_create(struct st_h2o_evloop_socket_t * sock)145 static void evloop_do_on_socket_create(struct st_h2o_evloop_socket_t *sock)
146 {
147     struct st_h2o_evloop_poll_t *loop = (struct st_h2o_evloop_poll_t *)sock->loop;
148 
149     if (sock->fd >= loop->socks.size) {
150         h2o_vector_reserve(NULL, &loop->socks, sock->fd + 1);
151         memset(loop->socks.entries + loop->socks.size, 0, (sock->fd + 1 - loop->socks.size) * sizeof(loop->socks.entries[0]));
152         loop->socks.size = sock->fd + 1;
153     }
154 
155     if (loop->socks.entries[sock->fd] != NULL)
156         assert(loop->socks.entries[sock->fd]->_flags == H2O_SOCKET_FLAG_IS_DISPOSED);
157 }
158 
evloop_do_on_socket_close(struct st_h2o_evloop_socket_t * sock)159 static void evloop_do_on_socket_close(struct st_h2o_evloop_socket_t *sock)
160 {
161     struct st_h2o_evloop_poll_t *loop = (struct st_h2o_evloop_poll_t *)sock->loop;
162 
163     if (sock->fd != -1)
164         loop->socks.entries[sock->fd] = NULL;
165 }
166 
evloop_do_on_socket_export(struct st_h2o_evloop_socket_t * sock)167 static void evloop_do_on_socket_export(struct st_h2o_evloop_socket_t *sock)
168 {
169     struct st_h2o_evloop_poll_t *loop = (struct st_h2o_evloop_poll_t *)sock->loop;
170     evloop_do_on_socket_close(sock);
171     loop->socks.entries[sock->fd] = NULL;
172 }
173 
evloop_do_dispose(h2o_evloop_t * loop)174 static void evloop_do_dispose(h2o_evloop_t *loop)
175 {
176 }
177 
h2o_evloop_create(void)178 h2o_evloop_t *h2o_evloop_create(void)
179 {
180     struct st_h2o_evloop_poll_t *loop = (struct st_h2o_evloop_poll_t *)create_evloop(sizeof(*loop));
181     return &loop->super;
182 }
183