Ignore:
File:
1 edited

Legend:

Unmodified
Added
Removed

  • benchmark/io/http/protocol.cfa

    rbf7c7ea r32d1383  
    1818extern "C" {
    1919      int snprintf ( char * s, size_t n, const char * format, ... );
     20        ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);
    2021        // #include <linux/io_uring.h>
    2122}
    … …  
    2930#define PLAINTEXT_MEMCPY
    3031#define PLAINTEXT_NOCOPY
    31 #define LINKED_IO
     32// #define LINKED_IO
     33#define TRUE_SENDFILE
    3234
    3335static inline __s32 wait_res( io_future_t & this ) {
    … …  
    7173                int ret = cfa_send(fd, it, len, 0, CFA_IO_LAZY);
    7274                if( ret < 0 ) {
    73                         if( errno == ECONNRESET || errno == EPIPE ) { close(fd); return -ECONNRESET; }
    74                         if( errno == EAGAIN || errno == EWOULDBLOCK) return -EAGAIN;
     75                        if( errno == ECONNRESET || errno == EPIPE || errno == EBADF ) {
     76                                ret = close(fd);
     77                                if( ret != 0 ) abort( "close in 'answer' error: (%d) %s\n", (int)errno, strerror(errno) );
     78                                return -ECONNRESET;
     79                        }
    7580
    7681                        abort( "'answer error' error: (%d) %s\n", (int)errno, strerror(errno) );
    … …  
    8489}
    8590
    86 int answer_error( int fd, HttpCode code ) {
    87         /* paranoid */ assert( code < KNOWN_CODES && code != OK200 );
    88         int idx = (int)code;
    89         return answer( fd, http_msgs[idx]->msg, http_msgs[idx]->len );
    90 }
     91// int answer_error( int fd, HttpCode code ) {
     92//      /* paranoid */ assert( code < KNOWN_CODES && code != OK200 );
     93//      int idx = (int)code;
     94//      return answer( fd, http_msgs[idx]->msg, http_msgs[idx]->len );
     95// }
    9196
    9297static int fill_header(char * it, size_t size) {
    … …  
    104109}
    105110
    106 #if defined(PLAINTEXT_NOCOPY)
    107 int answer_plaintext( int fd ) {
    108         return answer(fd, http_msgs[OK200_PlainText]->msg, http_msgs[OK200_PlainText]->len); // +1 cause snprintf doesn't count nullterminator
    109 }
    110 #elif defined(PLAINTEXT_MEMCPY)
    111 #define TEXTSIZE 15
    112 int answer_plaintext( int fd ) {
    113         char text[] = "Hello, World!\n\n";
    114         char ts[] = xstr(TEXTSIZE) " \n\n";
    115         _Static_assert(sizeof(text) - 1 == TEXTSIZE);
    116         char buffer[512 + TEXTSIZE];
    117         char * it = buffer;
    118         memcpy(it, http_msgs[OK200]->msg, http_msgs[OK200]->len);
    119         it += http_msgs[OK200]->len;
    120         int len = http_msgs[OK200]->len;
    121         memcpy(it, ts, sizeof(ts) - 1);
    122         it += sizeof(ts) - 1;
    123         len += sizeof(ts) - 1;
    124         memcpy(it, text, TEXTSIZE);
    125         return answer(fd, buffer, len + TEXTSIZE);
    126 }
    127 #elif defined(PLAINTEXT_1WRITE)
    128 int answer_plaintext( int fd ) {
    129         char text[] = "Hello, World!\n\n";
    130         char buffer[512 + sizeof(text)];
    131         char * it = buffer;
    132         memcpy(it, http_msgs[OK200]->msg, http_msgs[OK200]->len);
    133         it += http_msgs[OK200]->len;
    134         int len = http_msgs[OK200]->len;
    135         int r = snprintf(it, 512 - len, "%d \n\n", sizeof(text));
    136         it += r;
    137         len += r;
    138         memcpy(it, text, sizeof(text));
    139         return answer(fd, buffer, len + sizeof(text));
    140 }
    141 #else
    142 int answer_plaintext( int fd ) {
    143         char text[] = "Hello, World!\n\n";
    144         int ret = answer_header(fd, sizeof(text));
    145         if( ret < 0 ) return ret;
    146         return answer(fd, text, sizeof(text));
    147 }
    148 #endif
    149 
    150 int answer_empty( int fd ) {
    151         return answer_header(fd, 0);
    152 }
    153 
    154 static int sendfile( int pipe[2], int fd, int ans_fd, size_t count ) {
    155         unsigned sflags = SPLICE_F_MOVE; // | SPLICE_F_MORE;
    156         off_t offset = 0;
    157         ssize_t ret;
    158         SPLICE1: while(count > 0) {
    159                 ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, sflags, CFA_IO_LAZY);
    160                 if( ret < 0 ) {
    161                         if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE1;
    162                         if( errno == ECONNRESET ) return -ECONNRESET;
    163                         if( errno == EPIPE ) return -EPIPE;
    164                         abort( "splice [0] error: (%d) %s\n", (int)errno, strerror(errno) );
    165                 }
    166 
    167                 count -= ret;
    168                 size_t in_pipe = ret;
    169                 SPLICE2: while(in_pipe > 0) {
    170                         ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, sflags, CFA_IO_LAZY);
    171                         if( ret < 0 ) {
    172                                 if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE2;
    173                                 if( errno == ECONNRESET ) return -ECONNRESET;
    174                                 if( errno == EPIPE ) return -EPIPE;
    175                                 abort( "splice [1] error: (%d) %s\n", (int)errno, strerror(errno) );
     111// #if defined(PLAINTEXT_NOCOPY)
     112// int answer_plaintext( int fd ) {
     113//      return answer(fd, http_msgs[OK200_PlainText]->msg, http_msgs[OK200_PlainText]->len); // +1 cause snprintf doesn't count nullterminator
     114// }
     115// #elif defined(PLAINTEXT_MEMCPY)
     116// #define TEXTSIZE 15
     117// int answer_plaintext( int fd ) {
     118//      char text[] = "Hello, World!\n\n";
     119//      char ts[] = xstr(TEXTSIZE) " \n\n";
     120//      _Static_assert(sizeof(text) - 1 == TEXTSIZE);
     121//      char buffer[512 + TEXTSIZE];
     122//      char * it = buffer;
     123//      memcpy(it, http_msgs[OK200]->msg, http_msgs[OK200]->len);
     124//      it += http_msgs[OK200]->len;
     125//      int len = http_msgs[OK200]->len;
     126//      memcpy(it, ts, sizeof(ts) - 1);
     127//      it += sizeof(ts) - 1;
     128//      len += sizeof(ts) - 1;
     129//      memcpy(it, text, TEXTSIZE);
     130//      return answer(fd, buffer, len + TEXTSIZE);
     131// }
     132// #elif defined(PLAINTEXT_1WRITE)
     133// int answer_plaintext( int fd ) {
     134//      char text[] = "Hello, World!\n\n";
     135//      char buffer[512 + sizeof(text)];
     136//      char * it = buffer;
     137//      memcpy(it, http_msgs[OK200]->msg, http_msgs[OK200]->len);
     138//      it += http_msgs[OK200]->len;
     139//      int len = http_msgs[OK200]->len;
     140//      int r = snprintf(it, 512 - len, "%d \n\n", sizeof(text));
     141//      it += r;
     142//      len += r;
     143//      memcpy(it, text, sizeof(text));
     144//      return answer(fd, buffer, len + sizeof(text));
     145// }
     146// #else
     147// int answer_plaintext( int fd ) {
     148//      char text[] = "Hello, World!\n\n";
     149//      int ret = answer_header(fd, sizeof(text));
     150//      if( ret < 0 ) return ret;
     151//      return answer(fd, text, sizeof(text));
     152// }
     153// #endif
     154
     155// int answer_empty( int fd ) {
     156//      return answer_header(fd, 0);
     157// }
     158
     159static int sendfile( int pipe[2], int fd, int ans_fd, size_t count, sendfile_stats_t & stats ) {
     160        int zipf_idx = -1;
     161        STATS: for(i; zipf_cnts) {
     162                if(count <= zipf_sizes[i]) {
     163                        zipf_idx = i;
     164                        break STATS;
     165                }
     166        }
     167        if(zipf_idx < 0) mutex(serr) serr | "SENDFILE" | count | " greated than biggest zipf file";
     168
     169        #if defined(TRUE_SENDFILE)
     170                off_t offset = 0;
     171                ssize_t ret;
     172                int flags = fcntl(fd, F_GETFL);
     173                if(flags < 0) abort("getfl in 'true sendfile' error: (%d) %s\n", (int)errno, strerror(errno) );
     174                ret = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
     175                if(ret < 0) abort("setfl in 'true sendfile' error: (%d) %s\n", (int)errno, strerror(errno) );
     176
     177                while(count) {
     178                        ret = sendfile(fd, ans_fd, &offset, count);
     179                        if( ret <= 0 ) {
     180                                if( errno == EAGAIN || errno == EWOULDBLOCK ) {
     181                                        stats.eagain++;
     182                                        yield();
     183                                        continue;
     184                                }
     185                                if( errno == ECONNRESET || errno == EPIPE ) {
     186                                        ret = close(fd);
     187                                        if( ret != 0 ) abort( "close in 'true sendfile' error: (%d) %s\n", (int)errno, strerror(errno) );
     188                                        return -ECONNRESET;
     189                                }
     190                                abort( "sendfile error: %d (%d) %s\n", ret, (int)errno, strerror(errno) );
    176191                        }
    177                         in_pipe -= ret;
    178                 }
    179 
    180         }
     192                        count -= ret;
     193                        stats.splcin++;
     194                        if(count > 0) stats.avgrd[zipf_idx].calls++;
     195                        stats.avgrd[zipf_idx].bytes += ret;
     196                }
     197
     198                ret = fcntl(fd, F_SETFL, flags & ~O_NONBLOCK);
     199                if(ret < 0) abort("resetfl in 'true sendfile' error: (%d) %s\n", (int)errno, strerror(errno) );
     200        #else
     201                #error not implemented
     202                // unsigned sflags = SPLICE_F_MOVE; // | SPLICE_F_MORE;
     203                // off_t offset = 0;
     204                // ssize_t ret;
     205                // SPLICE1: while(count > 0) {
     206                //      stats.tries++;
     207                //      // ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, sflags, CFA_IO_LAZY);
     208                //      ret = splice(ans_fd, &offset, pipe[1], 0p, count, sflags);
     209                //      if( ret <= 0 ) {
     210                //              if( errno == ECONNRESET || errno == EPIPE ) {
     211                //                      ret = close(fd);
     212                //                      if( ret != 0 ) abort( "close in 'sendfile splice in' error: (%d) %s\n", (int)errno, strerror(errno) );
     213                //                      return -ECONNRESET;
     214                //              }
     215                //              abort( "splice [0] error: %d (%d) %s\n", ret, (int)errno, strerror(errno) );
     216                //      }
     217                //      count -= ret;
     218                //      stats.splcin++;
     219                //      if(count > 0) stats.avgrd[zipf_idx].calls++;
     220                //      stats.avgrd[zipf_idx].bytes += ret;
     221
     222                //      size_t in_pipe = ret;
     223                //      SPLICE2: while(in_pipe > 0) {
     224                //              ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, sflags, CFA_IO_LAZY);
     225                //              // ret = splice(pipe[0], 0p, fd, 0p, in_pipe, sflags);
     226                //              if( ret <= 0 ) {
     227                //                      if( errno == ECONNRESET || errno == EPIPE ) {
     228                //                              ret = close(fd);
     229                //                              if( ret != 0 ) abort( "close in 'sendfile splice out' error: (%d) %s\n", (int)errno, strerror(errno) );
     230                //                              return -ECONNRESET;
     231                //                      }
     232                //                      abort( "splice [1] error: %d (%d) %s\n", ret, (int)errno, strerror(errno) );
     233                //              }
     234                //              stats.splcot++;
     235                //              in_pipe -= ret;
     236                //      }
     237
     238                // }
     239        #endif
     240
    181241        return count;
    182242}
    183243
    184 enum FSM_STATE {
    185         Initial,
    186         Retry,
    187         Error,
    188         Done,
    189 };
    190 
    191 struct FSM_Result {
    192         FSM_STATE state;
    193         int error;
    194 };
    195 
    196 static inline void ?{}(FSM_Result & this) { this.state = Initial; this.error = 0; }
    197 static inline bool is_error(FSM_Result & this) { return Error == this.state; }
    198 static inline bool is_done(FSM_Result & this) { return Done == this.state; }
    199 
    200 static inline int error(FSM_Result & this, int error) {
    201         this.error = error;
    202         this.state = Error;
    203         return error;
    204 }
    205 
    206 static inline int done(FSM_Result & this) {
    207         this.state = Done;
    208         return 0;
    209 }
    210 
    211 static inline int retry(FSM_Result & this) {
    212         this.state = Retry;
    213         return 0;
    214 }
    215 
    216 static inline int need(FSM_Result & this) {
    217         switch(this.state) {
    218                 case Initial:
    219                 case Retry:
    220                         return 1;
    221                 case Error:
    222                         if(this.error == 0) mutex(serr) serr | "State marked error but code is 0";
    223                 case Done:
    224                         return 0;
    225         }
    226 }
    227 
    228 // Generator that handles sending the header
    229 generator header_g {
    230         io_future_t f;
    231         const char * next;
    232         int fd; size_t len;
    233         FSM_Result res;
    234 };
    235 
    236 static inline void ?{}(header_g & this, int fd, const char * it, size_t len ) {
    237         this.next = it;
    238         this.fd = fd;
    239         this.len = len;
    240 }
    241 
    242 static inline void fill(header_g & this, struct io_uring_sqe * sqe) {
    243         zero_sqe(sqe);
    244         sqe->opcode = IORING_OP_SEND;
    245         sqe->user_data = (uintptr_t)&this.f;
    246         sqe->flags = IOSQE_IO_LINK;
    247         sqe->fd = this.fd;
    248         sqe->addr = (uintptr_t)this.next;
    249         sqe->len = this.len;
    250 }
    251 
    252 static inline int error(header_g & this, int error) {
    253         int ret = close(this.fd);
    254         if( ret != 0 ) {
    255                 mutex(serr) serr | "Failed to close fd" | errno;
    256         }
    257         return error(this.res, error);
    258 }
    259 
    260 static inline int wait_and_process(header_g & this, sendfile_stats_t & stats) {
    261         wait(this.f);
    262 
    263         // Did something crazy happen?
    264         if(this.f.result > this.len) {
    265                 mutex(serr) serr | "HEADER sent too much!";
    266                 return error(this, -ERANGE);
    267         }
    268 
    269         // Something failed?
    270         if(this.f.result < 0) {
    271                 int error = -this.f.result;
    272                 if( error == ECONNRESET ) return error(this, -ECONNRESET);
    273                 if( error == EPIPE ) return error(this, -EPIPE);
    274                 if( error == ECANCELED ) {
    275                         mutex(serr) serr | "HEADER was cancelled, WTF!";
    276                         return error(this, -ECONNRESET);
    277                 }
    278                 if( error == EAGAIN || error == EWOULDBLOCK) {
    279                         mutex(serr) serr | "HEADER got eagain, WTF!";
    280                         return error(this, -ECONNRESET);
    281                 }
    282         }
    283 
    284         // Done?
    285         if(this.f.result == this.len) {
    286                 return done(this.res);
    287         }
    288 
    289         stats.header++;
    290 
    291         // It must be a Short read
    292         this.len  -= this.f.result;
    293         this.next += this.f.result;
    294         reset(this.f);
    295         return retry(this.res);
    296 }
    297 
    298 // Generator that handles splicing in a file
    299 struct splice_in_t {
    300         io_future_t f;
    301         int fd; int pipe; size_t len; off_t off;
    302         short zipf_idx;
    303         FSM_Result res;
    304 };
    305 
    306 static inline void ?{}(splice_in_t & this, int fd, int pipe, size_t len) {
    307         this.fd = fd;
    308         this.pipe = pipe;
    309         this.len = len;
    310         this.off = 0;
    311         this.zipf_idx = -1;
    312         STATS: for(i; zipf_cnts) {
    313                 if(len <= zipf_sizes[i]) {
    314                         this.zipf_idx = i;
    315                         break STATS;
    316                 }
    317         }
    318         if(this.zipf_idx < 0) mutex(serr) serr | "SPLICE IN" | len | " greated than biggest zipf file";
    319 }
    320 
    321 static inline void fill(splice_in_t & this, struct io_uring_sqe * sqe) {
    322         zero_sqe(sqe);
    323         sqe->opcode = IORING_OP_SPLICE;
    324         sqe->user_data = (uintptr_t)&this.f;
    325         sqe->flags = 0;
    326         sqe->splice_fd_in = this.fd;
    327         sqe->splice_off_in = this.off;
    328         sqe->fd = this.pipe;
    329         sqe->off = (__u64)-1;
    330         sqe->len = this.len;
    331         sqe->splice_flags = SPLICE_F_MOVE;
    332 }
    333 
    334 static inline int wait_and_process(splice_in_t & this, sendfile_stats_t & stats ) {
    335         wait(this.f);
    336 
    337         // Something failed?
    338         if(this.f.result < 0) {
    339                 int error = -this.f.result;
    340                 if( error == ECONNRESET ) return error(this.res, -ECONNRESET);
    341                 if( error == EPIPE ) return error(this.res, -EPIPE);
    342                 if( error == ECANCELED ) {
    343                         mutex(serr) serr | "SPLICE IN was cancelled, WTF!";
    344                         return error(this.res, -ECONNRESET);
    345                 }
    346                 if( error == EAGAIN || error == EWOULDBLOCK) {
    347                         mutex(serr) serr | "SPLICE IN got eagain, WTF!";
    348                         return error(this.res, -ECONNRESET);
    349                 }
    350                 mutex(serr) serr | "SPLICE IN got" | error | ", WTF!";
    351                 return error(this.res, -ECONNRESET);
    352         }
    353 
    354         // Did something crazy happen?
    355         if(this.f.result > this.len) {
    356                 mutex(serr) serr | "SPLICE IN spliced too much!";
    357                 return error(this.res, -ERANGE);
    358         }
    359 
    360         // Done?
    361         if(this.f.result == this.len) {
    362                 return done(this.res);
    363         }
    364 
    365         stats.splcin++;
    366         stats.avgrd[this.zipf_idx].calls++;
    367         stats.avgrd[this.zipf_idx].bytes += this.f.result;
    368 
    369         // It must be a Short read
    370         this.len -= this.f.result;
    371         this.off += this.f.result;
    372         reset(this.f);
    373         return retry(this.res);
    374 }
    375 
    376 generator splice_out_g {
    377         io_future_t f;
    378         int pipe; int fd; size_t len;
    379         FSM_Result res;
    380 };
    381 
    382 static inline void ?{}(splice_out_g & this, int pipe, int fd, size_t len) {
    383         this.pipe = pipe;
    384         this.fd = fd;
    385         this.len = len;
    386 }
    387 
    388 static inline void fill(splice_out_g & this, struct io_uring_sqe * sqe) {
    389         zero_sqe(sqe);
    390         sqe->opcode = IORING_OP_SPLICE;
    391         sqe->user_data = (uintptr_t)&this.f;
    392         sqe->flags = 0;
    393         sqe->splice_fd_in = this.pipe;
    394         sqe->splice_off_in = (__u64)-1;
    395         sqe->fd = this.fd;
    396         sqe->off = (__u64)-1;
    397         sqe->len = this.len;
    398         sqe->splice_flags = SPLICE_F_MOVE;
    399 }
    400 
    401 static inline int error(splice_out_g & this, int error) {
    402         int ret = close(this.fd);
    403         if( ret != 0 ) {
    404                 mutex(serr) serr | "Failed to close fd" | errno;
    405         }
    406         return error(this.res, error);
    407 }
    408 
    409 static inline void wait_and_process(splice_out_g & this, sendfile_stats_t & stats ) {
    410         wait(this.f);
    411 
    412         // Something failed?
    413         if(this.f.result < 0) {
    414                 int error = -this.f.result;
    415                 if( error == ECONNRESET ) return error(this, -ECONNRESET);
    416                 if( error == EPIPE ) return error(this, -EPIPE);
    417                 if( error == ECANCELED ) {
    418                         this.f.result = 0;
    419                         goto SHORT_WRITE;
    420                 }
    421                 if( error == EAGAIN || error == EWOULDBLOCK) {
    422                         mutex(serr) serr | "SPLICE OUT got eagain, WTF!";
    423                         return error(this, -ECONNRESET);
    424                 }
    425                 mutex(serr) serr | "SPLICE OUT got" | error | ", WTF!";
    426                 return error(this, -ECONNRESET);
    427         }
    428 
    429         // Did something crazy happen?
    430         if(this.f.result > this.len) {
    431                 mutex(serr) serr | "SPLICE OUT spliced too much!" | this.f.result | ">" | this.len;
    432                 return error(this.res, -ERANGE);
    433         }
    434 
    435         // Done?
    436         if(this.f.result == this.len) {
    437                 return done(this.res);
    438         }
    439 
    440 SHORT_WRITE:
    441         stats.splcot++;
    442 
    443         // It must be a Short Write
    444         this.len -= this.f.result;
    445         reset(this.f);
    446         return retry(this.res);
    447 }
     244// enum FSM_STATE {
     245//      Initial,
     246//      Retry,
     247//      Error,
     248//      Done,
     249// };
     250
     251// struct FSM_Result {
     252//      FSM_STATE state;
     253//      int error;
     254// };
     255
     256// static inline void ?{}(FSM_Result & this) { this.state = Initial; this.error = 0; }
     257// static inline bool is_error(FSM_Result & this) { return Error == this.state; }
     258// static inline bool is_done(FSM_Result & this) { return Done == this.state; }
     259
     260// static inline int error(FSM_Result & this, int error) {
     261//      this.error = error;
     262//      this.state = Error;
     263//      return error;
     264// }
     265
     266// static inline int done(FSM_Result & this) {
     267//      this.state = Done;
     268//      return 0;
     269// }
     270
     271// static inline int retry(FSM_Result & this) {
     272//      this.state = Retry;
     273//      return 0;
     274// }
     275
     276// static inline int need(FSM_Result & this) {
     277//      switch(this.state) {
     278//              case Initial:
     279//              case Retry:
     280//                      return 1;
     281//              case Error:
     282//                      if(this.error == 0) mutex(serr) serr | "State marked error but code is 0";
     283//              case Done:
     284//                      return 0;
     285//      }
     286// }
     287
     288// // Generator that handles sending the header
     289// generator header_g {
     290//      io_future_t f;
     291//      const char * next;
     292//      int fd; size_t len;
     293//      FSM_Result res;
     294// };
     295
     296// static inline void ?{}(header_g & this, int fd, const char * it, size_t len ) {
     297//      this.next = it;
     298//      this.fd = fd;
     299//      this.len = len;
     300// }
     301
     302// static inline void fill(header_g & this, struct io_uring_sqe * sqe) {
     303//      zero_sqe(sqe);
     304//      sqe->opcode = IORING_OP_SEND;
     305//      sqe->user_data = (uintptr_t)&this.f;
     306//      sqe->flags = IOSQE_IO_LINK;
     307//      sqe->fd = this.fd;
     308//      sqe->addr = (uintptr_t)this.next;
     309//      sqe->len = this.len;
     310// }
     311
     312// static inline int error(header_g & this, int error) {
     313//      int ret = close(this.fd);
     314//      if( ret != 0 ) {
     315//              mutex(serr) serr | "Failed to close fd" | errno;
     316//      }
     317//      return error(this.res, error);
     318// }
     319
     320// static inline int wait_and_process(header_g & this, sendfile_stats_t & stats) {
     321//      wait(this.f);
     322
     323//      // Did something crazy happen?
     324//      if(this.f.result > this.len) {
     325//              mutex(serr) serr | "HEADER sent too much!";
     326//              return error(this, -ERANGE);
     327//      }
     328
     329//      // Something failed?
     330//      if(this.f.result < 0) {
     331//              int error = -this.f.result;
     332//              if( error == ECONNRESET ) return error(this, -ECONNRESET);
     333//              if( error == EPIPE ) return error(this, -EPIPE);
     334//              if( error == ECANCELED ) {
     335//                      mutex(serr) serr | "HEADER was cancelled, WTF!";
     336//                      return error(this, -ECONNRESET);
     337//              }
     338//              if( error == EAGAIN || error == EWOULDBLOCK) {
     339//                      mutex(serr) serr | "HEADER got eagain, WTF!";
     340//                      return error(this, -ECONNRESET);
     341//              }
     342//      }
     343
     344//      // Done?
     345//      if(this.f.result == this.len) {
     346//              return done(this.res);
     347//      }
     348
     349//      stats.header++;
     350
     351//      // It must be a Short read
     352//      this.len  -= this.f.result;
     353//      this.next += this.f.result;
     354//      reset(this.f);
     355//      return retry(this.res);
     356// }
     357
     358// // Generator that handles splicing in a file
     359// struct splice_in_t {
     360//      io_future_t f;
     361//      int fd; int pipe; size_t len; off_t off;
     362//      short zipf_idx;
     363//      FSM_Result res;
     364// };
     365
     366// static inline void ?{}(splice_in_t & this, int fd, int pipe, size_t len) {
     367//      this.fd = fd;
     368//      this.pipe = pipe;
     369//      this.len = len;
     370//      this.off = 0;
     371//      this.zipf_idx = -1;
     372//      STATS: for(i; zipf_cnts) {
     373//              if(len <= zipf_sizes[i]) {
     374//                      this.zipf_idx = i;
     375//                      break STATS;
     376//              }
     377//      }
     378//      if(this.zipf_idx < 0) mutex(serr) serr | "SPLICE IN" | len | " greated than biggest zipf file";
     379// }
     380
     381// static inline void fill(splice_in_t & this, struct io_uring_sqe * sqe) {
     382//      zero_sqe(sqe);
     383//      sqe->opcode = IORING_OP_SPLICE;
     384//      sqe->user_data = (uintptr_t)&this.f;
     385//      sqe->flags = 0;
     386//      sqe->splice_fd_in = this.fd;
     387//      sqe->splice_off_in = this.off;
     388//      sqe->fd = this.pipe;
     389//      sqe->off = (__u64)-1;
     390//      sqe->len = this.len;
     391//      sqe->splice_flags = SPLICE_F_MOVE;
     392// }
     393
     394// static inline int wait_and_process(splice_in_t & this, sendfile_stats_t & stats ) {
     395//      wait(this.f);
     396
     397//      // Something failed?
     398//      if(this.f.result < 0) {
     399//              int error = -this.f.result;
     400//              if( error == ECONNRESET ) return error(this.res, -ECONNRESET);
     401//              if( error == EPIPE ) return error(this.res, -EPIPE);
     402//              if( error == ECANCELED ) {
     403//                      mutex(serr) serr | "SPLICE IN was cancelled, WTF!";
     404//                      return error(this.res, -ECONNRESET);
     405//              }
     406//              if( error == EAGAIN || error == EWOULDBLOCK) {
     407//                      mutex(serr) serr | "SPLICE IN got eagain, WTF!";
     408//                      return error(this.res, -ECONNRESET);
     409//              }
     410//              mutex(serr) serr | "SPLICE IN got" | error | ", WTF!";
     411//              return error(this.res, -ECONNRESET);
     412//      }
     413
     414//      // Did something crazy happen?
     415//      if(this.f.result > this.len) {
     416//              mutex(serr) serr | "SPLICE IN spliced too much!";
     417//              return error(this.res, -ERANGE);
     418//      }
     419
     420//      // Done?
     421//      if(this.f.result == this.len) {
     422//              return done(this.res);
     423//      }
     424
     425//      stats.splcin++;
     426//      stats.avgrd[this.zipf_idx].calls++;
     427//      stats.avgrd[this.zipf_idx].bytes += this.f.result;
     428
     429//      // It must be a Short read
     430//      this.len -= this.f.result;
     431//      this.off += this.f.result;
     432//      reset(this.f);
     433//      return retry(this.res);
     434// }
     435
     436// generator splice_out_g {
     437//      io_future_t f;
     438//      int pipe; int fd; size_t len;
     439//      FSM_Result res;
     440// };
     441
     442// static inline void ?{}(splice_out_g & this, int pipe, int fd, size_t len) {
     443//      this.pipe = pipe;
     444//      this.fd = fd;
     445//      this.len = len;
     446// }
     447
     448// static inline void fill(splice_out_g & this, struct io_uring_sqe * sqe) {
     449//      zero_sqe(sqe);
     450//      sqe->opcode = IORING_OP_SPLICE;
     451//      sqe->user_data = (uintptr_t)&this.f;
     452//      sqe->flags = 0;
     453//      sqe->splice_fd_in = this.pipe;
     454//      sqe->splice_off_in = (__u64)-1;
     455//      sqe->fd = this.fd;
     456//      sqe->off = (__u64)-1;
     457//      sqe->len = this.len;
     458//      sqe->splice_flags = SPLICE_F_MOVE;
     459// }
     460
     461// static inline int error(splice_out_g & this, int error) {
     462//      int ret = close(this.fd);
     463//      if( ret != 0 ) {
     464//              mutex(serr) serr | "Failed to close fd" | errno;
     465//      }
     466//      return error(this.res, error);
     467// }
     468
     469// static inline void wait_and_process(splice_out_g & this, sendfile_stats_t & stats ) {
     470//      wait(this.f);
     471
     472//      // Something failed?
     473//      if(this.f.result < 0) {
     474//              int error = -this.f.result;
     475//              if( error == ECONNRESET ) return error(this, -ECONNRESET);
     476//              if( error == EPIPE ) return error(this, -EPIPE);
     477//              if( error == ECANCELED ) {
     478//                      this.f.result = 0;
     479//                      goto SHORT_WRITE;
     480//              }
     481//              if( error == EAGAIN || error == EWOULDBLOCK) {
     482//                      mutex(serr) serr | "SPLICE OUT got eagain, WTF!";
     483//                      return error(this, -ECONNRESET);
     484//              }
     485//              mutex(serr) serr | "SPLICE OUT got" | error | ", WTF!";
     486//              return error(this, -ECONNRESET);
     487//      }
     488
     489//      // Did something crazy happen?
     490//      if(this.f.result > this.len) {
     491//              mutex(serr) serr | "SPLICE OUT spliced too much!" | this.f.result | ">" | this.len;
     492//              return error(this.res, -ERANGE);
     493//      }
     494
     495//      // Done?
     496//      if(this.f.result == this.len) {
     497//              return done(this.res);
     498//      }
     499
     500// SHORT_WRITE:
     501//      stats.splcot++;
     502
     503//      // It must be a Short Write
     504//      this.len -= this.f.result;
     505//      reset(this.f);
     506//      return retry(this.res);
     507// }
    448508
    449509int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t fsize, sendfile_stats_t & stats ) {
    450510        stats.calls++;
    451511        #if defined(LINKED_IO)
    452                 char buffer[512];
    453                 int len = fill_header(buffer, fsize);
    454                 header_g header = { fd, buffer, len };
    455                 splice_in_t splice_in = { ans_fd, pipe[1], fsize };
    456                 splice_out_g splice_out = { pipe[0], fd, fsize };
    457 
    458                 RETRY_LOOP: for() {
    459                         stats.tries++;
    460                         int have = need(header.res) + need(splice_in.res) + 1;
    461                         int idx = 0;
    462                         struct io_uring_sqe * sqes[3];
    463                         __u32 idxs[3];
    464                         struct $io_context * ctx = cfa_io_allocate(sqes, idxs, have);
    465 
    466                         if(need(splice_in.res)) { fill(splice_in, sqes[idx++]); }
    467                         if(need(   header.res)) { fill(header   , sqes[idx++]); }
    468                         fill(splice_out, sqes[idx]);
    469 
    470                         // Submit everything
    471                         asm volatile("": : :"memory");
    472                         cfa_io_submit( ctx, idxs, have, false );
    473 
    474                         // wait for the results
    475                         // Always wait for splice-in to complete as
    476                         // we may need to kill the connection if it fails
    477                         // If it already completed, this is a no-op
    478                         wait_and_process(splice_in, stats);
    479 
    480                         if(is_error(splice_in.res)) {
    481                                 if(splice_in.res.error == -EPIPE) return -ECONNRESET;
    482                                 mutex(serr) serr | "SPLICE IN failed with" | splice_in.res.error;
    483                                 close(fd);
    484                         }
    485 
    486                         // Process the other 2
    487                         wait_and_process(header, stats);
    488                         wait_and_process(splice_out, stats);
    489 
    490                         if(is_done(splice_out.res)) {
    491                                 break RETRY_LOOP;
    492                         }
    493 
    494                         // We need to wait for the completion if
    495                         // - both completed
    496                         // - the header failed
    497                         // -
    498 
    499                         if(  is_error(header.res)
    500                           || is_error(splice_in.res)
    501                           || is_error(splice_out.res)) {
    502                                 return -ECONNRESET;
    503                         }
    504                 }
    505 
    506                 return len + fsize;
     512                // char buffer[512];
     513                // int len = fill_header(buffer, fsize);
     514                // header_g header = { fd, buffer, len };
     515                // splice_in_t splice_in = { ans_fd, pipe[1], fsize };
     516                // splice_out_g splice_out = { pipe[0], fd, fsize };
     517
     518                // RETRY_LOOP: for() {
     519                //      stats.tries++;
     520                //      int have = need(header.res) + need(splice_in.res) + 1;
     521                //      int idx = 0;
     522                //      struct io_uring_sqe * sqes[3];
     523                //      __u32 idxs[3];
     524                //      struct $io_context * ctx = cfa_io_allocate(sqes, idxs, have);
     525
     526                //      if(need(splice_in.res)) { fill(splice_in, sqes[idx++]); }
     527                //      if(need(   header.res)) { fill(header   , sqes[idx++]); }
     528                //      fill(splice_out, sqes[idx]);
     529
     530                //      // Submit everything
     531                //      asm volatile("": : :"memory");
     532                //      cfa_io_submit( ctx, idxs, have, false );
     533
     534                //      // wait for the results
     535                //      // Always wait for splice-in to complete as
     536                //      // we may need to kill the connection if it fails
     537                //      // If it already completed, this is a no-op
     538                //      wait_and_process(splice_in, stats);
     539
     540                //      if(is_error(splice_in.res)) {
     541                //              if(splice_in.res.error == -EPIPE) return -ECONNRESET;
     542                //              mutex(serr) serr | "SPLICE IN failed with" | splice_in.res.error;
     543                //              int ret = close(fd);
     544                //              if( ret != 0 ) abort( "close in 'answer sendfile' error: (%d) %s\n", (int)errno, strerror(errno) );
     545                //      }
     546
     547                //      // Process the other 2
     548                //      wait_and_process(header, stats);
     549                //      wait_and_process(splice_out, stats);
     550
     551                //      if(is_done(splice_out.res)) {
     552                //              break RETRY_LOOP;
     553                //      }
     554
     555                //      // We need to wait for the completion if
     556                //      // - both completed
     557                //      // - the header failed
     558                //      // -
     559
     560                //      if(  is_error(header.res)
     561                //        || is_error(splice_in.res)
     562                //        || is_error(splice_out.res)) {
     563                //              return -ECONNRESET;
     564                //      }
     565                // }
     566
     567                // return len + fsize;
    507568        #else
    508                 stats.tries++;
    509569                int ret = answer_header(fd, fsize);
    510                 if( ret < 0 ) { close(fd); return ret; }
    511                 return sendfile(pipe, fd, ans_fd, fsize);
     570                if( ret < 0 ) { return ret; }
     571                return sendfile(pipe, fd, ans_fd, fsize, stats);
    512572        #endif
    513573}
    … …  
    528588                }
    529589                // int ret = read(fd, (void*)it, count);
    530                 if(ret == 0 ) return [OK200, true, 0, 0];
     590                if(ret == 0 ) {
     591                        ret = close(fd);
     592                        if( ret != 0 ) abort( "close in 'http read good' error: (%d) %s\n", (int)errno, strerror(errno) );
     593                        return [OK200, true, 0, 0];
     594                }
    531595                if(ret < 0 ) {
    532                         if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ;
    533                         if( errno == ECONNRESET ) { close(fd); return [E408, true, 0, 0]; }
    534                         if( errno == EPIPE ) { close(fd); return [E408, true, 0, 0]; }
     596                        if( errno == ECONNRESET || errno == EPIPE ) {
     597                                ret = close(fd);
     598                                if( ret != 0 ) abort( "close in 'http read bad' error: (%d) %s\n", (int)errno, strerror(errno) );
     599                                return [E408, true, 0, 0];
     600                        }
    535601                        abort( "read error: (%d) %s\n", (int)errno, strerror(errno) );
    536602                }
Note: See TracChangeset for help on using the changeset viewer.