diff --git a/examples/meson.build b/examples/meson.build index 96f70a2..a3705e2 100644 --- a/examples/meson.build +++ b/examples/meson.build @@ -14,7 +14,8 @@ dirs = [ 'arpresolver', 'mica', 'txgen', - 'mtcp' + 'mtcp', + 'mtcp-utils', ] def_deps = [include, log, nf, params, uds] diff --git a/examples/mtcp-utils/meson.build b/examples/mtcp-utils/meson.build new file mode 100644 index 0000000..e475a12 --- /dev/null +++ b/examples/mtcp-utils/meson.build @@ -0,0 +1,10 @@ +# SPDX-License-Identifier: Apache-2.0 +# Copyright (c) 2025 Debojeet Das + +sources = files('mtcp-enter.c') + +executable('mtcp-enter', sources, c_args: cflags, install: true, dependencies: deps) + +sources2 = files('mtcp-exit.c') + +executable('mtcp-exit', sources2, c_args: cflags, install: true, dependencies: deps) diff --git a/examples/mtcp-utils/mtcp-enter.c b/examples/mtcp-utils/mtcp-enter.c new file mode 100644 index 0000000..ace87ae --- /dev/null +++ b/examples/mtcp-utils/mtcp-enter.c @@ -0,0 +1,252 @@ +/* SPDX-License-Identifier: Apache-2.0 + * Copyright (c) 2025 Debojeet Das + * + * mtcp-enter: A simple NF that forwards pkt after adding a flash trailer + */ +#include +#include +#include + +#include +#include +#include + +volatile bool done = false; +struct config *cfg = NULL; +struct nf *nf = NULL; + +///////////////////////////////////// +static int src_nf = 0; +static int dst_nf = 0; + +static void int_exit(int sig) +{ + log_info("Received Signal: %d", sig); + done = true; +} + +struct appconf { + int cpu_start; + int cpu_end; + int stats_cpu; +} app_conf; + +// clang-format off +static const char *l2fwd_options[] = { + "-c \tStart CPU (default: 0)", + "-e \tEnd CPU (default: 0)", + "-s \tStats CPU (default: 1)", + "-S \tSource NF ID (default: 0)", + "-D \tDestination NF ID (default: 0)", + NULL +}; +// clang-format on + +static int parse_app_args(int argc, char **argv, struct appconf *app_conf, int shift) +{ + int c; + opterr = 0; + + app_conf->cpu_start = 0; + app_conf->cpu_end = 0; + app_conf->stats_cpu = 1; + + argc -= shift; + argv += shift; + + while ((c = getopt(argc, argv, "hc:e:s:S:D:")) != -1) + switch (c) { + case 'h': + printf("Usage: %s -h\n", argv[-shift]); + return -1; + case 'c': + app_conf->cpu_start = atoi(optarg); + break; + case 'e': + app_conf->cpu_end = atoi(optarg); + break; + case 's': + app_conf->stats_cpu = atoi(optarg); + break; + case 'S': + src_nf = atoi(optarg); + break; + case 'D': + dst_nf = atoi(optarg); + break; + default: + printf("Usage: %s -h\n", argv[-shift]); + return -1; + } + + return 0; +} + +static void do_nothing(void *data) +{ + /* This is stupid but it makes sure that compiler doesn't through any errors */ + (void)data; +} + +struct sock_args { + int socket_id; +}; + +static void *socket_routine(void *arg) +{ + int ret; + nfds_t nfds = 1; + struct socket *xsk; + struct xskvec *xskvecs; + struct pollfd fds[1] = {}; + uint32_t i, nrecv, nsend, nb_frags = 0; + struct sock_args *a = (struct sock_args *)arg; + + log_debug("Socket ID: %d", a->socket_id); + xsk = nf->thread[a->socket_id]->socket; + + xskvecs = calloc(cfg->xsk->batch_size, sizeof(struct xskvec)); + if (!xskvecs) { + log_error("Failed to allocate xskvecs array"); + return NULL; + } + + fds[0].fd = xsk->fd; + fds[0].events = POLLIN; + + for (;;) { + ret = flash__poll(cfg, xsk, fds, nfds); + if (!(ret == 1 || ret == -2)) + continue; + + nrecv = flash__recvmsg(cfg, xsk, xskvecs, cfg->xsk->batch_size); + for (i = 0; i < nrecv; i++) { + char *pkt = xskvecs[i].data; + + if (!nb_frags++) + do_nothing(pkt); + + if (IS_EOP_DESC(xskvecs[i].options)) + nb_frags = 0; + + + char *flash_header = pkt + xskvecs[i].len; + flash_header[0] = src_nf; + flash_header[1] = dst_nf; + xskvecs[i].len += 2; + xskvecs[i].options = 0; + } + + if (nrecv) { + nsend = flash__sendmsg(cfg, xsk, xskvecs, nrecv); + if (nsend != nrecv) { + log_error("errno: %d/\"%s\"", errno, strerror(errno)); + break; + } + } + + if (done) + break; + } + + free(xskvecs); + return NULL; +} + +int main(int argc, char **argv) +{ + int shift; + struct sock_args *args; + struct stats_conf stats_cfg = { NULL }; + cpu_set_t cpuset; + pthread_t socket_thread, stats_thread; + + cfg = calloc(1, sizeof(struct config)); + if (!cfg) { + log_error("ERROR: Memory allocation failed"); + exit(EXIT_FAILURE); + } + + cfg->app_name = "MTCP-enter"; + cfg->app_options = l2fwd_options; + cfg->done = &done; + + shift = flash__parse_cmdline_args(argc, argv, cfg); + if (shift < 0) + goto out_cfg; + + if (parse_app_args(argc, argv, &app_conf, shift) < 0) + goto out_cfg; + + if (flash__configure_nf(&nf, cfg) < 0) + goto out_cfg; + + log_info("Control Plane setup done..."); + + signal(SIGINT, int_exit); + signal(SIGTERM, int_exit); + signal(SIGABRT, int_exit); + + log_info("Starting Data Path..."); + + args = calloc(cfg->total_sockets, sizeof(struct sock_args)); + if (!args) { + log_error("ERROR: Memory allocation failed for sock_args"); + goto out_cfg_close; + } + + for (int i = 0; i < cfg->total_sockets; i++) { + args[i].socket_id = i; + + if (pthread_create(&socket_thread, NULL, socket_routine, &args[i])) { + log_error("Error creating socket thread"); + goto out_args; + } + + CPU_ZERO(&cpuset); + CPU_SET((i % (app_conf.cpu_end - app_conf.cpu_start + 1)) + app_conf.cpu_start, &cpuset); + if (pthread_setaffinity_np(socket_thread, sizeof(cpu_set_t), &cpuset) != 0) { + log_error("ERROR: Unable to set thread affinity: %s", strerror(errno)); + goto out_args; + } + + if (pthread_detach(socket_thread) != 0) { + log_error("ERROR: Unable to detach thread: %s", strerror(errno)); + goto out_args; + } + } + + stats_cfg.nf = nf; + stats_cfg.cfg = cfg; + + if (pthread_create(&stats_thread, NULL, flash__stats_thread, &stats_cfg)) { + log_error("Error creating statistics thread"); + goto out_args; + } + CPU_ZERO(&cpuset); + CPU_SET(app_conf.stats_cpu, &cpuset); + if (pthread_setaffinity_np(stats_thread, sizeof(cpu_set_t), &cpuset) != 0) { + log_error("ERROR: Unable to set thread affinity: %s", strerror(errno)); + goto out_args; + } + + if (pthread_detach(stats_thread) != 0) { + log_error("ERROR: Unable to detach thread: %s", strerror(errno)); + goto out_args; + } + + flash__wait(cfg); + flash__xsk_close(cfg, nf); + + exit(EXIT_SUCCESS); + +out_args: + done = true; + free(args); +out_cfg_close: + sleep(1); + flash__xsk_close(cfg, nf); +out_cfg: + free(cfg); + exit(EXIT_FAILURE); +} diff --git a/examples/mtcp-utils/mtcp-exit.c b/examples/mtcp-utils/mtcp-exit.c new file mode 100644 index 0000000..e418ea5 --- /dev/null +++ b/examples/mtcp-utils/mtcp-exit.c @@ -0,0 +1,245 @@ +/* SPDX-License-Identifier: Apache-2.0 + * Copyright (c) 2025 Debojeet Das + * + * mtcp-exit: A simple NF that forwards pkt outside the host after removing the flash trailer + */ +#include +#include +#include + +#include +#include +#include + +volatile bool done = false; +struct config *cfg = NULL; +struct nf *nf = NULL; + +static void int_exit(int sig) +{ + log_info("Received Signal: %d", sig); + done = true; +} + +struct appconf { + int cpu_start; + int cpu_end; + int stats_cpu; +} app_conf; + +// clang-format off +static const char *l2fwd_options[] = { + "-c \tStart CPU (default: 0)", + "-e \tEnd CPU (default: 0)", + "-s \tStats CPU (default: 1)", + NULL +}; +// clang-format on + +// static void hex_dump_pkt(void* pkt, int len) { +// unsigned char *data = (unsigned char *)pkt; +// for (int i = 0; i < len; i++) { +// printf("%02x ", data[i]); +// } +// printf("\n"); +// } + +static int parse_app_args(int argc, char **argv, struct appconf *app_conf, int shift) +{ + int c; + opterr = 0; + + app_conf->cpu_start = 0; + app_conf->cpu_end = 0; + app_conf->stats_cpu = 1; + + argc -= shift; + argv += shift; + + while ((c = getopt(argc, argv, "hc:e:s:S:D:")) != -1) + switch (c) { + case 'h': + printf("Usage: %s -h\n", argv[-shift]); + return -1; + case 'c': + app_conf->cpu_start = atoi(optarg); + break; + case 'e': + app_conf->cpu_end = atoi(optarg); + break; + case 's': + app_conf->stats_cpu = atoi(optarg); + break; + default: + printf("Usage: %s -h\n", argv[-shift]); + return -1; + } + + return 0; +} + +static void do_nothing(void *data) +{ + /* This is stupid but it makes sure that compiler doesn't through any errors */ + (void)data; +} + +struct sock_args { + int socket_id; +}; + +static void *socket_routine(void *arg) +{ + int ret; + nfds_t nfds = 1; + struct socket *xsk; + struct xskvec *xskvecs; + struct pollfd fds[1] = {}; + uint32_t i, nrecv, nsend, nb_frags = 0; + struct sock_args *a = (struct sock_args *)arg; + + log_debug("Socket ID: %d", a->socket_id); + xsk = nf->thread[a->socket_id]->socket; + + xskvecs = calloc(cfg->xsk->batch_size, sizeof(struct xskvec)); + if (!xskvecs) { + log_error("Failed to allocate xskvecs array"); + return NULL; + } + + fds[0].fd = xsk->fd; + fds[0].events = POLLIN; + + for (;;) { + ret = flash__poll(cfg, xsk, fds, nfds); + if (!(ret == 1 || ret == -2)) + continue; + + nrecv = flash__recvmsg(cfg, xsk, xskvecs, cfg->xsk->batch_size); + for (i = 0; i < nrecv; i++) { + char *pkt = xskvecs[i].data; + + if (!nb_frags++) + do_nothing(pkt); + + if (IS_EOP_DESC(xskvecs[i].options)) + nb_frags = 0; + + xskvecs[i].len -= 2; // Remove flash header + + // hex_dump_pkt(pkt, xskvecs[i].len); + } + + if (nrecv) { + nsend = flash__sendmsg(cfg, xsk, xskvecs, nrecv); + if (nsend != nrecv) { + log_error("errno: %d/\"%s\"", errno, strerror(errno)); + break; + } + } + + if (done) + break; + } + + free(xskvecs); + return NULL; +} + +int main(int argc, char **argv) +{ + int shift; + struct sock_args *args; + struct stats_conf stats_cfg = { NULL }; + cpu_set_t cpuset; + pthread_t socket_thread, stats_thread; + + cfg = calloc(1, sizeof(struct config)); + if (!cfg) { + log_error("ERROR: Memory allocation failed"); + exit(EXIT_FAILURE); + } + + cfg->app_name = "MTCP-exit"; + cfg->app_options = l2fwd_options; + cfg->done = &done; + + shift = flash__parse_cmdline_args(argc, argv, cfg); + if (shift < 0) + goto out_cfg; + + if (parse_app_args(argc, argv, &app_conf, shift) < 0) + goto out_cfg; + + if (flash__configure_nf(&nf, cfg) < 0) + goto out_cfg; + + log_info("Control Plane setup done..."); + + signal(SIGINT, int_exit); + signal(SIGTERM, int_exit); + signal(SIGABRT, int_exit); + + log_info("Starting Data Path..."); + + args = calloc(cfg->total_sockets, sizeof(struct sock_args)); + if (!args) { + log_error("ERROR: Memory allocation failed for sock_args"); + goto out_cfg_close; + } + + for (int i = 0; i < cfg->total_sockets; i++) { + args[i].socket_id = i; + + if (pthread_create(&socket_thread, NULL, socket_routine, &args[i])) { + log_error("Error creating socket thread"); + goto out_args; + } + + CPU_ZERO(&cpuset); + CPU_SET((i % (app_conf.cpu_end - app_conf.cpu_start + 1)) + app_conf.cpu_start, &cpuset); + if (pthread_setaffinity_np(socket_thread, sizeof(cpu_set_t), &cpuset) != 0) { + log_error("ERROR: Unable to set thread affinity: %s", strerror(errno)); + goto out_args; + } + + if (pthread_detach(socket_thread) != 0) { + log_error("ERROR: Unable to detach thread: %s", strerror(errno)); + goto out_args; + } + } + + stats_cfg.nf = nf; + stats_cfg.cfg = cfg; + + if (pthread_create(&stats_thread, NULL, flash__stats_thread, &stats_cfg)) { + log_error("Error creating statistics thread"); + goto out_args; + } + CPU_ZERO(&cpuset); + CPU_SET(app_conf.stats_cpu, &cpuset); + if (pthread_setaffinity_np(stats_thread, sizeof(cpu_set_t), &cpuset) != 0) { + log_error("ERROR: Unable to set thread affinity: %s", strerror(errno)); + goto out_args; + } + + if (pthread_detach(stats_thread) != 0) { + log_error("ERROR: Unable to detach thread: %s", strerror(errno)); + goto out_args; + } + + flash__wait(cfg); + flash__xsk_close(cfg, nf); + + exit(EXIT_SUCCESS); + +out_args: + done = true; + free(args); +out_cfg_close: + sleep(1); + flash__xsk_close(cfg, nf); +out_cfg: + free(cfg); + exit(EXIT_FAILURE); +} diff --git a/examples/mtcp/epserver.c b/examples/mtcp/epserver.c index 0a7639d..6d036b5 100644 --- a/examples/mtcp/epserver.c +++ b/examples/mtcp/epserver.c @@ -318,7 +318,9 @@ static struct thread_context *InitializeServerThread(int core) return NULL; } /* create mtcp context: this will spawn an mtcp thread */ - ctx->mctx = mtcp_create_context(core); + // h -> assuming single thread only + // h -> assuming epserver wants nic queue 0 + ctx->mctx = mtcp_create_context(core, 0); if (!ctx->mctx) { TRACE_ERROR("Failed to create mtcp context!\n"); free(ctx); diff --git a/lib/flash/mtcp/api.c b/lib/flash/mtcp/api.c index e3dac72..788035e 100644 --- a/lib/flash/mtcp/api.c +++ b/lib/flash/mtcp/api.c @@ -691,7 +691,7 @@ int mtcp_init_rss(mctx_t mctx, in_addr_t saddr_base, int num_addr, in_addr_t dad return 0; } /*----------------------------------------------------------------------------*/ -int mtcp_connect(mctx_t mctx, int sockid, const struct sockaddr *addr, socklen_t addrlen) +int mtcp_connect(mctx_t mctx, int sockid, const struct sockaddr *addr, socklen_t addrlen, uint8_t dst_flash_id) { mtcp_manager_t mtcp; socket_map_t socket; @@ -707,6 +707,12 @@ int mtcp_connect(mctx_t mctx, int sockid, const struct sockaddr *addr, socklen_t return -1; } +#ifdef MTCP_FLASH_ID_TRAILER + mtcp->ctx->flash_ctx.dst_flash_id = dst_flash_id; +#else + UNUSED(dst_flash_id); +#endif + if (sockid < 0 || sockid >= CONFIG.max_concurrency) { TRACE_API("Socket id %d out of range.\n", sockid); errno = EBADF; @@ -1041,7 +1047,7 @@ int mtcp_close(mctx_t mctx, int sockid) // return ERROR; // } else if (cur_stream->state == TCP_ST_SYN_SENT) { -// /* TODO: this should notify event failure to all +// /* TODO: this should notify event failure to all // previous read() or write() calls */ // cur_stream->state = TCP_ST_CLOSED; // cur_stream->close_reason = TCP_ACTIVE_CLOSE; @@ -1117,8 +1123,13 @@ static inline int CopyToUser(mtcp_manager_t mtcp, tcp_stream *cur_stream, char * prev_rcv_wnd = rcvvar->rcv_wnd; /* Copy data to user buffer and remove it from receiving buffer */ +#ifdef MTCP_RX_ZERO_COPY + RBRead_zc(mtcp->rbm_rcv, rcvvar->rcvbuf, buf, copylen); + RBRemove_zc(mtcp, mtcp->rbm_rcv, rcvvar->rcvbuf, copylen, AT_APP); +#else memcpy(buf, rcvvar->rcvbuf->head, copylen); RBRemove(mtcp->rbm_rcv, rcvvar->rcvbuf, copylen, AT_APP); +#endif rcvvar->rcv_wnd = rcvvar->rcvbuf->size - rcvvar->rcvbuf->merged_len; /* Advertise newly freed receive buffer */ @@ -1638,3 +1649,652 @@ int mtcp_writev(mctx_t mctx, int sockid, const struct iovec *iov, int numIOV) return to_write; } /*----------------------------------------------------------------------------*/ + +#ifdef MTCP_RX_ZERO_COPY +/*----------------------------------------------------------------------------*/ +static inline int ZeroCopyToUser(mtcp_manager_t mtcp, tcp_stream *cur_stream, char **buf, uint64_t *flash_addr) +{ + struct tcp_recv_vars *rcvvar = cur_stream->rcvvar; + uint32_t prev_rcv_wnd; + prev_rcv_wnd = rcvvar->rcv_wnd; + + size_t len = RBGetBuff_zc(mtcp->rbm_rcv, rcvvar->rcvbuf, (void **)buf, flash_addr); + if (*buf == NULL) { // or len == 0 + errno = EAGAIN; + return -1; + } + + rcvvar->rcv_wnd = rcvvar->rcvbuf->size - rcvvar->rcvbuf->merged_len; + + if (cur_stream->need_wnd_adv) { + if (rcvvar->rcv_wnd > cur_stream->sndvar->eff_mss) { + if (!cur_stream->sndvar->on_ackq) { + SQ_LOCK(&mtcp->ctx->ackq_lock); + cur_stream->sndvar->on_ackq = TRUE; + StreamEnqueue(mtcp->ackq, cur_stream); /* this always success */ + SQ_UNLOCK(&mtcp->ctx->ackq_lock); + cur_stream->need_wnd_adv = FALSE; + mtcp->wakeup_flag = TRUE; + } + } + } + + UNUSED(prev_rcv_wnd); + return len; +} +/*----------------------------------------------------------------------------*/ +ssize_t mtcp_recv_zc(mctx_t mctx, int sockid, char **buf, uint64_t *flash_addr, int flags) +{ + mtcp_manager_t mtcp; + socket_map_t socket; + tcp_stream *cur_stream; + struct tcp_recv_vars *rcvvar; + int event_remaining; + int ret = 0; + + mtcp = GetMTCPManager(mctx); + if (!mtcp) { + return -1; + } + + if (sockid < 0 || sockid >= CONFIG.max_concurrency) { + TRACE_API("Socket id %d out of range.\n", sockid); + errno = EBADF; + return -1; + } + + socket = &mtcp->smap[sockid]; + if (socket->socktype == MTCP_SOCK_UNUSED) { + TRACE_API("Invalid socket id: %d\n", sockid); + errno = EBADF; + return -1; + } + + // h-> pipes not supported in mtcp_recv_zc + // if (socket->socktype == MTCP_SOCK_PIPE) { + // return PipeRead(mctx, sockid, buf, len); + // } + + if (socket->socktype != MTCP_SOCK_STREAM) { + TRACE_API("Not an end socket. id: %d\n", sockid); + errno = ENOTSOCK; + return -1; + } + + /* stream should be in ESTABLISHED, FIN_WAIT_1, FIN_WAIT_2, CLOSE_WAIT */ + cur_stream = socket->stream; + if (!cur_stream || !(cur_stream->state >= TCP_ST_ESTABLISHED && cur_stream->state <= TCP_ST_CLOSE_WAIT)) { + errno = ENOTCONN; + return -1; + } + + rcvvar = cur_stream->rcvvar; + + /* if CLOSE_WAIT, return 0 if there is no payload */ + if (cur_stream->state == TCP_ST_CLOSE_WAIT) { + if (!rcvvar->rcvbuf) + return 0; + + if (rcvvar->rcvbuf->merged_len == 0) + return 0; + } + + /* return EAGAIN if no receive buffer */ + if (socket->opts & MTCP_NONBLOCK) { + if (!rcvvar->rcvbuf || rcvvar->rcvbuf->merged_len == 0) { + errno = EAGAIN; + return -1; + } + } + + SBUF_LOCK(&rcvvar->read_lock); + // h-> not sure if blocking support works with zero copy, but keeping it for now +#if BLOCKING_SUPPORT + if (!(socket->opts & MTCP_NONBLOCK)) { + while (rcvvar->rcvbuf->merged_len == 0) { + if (!cur_stream || cur_stream->state != TCP_ST_ESTABLISHED) { + SBUF_UNLOCK(&rcvvar->read_lock); + errno = EINTR; + return -1; + } + pthread_cond_wait(&rcvvar->read_cond, &rcvvar->read_lock); + } + } +#endif + + switch (flags) { + case 0: + ret = ZeroCopyToUser(mtcp, cur_stream, buf, flash_addr); + break; + default: + SBUF_UNLOCK(&rcvvar->read_lock); + ret = -1; + errno = EINVAL; + return ret; + } + + event_remaining = FALSE; + /* if there are remaining payload, generate EPOLLIN */ + /* (may due to insufficient user buffer) */ + if (socket->epoll & MTCP_EPOLLIN) { + if (!(socket->epoll & MTCP_EPOLLET) && rcvvar->rcvbuf->merged_len > 0) { + event_remaining = TRUE; + } + } + /* if waiting for close, notify it if no remaining data */ + if (cur_stream->state == TCP_ST_CLOSE_WAIT && rcvvar->rcvbuf->merged_len == 0 && ret > 0) { + event_remaining = TRUE; + } + + SBUF_UNLOCK(&rcvvar->read_lock); + + if (event_remaining) { + if (socket->epoll) { + AddEpollEvent(mtcp->ep, USR_SHADOW_EVENT_QUEUE, socket, MTCP_EPOLLIN); +#if BLOCKING_SUPPORT + } else if (!(socket->opts & MTCP_NONBLOCK)) { + if (!cur_stream->on_rcv_br_list) { + cur_stream->on_rcv_br_list = TRUE; + TAILQ_INSERT_TAIL(&mtcp->rcv_br_list, cur_stream, rcvvar->rcv_br_link); + mtcp->rcv_br_list_cnt++; + } +#endif + } + } + + TRACE_API("Stream %d: mtcp_recv() returning %d\n", cur_stream->id, ret); + return ret; +} +/*----------------------------------------------------------------------------*/ +inline ssize_t mtcp_read_zc(mctx_t mctx, int sockid, char **buf, uint64_t *flash_addr) +{ + return mtcp_recv_zc(mctx, sockid, buf, flash_addr, 0); +} +/*----------------------------------------------------------------------------*/ +ssize_t mtcp_zc_free(mctx_t mctx, uint64_t flash_addr) +{ + if (!flash_addr) { + errno = EINVAL; + return -1; + } + mtcp_manager_t mtcp = GetMTCPManager(mctx); + if (!mtcp) { + return -1; + } + + RBFreeBuff_zc(mtcp, mtcp->rbm_rcv, flash_addr); + + return 0; +} +/*----------------------------------------------------------------------------*/ +int mtcp_read_zc_batch(mctx_t mctx, int sockid, char **buf_array, uint64_t *addr_array, uint32_t *len_array, uint32_t max_pkts) +{ + mtcp_manager_t mtcp = GetMTCPManager(mctx); + if (!mtcp) + return -1; + + socket_map_t socket = &mtcp->smap[sockid]; + tcp_stream *cur_stream = socket->stream; + + if (!cur_stream || !(cur_stream->state >= TCP_ST_ESTABLISHED && cur_stream->state <= TCP_ST_CLOSE_WAIT)) { + errno = ENOTCONN; + return -1; + } + + struct tcp_recv_vars *rcvvar = cur_stream->rcvvar; + uint32_t pkts_read = 0; + + SBUF_LOCK(&rcvvar->read_lock); + + while (pkts_read < max_pkts) { + size_t len = RBGetBuff_zc(mtcp->rbm_rcv, rcvvar->rcvbuf, (void **)&buf_array[pkts_read], &addr_array[pkts_read]); + len_array[pkts_read] = len; + if (len == 0 || buf_array[pkts_read] == NULL) { + break; // No more packets in the queue + } + + rcvvar->rcv_wnd = rcvvar->rcvbuf->size - rcvvar->rcvbuf->merged_len; + pkts_read++; + } + + if (pkts_read > 0 && cur_stream->need_wnd_adv) { + if (rcvvar->rcv_wnd > cur_stream->sndvar->eff_mss) { + if (!cur_stream->sndvar->on_ackq) { + SQ_LOCK(&mtcp->ctx->ackq_lock); + cur_stream->sndvar->on_ackq = TRUE; + StreamEnqueue(mtcp->ackq, cur_stream); + SQ_UNLOCK(&mtcp->ctx->ackq_lock); + cur_stream->need_wnd_adv = FALSE; + mtcp->wakeup_flag = TRUE; + } + } + } + + SBUF_UNLOCK(&rcvvar->read_lock); + + if (pkts_read == 0 && (socket->opts & MTCP_NONBLOCK)) { + errno = EAGAIN; + return -1; + } + + return pkts_read; +} + +// /*----------------------------------------------------------------------------*/ +int mtcp_zc_free_batch(mctx_t mctx, uint64_t *addr_array, int num_pkts) +{ + mtcp_manager_t mtcp = GetMTCPManager(mctx); + + return RBFreeBuff_zc_batch(mtcp, mtcp->rbm_rcv, addr_array, num_pkts); // always successful since ring very big +} +/*----------------------------------------------------------------------------*/ +#endif /* MTCP_RX_ZERO_COPY */ + + +#ifdef MTCP_TX_ZERO_COPY +/*----------------------------------------------------------------------------*/ +static inline int ZeroCopyFromUser(mtcp_manager_t mtcp, tcp_stream *cur_stream, const char *buf, int len, uint64_t flash_addr) +{ + struct tcp_send_vars *sndvar = cur_stream->sndvar; + int ret; + + /* allocate send buffer if not exist */ + if (!sndvar->sndbuf) { + sndvar->sndbuf = SBInit(mtcp->rbm_snd, sndvar->iss + 1); + if (!sndvar->sndbuf) { + cur_stream->close_reason = TCP_NO_MEM; + /* notification may not required due to -1 return */ + errno = ENOMEM; + return -1; + } + } + + if (sndvar->sndbuf->available_zc_slots <= 0) { + errno = EAGAIN; + return -1; + } + + ret = SBPut_zc(mtcp->rbm_snd, sndvar->sndbuf, buf, len, flash_addr); + assert(ret == len); + if (ret <= 0) { + TRACE_ERROR("SBPut_zc failed. reason: %d (len: %u)\n", ret, len); + errno = EAGAIN; + return -1; + } + + if (sndvar->sndbuf->available_zc_slots <= 0) { + TRACE_SNDBUF("%u Sending buffer became full!! available_zc_slots: %u\n", cur_stream->id, + sndvar->sndbuf->available_zc_slots); + } + + return ret; +} +/*----------------------------------------------------------------------------*/ +ssize_t mtcp_get_zc_wptr(mctx_t mctx, int sockid, uint8_t **buf, uint64_t *flash_addr) +{ + mtcp_manager_t mtcp; + socket_map_t socket; + tcp_stream *cur_stream; + + *buf = NULL; + *flash_addr = 0; + + mtcp = GetMTCPManager(mctx); + if (!mtcp) { + return -1; + } + + if (sockid < 0 || sockid >= CONFIG.max_concurrency) { + TRACE_API("Socket id %d out of range.\n", sockid); + errno = EBADF; + return -1; + } + + socket = &mtcp->smap[sockid]; + if (socket->socktype == MTCP_SOCK_UNUSED) { + TRACE_API("Invalid socket id: %d\n", sockid); + errno = EBADF; + return -1; + } + + // h-> not supported for pipes yet + // if (socket->socktype == MTCP_SOCK_PIPE) { + // return PipeWrite(mctx, sockid, buf, len); + // } + + if (socket->socktype != MTCP_SOCK_STREAM) { + TRACE_API("Not an end socket. id: %d\n", sockid); + errno = ENOTSOCK; + return -1; + } + + cur_stream = socket->stream; + if (!cur_stream || !(cur_stream->state == TCP_ST_ESTABLISHED || cur_stream->state == TCP_ST_CLOSE_WAIT)) { + errno = ENOTCONN; + return -1; + } + + // ============= + struct zc_tx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_tx_ring; + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_RELAXED); + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_ACQUIRE); + + if (head == tail) { + // empty ring + errno = EAGAIN; + return -1; + } + + uint64_t addr = ring->flash_addrs[head & (ZC_TX_FREE_RING_SIZE - 1)]; + uint8_t *data = ring->flash_data[head & (ZC_TX_FREE_RING_SIZE - 1)]; + + __atomic_store_n(&ring->head, head + 1, __ATOMIC_RELEASE); + + // h-> for now, for simplicity, the header len is hardcoded for (TCP_OPT_TIMESTAMP_ENABLED = 1, TCP_OPT_SACK_ENABLED = 1,), where SACK flag is never set + uint8_t *payload_ptr = data + ETHERNET_HEADER_LEN + IP_HEADER_LEN + TCP_HEADER_LEN + (TCP_OPT_TIMESTAMP_LEN + 2); + + *buf = payload_ptr; + *flash_addr = addr; + return 0; +} + +ssize_t mtcp_write_zc(mctx_t mctx, int sockid, uint8_t *buf, int len, uint64_t flash_addr) +{ + UNUSED(mctx); + UNUSED(sockid); + UNUSED(buf); + UNUSED(len); + UNUSED(flash_addr); + + mtcp_manager_t mtcp; + socket_map_t socket; + tcp_stream *cur_stream; + struct tcp_send_vars *sndvar; + int ret; + + // app is the owner of pkt if ret <= 0 + // else mtcp thread is the owner + + mtcp = GetMTCPManager(mctx); + if (!mtcp) { + // should never happen + TRACE_ERROR("mtcp_write_zc: failed to get mtcp manager\n"); + return -1; + } + + if (sockid < 0 || sockid >= CONFIG.max_concurrency) { + TRACE_API("Socket id %d out of range.\n", sockid); + errno = EBADF; + return -1; + } + + socket = &mtcp->smap[sockid]; + if (socket->socktype == MTCP_SOCK_UNUSED) { + TRACE_API("Invalid socket id: %d\n", sockid); + errno = EBADF; + return -1; + } + + // h-> not supported for pipes yet + // if (socket->socktype == MTCP_SOCK_PIPE) { + // return PipeWrite(mctx, sockid, buf, len); + // } + + if (socket->socktype != MTCP_SOCK_STREAM) { + TRACE_API("Not an end socket. id: %d\n", sockid); + errno = ENOTSOCK; + return -1; + } + + cur_stream = socket->stream; + if (!cur_stream || !(cur_stream->state == TCP_ST_ESTABLISHED || cur_stream->state == TCP_ST_CLOSE_WAIT)) { + errno = ENOTCONN; + return -1; + } + + if (len <= 0) { + if (socket->opts & MTCP_NONBLOCK) { + errno = EAGAIN; + return -1; + } else { + return 0; + } + } + + sndvar = cur_stream->sndvar; + + SBUF_LOCK(&sndvar->write_lock); + ret = ZeroCopyFromUser(mtcp, cur_stream, buf, len, flash_addr); + // h-> blocking_support not implemented yet, skipping for now + SBUF_UNLOCK(&sndvar->write_lock); + + if (ret < 0) { + return -1; + } + + if (ret > 0 && !(sndvar->on_sendZCq || sndvar->on_sendZC_list)) { + SQ_LOCK(&mtcp->ctx->sendq_lock); + sndvar->on_sendZCq = TRUE; + StreamEnqueue(mtcp->sendZCq, cur_stream); /* this always success */ + SQ_UNLOCK(&mtcp->ctx->sendq_lock); + mtcp->wakeup_flag = TRUE; + } + + if (ret == 0 && (socket->opts & MTCP_NONBLOCK)) { + // should never happen + TRACE_ERROR("ZeroCopyFromUser failed. Should never reach here. reason: %d (len: %u)\n", ret, len); + assert(0); + ret = -1; + errno = EAGAIN; + } + + /* if there are remaining sending buffer, generate write event */ + if (sndvar->sndbuf->available_zc_slots > 0) { + if ((socket->epoll & MTCP_EPOLLOUT) && !(socket->epoll & MTCP_EPOLLET)) { + AddEpollEvent(mtcp->ep, USR_SHADOW_EVENT_QUEUE, socket, MTCP_EPOLLOUT); + // h-> no blocking support for now + } + } + + // TRACE_API("Stream %d: mtcp_write() returning %d\n", cur_stream->id, ret); + + return ret; +} + +/*----------------------------------------------------------------------------*/ +static inline int ZeroCopyFromUserBatch(mtcp_manager_t mtcp, tcp_stream *cur_stream, + uint8_t **buf_array, uint32_t *len_array, + uint64_t *addr_array, int max_pkts) +{ + struct tcp_send_vars *sndvar = cur_stream->sndvar; + int enqueued = 0; + + /* allocate send buffer if not exist */ + if (!sndvar->sndbuf) { + sndvar->sndbuf = SBInit(mtcp->rbm_snd, sndvar->iss + 1); + if (!sndvar->sndbuf) { + cur_stream->close_reason = TCP_NO_MEM; + errno = ENOMEM; + return -1; + } + } + + for (int i = 0; i < max_pkts; i++) { + assert(len_array[i] > 0); + if (len_array[i] <= 0) continue; // h-> there shouldn't be any empty pkts... + + if (sndvar->sndbuf->available_zc_slots <= 0) { + if (enqueued == 0) { + errno = EAGAIN; + return -1; + } + break; // stopping on full buffer + } + + int ret = SBPut_zc(mtcp->rbm_snd, sndvar->sndbuf, buf_array[i], len_array[i], addr_array[i]); + assert((uint32_t)ret == len_array[i]); + if (ret <= 0) { + if (enqueued == 0) { + errno = EAGAIN; + return -1; + } + break; + } + enqueued++; + } + + if (sndvar->sndbuf->available_zc_slots <= 0) { + TRACE_SNDBUF("%u Sending buffer became full!! available_zc_slots: %u\n", + cur_stream->id, sndvar->sndbuf->available_zc_slots); + } + + return enqueued; +} + +/*----------------------------------------------------------------------------*/ +ssize_t mtcp_get_zc_wptr_batch(mctx_t mctx, int sockid, uint8_t **buf_array, uint64_t *addr_array, int max_pkts) +{ + mtcp_manager_t mtcp; + socket_map_t socket; + tcp_stream *cur_stream; + + mtcp = GetMTCPManager(mctx); + if (!mtcp) { + return -1; + } + + if (sockid < 0 || sockid >= CONFIG.max_concurrency) { + TRACE_API("Socket id %d out of range.\n", sockid); + errno = EBADF; + return -1; + } + + socket = &mtcp->smap[sockid]; + if (socket->socktype == MTCP_SOCK_UNUSED) { + TRACE_API("Invalid socket id: %d\n", sockid); + errno = EBADF; + return -1; + } + + // h-> not supported for pipes yet + // if (socket->socktype == MTCP_SOCK_PIPE) { + // return PipeWrite(mctx, sockid, buf, len); + // } + + if (socket->socktype != MTCP_SOCK_STREAM) { + TRACE_API("Not an end socket. id: %d\n", sockid); + errno = ENOTSOCK; + return -1; + } + + cur_stream = socket->stream; + if (!cur_stream || !(cur_stream->state == TCP_ST_ESTABLISHED || cur_stream->state == TCP_ST_CLOSE_WAIT)) { + errno = ENOTCONN; + return -1; + } + + struct zc_tx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_tx_ring; + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_RELAXED); + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_ACQUIRE); + + if (head == tail) { + errno = EAGAIN; + return -1; + } + + int can_take = tail - head; + if (can_take > max_pkts) { + can_take = max_pkts; + } + + for (int i = 0; i < can_take; i++) { + buf_array[i] = ring->flash_data[(head + i) & (ZC_TX_FREE_RING_SIZE - 1)] + ETHERNET_HEADER_LEN + IP_HEADER_LEN + TCP_HEADER_LEN + (TCP_OPT_TIMESTAMP_LEN + 2); + addr_array[i] = ring->flash_addrs[(head + i) & (ZC_TX_FREE_RING_SIZE - 1)]; + } + + __atomic_store_n(&ring->head, head + can_take, __ATOMIC_RELEASE); + + return can_take; +} +/*----------------------------------------------------------------------------*/ +int mtcp_write_zc_batch(mctx_t mctx, int sockid, uint8_t **buf_array, + uint32_t *len_array, uint64_t *addr_array, int max_pkts) +{ + mtcp_manager_t mtcp; + socket_map_t socket; + tcp_stream *cur_stream; + struct tcp_send_vars *sndvar; + int ret; + + if (max_pkts <= 0) return 0; + + mtcp = GetMTCPManager(mctx); + if (!mtcp) { + TRACE_ERROR("mtcp_write_zc_batch: failed to get mtcp manager\n"); + return -1; + } + + if (sockid < 0 || sockid >= CONFIG.max_concurrency) { + TRACE_API("Socket id %d out of range.\n", sockid); + errno = EBADF; + return -1; + } + + socket = &mtcp->smap[sockid]; + if (socket->socktype == MTCP_SOCK_UNUSED) { + TRACE_API("Invalid socket id: %d\n", sockid); + errno = EBADF; + return -1; + } + + if (socket->socktype != MTCP_SOCK_STREAM) { + TRACE_API("Not an end socket. id: %d\n", sockid); + errno = ENOTSOCK; + return -1; + } + + cur_stream = socket->stream; + if (!cur_stream || !(cur_stream->state == TCP_ST_ESTABLISHED || cur_stream->state == TCP_ST_CLOSE_WAIT)) { + errno = ENOTCONN; + return -1; + } + + sndvar = cur_stream->sndvar; + + + SBUF_LOCK(&sndvar->write_lock); + ret = ZeroCopyFromUserBatch(mtcp, cur_stream, buf_array, len_array, addr_array, max_pkts); + SBUF_UNLOCK(&sndvar->write_lock); + + if (ret < 0) { + return -1; + } + + if (ret > 0 && !(sndvar->on_sendZCq || sndvar->on_sendZC_list)) { + SQ_LOCK(&mtcp->ctx->sendq_lock); + sndvar->on_sendZCq = TRUE; + StreamEnqueue(mtcp->sendZCq, cur_stream); + SQ_UNLOCK(&mtcp->ctx->sendq_lock); + mtcp->wakeup_flag = TRUE; + } + + if (ret == 0 && (socket->opts & MTCP_NONBLOCK)) { + TRACE_ERROR("ZeroCopyFromUserBatch failed. Should never reach here.\n"); + assert(0); + ret = -1; + errno = EAGAIN; + } + + /* if there is remaining sending buffer, generate write event */ + if (sndvar->sndbuf->available_zc_slots > 0) { + if ((socket->epoll & MTCP_EPOLLOUT) && !(socket->epoll & MTCP_EPOLLET)) { + AddEpollEvent(mtcp->ep, USR_SHADOW_EVENT_QUEUE, socket, MTCP_EPOLLOUT); + } + } + + return ret; // returns the NUMBER_OF_PKTS successfully enqueued + // h-> app still owns the unsent packets if ret < max_pkts +} + +#endif /* MTCP_TX_ZERO_COPY */ + diff --git a/lib/flash/mtcp/core.c b/lib/flash/mtcp/core.c index 9c66bae..b9c8bad 100644 --- a/lib/flash/mtcp/core.c +++ b/lib/flash/mtcp/core.c @@ -514,6 +514,14 @@ static inline void HandleApplicationCalls(mtcp_manager_t mtcp, uint32_t cur_ts) AddtoSendList(mtcp, stream); } +#ifdef MTCP_TX_ZERO_COPY + /* zero-copy send queue handling */ + while ((stream = StreamDequeue(mtcp->sendZCq))) { + stream->sndvar->on_sendZCq = FALSE; + AddtoZCSendList(mtcp, stream); + } +#endif /* MTCP_TX_ZERO_COPY */ + /* ack queue handling */ while ((stream = StreamDequeue(mtcp->ackq))) { stream->sndvar->on_ackq = FALSE; @@ -683,6 +691,11 @@ static inline void HandleApplicationCalls(mtcp_manager_t mtcp, uint32_t cur_ts) /*----------------------------------------------------------------------------*/ static inline void WritePacketsToChunks(mtcp_manager_t mtcp, uint32_t cur_ts) { + +#ifdef MTCP_TX_ZERO_COPY + mtcp->ctx->flash_ctx.tx_zc = FALSE; +#endif /* MTCP_TX_ZERO_COPY */ + int thresh = CONFIG.max_concurrency; int i; @@ -696,6 +709,15 @@ static inline void WritePacketsToChunks(mtcp_manager_t mtcp, uint32_t cur_ts) if (mtcp->g_sender->send_list_cnt) WriteTCPDataList(mtcp, mtcp->g_sender, cur_ts, thresh); + +#ifdef MTCP_TX_ZERO_COPY + if (mtcp->g_sender->sendZC_list_cnt) { + mtcp->ctx->flash_ctx.tx_zc = TRUE; + WriteTCPZCDataList(mtcp, mtcp->g_sender, cur_ts, thresh); + mtcp->ctx->flash_ctx.tx_zc = FALSE; + } +#endif + for (i = 0; i < CONFIG.eths_num; i++) { assert(mtcp->n_sender[i] != NULL); if (mtcp->n_sender[i]->control_list_cnt) @@ -704,6 +726,13 @@ static inline void WritePacketsToChunks(mtcp_manager_t mtcp, uint32_t cur_ts) WriteTCPACKList(mtcp, mtcp->n_sender[i], cur_ts, thresh); if (mtcp->n_sender[i]->send_list_cnt) WriteTCPDataList(mtcp, mtcp->n_sender[i], cur_ts, thresh); +#ifdef MTCP_TX_ZERO_COPY + if (mtcp->n_sender[i]->sendZC_list_cnt) { + mtcp->ctx->flash_ctx.tx_zc = TRUE; + WriteTCPZCDataList(mtcp, mtcp->n_sender[i], cur_ts, thresh); + mtcp->ctx->flash_ctx.tx_zc = FALSE; + } +#endif /* MTCP_TX_ZERO_COPY */ } } /*----------------------------------------------------------------------------*/ @@ -763,6 +792,49 @@ static void InterruptApplication(mtcp_manager_t mtcp) } } /*----------------------------------------------------------------------------*/ +#ifdef MTCP_RX_ZERO_COPY +// app thread produces, mtcp thread consumes +static inline void ReclaimFreeRXBuffers(mtcp_manager_t mtcp) +{ + struct zc_rx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_rx_ring; + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_RELAXED); + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_ACQUIRE); + + while (head != tail) { + uint64_t flash_addr = ring->flash_addrs[head & (ZC_RX_FREE_RING_SIZE - 1)]; + mtcp->iom->release_pkt(mtcp->ctx, 0, (uint8_t *)flash_addr, 0); + + head++; + } + + __atomic_store_n(&ring->head, head, __ATOMIC_RELEASE); +} +#endif /* MTCP_RX_ZERO_COPY */ +/*----------------------------------------------------------------------------*/ +#ifdef MTCP_TX_ZERO_COPY +static inline void ReplenishZCRing(mtcp_manager_t mtcp) { + // mtcp thread is producer + // app thread consumes using mtcp_get_zc_ptr + + struct zc_tx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_tx_ring; + + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_ACQUIRE); + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_RELAXED); + uint32_t space = ZC_TX_FREE_RING_SIZE - (tail - head); + + for (uint32_t i = 0; i < space; i++) { + uint8_t* ret = mtcp->iom->get_zc_wptr(mtcp->ctx, 0); + if (ret == NULL) break; + + ring->flash_addrs[tail & (ZC_TX_FREE_RING_SIZE - 1)] = mtcp->ctx->flash_ctx.flash_addr; + ring->flash_data[tail & (ZC_TX_FREE_RING_SIZE - 1)] = ret; + tail++; + } + + __atomic_store_n(&ring->tail, tail, __ATOMIC_RELEASE); +} +#endif /* MTCP_TX_ZERO_COPY */ +/*----------------------------------------------------------------------------*/ static void RunMainLoop(struct mtcp_thread_context *ctx) { mtcp_manager_t mtcp = ctx->mtcp_manager; @@ -792,16 +864,20 @@ static void RunMainLoop(struct mtcp_thread_context *ctx) for (i = 0; i < recv_cnt; i++) { pktbuf = mtcp->iom->get_rptr(mtcp->ctx, rx_inf, i, &len); + // IMP: get_rptr will set flash_ctx if (pktbuf != NULL) { - if (ProcessPacket(mtcp, rx_inf, ts, pktbuf, len) != TRUE) - mtcp->iom->release_pkt(mtcp->ctx, rx_inf, pktbuf, len); + if (ProcessPacket(mtcp, rx_inf, ts, pktbuf, len) != TRUE) { +#ifdef MTCP_RX_ZERO_COPY + mtcp->iom->release_pkt(mtcp->ctx, 0, (uint8_t *)mtcp->ctx->flash_ctx.flash_addr, len); +#endif + } } #ifdef NETSTAT else mtcp->nstat.rx_errors[rx_inf]++; #endif } -#ifndef DISABLE_AFXDP +#ifndef MTCP_RX_ZERO_COPY mtcp->iom->drop_pkts(mtcp->ctx); #endif } @@ -846,6 +922,13 @@ static void RunMainLoop(struct mtcp_thread_context *ctx) WritePacketsToChunks(mtcp, ts); +#ifdef MTCP_RX_ZERO_COPY + ReclaimFreeRXBuffers(mtcp); +#endif +#ifdef MTCP_TX_ZERO_COPY + ReplenishZCRing(mtcp); +#endif + /* send packets from write buffer */ /* send until tx is available */ for (tx_inf = 0; tx_inf < CONFIG.eths_num; tx_inf++) { @@ -895,6 +978,9 @@ static struct mtcp_sender *CreateMTCPSender(int ifidx) TAILQ_INIT(&sender->control_list); TAILQ_INIT(&sender->send_list); TAILQ_INIT(&sender->ack_list); +#ifdef MTCP_TX_ZERO_COPY + TAILQ_INIT(&sender->sendZC_list); +#endif sender->control_list_cnt = 0; sender->send_list_cnt = 0; @@ -1031,6 +1117,13 @@ static mtcp_manager_t InitializeMTCPManager(struct mtcp_thread_context *ctx) CTRACE_ERROR("Failed to create send queue.\n"); return NULL; } +#ifdef MTCP_TX_ZERO_COPY + mtcp->sendZCq = CreateStreamQueue(CONFIG.max_concurrency); + if (!mtcp->sendZCq) { + CTRACE_ERROR("Failed to create zero-copy send queue.\n"); + return NULL; + } +#endif mtcp->ackq = CreateStreamQueue(CONFIG.max_concurrency); if (!mtcp->ackq) { CTRACE_ERROR("Failed to create ack queue.\n"); @@ -1173,6 +1266,7 @@ static void *MTCPRunThread(void *arg) ctx->thread = pthread_self(); ctx->cpu = cpu; mtcp = ctx->mtcp_manager = InitializeMTCPManager(ctx); + ctx->flash_ctx.flash_nic_queue = mctx->flash_nic_queue; if (!mtcp) { TRACE_ERROR("Failed to initialize mtcp manager.\n"); exit(-1); @@ -1261,7 +1355,7 @@ int MTCPDPDKRunThread(void *arg) } #endif /*----------------------------------------------------------------------------*/ -mctx_t mtcp_create_context(int cpu) +mctx_t mtcp_create_context(int cpu, int flash_nic_queue) { mctx_t mctx; int ret; @@ -1291,6 +1385,7 @@ mctx_t mtcp_create_context(int cpu) return NULL; } mctx->cpu = cpu; + mctx->flash_nic_queue = flash_nic_queue; /* initialize logger */ g_logctx[cpu] = (struct log_thread_context *)calloc(1, sizeof(struct log_thread_context)); @@ -1450,6 +1545,25 @@ void mtcp_free_context(mctx_t mctx) mtcp->destroyq = NULL; } +#ifdef MTCP_TX_ZERO_COPY + if (mtcp->sendZCq) { + DestroyStreamQueue(mtcp->sendZCq); + mtcp->sendZCq = NULL; + } + + struct zc_tx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_tx_ring; + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_RELAXED); + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_ACQUIRE); + + while (head != tail) { + uint64_t flash_addr = ring->flash_addrs[head & (ZC_TX_FREE_RING_SIZE - 1)]; + mtcp->iom->release_pkt(mtcp->ctx, 0, (uint8_t *)flash_addr, 0); + head++; + } + __atomic_store_n(&ring->head, head, __ATOMIC_RELEASE); +#endif + + DestroyMTCPSender(mtcp->g_sender); for (i = 0; i < CONFIG.eths_num; i++) { DestroyMTCPSender(mtcp->n_sender[i]); diff --git a/lib/flash/mtcp/eth_in.c b/lib/flash/mtcp/eth_in.c index 2c4c028..5e9fb6c 100644 --- a/lib/flash/mtcp/eth_in.c +++ b/lib/flash/mtcp/eth_in.c @@ -67,7 +67,7 @@ int ProcessPacket(mtcp_manager_t mtcp, const int ifidx, uint32_t cur_ts, unsigne } else if (ip_proto == ETH_P_ARP) { ProcessARPPacket(mtcp, cur_ts, ifidx, pkt_data, len); - return TRUE; + return FALSE; // h-> FALSE TO FREE UP THE PACKET } else { //DumpPacket(mtcp, (char *)pkt_data, len, "??", ifidx); diff --git a/lib/flash/mtcp/eth_out.c b/lib/flash/mtcp/eth_out.c index 196450a..5d3c463 100644 --- a/lib/flash/mtcp/eth_out.c +++ b/lib/flash/mtcp/eth_out.c @@ -83,6 +83,18 @@ uint8_t *EthernetOutput(struct mtcp_manager *mtcp, uint16_t h_proto, int nif, un TRACE_INFO("No interface selected!\n"); return NULL; } +#ifdef MTCP_TX_ZERO_COPY + // if flash_ctx.data != NULL, ZC is active, give headroom for ethernet header + if (mtcp->ctx->flash_ctx.data != NULL) { + mtcp->ctx->flash_ctx.data -= ETHERNET_HEADER_LEN; + buf = mtcp->ctx->flash_ctx.data; + // enqueue the buffer now + // will send it later using mtcp->iom->send_pkts + buf = mtcp->iom->enqueue_zc_buf(mtcp->ctx, eidx, buf, mtcp->ctx->flash_ctx.flash_addr, iplen + ETHERNET_HEADER_LEN); + mtcp->ctx->flash_ctx.data = NULL; + } + else +#endif buf = mtcp->iom->get_wptr(mtcp->ctx, eidx, iplen + ETHERNET_HEADER_LEN); if (!buf) { diff --git a/lib/flash/mtcp/flash_module.c b/lib/flash/mtcp/flash_module.c index e12fa4b..2719a8c 100644 --- a/lib/flash/mtcp/flash_module.c +++ b/lib/flash/mtcp/flash_module.c @@ -40,11 +40,72 @@ struct afxdp_private_context { // private context on mTCP struct xskvec *recvvecs; struct xskvec *sendvecs; struct xskvec *dropvecs; + struct xskvec *allocvecs; uint32_t recv_index; uint32_t send_index; + uint32_t drop_index; + uint32_t alloc_index; + uint32_t alloc_count; struct pollfd fds[1]; + +#ifdef MTCP_TX_ZERO_COPY + struct xskvec *sendvecs_zc; + uint32_t send_index_zc; +#endif /* MTCP_TX_ZERO_COPY */ + + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + uint32_t descriptors_in_flight; // num of desc currently in use by the app and mtcp threads (not yet freed), i.e., not in FR, RX, TX, CR or flash__pool + uint32_t max_descriptors; // max desc allowed in flight, so that flash can function without running out of buffers (descriptors) in the UMEM +#endif + + // stats + uint64_t rx_pkts; + uint64_t tx_pkts; + + uint64_t total_rx_pkts; + uint64_t total_tx_pkts; + + uint64_t active_seconds; + + time_t last_print_time; + time_t window_start; } __attribute__((aligned(__WORDSIZE))); +/*----------------------------------------------------------------------------*/ +static void check_and_print_stats(struct afxdp_private_context *axpc) +{ + time_t now = time(NULL); + + if (now > axpc->last_print_time) { + if (axpc->rx_pkts > 0 || axpc->tx_pkts > 0) { + axpc->total_rx_pkts += axpc->rx_pkts; + axpc->total_tx_pkts += axpc->tx_pkts; + axpc->active_seconds++; + } + + axpc->rx_pkts = 0; + axpc->tx_pkts = 0; + axpc->last_print_time = now; + } + + // 10-second window + if (now - axpc->window_start >= 10) { + if (axpc->active_seconds > 0) { + double pps_rx = (double)axpc->total_rx_pkts / axpc->active_seconds; + double pps_tx = (double)axpc->total_tx_pkts / axpc->active_seconds; + + printf("[AFXDP LAST 10s] Active: %lu s | RX: %.0f pps | TX: %.0f pps\n", axpc->active_seconds, pps_rx, pps_tx); + } + + // reset window + axpc->total_rx_pkts = 0; + axpc->total_tx_pkts = 0; + axpc->active_seconds = 0; + axpc->window_start = now; + } +} + /*----------------------------------------------------------------------------*/ void afxdp_load_module(void); void afxdp_init_handle(struct mtcp_thread_context *ctxt); @@ -58,6 +119,10 @@ uint8_t *afxdp_get_rptr(struct mtcp_thread_context *ctxt, int ifidx, int index, int afxdp_select(struct mtcp_thread_context *ctxt); void afxdp_destroy_handle(struct mtcp_thread_context *ctxt); int afxdp_dev_ioctl(struct mtcp_thread_context *ctxt, int nif, int cmd, void *argp); +#ifdef MTCP_TX_ZERO_COPY +uint8_t *afxdp_get_zc_wptr(struct mtcp_thread_context *ctxt, int nif); +uint8_t *afxdp_enqueue_zc_buf(struct mtcp_thread_context *ctx, int ifidx, void *buf, uint64_t flash_addr, uint16_t len); +#endif /* MTCP_TX_ZERO_COPY */ /*----------------------------------------------------------------------------*/ void afxdp_load_module(void) @@ -93,7 +158,11 @@ void afxdp_init_handle(struct mtcp_thread_context *ctxt) argv[1] = strdup("-u"); argv[2] = strdup("0"); argv[3] = strdup("-f"); - argv[4] = strdup("0"); + + char flash_nic_queue_str[8]; + sprintf(flash_nic_queue_str, "%d", ctxt->flash_ctx.flash_nic_queue); + argv[4] = strdup(flash_nic_queue_str); + argv[5] = strdup("-t"); if (flash__parse_cmdline_args(6, argv, &axpc->cfg) < 0) @@ -158,6 +227,50 @@ void afxdp_init_handle(struct mtcp_thread_context *ctxt) free(axpc->recvvecs); goto out_cfg_close; } + axpc->drop_index = 0; + + axpc->allocvecs = calloc(axpc->cfg.xsk->batch_size, sizeof(struct xskvec)); + if (!axpc->allocvecs) { + log_error("Failed to allocate alloc xskvecs array"); + free(axpc->sendvecs); + free(axpc->recvvecs); + free(axpc->dropvecs); + goto out_cfg_close; + } + +#ifdef MTCP_TX_ZERO_COPY + axpc->sendvecs_zc = calloc(axpc->cfg.xsk->batch_size, sizeof(struct xskvec)); + if (!axpc->sendvecs_zc) { + log_error("Failed to allocate sendvecs_zc array"); + free(axpc->sendvecs); + free(axpc->recvvecs); + free(axpc->dropvecs); + free(axpc->allocvecs); + goto out_cfg_close; + } + axpc->send_index_zc = 0; +#endif /* MTCP_TX_ZERO_COPY */ + + axpc->alloc_index = 0; + axpc->alloc_count = 0; + + // Initialize statistics + axpc->rx_pkts = 0; + axpc->tx_pkts = 0; + + axpc->total_rx_pkts = 0; + axpc->total_tx_pkts = 0; + + axpc->active_seconds = 0; + + axpc->last_print_time = time(NULL); + axpc->window_start = axpc->last_print_time; + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + axpc->max_descriptors = 1024; // d-> hardcoded for now, but should be calculated based on the UMEM size and frame size + axpc->descriptors_in_flight = 0; + printf("Max Descriptors in Flight: %u\n", axpc->max_descriptors); +#endif return; @@ -189,7 +302,31 @@ int32_t afxdp_recv_pkts(struct mtcp_thread_context *ctxt, int ifidx) if (!(ret == 1 || ret == -2)) return 0; - nrecv = flash__recvmsg(&axpc->cfg, xsk, axpc->recvvecs, axpc->cfg.xsk->batch_size); + uint32_t to_recv = axpc->cfg.xsk->batch_size; +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + // throttling, to avoid deadlocks due to lack of umem buffers + if (axpc->descriptors_in_flight >= axpc->max_descriptors) { + return 0; + } + if (axpc->descriptors_in_flight + to_recv > axpc->max_descriptors) { + to_recv = axpc->max_descriptors - axpc->descriptors_in_flight; + } +#endif + + nrecv = flash__recvmsg(&axpc->cfg, xsk, axpc->recvvecs, to_recv); + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + axpc->descriptors_in_flight += nrecv; +#endif + +// h-> for zero copy (RB buffer library / application library is expected to free the descriptors) +#ifdef MTCP_RX_ZERO_COPY + axpc->recv_index = 0; +#endif + + axpc->rx_pkts += nrecv; + + check_and_print_stats(axpc); return nrecv; } @@ -210,23 +347,46 @@ uint8_t *afxdp_get_rptr(struct mtcp_thread_context *ctxt, int ifidx, int index, uint8_t *pktbuf = axpc->recvvecs[axpc->recv_index].data; *len = axpc->recvvecs[axpc->recv_index].len; - axpc->dropvecs[axpc->recv_index] = axpc->recvvecs[axpc->recv_index]; +#ifdef MTCP_RX_ZERO_COPY + // h-> this flash_addr will be needed by mtcp/app thread to free up the buffer + ctxt->flash_ctx.flash_addr = axpc->recvvecs[axpc->recv_index].addr; +#endif /* MTCP_RX_ZERO_COPY */ + +#ifdef MTCP_FLASH_ID_TRAILER + // flash_ids shaved off + // [pkt][src_flash_id][dst_flash_id] + // pkt's destination is me + // pkt's source is the other flash_id which we need to store in the context + ctxt->flash_ctx.dst_flash_id = *(uint8_t *)(pktbuf + *len - 2); + + *len -= 2; +#endif /* MTCP_FLASH_ID_TRAILER */ + axpc->recv_index++; return pktbuf; } /*----------------------------------------------------------------------------*/ +// h-> drops everything in recvvecs +// not invoked in mtcp-zero-copy void afxdp_drop_pkts(struct mtcp_thread_context *ctxt) { struct afxdp_private_context *axpc; axpc = (struct afxdp_private_context *)ctxt->io_private_context; - if (flash__dropmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->dropvecs, axpc->recv_index) != axpc->recv_index) { + if (flash__dropmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->recvvecs, axpc->recv_index) != axpc->recv_index) { log_error("Failed to drop messages"); axpc->recv_index = 0; return; } +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + if (axpc->descriptors_in_flight >= axpc->recv_index) { + axpc->descriptors_in_flight -= axpc->recv_index; + } else { + printf("Error: Attempted to drop more packets than descriptors in flight!\n"); + } +#endif axpc->recv_index = 0; } @@ -238,6 +398,34 @@ void afxdp_release_pkt(struct mtcp_thread_context *ctxt, int ifidx, unsigned cha (void)pkt_data; // d-> unused parameter (void)len; // d-> unused parameter /* not needed - drop packets is handled seperately */ + + // h-> added for rx zero copy path + struct afxdp_private_context *axpc; + axpc = (struct afxdp_private_context *)ctxt->io_private_context; + + // batching the drops + axpc->dropvecs[axpc->drop_index].addr = (uint64_t)pkt_data; + axpc->drop_index++; + + // flush if full + if (axpc->drop_index >= axpc->cfg.xsk->batch_size) { +#if defined(MTCP_TX_ZERO_COPY) + for (uint32_t i = 0; i < axpc->drop_index; i++) { + uint32_t frame_index = axpc->dropvecs[i].addr / axpc->cfg.umem->frame_size; + axpc->cfg.zc_tracker[frame_index] = 0; // mark the frame as free in the zero-copy tracker + } +#endif + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + if (axpc->descriptors_in_flight >= axpc->drop_index) { + axpc->descriptors_in_flight -= axpc->drop_index; + } else { + printf("Error: Attempted to release more packets than descriptors in flight!\n"); + } +#endif + flash__dropmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->dropvecs, axpc->drop_index); + axpc->drop_index = 0; + } } /*----------------------------------------------------------------------------*/ @@ -253,18 +441,53 @@ uint8_t *afxdp_get_wptr(struct mtcp_thread_context *ctxt, int nif, uint16_t pkts return NULL; } - struct xskvec tmpvec; - size_t ret = flash__allocmsg(&axpc->cfg, axpc->nf->thread[0]->socket, &tmpvec, 1); - if (ret == 0) { - return NULL; + // batched allocation + if (axpc->alloc_index >= axpc->alloc_count) { + uint32_t to_alloc = axpc->cfg.xsk->batch_size; +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + // h-> this is to avoid deadlocks due to lack of umem buffers + if (axpc->descriptors_in_flight >= axpc->max_descriptors) { + return NULL; + } + if (axpc->descriptors_in_flight + to_alloc > axpc->max_descriptors) { + to_alloc = axpc->max_descriptors - axpc->descriptors_in_flight; + } +#endif + + axpc->alloc_count = flash__allocmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->allocvecs, to_alloc); + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + axpc->descriptors_in_flight += axpc->alloc_count; +#endif + + axpc->alloc_index = 0; + if (axpc->alloc_count == 0) { + return NULL; + } } - uint8_t *pktbuf = tmpvec.data; + struct xskvec *tmpvec = &axpc->allocvecs[axpc->alloc_index++]; + + uint8_t *pktbuf = tmpvec->data; + uint32_t options = 0; + +#ifdef MTCP_FLASH_ID_TRAILER + // reserve 2 bytes at the end of the packet for flash ids (src and dst) + // [pkt][src_flash_id][dst_flash_id] + uint8_t src_flash_id = axpc->cfg.nf_id; + uint8_t dst_flash_id = ctxt->flash_ctx.dst_flash_id; + ctxt->flash_ctx.dst_flash_id = -1; + + *(uint8_t *)(pktbuf + pktsize) = src_flash_id; + *(uint8_t *)(pktbuf + pktsize + 1) = dst_flash_id; + pktsize += 2; + options = (dst_flash_id << 16) | (options & 0xFFFF); +#endif /* MTCP_FLASH_ID_TRAILER */ axpc->sendvecs[axpc->send_index].data = pktbuf; axpc->sendvecs[axpc->send_index].len = pktsize; - axpc->sendvecs[axpc->send_index].addr = tmpvec.addr; - axpc->sendvecs[axpc->send_index++].options = 0; + axpc->sendvecs[axpc->send_index].addr = tmpvec->addr; + axpc->sendvecs[axpc->send_index++].options = options; return pktbuf; } @@ -276,13 +499,56 @@ int afxdp_send_pkts(struct mtcp_thread_context *ctxt, int nif) struct afxdp_private_context *axpc; axpc = (struct afxdp_private_context *)ctxt->io_private_context; + // drop any pkts + if (axpc->drop_index > 0) { +#if defined(MTCP_TX_ZERO_COPY) + for (uint32_t i = 0; i < axpc->drop_index; i++) { + uint32_t frame_index = axpc->dropvecs[i].addr / axpc->cfg.umem->frame_size; + axpc->cfg.zc_tracker[frame_index] = 0; + } +#endif + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + if (axpc->descriptors_in_flight >= axpc->drop_index) { + axpc->descriptors_in_flight -= axpc->drop_index; + } else { + printf("Error: Attempted to release more packets than descriptors in flight!\n"); + } +#endif + + flash__dropmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->dropvecs, axpc->drop_index); + axpc->drop_index = 0; + } + if (flash__sendmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->sendvecs, axpc->send_index) != axpc->send_index) { log_error("Failed to send messages"); axpc->send_index = 0; return -1; } + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + if (axpc->descriptors_in_flight >= axpc->send_index) { + axpc->descriptors_in_flight -= axpc->send_index; + } else { + printf("Warning: Attempted to send packets when no descriptors are in flight!\n"); + } +#endif + + axpc->tx_pkts += axpc->send_index; axpc->send_index = 0; + +#ifdef MTCP_TX_ZERO_COPY + if (flash__sendmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->sendvecs_zc, axpc->send_index_zc) != axpc->send_index_zc) { + log_error("Failed to send zero-copy messages"); + axpc->send_index_zc = 0; + return -1; + } + axpc->tx_pkts += axpc->send_index_zc; + axpc->send_index_zc = 0; +#endif + + check_and_print_stats(axpc); return 1; } @@ -304,10 +570,93 @@ void afxdp_destroy_handle(struct mtcp_thread_context *ctxt) free(axpc->recvvecs); free(axpc->sendvecs); free(axpc->dropvecs); + free(axpc->allocvecs); +#ifdef MTCP_TX_ZERO_COPY + free(axpc->sendvecs_zc); +#endif flash__xsk_close(&axpc->cfg, axpc->nf); free(axpc); } +#ifdef MTCP_TX_ZERO_COPY +/*----------------------------------------------------------------------------*/ +uint8_t *afxdp_get_zc_wptr(struct mtcp_thread_context *ctxt, int nif) +{ + (void)nif; // d-> unused parameter + struct afxdp_private_context *axpc; + axpc = (struct afxdp_private_context *)ctxt->io_private_context; + + // refill allocated buffers if needed + uint32_t to_alloc = axpc->cfg.xsk->batch_size; + if (axpc->alloc_index >= axpc->alloc_count) { +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) // h-> always true, but keeping for clarity/consistency + if (axpc->descriptors_in_flight >= axpc->max_descriptors) { + return NULL; + } + if (axpc->descriptors_in_flight + to_alloc > axpc->max_descriptors) { + to_alloc = axpc->max_descriptors - axpc->descriptors_in_flight; + } +#endif + + axpc->alloc_count = flash__allocmsg(&axpc->cfg, axpc->nf->thread[0]->socket, axpc->allocvecs, to_alloc); + axpc->alloc_index = 0; + if (axpc->alloc_count == 0) { + return NULL; + } +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) // h-> always true, but keeping for clarity/consistency + axpc->descriptors_in_flight += axpc->alloc_count; +#endif + } + + + struct xskvec *tmpvec = &axpc->allocvecs[axpc->alloc_index++]; + uint8_t *pktbuf = tmpvec->data; + + // used by mtcp/app thread to free up the buffer + ctxt->flash_ctx.flash_addr = tmpvec->addr; + + return pktbuf; +} + +uint8_t *afxdp_enqueue_zc_buf(struct mtcp_thread_context *ctx, int ifidx, void *buf, uint64_t flash_addr, uint16_t len) +{ + UNUSED(ifidx); + struct afxdp_private_context *axpc; + axpc = (struct afxdp_private_context *)ctx->io_private_context; + + if (axpc->send_index_zc >= axpc->cfg.xsk->batch_size) { + return NULL; + } + + axpc->sendvecs_zc[axpc->send_index_zc].data = (uint8_t *)buf; + axpc->sendvecs_zc[axpc->send_index_zc].len = len; + axpc->sendvecs_zc[axpc->send_index_zc].addr = flash_addr; + axpc->sendvecs_zc[axpc->send_index_zc].options = 0; + + // for chaining, we can use the options field to pass the dst_flash_id +#ifdef MTCP_FLASH_ID_TRAILER + uint8_t dst_flash_id = ctx->flash_ctx.dst_flash_id; + ctx->flash_ctx.dst_flash_id = -1; // reset after using + + uint8_t src_flash_id = axpc->cfg.nf_id; + assert(dst_flash_id != axpc->cfg.nf_id); + + axpc->sendvecs_zc[axpc->send_index_zc].options = (dst_flash_id << 16) | + (axpc->sendvecs_zc[axpc->send_index_zc].options & 0xFFFF); +#endif + +#if defined(MTCP_TX_ZERO_COPY) + uint32_t frame_index = flash_addr / axpc->cfg.umem->frame_size; + axpc->cfg.zc_tracker[frame_index] = 1; +#endif + + axpc->send_index_zc++; + + return (uint8_t *)buf; +} + +#endif /* MTCP_TX_ZERO_COPY */ + /*----------------------------------------------------------------------------*/ io_module_func afxdp_module_func = { .load_module = afxdp_load_module, .init_handle = afxdp_init_handle, @@ -320,6 +669,10 @@ io_module_func afxdp_module_func = { .load_module = afxdp_load_module, .send_pkts = afxdp_send_pkts, .select = afxdp_select, .destroy_handle = afxdp_destroy_handle, +#ifdef MTCP_TX_ZERO_COPY + .get_zc_wptr = afxdp_get_zc_wptr, + .enqueue_zc_buf = afxdp_enqueue_zc_buf, +#endif /* MTCP_TX_ZERO_COPY */ .dev_ioctl = NULL }; /*----------------------------------------------------------------------------*/ #endif /* !DISABLE_AFXDP */ diff --git a/lib/flash/mtcp/include/io_module.h b/lib/flash/mtcp/include/io_module.h index 216e26b..b43a1ac 100644 --- a/lib/flash/mtcp/include/io_module.h +++ b/lib/flash/mtcp/include/io_module.h @@ -98,6 +98,10 @@ typedef struct io_module_func { int32_t (*dev_ioctl)(struct mtcp_thread_context *ctx, int nif, int cmd, void *argp); #ifndef DISABLE_AFXDP void (*drop_pkts)(struct mtcp_thread_context *ctxt); +#ifdef MTCP_TX_ZERO_COPY + uint8_t *(*get_zc_wptr)(struct mtcp_thread_context *ctxt, int nif); + uint8_t *(*enqueue_zc_buf)(struct mtcp_thread_context *ctx, int ifidx, void* buf, uint64_t flash_addr, uint16_t len); +#endif #endif } io_module_func __attribute__((aligned(__WORDSIZE))); /*----------------------------------------------------------------------------*/ diff --git a/lib/flash/mtcp/include/mtcp.h b/lib/flash/mtcp/include/mtcp.h index fb7bf37..e8aad6c 100644 --- a/lib/flash/mtcp/include/mtcp.h +++ b/lib/flash/mtcp/include/mtcp.h @@ -227,6 +227,9 @@ struct mtcp_config { /*----------------------------------------------------------------------------*/ struct mtcp_context { int cpu; + + // flash args + int flash_nic_queue; }; /*----------------------------------------------------------------------------*/ struct mtcp_sender { @@ -240,6 +243,11 @@ struct mtcp_sender { int control_list_cnt; int send_list_cnt; int ack_list_cnt; + +#ifdef MTCP_TX_ZERO_COPY + TAILQ_HEAD(sendZC_head, tcp_stream) sendZC_list; + int sendZC_list_cnt; +#endif }; /*----------------------------------------------------------------------------*/ struct mtcp_manager { @@ -281,6 +289,9 @@ struct mtcp_manager { stream_queue_t connectq; /* streams need to connect */ stream_queue_t sendq; /* streams need to send data */ +#ifdef MTCP_TX_ZERO_COPY + stream_queue_t sendZCq; /* streams need to send data */ +#endif stream_queue_t ackq; /* streams need to send ack */ stream_queue_t closeq; /* streams need to close */ @@ -339,6 +350,51 @@ typedef struct mtcp_manager *mtcp_manager_t; /*----------------------------------------------------------------------------*/ mtcp_manager_t GetMTCPManager(mctx_t mctx); /*----------------------------------------------------------------------------*/ + +#ifdef MTCP_RX_ZERO_COPY +// h-> buffer so big => RBFreeBuff_zc always succeeds +#define ZC_RX_FREE_RING_SIZE 2048 + +struct zc_rx_free_ring { + uint64_t flash_addrs[ZC_RX_FREE_RING_SIZE]; + _Atomic uint32_t head; // mtcp thread = consumer + _Atomic uint32_t tail; // nic thread = producer +}; + +#endif /* MTCP_RX_ZERO_COPY */ + +#ifdef MTCP_TX_ZERO_COPY +#define ZC_TX_FREE_RING_SIZE 128 +struct zc_tx_free_ring { + uint64_t flash_addrs[ZC_TX_FREE_RING_SIZE]; + uint8_t* flash_data[ZC_TX_FREE_RING_SIZE]; + _Atomic uint32_t head; /* App thread reads from here */ + _Atomic uint32_t tail; /* TCP thread writes to here */ +}; + +#endif + +struct flash_context { + int flash_nic_queue; +#ifdef MTCP_FLASH_ID_TRAILER + uint8_t dst_flash_id; +#endif + +#ifdef MTCP_RX_ZERO_COPY + struct zc_rx_free_ring zc_rx_ring; +#endif + +#if defined(MTCP_RX_ZERO_COPY) || defined(MTCP_TX_ZERO_COPY) + uint64_t flash_addr; +#endif + +#ifdef MTCP_TX_ZERO_COPY + int8_t tx_zc; + uint8_t* data; + struct zc_tx_free_ring zc_tx_ring; +#endif +}; + struct mtcp_thread_context { int cpu; pthread_t thread; @@ -351,6 +407,9 @@ struct mtcp_thread_context { pthread_mutex_t flow_pool_lock; pthread_mutex_t socket_pool_lock; + // flash vars + struct flash_context flash_ctx; + #if LOCK_STREAM_QUEUE #if USE_SPIN_LOCK pthread_spinlock_t connect_lock; diff --git a/lib/flash/mtcp/include/mtcp_api.h b/lib/flash/mtcp/include/mtcp_api.h index ecfce3f..f805c24 100644 --- a/lib/flash/mtcp/include/mtcp_api.h +++ b/lib/flash/mtcp/include/mtcp_api.h @@ -81,7 +81,7 @@ int mtcp_setconf(const struct mtcp_conf *conf); int mtcp_core_affinitize(int cpu); -mctx_t mtcp_create_context(int cpu); +mctx_t mtcp_create_context(int cpu, int flash_nic_queue); void mtcp_destroy_context(mctx_t mctx); @@ -111,7 +111,7 @@ int mtcp_accept(mctx_t mctx, int sockid, struct sockaddr *addr, socklen_t *addrl int mtcp_init_rss(mctx_t mctx, in_addr_t saddr_base, int num_addr, in_addr_t daddr, in_addr_t dport); -int mtcp_connect(mctx_t mctx, int sockid, const struct sockaddr *addr, socklen_t addrlen); +int mtcp_connect(mctx_t mctx, int sockid, const struct sockaddr *addr, socklen_t addrlen, uint8_t dst_flash_id); int mtcp_close(mctx_t mctx, int sockid); @@ -129,11 +129,29 @@ inline ssize_t mtcp_read(mctx_t mctx, int sockid, char *buf, size_t len); ssize_t mtcp_recv(mctx_t mctx, int sockid, char *buf, size_t len, int flags); +#ifdef MTCP_RX_ZERO_COPY +inline ssize_t mtcp_read_zc(mctx_t mctx, int sockid, char **buf, uint64_t *flash_addr); +ssize_t mtcp_recv_zc(mctx_t mctx, int sockid, char **buf, uint64_t *flash_addr, int flags); +ssize_t mtcp_zc_free(mctx_t mctx, uint64_t flash_addr); +int mtcp_read_zc_batch(mctx_t mctx, int sockid, char **buf_array, uint64_t *addr_array, uint32_t *len_array, uint32_t max_pkts); +int mtcp_zc_free_batch(mctx_t mctx, uint64_t *addr_array, int num_pkts); +#endif + + /* readv should work in atomic */ int mtcp_readv(mctx_t mctx, int sockid, const struct iovec *iov, int numIOV); ssize_t mtcp_write(mctx_t mctx, int sockid, const char *buf, size_t len); + +#ifdef MTCP_TX_ZERO_COPY +ssize_t mtcp_get_zc_wptr(mctx_t mctx, int sockid, uint8_t **buf, uint64_t *flash_addr); +ssize_t mtcp_get_zc_wptr_batch(mctx_t mctx, int sockid, uint8_t **buf_array, uint64_t *addr_array, int max_pkts); +ssize_t mtcp_write_zc(mctx_t mctx, int sockid, uint8_t *buf, int len, uint64_t flash_addr); +int mtcp_write_zc_batch(mctx_t mctx, int sockid, uint8_t **buf_array, uint32_t *len_array, uint64_t *addr_array, int max_pkts); +#endif + + /* writev should work in atomic */ int mtcp_writev(mctx_t mctx, int sockid, const struct iovec *iov, int numIOV); diff --git a/lib/flash/mtcp/include/tcp_out.h b/lib/flash/mtcp/include/tcp_out.h index 973b49b..6490f6e 100644 --- a/lib/flash/mtcp/include/tcp_out.h +++ b/lib/flash/mtcp/include/tcp_out.h @@ -64,4 +64,14 @@ extern inline void EnqueueACK(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint3 extern inline void DumpControlList(mtcp_manager_t mtcp, struct mtcp_sender *sender); +#ifdef MTCP_TX_ZERO_COPY + +extern inline int WriteTCPZCDataList(mtcp_manager_t mtcp, struct mtcp_sender *sender, uint32_t cur_ts, int thresh); + +extern inline void AddtoZCSendList(mtcp_manager_t mtcp, tcp_stream *cur_stream); + +extern inline void RemoveFromZCSendList(mtcp_manager_t mtcp, tcp_stream *cur_stream); + +#endif + #endif /* TCP_OUT_H */ diff --git a/lib/flash/mtcp/include/tcp_rb_frag_queue.h b/lib/flash/mtcp/include/tcp_rb_frag_queue.h index 1aa85f7..dd8e580 100644 --- a/lib/flash/mtcp/include/tcp_rb_frag_queue.h +++ b/lib/flash/mtcp/include/tcp_rb_frag_queue.h @@ -46,4 +46,13 @@ int RBFragEnqueue(rb_frag_queue_t rb_fragq, struct fragment_ctx *frag); struct fragment_ctx *RBFragDequeue(rb_frag_queue_t rb_fragq); /*---------------------------------------------------------------------------*/ +#ifdef MTCP_RX_ZERO_COPY +typedef struct rb_frag_queue_zc *rb_frag_queue_zc_t; +rb_frag_queue_zc_t CreateRBFragQueue_zc(int capacity); +void DestroyRBFragQueue_zc(rb_frag_queue_zc_t q); +int RBFragEnqueue_zc(rb_frag_queue_zc_t q, struct fragment_ctx_zc *frag); +struct fragment_ctx_zc *RBFragDequeue_zc(rb_frag_queue_zc_t q); +#endif + + #endif /* TCP_RB_FRAG_QUEUE */ diff --git a/lib/flash/mtcp/include/tcp_ring_buffer.h b/lib/flash/mtcp/include/tcp_ring_buffer.h index 9f7d2ac..0ae3017 100644 --- a/lib/flash/mtcp/include/tcp_ring_buffer.h +++ b/lib/flash/mtcp/include/tcp_ring_buffer.h @@ -78,6 +78,9 @@ struct tcp_ring_buffer { uint32_t init_seq; struct fragment_ctx *fctx; +#ifdef MTCP_RX_ZERO_COPY + struct fragment_ctx_zc *fctx_zc, *fctx_zc_tail; +#endif /* MTCP_RX_ZERO_COPY */ }; /*----------------------------------------------------------------------------*/ uint32_t RBGetCurnum(rb_manager_t rbm); @@ -88,7 +91,7 @@ void RBPrintHex(struct tcp_ring_buffer *buff); rb_manager_t RBManagerCreate(mtcp_manager_t mtcp, size_t chunk_size, uint32_t cnum); /*----------------------------------------------------------------------------*/ struct tcp_ring_buffer *RBInit(rb_manager_t rbm, uint32_t init_seq); -void RBFree(rb_manager_t rbm, struct tcp_ring_buffer *buff); +void RBFree(mtcp_manager_t mtcp, rb_manager_t rbm, struct tcp_ring_buffer *buff); uint32_t RBIsDanger(rb_manager_t rbm); /*----------------------------------------------------------------------------*/ /* data manupulation functions */ @@ -97,4 +100,36 @@ size_t RBGet(rb_manager_t rbm, struct tcp_ring_buffer *buff, size_t len); size_t RBRemove(rb_manager_t rbm, struct tcp_ring_buffer *buff, size_t len, int option); /*----------------------------------------------------------------------------*/ +/*----------------------------------------------------------------------------*/ +#ifdef MTCP_RX_ZERO_COPY +struct fragment_ctx_zc { + uint32_t seq; + uint32_t len; + struct fragment_ctx_zc *next; + + uint64_t flash_addr; // pointer passed to release_pkt() + unsigned char *payload; // current readable start + uint32_t payload_len; // current readable bytes +}; + +/* zero copy api */ +size_t RBAvailable_zc(struct tcp_ring_buffer *buff); +int RBPut_zc(mtcp_manager_t mtcp, rb_manager_t rbm, struct tcp_ring_buffer *buff, + uint64_t flash_addr, + unsigned char *payload, + uint32_t len, + uint32_t cur_seq); + +size_t RBRead_zc(rb_manager_t rbm, struct tcp_ring_buffer *buff, + void *dst, size_t len); + +size_t RBRemove_zc(mtcp_manager_t mtcp, rb_manager_t rbm, struct tcp_ring_buffer *buff, + size_t len, int option); + +size_t RBGetBuff_zc(rb_manager_t rbm, struct tcp_ring_buffer *buff, void **buf, uint64_t *flash_addr); +void RBFreeBuff_zc(mtcp_manager_t mtcp, rb_manager_t rbm, uint64_t flash_addr); +int RBFreeBuff_zc_batch(mtcp_manager_t mtcp, rb_manager_t rbm, uint64_t *addr_array, int num_pkts); + +#endif /* MTCP_RX_ZERO_COPY */ + #endif diff --git a/lib/flash/mtcp/include/tcp_send_buffer.h b/lib/flash/mtcp/include/tcp_send_buffer.h index 58dff83..fadb9f1 100644 --- a/lib/flash/mtcp/include/tcp_send_buffer.h +++ b/lib/flash/mtcp/include/tcp_send_buffer.h @@ -39,6 +39,16 @@ typedef struct sb_manager *sb_manager_t; typedef struct mtcp_manager *mtcp_manager_t; /*----------------------------------------------------------------------------*/ + +#ifdef MTCP_TX_ZERO_COPY +#define MAX_TX_SLOTS 32 +struct data_zc { + uint64_t flash_addr; + size_t len; + size_t acked_len; + uint8_t *payload; +}; +#endif /* MTCP_TX_ZERO_COPY */ struct tcp_send_buffer { unsigned char *data; unsigned char *head; @@ -51,6 +61,13 @@ struct tcp_send_buffer { uint32_t head_seq; uint32_t init_seq; + +#ifdef MTCP_TX_ZERO_COPY + // need to check overflow before inserting data into the buffer + struct data_zc zc_buf[MAX_TX_SLOTS]; + int q_front, q_back; + uint32_t available_zc_slots; +#endif /* MTCP_TX_ZERO_COPY */ }; /*----------------------------------------------------------------------------*/ uint32_t SBGetCurnum(sb_manager_t sbm); @@ -66,4 +83,10 @@ size_t SBPut(sb_manager_t sbm, struct tcp_send_buffer *buf, const void *data, si size_t SBRemove(sb_manager_t sbm, struct tcp_send_buffer *buf, size_t len); /*----------------------------------------------------------------------------*/ +#ifdef MTCP_TX_ZERO_COPY +size_t SBRemove_zc(mtcp_manager_t mtcp, struct tcp_send_buffer *buf, size_t acked_len); +size_t SBPut_zc(sb_manager_t sbm, struct tcp_send_buffer *buf, const void *data, size_t len, uint64_t flash_addr); +void SBClear_zc(mtcp_manager_t mtcp, struct tcp_send_buffer *buf); +#endif /* MTCP_TX_ZERO_COPY */ + #endif /* TCP_SEND_BUFFER_H */ diff --git a/lib/flash/mtcp/include/tcp_stream.h b/lib/flash/mtcp/include/tcp_stream.h index 27ebf57..cfda461 100644 --- a/lib/flash/mtcp/include/tcp_stream.h +++ b/lib/flash/mtcp/include/tcp_stream.h @@ -191,6 +191,12 @@ struct tcp_send_vars { TAILQ_ENTRY(tcp_stream) snd_br_link; pthread_cond_t write_cond; #endif + +#ifdef MTCP_TX_ZERO_COPY + uint8_t on_sendZC_list; + uint8_t on_sendZCq; + TAILQ_ENTRY(tcp_stream) sendZC_link; +#endif /* MTCP_TX_ZERO_COPY */ }; typedef struct tcp_stream { @@ -203,6 +209,11 @@ typedef struct tcp_stream { uint16_t sport; /* in network order */ uint16_t dport; /* in network order */ +#ifdef MTCP_FLASH_ID_TRAILER + uint8_t dst_flash_id; /* destination flash id */ + // h-> source flash_id is already there in flash_module.c +#endif + uint8_t state; /* tcp state */ uint8_t close_reason; /* close reason */ uint8_t on_hash_table; diff --git a/lib/flash/mtcp/ip_in.c b/lib/flash/mtcp/ip_in.c index 5072f53..3518db0 100644 --- a/lib/flash/mtcp/ip_in.c +++ b/lib/flash/mtcp/ip_in.c @@ -82,7 +82,8 @@ inline int ProcessIPv4Packet(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifi case IPPROTO_TCP: return ProcessTCPPacket(mtcp, cur_ts, ifidx, iph, ip_len); case IPPROTO_ICMP: - return ProcessICMPPacket(mtcp, iph, ip_len); + ProcessICMPPacket(mtcp, iph, ip_len); + return FALSE; // false to free up the packet in core.c for RX ZC. default: /* currently drop other protocols */ return FALSE; diff --git a/lib/flash/mtcp/ip_out.c b/lib/flash/mtcp/ip_out.c index 04af790..e1842f7 100644 --- a/lib/flash/mtcp/ip_out.c +++ b/lib/flash/mtcp/ip_out.c @@ -144,6 +144,10 @@ uint8_t *IPOutput(struct mtcp_manager *mtcp, tcp_stream *stream, uint16_t tcplen stream->is_external = is_external; } +#ifdef MTCP_FLASH_ID_TRAILER + mtcp->ctx->flash_ctx.dst_flash_id = stream->dst_flash_id; +#endif + haddr = GetDestinationHWaddr(stream->daddr, stream->is_external); if (!haddr) { #if 0 @@ -159,6 +163,14 @@ uint8_t *IPOutput(struct mtcp_manager *mtcp, tcp_stream *stream, uint16_t tcplen return NULL; } + + // if flash_ctx.data != NULL, ZC is active, give headroom for ip header +#ifdef MTCP_TX_ZERO_COPY + if (mtcp->ctx->flash_ctx.data != NULL) { + mtcp->ctx->flash_ctx.data = mtcp->ctx->flash_ctx.data - IP_HEADER_LEN; + } +#endif + iph = (struct iphdr *)EthernetOutput(mtcp, ETH_P_IP, stream->sndvar->nif_out, haddr, tcplen + IP_HEADER_LEN); if (!iph) { return NULL; diff --git a/lib/flash/mtcp/tcp_in.c b/lib/flash/mtcp/tcp_in.c index 47604e3..d448eac 100644 --- a/lib/flash/mtcp/tcp_in.c +++ b/lib/flash/mtcp/tcp_in.c @@ -464,6 +464,12 @@ static inline void ProcessACK(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint3 TRACE_DBG("Exceed MAX_RTX.\n"); } +#ifdef MTCP_TX_ZERO_COPY + if (sndvar->sndbuf->available_zc_slots < MAX_TX_SLOTS || sndvar->sndbuf->q_front != -1) + AddtoZCSendList(mtcp, cur_stream); + else +#endif + // standard fallback AddtoSendList(mtcp, cur_stream); } else if (cur_stream->rcvvar->dup_acks > 3) { @@ -504,6 +510,17 @@ static inline void ProcessACK(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint3 cur_stream->snd_nxt = ack_seq; TRACE_DBG("Sending again..., ack_seq=%u sndlen=%u cwnd=%u\n", ack_seq - sndvar->iss, sndvar->sndbuf->len, sndvar->cwnd / sndvar->mss); +#ifdef MTCP_TX_ZERO_COPY + // this condition tells me whether to use zero copy or not. If the sndbuf is empty, we can remove it from the list. If the sndbuf is not empty, we can add it to the list. + if (sndvar->sndbuf->available_zc_slots < MAX_TX_SLOTS || sndvar->sndbuf->q_front != -1) { + if (sndvar->sndbuf->len == 0) { + RemoveFromZCSendList(mtcp, cur_stream); + } else { + AddtoZCSendList(mtcp, cur_stream); + } + } else +#endif + // standard fallback if (sndvar->sndbuf->len == 0) { RemoveFromSendList(mtcp, cur_stream); } else { @@ -566,6 +583,14 @@ static inline void ProcessACK(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint3 perror("ProcessACK: write_lock blocked\n"); assert(0); } +#ifdef MTCP_TX_ZERO_COPY + uint32_t zc_slots_prev = 0; + if (sndvar->sndbuf->available_zc_slots < MAX_TX_SLOTS || sndvar->sndbuf->q_front != -1) { + zc_slots_prev = sndvar->sndbuf->available_zc_slots; + ret = SBRemove_zc(mtcp, sndvar->sndbuf, rmlen); + } else +#endif + // Standard fallback ret = SBRemove(mtcp->rbm_snd, sndvar->sndbuf, rmlen); sndvar->snd_una = ack_seq; snd_wnd_prev = sndvar->snd_wnd; @@ -574,7 +599,11 @@ static inline void ProcessACK(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint3 /* If there was no available sending window */ /* notify the newly available window to application */ #if SELECTIVE_WRITE_EVENT_NOTIFY +#ifdef MTCP_TX_ZERO_COPY + if (snd_wnd_prev <= 0 || (zc_slots_prev == 0 && sndvar->sndbuf->available_zc_slots > 0)) { +#else if (snd_wnd_prev <= 0) { +#endif /* MTCP_TX_ZERO_COPY */ #endif /* SELECTIVE_WRITE_EVENT_NOTIFY */ RaiseWriteEvent(mtcp, cur_stream); #if SELECTIVE_WRITE_EVENT_NOTIFY @@ -598,15 +627,18 @@ static inline int ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, (void)cur_ts; struct tcp_recv_vars *rcvvar = cur_stream->rcvvar; uint32_t prev_rcv_nxt; - int ret; + + // h-> i am returning two bits, bit0: original return value, bit1: app_will_drop + int ret = FALSE; + int app_will_drop = FALSE; /* if seq and segment length is lower than rcv_nxt, ignore and send ack */ if (TCP_SEQ_LT(seq + payloadlen, cur_stream->rcv_nxt)) { - return FALSE; + return (app_will_drop << 1) | ret; } /* if payload exceeds receiving buffer, drop and send ack */ if (TCP_SEQ_GT(seq + payloadlen, cur_stream->rcv_nxt + rcvvar->rcv_wnd)) { - return FALSE; + return (app_will_drop << 1) | ret; } /* allocate receive buffer if not exist */ @@ -618,7 +650,7 @@ static inline int ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, cur_stream->close_reason = TCP_NO_MEM; RaiseErrorEvent(mtcp, cur_stream); - return ERROR; + return (app_will_drop << 1) | ret; } } @@ -629,7 +661,12 @@ static inline int ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, } prev_rcv_nxt = cur_stream->rcv_nxt; +#ifdef MTCP_RX_ZERO_COPY + ret = RBPut_zc(mtcp, mtcp->rbm_rcv, rcvvar->rcvbuf, mtcp->ctx->flash_ctx.flash_addr, payload, (uint32_t)payloadlen, seq); + app_will_drop = TRUE; +#else ret = RBPut(mtcp->rbm_rcv, rcvvar->rcvbuf, payload, (uint32_t)payloadlen, seq); +#endif // MTCP_RX_ZERO_COPY if (ret < 0) { TRACE_ERROR("Cannot merge payload. reason: %d\n", ret); } @@ -637,7 +674,15 @@ static inline int ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, /* discard the buffer if the state is FIN_WAIT_1 or FIN_WAIT_2, meaning that the connection is already closed by the application */ if (cur_stream->state == TCP_ST_FIN_WAIT_1 || cur_stream->state == TCP_ST_FIN_WAIT_2) { +#ifdef MTCP_RX_ZERO_COPY + RBRemove_zc(mtcp, mtcp->rbm_rcv, rcvvar->rcvbuf, rcvvar->rcvbuf->merged_len, AT_MTCP); + // if (TCP_SEQ_LEQ(seq + payloadlen, rcvvar->rcvbuf->head_seq)) { + // app_will_drop = FALSE; + // } +#else RBRemove(mtcp->rbm_rcv, rcvvar->rcvbuf, rcvvar->rcvbuf->merged_len, AT_MTCP); +#endif // MTCP_RX_ZERO_COPY + } cur_stream->rcv_nxt = rcvvar->rcvbuf->head_seq + rcvvar->rcvbuf->merged_len; rcvvar->rcv_wnd = rcvvar->rcvbuf->size - rcvvar->rcvbuf->merged_len; @@ -646,7 +691,8 @@ static inline int ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, if (TCP_SEQ_LEQ(cur_stream->rcv_nxt, prev_rcv_nxt)) { /* There are some lost packets */ - return FALSE; + ret = FALSE; + return (app_will_drop << 1) | ret; } TRACE_EPOLL("Stream %d data arrived. " @@ -659,7 +705,8 @@ static inline int ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, RaiseReadEvent(mtcp, cur_stream); } - return TRUE; + ret = TRUE; + return (app_will_drop << 1) | ret; } /*----------------------------------------------------------------------------*/ static inline tcp_stream *CreateNewFlowHTEntry(mtcp_manager_t mtcp, uint32_t cur_ts, const struct iphdr *iph, int ip_len, @@ -871,7 +918,7 @@ static inline void Handle_TCP_ST_SYN_RCVD(mtcp_manager_t mtcp, uint32_t cur_ts, } } /*----------------------------------------------------------------------------*/ -static inline void Handle_TCP_ST_ESTABLISHED(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, +static inline int Handle_TCP_ST_ESTABLISHED(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, uint32_t seq, uint32_t ack_seq, uint8_t *payload, int payloadlen, uint16_t window) { if (tcph->syn) { @@ -880,11 +927,15 @@ static inline void Handle_TCP_ST_ESTABLISHED(mtcp_manager_t mtcp, uint32_t cur_t cur_stream->id, seq, cur_stream->rcv_nxt, ack_seq, cur_stream->snd_nxt); cur_stream->snd_nxt = ack_seq; AddtoControlList(mtcp, cur_stream, cur_ts); - return; + return FALSE; } - + int ret = FALSE; if (payloadlen > 0) { - if (ProcessTCPPayload(mtcp, cur_stream, cur_ts, payload, seq, payloadlen)) { + int ret_val = ProcessTCPPayload(mtcp, cur_stream, cur_ts, payload, seq, payloadlen); + if (ret_val & 2) { // will_app_drop + ret = TRUE; + } + if (ret_val & 1) { // original return value /* if return is TRUE, send ACK */ EnqueueACK(mtcp, cur_stream, cur_ts, ACK_OPT_AGGREGATE); } else { @@ -911,9 +962,9 @@ static inline void Handle_TCP_ST_ESTABLISHED(mtcp_manager_t mtcp, uint32_t cur_t RaiseReadEvent(mtcp, cur_stream); } else { EnqueueACK(mtcp, cur_stream, cur_ts, ACK_OPT_NOW); - return; } } + return ret; } /*----------------------------------------------------------------------------*/ static inline void Handle_TCP_ST_CLOSE_WAIT(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, @@ -988,7 +1039,7 @@ static inline void Handle_TCP_ST_LAST_ACK(mtcp_manager_t mtcp, uint32_t cur_ts, } } /*----------------------------------------------------------------------------*/ -static inline void Handle_TCP_ST_FIN_WAIT_1(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, +static inline int Handle_TCP_ST_FIN_WAIT_1(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, uint32_t seq, uint32_t ack_seq, uint8_t *payload, int payloadlen, uint16_t window) { if (TCP_SEQ_LT(seq, cur_stream->rcv_nxt)) { @@ -996,9 +1047,11 @@ static inline void Handle_TCP_ST_FIN_WAIT_1(mtcp_manager_t mtcp, uint32_t cur_ts "weird seq: %u, expected: %u\n", cur_stream->id, seq, cur_stream->rcv_nxt); AddtoControlList(mtcp, cur_stream, cur_ts); - return; + return FALSE; } + int ret = FALSE; + if (tcph->ack) { if (cur_stream->sndvar->sndbuf) { ProcessACK(mtcp, cur_stream, cur_ts, tcph, seq, ack_seq, window, payloadlen); @@ -1020,11 +1073,15 @@ static inline void Handle_TCP_ST_FIN_WAIT_1(mtcp_manager_t mtcp, uint32_t cur_ts } else { TRACE_DBG("Stream %d: does not contain an ack!\n", cur_stream->id); - return; + return FALSE; } if (payloadlen > 0) { - if (ProcessTCPPayload(mtcp, cur_stream, cur_ts, payload, seq, payloadlen)) { + int ret_val = ProcessTCPPayload(mtcp, cur_stream, cur_ts, payload, seq, payloadlen); + if (ret_val & 2) { + ret = TRUE; + } + if (ret_val & 1) { /* if return is TRUE, send ACK */ EnqueueACK(mtcp, cur_stream, cur_ts, ACK_OPT_AGGREGATE); } else { @@ -1050,22 +1107,28 @@ static inline void Handle_TCP_ST_FIN_WAIT_1(mtcp_manager_t mtcp, uint32_t cur_ts AddtoControlList(mtcp, cur_stream, cur_ts); } } + return ret; } /*----------------------------------------------------------------------------*/ -static inline void Handle_TCP_ST_FIN_WAIT_2(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, +static inline int Handle_TCP_ST_FIN_WAIT_2(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, uint32_t seq, uint32_t ack_seq, uint8_t *payload, int payloadlen, uint16_t window) { + int ret = FALSE; if (tcph->ack) { if (cur_stream->sndvar->sndbuf) { ProcessACK(mtcp, cur_stream, cur_ts, tcph, seq, ack_seq, window, payloadlen); } } else { TRACE_DBG("Stream %d: does not contain an ack!\n", cur_stream->id); - return; + return FALSE; } if (payloadlen > 0) { - if (ProcessTCPPayload(mtcp, cur_stream, cur_ts, payload, seq, payloadlen)) { + int ret_val = ProcessTCPPayload(mtcp, cur_stream, cur_ts, payload, seq, payloadlen); + if (ret_val & 2) { + ret = TRUE; + } + if (ret_val & 1) { /* if return is TRUE, send ACK */ EnqueueACK(mtcp, cur_stream, cur_ts, ACK_OPT_AGGREGATE); } else { @@ -1095,6 +1158,7 @@ static inline void Handle_TCP_ST_FIN_WAIT_2(mtcp_manager_t mtcp, uint32_t cur_ts #endif #endif } + return ret; } /*----------------------------------------------------------------------------*/ static inline void Handle_TCP_ST_CLOSING(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_stream, struct tcphdr *tcph, @@ -1185,11 +1249,15 @@ int ProcessTCPPacket(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifidx, cons s_stream.daddr = iph->saddr; s_stream.dport = tcph->source; +#ifdef MTCP_FLASH_ID_TRAILER + s_stream.dst_flash_id = mtcp->ctx->flash_ctx.dst_flash_id; +#endif + if (!(cur_stream = StreamHTSearch(mtcp->tcp_flow_table, &s_stream))) { /* not found in flow table */ cur_stream = CreateNewFlowHTEntry(mtcp, cur_ts, iph, ip_len, tcph, seq, ack_seq, payloadlen, window); if (!cur_stream) - return TRUE; + return FALSE; // false to free up the packet } /* Validate sequence. if not valid, ignore the packet */ @@ -1203,7 +1271,7 @@ int ProcessTCPPacket(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifidx, cons #ifdef DUMP_STREAM DumpStream(mtcp, cur_stream); #endif - return TRUE; + return FALSE; // false to free up the packet } } @@ -1222,7 +1290,7 @@ int ProcessTCPPacket(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifidx, cons cur_stream->have_reset = TRUE; if (cur_stream->state > TCP_ST_SYN_SENT) { if (ProcessRST(mtcp, cur_stream, ack_seq)) { - return TRUE; + return FALSE; // false to free up the packet } } } @@ -1243,13 +1311,13 @@ int ProcessTCPPacket(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifidx, cons else { Handle_TCP_ST_SYN_RCVD(mtcp, cur_ts, cur_stream, tcph, ack_seq); if (payloadlen > 0 && cur_stream->state == TCP_ST_ESTABLISHED) { - Handle_TCP_ST_ESTABLISHED(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); + return Handle_TCP_ST_ESTABLISHED(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); } } break; case TCP_ST_ESTABLISHED: - Handle_TCP_ST_ESTABLISHED(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); + return Handle_TCP_ST_ESTABLISHED(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); break; case TCP_ST_CLOSE_WAIT: @@ -1261,11 +1329,11 @@ int ProcessTCPPacket(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifidx, cons break; case TCP_ST_FIN_WAIT_1: - Handle_TCP_ST_FIN_WAIT_1(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); + return Handle_TCP_ST_FIN_WAIT_1(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); break; case TCP_ST_FIN_WAIT_2: - Handle_TCP_ST_FIN_WAIT_2(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); + return Handle_TCP_ST_FIN_WAIT_2(mtcp, cur_ts, cur_stream, tcph, seq, ack_seq, payload, payloadlen, window); break; case TCP_ST_CLOSING: @@ -1286,5 +1354,5 @@ int ProcessTCPPacket(mtcp_manager_t mtcp, uint32_t cur_ts, const int ifidx, cons break; } - return TRUE; + return FALSE; // to free up the packet } diff --git a/lib/flash/mtcp/tcp_out.c b/lib/flash/mtcp/tcp_out.c index a4bba01..4b55763 100644 --- a/lib/flash/mtcp/tcp_out.c +++ b/lib/flash/mtcp/tcp_out.c @@ -249,6 +249,26 @@ int SendTCPPacket(struct mtcp_manager *mtcp, tcp_stream *cur_stream, uint32_t cu int rc = -1; optlen = CalculateOptionLength(flags); +#ifdef MTCP_TX_ZERO_COPY + // h-> need to check mss length for pkt? for now i am experimenting with pkt sizes upto 1024 B + if (mtcp->ctx->flash_ctx.tx_zc == 0 && payloadlen + optlen > cur_stream->sndvar->mss) { + TRACE_ERROR("Payload size exceeds MSS\n"); + return ERROR; + } + if (mtcp->ctx->flash_ctx.tx_zc && payloadlen > 0) { + // data for IP and TCP header will be written in place + // h-> the header len is hardcoded for (TCP_OPT_TIMESTAMP_ENABLED = 1, TCP_OPT_SACK_ENABLED = 1,), where SACK flag is never set + assert(optlen == (TCP_OPT_TIMESTAMP_LEN + 2)); + mtcp->ctx->flash_ctx.data = payload - (TCP_HEADER_LEN + optlen); + } else { + mtcp->ctx->flash_ctx.data = NULL; + } +#else + if (payloadlen + optlen > cur_stream->sndvar->mss) { + TRACE_ERROR("Payload size exceeds MSS\n"); + return ERROR; + } +#endif /* MTCP_TX_ZERO_COPY */ if (payloadlen + optlen > cur_stream->sndvar->mss) { TRACE_ERROR("Payload size exceeds MSS\n"); return ERROR; @@ -327,7 +347,10 @@ int SendTCPPacket(struct mtcp_manager *mtcp, tcp_stream *cur_stream, uint32_t cu tcph->doff = (TCP_HEADER_LEN + optlen) >> 2; // copy payload if exist if (payloadlen > 0) { +#ifdef MTCP_TX_ZERO_COPY +#else memcpy((uint8_t *)tcph + TCP_HEADER_LEN + optlen, payload, payloadlen); +#endif /* MTCP_TX_ZERO_COPY */ #if defined(NETSTAT) && defined(ENABLELRO) mtcp->nstat.tx_gdptbytes += payloadlen; #endif /* NETSTAT */ @@ -1070,3 +1093,255 @@ inline void DumpControlList(mtcp_manager_t mtcp, struct mtcp_sender *sender) TRACE_DBG("Stream id: %u in control list\n", stream->id); } } + +#ifdef MTCP_TX_ZERO_COPY +/*----------------------------------------------------------------------------*/ +static int FlushTCPSendingBufferZC(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint32_t cur_ts) +{ + struct tcp_send_vars *sndvar = cur_stream->sndvar; + struct tcp_send_buffer *sndbuf; + uint8_t *data; + uint32_t total_len_left; + uint32_t seq = 0; + int remaining_window; + int sndlen; + int packets = 0; + uint8_t wack_sent = 0; + + if (!sndvar->sndbuf) { + TRACE_ERROR("Stream %d: No send buffer available.\n", cur_stream->id); + assert(0); + return 0; + } + + SBUF_LOCK(&sndvar->write_lock); + sndbuf = sndvar->sndbuf; + + /* empty queue */ + if (sndbuf->q_front == -1) { + goto out; + } + + while (1) { + seq = cur_stream->snd_nxt; + + // len that is left to send in the buffer + total_len_left = sndbuf->len - (seq - sndbuf->head_seq); + + /* sanity check */ + if (TCP_SEQ_LT(seq, sndvar->sndbuf->head_seq)) { + TRACE_ERROR("Stream %d: Invalid sequence to send. " + "state: %s, seq: %u, head_seq: %u.\n", + cur_stream->id, TCPStateToString(cur_stream), seq, sndvar->sndbuf->head_seq); + assert(0); + break; + } + if (TCP_SEQ_LT(seq, sndvar->snd_una)) { + TRACE_ERROR("Stream %d: Invalid sequence to send. " + "state: %s, seq: %u, snd_una: %u.\n", + cur_stream->id, TCPStateToString(cur_stream), seq, sndvar->snd_una); + assert(0); + break; + } + + if (sndbuf->len < (seq - sndbuf->head_seq)) { + TRACE_ERROR("Stream %d: len < 0\n", cur_stream->id); + assert(0); + break; + } + + if (total_len_left == 0) { + TRACE_DBG("Stream %d: No more data to send. seq: %u, head_seq: %u, len: %u\n", cur_stream->id, seq, + sndbuf->head_seq, sndbuf->len); + break; + } + + // locate the correct zero-copy slot for the current 'seq' + int send_index = sndbuf->q_front; + uint32_t curr_seq = sndbuf->head_seq - sndbuf->zc_buf[send_index].acked_len; + + // traverse until we find the slot where 'seq' belongs + while (TCP_SEQ_LEQ(curr_seq + sndbuf->zc_buf[send_index].len, seq)) { + curr_seq += sndbuf->zc_buf[send_index].len; + + if (send_index == sndbuf->q_back) { + // reached the end of valid queued data without encompassing 'seq' + break; + } + send_index = (send_index + 1) & (MAX_TX_SLOTS - 1); + } + +#if TCP_OPT_SACK_ENABLED + if (SeqIsSacked(cur_stream, seq)) { + /* + * already sacked, + * skip sending again + * h-> the buffer will be freed when the ACK is processed in tcp_in.c. + */ + TRACE_DBG("!! SKIPPING %u\n", seq - sndvar->iss); + // skip this umem buffer, move to the next one + cur_stream->snd_nxt += (sndbuf->zc_buf[send_index].len - (seq - curr_seq)); + continue; + } +#endif + + remaining_window = MIN(sndvar->cwnd, sndvar->peer_wnd) - (seq - sndvar->snd_una); + /* if there is no space in the window */ + if (remaining_window <= 0 || (remaining_window < sndvar->mss && seq - sndvar->snd_una > 0)) { + /* if peer window is full, send ACK and let its peer advertises new one */ + if (sndvar->peer_wnd <= sndvar->cwnd) { + if (!wack_sent && TS_TO_MSEC(cur_ts - sndvar->ts_lastack_sent) > 500) + EnqueueACK(mtcp, cur_stream, cur_ts, ACK_OPT_WACK); + else + wack_sent = 1; + } + packets = -3; + goto out; + } + + // ATOMIC ZERO COPY SEND + // h-> we have a constraint here, we cannot offset 'data' because prepending headers overwrites payload. + // therefore, if 'seq' is partially inside this slot, we MUST rewind and send the whole slot. + // h-> TODO: need to check mss before allocating and giving buffer to app + if (seq > curr_seq) { + TRACE_DBG("Partial ACK detected. Rewinding snd_nxt to %u to preserve ZC atomic slot.\n", curr_seq); + cur_stream->snd_nxt = curr_seq; + seq = curr_seq; + } + + data = sndbuf->zc_buf[send_index].payload; + uint32_t slot_send_len = sndbuf->zc_buf[send_index].len; + + // tell the lower layers about the buf + mtcp->ctx->flash_ctx.flash_addr = sndbuf->zc_buf[send_index].flash_addr; + // Send the entire pkt buffer + if ((sndlen = SendTCPPacket(mtcp, cur_stream, cur_ts, TCP_FLAG_ACK, data, slot_send_len)) < 0) { + /* no available tx buffers */ + packets = -3; + goto out; + } + + packets++; + } + +out: + SBUF_UNLOCK(&sndvar->write_lock); + return packets; +} + +/*----------------------------------------------------------------------------*/ +inline int WriteTCPZCDataList(mtcp_manager_t mtcp, struct mtcp_sender *sender, uint32_t cur_ts, int thresh) +{ + tcp_stream *cur_stream; + tcp_stream *next, *last; + int cnt = 0; + int ret; + + /* Send data */ + cnt = 0; + cur_stream = TAILQ_FIRST(&sender->sendZC_list); + last = TAILQ_LAST(&sender->sendZC_list, sendZC_head); + while (cur_stream) { + if (++cnt > thresh) + break; + + TRACE_LOOP("Inside send loop. cnt: %u, stream: %d\n", cnt, cur_stream->id); + next = TAILQ_NEXT(cur_stream, sndvar->sendZC_link); + + TAILQ_REMOVE(&sender->sendZC_list, cur_stream, sndvar->sendZC_link); + if (cur_stream->sndvar->on_sendZC_list) { + ret = 0; + + /* Send data here */ + /* Only can send data when ESTABLISHED or CLOSE_WAIT */ + if (cur_stream->state == TCP_ST_ESTABLISHED) { + if (cur_stream->sndvar->on_control_list) { + /* delay sending data after until on_control_list becomes off */ + //TRACE_DBG("Stream %u: delay sending data.\n", cur_stream->id); + ret = -1; + } else { + ret = FlushTCPSendingBufferZC(mtcp, cur_stream, cur_ts); + } + } else if (cur_stream->state == TCP_ST_CLOSE_WAIT || cur_stream->state == TCP_ST_FIN_WAIT_1 || + cur_stream->state == TCP_ST_LAST_ACK) { + ret = FlushTCPSendingBufferZC(mtcp, cur_stream, cur_ts); + } else { + TRACE_DBG("Stream %d: on_send_list at state %s\n", cur_stream->id, TCPStateToString(cur_stream)); +#if defined(DUMP_STREAM) + DumpStream(mtcp, cur_stream); +#endif + } + + if (ret < 0) { + TAILQ_INSERT_TAIL(&sender->sendZC_list, cur_stream, sndvar->sendZC_link); + /* since there is no available write buffer, break */ + break; + + } else { + cur_stream->sndvar->on_sendZC_list = FALSE; + sender->sendZC_list_cnt--; + /* the ret value is the number of packets sent. */ + /* decrease ack_cnt for the piggybacked acks */ +#if ACK_PIGGYBACK + if (cur_stream->sndvar->ack_cnt > 0) { + if (cur_stream->sndvar->ack_cnt > ret) { + cur_stream->sndvar->ack_cnt -= ret; + } else { + cur_stream->sndvar->ack_cnt = 0; + } + } +#endif + if (cur_stream->control_list_waiting) { + if (!cur_stream->sndvar->on_ack_list) { + cur_stream->control_list_waiting = FALSE; + AddtoControlList(mtcp, cur_stream, cur_ts); + } + } + } + } else { + TRACE_ERROR("Stream %d: not on send list.\n", cur_stream->id); +#ifdef DUMP_STREAM + DumpStream(mtcp, cur_stream); +#endif + } + + if (cur_stream == last) + break; + cur_stream = next; + } + + return cnt; +} + +/*----------------------------------------------------------------------------*/ +inline void AddtoZCSendList(mtcp_manager_t mtcp, tcp_stream *cur_stream) +{ + struct mtcp_sender *sender = GetSender(mtcp, cur_stream); + assert(sender != NULL); + + if (!cur_stream->sndvar->sndbuf) { + TRACE_ERROR("[%d] Stream %d: No send buffer available.\n", mtcp->ctx->cpu, cur_stream->id); + assert(0); + return; + } + + if (!cur_stream->sndvar->on_sendZC_list) { + cur_stream->sndvar->on_sendZC_list = TRUE; + TAILQ_INSERT_TAIL(&sender->sendZC_list, cur_stream, sndvar->sendZC_link); + sender->sendZC_list_cnt++; + } +} + +/*----------------------------------------------------------------------------*/ +inline void RemoveFromZCSendList(mtcp_manager_t mtcp, tcp_stream *cur_stream) +{ + struct mtcp_sender *sender = GetSender(mtcp, cur_stream); + assert(sender != NULL); + + if (cur_stream->sndvar->on_sendZC_list) { + cur_stream->sndvar->on_sendZC_list = FALSE; + TAILQ_REMOVE(&sender->sendZC_list, cur_stream, sndvar->sendZC_link); + sender->sendZC_list_cnt--; + } +} +#endif /* MTCP_TX_ZERO_COPY */ diff --git a/lib/flash/mtcp/tcp_rb_frag_queue.c b/lib/flash/mtcp/tcp_rb_frag_queue.c index 76aabd1..7a2a218 100644 --- a/lib/flash/mtcp/tcp_rb_frag_queue.c +++ b/lib/flash/mtcp/tcp_rb_frag_queue.c @@ -37,6 +37,16 @@ struct rb_frag_queue { struct fragment_ctx *volatile *_q; }; + +#ifdef MTCP_RX_ZERO_COPY +struct rb_frag_queue_zc { + index_type _capacity; + volatile index_type _head; + volatile index_type _tail; + struct fragment_ctx_zc *volatile *_q; +}; +#endif /* MTCP_RX_ZERO_COPY */ + /*----------------------------------------------------------------------------*/ static inline index_type NextIndex(rb_frag_queue_t rb_fragq, index_type i) { @@ -121,3 +131,88 @@ struct fragment_ctx *RBFragDequeue(rb_frag_queue_t rb_fragq) return NULL; } /*---------------------------------------------------------------------------*/ + +#ifdef MTCP_RX_ZERO_COPY +static inline index_type NextIndex_zc(rb_frag_queue_zc_t q, index_type i) +{ + return (i != q->_capacity ? i + 1 : 0); +} +/*---------------------------------------------------------------------------*/ +static inline void RBFragMemoryBarrier_zc(struct fragment_ctx_zc *volatile frag, + volatile index_type index) +{ + __asm__ volatile("" : : "m"(frag), "m"(index)); +} +/*---------------------------------------------------------------------------*/ +rb_frag_queue_zc_t CreateRBFragQueue_zc(int capacity) +{ + rb_frag_queue_zc_t q; + + q = (rb_frag_queue_zc_t)calloc(1, sizeof(struct rb_frag_queue_zc)); + if (!q) + return NULL; + + q->_q = (struct fragment_ctx_zc **) + calloc(capacity + 1, sizeof(struct fragment_ctx_zc *)); + if (!q->_q) { + free(q); + return NULL; + } + + q->_capacity = capacity; + q->_head = q->_tail = 0; + + return q; +} +/*---------------------------------------------------------------------------*/ +void DestroyRBFragQueue_zc(rb_frag_queue_zc_t q) +{ + if (!q) + return; + + if (q->_q) { + void *tmp = (void *)(uintptr_t)q->_q; + free(tmp); + q->_q = NULL; + } + + free(q); +} +/*---------------------------------------------------------------------------*/ +int RBFragEnqueue_zc(rb_frag_queue_zc_t q, struct fragment_ctx_zc *frag) +{ + index_type h = q->_head; + index_type t = q->_tail; + index_type nt = NextIndex_zc(q, t); + + if (nt != h) { + q->_q[t] = frag; + RBFragMemoryBarrier_zc(q->_q[t], q->_tail); + q->_tail = nt; + return 0; + } + + return -1; +} + +/*---------------------------------------------------------------------------*/ + +struct fragment_ctx_zc *RBFragDequeue_zc(rb_frag_queue_zc_t q) +{ + index_type h = q->_head; + index_type t = q->_tail; + + if (h != t) { + struct fragment_ctx_zc *frag = q->_q[h]; + RBFragMemoryBarrier_zc(q->_q[h], q->_head); + q->_head = NextIndex_zc(q, h); + assert(frag); + + return frag; + } + + return NULL; +} +/*---------------------------------------------------------------------------*/ +#endif /* MTCP_RX_ZERO_COPY */ + diff --git a/lib/flash/mtcp/tcp_ring_buffer.c b/lib/flash/mtcp/tcp_ring_buffer.c index 0089eb3..db466d9 100644 --- a/lib/flash/mtcp/tcp_ring_buffer.c +++ b/lib/flash/mtcp/tcp_ring_buffer.c @@ -62,6 +62,13 @@ struct rb_manager { rb_frag_queue_t free_fragq; /* free fragment queue (for app thread) */ rb_frag_queue_t free_fragq_int; /* free fragment quuee (only for mtcp) */ + +#ifdef MTCP_RX_ZERO_COPY + mem_pool_t frag_mp_zc; + rb_frag_queue_zc_t free_fragq_zc; // app thread + rb_frag_queue_zc_t free_fragq_zc_int; // mtcp thread +#endif /* MTCP_RX_ZERO_COPY */ + #ifdef ENABLELRO mtcp_manager_t mtcp; #endif @@ -154,6 +161,45 @@ rb_manager_t RBManagerCreate(mtcp_manager_t mtcp, size_t chunk_size, uint32_t cn return NULL; } +#ifdef MTCP_RX_ZERO_COPY + rbm->frag_mp_zc = (mem_pool_t)MPCreate(sizeof(struct fragment_ctx_zc), sizeof(struct fragment_ctx_zc) * cnum); + if (!rbm->frag_mp_zc) { + TRACE_ERROR("Failed to allocate frag_mp_zc pool.\n"); + MPDestroy(rbm->mp); + MPDestroy(rbm->frag_mp); + DestroyRBFragQueue(rbm->free_fragq); + DestroyRBFragQueue(rbm->free_fragq_int); + free(rbm); + return NULL; + } + + + rbm->free_fragq_zc = CreateRBFragQueue_zc(cnum); + if (!rbm->free_fragq_zc) { + TRACE_ERROR("Failed to create zero-copy free fragment queue.\n"); + MPDestroy(rbm->mp); + MPDestroy(rbm->frag_mp); + MPDestroy(rbm->frag_mp_zc); + DestroyRBFragQueue(rbm->free_fragq); + DestroyRBFragQueue(rbm->free_fragq_int); + free(rbm); + return NULL; + } + + rbm->free_fragq_zc_int = CreateRBFragQueue_zc(cnum); + if (!rbm->free_fragq_zc_int) { + TRACE_ERROR("Failed to create internal zero-copy free fragment queue.\n"); + MPDestroy(rbm->mp); + MPDestroy(rbm->frag_mp); + MPDestroy(rbm->frag_mp_zc); + DestroyRBFragQueue(rbm->free_fragq); + DestroyRBFragQueue(rbm->free_fragq_int); + DestroyRBFragQueue_zc(rbm->free_fragq_zc); + free(rbm); + return NULL; + } +#endif /* MTCP_RX_ZERO_COPY */ + #ifdef ENABLELRO rbm->mtcp = mtcp; #endif @@ -183,6 +229,19 @@ static void FreeFragmentContext(rb_manager_t rbm, struct fragment_ctx *fctx) } } /*----------------------------------------------------------------------------*/ +#ifdef MTCP_RX_ZERO_COPY +static void +FreeFragmentContext_zc(rb_manager_t rbm, + struct fragment_ctx_zc *frag, + int option) +{ + if (option == AT_APP) + RBFragEnqueue_zc(rbm->free_fragq_zc, frag); + else + RBFragEnqueue_zc(rbm->free_fragq_zc_int, frag); +} +#endif /* MTCP_RX_ZERO_COPY */ +/*----------------------------------------------------------------------------*/ static struct fragment_ctx *AllocateFragmentContext(rb_manager_t rbm) { /* this function should be called only in mtcp thread */ @@ -238,8 +297,9 @@ struct tcp_ring_buffer *RBInit(rb_manager_t rbm, uint32_t init_seq) return buff; } /*----------------------------------------------------------------------------*/ -void RBFree(rb_manager_t rbm, struct tcp_ring_buffer *buff) +void RBFree(mtcp_manager_t mtcp, rb_manager_t rbm, struct tcp_ring_buffer *buff) { + UNUSED(mtcp); assert(buff); if (buff->fctx) { FreeFragmentContext(rbm, buff->fctx); @@ -252,6 +312,16 @@ void RBFree(rb_manager_t rbm, struct tcp_ring_buffer *buff) rbm->cur_num--; +#ifdef MTCP_RX_ZERO_COPY + while (buff->fctx_zc) { + struct fragment_ctx_zc *tmp = buff->fctx_zc; + buff->fctx_zc = tmp->next; + mtcp->iom->release_pkt(mtcp->ctx, -2, (uint8_t *)tmp->flash_addr, tmp->len); + FreeFragmentContext_zc(rbm, tmp, AT_MTCP); // since RBFree is only called by mtcp thread only + } +#endif /* MTCP_RX_ZERO_COPY */ + + free(buff); } /*----------------------------------------------------------------------------*/ @@ -430,3 +500,271 @@ size_t RBRemove(rb_manager_t rbm, struct tcp_ring_buffer *buff, size_t len, int return len; } /*----------------------------------------------------------------------------*/ +#ifdef MTCP_RX_ZERO_COPY +static struct fragment_ctx_zc * +AllocateFragmentContext_zc(rb_manager_t rbm) +{ + struct fragment_ctx_zc *frag; + + frag = RBFragDequeue_zc(rbm->free_fragq_zc); + if (!frag) { + frag = RBFragDequeue_zc(rbm->free_fragq_zc_int); + if (!frag) { + frag = MPAllocateChunk(rbm->frag_mp_zc); + if (!frag) { + frag = calloc(1, sizeof(*frag)); + if (!frag) + exit(-1); + } + } + } + + memset(frag, 0, sizeof(*frag)); + return frag; +} + +size_t RBAvailable_zc(struct tcp_ring_buffer *buff) +{ + return buff->merged_len; +} + +int RBPut_zc(mtcp_manager_t mtcp, rb_manager_t rbm, struct tcp_ring_buffer *buff, + uint64_t flash_addr, + unsigned char *payload, + uint32_t len, + uint32_t cur_seq) +{ + struct fragment_ctx_zc *new_ctx; + uint32_t expect = buff->head_seq + buff->merged_len; + uint32_t end_seq = cur_seq + len; + + // 1. reject packets that wont contribute to merged_len + // h-> it is okay to reject out of order packets, not much drop in performance + // mtcp immediately sends an ack out, telling the sender to retransmit the missing packet + if (len == 0 || + GetMinSeq(end_seq, expect) == end_seq || // end_seq <= expect + GetMinSeq(expect, cur_seq) != cur_seq) { // expect < cur_seq + mtcp->iom->release_pkt(mtcp->ctx, -1, (uint8_t *)flash_addr, len); + return 0; + } + + // h-> at this point, cur_seq <= expect and expect < end_seq + + // overlap adjustment + // h-> i am making sure that the fragments do not overlap in data, + // therefore i am adjusting the payload and cur_seq to ensure that the new fragment starts exactly at expect + uint32_t offset = expect - cur_seq; + if (offset >= len) { + mtcp->iom->release_pkt(mtcp->ctx, -1, (uint8_t *)flash_addr, len); + return 0; + } + payload += offset; + cur_seq = expect; + len -= offset; + end_seq = cur_seq + len; + + new_ctx = AllocateFragmentContext_zc(rbm); + if (!new_ctx) { + mtcp->iom->release_pkt(mtcp->ctx, -1, (uint8_t *)flash_addr, len); + return 0; + } + new_ctx->seq = cur_seq; + new_ctx->len = len; + new_ctx->payload = payload; + new_ctx->payload_len = len; + new_ctx->flash_addr = flash_addr; + new_ctx->next = NULL; + + // h-> now we can simply append this pkt at the tail of the list + if (!buff->fctx_zc) { + buff->fctx_zc = new_ctx; + buff->fctx_zc_tail = new_ctx; + } else { + buff->fctx_zc_tail->next = new_ctx; + buff->fctx_zc_tail = new_ctx; + } + + // update merged_len + buff->merged_len += len; + return len; +} + +size_t RBRead_zc(rb_manager_t rbm, struct tcp_ring_buffer *buff, + void *dst, size_t len) +{ + struct fragment_ctx_zc *f; + size_t copied = 0; + unsigned char *out = dst; + + (void)rbm; + + if ((size_t)buff->merged_len < len) + len = (size_t)buff->merged_len; + + if (len == 0) + return 0; + + f = buff->fctx_zc; + + // h-> RBPut_zc guarantees: + // 1. fctx_zc starts exactly at head_seq + // 2. Each fragment's seq is exactly the previous fragment's end_seq + // 3. There are no overlaps + // Therefore we can simply iterate and copy. + + while (f && copied < len) { + uint32_t remaining_to_copy = len - copied; + uint32_t can_copy_from_this_frag = f->payload_len; + + if (can_copy_from_this_frag > remaining_to_copy) { + can_copy_from_this_frag = remaining_to_copy; + } + + // h-> fragments point to non-overlapping payloads, so we can safely copy + memcpy(out + copied, f->payload, can_copy_from_this_frag); + + copied += can_copy_from_this_frag; + f = f->next; + } + + return copied; +} + +size_t RBRemove_zc(mtcp_manager_t mtcp, rb_manager_t rbm, struct tcp_ring_buffer *buff, + size_t len, int option) +{ + /* this function should be called only in application thread */ + (void)rbm; + + if ((size_t)buff->merged_len < len) + len = buff->merged_len; + + if (len == 0) + return 0; + + struct fragment_ctx_zc *f; + size_t to_remove = len; + + while (to_remove > 0 && buff->fctx_zc) { + f = buff->fctx_zc; + + if (to_remove < f->payload_len) { + // Partial removal: + // We don't free the fragment yet, just move the logical start. + // This maintains the invariant: f->seq remains equal to head_seq. + f->seq += to_remove; + f->payload += to_remove; + f->payload_len -= to_remove; + + buff->head_seq += to_remove; + to_remove = 0; + } else { + // Full removal + buff->fctx_zc = f->next; + + if (buff->fctx_zc == NULL) { + buff->fctx_zc_tail = NULL; + } + + uint32_t consumed = f->payload_len; + + // Release the actual packet buffer and metadata + if (option == AT_APP) { + RBFreeBuff_zc(mtcp, rbm, f->flash_addr); + } else { + mtcp->iom->release_pkt(mtcp->ctx, -2, (uint8_t *)f->flash_addr, f->len); + } + FreeFragmentContext_zc(rbm, f, option); + + buff->head_seq += consumed; + to_remove -= consumed; + } + } + + // h-> since everything was already contiguous + buff->merged_len -= len; + + return len; +} + +size_t RBGetBuff_zc(rb_manager_t rbm, struct tcp_ring_buffer *buff, void **buf, uint64_t* flash_addr) +{ + if (!buff) { + *buf = NULL; + *flash_addr = 0; + return 0; + } + + /* this function should be called only in application thread */ + (void)rbm; + + if (!buff->fctx_zc) { + *buf = NULL; + *flash_addr = 0; + return 0; + } + + struct fragment_ctx_zc *f = buff->fctx_zc; + buff->fctx_zc = f->next; + if (!buff->fctx_zc) { + buff->fctx_zc_tail = NULL; + } + + *buf = f->payload; + *flash_addr = f->flash_addr; + buff->head_seq += f->payload_len; + buff->merged_len -= f->payload_len; + + FreeFragmentContext_zc(rbm, f, AT_APP); + + return f->payload_len; +} + +// app thread produces +// mtcp thread consumes +void RBFreeBuff_zc(mtcp_manager_t mtcp, rb_manager_t rbm, uint64_t flash_addr) +{ + UNUSED(mtcp); + UNUSED(rbm); + + struct zc_rx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_rx_ring; + + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_RELAXED); + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_ACQUIRE); + + // overflow + if (tail - head >= ZC_RX_FREE_RING_SIZE) { + printf("not possible: ring size was taken too big\n"); + errno = EAGAIN; + return; + } + + ring->flash_addrs[tail & (ZC_RX_FREE_RING_SIZE - 1)] = flash_addr; + + __atomic_store_n(&ring->tail, tail + 1, __ATOMIC_RELEASE); +} + +int RBFreeBuff_zc_batch(mtcp_manager_t mtcp, rb_manager_t rbm, uint64_t *addr_array, int num_pkts) +{ + (void) rbm; + struct zc_rx_free_ring *ring = &mtcp->ctx->flash_ctx.zc_rx_ring; + uint32_t tail = __atomic_load_n(&ring->tail, __ATOMIC_RELAXED); + uint32_t head = __atomic_load_n(&ring->head, __ATOMIC_ACQUIRE); + + + if (tail + num_pkts - head >= ZC_RX_FREE_RING_SIZE) { + printf("not possible: ring size was taken too big\n"); + errno = EAGAIN; + return -1; + } + + for (int i = 0; i < num_pkts; i++) { + ring->flash_addrs[(tail + i) & (ZC_RX_FREE_RING_SIZE - 1)] = addr_array[i]; + } + + __atomic_store_n(&ring->tail, tail + num_pkts, __ATOMIC_RELEASE); + + return 0; +} + +#endif /* MTCP_RX_ZERO_COPY */ diff --git a/lib/flash/mtcp/tcp_send_buffer.c b/lib/flash/mtcp/tcp_send_buffer.c index 5092119..6372f41 100644 --- a/lib/flash/mtcp/tcp_send_buffer.c +++ b/lib/flash/mtcp/tcp_send_buffer.c @@ -118,6 +118,11 @@ struct tcp_send_buffer *SBInit(sb_manager_t sbm, uint32_t init_seq) buf->init_seq = buf->head_seq = init_seq; +#ifdef MTCP_TX_ZERO_COPY + buf->q_front = buf->q_back = -1; + buf->available_zc_slots = MAX_TX_SLOTS; +#endif /* MTCP_TX_ZERO_COPY */ + return buf; } /*----------------------------------------------------------------------------*/ @@ -205,3 +210,111 @@ size_t SBRemove(sb_manager_t sbm, struct tcp_send_buffer *buf, size_t len) return to_remove; } /*---------------------------------------------------------------------------*/ + +#ifdef MTCP_TX_ZERO_COPY +/*----------------------------------------------------------------------------*/ +void SBClear_zc(mtcp_manager_t mtcp, struct tcp_send_buffer *buf) +{ + if (!buf || buf->q_front == -1) + return; + + int i = buf->q_front; + while (1) { + if (buf->zc_buf[i].payload != NULL) { + mtcp->iom->release_pkt(mtcp->ctx, 0, (void*)buf->zc_buf[i].flash_addr, buf->zc_buf[i].len); + + buf->zc_buf[i].payload = NULL; + buf->zc_buf[i].flash_addr = 0; + buf->zc_buf[i].len = 0; + buf->zc_buf[i].acked_len = 0; + } + + if (i == buf->q_back) break; + i = (i + 1) & (MAX_TX_SLOTS - 1); + } + + buf->q_front = buf->q_back = -1; + buf->available_zc_slots = MAX_TX_SLOTS; +} +/*----------------------------------------------------------------------------*/ +size_t SBPut_zc(sb_manager_t sbm, struct tcp_send_buffer *buf, const void *data, size_t len, uint64_t flash_addr) +{ + (void)sbm; + + if (len <= 0) /* should not happen */ + return 0; + + /* if no space, return -2 */ + if (buf->available_zc_slots <= 0) { + /* should not happen */ + return -2; + } + + if (buf->q_back == -1) { + buf->q_front = buf->q_back = 0; + } else { + buf->q_back = (buf->q_back + 1) & (MAX_TX_SLOTS - 1); + if (buf->q_back == buf->q_front) { + // should never happen since we check for available_zc_slots above + TRACE_ERROR("SBPut_zc: queue overflow. This should never happen. available_zc_slots: %u\n", buf->available_zc_slots); + assert(0); + return -2; + } + } + + buf->zc_buf[buf->q_back].payload = (uint8_t* )(uint64_t)data; + buf->zc_buf[buf->q_back].len = len; + buf->zc_buf[buf->q_back].flash_addr = flash_addr; + buf->zc_buf[buf->q_back].acked_len = 0; + + buf->available_zc_slots -= 1; + buf->len += len; + + return len; +} +/*---------------------------------------------------------------------------*/ +size_t SBRemove_zc(mtcp_manager_t mtcp, struct tcp_send_buffer *buf, size_t acked_len) { + size_t original_len = acked_len; + + while (acked_len > 0 && buf->q_front != -1) { + struct data_zc *zc = &buf->zc_buf[buf->q_front]; + + // how many bytes are left to be acknowledged in this specific umem buffer? + size_t unacked_in_slot = zc->len - zc->acked_len; + + if (acked_len >= unacked_in_slot) { + // umem buffer is fully acked, => we can release it + mtcp->iom->release_pkt(mtcp->ctx, 0, (void*)zc->flash_addr, zc->len); + + acked_len -= unacked_in_slot; + buf->head_seq += unacked_in_slot; + buf->len -= unacked_in_slot; + buf->available_zc_slots++; + + // clean up + zc->payload = NULL; + zc->len = 0; + zc->flash_addr = 0; + zc->acked_len = 0; + + + if (buf->q_front == buf->q_back) { + buf->q_front = buf->q_back = -1; + break; + } else { + buf->q_front = (buf->q_front + 1) & (MAX_TX_SLOTS - 1); + } + } else { + // umem buffer is partially acked, => advance seq numbers, but do not release the physical packet yet + zc->acked_len += acked_len; + buf->head_seq += acked_len; + buf->len -= acked_len; + acked_len = 0; // ack is consumed + break; + } + } + + return original_len - acked_len; // total bytes removed from send_buffer +} +/*---------------------------------------------------------------------------*/ +#endif /* MTCP_TX_ZERO_COPY */ diff --git a/lib/flash/mtcp/tcp_stream.c b/lib/flash/mtcp/tcp_stream.c index 20cbf4b..04529f2 100644 --- a/lib/flash/mtcp/tcp_stream.c +++ b/lib/flash/mtcp/tcp_stream.c @@ -262,6 +262,10 @@ tcp_stream *CreateTCPStream(mtcp_manager_t mtcp, socket_map_t socket, int type, stream->daddr = daddr; stream->dport = dport; +#ifdef MTCP_FLASH_ID_TRAILER + stream->dst_flash_id = mtcp->ctx->flash_ctx.dst_flash_id; +#endif + ret = StreamHTInsert(mtcp->tcp_flow_table, stream); if (ret < 0) { TRACE_ERROR("Stream %d: " @@ -364,10 +368,17 @@ tcp_stream *CreateTCPStream(mtcp_manager_t mtcp, socket_map_t socket, int type, sa = (uint8_t *)&stream->saddr; da = (uint8_t *)&stream->daddr; +#ifndef MTCP_FLASH_ID_TRAILER TRACE_STREAM("CREATED NEW TCP STREAM %d: " "%u.%u.%u.%u(%d) -> %u.%u.%u.%u(%d) (ISS: %u)\n", stream->id, sa[0], sa[1], sa[2], sa[3], ntohs(stream->sport), da[0], da[1], da[2], da[3], ntohs(stream->dport), stream->sndvar->iss); +#else + TRACE_STREAM("CREATED NEW TCP STREAM %d: " + "%u.%u.%u.%u(%d) -> %u.%u.%u.%u(%d) (ISS: %u) (FLASH_IDS: %d)\n", + stream->id, sa[0], sa[1], sa[2], sa[3], ntohs(stream->sport), da[0], da[1], da[2], da[3], ntohs(stream->dport), + stream->sndvar->iss, stream->dst_flash_id); +#endif #if RATE_LIMIT_ENABLED stream->bucket = NewTokenBucket(); @@ -469,6 +480,9 @@ void DestroyTCPStream(mtcp_manager_t mtcp, tcp_stream *stream) RemoveFromControlList(mtcp, stream); RemoveFromSendList(mtcp, stream); +#ifdef MTCP_TX_ZERO_COPY + RemoveFromZCSendList(mtcp, stream); +#endif /* MTCP_TX_ZERO_COPY */ RemoveFromACKList(mtcp, stream); if (stream->on_rto_idx >= 0) @@ -511,11 +525,14 @@ void DestroyTCPStream(mtcp_manager_t mtcp, tcp_stream *stream) /* free ring buffers */ if (stream->sndvar->sndbuf) { +#ifdef MTCP_TX_ZERO_COPY + SBClear_zc(mtcp, stream->sndvar->sndbuf); +#endif /* MTCP_TX_ZERO_COPY */ SBFree(mtcp->rbm_snd, stream->sndvar->sndbuf); stream->sndvar->sndbuf = NULL; } if (stream->rcvvar->rcvbuf) { - RBFree(mtcp->rbm_rcv, stream->rcvvar->rcvbuf); + RBFree(mtcp, mtcp->rbm_rcv, stream->rcvvar->rcvbuf); stream->rcvvar->rcvbuf = NULL; } diff --git a/lib/flash/mtcp/timer.c b/lib/flash/mtcp/timer.c index c9445f1..bfd9063 100644 --- a/lib/flash/mtcp/timer.c +++ b/lib/flash/mtcp/timer.c @@ -338,6 +338,12 @@ static int HandleRTO(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_strea cur_stream->snd_nxt = cur_stream->sndvar->snd_una; if (cur_stream->state == TCP_ST_ESTABLISHED || cur_stream->state == TCP_ST_CLOSE_WAIT) { /* retransmit data at ESTABLISHED state */ +#ifdef MTCP_TX_ZERO_COPY + if (cur_stream->sndvar->sndbuf && (cur_stream->sndvar->sndbuf->available_zc_slots < MAX_TX_SLOTS || cur_stream->sndvar->sndbuf->q_front != -1)) { + AddtoZCSendList(mtcp, cur_stream); + } else +#endif /* MTCP_TX_ZERO_COPY */ + // standard path AddtoSendList(mtcp, cur_stream); } else if (cur_stream->state == TCP_ST_FIN_WAIT_1 || cur_stream->state == TCP_ST_CLOSING || @@ -352,6 +358,11 @@ static int HandleRTO(mtcp_manager_t mtcp, uint32_t cur_ts, tcp_stream *cur_strea RemoveFromControlList(mtcp, cur_stream); } cur_stream->control_list_waiting = TRUE; +#ifdef MTCP_TX_ZERO_COPY + if (cur_stream->sndvar->sndbuf && (cur_stream->sndvar->sndbuf->available_zc_slots < MAX_TX_SLOTS || cur_stream->sndvar->sndbuf->q_front != -1)) { + AddtoZCSendList(mtcp, cur_stream); + } else +#endif /* MTCP_TX_ZERO_COPY */ AddtoSendList(mtcp, cur_stream); } else { diff --git a/lib/flash/nf/flash_nf.c b/lib/flash/nf/flash_nf.c index cfd0e78..1c25c15 100644 --- a/lib/flash/nf/flash_nf.c +++ b/lib/flash/nf/flash_nf.c @@ -601,6 +601,14 @@ int flash__configure_nf(struct nf **_nf, struct config *cfg) } } +#if defined(MTCP_TX_ZERO_COPY) + cfg->zc_tracker = (uint8_t *)calloc(cfg->umem->size / cfg->umem->frame_size, sizeof(uint8_t)); + if (!cfg->zc_tracker) { + log_error("ERROR: Memory allocation failed for zero-copy tracker"); + goto out_error; + } +#endif + free(sockfd); *_nf = nf; return 0; diff --git a/lib/flash/nf/flash_txrx.c b/lib/flash/nf/flash_txrx.c index 6c380a6..3aab738 100644 --- a/lib/flash/nf/flash_txrx.c +++ b/lib/flash/nf/flash_txrx.c @@ -236,7 +236,14 @@ static inline void __complete_tx_completions(struct config *cfg, struct socket * *data_ptr = 0; } +#if defined(MTCP_TX_ZERO_COPY) + uint32_t frame_index = addr / cfg->umem->frame_size; + if (cfg->zc_tracker[frame_index] == 0) { + flash_pool__put(xsk->flash_pool, addr); + } +#else flash_pool__put(xsk->flash_pool, addr); +#endif } } diff --git a/lib/include/flash_defines.h b/lib/include/flash_defines.h index f2f6315..c062d11 100644 --- a/lib/include/flash_defines.h +++ b/lib/include/flash_defines.h @@ -82,6 +82,11 @@ struct config { int prev_size; bool track_tx_budget; int max_outstanding_tx; + +#if defined(MTCP_TX_ZERO_COPY) + uint8_t *zc_tracker; +#endif + #ifdef STATS clockid_t clock; int verbose;