Changes in benchmark/io/http/main.cfa [ef3c383:329e26a]
- File:
-
- 1 edited
-
benchmark/io/http/main.cfa (modified) (16 diffs)
Legend:
- Unmodified
- Added
- Removed
-
benchmark/io/http/main.cfa
ref3c383 r329e26a 2 2 3 3 #include <errno.h> 4 #include <signal.h> 4 5 #include <stdio.h> 5 6 #include <string.h> … … 8 9 #include <sched.h> 9 10 #include <signal.h> 11 #include <sys/eventfd.h> 10 12 #include <sys/socket.h> 11 13 #include <netinet/in.h> … … 14 16 #include <fstream.hfa> 15 17 #include <kernel.hfa> 18 #include <locks.hfa> 16 19 #include <iofwd.hfa> 17 20 #include <stats.hfa> … … 21 24 #include "filecache.hfa" 22 25 #include "options.hfa" 26 #include "socket.hfa" 27 #include "printer.hfa" 23 28 #include "worker.hfa" 24 29 … … 30 35 31 36 //============================================================================================= 32 // Stats Printer33 //============================================================================================='34 35 thread StatsPrinter {36 Worker * workers;37 int worker_cnt;38 };39 40 void ?{}( StatsPrinter & this, cluster & cl ) {41 ((thread&)this){ "Stats Printer Thread", cl };42 this.worker_cnt = 0;43 }44 45 void ^?{}( StatsPrinter & mutex this ) {}46 47 #define eng3(X) (ws(3, 3, unit(eng( X ))))48 49 void main(StatsPrinter & this) {50 LOOP: for() {51 waitfor( ^?{} : this) {52 break LOOP;53 }54 or else {}55 56 sleep(10`s);57 58 print_stats_now( *active_cluster(), CFA_STATS_READY_Q | CFA_STATS_IO );59 if(this.worker_cnt != 0) {60 uint64_t tries = 0;61 uint64_t calls = 0;62 uint64_t header = 0;63 uint64_t splcin = 0;64 uint64_t splcot = 0;65 struct {66 volatile uint64_t calls;67 volatile uint64_t bytes;68 } avgrd[zipf_cnts];69 memset(avgrd, 0, sizeof(avgrd));70 71 for(i; this.worker_cnt) {72 tries += this.workers[i].stats.sendfile.tries;73 calls += this.workers[i].stats.sendfile.calls;74 header += this.workers[i].stats.sendfile.header;75 splcin += this.workers[i].stats.sendfile.splcin;76 splcot += this.workers[i].stats.sendfile.splcot;77 for(j; zipf_cnts) {78 avgrd[j].calls += this.workers[i].stats.sendfile.avgrd[j].calls;79 avgrd[j].bytes += this.workers[i].stats.sendfile.avgrd[j].bytes;80 }81 }82 83 double ratio = ((double)tries) / calls;84 85 sout | "----- Worker Stats -----";86 sout | "sendfile : " | calls | "calls," | tries | "tries (" | ratio | " try/call)";87 sout | " " | header | "header," | splcin | "splice in," | splcot | "splice out";88 sout | " - zipf sizes:";89 for(i; zipf_cnts) {90 double written = avgrd[i].calls > 0 ? ((double)avgrd[i].bytes) / avgrd[i].calls : 0;91 sout | " " | zipf_sizes[i] | "bytes," | avgrd[i].calls | "shorts," | written | "written";92 }93 }94 else {95 sout | "No Workers!";96 }97 }98 }99 100 //=============================================================================================101 37 // Globals 102 38 //============================================================================================= 103 struct ServerCluster {104 cluster self;105 processor * procs;106 // io_context * ctxs;107 StatsPrinter * prnt;108 109 };110 111 39 void ?{}( ServerCluster & this ) { 112 40 (this.self){ "Server Cluster", options.clopts.params }; … … 122 50 (this.procs[i]){ "Benchmark Processor", this.self }; 123 51 124 int c = 0;125 int n = 1 + (i % cnt);126 for(int j = 0; j < CPU_SETSIZE; j++) {127 if(CPU_ISSET(j, &fullset)) n--;128 if(n == 0) {129 c = j;130 break;131 }132 }133 cpu_set_t localset;134 CPU_ZERO(&localset);135 CPU_SET(c, &localset);136 ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset);137 if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret );52 // int c = 0; 53 // int n = 1 + (i % cnt); 54 // for(int j = 0; j < CPU_SETSIZE; j++) { 55 // if(CPU_ISSET(j, &fullset)) n--; 56 // if(n == 0) { 57 // c = j; 58 // break; 59 // } 60 // } 61 // cpu_set_t localset; 62 // CPU_ZERO(&localset); 63 // CPU_SET(c, &localset); 64 // ret = pthread_setaffinity_np(this.procs[i].kernel_thread, sizeof(localset), &localset); 65 // if( ret != 0 ) abort | "sched_getaffinity failed with" | ret | strerror( ret ); 138 66 139 67 #if !defined(__CFA_NO_STATISTICS__) … … 147 75 } 148 76 149 if(options.stats) {150 this.prnt = alloc();151 (*this.prnt){ this.self };152 } else {153 this.prnt = 0p;154 }155 156 77 #if !defined(__CFA_NO_STATISTICS__) 157 78 print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO ); … … 163 84 164 85 void ^?{}( ServerCluster & this ) { 165 delete(this.prnt);166 167 86 for(i; options.clopts.nprocs) { 168 87 ^(this.procs[i]){}; … … 175 94 extern void init_protocol(void); 176 95 extern void deinit_protocol(void); 96 97 //============================================================================================= 98 // REUSEPORT 99 //============================================================================================= 100 101 size_t sockarr_size; 102 struct __attribute__((aligned(128))) Q { 103 mpsc_queue(PendingRead) q; 104 }; 105 106 //============================================================================================= 107 // Termination 108 //============================================================================================= 109 110 int closefd; 111 void cleanstop(int) { 112 eventfd_t buffer = 1; 113 char * buffer_s = (char*)&buffer; 114 int ret = write(closefd, buffer_s, sizeof(buffer)); 115 if(ret < 0) abort( "eventfd write error: (%d) %s\n", (int)errno, strerror(errno) ); 116 return; 117 } 177 118 178 119 //============================================================================================= … … 180 121 //=============================================================================================' 181 122 int main( int argc, char * argv[] ) { 123 int ret; 182 124 __sighandler_t s = 1p; 183 125 signal(SIGPIPE, s); … … 186 128 // Parse args 187 129 parse_options(argc, argv); 130 131 //=================== 132 // Setup non-interactive termination 133 if(!options.interactive) { 134 closefd = eventfd(0, 0); 135 if(closefd < 0) abort( "eventfd error: (%d) %s\n", (int)errno, strerror(errno) ); 136 137 sighandler_t prev = signal(SIGTERM, cleanstop); 138 intptr_t prev_workaround = (intptr_t) prev; 139 // can't use SIG_ERR it crashes the compiler 140 if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) ); 141 142 sout | "Signal termination ready"; 143 } 188 144 189 145 //=================== … … 197 153 // Open Socket 198 154 sout | getpid() | ": Listening on port" | options.socket.port; 199 int server_fd = socket(AF_INET, SOCK_STREAM, 0); 200 if(server_fd < 0) { 201 abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) ); 202 } 203 204 int ret = 0; 155 205 156 struct sockaddr_in address; 206 int addrlen = sizeof(address); 207 memset( (char *)&address, '\0' ); 208 address.sin_family = AF_INET; 209 address.sin_addr.s_addr = htonl(INADDR_ANY); 210 address.sin_port = htons( options.socket.port ); 211 212 int waited = 0; 213 for() { 214 int sockfd = server_fd; 215 __CONST_SOCKADDR_ARG addr; 216 addr.__sockaddr__ = (struct sockaddr *)&address; 217 socklen_t addrlen = sizeof(address); 218 ret = bind( sockfd, addr, addrlen ); 219 if(ret < 0) { 220 if(errno == EADDRINUSE) { 221 if(waited == 0) { 222 if(!options.interactive) abort | "Port already in use in non-interactive mode. Aborting"; 223 sout | "Waiting for port"; 224 } else { 225 sout | "\r" | waited | nonl; 226 flush( sout ); 227 } 228 waited ++; 229 sleep( 1`s ); 230 continue; 231 } 232 abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) ); 233 } 234 break; 235 } 236 237 ret = listen( server_fd, options.socket.backlog ); 238 if(ret < 0) { 239 abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) ); 240 } 157 int addrlen = prepaddr(address); 158 159 int server_fd; 241 160 242 161 //=================== … … 257 176 258 177 { 178 // Stats printer makes a copy so this needs to persist longer than normal 179 connection ** conns; 180 AcceptWorker * aworkers = 0p; 181 ChannelWorker * cworkers = 0p; 182 Acceptor * acceptors = 0p; 183 Q * queues = 0p; 259 184 ServerCluster cl[options.clopts.nclusters]; 185 186 if(options.stats) { 187 stats_thrd = alloc(); 188 (*stats_thrd){ cl }; 189 } else { 190 stats_thrd = 0p; 191 } 260 192 261 193 init_protocol(); 262 194 { 263 Worker * workers = anew(options.clopts.nworkers); 264 cl[0].prnt->workers = workers; 265 cl[0].prnt->worker_cnt = options.clopts.nworkers; 266 for(i; options.clopts.nworkers) { 267 // if( options.file_cache.fixed_fds ) { 268 // workers[i].pipe[0] = pipe_off + (i * 2) + 0; 269 // workers[i].pipe[1] = pipe_off + (i * 2) + 1; 270 // } 271 // else 272 { 273 workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0]; 274 workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1]; 275 workers[i].sockfd = server_fd; 276 workers[i].addr = (struct sockaddr *)&address; 277 workers[i].addrlen = (socklen_t*)&addrlen; 278 workers[i].flags = 0; 279 } 280 unpark( workers[i] ); 195 int nacceptors = options.clopts.nprocs * options.clopts.nclusters; 196 conns = alloc(options.clopts.nworkers); 197 if(options.socket.reuseport) { 198 queues = alloc(nacceptors); 199 acceptors = alloc(nacceptors); 200 sout | "Creating" | nacceptors | "Acceptors"; 201 for(i; nacceptors) { 202 (acceptors[i]){ i % options.clopts.nclusters }; 203 } 204 for(i; nacceptors) { 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 % nacceptors].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 } 281 247 } 282 248 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters"; … … 285 251 } 286 252 sout | nl; 287 if(!options.interactive) park();288 253 { 289 char buffer[128]; 290 for() { 291 int ret = cfa_read(0, buffer, 128, 0); 292 if(ret == 0) break; 254 if(options.interactive) { 255 char buffer[128]; 256 for() { 257 int ret = cfa_read(0, buffer, 128, 0); 258 if(ret == 0) break; 259 if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) ); 260 sout | "User wrote '" | "" | nonl; 261 write(sout, buffer, ret - 1); 262 sout | "'"; 263 } 264 } 265 else { 266 char buffer[sizeof(eventfd_t)]; 267 int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0); 293 268 if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) ); 294 sout | "User wrote '" | "" | nonl;295 write(sout, buffer, ret - 1);296 sout | "'";297 269 } 298 270 … … 300 272 } 301 273 302 sout | "Notifying connections..." | nonl; flush( sout );303 for(i; options.clopts.nworkers) {304 workers[i].done = true;305 }306 sout | "done";307 308 sout | "Shutting down socket..." | nonl; flush( sout );309 int ret = shutdown( server_fd, SHUT_RD );310 if( ret < 0 ) {311 abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) );312 }313 sout | "done";314 315 274 //=================== 316 // Close Socket 317 sout | "Closing Socket..." | nonl; flush( sout ); 318 ret = close( server_fd ); 319 if(ret < 0) { 320 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); 321 } 322 sout | "done"; 323 324 sout | "Stopping connection threads..." | nonl; flush( sout ); 325 adelete(workers); 275 // Close Socket and join 276 if(options.socket.reuseport) { 277 sout | "Notifying connections..." | nonl; flush( sout ); 278 for(i; nacceptors) { 279 acceptors[i].done = true; 280 } 281 for(i; options.clopts.nworkers) { 282 cworkers[i].done = true; 283 } 284 sout | "done"; 285 286 sout | "Shutting down Socket..." | nonl; flush( sout ); 287 for(i; nacceptors) { 288 ret = shutdown( acceptors[i].sockfd, SHUT_RD ); 289 if( ret < 0 ) { 290 abort( "shutdown1 error: (%d) %s\n", (int)errno, strerror(errno) ); 291 } 292 } 293 sout | "done"; 294 295 sout | "Closing Socket..." | nonl; flush( sout ); 296 for(i; nacceptors) { 297 ret = close( acceptors[i].sockfd ); 298 if( ret < 0) { 299 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); 300 } 301 } 302 sout | "done"; 303 304 sout | "Stopping accept threads..." | nonl; flush( sout ); 305 for(i; nacceptors) { 306 join(acceptors[i]); 307 } 308 sout | "done"; 309 310 sout | "Draining worker queues..." | nonl; flush( sout ); 311 for(i; nacceptors) { 312 PendingRead * p = 0p; 313 while(p = pop(queues[i].q)) { 314 fulfil(p->f, -ECONNRESET); 315 } 316 } 317 sout | "done"; 318 319 sout | "Stopping worker threads..." | nonl; flush( sout ); 320 for(i; options.clopts.nworkers) { 321 for(j; 2) { 322 ret = close(cworkers[i].conn.pipe[j]); 323 if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); 324 } 325 join(cworkers[i]); 326 } 327 } 328 else { 329 sout | "Notifying connections..." | nonl; flush( sout ); 330 for(i; options.clopts.nworkers) { 331 aworkers[i].done = true; 332 } 333 sout | "done"; 334 335 sout | "Shutting down Socket..." | nonl; flush( sout ); 336 ret = shutdown( server_fd, SHUT_RD ); 337 if( ret < 0 ) { 338 abort( "shutdown2 error: (%d) %s\n", (int)errno, strerror(errno) ); 339 } 340 sout | "done"; 341 342 sout | "Closing Socket..." | nonl; flush( sout ); 343 ret = close( server_fd ); 344 if(ret < 0) { 345 abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) ); 346 } 347 sout | "done"; 348 349 sout | "Stopping connection threads..." | nonl; flush( sout ); 350 for(i; options.clopts.nworkers) { 351 for(j; 2) { 352 ret = close(aworkers[i].conn.pipe[j]); 353 if(ret < 0) abort( "close pipe %d error: (%d) %s\n", j, (int)errno, strerror(errno) ); 354 } 355 join(aworkers[i]); 356 } 357 } 326 358 } 327 359 sout | "done"; … … 331 363 sout | "done"; 332 364 365 sout | "Stopping printer threads..." | nonl; flush( sout ); 366 if(stats_thrd) { 367 notify_one(stats_thrd->var); 368 } 369 delete(stats_thrd); 370 sout | "done"; 371 372 // Now that the stats printer is stopped, we can reclaim this 373 adelete(aworkers); 374 adelete(cworkers); 375 adelete(acceptors); 376 adelete(queues); 377 free(conns); 378 333 379 sout | "Stopping processors/clusters..." | nonl; flush( sout ); 334 380 } 335 381 sout | "done"; 336 382 337 sout | "Closing splice fds..." | nonl; flush( sout );338 for(i; pipe_cnt) {339 ret = close( fds[pipe_off + i] );340 if(ret < 0) {341 abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );342 }343 }344 383 free(fds); 345 sout | "done";346 384 347 385 sout | "Stopping processors..." | nonl; flush( sout );
Note:
See TracChangeset
for help on using the changeset viewer.
