Merge pull request #83 from kevinGC/rx-zerocopy

TCP RX zerocopy

Author: hi-rajat-kumar

check_all_options.c

@@ -68,6 +68,8 @@ void check_options_tcp(struct options *opts, struct callbacks *cb)
                         "Source ports need to be in the range of 1..0xFFFF. "
                         "Best larger than 1024.");
         }
+        CHECK(cb, !(opts->rx_zerocopy && opts->skip_rx_copy),
+              "TCP RX zerocopy is not supported in conjunction RX copy skip.");
 }
 
 void check_options_udp(struct options *opts, struct callbacks *cb)

common.c

@@ -14,13 +14,18 @@
  * limitations under the License.
  */
 
-#include <ctype.h>
 #include <fcntl.h>
 #include <netinet/tcp.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/mman.h>
 
 #include "common.h"
-#include "hexdump.h"
-#include "parse.h"
+#include "lib.h"
+#include "logging.h"
+#include "or_die.h"
 
 #define kilo (1000)
 #define kibi (1024)
@@ -187,7 +192,7 @@ void set_reuseaddr(int fd, int on, struct callbacks *cb)
                 PLOG_ERROR(cb, "setsockopt(SO_REUSEADDR)");
 }
 
-void set_zerocopy(int fd, int on, struct callbacks *cb)
+void set_tx_zerocopy(int fd, int on, struct callbacks *cb)
 {
 #ifndef SO_ZEROCOPY
 #define SO_ZEROCOPY 60

common.h

@@ -152,7 +152,7 @@ struct addrinfo **parse_local_hosts(const struct options *opts, int n,
 void set_reuseport(int fd, struct callbacks *cb);
 void set_nonblocking(int fd, struct callbacks *cb);
 void set_reuseaddr(int fd, int on, struct callbacks *cb);
-void set_zerocopy(int fd, int on, struct callbacks *cb);
+void set_tx_zerocopy(int fd, int on, struct callbacks *cb);
 void set_freebind(int fd, struct callbacks *cb);
 void set_debug(int fd, int onoff, struct callbacks *cb);
 void set_mark(int fd, int mark, struct callbacks *cb);

define_all_flags.c

@@ -106,7 +106,7 @@ struct flags_parser *add_flags_stream(struct flags_parser *fp)
         DEFINE_FLAG(fp, int,           test_length,     10,      'l', "Test length in seconds");
         DEFINE_FLAG(fp, bool,          edge_trigger,    false,   'E', "Edge-triggered epoll");
         DEFINE_FLAG(fp, bool,          reuseaddr,       false,   'R', "Use SO_REUSEADDR on sockets");
-        DEFINE_FLAG(fp, bool,          zerocopy,        false,   'Z', "Set MSG_ZEROCOPY when sending");
+        DEFINE_FLAG_NAMED(fp, bool,    tx_zerocopy,     false, "zerocopy", 'Z', "Set MSG_ZEROCOPY when sending");
         DEFINE_FLAG(fp, const struct rate_conversion *, throughput_opt, neper_units_mb_pointer_hack, 0, "Units to display for throughput");
         DEFINE_FLAG_PARSER(fp,                          throughput_opt, parse_unit);
         DEFINE_FLAG_PRINTER(fp,                         throughput_opt, print_unit);
@@ -143,6 +143,7 @@ struct flags_parser *add_flags_tcp_stream(struct flags_parser *fp)
         DEFINE_FLAG(fp, unsigned long, delay,           0,       'D', "Nanosecond delay between each send()/write()");
         DEFINE_FLAG(fp, int,           buffer_size,     16384,   'B', "Number of bytes that each read/write uses as the buffer");
         DEFINE_FLAG(fp, bool,          skip_rx_copy,    false,    0,  "Skip kernel->user payload copy on receives");
+        DEFINE_FLAG(fp, bool,          rx_zerocopy,     false,   'z', "Use TCP RX zerocopy");
         DEFINE_FLAG(fp, bool,          enable_read,     false,   'r', "Read from flows? enabled by default for the server");
         DEFINE_FLAG(fp, bool,          enable_write,    false,   'w', "Write to flows? Enabled by default for the client");
         DEFINE_FLAG(fp, bool,          split_bidir ,    false,    0,  "Bidirectional using separate tx/rx sockets");

flags.h

@@ -39,15 +39,30 @@ void flags_parser_run(struct flags_parser *fp, int argc, char **argv);
 void flags_parser_dump(struct flags_parser *fp);
 void flags_parser_destroy(struct flags_parser *fp);
 
-#define DEFINE_FLAG(FP, TYPE, VAR, DEFAULT_VALUE, SHORT_NAME, USAGE) \
-  do {                                                               \
-    TYPE default_value = DEFAULT_VALUE;                              \
-    struct options *opts = flags_parser_opts(FP);                    \
-    opts->VAR = default_value;                                       \
-    flags_parser_add(FP, SHORT_NAME, #VAR,                           \
-                     USAGE " (default " #DEFAULT_VALUE ")", #TYPE,   \
-                     &opts->VAR);                                    \
-  } while (0)
+/* You probably want DEFINE_FLAG or DEFINE_FLAG_NAMED instead of this. This
+ * macro takes extra arguments -- separating TYPE and TYPESTR -- because the
+ * preprocessor will convert a TYPE of "bool" too "_Bool" *except* in the
+ * specific case where the macro contains #TYPE. This was not fun to debug.
+ */
+#define DEFINE_FLAG_NAMED_TYPED(FP, TYPE, TYPESTR, VAR, DEFAULT_VALUE, NAME,  \
+                SHORT_NAME, USE)                                              \
+        do {                                                                  \
+                TYPE default_value = DEFAULT_VALUE;                           \
+                struct options *opts = flags_parser_opts(FP);                 \
+                opts->VAR = default_value;                                    \
+                flags_parser_add(FP, SHORT_NAME, NAME,                        \
+                                USE " (default " #DEFAULT_VALUE ")", TYPESTR, \
+                                &opts->VAR);                                  \
+        } while (0);
+
+/* DEFINE_FLAG_NAMED is like DEFINE_FLAG, but uses a custom flag name. */
+#define DEFINE_FLAG_NAMED(FP, TYPE, VAR, DEFAULT_VALUE, NAME, SHORT_NAME, USE) \
+        DEFINE_FLAG_NAMED_TYPED(FP, TYPE, #TYPE, VAR, DEFAULT_VALUE, NAME,     \
+                        SHORT_NAME, USE)
+
+#define DEFINE_FLAG(FP, TYPE, VAR, DEFAULT_VALUE, SHORT_NAME, USAGE)       \
+        DEFINE_FLAG_NAMED_TYPED(FP, TYPE, #TYPE, VAR, DEFAULT_VALUE, #VAR, \
+                        SHORT_NAME, USAGE)
 
 #define DEFINE_FLAG_PARSER(FP, VAR, PARSER) do { \
         struct options *opts = flags_parser_opts(FP); \

flow.c

@@ -14,14 +14,19 @@
  * limitations under the License.
  */
 
+#include <linux/tcp.h>
 #include <stdint.h>
+#include <stdlib.h>
+#include <sys/mman.h>
 #include <time.h>
+#include <unistd.h>
 
 #include "common.h"
 #include "flow.h"
+#include "logging.h"
 #include "socket.h"
-#include "thread.h"
 #include "stats.h"
+#include "thread.h"
 
 /*
  * We define the flow struct locally to this file to force outside users to go
@@ -41,6 +46,10 @@ struct flow {
         uint32_t        f_events;   /* pending epoll events */
 
         struct neper_stat *f_stat;
+
+	/* TCP RX zerocopy state. */
+	void *f_rx_zerocopy_buffer;
+	size_t f_rx_zerocopy_buffer_sz;
 };
 
 int flow_fd(const struct flow *f)
@@ -120,7 +129,24 @@ void flow_reconnect(struct flow *f, flow_handler fh, uint32_t events)
         flow_ctl(f, EPOLL_CTL_ADD, fh, events, true);
 }
 
-void flow_create(const struct flow_create_args *args)
+void flow_init_rx_zerocopy(struct flow *f, int buffer_size, struct callbacks *cb)
+{
+        // Use RCVLOWAT to reduce syscall overhead.
+        int rcvlowat = buffer_size;
+        if (setsockopt(f->f_fd, SOL_SOCKET, SO_RCVLOWAT, &rcvlowat,
+                       sizeof(rcvlowat)) == -1)
+                PLOG_FATAL(cb, "setsockopt(SO_RCVLOWAT)");
+
+        // Zerocopy requires mmap'd pages. Each flow has its own pages.
+        f->f_rx_zerocopy_buffer = mmap(NULL, buffer_size, PROT_READ,
+                        MAP_SHARED, f->f_fd, 0);
+        if (f->f_rx_zerocopy_buffer == (void *)-1)
+                PLOG_FATAL(cb, "failed to map RX zerocopy buffer");
+
+        f->f_rx_zerocopy_buffer_sz = buffer_size;
+}
+
+struct flow *flow_create(const struct flow_create_args *args)
 {
         struct thread *t = args->thread;
         struct flow *f = calloc_or_die(1, sizeof(struct flow), t->cb);
@@ -164,6 +190,7 @@ void flow_create(const struct flow_create_args *args)
                 events &= (f->f_id & 1) ? EPOLLOUT : EPOLLIN;
 
         flow_ctl(f, EPOLL_CTL_ADD, args->handler, events, true);
+	return f;
 }
 
 /* Returns true if the deadline for the flow has expired.
@@ -292,10 +319,58 @@ void flow_delete(struct flow *f)
          */
         if (f->f_mbuf != f->f_thread->f_mbuf)
                 free(f->f_mbuf);
+
+        /* Cleanup TCP RX zerocopy. */
+        if (f->f_rx_zerocopy_buffer)
+                munmap(f->f_rx_zerocopy_buffer, f->f_rx_zerocopy_buffer_sz);
+
         free(f);
 }
 
 void flow_update_next_event(struct flow *f, uint64_t duration)
 {
         f->f_next_event += duration;
 }
+
+ssize_t flow_recv_zerocopy(struct flow *f, void *copybuf, size_t copybuf_len) {
+        struct tcp_zerocopy_receive zc = {0};
+        socklen_t zc_len = sizeof(zc);
+        int result;
+
+        /* Setup both the mmap address and extra buffer for bytes that aren't
+         * zerocopy-able.
+         */
+        zc.address = (__u64)f->f_rx_zerocopy_buffer;
+        zc.length = copybuf_len; /* Same size used as zerocopy buffer. */
+
+        /* The kernel will effectively use copybuf_len as a hint as to what the
+         * cutoff point between zerocopy and recv is. So passing a large copybuf
+         * causes less zerocopy. Thus we pass just under a page to maximize
+         * zerocopying.
+         */
+        zc.copybuf_address = (__u64)copybuf;
+        zc.copybuf_len = copybuf_len < 4096 ? copybuf_len : 4095;
+
+        result = getsockopt(f->f_fd, IPPROTO_TCP, TCP_ZEROCOPY_RECEIVE, &zc,
+                            &zc_len);
+        if (result == -1)
+                return result;
+
+        /* Handle overflow data, i.e. bytes that couldn't be zerocopied. */
+        if (zc.recv_skip_hint) {
+                int read_len = zc.recv_skip_hint < copybuf_len ?
+                        zc.recv_skip_hint : copybuf_len;
+                result = read(f->f_fd, copybuf, read_len);
+                if (result < 0)
+                        PLOG_FATAL(f->f_thread->cb, "failed to read extra "
+                                        "bytes");
+        }
+
+        /* Handle zerocopy data. */
+        if (zc.length) {
+                flow_thread(f)->io_stats.rx_zc_bytes += zc.length;
+                madvise(f->f_rx_zerocopy_buffer, zc.length, MADV_DONTNEED);
+        }
+
+        return zc.recv_skip_hint + zc.length + zc.copybuf_len;
+}

flow.h

@@ -56,9 +56,15 @@ struct flow_create_args {
         struct neper_stat *(*stat)(struct flow *); /* stats callback */
 };
 
-void flow_create(const struct flow_create_args *);
+struct flow *flow_create(const struct flow_create_args *);
 void flow_delete(struct flow *);
 /* Adds duration to f's next event time. */
 void flow_update_next_event(struct flow *f, uint64_t duration);
 
+/* Initialize RX zerocopy state for a flow. */
+void flow_init_rx_zerocopy(struct flow *f, int buffer_size,
+                          struct callbacks *cb);
+/* Perform a zerocopy receive. */
+ssize_t flow_recv_zerocopy(struct flow *f, void *copybuf, size_t copybuf_len);
+
 #endif

lib.h

@@ -92,7 +92,8 @@ struct options {
         bool freebind;
         bool tcp_fastopen;
         bool skip_rx_copy;
-        bool zerocopy;
+        bool tx_zerocopy;
+        bool rx_zerocopy;
         bool time_wait;
         double interval;
         long long max_pacing_rate;

psp_stream_main.c

@@ -61,7 +61,7 @@ int main(int argc, char **argv)
 
         if (opts.skip_rx_copy)
                 opts.recv_flags = MSG_TRUNC;
-        if (opts.zerocopy)
+        if (opts.tx_zerocopy)
                 opts.send_flags = MSG_ZEROCOPY;
 
         flags_parser_dump(fp);

rr.c

@@ -101,7 +101,11 @@ static ssize_t rr_fn_recv(struct flow *f, char *buf, size_t len)
         struct thread *t = flow_thread(f);
         ssize_t ret;
 
-        ret = recv(flow_fd(f), buf, len, 0);
+        if (t->opts->rx_zerocopy) {
+                ret = flow_recv_zerocopy(f, buf, len);
+	} else {
+		ret = recv(flow_fd(f), buf, len, 0);
+	}
         t->io_stats.rx_ops++;
         t->io_stats.rx_bytes += ret > 0 ? ret : 0;
         return ret;