use libev instead of libuv

1 files changed, 361 insertions, 208 deletions

diff --git a/dns2tcp.c b/dns2tcp.c
index 79b5d41..be45d83 100644
--- a/dns2tcp.c
+++ b/dns2tcp.c
@@ -1,50 +1,204 @@
 #define _GNU_SOURCE
-#include "logutils.h"
-#include "netutils.h"
-#include <uv.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <stdint.h>
 #include <stdbool.h>
 #include <string.h>
-#include <unistd.h>
+#include <time.h>
+#include <errno.h>
 #include <signal.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <arpa/inet.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include "libev/ev.h"
 #undef _GNU_SOURCE
 
+#define DNS2TCP_VER "dns2tcp v1.1.0"
+
+#ifndef IPV6_V6ONLY
+  #define IPV6_V6ONLY 26
+#endif
+#ifndef SO_REUSEPORT
+  #define SO_REUSEPORT 15
+#endif
+#ifndef TCP_QUICKACK
+  #define TCP_QUICKACK 12
+#endif
+#ifndef TCP_SYNCNT
+  #define TCP_SYNCNT 7
+#endif
+#ifndef MSG_FASTOPEN
+  #define MSG_FASTOPEN 0x20000000
+#endif
+
+#define IP4STRLEN INET_ADDRSTRLEN /* ipv4addr max strlen */
+#define IP6STRLEN INET6_ADDRSTRLEN /* ipv6addr max strlen */
+#define PORTSTRLEN 6 /* "65535", include the null character */
+#define UDPDGRAM_MAXSIZ 1472 /* mtu:1500 - iphdr:20 - udphdr:8 */
+
+typedef uint16_t portno_t; /* 16bit */
+typedef struct sockaddr_in  skaddr4_t;
+typedef struct sockaddr_in6 skaddr6_t;
+
 #define IF_VERBOSE if (g_verbose)
-#define DNS2TCP_VERSION "dns2tcp v1.0"
+
+#define LOGINF(fmt, ...)                                                    \
+    do {                                                                    \
+        struct tm *tm = localtime(&(time_t){time(NULL)});                   \
+        printf("\e[1;32m%04d-%02d-%02d %02d:%02d:%02d INF:\e[0m " fmt "\n", \
+                tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday,            \
+                tm->tm_hour,        tm->tm_min,     tm->tm_sec,             \
+                ##__VA_ARGS__);                                             \
+    } while (0)
+
+#define LOGERR(fmt, ...)                                                    \
+    do {                                                                    \
+        struct tm *tm = localtime(&(time_t){time(NULL)});                   \
+        printf("\e[1;35m%04d-%02d-%02d %02d:%02d:%02d ERR:\e[0m " fmt "\n", \
+                tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday,            \
+                tm->tm_hour,        tm->tm_min,     tm->tm_sec,             \
+                ##__VA_ARGS__);                                             \
+    } while (0)
 
 typedef struct {
-    void *buffer;
-    uint16_t nread;
+    evio_t    watcher;
+    uint8_t   buffer[2 + UDPDGRAM_MAXSIZ]; /* msglen(16bit) + msgbuf */
+    uint16_t  nrcvsnd;
     skaddr6_t srcaddr;
-} tcp_data_t;
+} tcpwatcher_t;
+
+enum {
+    OPT_IPV6_V6ONLY = 1 << 0,
+    OPT_REUSE_PORT  = 1 << 1,
+    OPT_QUICK_ACK   = 1 << 2,
+    OPT_FAST_OPEN   = 1 << 3,
+};
 
 static bool       g_verbose                 = false;
-static uv_loop_t *g_evloop                  = NULL;
-static uv_udp_t  *g_udp_server              = NULL;
+static uint8_t    g_options                 = 0;
+static uint8_t    g_syn_maxcnt              = 0;
+static int        g_udp_sockfd              = -1;
 static char       g_listen_ipstr[IP6STRLEN] = {0};
 static portno_t   g_listen_portno           = 0;
 static skaddr6_t  g_listen_skaddr           = {0};
 static char       g_remote_ipstr[IP6STRLEN] = {0};
 static portno_t   g_remote_portno           = 0;
 static skaddr6_t  g_remote_skaddr           = {0};
+static char       g_ipstr_buf[IP6STRLEN]    = {0};
+
+static void udp_recvmsg_cb(evloop_t *evloop, evio_t *watcher, int events);
+static void tcp_connect_cb(evloop_t *evloop, evio_t *watcher, int events);
+static void tcp_sendmsg_cb(evloop_t *evloop, evio_t *watcher, int events);
+static void tcp_recvmsg_cb(evloop_t *evloop, evio_t *watcher, int events);
+
+static void set_nonblock(int sockfd) {
+    int flags = fcntl(sockfd, F_GETFL, 0);
+    if (flags < 0) {
+        LOGERR("[set_nonblock] fcntl(%d, F_GETFL): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+    if (fcntl(sockfd, F_SETFL, flags | O_NONBLOCK) < 0) {
+        LOGERR("[set_nonblock] fcntl(%d, F_SETFL): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static void set_ipv6only(int sockfd) {
+    if (setsockopt(sockfd, IPPROTO_IPV6, IPV6_V6ONLY, &(int){1}, sizeof(int)) < 0) {
+        LOGERR("[set_ipv6only] setsockopt(%d, IPV6_V6ONLY): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static void set_reuseaddr(int sockfd) {
+    if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEADDR, &(int){1}, sizeof(int)) < 0) {
+        LOGERR("[set_reuseaddr] setsockopt(%d, SO_REUSEADDR): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static void set_reuseport(int sockfd) {
+    if (setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &(int){1}, sizeof(int)) < 0) {
+        LOGERR("[set_reuseport] setsockopt(%d, SO_REUSEPORT): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static void set_nodelay(int sockfd) {
+    if (setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &(int){1}, sizeof(int)) < 0) {
+        LOGERR("[set_nodelay] setsockopt(%d, TCP_NODELAY): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static void set_quickack(int sockfd) {
+    if (setsockopt(sockfd, IPPROTO_TCP, TCP_QUICKACK, &(int){1}, sizeof(int)) < 0) {
+        LOGERR("[set_quickack] setsockopt(%d, TCP_QUICKACK): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static void set_syncnt(int sockfd, int syncnt) {
+    if (setsockopt(sockfd, IPPROTO_TCP, TCP_SYNCNT, &syncnt, sizeof(syncnt)) < 0) {
+        LOGERR("[set_syncnt] setsockopt(%d, TCP_SYNCNT): (%d) %s", sockfd, errno, strerror(errno));
+        exit(errno);
+    }
+}
+
+static int get_ipstr_family(const char *ipstr) {
+    if (!ipstr) return -1; /* invalid */
+    uint8_t ipaddr[16]; /* 16-bytes */
+    if (inet_pton(AF_INET, ipstr, &ipaddr) == 1) {
+        return AF_INET;
+    } else if (inet_pton(AF_INET6, ipstr, &ipaddr) == 1) {
+        return AF_INET6;
+    } else {
+        return -1; /* invalid */
+    }
+}
 
-static void udp_alloc_cb(uv_handle_t *udp_server, size_t sugsize, uv_buf_t *uvbuf);
-static void udp_recv_cb(uv_udp_t *udp_server, ssize_t nread, const uv_buf_t *uvbuf, const skaddr_t *skaddr, unsigned flags);
+static void build_socket_addr(int ipfamily, void *skaddr, const char *ipstr, portno_t portno) {
+    if (ipfamily == AF_INET) {
+        skaddr4_t *addr = skaddr;
+        addr->sin_family = AF_INET;
+        inet_pton(AF_INET, ipstr, &addr->sin_addr);
+        addr->sin_port = htons(portno);
+    } else {
+        skaddr6_t *addr = skaddr;
+        addr->sin6_family = AF_INET6;
+        inet_pton(AF_INET6, ipstr, &addr->sin6_addr);
+        addr->sin6_port = htons(portno);
+    }
+}
 
-static void tcp_connect_cb(uv_connect_t *connreq, int status);
-static void tcp_write_cb(uv_write_t *writereq, int status);
-static void tcp_alloc_cb(uv_handle_t *tcp_client, size_t sugsize, uv_buf_t *uvbuf);
-static void tcp_read_cb(uv_stream_t *tcp_client, ssize_t nread, const uv_buf_t *uvbuf);
-static void tcp_close_cb(uv_handle_t *tcp_client);
+static void parse_socket_addr(const void *skaddr, char *ipstr, portno_t *portno) {
+    if (((skaddr4_t *)skaddr)->sin_family == AF_INET) {
+        const skaddr4_t *addr = skaddr;
+        inet_ntop(AF_INET, &addr->sin_addr, ipstr, IP4STRLEN);
+        *portno = ntohs(addr->sin_port);
+    } else {
+        const skaddr6_t *addr = skaddr;
+        inet_ntop(AF_INET6, &addr->sin6_addr, ipstr, IP6STRLEN);
+        *portno = ntohs(addr->sin6_port);
+    }
+}
 
 static void print_command_help(void) {
-    printf("usage: dns2tcp <-L LISTEN_ADDR> <-R REMOTE_ADDR> [-vVh]\n"
-           " -L <ip#port>           udp listen address, it is required\n"
-           " -R <ip#port>           tcp remote address, it is required\n"
-           " -v                     print verbose log, default: <disabled>\n"
-           " -V                     print version number of dns2tcp and exit\n"
-           " -h                     print help information of dns2tcp and exit\n"
+    printf("usage: dns2tcp <-L listen> <-R remote> [-s syncnt] [-6rafvVh]\n"
+           " -L <ip#port>            udp listen address, this is required\n"
+           " -R <ip#port>            tcp remote address, this is required\n"
+           " -s <syncnt>             set TCP_SYNCNT(max) for remote socket\n"
+           " -6                      enable IPV6_V6ONLY for listen socket\n"
+           " -r                      enable SO_REUSEPORT for listen socket\n"
+           " -a                      enable TCP_QUICKACK for remote socket\n"
+           " -f                      enable TCP_FASTOPEN for remote socket\n"
+           " -v                      print verbose log, default: <disabled>\n"
+           " -V                      print version number of dns2tcp and exit\n"
+           " -h                      print help information of dns2tcp and exit\n"
            "bug report: https://github.com/zfl9/dns2tcp. email: zfl9.com@gmail.com\n"
     );
 }
@@ -54,11 +208,15 @@ static void parse_address_opt(char *ip_port_str, bool is_listen_addr) {
 
     char *portstr = strchr(ip_port_str, '#');
     if (!portstr) {
-        printf("[parse_address_opt] %s port is not specified\n", opt_name);
+        printf("[parse_address_opt] %s port not provided\n", opt_name);
         goto PRINT_HELP_AND_EXIT;
     }
     if (portstr == ip_port_str) {
-        printf("[parse_address_opt] %s addr is not specified\n", opt_name);
+        printf("[parse_address_opt] %s addr not provided\n", opt_name);
+        goto PRINT_HELP_AND_EXIT;
+    }
+    if (portstr == ip_port_str + strlen(ip_port_str) - 1) {
+        printf("[parse_address_opt] %s port not provided\n", opt_name);
         goto PRINT_HELP_AND_EXIT;
     }
 
@@ -67,13 +225,13 @@ static void parse_address_opt(char *ip_port_str, bool is_listen_addr) {
         printf("[parse_address_opt] %s port is invalid: %s\n", opt_name, portstr);
         goto PRINT_HELP_AND_EXIT;
     }
-    portno_t portno = strtol(portstr, NULL, 10);
+    portno_t portno = strtoul(portstr, NULL, 10);
     if (portno == 0) {
         printf("[parse_address_opt] %s port is invalid: %s\n", opt_name, portstr);
         goto PRINT_HELP_AND_EXIT;
     }
 
-    char *ipstr = ip_port_str;
+    const char *ipstr = ip_port_str;
     if (strlen(ipstr) + 1 > IP6STRLEN) {
         printf("[parse_address_opt] %s addr is invalid: %s\n", opt_name, ipstr);
         goto PRINT_HELP_AND_EXIT;
@@ -84,19 +242,14 @@ static void parse_address_opt(char *ip_port_str, bool is_listen_addr) {
         goto PRINT_HELP_AND_EXIT;
     }
 
-    void *skaddr_ptr = is_listen_addr ? &g_listen_skaddr : &g_remote_skaddr;
-    if (ipfamily == AF_INET) {
-        build_ipv4_addr(skaddr_ptr, ipstr, portno);
-    } else {
-        build_ipv6_addr(skaddr_ptr, ipstr, portno);
-    }
-
     if (is_listen_addr) {
         strcpy(g_listen_ipstr, ipstr);
         g_listen_portno = portno;
+        build_socket_addr(ipfamily, &g_listen_skaddr, ipstr, portno);
     } else {
         strcpy(g_remote_ipstr, ipstr);
         g_remote_portno = portno;
+        build_socket_addr(ipfamily, &g_remote_skaddr, ipstr, portno);
     }
     return;
 
@@ -106,25 +259,52 @@ PRINT_HELP_AND_EXIT:
 }
 
 static void parse_command_args(int argc, char *argv[]) {
-    char *opt_listen_addr = NULL;
-    char *opt_remote_addr = NULL;
+    char opt_listen_addr[IP6STRLEN + PORTSTRLEN] = {0};
+    char opt_remote_addr[IP6STRLEN + PORTSTRLEN] = {0};
 
     opterr = 0;
     int shortopt = -1;
-    const char *optstr = "L:R:vVh";
+    const char *optstr = "L:R:s:6rafvVh";
     while ((shortopt = getopt(argc, argv, optstr)) != -1) {
         switch (shortopt) {
             case 'L':
-                opt_listen_addr = optarg;
+                if (strlen(optarg) + 1 > IP6STRLEN + PORTSTRLEN) {
+                    printf("[parse_command_args] invalid listen addr: %s\n", optarg);
+                    goto PRINT_HELP_AND_EXIT;
+                }
+                strcpy(opt_listen_addr, optarg);
                 break;
             case 'R':
-                opt_remote_addr = optarg;
+                if (strlen(optarg) + 1 > IP6STRLEN + PORTSTRLEN) {
+                    printf("[parse_command_args] invalid remote addr: %s\n", optarg);
+                    goto PRINT_HELP_AND_EXIT;
+                }
+                strcpy(opt_remote_addr, optarg);
+                break;
+            case 's':
+                g_syn_maxcnt = strtoul(optarg, NULL, 10);
+                if (g_syn_maxcnt == 0) {
+                    printf("[parse_command_args] invalid tcp syn cnt: %s\n", optarg);
+                    goto PRINT_HELP_AND_EXIT;
+                }
+                break;
+            case '6':
+                g_options |= OPT_IPV6_V6ONLY;
+                break;
+            case 'r':
+                g_options |= OPT_REUSE_PORT;
+                break;
+            case 'a':
+                g_options |= OPT_QUICK_ACK;
+                break;
+            case 'f':
+                g_options |= OPT_FAST_OPEN;
                 break;
             case 'v':
                 g_verbose = true;
                 break;
             case 'V':
-                printf(DNS2TCP_VERSION"\n");
+                printf(DNS2TCP_VER"\n");
                 exit(0);
             case 'h':
                 print_command_help();
@@ -139,23 +319,17 @@ static void parse_command_args(int argc, char *argv[]) {
         }
     }
 
-    if (!opt_listen_addr) {
+    if (strlen(opt_listen_addr) == 0) {
         printf("[parse_command_args] missing option: '-L'\n");
         goto PRINT_HELP_AND_EXIT;
     }
-    if (!opt_remote_addr) {
+    if (strlen(opt_remote_addr) == 0) {
         printf("[parse_command_args] missing option: '-R'\n");
         goto PRINT_HELP_AND_EXIT;
     }
 
-    do {
-        char listenaddr_optstring[strlen(opt_listen_addr) + 1];
-        char remoteaddr_optstring[strlen(opt_remote_addr) + 1];
-        strcpy(listenaddr_optstring, opt_listen_addr);
-        strcpy(remoteaddr_optstring, opt_remote_addr);
-        parse_address_opt(listenaddr_optstring, true);
-        parse_address_opt(remoteaddr_optstring, false);
-    } while (0);
+    parse_address_opt(opt_listen_addr, true);
+    parse_address_opt(opt_remote_addr, false);
     return;
 
 PRINT_HELP_AND_EXIT:
@@ -170,194 +344,173 @@ int main(int argc, char *argv[]) {
 
     LOGINF("[main] udp listen addr: %s#%hu", g_listen_ipstr, g_listen_portno);
     LOGINF("[main] tcp remote addr: %s#%hu", g_remote_ipstr, g_remote_portno);
+    if (g_syn_maxcnt) LOGINF("[main] enable TCP_SYNCNT:%hhu sockopt", g_syn_maxcnt);
+    if (g_options & OPT_IPV6_V6ONLY) LOGINF("[main] enable IPV6_V6ONLY sockopt");
+    if (g_options & OPT_REUSE_PORT) LOGINF("[main] enable SO_REUSEPORT sockopt");
+    if (g_options & OPT_QUICK_ACK) LOGINF("[main] enable TCP_QUICKACK sockopt");
+    if (g_options & OPT_FAST_OPEN) LOGINF("[main] enable TCP_FASTOPEN sockopt");
     IF_VERBOSE LOGINF("[main] verbose mode, affect performance");
 
-    g_evloop = uv_default_loop();
-    g_udp_server = &(uv_udp_t){0};
-    uv_udp_init(g_evloop, g_udp_server);
-
-    int retval = uv_udp_bind(g_udp_server, (void *)&g_listen_skaddr, (g_listen_skaddr.sin6_family == AF_INET) ? 0 : UV_UDP_IPV6ONLY);
-    if (retval < 0) {
-        LOGERR("[main] bind failed: (%d) %s", -retval, uv_strerror(retval));
-        return -retval;
+    g_udp_sockfd = socket(g_listen_skaddr.sin6_family, SOCK_DGRAM, 0);
+    if (g_udp_sockfd < 0) {
+        LOGERR("[main] create udp socket: (%d) %s", errno, strerror(errno));
+        return errno;
     }
-    uv_udp_recv_start(g_udp_server, udp_alloc_cb, udp_recv_cb);
 
-    uv_run(g_evloop, UV_RUN_DEFAULT);
-    return 0;
-}
+    set_nonblock(g_udp_sockfd);
+    set_reuseaddr(g_udp_sockfd);
+    if (g_options & OPT_REUSE_PORT) set_reuseport(g_udp_sockfd);
+    if ((g_options & OPT_IPV6_V6ONLY) && g_listen_skaddr.sin6_family == AF_INET6) set_ipv6only(g_udp_sockfd);
 
-static void udp_alloc_cb(uv_handle_t *udp_server __attribute__((unused)), size_t sugsize __attribute__((unused)), uv_buf_t *uvbuf) {
-    uvbuf->base = malloc(DNS_PACKET_MAXSIZE + 2) + 2;
-    uvbuf->len = DNS_PACKET_MAXSIZE;
-}
+    if (bind(g_udp_sockfd, (void *)&g_listen_skaddr, g_listen_skaddr.sin6_family == AF_INET ? sizeof(skaddr4_t) : sizeof(skaddr6_t)) < 0) {
+        LOGERR("[main] bind udp address: (%d) %s", errno, strerror(errno));
+        return errno;
+    }
 
-static void udp_recv_cb(uv_udp_t *udp_server __attribute__((unused)), ssize_t nread, const uv_buf_t *uvbuf, const skaddr_t *skaddr, unsigned flags) {
-    if (nread == 0) goto FREE_UVBUF;
+    evloop_t *evloop = ev_default_loop(0);
+    evio_t *watcher = &(evio_t){0};
+    ev_io_init(watcher, udp_recvmsg_cb, g_udp_sockfd, EV_READ);
+    ev_io_start(evloop, watcher);
 
-    if (nread < 0) {
-        LOGERR("[udp_recv_cb] recv failed: (%zd) %s", -nread, uv_strerror(nread));
-        goto FREE_UVBUF;
-    }
+    ev_run(evloop, 0);
+    return 0;
+}
 
-    if (flags & UV_UDP_PARTIAL) {
-        LOGERR("[udp_recv_cb] received a partial packet, discard it");
-        goto FREE_UVBUF;
+static void udp_recvmsg_cb(evloop_t *evloop, evio_t *watcher __attribute__((unused)), int events __attribute__((unused))) {
+    tcpwatcher_t *tcpw = malloc(sizeof(*tcpw));
+    ssize_t nrecv = recvfrom(g_udp_sockfd, (void *)tcpw->buffer + 2, UDPDGRAM_MAXSIZ, 0, (void *)&tcpw->srcaddr, &(socklen_t){sizeof(tcpw->srcaddr)});
+    if (nrecv < 0) {
+        if (errno != EAGAIN && errno != EWOULDBLOCK) {
+            LOGERR("[udp_recvmsg_cb] recv from udp socket: (%d) %s", errno, strerror(errno));
+        }
+        goto FREE_TCP_WATCHER;
     }
-
-    bool is_ipv4 = skaddr->sa_family == AF_INET;
     IF_VERBOSE {
-        char ipstr[IP6STRLEN]; portno_t portno;
-        if (is_ipv4) {
-            parse_ipv4_addr((void *)skaddr, ipstr, &portno);
+        portno_t portno;
+        parse_socket_addr(&tcpw->srcaddr, g_ipstr_buf, &portno);
+        LOGINF("[udp_recvmsg_cb] recv from %s#%hu, nrecv:%zd", g_ipstr_buf, portno, nrecv);
+    }
+    *(uint16_t *)tcpw->buffer = htons(nrecv);
+    nrecv += 2; /* msglen + msgbuf */
+    tcpw->nrcvsnd = 0;
+
+    int sockfd = socket(g_remote_skaddr.sin6_family, SOCK_STREAM, 0);
+    if (sockfd < 0) {
+        LOGERR("[udp_recvmsg_cb] create tcp socket: (%d) %s", errno, strerror(errno));
+        goto FREE_TCP_WATCHER;
+    }
+    set_nonblock(sockfd);
+    set_reuseaddr(sockfd);
+    set_nodelay(sockfd);
+    if (g_syn_maxcnt) set_syncnt(sockfd, g_syn_maxcnt);
+    if (g_options & OPT_QUICK_ACK) set_quickack(sockfd);
+
+    bool tfo_succ = false;
+    if (g_options & OPT_FAST_OPEN) {
+        ssize_t nsend = sendto(sockfd, tcpw->buffer, nrecv, MSG_FASTOPEN, (void *)&g_remote_skaddr, g_remote_skaddr.sin6_family == AF_INET ? sizeof(skaddr4_t) : sizeof(skaddr6_t));
+        if (nsend < 0) {
+            if (errno != EINPROGRESS) {
+                LOGERR("[udp_recvmsg_cb] connect to %s#%hu: (%d) %s", g_remote_ipstr, g_remote_portno, errno, strerror(errno));
+                goto CLOSE_TCP_SOCKFD;
+            }
+            IF_VERBOSE LOGINF("[udp_recvmsg_cb] try to connect to %s#%hu", g_remote_ipstr, g_remote_portno);
         } else {
-            parse_ipv6_addr((void *)skaddr, ipstr, &portno);
+            tfo_succ = true;
+            tcpw->nrcvsnd = nsend;
+            IF_VERBOSE LOGINF("[udp_recvmsg_cb] tcp_fastopen connect, nsend:%zd", nsend);
+        }
+    } else {
+        if (connect(sockfd, (void *)&g_remote_skaddr, g_remote_skaddr.sin6_family == AF_INET ? sizeof(skaddr4_t) : sizeof(skaddr6_t)) < 0 && errno != EINPROGRESS) {
+            LOGERR("[udp_recvmsg_cb] connect to %s#%hu: (%d) %s", g_remote_ipstr, g_remote_portno, errno, strerror(errno));
+            goto CLOSE_TCP_SOCKFD;
         }
-        LOGINF("[udp_recv_cb] recv %zdB data from %s#%hu", nread, ipstr, portno);
+        IF_VERBOSE LOGINF("[udp_recvmsg_cb] try to connect to %s#%hu", g_remote_ipstr, g_remote_portno);
     }
 
-    uint16_t *msglen_ptr = (void *)uvbuf->base - 2;
-    *msglen_ptr = htons(nread);
-
-    uv_tcp_t *tcp_client = malloc(sizeof(uv_tcp_t));
-    uv_tcp_init(g_evloop, tcp_client);
-    uv_tcp_nodelay(tcp_client, 1);
-
-    tcp_data_t *tcp_data = calloc(1, sizeof(tcp_data_t));
-    tcp_data->buffer = uvbuf->base - 2;
-    memcpy(&tcp_data->srcaddr, skaddr, is_ipv4 ? sizeof(skaddr4_t) : sizeof(skaddr6_t));
-    tcp_client->data = tcp_data;
-
-    uv_connect_t *connreq = malloc(sizeof(uv_connect_t));
-    int retval = uv_tcp_connect(connreq, tcp_client, (void *)&g_remote_skaddr, tcp_connect_cb);
-    if (retval < 0) {
-        LOGERR("[udp_recv_cb] connect failed: (%d) %s", -retval, uv_strerror(retval));
-        uv_close((void *)tcp_client, tcp_close_cb);
-        free(connreq);
-        return;
+    if (tfo_succ && tcpw->nrcvsnd >= nrecv) {
+        tcpw->nrcvsnd = 0; /* reset to zero for recv data */
+        ev_io_init((evio_t *)tcpw, tcp_recvmsg_cb, sockfd, EV_READ);
+    } else {
+        ev_io_init((evio_t *)tcpw, tfo_succ ? tcp_sendmsg_cb : tcp_connect_cb, sockfd, EV_WRITE);
     }
-    IF_VERBOSE LOGINF("[udp_recv_cb] connecting to %s#%hu", g_remote_ipstr, g_remote_portno);
+    ev_io_start(evloop, (evio_t *)tcpw);
     return;
 
-FREE_UVBUF:
-    free(uvbuf->base - 2);
+CLOSE_TCP_SOCKFD:
+    close(sockfd);
+FREE_TCP_WATCHER:
+    free(tcpw);
 }
 
-static void tcp_connect_cb(uv_connect_t *connreq, int status) {
-    uv_stream_t *tcp_client = connreq->handle;
-    tcp_data_t *tcp_data = tcp_client->data;
-    free(connreq);
-
-    if (status < 0) {
-        LOGERR("[tcp_connect_cb] connect failed: (%d) %s", -status, uv_strerror(status));
-        goto CLOSE_TCPCLIENT;
-    }
-    IF_VERBOSE LOGINF("[tcp_connect_cb] connected to %s#%hu", g_remote_ipstr, g_remote_portno);
-
-    uint16_t msglen = ntohs(*(uint16_t *)tcp_data->buffer);
-    uv_buf_t uvbufs[] = {{
-        .base = tcp_data->buffer,
-        .len = msglen + 2,
-    }};
-    uv_write_t *writereq = malloc(sizeof(uv_write_t));
-    status = uv_write(writereq, tcp_client, uvbufs, 1, tcp_write_cb);
-    if (status < 0) {
-        LOGERR("[tcp_connect_cb] write failed: (%d) %s", -status, uv_strerror(status));
-        free(writereq);
-        goto CLOSE_TCPCLIENT;
-    }
-    IF_VERBOSE LOGINF("[tcp_connect_cb] writing %huB data to %s#%hu", msglen, g_remote_ipstr, g_remote_portno);
-    return;
-
-CLOSE_TCPCLIENT:
-    uv_close((void *)tcp_client, tcp_close_cb);
-}
-
-static void tcp_write_cb(uv_write_t *writereq, int status) {
-    uv_stream_t *tcp_client = writereq->handle;
-    free(writereq);
-
-    if (status < 0) {
-        LOGERR("[tcp_write_cb] write failed: (%d) %s", -status, uv_strerror(status));
-        uv_close((void *)tcp_client, tcp_close_cb);
+static void tcp_connect_cb(evloop_t *evloop, evio_t *watcher, int events __attribute__((unused))) {
+    if (getsockopt(watcher->fd, SOL_SOCKET, SO_ERROR, &errno, &(socklen_t){sizeof(errno)}) < 0 || errno) {
+        LOGERR("[tcp_connect_cb] connect to %s#%hu: (%d) %s", g_remote_ipstr, g_remote_portno, errno, strerror(errno));
+        ev_io_stop(evloop, watcher);
+        close(watcher->fd);
+        free(watcher);
         return;
     }
-    IF_VERBOSE LOGINF("[tcp_write_cb] data has been written to %s#%hu", g_remote_ipstr, g_remote_portno);
-
-    uv_read_start(tcp_client, tcp_alloc_cb, tcp_read_cb);
-}
-
-static void tcp_alloc_cb(uv_handle_t *tcp_client, size_t sugsize __attribute__((unused)), uv_buf_t *uvbuf) {
-    tcp_data_t *tcp_data = tcp_client->data;
-    uvbuf->base = tcp_data->buffer + tcp_data->nread;
-    uvbuf->len = DNS_PACKET_MAXSIZE + 2 - tcp_data->nread;
+    IF_VERBOSE LOGINF("[tcp_connect_cb] connect to %s#%hu succeed", g_remote_ipstr, g_remote_portno);
+    ev_set_cb(watcher, tcp_sendmsg_cb);
+    ev_invoke(evloop, watcher, EV_WRITE);
 }
 
-static void tcp_read_cb(uv_stream_t *tcp_client, ssize_t nread, const uv_buf_t *uvbuf __attribute__((unused))) {
-    tcp_data_t *tcp_data = tcp_client->data;
-
-    if (nread == 0) return;
-    if (nread < 0) {
-        if (nread != UV_EOF) LOGERR("[tcp_read_cb] read failed: (%zd) %s", -nread, uv_strerror(nread));
-        goto CLOSE_TCPCLIENT;
+static void tcp_sendmsg_cb(evloop_t *evloop, evio_t *watcher, int events __attribute__((unused))) {
+    tcpwatcher_t *tcpw = (void *)watcher;
+    uint16_t datalen = 2 + ntohs(*(uint16_t *)tcpw->buffer);
+    ssize_t nsend = send(watcher->fd, (void *)tcpw->buffer + tcpw->nrcvsnd, datalen - tcpw->nrcvsnd, 0);
+    if (nsend < 0) {
+        if (errno == EAGAIN || errno == EWOULDBLOCK) return;
+        LOGERR("[tcp_sendmsg_cb] send to %s#%hu: (%d) %s", g_remote_ipstr, g_remote_portno, errno, strerror(errno));
+        ev_io_stop(evloop, watcher);
+        close(watcher->fd);
+        free(watcher);
+        return;
     }
-    if (tcp_data->nread == 0 && nread < 2) {
-        LOGERR("[tcp_read_cb] message length is too small, discard it");
-        goto CLOSE_TCPCLIENT;
+    IF_VERBOSE LOGINF("[tcp_sendmsg_cb] send to %s#%hu, nsend:%zd", g_remote_ipstr, g_remote_portno, nsend);
+    tcpw->nrcvsnd += nsend;
+    if (tcpw->nrcvsnd >= datalen) {
+        tcpw->nrcvsnd = 0; /* reset to zero for recv data */
+        ev_io_stop(evloop, watcher);
+        ev_io_init(watcher, tcp_recvmsg_cb, watcher->fd, EV_READ);
+        ev_io_start(evloop, watcher);
     }
-    tcp_data->nread += nread;
-
-    uint16_t msglen = ntohs(*(uint16_t *)tcp_data->buffer);
-    if (tcp_data->nread - nread == 0) {
-        if (msglen > DNS_PACKET_MAXSIZE) {
-            LOGERR("[tcp_read_cb] message length is too large, discard it");
-            goto CLOSE_TCPCLIENT;
-        }
-        if (tcp_data->nread > msglen + 2) {
-            LOGERR("[tcp_read_cb] message length is incorrect, discard it");
-            goto CLOSE_TCPCLIENT;
-        }
-    }
-    if (tcp_data->nread < msglen + 2) return; /* received partial data */
+}
 
-    uv_buf_t uvbufs[] = {{
-        .base = tcp_data->buffer + 2,
-        .len = msglen,
-    }};
+static void tcp_recvmsg_cb(evloop_t *evloop, evio_t *watcher, int events __attribute__((unused))) {
+    tcpwatcher_t *tcpw = (void *)watcher;
+    void *buffer = tcpw->buffer;
 
-    IF_VERBOSE {
-        LOGINF("[tcp_read_cb] recv %huB data from %s#%hu", msglen, g_remote_ipstr, g_remote_portno);
-        char ipstr[IP6STRLEN]; portno_t portno;
-        if (tcp_data->srcaddr.sin6_family == AF_INET) {
-            parse_ipv4_addr((void *)&tcp_data->srcaddr, ipstr, &portno);
-        } else {
-            parse_ipv6_addr((void *)&tcp_data->srcaddr, ipstr, &portno);
-        }
-        LOGINF("[tcp_read_cb] send %huB data to %s#%hu", msglen, ipstr, portno);
+    ssize_t nrecv = recv(watcher->fd, buffer + tcpw->nrcvsnd, 2 + UDPDGRAM_MAXSIZ - tcpw->nrcvsnd, 0);
+    if (nrecv < 0) {
+        if (errno == EAGAIN || errno == EWOULDBLOCK) return;
+        LOGERR("[tcp_recvmsg_cb] recv from %s#%hu: (%d) %s", g_remote_ipstr, g_remote_portno, errno, strerror(errno));
+        goto FREE_TCP_WATCHER;
     }
-
-    nread = uv_udp_try_send(g_udp_server, uvbufs, 1, (void *)&tcp_data->srcaddr);
-    if (nread < 0) {
-        LOGERR("[tcp_read_cb] send failed: (%zd) %s", -nread, uv_strerror(nread));
+    if (nrecv == 0) {
+        LOGERR("[tcp_recvmsg_cb] recv from %s#%hu: connection is closed", g_remote_ipstr, g_remote_portno);
+        goto FREE_TCP_WATCHER;
+    }
+    tcpw->nrcvsnd += nrecv;
+    IF_VERBOSE LOGINF("[tcp_recvmsg_cb] recv from %s#%hu, nrecv:%zd", g_remote_ipstr, g_remote_portno, nrecv);
+    if (tcpw->nrcvsnd < 2 || tcpw->nrcvsnd < 2 + ntohs(*(uint16_t *)buffer)) return;
+
+    const void *sendto_skaddr = &tcpw->srcaddr;
+    socklen_t sendto_skaddrlen = tcpw->srcaddr.sin6_family == AF_INET ? sizeof(skaddr4_t) : sizeof(skaddr6_t);
+    ssize_t nsend = sendto(g_udp_sockfd, buffer + 2, ntohs(*(uint16_t *)buffer), 0, sendto_skaddr, sendto_skaddrlen);
+    if (nsend < 0) {
+        portno_t portno;
+        parse_socket_addr(&tcpw->srcaddr, g_ipstr_buf, &portno);
+        LOGERR("[tcp_recvmsg_cb] send to %s#%hu: (%d) %s", g_ipstr_buf, portno, errno, strerror(errno));
     } else {
         IF_VERBOSE {
-            char ipstr[IP6STRLEN]; portno_t portno;
-            if (tcp_data->srcaddr.sin6_family == AF_INET) {
-                parse_ipv4_addr((void *)&tcp_data->srcaddr, ipstr, &portno);
-            } else {
-                parse_ipv6_addr((void *)&tcp_data->srcaddr, ipstr, &portno);
-            }
-            LOGINF("[tcp_read_cb] data has been written to %s#%hu", ipstr, portno);
+            portno_t portno;
+            parse_socket_addr(&tcpw->srcaddr, g_ipstr_buf, &portno);
+            LOGINF("[tcp_recvmsg_cb] send to %s#%hu, nsend:%zd", g_ipstr_buf, portno, nsend);
         }
     }
-
-CLOSE_TCPCLIENT:
-    uv_close((void *)tcp_client, tcp_close_cb);
-}
-
-static void tcp_close_cb(uv_handle_t *tcp_client) {
-    tcp_data_t *tcp_data = tcp_client->data;
-    free(tcp_data->buffer);
-    free(tcp_data);
-    free(tcp_client);
+FREE_TCP_WATCHER:
+    ev_io_stop(evloop, watcher);
+    close(watcher->fd);
+    free(watcher);
 }