1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
/* Four ways to get a thread out of a blocking recvfrom, tested rather than
assumed: close the socket, shut it down, give it a receive timeout, or
send it a signal.
This decides how udp workers are stopped. The sockets of a SO_REUSEPORT
group all share one address, so a worker cannot be singled out by sending
it a datagram, and the two mechanisms differ in what they cost while the
tracker is running: a receive timeout wakes every worker on every tick
forever, a signal and a shutdown cost nothing until the one moment they
are used.
cc -O2 -pthread -o /tmp/wake tests/udp_wake_probe.c && /tmp/wake
On Linux 6.1: close does not wake it, the other three do. */
#include <errno.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <string.h>
#include <sys/socket.h>
#include <signal.h>
#include <sys/time.h>
#include <unistd.h>
#define WAKE_SIGNAL SIGUSR1
static _Atomic int returned;
static _Atomic int saved_errno;
static int sock;
static void *reader(void *args) {
char buffer[64];
(void)args;
if (recvfrom(sock, buffer, sizeof(buffer), 0, NULL, NULL) < 0)
atomic_store(&saved_errno, errno);
atomic_store(&returned, 1);
return NULL;
}
/* A handler that does nothing is all it takes: the point is the return from
recvfrom with EINTR, not anything the handler does. SA_RESTART must stay
off or the kernel resumes the call instead of reporting it. */
static void noop(int s) { (void)s; }
static int bind_udp(int timeout_ms) {
struct sockaddr_in address;
int s = socket(AF_INET, SOCK_DGRAM, 0);
memset(&address, 0, sizeof(address));
address.sin_family = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
address.sin_port = 0;
bind(s, (struct sockaddr *)&address, sizeof(address));
if (timeout_ms) {
struct timeval how_long;
how_long.tv_sec = timeout_ms / 1000;
how_long.tv_usec = (timeout_ms % 1000) * 1000;
setsockopt(s, SOL_SOCKET, SO_RCVTIMEO, &how_long, sizeof(how_long));
}
return s;
}
static void try_it(const char *name, int timeout_ms, int do_close, int do_shutdown, int do_signal) {
pthread_t thread;
int waited;
atomic_store(&returned, 0);
atomic_store(&saved_errno, 0);
sock = bind_udp(timeout_ms);
pthread_create(&thread, NULL, reader, NULL);
usleep(300000); /* let it reach recvfrom */
if (do_shutdown)
shutdown(sock, SHUT_RD);
if (do_signal)
pthread_kill(thread, WAKE_SIGNAL);
if (do_close)
close(sock);
for (waited = 0; waited < 2000 && !atomic_load(&returned); ++waited)
usleep(1000);
if (atomic_load(&returned)) {
printf(" %-24s returned after %4d ms, errno %s\n", name, waited,
atomic_load(&saved_errno) ? strerror(atomic_load(&saved_errno)) : "none");
pthread_join(thread, NULL);
} else {
printf(" %-24s STILL BLOCKED after 2000 ms\n", name);
/* unblock it so the probe can finish */
struct sockaddr_in to;
socklen_t length = sizeof(to);
int poke = socket(AF_INET, SOCK_DGRAM, 0);
if (!getsockname(sock, (struct sockaddr *)&to, &length))
sendto(poke, "x", 1, 0, (struct sockaddr *)&to, length);
close(poke);
usleep(200000);
if (atomic_load(&returned))
pthread_join(thread, NULL);
else
pthread_detach(thread);
}
if (!do_close)
close(sock);
}
int main(void) {
struct sigaction sa;
memset(&sa, 0, sizeof(sa));
sa.sa_handler = noop;
sa.sa_flags = 0;
sigaction(WAKE_SIGNAL, &sa, NULL);
try_it("close() the socket", 0, 1, 0, 0);
try_it("shutdown(SHUT_RD)", 0, 0, 1, 0);
try_it("SO_RCVTIMEO 500 ms", 500, 0, 0, 0);
try_it("pthread_kill(SIGUSR1)", 0, 0, 0, 1);
return 0;
}
|
