Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, Google Inc.
4 : * All rights reserved.
5 : *
6 : * Redistribution and use in source and binary forms, with or without
7 : * modification, are permitted provided that the following conditions are
8 : * met:
9 : *
10 : * * Redistributions of source code must retain the above copyright
11 : * notice, this list of conditions and the following disclaimer.
12 : * * Redistributions in binary form must reproduce the above
13 : * copyright notice, this list of conditions and the following disclaimer
14 : * in the documentation and/or other materials provided with the
15 : * distribution.
16 : * * Neither the name of Google Inc. nor the names of its
17 : * contributors may be used to endorse or promote products derived from
18 : * this software without specific prior written permission.
19 : *
20 : * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 : * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 : * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 : * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 : * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 : * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 : * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 : * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 : * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 : * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 : * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 : *
32 : */
33 :
34 : #include <grpc/support/port_platform.h>
35 :
36 : #ifdef GPR_POSIX_SOCKET
37 :
38 : #include "src/core/iomgr/fd_posix.h"
39 :
40 : #include <assert.h>
41 : #include <sys/socket.h>
42 : #include <unistd.h>
43 :
44 : #include <grpc/support/alloc.h>
45 : #include <grpc/support/log.h>
46 : #include <grpc/support/useful.h>
47 :
48 : #define CLOSURE_NOT_READY ((grpc_closure *)0)
49 : #define CLOSURE_READY ((grpc_closure *)1)
50 :
51 : /* We need to keep a freelist not because of any concerns of malloc performance
52 : * but instead so that implementations with multiple threads in (for example)
53 : * epoll_wait deal with the race between pollset removal and incoming poll
54 : * notifications.
55 : *
56 : * The problem is that the poller ultimately holds a reference to this
57 : * object, so it is very difficult to know when is safe to free it, at least
58 : * without some expensive synchronization.
59 : *
60 : * If we keep the object freelisted, in the worst case losing this race just
61 : * becomes a spurious read notification on a reused fd.
62 : */
63 : /* TODO(klempner): We could use some form of polling generation count to know
64 : * when these are safe to free. */
65 : /* TODO(klempner): Consider disabling freelisting if we don't have multiple
66 : * threads in poll on the same fd */
67 : /* TODO(klempner): Batch these allocations to reduce fragmentation */
68 : static grpc_fd *fd_freelist = NULL;
69 : static gpr_mu fd_freelist_mu;
70 :
71 20536 : static void freelist_fd(grpc_fd *fd) {
72 20536 : gpr_mu_lock(&fd_freelist_mu);
73 20537 : fd->freelist_next = fd_freelist;
74 20537 : fd_freelist = fd;
75 20537 : grpc_iomgr_unregister_object(&fd->iomgr_object);
76 20537 : gpr_mu_unlock(&fd_freelist_mu);
77 20537 : }
78 :
79 20634 : static grpc_fd *alloc_fd(int fd) {
80 20461 : grpc_fd *r = NULL;
81 20634 : gpr_mu_lock(&fd_freelist_mu);
82 20634 : if (fd_freelist != NULL) {
83 12758 : r = fd_freelist;
84 12837 : fd_freelist = fd_freelist->freelist_next;
85 : }
86 20634 : gpr_mu_unlock(&fd_freelist_mu);
87 20634 : if (r == NULL) {
88 7797 : r = gpr_malloc(sizeof(grpc_fd));
89 7797 : gpr_mu_init(&r->mu);
90 : }
91 :
92 20634 : gpr_atm_rel_store(&r->refst, 1);
93 20634 : r->shutdown = 0;
94 20634 : r->read_closure = CLOSURE_NOT_READY;
95 20634 : r->write_closure = CLOSURE_NOT_READY;
96 20634 : r->fd = fd;
97 20634 : r->inactive_watcher_root.next = r->inactive_watcher_root.prev =
98 20634 : &r->inactive_watcher_root;
99 20634 : r->freelist_next = NULL;
100 20634 : r->read_watcher = r->write_watcher = NULL;
101 20634 : r->on_done_closure = NULL;
102 20634 : r->closed = 0;
103 20634 : return r;
104 : }
105 :
106 7689 : static void destroy(grpc_fd *fd) {
107 7689 : gpr_mu_destroy(&fd->mu);
108 7689 : gpr_free(fd);
109 7689 : }
110 :
111 : #ifdef GRPC_FD_REF_COUNT_DEBUG
112 : #define REF_BY(fd, n, reason) ref_by(fd, n, reason, __FILE__, __LINE__)
113 : #define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__)
114 : static void ref_by(grpc_fd *fd, int n, const char *reason, const char *file,
115 : int line) {
116 : gpr_log(GPR_DEBUG, "FD %d %p ref %d %d -> %d [%s; %s:%d]", fd->fd, fd, n,
117 : gpr_atm_no_barrier_load(&fd->refst),
118 : gpr_atm_no_barrier_load(&fd->refst) + n, reason, file, line);
119 : #else
120 : #define REF_BY(fd, n, reason) ref_by(fd, n)
121 : #define UNREF_BY(fd, n, reason) unref_by(fd, n)
122 6146958 : static void ref_by(grpc_fd *fd, int n) {
123 : #endif
124 6146958 : GPR_ASSERT(gpr_atm_no_barrier_fetch_add(&fd->refst, n) > 0);
125 6146958 : }
126 :
127 : #ifdef GRPC_FD_REF_COUNT_DEBUG
128 : static void unref_by(grpc_fd *fd, int n, const char *reason, const char *file,
129 : int line) {
130 : gpr_atm old;
131 : gpr_log(GPR_DEBUG, "FD %d %p unref %d %d -> %d [%s; %s:%d]", fd->fd, fd, n,
132 : gpr_atm_no_barrier_load(&fd->refst),
133 : gpr_atm_no_barrier_load(&fd->refst) - n, reason, file, line);
134 : #else
135 6146851 : static void unref_by(grpc_fd *fd, int n) {
136 : gpr_atm old;
137 : #endif
138 6146851 : old = gpr_atm_full_fetch_add(&fd->refst, -n);
139 6146851 : if (old == n) {
140 20537 : freelist_fd(fd);
141 : } else {
142 6126314 : GPR_ASSERT(old > n);
143 : }
144 6146851 : }
145 :
146 3457 : void grpc_fd_global_init(void) { gpr_mu_init(&fd_freelist_mu); }
147 :
148 3455 : void grpc_fd_global_shutdown(void) {
149 3455 : gpr_mu_lock(&fd_freelist_mu);
150 3455 : gpr_mu_unlock(&fd_freelist_mu);
151 14599 : while (fd_freelist != NULL) {
152 7689 : grpc_fd *fd = fd_freelist;
153 7689 : fd_freelist = fd_freelist->freelist_next;
154 7689 : destroy(fd);
155 : }
156 3455 : gpr_mu_destroy(&fd_freelist_mu);
157 3455 : }
158 :
159 20634 : grpc_fd *grpc_fd_create(int fd, const char *name) {
160 20634 : grpc_fd *r = alloc_fd(fd);
161 20634 : grpc_iomgr_register_object(&r->iomgr_object, name);
162 : #ifdef GRPC_FD_REF_COUNT_DEBUG
163 : gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, r, name);
164 : #endif
165 20634 : return r;
166 : }
167 :
168 6646676 : int grpc_fd_is_orphaned(grpc_fd *fd) {
169 6646676 : return (gpr_atm_acq_load(&fd->refst) & 1) == 0;
170 : }
171 :
172 89045 : static void pollset_kick_locked(grpc_fd_watcher *watcher) {
173 89045 : gpr_mu_lock(GRPC_POLLSET_MU(watcher->pollset));
174 89045 : GPR_ASSERT(watcher->worker);
175 89045 : grpc_pollset_kick_ext(watcher->pollset, watcher->worker,
176 : GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP);
177 89041 : gpr_mu_unlock(GRPC_POLLSET_MU(watcher->pollset));
178 89044 : }
179 :
180 1921606 : static void maybe_wake_one_watcher_locked(grpc_fd *fd) {
181 1921606 : if (fd->inactive_watcher_root.next != &fd->inactive_watcher_root) {
182 86284 : pollset_kick_locked(fd->inactive_watcher_root.next);
183 1835322 : } else if (fd->read_watcher) {
184 1362 : pollset_kick_locked(fd->read_watcher);
185 1833960 : } else if (fd->write_watcher) {
186 0 : pollset_kick_locked(fd->write_watcher);
187 : }
188 1921605 : }
189 :
190 1395 : static void wake_all_watchers_locked(grpc_fd *fd) {
191 : grpc_fd_watcher *watcher;
192 4180 : for (watcher = fd->inactive_watcher_root.next;
193 1390 : watcher != &fd->inactive_watcher_root; watcher = watcher->next) {
194 1390 : pollset_kick_locked(watcher);
195 : }
196 1395 : if (fd->read_watcher) {
197 9 : pollset_kick_locked(fd->read_watcher);
198 : }
199 1395 : if (fd->write_watcher && fd->write_watcher != fd->read_watcher) {
200 0 : pollset_kick_locked(fd->write_watcher);
201 : }
202 1395 : }
203 :
204 21860 : static int has_watchers(grpc_fd *fd) {
205 43964 : return fd->read_watcher != NULL || fd->write_watcher != NULL ||
206 21976 : fd->inactive_watcher_root.next != &fd->inactive_watcher_root;
207 : }
208 :
209 20589 : void grpc_fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure *on_done,
210 : int *release_fd, const char *reason) {
211 20589 : fd->on_done_closure = on_done;
212 20589 : fd->released = release_fd != NULL;
213 20589 : if (!fd->released) {
214 20588 : shutdown(fd->fd, SHUT_RDWR);
215 : } else {
216 1 : *release_fd = fd->fd;
217 : }
218 20589 : gpr_mu_lock(&fd->mu);
219 20589 : REF_BY(fd, 1, reason); /* remove active status, but keep referenced */
220 20589 : if (!has_watchers(fd)) {
221 19194 : fd->closed = 1;
222 19194 : if (!fd->released) {
223 19193 : close(fd->fd);
224 : }
225 19193 : grpc_exec_ctx_enqueue(exec_ctx, fd->on_done_closure, 1);
226 : } else {
227 1395 : wake_all_watchers_locked(fd);
228 : }
229 20588 : gpr_mu_unlock(&fd->mu);
230 20588 : UNREF_BY(fd, 2, reason); /* drop the reference */
231 20589 : }
232 :
233 : /* increment refcount by two to avoid changing the orphan bit */
234 : #ifdef GRPC_FD_REF_COUNT_DEBUG
235 : void grpc_fd_ref(grpc_fd *fd, const char *reason, const char *file, int line) {
236 : ref_by(fd, 2, reason, file, line);
237 : }
238 :
239 : void grpc_fd_unref(grpc_fd *fd, const char *reason, const char *file,
240 : int line) {
241 : unref_by(fd, 2, reason, file, line);
242 : }
243 : #else
244 6124982 : void grpc_fd_ref(grpc_fd *fd) { ref_by(fd, 2); }
245 :
246 6124843 : void grpc_fd_unref(grpc_fd *fd) { unref_by(fd, 2); }
247 : #endif
248 :
249 2378668 : static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
250 : grpc_closure **st, grpc_closure *closure) {
251 2378668 : if (*st == CLOSURE_NOT_READY) {
252 : /* not ready ==> switch to a waiting state by setting the closure */
253 2354780 : *st = closure;
254 23888 : } else if (*st == CLOSURE_READY) {
255 : /* already ready ==> queue the closure to run immediately */
256 23888 : *st = CLOSURE_NOT_READY;
257 23888 : grpc_exec_ctx_enqueue(exec_ctx, closure, !fd->shutdown);
258 23887 : maybe_wake_one_watcher_locked(fd);
259 : } else {
260 : /* upcallptr was set to a different closure. This is an error! */
261 0 : gpr_log(GPR_ERROR,
262 : "User called a notify_on function with a previous callback still "
263 : "pending");
264 0 : abort();
265 : }
266 2378667 : }
267 :
268 : /* returns 1 if state becomes not ready */
269 4956224 : static int set_ready_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
270 : grpc_closure **st) {
271 4956224 : if (*st == CLOSURE_READY) {
272 : /* duplicate ready ==> ignore */
273 2553298 : return 0;
274 2402298 : } else if (*st == CLOSURE_NOT_READY) {
275 : /* not ready, and not waiting ==> flag ready */
276 47589 : *st = CLOSURE_READY;
277 47390 : return 0;
278 : } else {
279 : /* waiting ==> queue closure */
280 2354709 : grpc_exec_ctx_enqueue(exec_ctx, *st, !fd->shutdown);
281 2354718 : *st = CLOSURE_NOT_READY;
282 2354008 : return 1;
283 : }
284 : }
285 :
286 3812028 : static void set_ready(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_closure **st) {
287 : /* only one set_ready can be active at once (but there may be a racing
288 : notify_on) */
289 3812028 : gpr_mu_lock(&fd->mu);
290 3813182 : set_ready_locked(exec_ctx, fd, st);
291 3813179 : gpr_mu_unlock(&fd->mu);
292 3813116 : }
293 :
294 11027 : void grpc_fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
295 11027 : gpr_mu_lock(&fd->mu);
296 11027 : GPR_ASSERT(!fd->shutdown);
297 11027 : fd->shutdown = 1;
298 11027 : set_ready_locked(exec_ctx, fd, &fd->read_closure);
299 11027 : set_ready_locked(exec_ctx, fd, &fd->write_closure);
300 11027 : gpr_mu_unlock(&fd->mu);
301 11027 : }
302 :
303 2365718 : void grpc_fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
304 : grpc_closure *closure) {
305 2365718 : gpr_mu_lock(&fd->mu);
306 2365723 : notify_on_locked(exec_ctx, fd, &fd->read_closure, closure);
307 2365723 : gpr_mu_unlock(&fd->mu);
308 2365721 : }
309 :
310 12945 : void grpc_fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
311 : grpc_closure *closure) {
312 12945 : gpr_mu_lock(&fd->mu);
313 12945 : notify_on_locked(exec_ctx, fd, &fd->write_closure, closure);
314 12945 : gpr_mu_unlock(&fd->mu);
315 12945 : }
316 :
317 4425566 : gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset,
318 : grpc_pollset_worker *worker, gpr_uint32 read_mask,
319 : gpr_uint32 write_mask, grpc_fd_watcher *watcher) {
320 4425052 : gpr_uint32 mask = 0;
321 : grpc_closure *cur;
322 : int requested;
323 : /* keep track of pollers that have requested our events, in case they change
324 : */
325 4425566 : GRPC_FD_REF(fd, "poll");
326 :
327 4425630 : gpr_mu_lock(&fd->mu);
328 :
329 : /* if we are shutdown, then don't add to the watcher set */
330 4425634 : if (fd->shutdown) {
331 1 : watcher->fd = NULL;
332 1 : watcher->pollset = NULL;
333 1 : watcher->worker = NULL;
334 1 : gpr_mu_unlock(&fd->mu);
335 1 : GRPC_FD_UNREF(fd, "poll");
336 1 : return 0;
337 : }
338 :
339 : /* if there is nobody polling for read, but we need to, then start doing so */
340 4425633 : cur = fd->read_closure;
341 4425119 : requested = cur != CLOSURE_READY;
342 4425633 : if (read_mask && fd->read_watcher == NULL && requested) {
343 1904761 : fd->read_watcher = watcher;
344 1904759 : mask |= read_mask;
345 : }
346 : /* if there is nobody polling for write, but we need to, then start doing so
347 : */
348 4425633 : cur = fd->write_closure;
349 4425119 : requested = cur != CLOSURE_READY;
350 4425633 : if (write_mask && fd->write_watcher == NULL && requested) {
351 334289 : fd->write_watcher = watcher;
352 334289 : mask |= write_mask;
353 : }
354 : /* if not polling, remember this watcher in case we need someone to later */
355 4425633 : if (mask == 0 && worker != NULL) {
356 282323 : watcher->next = &fd->inactive_watcher_root;
357 282323 : watcher->prev = watcher->next->prev;
358 282323 : watcher->next->prev = watcher->prev->next = watcher;
359 : }
360 4425633 : watcher->pollset = pollset;
361 4425633 : watcher->worker = worker;
362 4425633 : watcher->fd = fd;
363 4425633 : gpr_mu_unlock(&fd->mu);
364 :
365 4425629 : return mask;
366 : }
367 :
368 4425092 : void grpc_fd_end_poll(grpc_exec_ctx *exec_ctx, grpc_fd_watcher *watcher,
369 : int got_read, int got_write) {
370 4424578 : int was_polling = 0;
371 4424578 : int kick = 0;
372 4425092 : grpc_fd *fd = watcher->fd;
373 :
374 4425092 : if (fd == NULL) {
375 4425632 : return;
376 : }
377 :
378 4425091 : gpr_mu_lock(&fd->mu);
379 :
380 4425503 : if (watcher == fd->read_watcher) {
381 : /* remove read watcher, kick if we still need a read */
382 1904754 : was_polling = 1;
383 1904756 : if (!got_read) {
384 801575 : kick = 1;
385 : }
386 1904756 : fd->read_watcher = NULL;
387 : }
388 4425503 : if (watcher == fd->write_watcher) {
389 : /* remove write watcher, kick if we still need a write */
390 334287 : was_polling = 1;
391 334289 : if (!got_write) {
392 317885 : kick = 1;
393 : }
394 334289 : fd->write_watcher = NULL;
395 : }
396 4425503 : if (!was_polling && watcher->worker != NULL) {
397 : /* remove from inactive list */
398 282300 : watcher->next->prev = watcher->prev;
399 282300 : watcher->prev->next = watcher->next;
400 : }
401 4425503 : if (got_read) {
402 1104564 : if (set_ready_locked(exec_ctx, fd, &fd->read_closure)) {
403 1092881 : kick = 1;
404 : }
405 : }
406 4425503 : if (got_write) {
407 16435 : if (set_ready_locked(exec_ctx, fd, &fd->write_closure)) {
408 6482 : kick = 1;
409 : }
410 : }
411 4425503 : if (kick) {
412 1897715 : maybe_wake_one_watcher_locked(fd);
413 : }
414 4425493 : if (grpc_fd_is_orphaned(fd) && !has_watchers(fd) && !fd->closed) {
415 1395 : fd->closed = 1;
416 1395 : if (!fd->released) {
417 1395 : close(fd->fd);
418 : }
419 1395 : grpc_exec_ctx_enqueue(exec_ctx, fd->on_done_closure, 1);
420 : }
421 4425356 : gpr_mu_unlock(&fd->mu);
422 :
423 4425463 : GRPC_FD_UNREF(fd, "poll");
424 : }
425 :
426 1854398 : void grpc_fd_become_readable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
427 1854398 : set_ready(exec_ctx, fd, &fd->read_closure);
428 1854710 : }
429 :
430 1958496 : void grpc_fd_become_writable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
431 1958496 : set_ready(exec_ctx, fd, &fd->write_closure);
432 1958480 : }
433 :
434 : #endif
|
