Add timer, async wait

conev.c

@@ -92,13 +92,13 @@ void del_event(struct poolhd *pool, struct eval *val)
     #else
     epoll_ctl(pool->efd, EPOLL_CTL_DEL, val->fd, 0);
     #endif
-    if (val->buff && val->buff->lock) {
-        val->buff->lock = 0;
-        val->buff->offset = 0;
+    if (val->buff) {
+        buff_unlock(val);
     }
     close(val->fd);
     val->fd = -1;
     val->mod_iter = pool->iters;
+    remove_timer(pool, val);
     pool->count--;
     
     struct eval *ev = pool->links[pool->count];
@@ -147,16 +147,18 @@ void destroy_pool(struct poolhd *pool)
 
 
 #ifndef NOEPOLL
-struct eval *next_event(struct poolhd *pool, int *offs, int *type)
+struct eval *next_event(struct poolhd *pool, int *offs, int *type, int ms)
 {
     while (1) {
         int i = *offs;
         assert(i >= -1 && i < pool->max);
         if (i < 0) {
-            i = (epoll_wait(pool->efd, pool->pevents, pool->max, -1) - 1);
-            if (i < 0) {
+            i = epoll_wait(pool->efd, pool->pevents, pool->max, ms);
+            if (!i) *type = POLLTIMEOUT;
+            if (i <= 0) {
                 return 0;
             }
+            i--;
             pool->iters++;
         }
         struct eval *val = pool->pevents[i].data.ptr;
@@ -180,12 +182,14 @@ int mod_etype(struct poolhd *pool, struct eval *val, int type)
 }
 
 #else
-struct eval *next_event(struct poolhd *pool, int *offs, int *typel)
+struct eval *next_event(struct poolhd *pool, int *offs, int *typel, int ms)
 {
     for (int i = *offs; ; i--) {
         assert(i >= -1 && i < pool->max);
         if (i < 0) {
-            if (poll(pool->pevents, pool->count, -1) <= 0) {
+            int ret = poll(pool->pevents, pool->count, ms);
+            if (!ret) *typel = POLLTIMEOUT;
+            if (ret <= 0) {
                 return 0;
             }
             i = pool->count - 1;
@@ -216,20 +220,105 @@ int mod_etype(struct poolhd *pool, struct eval *val, int type)
 }
 #endif
 
+static long time_ms(void)
+{
+    #ifndef _WIN32
+    struct timespec t;
+    clock_gettime(CLOCK_MONOTONIC_RAW, &t);
+    return t.tv_sec * 1e3 + (t.tv_nsec / 1e6);
+    #else
+    FILETIME st;
+    GetSystemTimeAsFileTime(&st);
+    return (((((uint64_t)st.dwHighDateTime) << 32) | st.dwLowDateTime) / 1e4);
+    #endif
+}
+
+
+void set_timer(struct poolhd *pool, struct eval *val, long ms)
+{
+    if (val->tv_ms) {
+        return;
+    }
+    struct eval *next = 0, *prev = pool->tv_end;
+    val->tv_ms = time_ms() + ms;
+    
+    while (prev && prev->tv_ms >= val->tv_ms) {
+        next = prev;
+        prev = prev->tv_prev;
+    }
+    val->tv_next = next;
+    val->tv_prev = prev;
+    
+    if (next) {
+        next->tv_prev = val;
+    }
+    if (prev) {
+        prev->tv_next = val;
+    }
+    if (!pool->tv_start || next == pool->tv_start) {
+        pool->tv_start = val;
+    }
+    if (!pool->tv_end || prev == pool->tv_end) {
+        pool->tv_end = val;
+    }
+}
+
+
+void remove_timer(struct poolhd *pool, struct eval *val)
+{
+    if (val->tv_prev) {
+        val->tv_prev->tv_next = val->tv_next;
+    }
+    if (val->tv_next) {
+        val->tv_next->tv_prev = val->tv_prev;
+    }
+    if (pool->tv_start == val) {
+        pool->tv_start = val->tv_next;
+    }
+    if (pool->tv_end == val) {
+        pool->tv_end = val->tv_prev;
+    }
+    val->tv_ms = 0;
+    val->tv_next = 0;
+    val->tv_prev = 0;
+}
+
+
+struct eval *next_event_tv(struct poolhd *pool, int *offs, int *type)
+{
+    int ms = 0;
+    struct eval *val = 0;
+    
+    if (pool->tv_start) {
+         ms = pool->tv_start->tv_ms - time_ms();
+    }
+    if (ms >= 0) {
+        val = next_event(pool, offs, type, ms ? ms : -1);
+    }
+    else *type = POLLTIMEOUT;
+    
+    if (!val && pool->tv_start && *type == POLLTIMEOUT) {
+        val = pool->tv_start;
+        remove_timer(pool, val);
+    }
+    return val;
+}
+
 
 void loop_event(struct poolhd *pool)
 {
-    struct eval *val;
-    int i = -1, etype;
+    int i = -1, etype = -1;
     
     while (!pool->brk) {
-        val = next_event(pool, &i, &etype);
+        struct eval *val = next_event_tv(pool, &i, &etype);
         if (!val) {
             if (get_e() == EINTR) 
                 continue;
             uniperror("(e)poll");
             break;
         }
+        LOG(LOG_L, "new event: fd: %d, type: %d\n", val->fd, etype);
+        
         int ret = (*val->cb)(pool, val, etype);
         if (ret < 0) {
             del_event(pool, val);
@@ -237,6 +326,7 @@ void loop_event(struct poolhd *pool)
     }
 }
 
+
 struct buffer *buff_get(struct buffer *root, size_t size)
 {
     struct buffer *prev = root;
@@ -256,7 +346,6 @@ struct buffer *buff_get(struct buffer *root, size_t size)
     LOG(LOG_S, "alloc new buffer\n");
     
     memset(buff, 0, sizeof(struct buffer));
-    buff->data = (char *)buff + sizeof(struct buffer);
     buff->size = size;
     
     if (prev) {

conev.h

@@ -16,6 +16,7 @@
     #include <netinet/in.h>
     #include <sys/socket.h>
     #include <unistd.h>
+    #include <time.h>
     
     #ifndef NOEPOLL
         #include <sys/epoll.h>
@@ -32,6 +33,7 @@
 #ifndef POLLRDHUP
     #define POLLRDHUP 0
 #endif
+#define POLLTIMEOUT 0
 
 struct poolhd;
 struct eval;
@@ -51,9 +53,9 @@ union sockaddr_u {
 struct buffer {
     size_t size;
     unsigned int offset;
-    char *data;
     size_t lock;
     struct buffer *next;
+    char data[];
 };
 
 struct eval {
@@ -61,6 +63,10 @@ struct eval {
     int index;
     unsigned long long mod_iter;
     evcb_t cb;
+    
+    long tv_ms;
+    struct eval *tv_next, *tv_prev;
+    
     struct eval *pair;
     struct buffer *buff;
     int flag;
@@ -87,6 +93,7 @@ struct poolhd {
     unsigned long long iters;
     bool brk;
     
+    struct eval *tv_start, *tv_end;
     struct buffer *root_buff;
 };
 
@@ -100,13 +107,23 @@ void del_event(struct poolhd *pool, struct eval *val);
 
 void destroy_pool(struct poolhd *pool);
 
-struct eval *next_event(struct poolhd *pool, int *offs, int *type);
+struct eval *next_event(struct poolhd *pool, int *offs, int *type, int ms);
 
 int mod_etype(struct poolhd *pool, struct eval *val, int type);
 
+void set_timer(struct poolhd *pool, struct eval *val, long ms);
+
+void remove_timer(struct poolhd *pool, struct eval *val);
+
 void loop_event(struct poolhd *pool);
 
 struct buffer *buff_get(struct buffer *root, size_t size);
 
 void buff_destroy(struct buffer *root);
+
+#define buff_unlock(val) \
+    val->buff->lock = 0; \
+    val->buff->offset = 0; \
+    val->buff = 0;
+    
 #endif

desync.c

@@ -33,7 +33,7 @@
 
 #define WAIT_LIMIT_MS 500
 #define DEFAULT_TTL 8
-
+#define ERR_WAIT -12
 
 int setttl(int fd, int ttl)
 {
@@ -86,48 +86,28 @@ static int drop_sack(int fd)
 }
 
 
-static inline void delay(long ms)
+static bool sock_has_notsent(int sfd)
 {
-    struct timespec time = { 
-         .tv_nsec = ms * 1e6
-    };
-    nanosleep(&time, 0);
-}
-
-
-static void wait_send_if_support(int sfd)
-{
-    int i = 0;
-    for (; params.wait_send && i < WAIT_LIMIT_MS; i++) {
     struct tcp_info tcpi;
     socklen_t ts = sizeof(tcpi);
     
     if (getsockopt(sfd, IPPROTO_TCP,
             TCP_INFO, (char *)&tcpi, &ts) < 0) {
         perror("getsockopt TCP_INFO");
-            break;
+        return 0;
     }
     if (tcpi.tcpi_state != 1) {
         LOG(LOG_E, "state: %d\n", tcpi.tcpi_state);
-            break;
+        return 0;
     }
     if (ts <= offsetof(struct tcp_info, tcpi_notsent_bytes)) {
         LOG(LOG_E, "tcpi_notsent_bytes not provided\n");
-            params.wait_send = 0;
-            break;
+        return 0;
     }
-        if (tcpi.tcpi_notsent_bytes == 0) {
-            break;
+    return tcpi.tcpi_notsent_bytes != 0;
 }
-        delay(1);
-    }
-    if (i) LOG(LOG_S, "waiting for send: %d ms\n", i);
-}
-#else
-#define wait_send_if_support(sfd) {}
 #endif
 
-
 static struct packet get_tcp_fake(const char *buffer, size_t n,
         struct proto_info *info, const struct desync_params *opt)
 {
@@ -217,7 +197,6 @@ static ssize_t send_fake(int sfd, const char *buffer,
             uniperror("splice");
             break;
         }
-        wait_send_if_support(sfd);
         memcpy(p, buffer, pos);
         
         if (setttl(sfd, params.def_ttl) < 0) {
@@ -251,6 +230,7 @@ static ssize_t send_fake(int sfd, const char *buffer,
 #endif
 
 #ifdef _WIN32
+#define sock_has_notsent(sfd) 0
 #define MAX_TF 2
 
 struct tf_s {
@@ -272,7 +252,7 @@ static struct tf_s *getTFE(void)
         for (int i = 0; i < MAX_TF; i++) {
             events[i] = tf_exems[i].ov.hEvent;
         }
-        DWORD ret = WaitForMultipleObjects(MAX_TF, events, FALSE, INFINITE);
+        DWORD ret = WaitForMultipleObjects(MAX_TF, events, FALSE, 0);
         
         if (ret >= WAIT_OBJECT_0 && ret < WAIT_OBJECT_0 + MAX_TF) {
             s = &tf_exems[ret - WAIT_OBJECT_0];
@@ -314,7 +294,7 @@ static ssize_t send_fake(int sfd, const char *buffer,
 {
     struct tf_s *s = getTFE();
     if (!s) {
-        return -1;
+        return ERR_WAIT;
     }
     HANDLE hfile = openTempFile();
     if (!hfile) {
@@ -398,7 +378,6 @@ static ssize_t send_oob(int sfd, char *buffer,
         uniperror("send");
         return -1;
     }
-    wait_send_if_support(sfd);
     
     len--;
     if (len != pos) {
@@ -420,8 +399,6 @@ static ssize_t send_disorder(int sfd,
     if (len < 0) {
         uniperror("send");
     }
-    else wait_send_if_support(sfd);
-    
     if (setttl(sfd, params.def_ttl) < 0) {
         return -1;
     }
@@ -543,13 +520,18 @@ static ssize_t tamp(char *buffer, size_t bfsize, ssize_t n,
 }
 
 
-ssize_t desync(int sfd, char *buffer, size_t bfsize,
-        ssize_t n, ssize_t offset, int dp_c)
+ssize_t desync(struct poolhd *pool, 
+        struct eval *val, struct buffer *buff, ssize_t n)
 {
-    struct desync_params dp = params.dp[dp_c];
+    struct desync_params dp = params.dp[val->pair->attempt];
     struct proto_info info = { 0 };
     
-    if (offset == 0 && params.debug) {
+    int sfd = val->fd;
+    char *buffer = buff->data;
+    size_t bfsize = buff->size;
+    ssize_t offset = buff->offset;
+    
+    if (!val->recv_count && params.debug) {
         init_proto_info(buffer, n, &info);
         
         if (info.host_pos) {
@@ -561,12 +543,8 @@ ssize_t desync(int sfd, char *buffer, size_t bfsize,
         }
     }
     n = tamp(buffer, bfsize, n, &dp, &info);
-    #ifdef __linux__
-    if (!offset && dp.drop_sack && drop_sack(sfd)) {
-        return -1;
-    }
-    #endif
-    long lp = 0;
+    
+    long lp = offset;
     struct part part;
     int i = 0, r = 0;
     
@@ -578,17 +556,20 @@ ssize_t desync(int sfd, char *buffer, size_t bfsize,
         long pos = gen_offset(part.pos, part.flag, buffer, n, lp, &info);
         pos += (long )part.s * (part.r - r);
         
-        if (!(part.flag & OFFSET_START) && offset && pos <= offset) {
-            LOG(LOG_S, "offset: %zd, skip\n", offset);
+        if (offset && pos <= offset) {
             continue;
         }
         if (pos < 0 || pos > n || pos < lp) {
             LOG(LOG_E, "split cancel: pos=%ld-%ld, n=%zd\n", lp, pos, n);
             break;
         }
-        
         ssize_t s = 0;
-        switch (part.m) {
+        
+        if (sock_has_notsent(sfd)) {
+            LOG(LOG_S, "sock_has_notsent\n");
+            s = ERR_WAIT;
+        }
+        else switch (part.m) {
             #ifdef FAKE_SUPPORT
             case DESYNC_FAKE:
                 if (pos != lp) s = send_fake(sfd, 
@@ -612,24 +593,25 @@ ssize_t desync(int sfd, char *buffer, size_t bfsize,
                 
             case DESYNC_SPLIT:
             case DESYNC_NONE:
-                s = send(sfd, buffer + lp, pos - lp, 0);
-                wait_send_if_support(sfd);
-                break;
-                
             default:
-                return -1;
+                s = send(sfd, buffer + lp, pos - lp, 0);
+                break;
         }
         LOG(LOG_S, "split: pos=%ld-%ld (%zd), m: %s\n", lp, pos, s, demode_str[part.m]);
         
+        if (s == ERR_WAIT) {
+            set_timer(pool, val, 10);
+            return lp - offset;
+        }
         if (s < 0) {
             if (get_e() == EAGAIN) {
-                return lp;
+                return lp - offset;
             }
             return -1;
         } 
         else if (s != (pos - lp)) {
             LOG(LOG_E, "%zd != %ld\n", s, pos - lp);
-            return lp + s;
+            return lp + s - offset;
         }
         lp = pos;
     }
@@ -638,16 +620,28 @@ ssize_t desync(int sfd, char *buffer, size_t bfsize,
         LOG((lp ? LOG_S : LOG_L), "send: pos=%ld-%zd\n", lp, n);
         if (send(sfd, buffer + lp, n - lp, 0) < 0) {
             if (get_e() == EAGAIN) {
-                return lp;
+                return lp - offset;
             }
             uniperror("send");
             return -1;
         }
     }
-    return n;
+    return n - offset;
 }
 
 
+int pre_desync(int sfd, int dp_c)
+{
+    struct desync_params *dp = &params.dp[dp_c];
+    
+    #ifdef __linux__
+    if (dp->drop_sack && drop_sack(sfd)) {
+        return -1;
+    }
+    #endif
+    return 0;
+}
+
 int post_desync(int sfd, int dp_c)
 {
     struct desync_params *dp = &params.dp[dp_c];

desync.h

@@ -3,6 +3,7 @@
 
 #include <stdint.h>
 #include <stddef.h>
+#include "conev.h"
 
 #ifdef _WIN32
     #include <winsock2.h>
@@ -10,12 +11,14 @@
     #include <sys/socket.h>
 #endif
 
-ssize_t desync(int sfd, char *buffer, size_t bfsize, ssize_t n, ssize_t offset, int dp_c);
+ssize_t desync(struct poolhd *pool, struct eval *val, struct buffer *buff, ssize_t n);
 
 ssize_t desync_udp(int sfd, char *buffer, ssize_t n, const struct sockaddr *dst, int dp_c);
 
 int setttl(int fd, int ttl);
 
+int pre_desync(int sfd, int dp_c);
+
 int post_desync(int sfd, int dp_c);
 
 struct proto_info {

extend.c

@@ -161,7 +161,6 @@ static int reconnect(struct poolhd *pool, struct eval *val, int m)
     val->pair = 0;
     del_event(pool, val);
     
-    //client->type = EV_IGNORE;
     client->attempt = m;
     client->cache = 1;
     client->buff->offset = 0;
@@ -347,8 +346,7 @@ static inline void free_first_req(struct eval *client)
 {
     client->cb = &on_tunnel;
     client->pair->cb = &on_tunnel;
-    client->buff->lock = 0;
-    client->buff->offset = 0;
+    buff_unlock(client);
 }
 
 
@@ -378,6 +376,9 @@ static int setup_conn(struct eval *client, const char *buffer, ssize_t n)
             && set_timeout(client->pair->fd, params.timeout)) {
         return -1;
     }
+    if (pre_desync(client->pair->fd, client->attempt)) {
+        return -1;
+    }
     return 0;
 }
 
@@ -397,24 +398,26 @@ static int cancel_setup(struct eval *remote)
 
 static int send_saved_req(struct poolhd *pool, struct eval *client)
 {
-    struct buffer *buff = buff_get(pool->root_buff, params.bfsize);
+    struct buffer *buff = buff_get(pool->root_buff, params.bfsize),
+        *cb = client->buff;
         
-    ssize_t offset = client->buff->offset;
-    ssize_t n = client->buff->lock - offset;
-    memcpy(buff->data, client->buff->data + offset, n);
+    ssize_t n = cb->lock - cb->offset;
+    memcpy(buff->data, cb->data, cb->lock);
+    buff->offset = cb->offset;
     
-    ssize_t sn = tcp_send_hook(pool, client->pair, buff, n);
+    ssize_t sn = tcp_send_hook(pool, client->pair, buff, cb->lock);
     if (sn < 0) {
         return -1;
     }
-    client->buff->offset += sn;
+    cb->offset += sn;
     if (sn < n) {
-        if (mod_etype(pool, client->pair, POLLOUT) ||
+        if (mod_etype(pool, client->pair, !client->pair->tv_ms ? POLLOUT : 0) ||
                 mod_etype(pool, client, 0)) {
             uniperror("mod_etype");
             return -1;
         }
     }
+    buff->offset = 0;
     return 0;
 }
 
@@ -423,7 +426,8 @@ int on_first_tunnel(struct poolhd *pool, struct eval *val, int etype)
 {
     struct buffer *buff = buff_get(pool->root_buff, params.bfsize);
     
-    if ((etype & POLLOUT) && val->flag == FLAG_CONN) {
+    if (val->flag == FLAG_CONN 
+            && ((etype & POLLOUT) || etype == POLLTIMEOUT)) {
         if (mod_etype(pool, val, POLLIN) ||
                 mod_etype(pool, val->pair, POLLIN)) {
             uniperror("mod_etype");
@@ -488,16 +492,11 @@ ssize_t tcp_send_hook(struct poolhd *pool,
         }
         else {
             LOG(LOG_S, "desync TCP: group=%d, round=%d, fd=%d\n", m, r, remote->fd);
-            
-            ssize_t offset = remote->pair->round_sent;
-            if (!offset && remote->round_count) offset = -1;
-            
-            sn = desync(remote->fd, 
-                buff->data + off, buff->size - off, n, offset, m);
+            sn = desync(pool, remote, buff, n);
         }
     }
     if (skip) {
-        sn = send(remote->fd, buff->data + off, n, 0);
+        sn = send(remote->fd, buff->data + off, n - off, 0);
         if (sn < 0 && get_e() == EAGAIN) {
             return 0;
         }

proxy.c

@@ -18,6 +18,7 @@
     #include <ws2tcpip.h>
     
     #define close(fd) closesocket(fd)
+    #define SHUT_RDWR SD_BOTH
 #else
     #include <errno.h>
     #include <unistd.h>
@@ -650,7 +651,7 @@ int on_tunnel(struct poolhd *pool, struct eval *val, int etype)
     ssize_t n = 0;
     struct eval *pair = val->pair;
     
-    if (etype & POLLOUT) {
+    if (etype & POLLOUT || etype == POLLTIMEOUT) {
         LOG(LOG_S, "pollout (fd=%d)\n", val->fd);
         val = pair;
         pair = val->pair;
@@ -661,7 +662,7 @@ int on_tunnel(struct poolhd *pool, struct eval *val, int etype)
         }
         n = val->buff->lock - val->buff->offset;
         
-        ssize_t sn = tcp_send_hook(pool, pair, val->buff, n);
+        ssize_t sn = tcp_send_hook(pool, pair, val->buff, val->buff->lock);
         if (sn < 0) {
             uniperror("send");
             return -1;
@@ -670,8 +671,7 @@ int on_tunnel(struct poolhd *pool, struct eval *val, int etype)
             val->buff->offset += sn;
             return 0;
         }
-        val->buff->lock = 0;
-        val->buff->offset = 0;
+        buff_unlock(val);
         
         if (mod_etype(pool, val, POLLIN) ||
                 mod_etype(pool, pair, POLLIN)) {
@@ -702,7 +702,7 @@ int on_tunnel(struct poolhd *pool, struct eval *val, int etype)
             buff->offset = sn;
             
             if (mod_etype(pool, val, 0) ||
-                    mod_etype(pool, pair, POLLOUT)) {
+                    mod_etype(pool, pair, !pair->tv_ms ? POLLOUT : 0)) {
                 uniperror("mod_etype");
                 return -1;
             }