Changes in benchmark/io/http/main.cfa [329e26a:ef3c383]
- File:
-
- 1 edited
-
benchmark/io/http/main.cfa (modified) (16 diffs)
Legend:
- Unmodified
- Added
- Removed
-
benchmark/io/http/main.cfa
r329e26a ref3c383 2 2 3 3 #include <errno.h> 4 #include <signal.h>5 4 #include <stdio.h> 6 5 #include <string.h> … … 9 8 #include <sched.h> 10 9 #include <signal.h> 11 #include <sys/eventfd.h>12 10 #include <sys/socket.h> 13 11 #include <netinet/in.h> … … 16 14 #include <fstream.hfa> 17 15 #include <kernel.hfa> 18 #include <locks.hfa>19 16 #include <iofwd.hfa> 20 17 #include <stats.hfa> … … 24 21 #include "filecache.hfa" 25 22 #include "options.hfa" 26 #include "socket.hfa"27 #include "printer.hfa"28 23 #include "worker.hfa" 29 24 … … 32 27 Duration default_preemption() { 33 28 return 0; 29 } 30 31 //============================================================================================= 32 // Stats Printer 33 //=============================================================================================' 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 } 34 98 } 35 99 … … 37 101 // Globals 38 102 //============================================================================================= 103 struct ServerCluster { 104 cluster self; 105 processor * procs; 106 // io_context * ctxs; 107 StatsPrinter * prnt; 108 109 }; 110 39 111 void ?{}( ServerCluster & this ) { 40 112 (this.self){ "Server Cluster", options.clopts.params }; … … 50 122 (this.procs[i]){ "Benchmark Processor", this.self }; 51 123 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 );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 ); 66 138 67 139 #if !defined(__CFA_NO_STATISTICS__) … … 75 147 } 76 148 149 if(options.stats) { 150 this.prnt = alloc(); 151 (*this.prnt){ this.self }; 152 } else { 153 this.prnt = 0p; 154 } 155 77 156 #if !defined(__CFA_NO_STATISTICS__) 78 157 print_stats_at_exit( this.self, CFA_STATS_READY_Q | CFA_STATS_IO ); … … 84 163 85 164 void ^?{}( ServerCluster & this ) { 165 delete(this.prnt); 166 86 167 for(i; options.clopts.nprocs) { 87 168 ^(this.procs[i]){}; … … 94 175 extern void init_protocol(void); 95 176 extern void deinit_protocol(void); 96 97 //=============================================================================================98 // REUSEPORT99 //=============================================================================================100 101 size_t sockarr_size;102 struct __attribute__((aligned(128))) Q {103 mpsc_queue(PendingRead) q;104 };105 106 //=============================================================================================107 // Termination108 //=============================================================================================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 }118 177 119 178 //============================================================================================= … … 121 180 //=============================================================================================' 122 181 int main( int argc, char * argv[] ) { 123 int ret;124 182 __sighandler_t s = 1p; 125 183 signal(SIGPIPE, s); … … 128 186 // Parse args 129 187 parse_options(argc, argv); 130 131 //===================132 // Setup non-interactive termination133 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 compiler140 if(prev_workaround == -1) abort( "signal setup error: (%d) %s\n", (int)errno, strerror(errno) );141 142 sout | "Signal termination ready";143 }144 188 145 189 //=================== … … 153 197 // Open Socket 154 198 sout | getpid() | ": Listening on port" | options.socket.port; 155 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; 156 205 struct sockaddr_in address; 157 int addrlen = prepaddr(address); 158 159 int server_fd; 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 } 160 241 161 242 //=================== … … 176 257 177 258 { 178 // Stats printer makes a copy so this needs to persist longer than normal179 connection ** conns;180 AcceptWorker * aworkers = 0p;181 ChannelWorker * cworkers = 0p;182 Acceptor * acceptors = 0p;183 Q * queues = 0p;184 259 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 }192 260 193 261 init_protocol(); 194 262 { 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 }; 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; 203 279 } 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 } 280 unpark( workers[i] ); 247 281 } 248 282 sout | options.clopts.nworkers | "workers started on" | options.clopts.nprocs | "processors /" | options.clopts.nclusters | "clusters"; … … 251 285 } 252 286 sout | nl; 287 if(!options.interactive) park(); 253 288 { 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 } 289 char buffer[128]; 290 for() { 291 int ret = cfa_read(0, buffer, 128, 0); 292 if(ret == 0) break; 293 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 | "'"; 264 297 } 265 else {266 char buffer[sizeof(eventfd_t)];267 int ret = cfa_read(closefd, buffer, sizeof(eventfd_t), 0);268 if(ret < 0) abort( "main read error: (%d) %s\n", (int)errno, strerror(errno) );269 }270 298 271 299 sout | "Shutdown received"; 272 300 } 273 301 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 274 315 //=================== 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 } 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); 358 326 } 359 327 sout | "done"; … … 363 331 sout | "done"; 364 332 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 this373 adelete(aworkers);374 adelete(cworkers);375 adelete(acceptors);376 adelete(queues);377 free(conns);378 379 333 sout | "Stopping processors/clusters..." | nonl; flush( sout ); 380 334 } 381 335 sout | "done"; 382 336 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 } 383 344 free(fds); 345 sout | "done"; 384 346 385 347 sout | "Stopping processors..." | nonl; flush( sout );
Note:
See TracChangeset
for help on using the changeset viewer.
