1 /* Copyright libuv project 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 "uv.h"
23 #include "task.h"
24
25 #include <stdio.h>
26 #include <stdlib.h>
27 #include <string.h>
28
29 #if !defined(__linux__) && !defined(__FreeBSD__) && \
30 !defined(__DragonFly__) && !defined(__sun) && !defined(_AIX73)
31
TEST_IMPL(tcp_reuseport)32 TEST_IMPL(tcp_reuseport) {
33 struct sockaddr_in addr;
34 uv_loop_t* loop;
35 uv_tcp_t handle;
36 int r;
37
38 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
39
40 loop = uv_default_loop();
41 ASSERT_NOT_NULL(loop);
42
43 r = uv_tcp_init(loop, &handle);
44 ASSERT_OK(r);
45
46 r = uv_tcp_bind(&handle, (const struct sockaddr*) &addr, UV_TCP_REUSEPORT);
47 ASSERT_EQ(r, UV_ENOTSUP);
48
49 MAKE_VALGRIND_HAPPY(loop);
50
51 return 0;
52 }
53
54 #else
55
56 #define NUM_LISTENING_THREADS 2
57 #define MAX_TCP_CLIENTS 10
58
59 static uv_tcp_t tcp_connect_handles[MAX_TCP_CLIENTS];
60 static uv_connect_t tcp_connect_requests[MAX_TCP_CLIENTS];
61
62 static uv_sem_t semaphore;
63
64 static uv_mutex_t mutex;
65 static unsigned int accepted;
66
67 static unsigned int thread_loop1_accepted;
68 static unsigned int thread_loop2_accepted;
69 static unsigned int connected;
70
71 static uv_loop_t* main_loop;
72 static uv_loop_t thread_loop1;
73 static uv_loop_t thread_loop2;
74 static uv_tcp_t thread_handle1;
75 static uv_tcp_t thread_handle2;
76 static uv_timer_t thread_timer_handle1;
77 static uv_timer_t thread_timer_handle2;
78
on_close(uv_handle_t * handle)79 static void on_close(uv_handle_t* handle) {
80 free(handle);
81 }
82
ticktack(uv_timer_t * timer)83 static void ticktack(uv_timer_t* timer) {
84 ASSERT(timer == &thread_timer_handle1 || timer == &thread_timer_handle2);
85
86 int done = 0;
87 uv_mutex_lock(&mutex);
88 if (accepted == MAX_TCP_CLIENTS) {
89 done = 1;
90 }
91 uv_mutex_unlock(&mutex);
92
93 if (done) {
94 uv_close((uv_handle_t*) timer, NULL);
95 if (timer->loop == &thread_loop1)
96 uv_close((uv_handle_t*) &thread_handle1, NULL);
97 if (timer->loop == &thread_loop2)
98 uv_close((uv_handle_t*) &thread_handle2, NULL);
99 }
100 }
101
on_connection(uv_stream_t * server,int status)102 static void on_connection(uv_stream_t* server, int status)
103 {
104 ASSERT_OK(status);
105 ASSERT(server == (uv_stream_t*) &thread_handle1 || \
106 server == (uv_stream_t*) &thread_handle2);
107
108 uv_tcp_t *client = malloc(sizeof(uv_tcp_t));
109 ASSERT_OK(uv_tcp_init(server->loop, client));
110 ASSERT_OK(uv_accept(server, (uv_stream_t*) client));
111 uv_close((uv_handle_t*) client, on_close);
112
113 if (server->loop == &thread_loop1)
114 thread_loop1_accepted++;
115
116 if (server->loop == &thread_loop2)
117 thread_loop2_accepted++;
118
119 uv_mutex_lock(&mutex);
120 accepted++;
121 uv_mutex_unlock(&mutex);
122 }
123
on_connect(uv_connect_t * req,int status)124 static void on_connect(uv_connect_t* req, int status) {
125 ASSERT_OK(status);
126 ASSERT_NOT_NULL(req->handle);
127 ASSERT_PTR_EQ(req->handle->loop, main_loop);
128
129 connected++;
130 uv_close((uv_handle_t*) req->handle, NULL);
131 }
132
create_listener(uv_loop_t * loop,uv_tcp_t * handle)133 static void create_listener(uv_loop_t* loop, uv_tcp_t* handle) {
134 struct sockaddr_in addr;
135 int r;
136
137 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
138
139 r = uv_tcp_init(loop, handle);
140 ASSERT_OK(r);
141
142 r = uv_tcp_bind(handle, (const struct sockaddr*) &addr, UV_TCP_REUSEPORT);
143 ASSERT_OK(r);
144
145 r = uv_listen((uv_stream_t*) handle, 128, on_connection);
146 ASSERT_OK(r);
147 }
148
run_event_loop(void * arg)149 static void run_event_loop(void* arg) {
150 int r;
151 uv_tcp_t* handle;
152 uv_timer_t* timer;
153 uv_loop_t* loop = (uv_loop_t*) arg;
154 ASSERT(loop == &thread_loop1 || loop == &thread_loop2);
155
156 if (loop == &thread_loop1) {
157 handle = &thread_handle1;
158 timer = &thread_timer_handle1;
159 } else {
160 handle = &thread_handle2;
161 timer = &thread_timer_handle2;
162 }
163
164 create_listener(loop, handle);
165 r = uv_timer_init(loop, timer);
166 ASSERT_OK(r);
167 r = uv_timer_start(timer, ticktack, 0, 10);
168 ASSERT_OK(r);
169
170 /* Notify the main thread to start connecting. */
171 uv_sem_post(&semaphore);
172 r = uv_run(loop, UV_RUN_DEFAULT);
173 ASSERT_OK(r);
174 }
175
TEST_IMPL(tcp_reuseport)176 TEST_IMPL(tcp_reuseport) {
177 struct sockaddr_in addr;
178 int r;
179 int i;
180
181 r = uv_mutex_init(&mutex);
182 ASSERT_OK(r);
183
184 r = uv_sem_init(&semaphore, 0);
185 ASSERT_OK(r);
186
187 main_loop = uv_default_loop();
188 ASSERT_NOT_NULL(main_loop);
189
190 /* Run event loops of listeners in separate threads. */
191 uv_loop_init(&thread_loop1);
192 uv_loop_init(&thread_loop2);
193 uv_thread_t thread_loop_id1;
194 uv_thread_t thread_loop_id2;
195 uv_thread_create(&thread_loop_id1, run_event_loop, &thread_loop1);
196 uv_thread_create(&thread_loop_id2, run_event_loop, &thread_loop2);
197
198 /* Wait until all threads to poll for accepting connections
199 * before we start to connect. Otherwise the incoming connections
200 * might not be distributed across all listening threads. */
201 for (i = 0; i < NUM_LISTENING_THREADS; i++)
202 uv_sem_wait(&semaphore);
203 /* Now we know all threads are up and entering the uv_run(),
204 * but we still sleep a little bit just for dual fail-safe. */
205 uv_sleep(100);
206
207 /* Start connecting to the peers. */
208 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr));
209 for (i = 0; i < MAX_TCP_CLIENTS; i++) {
210 r = uv_tcp_init(main_loop, &tcp_connect_handles[i]);
211 ASSERT_OK(r);
212 r = uv_tcp_connect(&tcp_connect_requests[i],
213 &tcp_connect_handles[i],
214 (const struct sockaddr*) &addr,
215 on_connect);
216 ASSERT_OK(r);
217 }
218
219 r = uv_run(main_loop, UV_RUN_DEFAULT);
220 ASSERT_OK(r);
221
222 /* Wait for all threads to exit. */
223 uv_thread_join(&thread_loop_id1);
224 uv_thread_join(&thread_loop_id2);
225
226 /* Verify if each listener per event loop accepted connections
227 * and the amount of accepted connections matches the one of
228 * connected connections.
229 */
230 ASSERT_EQ(accepted, MAX_TCP_CLIENTS);
231 ASSERT_EQ(connected, MAX_TCP_CLIENTS);
232 ASSERT_GT(thread_loop1_accepted, 0);
233 ASSERT_GT(thread_loop2_accepted, 0);
234 ASSERT_EQ(thread_loop1_accepted + thread_loop2_accepted, connected);
235
236 /* Clean up. */
237 uv_mutex_destroy(&mutex);
238
239 uv_sem_destroy(&semaphore);
240
241 uv_loop_close(&thread_loop1);
242 uv_loop_close(&thread_loop2);
243 MAKE_VALGRIND_HAPPY(main_loop);
244
245 return 0;
246 }
247
248 #endif
249
