1/* Copyright Joyent, Inc. and other Node contributors. All rights reserved.
2 *
3 * Permission is hereby granted, free of charge, to any person obtaining a copy
4 * of this software and associated documentation files (the "Software"), to
5 * deal in the Software without restriction, including without limitation the
6 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
7 * sell copies of the Software, and to permit persons to whom the Software is
8 * furnished to do so, subject to the following conditions:
9 *
10 * The above copyright notice and this permission notice shall be included in
11 * all copies or substantial portions of the Software.
12 *
13 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
18 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
19 * IN THE SOFTWARE.
20 */
21
22#include <errno.h>
23
24#ifdef _WIN32
25# include <fcntl.h>
26#else
27# include <sys/socket.h>
28# include <unistd.h>
29#endif
30
31#include "uv.h"
32#include "task.h"
33
34#ifdef __linux__
35# include <sys/epoll.h>
36#endif
37
38#ifdef UV_HAVE_KQUEUE
39# include <sys/types.h>
40# include <sys/event.h>
41# include <sys/time.h>
42#endif
43
44
45#define NUM_CLIENTS 5
46#define TRANSFER_BYTES (1 << 16)
47
48#undef MIN
49#define MIN(a, b) (((a) < (b)) ? (a) : (b));
50
51
52typedef enum {
53 UNIDIRECTIONAL,
54 DUPLEX
55} test_mode_t;
56
57typedef struct connection_context_s {
58 uv_poll_t poll_handle;
59 uv_timer_t timer_handle;
60 uv_os_sock_t sock;
61 size_t read, sent;
62 int is_server_connection;
63 int open_handles;
64 int got_fin, sent_fin, got_disconnect;
65 unsigned int events, delayed_events;
66} connection_context_t;
67
68typedef struct server_context_s {
69 uv_poll_t poll_handle;
70 uv_os_sock_t sock;
71 int connections;
72} server_context_t;
73
74
75static void delay_timer_cb(uv_timer_t* timer);
76
77
78static test_mode_t test_mode = DUPLEX;
79
80static int closed_connections = 0;
81
82static int valid_writable_wakeups = 0;
83static int spurious_writable_wakeups = 0;
84
85#if !defined(__sun) && !defined(_AIX) && !defined(__MVS__)
86static int disconnects = 0;
87#endif /* !__sun && !_AIX && !__MVS__ */
88
89static int got_eagain(void) {
90#ifdef _WIN32
91 return WSAGetLastError() == WSAEWOULDBLOCK;
92#else
93 return errno == EAGAIN
94 || errno == EINPROGRESS
95#ifdef EWOULDBLOCK
96 || errno == EWOULDBLOCK;
97#endif
98 ;
99#endif
100}
101
102
103static uv_os_sock_t create_bound_socket (struct sockaddr_in bind_addr) {
104 uv_os_sock_t sock;
105 int r;
106
107 sock = socket(AF_INET, SOCK_STREAM, IPPROTO_IP);
108#ifdef _WIN32
109 ASSERT(sock != INVALID_SOCKET);
110#else
111 ASSERT(sock >= 0);
112#endif
113
114#ifndef _WIN32
115 {
116 /* Allow reuse of the port. */
117 int yes = 1;
118 r = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof yes);
119 ASSERT(r == 0);
120 }
121#endif
122
123 r = bind(sock, (const struct sockaddr*) &bind_addr, sizeof bind_addr);
124 ASSERT(r == 0);
125
126 return sock;
127}
128
129
130static void close_socket(uv_os_sock_t sock) {
131 int r;
132#ifdef _WIN32
133 r = closesocket(sock);
134#else
135 r = close(sock);
136#endif
137 /* On FreeBSD close() can fail with ECONNRESET if the socket was shutdown by
138 * the peer before all pending data was delivered.
139 */
140 ASSERT(r == 0 || errno == ECONNRESET);
141}
142
143
144static connection_context_t* create_connection_context(
145 uv_os_sock_t sock, int is_server_connection) {
146 int r;
147 connection_context_t* context;
148
149 context = (connection_context_t*) malloc(sizeof *context);
150 ASSERT(context != NULL);
151
152 context->sock = sock;
153 context->is_server_connection = is_server_connection;
154 context->read = 0;
155 context->sent = 0;
156 context->open_handles = 0;
157 context->events = 0;
158 context->delayed_events = 0;
159 context->got_fin = 0;
160 context->sent_fin = 0;
161 context->got_disconnect = 0;
162
163 r = uv_poll_init_socket(uv_default_loop(), &context->poll_handle, sock);
164 context->open_handles++;
165 context->poll_handle.data = context;
166 ASSERT(r == 0);
167
168 r = uv_timer_init(uv_default_loop(), &context->timer_handle);
169 context->open_handles++;
170 context->timer_handle.data = context;
171 ASSERT(r == 0);
172
173 return context;
174}
175
176
177static void connection_close_cb(uv_handle_t* handle) {
178 connection_context_t* context = (connection_context_t*) handle->data;
179
180 if (--context->open_handles == 0) {
181 if (test_mode == DUPLEX || context->is_server_connection) {
182 ASSERT(context->read == TRANSFER_BYTES);
183 } else {
184 ASSERT(context->read == 0);
185 }
186
187 if (test_mode == DUPLEX || !context->is_server_connection) {
188 ASSERT(context->sent == TRANSFER_BYTES);
189 } else {
190 ASSERT(context->sent == 0);
191 }
192
193 closed_connections++;
194
195 free(context);
196 }
197}
198
199
200static void destroy_connection_context(connection_context_t* context) {
201 uv_close((uv_handle_t*) &context->poll_handle, connection_close_cb);
202 uv_close((uv_handle_t*) &context->timer_handle, connection_close_cb);
203}
204
205
206static void connection_poll_cb(uv_poll_t* handle, int status, int events) {
207 connection_context_t* context = (connection_context_t*) handle->data;
208 unsigned int new_events;
209 int r;
210
211 ASSERT(status == 0);
212 ASSERT(events & context->events);
213 ASSERT(!(events & ~context->events));
214
215 new_events = context->events;
216
217 if (events & UV_READABLE) {
218 int action = rand() % 7;
219
220 switch (action) {
221 case 0:
222 case 1: {
223 /* Read a couple of bytes. */
224 static char buffer[74];
225 r = recv(context->sock, buffer, sizeof buffer, 0);
226 ASSERT(r >= 0);
227
228 if (r > 0) {
229 context->read += r;
230 } else {
231 /* Got FIN. */
232 context->got_fin = 1;
233 new_events &= ~UV_READABLE;
234 }
235
236 break;
237 }
238
239 case 2:
240 case 3: {
241 /* Read until EAGAIN. */
242 static char buffer[931];
243 r = recv(context->sock, buffer, sizeof buffer, 0);
244 ASSERT(r >= 0);
245
246 while (r > 0) {
247 context->read += r;
248 r = recv(context->sock, buffer, sizeof buffer, 0);
249 }
250
251 if (r == 0) {
252 /* Got FIN. */
253 context->got_fin = 1;
254 new_events &= ~UV_READABLE;
255 } else {
256 ASSERT(got_eagain());
257 }
258
259 break;
260 }
261
262 case 4:
263 /* Ignore. */
264 break;
265
266 case 5:
267 /* Stop reading for a while. Restart in timer callback. */
268 new_events &= ~UV_READABLE;
269 if (!uv_is_active((uv_handle_t*) &context->timer_handle)) {
270 context->delayed_events = UV_READABLE;
271 uv_timer_start(&context->timer_handle, delay_timer_cb, 10, 0);
272 } else {
273 context->delayed_events |= UV_READABLE;
274 }
275 break;
276
277 case 6:
278 /* Fudge with the event mask. */
279 uv_poll_start(&context->poll_handle, UV_WRITABLE, connection_poll_cb);
280 uv_poll_start(&context->poll_handle, UV_READABLE, connection_poll_cb);
281 context->events = UV_READABLE;
282 break;
283
284 default:
285 ASSERT(0);
286 }
287 }
288
289 if (events & UV_WRITABLE) {
290 if (context->sent < TRANSFER_BYTES &&
291 !(test_mode == UNIDIRECTIONAL && context->is_server_connection)) {
292 /* We have to send more bytes. */
293 int action = rand() % 7;
294
295 switch (action) {
296 case 0:
297 case 1: {
298 /* Send a couple of bytes. */
299 static char buffer[103];
300
301 int send_bytes = MIN(TRANSFER_BYTES - context->sent, sizeof buffer);
302 ASSERT(send_bytes > 0);
303
304 r = send(context->sock, buffer, send_bytes, 0);
305
306 if (r < 0) {
307 ASSERT(got_eagain());
308 spurious_writable_wakeups++;
309 break;
310 }
311
312 ASSERT(r > 0);
313 context->sent += r;
314 valid_writable_wakeups++;
315 break;
316 }
317
318 case 2:
319 case 3: {
320 /* Send until EAGAIN. */
321 static char buffer[1234];
322
323 int send_bytes = MIN(TRANSFER_BYTES - context->sent, sizeof buffer);
324 ASSERT(send_bytes > 0);
325
326 r = send(context->sock, buffer, send_bytes, 0);
327
328 if (r < 0) {
329 ASSERT(got_eagain());
330 spurious_writable_wakeups++;
331 break;
332 }
333
334 ASSERT(r > 0);
335 valid_writable_wakeups++;
336 context->sent += r;
337
338 while (context->sent < TRANSFER_BYTES) {
339 send_bytes = MIN(TRANSFER_BYTES - context->sent, sizeof buffer);
340 ASSERT(send_bytes > 0);
341
342 r = send(context->sock, buffer, send_bytes, 0);
343
344 if (r <= 0) break;
345 context->sent += r;
346 }
347 ASSERT(r > 0 || got_eagain());
348 break;
349 }
350
351 case 4:
352 /* Ignore. */
353 break;
354
355 case 5:
356 /* Stop sending for a while. Restart in timer callback. */
357 new_events &= ~UV_WRITABLE;
358 if (!uv_is_active((uv_handle_t*) &context->timer_handle)) {
359 context->delayed_events = UV_WRITABLE;
360 uv_timer_start(&context->timer_handle, delay_timer_cb, 100, 0);
361 } else {
362 context->delayed_events |= UV_WRITABLE;
363 }
364 break;
365
366 case 6:
367 /* Fudge with the event mask. */
368 uv_poll_start(&context->poll_handle,
369 UV_READABLE,
370 connection_poll_cb);
371 uv_poll_start(&context->poll_handle,
372 UV_WRITABLE,
373 connection_poll_cb);
374 context->events = UV_WRITABLE;
375 break;
376
377 default:
378 ASSERT(0);
379 }
380
381 } else {
382 /* Nothing more to write. Send FIN. */
383 int r;
384#ifdef _WIN32
385 r = shutdown(context->sock, SD_SEND);
386#else
387 r = shutdown(context->sock, SHUT_WR);
388#endif
389 ASSERT(r == 0);
390 context->sent_fin = 1;
391 new_events &= ~UV_WRITABLE;
392 }
393 }
394#if !defined(__sun) && !defined(_AIX) && !defined(__MVS__)
395 if (events & UV_DISCONNECT) {
396 context->got_disconnect = 1;
397 ++disconnects;
398 new_events &= ~UV_DISCONNECT;
399 }
400
401 if (context->got_fin && context->sent_fin && context->got_disconnect) {
402#else /* __sun && _AIX && __MVS__ */
403 if (context->got_fin && context->sent_fin) {
404#endif /* !__sun && !_AIX && !__MVS__ */
405 /* Sent and received FIN. Close and destroy context. */
406 close_socket(context->sock);
407 destroy_connection_context(context);
408 context->events = 0;
409
410 } else if (new_events != context->events) {
411 /* Poll mask changed. Call uv_poll_start again. */
412 context->events = new_events;
413 uv_poll_start(handle, new_events, connection_poll_cb);
414 }
415
416 /* Assert that uv_is_active works correctly for poll handles. */
417 if (context->events != 0) {
418 ASSERT(1 == uv_is_active((uv_handle_t*) handle));
419 } else {
420 ASSERT(0 == uv_is_active((uv_handle_t*) handle));
421 }
422}
423
424
425static void delay_timer_cb(uv_timer_t* timer) {
426 connection_context_t* context = (connection_context_t*) timer->data;
427 int r;
428
429 /* Timer should auto stop. */
430 ASSERT(0 == uv_is_active((uv_handle_t*) timer));
431
432 /* Add the requested events to the poll mask. */
433 ASSERT(context->delayed_events != 0);
434 context->events |= context->delayed_events;
435 context->delayed_events = 0;
436
437 r = uv_poll_start(&context->poll_handle,
438 context->events,
439 connection_poll_cb);
440 ASSERT(r == 0);
441}
442
443
444static server_context_t* create_server_context(
445 uv_os_sock_t sock) {
446 int r;
447 server_context_t* context;
448
449 context = (server_context_t*) malloc(sizeof *context);
450 ASSERT(context != NULL);
451
452 context->sock = sock;
453 context->connections = 0;
454
455 r = uv_poll_init_socket(uv_default_loop(), &context->poll_handle, sock);
456 context->poll_handle.data = context;
457 ASSERT(r == 0);
458
459 return context;
460}
461
462
463static void server_close_cb(uv_handle_t* handle) {
464 server_context_t* context = (server_context_t*) handle->data;
465 free(context);
466}
467
468
469static void destroy_server_context(server_context_t* context) {
470 uv_close((uv_handle_t*) &context->poll_handle, server_close_cb);
471}
472
473
474static void server_poll_cb(uv_poll_t* handle, int status, int events) {
475 server_context_t* server_context = (server_context_t*)
476 handle->data;
477 connection_context_t* connection_context;
478 struct sockaddr_in addr;
479 socklen_t addr_len;
480 uv_os_sock_t sock;
481 int r;
482
483 addr_len = sizeof addr;
484 sock = accept(server_context->sock, (struct sockaddr*) &addr, &addr_len);
485#ifdef _WIN32
486 ASSERT(sock != INVALID_SOCKET);
487#else
488 ASSERT(sock >= 0);
489#endif
490
491 connection_context = create_connection_context(sock, 1);
492 connection_context->events = UV_READABLE | UV_WRITABLE | UV_DISCONNECT;
493 r = uv_poll_start(&connection_context->poll_handle,
494 UV_READABLE | UV_WRITABLE | UV_DISCONNECT,
495 connection_poll_cb);
496 ASSERT(r == 0);
497
498 if (++server_context->connections == NUM_CLIENTS) {
499 close_socket(server_context->sock);
500 destroy_server_context(server_context);
501 }
502}
503
504
505static void start_server(void) {
506 server_context_t* context;
507 struct sockaddr_in addr;
508 uv_os_sock_t sock;
509 int r;
510
511 ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
512 sock = create_bound_socket(addr);
513 context = create_server_context(sock);
514
515 r = listen(sock, 100);
516 ASSERT(r == 0);
517
518 r = uv_poll_start(&context->poll_handle, UV_READABLE, server_poll_cb);
519 ASSERT(r == 0);
520}
521
522
523static void start_client(void) {
524 uv_os_sock_t sock;
525 connection_context_t* context;
526 struct sockaddr_in server_addr;
527 struct sockaddr_in addr;
528 int r;
529
530 ASSERT(0 == uv_ip4_addr("127.0.0.1", TEST_PORT, &server_addr));
531 ASSERT(0 == uv_ip4_addr("0.0.0.0", 0, &addr));
532
533 sock = create_bound_socket(addr);
534 context = create_connection_context(sock, 0);
535
536 context->events = UV_READABLE | UV_WRITABLE | UV_DISCONNECT;
537 r = uv_poll_start(&context->poll_handle,
538 UV_READABLE | UV_WRITABLE | UV_DISCONNECT,
539 connection_poll_cb);
540 ASSERT(r == 0);
541
542 r = connect(sock, (struct sockaddr*) &server_addr, sizeof server_addr);
543 ASSERT(r == 0 || got_eagain());
544}
545
546
547static void start_poll_test(void) {
548 int i, r;
549
550#ifdef _WIN32
551 {
552 struct WSAData wsa_data;
553 int r = WSAStartup(MAKEWORD(2, 2), &wsa_data);
554 ASSERT(r == 0);
555 }
556#endif
557
558 start_server();
559
560 for (i = 0; i < NUM_CLIENTS; i++)
561 start_client();
562
563 r = uv_run(uv_default_loop(), UV_RUN_DEFAULT);
564 ASSERT(r == 0);
565
566 /* Assert that at most five percent of the writable wakeups was spurious. */
567 ASSERT(spurious_writable_wakeups == 0 ||
568 (valid_writable_wakeups + spurious_writable_wakeups) /
569 spurious_writable_wakeups > 20);
570
571 ASSERT(closed_connections == NUM_CLIENTS * 2);
572#if !defined(__sun) && !defined(_AIX) && !defined(__MVS__)
573 ASSERT(disconnects == NUM_CLIENTS * 2);
574#endif
575 MAKE_VALGRIND_HAPPY();
576}
577
578
579TEST_IMPL(poll_duplex) {
580#if defined(NO_SELF_CONNECT)
581 RETURN_SKIP(NO_SELF_CONNECT);
582#endif
583 test_mode = DUPLEX;
584 start_poll_test();
585 return 0;
586}
587
588
589TEST_IMPL(poll_unidirectional) {
590#if defined(NO_SELF_CONNECT)
591 RETURN_SKIP(NO_SELF_CONNECT);
592#endif
593 test_mode = UNIDIRECTIONAL;
594 start_poll_test();
595 return 0;
596}
597
598
599/* Windows won't let you open a directory so we open a file instead.
600 * OS X lets you poll a file so open the $PWD instead. Both fail
601 * on Linux so it doesn't matter which one we pick. Both succeed
602 * on FreeBSD, Solaris and AIX so skip the test on those platforms.
603 */
604TEST_IMPL(poll_bad_fdtype) {
605#if !defined(__DragonFly__) && !defined(__FreeBSD__) && !defined(__sun) && \
606 !defined(_AIX) && !defined(__MVS__) && !defined(__FreeBSD_kernel__) && \
607 !defined(__OpenBSD__) && !defined(__CYGWIN__) && !defined(__MSYS__) && \
608 !defined(__NetBSD__)
609 uv_poll_t poll_handle;
610 int fd;
611
612#if defined(_WIN32)
613 fd = open("test/fixtures/empty_file", O_RDONLY);
614#else
615 fd = open(".", O_RDONLY);
616#endif
617 ASSERT(fd != -1);
618 ASSERT(0 != uv_poll_init(uv_default_loop(), &poll_handle, fd));
619 ASSERT(0 == close(fd));
620#endif
621
622 MAKE_VALGRIND_HAPPY();
623 return 0;
624}
625
626
627#ifdef __linux__
628TEST_IMPL(poll_nested_epoll) {
629 uv_poll_t poll_handle;
630 int fd;
631
632 fd = epoll_create(1);
633 ASSERT(fd != -1);
634
635 ASSERT(0 == uv_poll_init(uv_default_loop(), &poll_handle, fd));
636 ASSERT(0 == uv_poll_start(&poll_handle, UV_READABLE, (uv_poll_cb) abort));
637 ASSERT(0 != uv_run(uv_default_loop(), UV_RUN_NOWAIT));
638
639 uv_close((uv_handle_t*) &poll_handle, NULL);
640 ASSERT(0 == uv_run(uv_default_loop(), UV_RUN_DEFAULT));
641 ASSERT(0 == close(fd));
642
643 MAKE_VALGRIND_HAPPY();
644 return 0;
645}
646#endif /* __linux__ */
647
648
649#ifdef UV_HAVE_KQUEUE
650TEST_IMPL(poll_nested_kqueue) {
651 uv_poll_t poll_handle;
652 int fd;
653
654 fd = kqueue();
655 ASSERT(fd != -1);
656
657 ASSERT(0 == uv_poll_init(uv_default_loop(), &poll_handle, fd));
658 ASSERT(0 == uv_poll_start(&poll_handle, UV_READABLE, (uv_poll_cb) abort));
659 ASSERT(0 != uv_run(uv_default_loop(), UV_RUN_NOWAIT));
660
661 uv_close((uv_handle_t*) &poll_handle, NULL);
662 ASSERT(0 == uv_run(uv_default_loop(), UV_RUN_DEFAULT));
663 ASSERT(0 == close(fd));
664
665 MAKE_VALGRIND_HAPPY();
666 return 0;
667}
668#endif /* UV_HAVE_KQUEUE */
669