Changes in benchmark/io/http/worker.cfa [07997cd:0f1336c]
- File:
-
- 1 edited
-
benchmark/io/http/worker.cfa (modified) (17 diffs)
Legend:
- Unmodified
- Added
- Removed
-
benchmark/io/http/worker.cfa
r07997cd r0f1336c 14 14 #include "filecache.hfa" 15 15 16 static const unsigned long long period = 5_000_000; 17 16 18 //============================================================================================= 17 19 // Generic connection handling … … 31 33 32 34 // if we are done, break out of the loop 33 if( closed ) break REQUEST; 35 if( closed ) { 36 if( code != OK200 ) this.stats.sendfile.error++; 37 break REQUEST; 38 } 34 39 35 40 // If this wasn't a request retrun 400 36 41 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 ) mutex(sout) sout | "=== Request for /plaintext ===";44 45 int ret = answer_plaintext(fd);46 if( ret == -ECONNRESET ) break REQUEST;47 48 if( options.log ) mutex(sout) sout | "=== Answer sent ===";49 continue REQUEST;50 }51 52 if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {53 if( options.log ) mutex(sout) sout | "=== Request for /ping ===";54 55 // Send the header56 int ret = answer_empty(fd);57 if( ret == -ECONNRESET ) break REQUEST;58 59 if( options.log ) mutex(sout) sout | "=== Answer sent ===";60 continue REQUEST;61 }42 abort | "=== Invalid Request :" | code_val(code) | "==="; 43 // answer_error(fd, code); 44 // continue REQUEST; 45 } 46 47 // if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) { 48 // if( options.log ) mutex(sout) sout | "=== Request for /plaintext ==="; 49 50 // int ret = answer_plaintext(fd); 51 // if( ret == -ECONNRESET ) { this.stats.sendfile.error++; break REQUEST; } 52 53 // if( options.log ) mutex(sout) sout | "=== Answer sent ==="; 54 // continue REQUEST; 55 // } 56 57 // if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) { 58 // if( options.log ) mutex(sout) sout | "=== Request for /ping ==="; 59 60 // // Send the header 61 // int ret = answer_empty(fd); 62 // if( ret == -ECONNRESET ) { this.stats.sendfile.error++; break REQUEST; } 63 64 // if( options.log ) mutex(sout) sout | "=== Answer sent ==="; 65 // continue REQUEST; 66 // } 62 67 63 68 if( options.log ) { … … 68 73 69 74 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; 75 // if( options.log ) { 76 serr | "=== File Not Found (" | nonl; 77 write(serr, file, name_size); 78 serr | ") ==="; 79 abort(); 80 // } 81 // answer_error(fd, E405); 82 // continue REQUEST; 77 83 } 78 84 … … 84 90 // If we can't find the file, return 404 85 91 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; 92 // if( options.log ) { 93 serr | "=== File Not Found 2 (" | nonl; 94 write(serr, file, name_size); 95 serr | ") ==="; 96 abort(); 97 // } 98 // answer_error(fd, E404); 99 // continue REQUEST; 93 100 } 94 101 95 102 // Send the desired file 96 103 int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile ); 97 if( ret == -ECONNRESET ) break REQUEST; 104 if(ret < 0) { 105 if( ret == -ECONNABORTED ) { this.stats.sendfile.error++; break REQUEST; } 106 if( ret == -ECONNRESET ) { this.stats.sendfile.error++; break REQUEST; } 107 if( ret == -EPIPE ) { this.stats.sendfile.error++; break REQUEST; } 108 if( ret == -EBADF ) { this.stats.sendfile.error++; break REQUEST; } 109 abort( "answer sendfile error: %d (%d) %s\n", ret, (int)errno, strerror(errno) ); 110 } 98 111 99 112 if( options.log ) mutex(sout) sout | "=== Answer sent ==="; 100 113 } 101 114 115 this.stats.sendfile.close++; 116 102 117 if (stats_thrd) { 103 unsigned long long next = rdtscl();104 if(next > (last + 500000000)) {118 // unsigned long long next = rdtscl(); 119 // if(next > (last + period)) { 105 120 if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) { 106 121 push(this.stats.sendfile, stats_thrd->stats.send); 107 122 unlock(stats_thrd->stats.lock); 108 last = next;109 } 110 }123 // last = next; 124 } 125 // } 111 126 } 112 127 } … … 130 145 if( options.log ) mutex(sout) sout | "=== Accepting connection ==="; 131 146 int fd = cfa_accept4( this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY ); 132 if(fd < 0) {147 if(fd <= 0) { 133 148 if( errno == ECONNABORTED ) break; 134 149 if( this.done && (errno == EINVAL || errno == EBADF) ) break; 135 abort( "accept error : (%d) %s\n", (int)errno, strerror(errno) );150 abort( "accept error %d: (%d) %s\n", fd, (int)errno, strerror(errno) ); 136 151 } 137 152 if(this.done) break; 153 154 this.stats.accepts++; 155 if (stats_thrd && try_lock(stats_thrd->stats.lock)) { 156 push(this.stats, stats_thrd->stats.accpt); 157 unlock(stats_thrd->stats.lock); 158 } 138 159 139 160 if( options.log ) mutex(sout) sout | "=== New connection" | fd | "" | ", waiting for requests ==="; … … 141 162 char buffer[len]; 142 163 handle_connection( this.conn, fd, buffer, len, 0p, last ); 164 this.conn.stats.sendfile.maxfd = max(this.conn.stats.sendfile.maxfd, fd); 143 165 144 166 if( options.log ) mutex(sout) sout | "=== Connection closed ==="; … … 162 184 /* paranoid */ assert( this.conn.pipe[0] != -1 ); 163 185 /* paranoid */ assert( this.conn.pipe[1] != -1 ); 186 this.conn.stats.sendfile.maxfd = max(this.conn.pipe[0], this.conn.pipe[1]); 187 // this.conn.stats.sendfile.maxfd = 0; 164 188 for() { 165 189 size_t len = options.socket.buflen; … … 173 197 if( options.log ) mutex(sout) sout | "=== Waiting new connection ==="; 174 198 handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last ); 199 if(this.done) break; 200 this.conn.stats.sendfile.maxfd = max(this.conn.stats.sendfile.maxfd, p.out.fd); 201 this.conn.stats.sendfile.close++; 175 202 176 203 if( options.log ) mutex(sout) sout | "=== Connection closed ==="; 177 if(this.done) break; 178 } 204 } 205 206 lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2); 207 push(this.conn.stats.sendfile, stats_thrd->stats.send); 208 unlock(stats_thrd->stats.lock); 179 209 } 180 210 … … 198 228 199 229 static inline void push_connection( Acceptor & this, int fd ) { 230 this.stats.accepts++; 200 231 PendingRead * p = 0p; 201 232 for() { … … 203 234 p = pop(*this.queue); 204 235 if(p) break; 236 // abort( "Too few threads" ); 205 237 yield(); 206 238 this.stats.creates++; … … 212 244 213 245 // #define ACCEPT_SPIN 214 #define ACCEPT_MANY 246 #define ACCEPT_ONE 247 // #define ACCEPT_MANY 215 248 216 249 void main( Acceptor & this ) { … … 232 265 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) ); 233 266 } 234 this.stats.accepts++;235 267 236 268 if(this.done) return; … … 242 274 if (stats_thrd) { 243 275 unsigned long long next = rdtscl(); 244 if(next > (last + 500000000)) {276 if(next > (last + period)) { 245 277 if(try_lock(stats_thrd->stats.lock)) { 246 278 push(this.stats, stats_thrd->stats.accpt); … … 249 281 } 250 282 } 283 } 284 285 if( options.log ) sout | "=== Accepting connection ==="; 286 } 287 288 #elif defined(ACCEPT_ONE) 289 if( options.log ) sout | "=== Accepting connection ==="; 290 for() { 291 int fd = cfa_accept4(this.sockfd, this.[addr, addrlen, flags], 0); 292 if(fd < 0) { 293 if( errno == ECONNABORTED ) break; 294 if( this.done && (errno == EINVAL || errno == EBADF) ) break; 295 abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) ); 296 } 297 298 if(this.done) return; 299 300 if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ==="; 301 302 if(fd) push_connection(this, fd); 303 304 if (stats_thrd) { 305 // unsigned long long next = rdtscl(); 306 // if(next > (last + period)) { 307 if(try_lock(stats_thrd->stats.lock)) { 308 push(this.stats, stats_thrd->stats.accpt); 309 unlock(stats_thrd->stats.lock); 310 // last = next; 311 } 312 // } 251 313 } 252 314 … … 269 331 if (stats_thrd) { 270 332 unsigned long long next = rdtscl(); 271 if(next > (last + 500000000)) {333 if(next > (last + period)) { 272 334 if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) { 273 335 push(this.stats, stats_thrd->stats.accpt); … … 284 346 int fd = get_res(res); 285 347 reset(res); 286 this.stats.accepts++;287 348 if(fd < 0) { 288 349 if( errno == ECONNABORTED ) continue; … … 319 380 #error no accept algorithm specified 320 381 #endif 321 } 382 lock(stats_thrd->stats.lock); 383 push(this.stats, stats_thrd->stats.accpt); 384 unlock(stats_thrd->stats.lock); 385 }
Note:
See TracChangeset
for help on using the changeset viewer.
