Ignore:
File:
1 edited

Legend:

Unmodified
Added
Removed

  • benchmark/io/http/protocol.cfa

    r32d1383 rbf7c7ea  
    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);
    2120        // #include <linux/io_uring.h>
    2221}
    … …  
    3029#define PLAINTEXT_MEMCPY
    3130#define PLAINTEXT_NOCOPY
    32 // #define LINKED_IO
    33 #define TRUE_SENDFILE
     31#define LINKED_IO
    3432
    3533static inline __s32 wait_res( io_future_t & this ) {
    … …  
    7371                int ret = cfa_send(fd, it, len, 0, CFA_IO_LAZY);
    7472                if( ret < 0 ) {
    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                         }
     73                        if( errno == ECONNRESET || errno == EPIPE ) { close(fd); return -ECONNRESET; }
     74                        if( errno == EAGAIN || errno == EWOULDBLOCK) return -EAGAIN;
    8075
    8176                        abort( "'answer error' error: (%d) %s\n", (int)errno, strerror(errno) );
    … …  
    8984}
    9085
    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 // }
     86int 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}
    9691
    9792static int fill_header(char * it, size_t size) {
    … …  
    109104}
    110105
    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 
    159 static int sendfile( int pipe[2], int fd, int ans_fd, size_t count, sendfile_stats_t & stats ) {
    160         int zipf_idx = -1;
     106#if defined(PLAINTEXT_NOCOPY)
     107int 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
     112int 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)
     128int 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
     142int 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
     150int answer_empty( int fd ) {
     151        return answer_header(fd, 0);
     152}
     153
     154static 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) );
     176                        }
     177                        in_pipe -= ret;
     178                }
     179
     180        }
     181        return count;
     182}
     183
     184enum FSM_STATE {
     185        Initial,
     186        Retry,
     187        Error,
     188        Done,
     189};
     190
     191struct FSM_Result {
     192        FSM_STATE state;
     193        int error;
     194};
     195
     196static inline void ?{}(FSM_Result & this) { this.state = Initial; this.error = 0; }
     197static inline bool is_error(FSM_Result & this) { return Error == this.state; }
     198static inline bool is_done(FSM_Result & this) { return Done == this.state; }
     199
     200static inline int error(FSM_Result & this, int error) {
     201        this.error = error;
     202        this.state = Error;
     203        return error;
     204}
     205
     206static inline int done(FSM_Result & this) {
     207        this.state = Done;
     208        return 0;
     209}
     210
     211static inline int retry(FSM_Result & this) {
     212        this.state = Retry;
     213        return 0;
     214}
     215
     216static 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
     229generator header_g {
     230        io_future_t f;
     231        const char * next;
     232        int fd; size_t len;
     233        FSM_Result res;
     234};
     235
     236static 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
     242static 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
     252static 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
     260static 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
     299struct 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
     306static 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;
    161312        STATS: for(i; zipf_cnts) {
    162                 if(count <= zipf_sizes[i]) {
    163                         zipf_idx = i;
     313                if(len <= zipf_sizes[i]) {
     314                        this.zipf_idx = i;
    164315                        break STATS;
    165316                }
    166317        }
    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) );
    191                         }
    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 
    241         return count;
    242 }
    243 
    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 // }
     318        if(this.zipf_idx < 0) mutex(serr) serr | "SPLICE IN" | len | " greated than biggest zipf file";
     319}
     320
     321static 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
     334static 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
     376generator splice_out_g {
     377        io_future_t f;
     378        int pipe; int fd; size_t len;
     379        FSM_Result res;
     380};
     381
     382static 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
     388static 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
     401static 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
     409static 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
     440SHORT_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}
    508448
    509449int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t fsize, sendfile_stats_t & stats ) {
    510450        stats.calls++;
    511451        #if defined(LINKED_IO)
    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;
     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;
    568507        #else
     508                stats.tries++;
    569509                int ret = answer_header(fd, fsize);
    570                 if( ret < 0 ) { return ret; }
    571                 return sendfile(pipe, fd, ans_fd, fsize, stats);
     510                if( ret < 0 ) { close(fd); return ret; }
     511                return sendfile(pipe, fd, ans_fd, fsize);
    572512        #endif
    573513}
    … …  
    588528                }
    589529                // int ret = read(fd, (void*)it, count);
    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                 }
     530                if(ret == 0 ) return [OK200, true, 0, 0];
    595531                if(ret < 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                         }
     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]; }
    601535                        abort( "read error: (%d) %s\n", (int)errno, strerror(errno) );
    602536                }
Note: See TracChangeset for help on using the changeset viewer.