From d57a0260e85d59f51f67cd18dd69111510400f03 Mon Sep 17 00:00:00 2001 From: wxj Date: Tue, 30 Jun 2026 16:38:55 +0000 Subject: [PATCH] stable: revert multi-core, restore working single-core optimizations Keep P0-P4 optimizations: - Fragment mempool (frag_alloc/frag_free) - WS frame alloca() dynamic stack - Fragment read_offset (eliminate memmove) - nclose busy-poll (replace cond_timedwait) - Remove dead ng_tcp_out() Revert experimental multi-core changes (moved to feature/multi-core): - Worker arrays, TLS context switching - RX hash distribution, TX multi-ring collection Co-Authored-By: Claude --- src/main.c | 109 +++++++++++++++++------------------------------ src/tcp_core.c | 98 +++++++++--------------------------------- src/tcp_core.h | 15 ------- src/tcp_socket.c | 17 +++----- 4 files changed, 67 insertions(+), 172 deletions(-) diff --git a/src/main.c b/src/main.c index c19ff15..07f528a 100755 --- a/src/main.c +++ b/src/main.c @@ -179,33 +179,28 @@ int main(int argc, char *argv[]) ng_init_port(mbuf_pool); rte_eth_macaddr_get(gDpdkPortId, (struct rte_ether_addr *)gSrcMac); - /* ── Init per-worker ring buffers ── */ - int num_workers = ng_get_num_workers(); - for (int w = 1; w <= num_workers; w++) + /* ── Init ring buffers ── */ + struct inout_ring *ring = ringInstance(); + if (ring->in == NULL) { - struct inout_ring *r = ringInstanceFor(w); - char name_in[32], name_out[32]; - snprintf(name_in, sizeof(name_in), "in_ring_w%d", w); - snprintf(name_out, sizeof(name_out), "out_ring_w%d", w); - r->in = rte_ring_create(name_in, RING_SIZE, rte_socket_id(), - RING_F_SP_ENQ | RING_F_SC_DEQ); - r->out = rte_ring_create(name_out, RING_SIZE, rte_socket_id(), - RING_F_SP_ENQ | RING_F_SC_DEQ); + ring->in = rte_ring_create("in_ring", RING_SIZE, + rte_socket_id(), + RING_F_SP_ENQ | RING_F_SC_DEQ); + } + if (ring->out == NULL) + { + ring->out = rte_ring_create("out_ring", RING_SIZE, + rte_socket_id(), + RING_F_SP_ENQ | RING_F_SC_DEQ); } - /* ── Launch pkt_process workers on lcores 1..N ── */ - static struct worker_args wargs[MAX_WORKERS + 1]; + /* ── Launch pkt_process on secondary lcore ── */ unsigned lcore_id = rte_lcore_id(); - for (int w = 1; w <= num_workers; w++) - { - lcore_id = rte_get_next_lcore(lcore_id, 1, 0); - wargs[w].mbuf_pool = mbuf_pool; - wargs[w].worker_id = w; - printf("[MAIN] Launching worker %d on lcore %u\n", w, lcore_id); - rte_eal_remote_launch(pkt_process, &wargs[w], lcore_id); - } + lcore_id = rte_get_next_lcore(lcore_id, 1, 0); + printf("[MAIN] Launching pkt_process on lcore %u\n", lcore_id); + rte_eal_remote_launch(pkt_process, mbuf_pool, lcore_id); - /* Give workers time to start */ + /* Give pkt_process time to start */ rte_delay_ms(200); /* ══════════════════════════════════════════════════════════ @@ -251,41 +246,25 @@ int main(int argc, char *argv[]) int running = 1; while (running) { - /* ── DPDK RX (distribute by flow hash to workers) ── */ + /* ── DPDK RX ── */ + struct rte_mbuf *rx[BURST_SIZE]; + unsigned num_recvd = rte_eth_rx_burst(gDpdkPortId, 0, rx, BURST_SIZE); + if (num_recvd > 0) { - struct rte_mbuf *rx[BURST_SIZE]; - unsigned num_recvd = rte_eth_rx_burst(gDpdkPortId, 0, rx, BURST_SIZE); - for (unsigned i = 0; i < num_recvd; i++) - { - struct rte_ether_hdr *ehdr = rte_pktmbuf_mtod(rx[i], struct rte_ether_hdr *); - int wid = 1; /* default to worker 1 for non-IP */ - if (ehdr->ether_type == rte_cpu_to_be_16(RTE_ETHER_TYPE_IPV4)) - { - struct rte_ipv4_hdr *iph = rte_pktmbuf_mtod_offset(rx[i], - struct rte_ipv4_hdr *, sizeof(struct rte_ether_hdr)); - if (iph->next_proto_id == IPPROTO_TCP) - { - struct rte_tcp_hdr *th = (struct rte_tcp_hdr *)((uint8_t *)iph + - ((iph->version_ihl & 0x0F) << 2)); - wid = ng_compute_worker(iph->src_addr, iph->dst_addr, - th->src_port, th->dst_port); - } - } - rte_ring_sp_enqueue(ringInstanceFor(wid)->in, rx[i]); - } + rte_ring_sp_enqueue_burst(ring->in, (void **)rx, + num_recvd, NULL); } - /* ── DPDK TX (collect from all workers) ── */ - for (int w = 1; w <= num_workers; w++) + /* ── DPDK TX ── */ + struct rte_mbuf *tx[BURST_SIZE]; + unsigned nb_tx = rte_ring_sc_dequeue_burst(ring->out, + (void **)tx, BURST_SIZE, NULL); + if (nb_tx > 0) { - struct rte_mbuf *tx[BURST_SIZE]; - unsigned nb_tx = rte_ring_sc_dequeue_burst(ringInstanceFor(w)->out, - (void **)tx, BURST_SIZE, NULL); - if (nb_tx > 0) + rte_eth_tx_burst(gDpdkPortId, 0, tx, nb_tx); + for (unsigned i = 0; i < nb_tx; i++) { - rte_eth_tx_burst(gDpdkPortId, 0, tx, nb_tx); - for (unsigned i = 0; i < nb_tx; i++) - rte_pktmbuf_free(tx[i]); + rte_pktmbuf_free(tx[i]); } } @@ -369,27 +348,19 @@ int main(int argc, char *argv[]) nclose(sockfd); - /* Service TX until FIN and pending ACKs are sent (all workers) */ + /* Service TX until FIN and pending ACKs are sent */ for (int flush = 0; flush < 500; flush++) { - int any = 0; - for (int w = 1; w <= num_workers; w++) { - struct rte_mbuf *txf[BURST_SIZE]; - unsigned nb = rte_ring_sc_dequeue_burst(ringInstanceFor(w)->out, (void **)txf, BURST_SIZE, NULL); - if (nb > 0) { - rte_eth_tx_burst(gDpdkPortId, 0, txf, nb); - for (unsigned i = 0; i < nb; i++) rte_pktmbuf_free(txf[i]); - any = 1; - } + struct rte_mbuf *txf[BURST_SIZE]; + unsigned nb = rte_ring_sc_dequeue_burst(ring->out, (void **)txf, BURST_SIZE, NULL); + if (nb > 0) { + rte_eth_tx_burst(gDpdkPortId, 0, txf, nb); + for (unsigned i = 0; i < nb; i++) rte_pktmbuf_free(txf[i]); } - /* RX for any remaining packets */ + /* Also do one more RX to catch any final packets */ struct rte_mbuf *rxf[BURST_SIZE]; unsigned nr = rte_eth_rx_burst(gDpdkPortId, 0, rxf, BURST_SIZE); - if (nr > 0) { - for (unsigned i = 0; i < nr; i++) - rte_ring_sp_enqueue(ringInstanceFor(1)->in, rxf[i]); - any = 1; - } - if (!any) rte_delay_us(200); + if (nr > 0) rte_ring_sp_enqueue_burst(ring->in, (void **)rxf, nr, NULL); + if (nb == 0 && nr == 0) rte_delay_us(200); } rte_eal_mp_wait_lcore(); diff --git a/src/tcp_core.c b/src/tcp_core.c index 6d89c05..ee80548 100755 --- a/src/tcp_core.c +++ b/src/tcp_core.c @@ -25,83 +25,40 @@ uint8_t gDefaultArpMac[RTE_ETHER_ADDR_LEN] = {0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF struct lat_trace g_lat_trace[LAT_TRACE_SIZE]; int g_lat_trace_idx = 0; -static struct ng_tcp_table *g_tcp_tables[MAX_WORKERS + 1]; -static struct inout_ring *g_rings[MAX_WORKERS + 1]; -static int g_num_workers = DEFAULT_WORKERS; +static struct ng_tcp_table *tInst = NULL; +static struct inout_ring *rInst = NULL; /* Fragment mempools (replaces rte_malloc for hot-path allocations) */ struct rte_mempool *g_frag_pool = NULL; struct rte_mempool *g_ofo_pool = NULL; -/* fd → TCB index array (O(1) lookup from socket fd) — shared across all workers */ +/* fd → TCB index array (O(1) lookup from socket fd) */ static struct ng_tcp_stream *fd_to_tcb[MAX_FD_ARRAY]; -/* Current worker context (set by worker thread at startup, saved/restored by main) */ -static int g_cur_worker = 0; - -/* ── Multi-worker API ──────────────────────────────────────────── */ - -void ng_set_num_workers(int n) { if (n >= 1 && n <= MAX_WORKERS) g_num_workers = n; } -int ng_get_num_workers(void) { return g_num_workers; } -void ng_thread_set_worker(int w) { g_cur_worker = w; } -int ng_thread_get_worker(void) { return g_cur_worker; } - -/* Forward declarations for lazy init */ -static void tcpInstanceInit(int wid); -static void ringInstanceInit(int wid); +/* ══════════════════════════════════════════════════════════════════ + * TCP table singleton + * ══════════════════════════════════════════════════════════════════ */ struct ng_tcp_table *tcpInstance(void) { - int wid = g_cur_worker; - if (g_tcp_tables[wid] == NULL) tcpInstanceInit(wid); - return g_tcp_tables[wid]; + if (tInst == NULL) + { + tInst = rte_malloc("ng_tcp_table", sizeof(struct ng_tcp_table), 0); + if (tInst == NULL) + rte_exit(EXIT_FAILURE, "tcpInstance malloc failed\n"); + memset(tInst, 0, sizeof(struct ng_tcp_table)); + } + return tInst; } struct inout_ring *ringInstance(void) { - int wid = g_cur_worker; - if (g_rings[wid] == NULL) ringInstanceInit(wid); - return g_rings[wid]; -} - -/* Direct access for main thread to any worker's instances */ -struct ng_tcp_table *tcpInstanceFor(int wid) -{ - if (wid < 0 || wid > MAX_WORKERS) return NULL; - if (g_tcp_tables[wid] == NULL) tcpInstanceInit(wid); - return g_tcp_tables[wid]; -} - -struct inout_ring *ringInstanceFor(int wid) -{ - if (wid < 0 || wid > MAX_WORKERS) return NULL; - if (g_rings[wid] == NULL) ringInstanceInit(wid); - return g_rings[wid]; -} - -/* Compute which worker should handle a given 4-tuple flow */ -int ng_compute_worker(uint32_t sip, uint32_t dip, uint16_t sport, uint16_t dport) -{ - uint32_t h = sip ^ dip; - h ^= ((uint32_t)sport << 16) | dport; - h ^= (h >> 16); - return (int)(h % g_num_workers) + 1; /* workers are 1..N */ -} - -/* ── Init helpers (lazy allocation) ────────────────────────────── */ - -static void tcpInstanceInit(int wid) -{ - g_tcp_tables[wid] = rte_malloc("ng_tcp_table", sizeof(struct ng_tcp_table), 0); - if (g_tcp_tables[wid] == NULL) - rte_exit(EXIT_FAILURE, "tcpInstance[%d] malloc failed\n", wid); - memset(g_tcp_tables[wid], 0, sizeof(struct ng_tcp_table)); -} - -static void ringInstanceInit(int wid) -{ - g_rings[wid] = rte_malloc("inout_ring", sizeof(struct inout_ring), 0); - memset(g_rings[wid], 0, sizeof(struct inout_ring)); + if (rInst == NULL) + { + rInst = rte_malloc("inout_ring", sizeof(struct inout_ring), 0); + memset(rInst, 0, sizeof(struct inout_ring)); + } + return rInst; } /* ══════════════════════════════════════════════════════════════════ @@ -224,7 +181,6 @@ struct ng_tcp_stream *ng_tcp_stream_create(uint32_t sip, uint32_t dip, s->protocol = IPPROTO_TCP; s->fd = -1; s->status = NG_TCP_STATUS_CLOSED; - s->worker_id = ng_compute_worker(sip, dip, sport, dport); /* Sequence numbers (random ISN) */ uint32_t seed = time(NULL); @@ -634,7 +590,6 @@ static struct rte_mbuf *ng_send_icmp(struct rte_mempool *mbuf_pool, int ng_tcp_process(struct rte_mbuf *tcpmbuf) { - if (!tcpmbuf) { printf("[BUG] ng_tcp_process: NULL mbuf\n"); return 0; } struct rte_ipv4_hdr *iphdr = rte_pktmbuf_mtod_offset(tcpmbuf, struct rte_ipv4_hdr *, sizeof(struct rte_ether_hdr)); struct rte_tcp_hdr *tcphdr = (struct rte_tcp_hdr *)(iphdr + 1); @@ -726,18 +681,10 @@ int ng_tcp_process(struct rte_mbuf *tcpmbuf) int pkt_process(void *arg) { - struct worker_args *wa = (struct worker_args *)arg; - struct rte_mempool *mbuf_pool = wa->mbuf_pool; - int worker_id = wa->worker_id; - - /* Set thread-local worker for tcpInstance()/ringInstance() routing */ - ng_thread_set_worker(worker_id); - struct sched_param sp = { .sched_priority = 90 }; sched_setscheduler(0, SCHED_FIFO, &sp); - + struct rte_mempool *mbuf_pool = (struct rte_mempool *)arg; struct inout_ring *ring = ringInstance(); - printf("[WORKER %d] pkt_process started on lcore %d\n", worker_id, rte_lcore_id()); while (1) @@ -745,7 +692,6 @@ int pkt_process(void *arg) /* ── Phase 1: RX processing ── */ struct rte_mbuf *mbufs[BURST_SIZE]; - if (!ring || !ring->in) { printf("\[BUG\] pkt_process: NULL ring\n"); rte_delay_ms(1000); continue; } unsigned num_recvd = rte_ring_mc_dequeue_burst(ring->in, (void **)mbufs, BURST_SIZE, NULL); @@ -818,8 +764,6 @@ int pkt_process(void *arg) uint64_t now = rte_get_timer_cycles(); struct ng_tcp_table *table = tcpInstance(); struct ng_tcp_stream *s = table->active_head; - static int dbg_active = 0; - if (s && !dbg_active) { printf("[WORKER %d] active_head non-null, processing\n", worker_id); dbg_active = 1; } while (s) { struct ng_tcp_stream *next = s->active_next; diff --git a/src/tcp_core.h b/src/tcp_core.h index c19d41e..690051f 100755 --- a/src/tcp_core.h +++ b/src/tcp_core.h @@ -38,8 +38,6 @@ #define TIMER_RESOLUTION_CYCLES 120000000000ULL #define TCP_HTABLE_SIZE 256 /* MUST be power of 2 */ -#define MAX_WORKERS 4 /* max TCP processing lcores */ -#define DEFAULT_WORKERS 1 /* default: single worker (backward compat) */ #define TCP_HTABLE_MASK (TCP_HTABLE_SIZE - 1) #define TCP_MAX_SEQ 4294967295U @@ -191,7 +189,6 @@ struct ng_tcp_stream { /* ── Identity ── */ int fd; - int worker_id; /* owning worker lcore (1..N) */ uint32_t sip; uint32_t dip; uint16_t sport; @@ -472,18 +469,6 @@ void ng_frag_pool_init(void); void ng_init_port(struct rte_mempool *mbuf_pool); int pkt_process(void *arg); -/* Worker startup arguments */ -struct worker_args { struct rte_mempool *mbuf_pool; int worker_id; }; - -/* Multi-worker support */ -void ng_set_num_workers(int n); -int ng_get_num_workers(void); -void ng_thread_set_worker(int wid); -int ng_thread_get_worker(void); -struct ng_tcp_table *tcpInstanceFor(int wid); -struct inout_ring *ringInstanceFor(int wid); -int ng_compute_worker(uint32_t sip, uint32_t dip, uint16_t sport, uint16_t dport); - /* fd → TCB index */ void ng_tcp_fd_set(int fd, struct ng_tcp_stream *s); void ng_tcp_fd_clear(int fd); diff --git a/src/tcp_socket.c b/src/tcp_socket.c index 81dc079..f812021 100755 --- a/src/tcp_socket.c +++ b/src/tcp_socket.c @@ -243,22 +243,17 @@ int nconnect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) srand(seed); } s->sport = htons((rand() % 16384) + 49152); - s->worker_id = ng_compute_worker(s->sip, s->dip, s->sport, s->dport); rte_memcpy(s->localmac, gSrcMac, RTE_ETHER_ADDR_LEN); - /* Switch to stream's worker context for hash table operations */ - int saved_worker = ng_thread_get_worker(); - ng_thread_set_worker(s->worker_id); ng_tcp_hash_remove(s); ng_tcp_hash_insert(s); - ng_tcp_stream_activate(s); - s->status = NG_TCP_STATUS_SYN_SENT; - ng_thread_set_worker(saved_worker); s->iss = rand() % TCP_MAX_SEQ; s->snd_una = s->iss; s->snd_nxt = s->iss; s->snd_max = s->iss; + s->status = NG_TCP_STATUS_SYN_SENT; + ng_tcp_stream_activate(s); printf("[nconnect] Connecting to %s:%d ...\n", inet_ntoa(in->sin_addr), ntohs(s->dport)); @@ -283,7 +278,7 @@ int nconnect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += 5; - struct inout_ring *ring = ringInstanceFor(s->worker_id); + struct inout_ring *ring = ringInstance(); int spin_count = 0; while (s->status != NG_TCP_STATUS_ESTABLISHED) @@ -393,7 +388,7 @@ ssize_t nrecv(int sockfd, void *buf, size_t len, int flags) return -1; /* Service RX/TX */ - struct inout_ring *ring = ringInstanceFor(s->worker_id); + struct inout_ring *ring = ringInstance(); struct rte_mbuf *tx[BURST_SIZE]; unsigned nb_tx = rte_ring_sc_dequeue_burst(ring->out, (void **)tx, BURST_SIZE, NULL); if (nb_tx > 0) @@ -498,7 +493,7 @@ int nclose(int fd) clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += 5; int spin = 0; - struct inout_ring *ring = ringInstanceFor(s->worker_id); + struct inout_ring *ring = ringInstance(); while (s->status != NG_TCP_STATUS_CLOSED && s->status != NG_TCP_STATUS_TIME_WAIT) { @@ -534,7 +529,7 @@ int nclose(int fd) clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += 5; int spin = 0; - struct inout_ring *ring = ringInstanceFor(s->worker_id); + struct inout_ring *ring = ringInstance(); while (s->status != NG_TCP_STATUS_CLOSED) { /* Service TX */