Changes in / [aeb20a4:db7a3ad]
- Files:
-
- 3 deleted
- 10 edited
-
benchmark/io/http/Makefile.am (modified) (1 diff)
-
benchmark/io/http/main.cfa (modified) (10 diffs)
-
benchmark/io/http/options.cfa (modified) (4 diffs)
-
benchmark/io/http/options.hfa (modified) (2 diffs)
-
benchmark/io/http/printer.cfa (deleted)
-
benchmark/io/http/printer.hfa (deleted)
-
benchmark/io/http/protocol.cfa (modified) (5 diffs)
-
benchmark/io/http/protocol.hfa (modified) (2 diffs)
-
benchmark/io/http/socket.cfa (modified) (1 diff)
-
benchmark/io/http/worker.cfa (modified) (3 diffs)
-
benchmark/io/http/worker.hfa (modified) (3 diffs)
-
benchmark/io/sendfile/producer.cfa (deleted)
-
libcfa/src/containers/queueLockFree.hfa (modified) (1 diff)
Legend:
- Unmodified
- Added
- Removed
-
benchmark/io/http/Makefile.am
raeb20a4 rdb7a3ad 37 37 options.cfa \ 38 38 options.hfa \ 39 printer.cfa \40 printer.hfa \41 39 protocol.cfa \ 42 40 protocol.hfa \ -
benchmark/io/http/main.cfa
raeb20a4 rdb7a3ad 25 25 #include "options.hfa" 26 26 #include "socket.hfa" 27 #include "printer.hfa"28 27 #include "worker.hfa" 29 28 … … 32 31 Duration default_preemption() { 33 32 return 0; 33 } 34 35 //============================================================================================= 36 // Stats Printer 37 //=============================================================================================' 38 39 thread StatsPrinter { 40 Worker * workers; 41 int worker_cnt; 42 condition_variable(fast_block_lock) var; 43 }; 44 45 void ?{}( StatsPrinter & this, cluster & cl ) { 46 ((thread&)this){ "Stats Printer Thread", cl }; 47 this.worker_cnt = 0; 48 } 49 50 void ^?{}( StatsPrinter & mutex this ) {} 51 52 #define eng3(X) (ws(3, 3, unit(eng( X )))) 53 54 void main(StatsPrinter & this) { 55 LOOP: for() { 56 waitfor( ^?{} : this) { 57 break LOOP; 58 } 59 or else {} 60 61 wait(this.var, 10`s); 62 63 print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO ); 64 if(this.worker_cnt != 0) { 65 uint64_t tries = 0; 66 uint64_t calls = 0; 67 uint64_t header = 0; 68 uint64_t splcin = 0; 69 uint64_t splcot = 0; 70 struct { 71 volatile uint64_t calls; 72 volatile uint64_t bytes; 73 } avgrd[zipf_cnts]; 74 memset(avgrd, 0, sizeof(avgrd)); 75 76 for(i; this.worker_cnt) { 77 tries += this.workers[i].stats.sendfile.tries; 78 calls += this.workers[i].stats.sendfile.calls; 79 header += this.workers[i].stats.sendfile.header; 80 splcin += this.workers[i].stats.sendfile.splcin; 81 splcot += this.workers[i].stats.sendfile.splcot; 82 for(j; zipf_cnts) { 83 avgrd[j].calls += this.workers[i].stats.sendfile.avgrd[j].calls; 84 avgrd[j].bytes += this.workers[i].stats.sendfile.avgrd[j].bytes; 85 } 86 } 87 88 double ratio = ((double)tries) / calls; 89 90 sout | "----- Worker Stats -----"; 91 sout | "sendfile : " | calls | "calls," | tries | "tries (" | ratio | " try/call)"; 92 sout | " " | header | "header," | splcin | "splice in," | splcot | "splice out"; 93 sout | " - zipf sizes:"; 94 for(i; zipf_cnts) { 95 double written = avgrd[i].calls > 0 ? ((double)avgrd[i].bytes) / avgrd[i].calls : 0; 96 sout | " " | zipf_sizes[i] | "bytes," | avgrd[i].calls | "shorts," | written | "written"; 97 } 98 } 99 else { 100 sout | "No Workers!"; 101 } 102 } 34 103 } 35 104 … … 40 109 cluster self; 41 110 processor * procs; 111 // io_context * ctxs; 112 StatsPrinter * prnt; 42 113 43 114 }; … … 81 152 } 82 153 154 if(options.stats) { 155 this.prnt = alloc(); 156 (*this.prnt){ this.self }; 157 } else { 158 this.prnt = 0p; 159 } 160 83 161 #if !defined(__CFA_NO_STATISTICS__) 84 162 print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO ); 85 163 #endif 86 164 87 options.clopts.instance = &this.self; 165 options.clopts.instance[options.clopts.cltr_cnt] = &this.self; 166 options.clopts.cltr_cnt++; 88 167 } 89 168 90 169 void ^?{}( ServerCluster & this ) { 170 delete(this.prnt); 171 91 172 for(i; options.clopts.nprocs) { 92 173 ^(this.procs[i]){}; … … 99 180 extern void init_protocol(void); 100 181 extern void deinit_protocol(void); 101 102 //=============================================================================================103 // REUSEPORT104 //=============================================================================================105 106 size_t sockarr_size;107 struct __attribute__((aligned(128))) Q {108 mpsc_queue(PendingRead) q;109 };110 182 111 183 //============================================================================================= … … 163 235 164 236 int server_fd; 237 if(!options.socket.manyreuse) { 238 server_fd = listener(address, addrlen); 239 } 165 240 166 241 //=================== … … 182 257 { 183 258 // Stats printer makes a copy so this needs to persist longer than normal 184 connection ** conns; 185 AcceptWorker * aworkers = 0p; 186 ChannelWorker * cworkers = 0p; 187 Acceptor * acceptors = 0p; 188 Q * queues = 0p; 189 ServerCluster cl; 190 191 if(options.stats) { 192 stats_thrd = alloc(); 193 (*stats_thrd){ cl.self }; 194 } else { 195 stats_thrd = 0p; 196 } 259 Worker * workers; 260 ServerCluster cl[options.clopts.nclusters]; 197 261 198 262 init_protocol(); 199 263 { 200 conns = alloc(options.clopts.nworkers); 201 if(options.socket.reuseport) { 202 queues = alloc(options.clopts.nprocs); 203 acceptors = anew(options.clopts.nprocs); 204 for(i; options.clopts.nprocs) { 205 (queues[i]){}; 206 { 207 acceptors[i].sockfd = listener(address, addrlen); 208 acceptors[i].addr = (struct sockaddr *)&address; 209 acceptors[i].addrlen = (socklen_t*)&addrlen; 210 acceptors[i].flags = 0; 211 acceptors[i].queue = &queues[i].q; 212 } 213 unpark( acceptors[i] ); 214 } 215 216 cworkers = anew(options.clopts.nworkers); 217 for(i; options.clopts.nworkers) { 218 { 219 cworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0]; 220 cworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1]; 221 cworkers[i].queue = &queues[i % options.clopts.nprocs].q; 222 conns[i] = &cworkers[i].conn; 223 } 224 unpark( cworkers[i] ); 225 } 226 } 227 else { 228 server_fd = listener(address, addrlen); 229 aworkers = anew(options.clopts.nworkers); 230 for(i; options.clopts.nworkers) { 231 // if( options.file_cache.fixed_fds ) { 232 // workers[i].pipe[0] = pipe_off + (i * 2) + 0; 233 // workers[i].pipe[1] = pipe_off + (i * 2) + 1; 234 // } 235 // else 236 { 237 aworkers[i].conn.pipe[0] = fds[pipe_off + (i * 2) + 0]; 238 aworkers[i].conn.pipe[1] = fds[pipe_off + (i * 2) + 1]; 239 aworkers[i].sockfd = server_fd; 240 aworkers[i].addr = (struct sockaddr *)&address; 241 aworkers[i].addrlen = (socklen_t*)&addrlen; 242 aworkers[i].flags = 0; 243 conns[i] = &aworkers[i].conn; 244 } 245 unpark( aworkers[i] ); 246 } 247 } 248 249 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors"; 264 workers = anew(options.clopts.nworkers); 265 cl[0].prnt->workers = workers; 266 cl[0].prnt->worker_cnt = options.clopts.nworkers; 267 for(i; options.clopts.nworkers) { 268 // if( options.file_cache.fixed_fds ) { 269 // workers[i].pipe[0] = pipe_off + (i * 2) + 0; 270 // workers[i].pipe[1] = pipe_off + (i * 2) + 1; 271 // } 272 // else 273 { 274 workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0]; 275 workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1]; 276 workers[i].sockfd = options.socket.manyreuse ? listener(address, addrlen) : server_fd; 277 workers[i].addr = (struct sockaddr *)&address; 278 workers[i].addrlen = (socklen_t*)&addrlen; 279 workers[i].flags = 0; 280 } 281 unpark( workers[i] ); 282 } 283 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters"; 284 for(i; options.clopts.nclusters) { 285 sout | options.clopts.thrd_cnt[i] | nonl; 286 } 250 287 sout | nl; 251 288 { … … 270 307 } 271 308 272 //=================== 273 // Close Socket and join 274 if(options.socket.reuseport) { 275 sout | "Notifying connections..." | nonl; flush( sout ); 276 for(i; options.clopts.nprocs) { 277 acceptors[i].done = true; 278 } 309 sout | "Notifying connections..." | nonl; flush( sout ); 310 for(i; options.clopts.nworkers) { 311 workers[i].done = true; 312 } 313 sout | "done"; 314 315 sout | "Shutting down socket..." | nonl; flush( sout ); 316 if(options.socket.manyreuse) { 279 317 for(i; options.clopts.nworkers) { 280 cworkers[i].done = true; 281 } 282 sout | "done"; 283 284 sout | "Shutting down Socket..." | nonl; flush( sout ); 285 for(i; options.clopts.nprocs) { 286 ret = shutdown( acceptors[i].sockfd, SHUT_RD ); 287 if( ret < 0 ) { 288 abort( "shutdown1 error: (%d) %s\n", (int)errno, strerror(errno) ); 289 } 290 } 291 sout | "done"; 292 293 sout | "Closing Socket..." | nonl; flush( sout ); 294 for(i; options.clopts.nprocs) { 295 ret = close( acceptors[i].sockfd ); 296 if( ret < 0) { 297 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); 298 } 299 } 300 sout | "done"; 301 302 sout | "Stopping accept threads..." | nonl; flush( sout ); 303 for(i; options.clopts.nprocs) { 304 join(acceptors[i]); 305 } 306 sout | "done"; 307 308 sout | "Draining worker queues..." | nonl; flush( sout ); 309 for(i; options.clopts.nprocs) { 310 PendingRead * p = 0p; 311 while(p = pop(queues[i].q)) { 312 fulfil(p->f, -ECONNRESET); 313 } 314 } 315 sout | "done"; 316 317 sout | "Stopping worker threads..." | nonl; flush( sout ); 318 for(i; options.clopts.nworkers) { 319 for(j; 2) { 320 ret = close(cworkers[i].conn.pipe[j]); 321 if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); 322 } 323 join(cworkers[i]); 318 ret = shutdown( workers[i].sockfd, SHUT_RD ); 319 if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) ); 324 320 } 325 321 } 326 322 else { 327 sout | "Notifying connections..." | nonl; flush( sout );328 for(i; options.clopts.nworkers) {329 aworkers[i].done = true;330 }331 sout | "done";332 333 sout | "Shutting down Socket..." | nonl; flush( sout );334 323 ret = shutdown( server_fd, SHUT_RD ); 335 324 if( ret < 0 ) { 336 abort( "shutdown2 error: (%d) %s\n", (int)errno, strerror(errno) ); 337 } 338 sout | "done"; 339 340 sout | "Closing Socket..." | nonl; flush( sout ); 325 abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); 326 } 327 } 328 sout | "done"; 329 330 //=================== 331 // Close Socket 332 sout | "Closing Socket..." | nonl; flush( sout ); 333 if(options.socket.manyreuse) { 334 for(i; options.clopts.nworkers) { 335 ret = close(workers[i].sockfd); 336 if(ret < 0) abort( "close socket %d error: (%d) %s\n", i, (int)errno, strerror(errno) ); 337 } 338 } 339 else { 341 340 ret = close( server_fd ); 342 341 if(ret < 0) { 343 342 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); 344 343 } 345 sout | "done";346 347 sout | "Stopping connection threads..." | nonl; flush( sout ); 348 for(i; options.clopts.nworkers) {349 for(j; 2) {350 ret = close(aworkers[i].conn.pipe[j]);351 if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno));352 }353 join(aworkers[i]);354 }344 } 345 sout | "done"; 346 347 sout | "Stopping connection threads..." | nonl; flush( sout ); 348 for(i; options.clopts.nworkers) { 349 for(j; 2) { 350 ret = close(workers[i].pipe[j]); 351 if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); 352 } 353 join(workers[i]); 355 354 } 356 355 } … … 362 361 363 362 sout | "Stopping printer threads..." | nonl; flush( sout ); 364 if(stats_thrd) { 365 notify_one(stats_thrd->var); 366 } 367 delete(stats_thrd); 363 for(i; options.clopts.nclusters) { 364 StatsPrinter * p = cl[i].prnt; 365 if(p) { 366 notify_one(p->var); 367 join(*p); 368 } 369 } 368 370 sout | "done"; 369 371 370 372 // Now that the stats printer is stopped, we can reclaim this 371 adelete(aworkers); 372 adelete(cworkers); 373 adelete(acceptors); 374 adelete(queues); 375 free(conns); 373 adelete(workers); 376 374 377 375 sout | "Stopping processors/clusters..." | nonl; flush( sout ); … … 379 377 sout | "done"; 380 378 379 // sout | "Closing splice fds..." | nonl; flush( sout ); 380 // for(i; pipe_cnt) { 381 // ret = close( fds[pipe_off + i] ); 382 // if(ret < 0) { 383 // abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) ); 384 // } 385 // } 381 386 free(fds); 387 sout | "done"; 382 388 383 389 sout | "Stopping processors..." | nonl; flush( sout ); -
benchmark/io/http/options.cfa
raeb20a4 rdb7a3ad 38 38 10, // backlog 39 39 1024, // buflen 40 false // reuseport 40 false, // onereuse 41 false // manyreuse 41 42 }, 42 43 43 44 { // cluster 45 1, // nclusters; 44 46 1, // nprocs; 45 47 1, // nworkers; … … 52 54 53 55 void parse_options( int argc, char * argv[] ) { 56 // bool fixedfd = false; 57 // bool sqkpoll = false; 58 // bool iokpoll = false; 54 59 unsigned nentries = 0; 60 bool isolate = false; 61 62 55 63 static cfa_option opt[] = { 56 64 { 'p', "port", "Port the server will listen on", options.socket.port}, 57 65 { 'c', "cpus", "Number of processors to use", options.clopts.nprocs}, 58 66 { 't', "threads", "Number of worker threads to use", options.clopts.nworkers}, 67 {'\0', "isolate", "Create one cluster per processor", isolate, parse_settrue}, 59 68 {'\0', "log", "Enable logs", options.log, parse_settrue}, 60 69 {'\0', "sout", "Redirect standard out to file", options.reopen_stdout}, … … 63 72 {'\0', "shell", "Disable interactive mode", options.interactive, parse_setfalse}, 64 73 {'\0', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog}, 65 {'\0', "reuseport", "Use acceptor threads with reuse port SO_REUSEPORT", options.socket.reuseport, parse_settrue}, 74 {'\0', "reuseport-one", "Create a single listen socket with SO_REUSEPORT", options.socket.onereuse, parse_settrue}, 75 {'\0', "reuseport", "Use many listen sockets with SO_REUSEPORT", options.socket.manyreuse, parse_settrue}, 66 76 {'\0', "request_len", "Maximum number of bytes in the http request, requests with more data will be answered with Http Code 414", options.socket.buflen}, 67 77 {'\0', "seed", "seed to use for hashing", options.file_cache.hash_seed }, … … 91 101 nentries = v; 92 102 } 103 if(isolate) { 104 options.clopts.nclusters = options.clopts.nprocs; 105 options.clopts.nprocs = 1; 106 } 93 107 options.clopts.params.num_entries = nentries; 94 options.clopts.instance = 0p; 95 options.clopts.thrd_cnt = 0; 108 options.clopts.instance = alloc(options.clopts.nclusters); 109 options.clopts.thrd_cnt = alloc(options.clopts.nclusters); 110 options.clopts.cltr_cnt = 0; 111 for(i; options.clopts.nclusters) { 112 options.clopts.thrd_cnt[i] = 0; 113 } 96 114 97 115 -
benchmark/io/http/options.hfa
raeb20a4 rdb7a3ad 27 27 int backlog; 28 28 int buflen; 29 bool reuseport; 29 bool onereuse; 30 bool manyreuse; 30 31 } socket; 31 32 32 33 struct { 34 int nclusters; 33 35 int nprocs; 34 36 int nworkers; … … 36 38 bool procstats; 37 39 bool viewhalts; 38 cluster * instance; 39 size_t thrd_cnt; 40 cluster ** instance; 41 size_t * thrd_cnt; 42 size_t cltr_cnt; 40 43 } clopts; 41 44 }; -
benchmark/io/http/protocol.cfa
raeb20a4 rdb7a3ad 30 30 #define PLAINTEXT_NOCOPY 31 31 #define LINKED_IO 32 33 static inline __s32 wait_res( io_future_t & this ) {34 wait( this );35 if( this.result < 0 ) {{36 errno = -this.result;37 return -1;38 }}39 return this.result;40 }41 32 42 33 struct https_msg_str { … … 479 470 480 471 if(is_error(splice_in.res)) { 481 if(splice_in.res.error == -EPIPE) return -ECONNRESET;482 472 mutex(serr) serr | "SPLICE IN failed with" | splice_in.res.error; 483 473 close(fd); … … 513 503 } 514 504 515 [HttpCode code, bool closed, * const char file, size_t len] http_read( volatile int & fd, []char buffer, size_t len, io_future_t * f) {505 [HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) { 516 506 char * it = buffer; 517 507 size_t count = len - 1; … … 519 509 READ: 520 510 for() { 521 int ret; 522 if( f ) { 523 ret = wait_res(*f); 524 reset(*f); 525 f = 0p; 526 } else { 527 ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY); 528 } 511 int ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY); 529 512 // int ret = read(fd, (void*)it, count); 530 513 if(ret == 0 ) return [OK200, true, 0, 0]; … … 587 570 588 571 void ?{}( DateFormater & this ) { 589 ((thread&)this){ "Server Date Thread", *options.clopts.instance };572 ((thread&)this){ "Server Date Thread", *options.clopts.instance[0] }; 590 573 this.idx = 0; 591 574 memset( &this.buffers[0], 0, sizeof(this.buffers[0]) ); -
benchmark/io/http/protocol.hfa
raeb20a4 rdb7a3ad 1 1 #pragma once 2 2 3 struct io_future_t;4 3 struct sendfile_stats_t; 5 4 … … 23 22 int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t count, struct sendfile_stats_t & ); 24 23 25 [HttpCode code, bool closed, * const char file, size_t len] http_read( volatile int & fd, []char buffer, size_t len, io_future_t * f);24 [HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len); -
benchmark/io/http/socket.cfa
raeb20a4 rdb7a3ad 26 26 27 27 int listener(struct sockaddr_in & address, int addrlen) { 28 int type = SOCK_STREAM; 29 if(options.socket.reuseport) type |= SOCK_NONBLOCK; 30 int sockfd = socket(AF_INET, type, 0); 28 int sockfd = socket(AF_INET, SOCK_STREAM, 0); 31 29 if(sockfd < 0) { 32 30 abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) ); 33 31 } 34 32 35 if(options.socket. reuseport) {33 if(options.socket.onereuse || options.socket.manyreuse) { 36 34 int value = 1; 37 35 // if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (const void*)&on, sizeof(on))) -
benchmark/io/http/worker.cfa
raeb20a4 rdb7a3ad 8 8 #include <fstream.hfa> 9 9 #include <iofwd.hfa> 10 #include <mutex_stmt.hfa>11 10 12 11 #include "options.hfa" … … 15 14 16 15 //============================================================================================= 17 // Generic connection handling16 // Worker Thread 18 17 //============================================================================================= 19 static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f, unsigned long long & last) {20 REQUEST:21 for() {22 bool closed;23 HttpCode code;24 const char * file;25 size_t name_size;18 void ?{}( Worker & this ) { 19 size_t cli = rand() % options.clopts.cltr_cnt; 20 ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 }; 21 options.clopts.thrd_cnt[cli]++; 22 this.pipe[0] = -1; 23 this.pipe[1] = -1; 24 this.done = false; 26 25 27 // Read the http request 28 if( options.log ) sout | "=== Reading request ==="; 29 [code, closed, file, name_size] = http_read(fd, buffer, len, f); 30 f = 0p; 31 32 // if we are done, break out of the loop 33 if( closed ) break REQUEST; 34 35 // If this wasn't a request retrun 400 36 if( code != OK200 ) { 37 sout | "=== Invalid Request :" | code_val(code) | "==="; 38 answer_error(fd, code); 39 continue REQUEST; 40 } 41 42 if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) { 43 if( options.log ) sout | "=== Request for /plaintext ==="; 44 45 int ret = answer_plaintext(fd); 46 if( ret == -ECONNRESET ) break REQUEST; 47 48 if( options.log ) sout | "=== Answer sent ==="; 49 continue REQUEST; 50 } 51 52 if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) { 53 if( options.log ) sout | "=== Request for /ping ==="; 54 55 // Send the header 56 int ret = answer_empty(fd); 57 if( ret == -ECONNRESET ) break REQUEST; 58 59 if( options.log ) sout | "=== Answer sent ==="; 60 continue REQUEST; 61 } 62 63 if( options.log ) { 64 sout | "=== Request for file " | nonl; 65 write(sout, file, name_size); 66 sout | " ==="; 67 } 68 69 if( !options.file_cache.path ) { 70 if( options.log ) { 71 sout | "=== File Not Found (" | nonl; 72 write(sout, file, name_size); 73 sout | ") ==="; 74 } 75 answer_error(fd, E405); 76 continue REQUEST; 77 } 78 79 // Get the fd from the file cache 80 int ans_fd; 81 size_t count; 82 [ans_fd, count] = get_file( file, name_size ); 83 84 // If we can't find the file, return 404 85 if( ans_fd < 0 ) { 86 if( options.log ) { 87 sout | "=== File Not Found (" | nonl; 88 write(sout, file, name_size); 89 sout | ") ==="; 90 } 91 answer_error(fd, E404); 92 continue REQUEST; 93 } 94 95 // Send the desired file 96 int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile ); 97 if( ret == -ECONNRESET ) break REQUEST; 98 99 if( options.log ) sout | "=== Answer sent ==="; 100 } 101 102 if (stats_thrd) { 103 unsigned long long next = rdtscl(); 104 if(next > (last + 500000000)) { 105 if(try_lock(stats_thrd->stats.lock)) { 106 push(this.stats.sendfile, stats_thrd->stats.send); 107 unlock(stats_thrd->stats.lock); 108 last = next; 109 } 110 } 26 this.stats.sendfile.calls = 0; 27 this.stats.sendfile.tries = 0; 28 this.stats.sendfile.header = 0; 29 this.stats.sendfile.splcin = 0; 30 this.stats.sendfile.splcot = 0; 31 for(i; zipf_cnts) { 32 this.stats.sendfile.avgrd[i].calls = 0; 33 this.stats.sendfile.avgrd[i].bytes = 0; 111 34 } 112 35 } 113 36 114 //============================================================================================= 115 // Self Accepting Worker Thread 116 //============================================================================================= 117 void ?{}( AcceptWorker & this ) { 118 ((thread&)this){ "Server Worker Thread", *options.clopts.instance, 64000 }; 119 options.clopts.thrd_cnt++; 120 this.done = false; 37 extern "C" { 38 extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags); 121 39 } 122 40 123 void main( AcceptWorker & this ) {41 void main( Worker & this ) { 124 42 park(); 125 unsigned long long last = rdtscl(); 126 /* paranoid */ assert( this.conn.pipe[0] != -1 ); 127 /* paranoid */ assert( this.conn.pipe[1] != -1 ); 43 /* paranoid */ assert( this.pipe[0] != -1 ); 44 /* paranoid */ assert( this.pipe[1] != -1 ); 45 46 const bool reuse = options.socket.manyreuse; 47 48 CONNECTION: 128 49 for() { 129 50 if( options.log ) sout | "=== Accepting connection ==="; 130 int fd = cfa_accept4( this. sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY );51 int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], CFA_IO_LAZY ); 131 52 if(fd < 0) { 132 53 if( errno == ECONNABORTED ) break; … … 137 58 138 59 if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ==="; 139 size_t len = options.socket.buflen; 140 char buffer[len]; 141 handle_connection( this.conn, fd, buffer, len, 0p, last ); 60 REQUEST: 61 for() { 62 bool closed; 63 HttpCode code; 64 const char * file; 65 size_t name_size; 66 67 // Read the http request 68 size_t len = options.socket.buflen; 69 char buffer[len]; 70 if( options.log ) sout | "=== Reading request ==="; 71 [code, closed, file, name_size] = http_read(fd, buffer, len); 72 73 // if we are done, break out of the loop 74 if( closed ) break REQUEST; 75 76 // If this wasn't a request retrun 400 77 if( code != OK200 ) { 78 sout | "=== Invalid Request :" | code_val(code) | "==="; 79 answer_error(fd, code); 80 continue REQUEST; 81 } 82 83 if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) { 84 if( options.log ) sout | "=== Request for /plaintext ==="; 85 86 int ret = answer_plaintext(fd); 87 if( ret == -ECONNRESET ) break REQUEST; 88 89 if( options.log ) sout | "=== Answer sent ==="; 90 continue REQUEST; 91 } 92 93 if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) { 94 if( options.log ) sout | "=== Request for /ping ==="; 95 96 // Send the header 97 int ret = answer_empty(fd); 98 if( ret == -ECONNRESET ) break REQUEST; 99 100 if( options.log ) sout | "=== Answer sent ==="; 101 continue REQUEST; 102 } 103 104 if( options.log ) { 105 sout | "=== Request for file " | nonl; 106 write(sout, file, name_size); 107 sout | " ==="; 108 } 109 110 if( !options.file_cache.path ) { 111 if( options.log ) { 112 sout | "=== File Not Found (" | nonl; 113 write(sout, file, name_size); 114 sout | ") ==="; 115 } 116 answer_error(fd, E405); 117 continue REQUEST; 118 } 119 120 // Get the fd from the file cache 121 int ans_fd; 122 size_t count; 123 [ans_fd, count] = get_file( file, name_size ); 124 125 // If we can't find the file, return 404 126 if( ans_fd < 0 ) { 127 if( options.log ) { 128 sout | "=== File Not Found (" | nonl; 129 write(sout, file, name_size); 130 sout | ") ==="; 131 } 132 answer_error(fd, E404); 133 continue REQUEST; 134 } 135 136 // Send the desired file 137 int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile ); 138 if( ret == -ECONNRESET ) break REQUEST; 139 140 if( options.log ) sout | "=== Answer sent ==="; 141 } 142 142 143 143 if( options.log ) sout | "=== Connection closed ==="; 144 continue CONNECTION; 144 145 } 145 146 } 146 147 148 //=============================================================================================149 // Channel Worker Thread150 //=============================================================================================151 void ?{}( ChannelWorker & this ) {152 ((thread&)this){ "Server Worker Thread", *options.clopts.instance, 64000 };153 options.clopts.thrd_cnt++;154 this.done = false;155 }156 157 void main( ChannelWorker & this ) {158 park();159 unsigned long long last = rdtscl();160 /* paranoid */ assert( this.conn.pipe[0] != -1 );161 /* paranoid */ assert( this.conn.pipe[1] != -1 );162 for() {163 size_t len = options.socket.buflen;164 char buffer[len];165 PendingRead p;166 p.in.buf = (void*)buffer;167 p.in.len = len;168 push(*this.queue, &p);169 170 if( options.log ) sout | "=== Waiting new connection ===";171 handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last );172 173 if( options.log ) sout | "=== Connection closed ===";174 if(this.done) break;175 }176 }177 178 extern "C" {179 extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);180 }181 182 void ?{}( Acceptor & this ) {183 ((thread&)this){ "Server Worker Thread", *options.clopts.instance, 64000 };184 options.clopts.thrd_cnt++;185 this.done = false;186 }187 188 void main( Acceptor & this ) {189 park();190 unsigned long long last = rdtscl();191 if( options.log ) sout | "=== Accepting connection ===";192 for() {193 int fd = accept4(this.sockfd, this.[addr, addrlen, flags]);194 if(fd < 0) {195 if( errno == EWOULDBLOCK) {196 this.stats.eagains++;197 yield();198 continue;199 }200 if( errno == ECONNABORTED ) break;201 if( this.done && (errno == EINVAL || errno == EBADF) ) break;202 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );203 }204 this.stats.accepts++;205 206 if(this.done) return;207 208 if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";209 210 if(fd) {211 PendingRead * p = 0p;212 for() {213 if(this.done) return;214 p = pop(*this.queue);215 if(p) break;216 yield();217 this.stats.creates++;218 };219 220 p->out.fd = fd;221 async_recv(p->f, p->out.fd, p->in.buf, p->in.len, 0, CFA_IO_LAZY);222 }223 224 if (stats_thrd) {225 unsigned long long next = rdtscl();226 if(next > (last + 500000000)) {227 if(try_lock(stats_thrd->stats.lock)) {228 push(this.stats, stats_thrd->stats.accpt);229 unlock(stats_thrd->stats.lock);230 last = next;231 }232 }233 }234 235 if( options.log ) sout | "=== Accepting connection ===";236 }237 } -
benchmark/io/http/worker.hfa
raeb20a4 rdb7a3ad 1 1 #pragma once 2 2 3 #include <iofwd.hfa>4 #include <queueLockFree.hfa>5 3 #include <thread.hfa> 6 4 … … 9 7 } 10 8 11 #include "printer.hfa"12 13 9 //============================================================================================= 14 10 // Worker Thread 15 11 //============================================================================================= 16 12 17 struct connection { 18 int pipe[2]; 13 extern const size_t zipf_sizes[]; 14 enum { zipf_cnts = 36, }; 15 16 struct sendfile_stats_t { 17 volatile uint64_t calls; 18 volatile uint64_t tries; 19 volatile uint64_t header; 20 volatile uint64_t splcin; 21 volatile uint64_t splcot; 19 22 struct { 20 sendfile_stats_t sendfile; 21 } stats; 23 volatile uint64_t calls; 24 volatile uint64_t bytes; 25 } avgrd[zipf_cnts]; 22 26 }; 23 27 24 static inline void ?{}( connection & this ) { 25 this.pipe[0] = -1; 26 this.pipe[1] = -1; 27 } 28 29 thread AcceptWorker { 30 connection conn; 28 thread Worker { 29 int pipe[2]; 31 30 int sockfd; 32 31 struct sockaddr * addr; … … 34 33 int flags; 35 34 volatile bool done; 35 struct { 36 sendfile_stats_t sendfile; 37 } stats; 36 38 }; 37 void ?{}( AcceptWorker & this); 38 void main( AcceptWorker & ); 39 40 41 struct PendingRead { 42 PendingRead * volatile next; 43 io_future_t f; 44 struct { 45 void * buf; 46 size_t len; 47 } in; 48 struct { 49 volatile int fd; 50 } out; 51 }; 52 53 static inline PendingRead * volatile & ?`next ( PendingRead * node ) { 54 return node->next; 55 } 56 57 thread ChannelWorker { 58 connection conn; 59 volatile bool done; 60 mpsc_queue(PendingRead) * queue; 61 }; 62 void ?{}( ChannelWorker & ); 63 void main( ChannelWorker & ); 64 65 thread Acceptor { 66 mpsc_queue(PendingRead) * queue; 67 int sockfd; 68 struct sockaddr * addr; 69 socklen_t * addrlen; 70 int flags; 71 volatile bool done; 72 acceptor_stats_t stats; 73 }; 74 void ?{}( Acceptor & ); 75 void main( Acceptor & ); 39 void ?{}( Worker & this); 40 void main( Worker & ); -
libcfa/src/containers/queueLockFree.hfa
raeb20a4 rdb7a3ad 2 2 3 3 #include <assert.h> 4 5 #include <bits/defs.hfa>6 4 7 5 forall( T &) {
Note:
See TracChangeset
for help on using the changeset viewer.
