[efdfdee] | 1 | #include <cstdio> |
---|
| 2 | #include <cstdlib> |
---|
| 3 | #include <cstring> |
---|
| 4 | |
---|
| 5 | #include <iostream> |
---|
| 6 | |
---|
| 7 | #include <signal.h> |
---|
| 8 | #include <unistd.h> |
---|
| 9 | #include <liburing.h> |
---|
| 10 | |
---|
| 11 | typedef enum { |
---|
| 12 | EVENT_END, |
---|
| 13 | EVENT_ACCEPT, |
---|
| 14 | EVENT_REQUEST, |
---|
| 15 | EVENT_ANSWER |
---|
| 16 | } event_t; |
---|
| 17 | |
---|
[3acbf89] | 18 | struct __attribute__((aligned(128))) request_t { |
---|
[efdfdee] | 19 | event_t type; |
---|
| 20 | int fd; |
---|
| 21 | size_t length; |
---|
[3acbf89] | 22 | char * buff; |
---|
| 23 | char data[0]; |
---|
[efdfdee] | 24 | |
---|
| 25 | static struct request_t * create(event_t type, size_t extra) { |
---|
| 26 | auto ret = (struct request_t *)malloc(sizeof(struct request_t) + extra); |
---|
| 27 | ret->type = type; |
---|
| 28 | ret->length = extra; |
---|
[3acbf89] | 29 | ret->buff = ret->data; |
---|
[efdfdee] | 30 | return ret; |
---|
| 31 | } |
---|
| 32 | |
---|
| 33 | static struct request_t * create(event_t type) { |
---|
| 34 | return create(type, 0); |
---|
| 35 | } |
---|
| 36 | }; |
---|
| 37 | |
---|
[3acbf89] | 38 | struct __attribute__((aligned(128))) options_t { |
---|
[efdfdee] | 39 | struct { |
---|
| 40 | int sockfd; |
---|
| 41 | struct sockaddr *addr; |
---|
| 42 | socklen_t *addrlen; |
---|
| 43 | int flags; |
---|
| 44 | } acpt; |
---|
| 45 | |
---|
| 46 | int endfd; |
---|
[f3e87af] | 47 | struct io_uring * ring; |
---|
[3acbf89] | 48 | |
---|
| 49 | struct { |
---|
| 50 | size_t subs = 0; |
---|
| 51 | size_t cnts = 0; |
---|
| 52 | } result; |
---|
[efdfdee] | 53 | }; |
---|
| 54 | |
---|
| 55 | //========================================================= |
---|
[3acbf89] | 56 | static struct io_uring_sqe * get_sqe(struct io_uring * ring) { |
---|
[efdfdee] | 57 | struct io_uring_sqe * sqe = io_uring_get_sqe(ring); |
---|
[3acbf89] | 58 | if(!sqe) { |
---|
| 59 | std::cerr << "Insufficient entries in ring" << std::endl; |
---|
| 60 | exit(EXIT_FAILURE); |
---|
| 61 | } |
---|
| 62 | return sqe; |
---|
| 63 | } |
---|
| 64 | |
---|
| 65 | static void submit(struct io_uring * ) { |
---|
| 66 | // io_uring_submit(ring); |
---|
| 67 | } |
---|
| 68 | |
---|
| 69 | //========================================================= |
---|
| 70 | static void ring_end(struct io_uring * ring, int fd, char * buffer, size_t len) { |
---|
| 71 | struct io_uring_sqe * sqe = get_sqe(ring); |
---|
[efdfdee] | 72 | io_uring_prep_read(sqe, fd, buffer, len, 0); |
---|
| 73 | io_uring_sqe_set_data(sqe, request_t::create(EVENT_END)); |
---|
[3acbf89] | 74 | submit(ring); |
---|
[efdfdee] | 75 | } |
---|
| 76 | |
---|
| 77 | static void ring_accept(struct io_uring * ring, int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) { |
---|
| 78 | auto req = request_t::create(EVENT_ACCEPT); |
---|
[3acbf89] | 79 | struct io_uring_sqe * sqe = get_sqe(ring); |
---|
[efdfdee] | 80 | io_uring_prep_accept(sqe, sockfd, addr, addrlen, flags); |
---|
| 81 | io_uring_sqe_set_data(sqe, req); |
---|
[3acbf89] | 82 | submit(ring); |
---|
| 83 | // std::cout << "Submitted accept: " << req << std::endl; |
---|
[efdfdee] | 84 | } |
---|
| 85 | |
---|
| 86 | static void ring_request(struct io_uring * ring, int fd) { |
---|
| 87 | size_t size = 1024; |
---|
| 88 | auto req = request_t::create(EVENT_REQUEST, size); |
---|
| 89 | req->fd = fd; |
---|
| 90 | |
---|
[3acbf89] | 91 | struct io_uring_sqe * sqe = get_sqe(ring); |
---|
| 92 | io_uring_prep_read(sqe, fd, req->buff, size, 0); |
---|
[efdfdee] | 93 | io_uring_sqe_set_data(sqe, req); |
---|
[3acbf89] | 94 | submit(ring); |
---|
| 95 | // std::cout << "Submitted request: " << req << " (" << (void*)req->buffer << ")"<<std::endl; |
---|
[efdfdee] | 96 | } |
---|
| 97 | |
---|
| 98 | //========================================================= |
---|
| 99 | enum HttpCode { |
---|
| 100 | OK200 = 0, |
---|
| 101 | E400, |
---|
| 102 | E404, |
---|
| 103 | E405, |
---|
| 104 | E408, |
---|
| 105 | E413, |
---|
| 106 | E414, |
---|
| 107 | KNOWN_CODES |
---|
| 108 | }; |
---|
| 109 | |
---|
| 110 | const char * http_msgs[] = { |
---|
| 111 | "HTTP/1.1 200 OK\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: %zu \n\n%s", |
---|
| 112 | "HTTP/1.1 400 Bad Request\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n", |
---|
| 113 | "HTTP/1.1 404 Not Found\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n", |
---|
| 114 | "HTTP/1.1 405 Method Not Allowed\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n", |
---|
| 115 | "HTTP/1.1 408 Request Timeout\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n", |
---|
| 116 | "HTTP/1.1 413 Payload Too Large\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n", |
---|
| 117 | "HTTP/1.1 414 URI Too Long\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n", |
---|
| 118 | }; |
---|
| 119 | |
---|
| 120 | static_assert( KNOWN_CODES == (sizeof(http_msgs ) / sizeof(http_msgs [0]))); |
---|
| 121 | |
---|
| 122 | const int http_codes[] = { |
---|
| 123 | 200, |
---|
| 124 | 400, |
---|
| 125 | 404, |
---|
| 126 | 405, |
---|
| 127 | 408, |
---|
| 128 | 413, |
---|
| 129 | 414, |
---|
| 130 | }; |
---|
| 131 | |
---|
| 132 | static_assert( KNOWN_CODES == (sizeof(http_codes) / sizeof(http_codes[0]))); |
---|
| 133 | |
---|
| 134 | int code_val(HttpCode code) { |
---|
| 135 | return http_codes[code]; |
---|
| 136 | } |
---|
| 137 | |
---|
| 138 | static void ring_answer(struct io_uring * ring, int fd, HttpCode code) { |
---|
| 139 | size_t size = 256; |
---|
| 140 | auto req = request_t::create(EVENT_ANSWER, size); |
---|
| 141 | req->fd = fd; |
---|
| 142 | |
---|
| 143 | const char * fmt = http_msgs[code]; |
---|
| 144 | const char * date = ""; |
---|
[3acbf89] | 145 | size = snprintf(req->buff, size, fmt, date, size); |
---|
[efdfdee] | 146 | |
---|
[3acbf89] | 147 | struct io_uring_sqe * sqe = get_sqe(ring); |
---|
| 148 | io_uring_prep_write(sqe, fd, req->buff, size, 0); |
---|
[efdfdee] | 149 | io_uring_sqe_set_data(sqe, req); |
---|
[3acbf89] | 150 | submit(ring); |
---|
| 151 | // std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl; |
---|
[efdfdee] | 152 | } |
---|
| 153 | |
---|
[3acbf89] | 154 | static void ring_answer(struct io_uring * ring, int fd, const std::string &) { |
---|
| 155 | // size_t size = 256; |
---|
| 156 | // auto req = request_t::create(EVENT_ANSWER, size); |
---|
| 157 | // req->fd = fd; |
---|
| 158 | |
---|
| 159 | // const char * fmt = http_msgs[OK200]; |
---|
| 160 | // const char * date = ""; |
---|
| 161 | // size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str()); |
---|
| 162 | // req->length = len; |
---|
| 163 | |
---|
| 164 | // struct io_uring_sqe * sqe = get_sqe(ring); |
---|
| 165 | // io_uring_prep_write(sqe, fd, req->buffer, len, 0); |
---|
| 166 | // io_uring_sqe_set_data(sqe, req); |
---|
| 167 | // submit(ring); |
---|
| 168 | // std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl; |
---|
| 169 | |
---|
| 170 | |
---|
| 171 | static const char* RESPONSE = "HTTP/1.1 200 OK\r\n" \ |
---|
| 172 | "Content-Length: 15\r\n" \ |
---|
| 173 | "Content-Type: text/html\r\n" \ |
---|
| 174 | "Connection: keep-alive\r\n" \ |
---|
| 175 | "Server: testserver\r\n" \ |
---|
| 176 | "\r\n" \ |
---|
| 177 | "Hello, World!\r\n"; |
---|
| 178 | |
---|
| 179 | static const size_t RLEN = strlen(RESPONSE); |
---|
| 180 | |
---|
[efdfdee] | 181 | size_t size = 256; |
---|
| 182 | auto req = request_t::create(EVENT_ANSWER, size); |
---|
| 183 | req->fd = fd; |
---|
[3acbf89] | 184 | req->buff = (char*)RESPONSE; |
---|
| 185 | req->length = RLEN; |
---|
[efdfdee] | 186 | |
---|
[3acbf89] | 187 | // const char * fmt = http_msgs[OK200]; |
---|
| 188 | // const char * date = ""; |
---|
| 189 | // size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str()); |
---|
| 190 | // req->length = len; |
---|
[efdfdee] | 191 | |
---|
[3acbf89] | 192 | struct io_uring_sqe * sqe = get_sqe(ring); |
---|
| 193 | io_uring_prep_write(sqe, fd, RESPONSE, RLEN, 0); |
---|
[efdfdee] | 194 | io_uring_sqe_set_data(sqe, req); |
---|
[3acbf89] | 195 | submit(ring); |
---|
[efdfdee] | 196 | } |
---|
| 197 | |
---|
| 198 | //========================================================= |
---|
| 199 | static void handle_new_conn(struct io_uring * ring, int fd) { |
---|
| 200 | if( fd < 0 ) { |
---|
| 201 | int err = -fd; |
---|
| 202 | if( err == ECONNABORTED ) return; |
---|
| 203 | std::cerr << "accept error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 204 | exit(EXIT_FAILURE); |
---|
| 205 | } |
---|
| 206 | |
---|
| 207 | ring_request(ring, fd); |
---|
| 208 | } |
---|
| 209 | |
---|
| 210 | static void handle_request(struct io_uring * ring, struct request_t * in, int res) { |
---|
| 211 | if( res < 0 ) { |
---|
| 212 | int err = -res; |
---|
| 213 | switch(err) { |
---|
| 214 | case EPIPE: |
---|
| 215 | case ECONNRESET: |
---|
| 216 | close(in->fd); |
---|
| 217 | free(in); |
---|
| 218 | return; |
---|
| 219 | default: |
---|
[3acbf89] | 220 | std::cerr << "request error: (" << err << ") " << strerror(err) << std::endl; |
---|
[efdfdee] | 221 | exit(EXIT_FAILURE); |
---|
| 222 | } |
---|
| 223 | } |
---|
| 224 | |
---|
| 225 | if(res == 0) { |
---|
| 226 | close(in->fd); |
---|
| 227 | free(in); |
---|
| 228 | return; |
---|
| 229 | } |
---|
| 230 | |
---|
[3acbf89] | 231 | const char * it = in->buff; |
---|
[efdfdee] | 232 | if( !strstr( it, "\r\n\r\n" ) ) { |
---|
| 233 | std::cout << "Incomplete request" << std::endl; |
---|
| 234 | close(in->fd); |
---|
| 235 | free(in); |
---|
| 236 | return; |
---|
| 237 | } |
---|
| 238 | |
---|
[3acbf89] | 239 | it = in->buff; |
---|
[efdfdee] | 240 | const std::string reply = "Hello, World!\n"; |
---|
| 241 | int ret = memcmp(it, "GET ", 4); |
---|
| 242 | if( ret != 0 ) { |
---|
| 243 | ring_answer(ring, in->fd, E400); |
---|
| 244 | goto NEXT; |
---|
| 245 | } |
---|
| 246 | |
---|
| 247 | it += 4; |
---|
| 248 | ret = memcmp(it, "/plaintext", 10); |
---|
| 249 | if( ret != 0 ) { |
---|
| 250 | ring_answer(ring, in->fd, E404); |
---|
| 251 | goto NEXT; |
---|
| 252 | } |
---|
| 253 | |
---|
| 254 | ring_answer(ring, in->fd, reply); |
---|
| 255 | |
---|
| 256 | NEXT: |
---|
| 257 | ring_request(ring, in->fd); |
---|
| 258 | return; |
---|
| 259 | } |
---|
| 260 | |
---|
| 261 | static void handle_answer(struct io_uring * ring, struct request_t * in, int res) { |
---|
| 262 | if( res < 0 ) { |
---|
| 263 | int err = -res; |
---|
| 264 | switch(err) { |
---|
| 265 | case EPIPE: |
---|
| 266 | case ECONNRESET: |
---|
| 267 | close(in->fd); |
---|
| 268 | free(in); |
---|
| 269 | return; |
---|
| 270 | default: |
---|
| 271 | std::cerr << "answer error: (" << err << ") " << strerror(err) << std::endl; |
---|
| 272 | exit(EXIT_FAILURE); |
---|
| 273 | } |
---|
| 274 | } |
---|
| 275 | |
---|
| 276 | if( res >= in->length ) { |
---|
| 277 | free(in); |
---|
| 278 | return; |
---|
| 279 | } |
---|
| 280 | |
---|
[3acbf89] | 281 | struct io_uring_sqe * sqe = get_sqe(ring); |
---|
| 282 | io_uring_prep_write(sqe, in->fd, in->buff + res, in->length - res, 0); |
---|
[efdfdee] | 283 | io_uring_sqe_set_data(sqe, in); |
---|
[3acbf89] | 284 | submit(ring); |
---|
| 285 | // std::cout << "Re-Submitted request: " << in << " (" << (void*)in->buffer << ")"<<std::endl; |
---|
[efdfdee] | 286 | |
---|
| 287 | ring_request(ring, in->fd); |
---|
| 288 | } |
---|
| 289 | |
---|
| 290 | //========================================================= |
---|
[3acbf89] | 291 | extern "C" { |
---|
| 292 | extern int __io_uring_flush_sq(struct io_uring *ring); |
---|
| 293 | } |
---|
| 294 | |
---|
[efdfdee] | 295 | void * proc_loop(void * arg) { |
---|
[3acbf89] | 296 | size_t count = 0; |
---|
| 297 | struct options_t & opt = *(struct options_t *)arg; |
---|
[efdfdee] | 298 | |
---|
[f3e87af] | 299 | struct io_uring * ring = opt.ring; |
---|
[efdfdee] | 300 | |
---|
| 301 | char endfd_buf[8]; |
---|
| 302 | ring_end(ring, opt.endfd, endfd_buf, 8); |
---|
| 303 | |
---|
| 304 | ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags); |
---|
| 305 | |
---|
| 306 | bool done = false; |
---|
| 307 | while(!done) { |
---|
| 308 | struct io_uring_cqe *cqe; |
---|
[3acbf89] | 309 | int ret; |
---|
| 310 | while(-EAGAIN == (ret = io_uring_wait_cqe_nr(ring, &cqe, 0))) { |
---|
| 311 | ret = io_uring_submit_and_wait(ring, 1); |
---|
| 312 | if (ret < 0) { |
---|
| 313 | fprintf( stderr, "io_uring get error: (%d) %s\n", (int)-ret, strerror(-ret) ); |
---|
| 314 | exit(EXIT_FAILURE); |
---|
| 315 | } |
---|
| 316 | opt.result.subs += ret; |
---|
| 317 | opt.result.cnts++; |
---|
| 318 | } |
---|
| 319 | |
---|
| 320 | if (ret < 0 && -EAGAIN != ret) { |
---|
| 321 | fprintf( stderr, "io_uring peek error: (%d) %s\n", (int)-ret, strerror(-ret) ); |
---|
[efdfdee] | 322 | exit(EXIT_FAILURE); |
---|
| 323 | } |
---|
| 324 | |
---|
| 325 | auto req = (struct request_t *)cqe->user_data; |
---|
[3acbf89] | 326 | // std::cout << req << " completed with " << cqe->res << std::endl; |
---|
[efdfdee] | 327 | |
---|
| 328 | switch(req->type) { |
---|
| 329 | case EVENT_END: |
---|
| 330 | done = true; |
---|
| 331 | break; |
---|
| 332 | case EVENT_ACCEPT: |
---|
| 333 | handle_new_conn(ring, cqe->res); |
---|
| 334 | free(req); |
---|
| 335 | ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags); |
---|
| 336 | break; |
---|
| 337 | case EVENT_REQUEST: |
---|
| 338 | handle_request(ring, req, cqe->res); |
---|
| 339 | break; |
---|
| 340 | case EVENT_ANSWER: |
---|
| 341 | handle_answer(ring, req, cqe->res); |
---|
| 342 | break; |
---|
| 343 | } |
---|
| 344 | |
---|
| 345 | io_uring_cqe_seen(ring, cqe); |
---|
| 346 | } |
---|
| 347 | |
---|
[3acbf89] | 348 | return (void*)count; |
---|
[efdfdee] | 349 | } |
---|
| 350 | |
---|
| 351 | //========================================================= |
---|
[f3e87af] | 352 | struct __attribute__((aligned(128))) aligned_ring { |
---|
| 353 | struct io_uring storage; |
---|
| 354 | }; |
---|
| 355 | |
---|
[3acbf89] | 356 | #include <bit> |
---|
| 357 | |
---|
[efdfdee] | 358 | #include <pthread.h> |
---|
| 359 | extern "C" { |
---|
| 360 | #include <signal.h> |
---|
| 361 | #include <sys/eventfd.h> |
---|
| 362 | #include <sys/socket.h> |
---|
| 363 | #include <netinet/in.h> |
---|
| 364 | } |
---|
| 365 | |
---|
[3acbf89] | 366 | int main(int argc, char * argv[]) { |
---|
[efdfdee] | 367 | signal(SIGPIPE, SIG_IGN); |
---|
| 368 | |
---|
| 369 | unsigned nthreads = 1; |
---|
| 370 | unsigned port = 8800; |
---|
| 371 | unsigned entries = 256; |
---|
| 372 | unsigned backlog = 10; |
---|
[f3e87af] | 373 | bool attach = false; |
---|
[efdfdee] | 374 | |
---|
[3acbf89] | 375 | //=================== |
---|
| 376 | // Arguments |
---|
| 377 | int c; |
---|
[f3e87af] | 378 | while ((c = getopt (argc, argv, "t:p:e:b:a")) != -1) { |
---|
[3acbf89] | 379 | switch (c) |
---|
| 380 | { |
---|
| 381 | case 't': |
---|
| 382 | nthreads = atoi(optarg); |
---|
| 383 | break; |
---|
| 384 | case 'p': |
---|
| 385 | port = atoi(optarg); |
---|
| 386 | break; |
---|
| 387 | case 'e': |
---|
| 388 | entries = atoi(optarg); |
---|
| 389 | break; |
---|
| 390 | case 'b': |
---|
| 391 | backlog = atoi(optarg); |
---|
| 392 | break; |
---|
[f3e87af] | 393 | case 'a': |
---|
| 394 | attach = true; |
---|
| 395 | break; |
---|
[3acbf89] | 396 | case '?': |
---|
| 397 | default: |
---|
[f3e87af] | 398 | std::cerr << "Usage: -t <threads> -p <port> -e <entries> -b <backlog> -a" << std::endl; |
---|
[3acbf89] | 399 | return EXIT_FAILURE; |
---|
| 400 | } |
---|
| 401 | } |
---|
| 402 | |
---|
| 403 | if( !std::ispow2(entries) ) { |
---|
| 404 | unsigned v = entries; |
---|
| 405 | v--; |
---|
| 406 | v |= v >> 1; |
---|
| 407 | v |= v >> 2; |
---|
| 408 | v |= v >> 4; |
---|
| 409 | v |= v >> 8; |
---|
| 410 | v |= v >> 16; |
---|
| 411 | v++; |
---|
| 412 | std::cerr << "Warning: num_entries not a power of 2 (" << entries << ") raising to " << v << std::endl; |
---|
| 413 | entries = v; |
---|
| 414 | } |
---|
| 415 | |
---|
[efdfdee] | 416 | //=================== |
---|
| 417 | // End FD |
---|
| 418 | int efd = eventfd(0, EFD_SEMAPHORE); |
---|
| 419 | if (efd < 0) { |
---|
| 420 | std::cerr << "eventfd error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 421 | exit(EXIT_FAILURE); |
---|
| 422 | } |
---|
| 423 | |
---|
| 424 | //=================== |
---|
| 425 | // Open Socket |
---|
| 426 | std::cout << getpid() << " : Listening on port " << port << std::endl; |
---|
| 427 | int server_fd = socket(AF_INET, SOCK_STREAM, 0); |
---|
| 428 | if(server_fd < 0) { |
---|
| 429 | std::cerr << "socket error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 430 | exit(EXIT_FAILURE); |
---|
| 431 | } |
---|
| 432 | |
---|
| 433 | int ret = 0; |
---|
| 434 | struct sockaddr_in address; |
---|
| 435 | int addrlen = sizeof(address); |
---|
| 436 | memset( (char *)&address, '\0', addrlen ); |
---|
| 437 | address.sin_family = AF_INET; |
---|
| 438 | address.sin_addr.s_addr = htonl(INADDR_ANY); |
---|
| 439 | address.sin_port = htons( port ); |
---|
| 440 | |
---|
| 441 | int waited = 0; |
---|
| 442 | while(true) { |
---|
| 443 | ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) ); |
---|
| 444 | if(ret < 0) { |
---|
| 445 | if(errno == EADDRINUSE) { |
---|
| 446 | if(waited == 0) { |
---|
| 447 | std::cerr << "Waiting for port" << std::endl; |
---|
| 448 | } else { |
---|
| 449 | std::cerr << "\r" << waited; |
---|
| 450 | std::cerr.flush(); |
---|
| 451 | } |
---|
| 452 | waited ++; |
---|
| 453 | usleep( 1000000 ); |
---|
| 454 | continue; |
---|
| 455 | } |
---|
| 456 | std::cerr << "bind error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 457 | exit(EXIT_FAILURE); |
---|
| 458 | } |
---|
| 459 | break; |
---|
| 460 | } |
---|
| 461 | |
---|
| 462 | ret = listen( server_fd, backlog ); |
---|
| 463 | if(ret < 0) { |
---|
| 464 | std::cerr << "listen error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 465 | exit(EXIT_FAILURE); |
---|
| 466 | } |
---|
| 467 | |
---|
| 468 | //=================== |
---|
| 469 | // Run Server Threads |
---|
[f3e87af] | 470 | std::cout << "Starting " << nthreads << " Threads"; |
---|
| 471 | if(attach) { |
---|
| 472 | std::cout << " with attached Rings"; |
---|
| 473 | } |
---|
| 474 | std::cout << std::endl; |
---|
| 475 | |
---|
| 476 | aligned_ring thrd_rings[nthreads]; |
---|
| 477 | pthread_t thrd_hdls[nthreads]; |
---|
| 478 | options_t thrd_opts[nthreads]; |
---|
[efdfdee] | 479 | for(unsigned i = 0; i < nthreads; i++) { |
---|
[f3e87af] | 480 | if(!attach || i == 0) { |
---|
| 481 | io_uring_queue_init(entries, &thrd_rings[i].storage, 0); |
---|
| 482 | } |
---|
| 483 | else { |
---|
| 484 | struct io_uring_params p; |
---|
| 485 | memset(&p, 0, sizeof(p)); |
---|
| 486 | p.flags = IORING_SETUP_ATTACH_WQ; |
---|
| 487 | p.wq_fd = thrd_rings[0].storage.ring_fd; |
---|
| 488 | io_uring_queue_init_params(entries, &thrd_rings[i].storage, &p); |
---|
| 489 | } |
---|
| 490 | |
---|
[efdfdee] | 491 | thrd_opts[i].acpt.sockfd = server_fd; |
---|
| 492 | thrd_opts[i].acpt.addr = (struct sockaddr *)&address; |
---|
| 493 | thrd_opts[i].acpt.addrlen = (socklen_t*)&addrlen; |
---|
| 494 | thrd_opts[i].acpt.flags = 0; |
---|
[f3e87af] | 495 | thrd_opts[i].endfd = efd; |
---|
| 496 | thrd_opts[i].ring = &thrd_rings[i].storage; |
---|
[efdfdee] | 497 | |
---|
| 498 | int ret = pthread_create(&thrd_hdls[i], nullptr, proc_loop, &thrd_opts[i]); |
---|
| 499 | if (ret < 0) { |
---|
| 500 | std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 501 | exit(EXIT_FAILURE); |
---|
| 502 | } |
---|
| 503 | } |
---|
| 504 | |
---|
| 505 | //=================== |
---|
| 506 | // Server Started |
---|
| 507 | std::cout << "Server Started" << std::endl; |
---|
| 508 | { |
---|
| 509 | char buffer[128]; |
---|
| 510 | int ret; |
---|
| 511 | do { |
---|
| 512 | ret = read(STDIN_FILENO, buffer, 128); |
---|
| 513 | if(ret < 0) { |
---|
| 514 | std::cerr << "main read error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 515 | exit(EXIT_FAILURE); |
---|
| 516 | } |
---|
| 517 | else if(ret > 0) { |
---|
| 518 | std::cout << "User inputed '"; |
---|
| 519 | std::cout.write(buffer, ret); |
---|
| 520 | std::cout << "'" << std::endl; |
---|
| 521 | } |
---|
| 522 | } while(ret != 0); |
---|
| 523 | |
---|
| 524 | std::cout << "Shutdown received" << std::endl; |
---|
| 525 | } |
---|
| 526 | |
---|
| 527 | //=================== |
---|
| 528 | (std::cout << "Sending Shutdown to Threads... ").flush(); |
---|
| 529 | ret = eventfd_write(efd, nthreads); |
---|
| 530 | if (ret < 0) { |
---|
| 531 | std::cerr << "eventfd close error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 532 | exit(EXIT_FAILURE); |
---|
| 533 | } |
---|
| 534 | std::cout << "done" << std::endl; |
---|
| 535 | |
---|
| 536 | //=================== |
---|
| 537 | (std::cout << "Stopping Threads Done... ").flush(); |
---|
[3acbf89] | 538 | size_t total = 0; |
---|
| 539 | size_t count = 0; |
---|
[efdfdee] | 540 | for(unsigned i = 0; i < nthreads; i++) { |
---|
| 541 | void * retval; |
---|
| 542 | int ret = pthread_join(thrd_hdls[i], &retval); |
---|
| 543 | if (ret < 0) { |
---|
| 544 | std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 545 | exit(EXIT_FAILURE); |
---|
| 546 | } |
---|
[3acbf89] | 547 | // total += (size_t)retval; |
---|
| 548 | total += thrd_opts[i].result.subs; |
---|
| 549 | count += thrd_opts[i].result.cnts; |
---|
[f3e87af] | 550 | |
---|
| 551 | io_uring_queue_exit(thrd_opts[i].ring); |
---|
[efdfdee] | 552 | } |
---|
| 553 | std::cout << "done" << std::endl; |
---|
[3acbf89] | 554 | std::cout << "Submit average: " << total << "/" << count << "(" << (((double)total) / count) << ")" << std::endl; |
---|
[efdfdee] | 555 | |
---|
| 556 | //=================== |
---|
| 557 | (std::cout << "Closing Socket... ").flush(); |
---|
| 558 | ret = shutdown( server_fd, SHUT_RD ); |
---|
| 559 | if( ret < 0 ) { |
---|
| 560 | std::cerr << "shutdown socket error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 561 | exit(EXIT_FAILURE); |
---|
| 562 | } |
---|
| 563 | |
---|
| 564 | ret = close(server_fd); |
---|
| 565 | if (ret < 0) { |
---|
| 566 | std::cerr << "close socket error: (" << errno << ") " << strerror(errno) << std::endl; |
---|
| 567 | exit(EXIT_FAILURE); |
---|
| 568 | } |
---|
| 569 | std::cout << "done" << std::endl; |
---|
| 570 | } |
---|