2 * Copyright (c) 2008-2009 Robert N. M. Watson
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions
8 * 1. Redistributions of source code must retain the above copyright
9 * notice, this list of conditions and the following disclaimer.
10 * 2. Redistributions in binary form must reproduce the above copyright
11 * notice, this list of conditions and the following disclaimer in the
12 * documentation and/or other materials provided with the distribution.
14 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
15 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
16 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
17 * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
18 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
19 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
20 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
21 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
22 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
23 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
29 #include <sys/types.h>
30 #include <sys/endian.h>
31 #include <sys/event.h>
32 #include <sys/resource.h>
33 #include <sys/sched.h>
34 #include <sys/socket.h>
35 #include <sys/sysctl.h>
39 #include <netinet/in.h>
53 * Server side -- create a pool of processes, each listening on its own TCP
54 * port number for new connections. The first 8 bytes of each connection
55 * will be a network byte order length, then there will be that number of
56 * bytes of data. We use non-blocking sockets with kqueue to to avoid the
57 * overhead of threading or more than one process per processor, which makes
58 * things a bit awkward when dealing with data we care about. As such, we
59 * read into a small character buffer which we then convert to a length once
60 * we have all the data.
62 #define CONNECTION_MAGIC 0x6392af27
64 uint32_t conn_magic; /* Just magic. */
66 struct tcpp_header conn_header; /* Header buffer. */
67 u_int conn_header_len; /* Bytes so far. */
68 u_int64_t conn_data_len; /* How much to sink. */
69 u_int64_t conn_data_received; /* How much so far. */
72 static pid_t *pid_list;
75 static struct connection *
76 tcpp_server_newconn(int listen_fd)
78 struct connection *conn;
82 fd = accept(listen_fd, NULL, NULL);
88 if (fcntl(fd, F_SETFL, O_NONBLOCK) < 0)
91 conn = malloc(sizeof(*conn));
94 bzero(conn, sizeof(*conn));
95 conn->conn_magic = CONNECTION_MAGIC;
99 * Register to read on the socket, and set our conn pointer as the
100 * udata so we can find it quickly in the future.
102 EV_SET(&kev, fd, EVFILT_READ, EV_ADD, 0, 0, conn);
103 if (kevent(kq, &kev, 1, NULL, 0, NULL) < 0)
110 tcpp_server_closeconn(struct connection *conn)
114 * Kqueue cleans up after itself once we close the socket, and since
115 * we are processing only one kevent at a time, we don't need to
116 * worry about watching out for future kevents referring to it.
120 close(conn->conn_fd);
121 bzero(conn, sizeof(*conn));
125 static u_char buffer[256*1024]; /* Buffer in which to sink data. */
127 tcpp_server_handleconn(struct kevent *kev)
129 struct connection *conn;
133 if (conn->conn_magic != CONNECTION_MAGIC)
134 errx(-1, "tcpp_server_handleconn: magic");
136 if (conn->conn_header_len < sizeof(conn->conn_header)) {
137 len = read(conn->conn_fd,
138 ((u_char *)&conn->conn_header) + conn->conn_header_len,
139 sizeof(conn->conn_header) - conn->conn_header_len);
141 warn("tcpp_server_handleconn: header read");
142 tcpp_server_closeconn(conn);
146 warnx("tcpp_server_handleconn: header premature eof");
147 tcpp_server_closeconn(conn);
150 conn->conn_header_len += len;
151 if (conn->conn_header_len == sizeof(conn->conn_header)) {
152 tcpp_header_decode(&conn->conn_header);
153 if (conn->conn_header.th_magic != TCPP_MAGIC) {
154 warnx("tcpp_server_handleconn: bad magic");
155 tcpp_server_closeconn(conn);
161 * Drain up to a buffer from the connection, so that we pay
162 * attention to other connections too.
164 len = read(conn->conn_fd, buffer, sizeof(buffer));
166 warn("tcpp_server_handleconn: data bad read");
167 tcpp_server_closeconn(conn);
170 if (len == 0 && conn->conn_data_received <
171 conn->conn_header.th_len) {
172 warnx("tcpp_server_handleconn: data premature eof");
173 tcpp_server_closeconn(conn);
176 conn->conn_data_received += len;
177 if (conn->conn_data_received > conn->conn_header.th_len) {
178 warnx("tcpp_server_handleconn: too much data");
179 tcpp_server_closeconn(conn);
182 if (conn->conn_data_received == conn->conn_header.th_len) {
186 tcpp_server_closeconn(conn);
193 tcpp_server_worker(int workernum)
195 int i, listen_sock, numevents;
196 struct kevent kev, *kev_array;
198 #if defined(CPU_SETSIZE) && 0
204 if (sysctlbyname(SYSCTLNAME_CPUS, &ncpus, &len, NULL, 0) < 0)
205 err(-1, "sysctlbyname: %s", SYSCTLNAME_CPUS);
206 if (len != sizeof(ncpus))
207 errx(-1, "sysctlbyname: %s: len %jd", SYSCTLNAME_CPUS,
211 CPU_SET(workernum % ncpus, &mask);
212 if (sched_setaffinity(0, CPU_SETSIZE, &mask) < 0)
213 err(-1, "sched_setaffinity");
215 setproctitle("tcpp_server %d", workernum);
217 /* Allow an extra kevent for the listen socket. */
218 kev_bytes = sizeof(*kev_array) * (mflag + 1);
219 kev_array = malloc(kev_bytes);
220 if (kev_array == NULL)
222 bzero(kev_array, kev_bytes);
224 /* XXXRW: Want to set and pin the CPU here. */
227 * Add the worker number to the local port.
229 localipbase.sin_port = htons(rflag + workernum);
231 listen_sock = socket(PF_INET, SOCK_STREAM, 0);
235 if (setsockopt(listen_sock, SOL_SOCKET, SO_NOSIGPIPE, &i, sizeof(i))
237 err(-1, "setsockopt");
239 if (setsockopt(listen_sock, SOL_SOCKET, SO_REUSEPORT, &i, sizeof(i))
241 err(-1, "setsockopt");
242 if (bind(listen_sock, (struct sockaddr *)&localipbase,
243 sizeof(localipbase)) < 0)
245 if (listen(listen_sock, 16384))
247 if (fcntl(listen_sock, F_SETFL, O_NONBLOCK) < 0)
254 EV_SET(&kev, listen_sock, EVFILT_READ, EV_ADD, 0, 0, NULL);
255 if (kevent(kq, &kev, 1, NULL, 0, NULL) < 0)
258 while ((numevents = kevent(kq, NULL, 0, kev_array, mflag + 1, NULL))
260 for (i = 0; i < numevents; i++) {
261 if (kev_array[i].ident == (u_int)listen_sock)
262 (void)tcpp_server_newconn(listen_sock);
264 tcpp_server_handleconn(&kev_array[i]);
267 printf("Worker %d done\n", workernum);
273 long cp_time_last[CPUSTATES], cp_time_now[CPUSTATES], ticks;
278 pid_list = malloc(sizeof(*pid_list) * pflag);
279 if (pid_list == NULL)
280 err(-1, "malloc pid_list");
281 bzero(pid_list, sizeof(*pid_list) * pflag);
286 for (i = 0; i < pflag; i++) {
290 for (i = 0; i < pflag; i++) {
291 if (pid_list[i] != 0)
292 (void)kill(pid_list[i], SIGKILL);
297 tcpp_server_worker(i);
304 size = sizeof(cp_time_last);
305 if (sysctlbyname(SYSCTLNAME_CPTIME, &cp_time_last, &size,
307 err(-1, "sysctlbyname: %s", SYSCTLNAME_CPTIME);
310 size = sizeof(cp_time_last);
311 if (sysctlbyname(SYSCTLNAME_CPTIME, &cp_time_now,
313 err(-1, "sysctlbyname: %s",
316 for (i = 0; i < CPUSTATES; i++) {
317 cp_time_last[i] = cp_time_now[i] -
319 ticks += cp_time_last[i];
321 printf("user%% %lu nice%% %lu sys%% %lu intr%% %lu "
323 (100 * cp_time_last[CP_USER]) / ticks,
324 (100 * cp_time_last[CP_NICE]) / ticks,
325 (100 * cp_time_last[CP_SYS]) / ticks,
326 (100 * cp_time_last[CP_INTR]) / ticks,
327 (100 * cp_time_last[CP_IDLE]) / ticks);
328 bcopy(cp_time_now, cp_time_last, sizeof(cp_time_last));
335 for (i = 0; i < pflag; i++) {
336 if (pid_list[i] != 0) {
337 while (waitpid(pid_list[i], NULL, 0) != pid_list[i]);