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 32 TEST_IMPL(udp_reuseport) { 33 struct sockaddr_in addr1, addr2, addr3; 34 uv_loop_t* loop; 35 uv_udp_t handle1, handle2, handle3; 36 int r; 37 38 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr1)); 39 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT_2, &addr2)); 40 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT_3, &addr3)); 41 42 loop = uv_default_loop(); 43 ASSERT_NOT_NULL(loop); 44 45 r = uv_udp_init(loop, &handle1); 46 ASSERT_OK(r); 47 48 r = uv_udp_bind(&handle1, (const struct sockaddr*) &addr1, UV_UDP_REUSEADDR); 49 ASSERT_OK(r); 50 51 r = uv_udp_init(loop, &handle2); 52 ASSERT_OK(r); 53 54 r = uv_udp_bind(&handle2, (const struct sockaddr*) &addr2, UV_UDP_REUSEPORT); 55 ASSERT_EQ(r, UV_ENOTSUP); 56 57 r = uv_udp_init(loop, &handle3); 58 ASSERT_OK(r); 59 60 /* For platforms where SO_REUSEPORTs don't have the capability of 61 * load balancing, specifying both UV_UDP_REUSEADDR and UV_UDP_REUSEPORT 62 * in flags will fail, returning an UV_ENOTSUP error. */ 63 r = uv_udp_bind(&handle3, (const struct sockaddr*) &addr3, 64 UV_UDP_REUSEADDR | UV_UDP_REUSEPORT); 65 ASSERT_EQ(r, UV_ENOTSUP); 66 67 MAKE_VALGRIND_HAPPY(loop); 68 69 return 0; 70 } 71 72 #else 73 74 #define NUM_RECEIVING_THREADS 2 75 #define MAX_UDP_DATAGRAMS 10 76 77 static uv_udp_t udp_send_handles[MAX_UDP_DATAGRAMS]; 78 static uv_udp_send_t udp_send_requests[MAX_UDP_DATAGRAMS]; 79 80 static uv_sem_t semaphore; 81 82 static uv_mutex_t mutex; 83 static unsigned int received; 84 85 static unsigned int thread_loop1_recv; 86 static unsigned int thread_loop2_recv; 87 static unsigned int sent; 88 89 static uv_loop_t* main_loop; 90 static uv_loop_t thread_loop1; 91 static uv_loop_t thread_loop2; 92 static uv_udp_t thread_handle1; 93 static uv_udp_t thread_handle2; 94 static uv_timer_t thread_timer_handle1; 95 static uv_timer_t thread_timer_handle2; 96 97 static void alloc_cb(uv_handle_t* handle, 98 size_t suggested_size, 99 uv_buf_t* buf) { 100 buf->base = malloc(suggested_size); 101 buf->len = (int) suggested_size; 102 } 103 104 static void ticktack(uv_timer_t* timer) { 105 int done = 0; 106 107 ASSERT(timer == &thread_timer_handle1 || timer == &thread_timer_handle2); 108 109 uv_mutex_lock(&mutex); 110 if (received == MAX_UDP_DATAGRAMS) { 111 done = 1; 112 } 113 uv_mutex_unlock(&mutex); 114 115 if (done) { 116 uv_close((uv_handle_t*) timer, NULL); 117 if (timer->loop == &thread_loop1) 118 uv_close((uv_handle_t*) &thread_handle1, NULL); 119 if (timer->loop == &thread_loop2) 120 uv_close((uv_handle_t*) &thread_handle2, NULL); 121 } 122 } 123 124 static void on_recv(uv_udp_t* handle, 125 ssize_t nr, 126 const uv_buf_t* buf, 127 const struct sockaddr* addr, 128 unsigned flags) { 129 ASSERT_OK(flags); 130 ASSERT(handle == &thread_handle1 || handle == &thread_handle2); 131 132 ASSERT_GE(nr, 0); 133 134 if (nr == 0) { 135 ASSERT_NULL(addr); 136 free(buf->base); 137 return; 138 } 139 140 ASSERT_NOT_NULL(addr); 141 ASSERT_EQ(5, nr); 142 ASSERT(!memcmp("Hello", buf->base, nr)); 143 free(buf->base); 144 145 if (handle->loop == &thread_loop1) 146 thread_loop1_recv++; 147 148 if (handle->loop == &thread_loop2) 149 thread_loop2_recv++; 150 151 uv_mutex_lock(&mutex); 152 received++; 153 uv_mutex_unlock(&mutex); 154 } 155 156 static void on_send(uv_udp_send_t* req, int status) { 157 ASSERT_OK(status); 158 ASSERT_PTR_EQ(req->handle->loop, main_loop); 159 160 if (++sent == MAX_UDP_DATAGRAMS) 161 uv_close((uv_handle_t*) req->handle, NULL); 162 } 163 164 static void bind_socket_and_prepare_recv(uv_loop_t* loop, uv_udp_t* handle) { 165 struct sockaddr_in addr; 166 int r; 167 168 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr)); 169 170 r = uv_udp_init(loop, handle); 171 ASSERT_OK(r); 172 173 /* For platforms where SO_REUSEPORTs have the capability of 174 * load balancing, specifying both UV_UDP_REUSEADDR and 175 * UV_UDP_REUSEPORT in flags is allowed and SO_REUSEPORT will 176 * always override the behavior of SO_REUSEADDR. */ 177 r = uv_udp_bind(handle, (const struct sockaddr*) &addr, 178 UV_UDP_REUSEADDR | UV_UDP_REUSEPORT); 179 ASSERT_OK(r); 180 181 r = uv_udp_recv_start(handle, alloc_cb, on_recv); 182 ASSERT_OK(r); 183 } 184 185 static void run_event_loop(void* arg) { 186 int r; 187 uv_udp_t* handle; 188 uv_timer_t* timer; 189 uv_loop_t* loop = (uv_loop_t*) arg; 190 ASSERT(loop == &thread_loop1 || loop == &thread_loop2); 191 192 if (loop == &thread_loop1) { 193 handle = &thread_handle1; 194 timer = &thread_timer_handle1; 195 } else { 196 handle = &thread_handle2; 197 timer = &thread_timer_handle2; 198 } 199 200 bind_socket_and_prepare_recv(loop, handle); 201 r = uv_timer_init(loop, timer); 202 ASSERT_OK(r); 203 r = uv_timer_start(timer, ticktack, 0, 10); 204 ASSERT_OK(r); 205 206 /* Notify the main thread to start sending data. */ 207 uv_sem_post(&semaphore); 208 r = uv_run(loop, UV_RUN_DEFAULT); 209 ASSERT_OK(r); 210 } 211 212 TEST_IMPL(udp_reuseport) { 213 struct sockaddr_in addr; 214 uv_buf_t buf; 215 int r; 216 int i; 217 218 r = uv_mutex_init(&mutex); 219 ASSERT_OK(r); 220 221 r = uv_sem_init(&semaphore, 0); 222 ASSERT_OK(r); 223 224 main_loop = uv_default_loop(); 225 ASSERT_NOT_NULL(main_loop); 226 227 /* Run event loops of receiving sockets in separate threads. */ 228 uv_loop_init(&thread_loop1); 229 uv_loop_init(&thread_loop2); 230 uv_thread_t thread_loop_id1; 231 uv_thread_t thread_loop_id2; 232 uv_thread_create(&thread_loop_id1, run_event_loop, &thread_loop1); 233 uv_thread_create(&thread_loop_id2, run_event_loop, &thread_loop2); 234 235 /* Wait until all threads to poll for receiving datagrams 236 * before we start to sending. Otherwise the incoming datagrams 237 * might not be distributed across all receiving threads. */ 238 for (i = 0; i < NUM_RECEIVING_THREADS; i++) 239 uv_sem_wait(&semaphore); 240 /* Now we know all threads are up and entering the uv_run(), 241 * but we still sleep a little bit just for dual fail-safe. */ 242 uv_sleep(100); 243 244 /* Start sending datagrams to the peers. */ 245 buf = uv_buf_init("Hello", 5); 246 ASSERT_OK(uv_ip4_addr("127.0.0.1", TEST_PORT, &addr)); 247 for (i = 0; i < MAX_UDP_DATAGRAMS; i++) { 248 r = uv_udp_init(main_loop, &udp_send_handles[i]); 249 ASSERT_OK(r); 250 r = uv_udp_send(&udp_send_requests[i], 251 &udp_send_handles[i], 252 &buf, 253 1, 254 (const struct sockaddr*) &addr, 255 on_send); 256 ASSERT_OK(r); 257 } 258 259 r = uv_run(main_loop, UV_RUN_DEFAULT); 260 ASSERT_OK(r); 261 262 /* Wait for all threads to exit. */ 263 uv_thread_join(&thread_loop_id1); 264 uv_thread_join(&thread_loop_id2); 265 266 /* Verify if each receiving socket per event loop received datagrams 267 * and the amount of received datagrams matches the one of sent datagrams. 268 */ 269 ASSERT_EQ(received, MAX_UDP_DATAGRAMS); 270 ASSERT_EQ(sent, MAX_UDP_DATAGRAMS); 271 ASSERT_GT(thread_loop1_recv, 0); 272 ASSERT_GT(thread_loop2_recv, 0); 273 ASSERT_EQ(thread_loop1_recv + thread_loop2_recv, sent); 274 275 /* Clean up. */ 276 uv_mutex_destroy(&mutex); 277 278 uv_sem_destroy(&semaphore); 279 280 uv_loop_close(&thread_loop1); 281 uv_loop_close(&thread_loop2); 282 MAKE_VALGRIND_HAPPY(main_loop); 283 284 return 0; 285 } 286 287 #endif 288
Indexes created Mon Mar 02 05:31:46 UTC 2026