wxj/dpdk-ws-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

stable: revert multi-core, restore working single-core optimizations

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
wxj
2026-06-30 16:38:55 +00:00
co-authored by Claude
parent 5e57d06b1f
commit d57a0260e8
4 changed files with 67 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main.c
+26 -55
View File
@@ -179,33 +179,28 @@ int main(int argc, char *argv[])
ng_init_port(mbuf_pool); ng_init_port(mbuf_pool);
rte_eth_macaddr_get(gDpdkPortId, (struct rte_ether_addr *)gSrcMac); rte_eth_macaddr_get(gDpdkPortId, (struct rte_ether_addr *)gSrcMac);
/* ── Init per-worker ring buffers ── */ /* ── Init ring buffers ── */
int num_workers = ng_get_num_workers(); struct inout_ring *ring = ringInstance();
for (int w = 1; w <= num_workers; w++) if (ring->in == NULL)
{ {
struct inout_ring *r = ringInstanceFor(w); ring->in = rte_ring_create("in_ring", RING_SIZE,
char name_in[32], name_out[32]; rte_socket_id(),
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); RING_F_SP_ENQ | RING_F_SC_DEQ);
r->out = rte_ring_create(name_out, RING_SIZE, rte_socket_id(), }
if (ring->out == NULL)
{
ring->out = rte_ring_create("out_ring", RING_SIZE,
rte_socket_id(),
RING_F_SP_ENQ | RING_F_SC_DEQ); RING_F_SP_ENQ | RING_F_SC_DEQ);
} }
/* ── Launch pkt_process workers on lcores 1..N ── */ /* ── Launch pkt_process on secondary lcore ── */
static struct worker_args wargs[MAX_WORKERS + 1];
unsigned lcore_id = rte_lcore_id(); unsigned lcore_id = rte_lcore_id();
for (int w = 1; w <= num_workers; w++)
{
lcore_id = rte_get_next_lcore(lcore_id, 1, 0); lcore_id = rte_get_next_lcore(lcore_id, 1, 0);
wargs[w].mbuf_pool = mbuf_pool; printf("[MAIN] Launching pkt_process on lcore %u\n", lcore_id);
wargs[w].worker_id = w; rte_eal_remote_launch(pkt_process, mbuf_pool, lcore_id);
printf("[MAIN] Launching worker %d on lcore %u\n", w, lcore_id);
rte_eal_remote_launch(pkt_process, &wargs[w], lcore_id);
}
/* Give workers time to start */ /* Give pkt_process time to start */
rte_delay_ms(200); rte_delay_ms(200);
/* ══════════════════════════════════════════════════════════ /* ══════════════════════════════════════════════════════════
@@ -251,40 +246,24 @@ int main(int argc, char *argv[])
int running = 1; int running = 1;
while (running) while (running)
{ {
/* ── DPDK RX (distribute by flow hash to workers) ── */ /* ── DPDK RX ── */
{
struct rte_mbuf *rx[BURST_SIZE]; struct rte_mbuf *rx[BURST_SIZE];
unsigned num_recvd = rte_eth_rx_burst(gDpdkPortId, 0, rx, BURST_SIZE); unsigned num_recvd = rte_eth_rx_burst(gDpdkPortId, 0, rx, BURST_SIZE);
for (unsigned i = 0; i < num_recvd; i++) if (num_recvd > 0)
{ {
struct rte_ether_hdr *ehdr = rte_pktmbuf_mtod(rx[i], struct rte_ether_hdr *); rte_ring_sp_enqueue_burst(ring->in, (void **)rx,
int wid = 1; /* default to worker 1 for non-IP */ num_recvd, NULL);
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]);
}
} }
/* ── DPDK TX (collect from all workers) ── */ /* ── DPDK TX ── */
for (int w = 1; w <= num_workers; w++)
{
struct rte_mbuf *tx[BURST_SIZE]; struct rte_mbuf *tx[BURST_SIZE];
unsigned nb_tx = rte_ring_sc_dequeue_burst(ringInstanceFor(w)->out, unsigned nb_tx = rte_ring_sc_dequeue_burst(ring->out,
(void **)tx, BURST_SIZE, NULL); (void **)tx, BURST_SIZE, NULL);
if (nb_tx > 0) if (nb_tx > 0)
{ {
rte_eth_tx_burst(gDpdkPortId, 0, tx, nb_tx); rte_eth_tx_burst(gDpdkPortId, 0, tx, nb_tx);
for (unsigned i = 0; i < nb_tx; i++) 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); 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++) { for (int flush = 0; flush < 500; flush++) {
int any = 0;
for (int w = 1; w <= num_workers; w++) {
struct rte_mbuf *txf[BURST_SIZE]; struct rte_mbuf *txf[BURST_SIZE];
unsigned nb = rte_ring_sc_dequeue_burst(ringInstanceFor(w)->out, (void **)txf, BURST_SIZE, NULL); unsigned nb = rte_ring_sc_dequeue_burst(ring->out, (void **)txf, BURST_SIZE, NULL);
if (nb > 0) { if (nb > 0) {
rte_eth_tx_burst(gDpdkPortId, 0, txf, nb); rte_eth_tx_burst(gDpdkPortId, 0, txf, nb);
for (unsigned i = 0; i < nb; i++) rte_pktmbuf_free(txf[i]); for (unsigned i = 0; i < nb; i++) rte_pktmbuf_free(txf[i]);
any = 1;
} }
} /* Also do one more RX to catch any final packets */
/* RX for any remaining packets */
struct rte_mbuf *rxf[BURST_SIZE]; struct rte_mbuf *rxf[BURST_SIZE];
unsigned nr = rte_eth_rx_burst(gDpdkPortId, 0, rxf, BURST_SIZE); unsigned nr = rte_eth_rx_burst(gDpdkPortId, 0, rxf, BURST_SIZE);
if (nr > 0) { if (nr > 0) rte_ring_sp_enqueue_burst(ring->in, (void **)rxf, nr, NULL);
for (unsigned i = 0; i < nr; i++) if (nb == 0 && nr == 0) rte_delay_us(200);
rte_ring_sp_enqueue(ringInstanceFor(1)->in, rxf[i]);
any = 1;
}
if (!any) rte_delay_us(200);
} }
rte_eal_mp_wait_lcore(); rte_eal_mp_wait_lcore();
src/tcp_core.c
+19 -75
View File
@@ -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]; struct lat_trace g_lat_trace[LAT_TRACE_SIZE];
int g_lat_trace_idx = 0; int g_lat_trace_idx = 0;
static struct ng_tcp_table *g_tcp_tables[MAX_WORKERS + 1]; static struct ng_tcp_table *tInst = NULL;
static struct inout_ring *g_rings[MAX_WORKERS + 1]; static struct inout_ring *rInst = NULL;
static int g_num_workers = DEFAULT_WORKERS;
/* Fragment mempools (replaces rte_malloc for hot-path allocations) */ /* Fragment mempools (replaces rte_malloc for hot-path allocations) */
struct rte_mempool *g_frag_pool = NULL; struct rte_mempool *g_frag_pool = NULL;
struct rte_mempool *g_ofo_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]; 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; * TCP table singleton
* ══════════════════════════════════════════════════════════════════ */
/* ── 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);
struct ng_tcp_table *tcpInstance(void) struct ng_tcp_table *tcpInstance(void)
{ {
int wid = g_cur_worker; if (tInst == NULL)
if (g_tcp_tables[wid] == NULL) tcpInstanceInit(wid); {
return g_tcp_tables[wid]; 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) struct inout_ring *ringInstance(void)
{ {
int wid = g_cur_worker; if (rInst == NULL)
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; rInst = rte_malloc("inout_ring", sizeof(struct inout_ring), 0);
if (g_tcp_tables[wid] == NULL) tcpInstanceInit(wid); memset(rInst, 0, sizeof(struct inout_ring));
return g_tcp_tables[wid];
} }
return rInst;
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));
} }
/* ══════════════════════════════════════════════════════════════════ /* ══════════════════════════════════════════════════════════════════
@@ -224,7 +181,6 @@ struct ng_tcp_stream *ng_tcp_stream_create(uint32_t sip, uint32_t dip,
s->protocol = IPPROTO_TCP; s->protocol = IPPROTO_TCP;
s->fd = -1; s->fd = -1;
s->status = NG_TCP_STATUS_CLOSED; s->status = NG_TCP_STATUS_CLOSED;
s->worker_id = ng_compute_worker(sip, dip, sport, dport);
/* Sequence numbers (random ISN) */ /* Sequence numbers (random ISN) */
uint32_t seed = time(NULL); 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) 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 *iphdr = rte_pktmbuf_mtod_offset(tcpmbuf,
struct rte_ipv4_hdr *, sizeof(struct rte_ether_hdr)); struct rte_ipv4_hdr *, sizeof(struct rte_ether_hdr));
struct rte_tcp_hdr *tcphdr = (struct rte_tcp_hdr *)(iphdr + 1); 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) 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 }; struct sched_param sp = { .sched_priority = 90 };
sched_setscheduler(0, SCHED_FIFO, &sp); sched_setscheduler(0, SCHED_FIFO, &sp);
struct rte_mempool *mbuf_pool = (struct rte_mempool *)arg;
struct inout_ring *ring = ringInstance(); struct inout_ring *ring = ringInstance();
printf("[WORKER %d] pkt_process started on lcore %d\n", worker_id, rte_lcore_id());
while (1) while (1)
@@ -745,7 +692,6 @@ int pkt_process(void *arg)
/* ── Phase 1: RX processing ── */ /* ── Phase 1: RX processing ── */
struct rte_mbuf *mbufs[BURST_SIZE]; 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, unsigned num_recvd = rte_ring_mc_dequeue_burst(ring->in,
(void **)mbufs, BURST_SIZE, NULL); (void **)mbufs, BURST_SIZE, NULL);
@@ -818,8 +764,6 @@ int pkt_process(void *arg)
uint64_t now = rte_get_timer_cycles(); uint64_t now = rte_get_timer_cycles();
struct ng_tcp_table *table = tcpInstance(); struct ng_tcp_table *table = tcpInstance();
struct ng_tcp_stream *s = table->active_head; 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) while (s)
{ {
struct ng_tcp_stream *next = s->active_next; struct ng_tcp_stream *next = s->active_next;
src/tcp_core.h
-15
View File
@@ -38,8 +38,6 @@
#define TIMER_RESOLUTION_CYCLES 120000000000ULL #define TIMER_RESOLUTION_CYCLES 120000000000ULL
#define TCP_HTABLE_SIZE 256 /* MUST be power of 2 */ #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_HTABLE_MASK (TCP_HTABLE_SIZE - 1)
#define TCP_MAX_SEQ 4294967295U #define TCP_MAX_SEQ 4294967295U
@@ -191,7 +189,6 @@ struct ng_tcp_stream
{ {
/* ── Identity ── */ /* ── Identity ── */
int fd; int fd;
int worker_id; /* owning worker lcore (1..N) */
uint32_t sip; uint32_t sip;
uint32_t dip; uint32_t dip;
uint16_t sport; uint16_t sport;
@@ -472,18 +469,6 @@ void ng_frag_pool_init(void);
void ng_init_port(struct rte_mempool *mbuf_pool); void ng_init_port(struct rte_mempool *mbuf_pool);
int pkt_process(void *arg); 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 */ /* fd → TCB index */
void ng_tcp_fd_set(int fd, struct ng_tcp_stream *s); void ng_tcp_fd_set(int fd, struct ng_tcp_stream *s);
void ng_tcp_fd_clear(int fd); void ng_tcp_fd_clear(int fd);
src/tcp_socket.c
+6 -11
View File
@@ -243,22 +243,17 @@ int nconnect(int sockfd, const struct sockaddr *addr, socklen_t addrlen)
srand(seed); srand(seed);
} }
s->sport = htons((rand() % 16384) + 49152); 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); 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_remove(s);
ng_tcp_hash_insert(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->iss = rand() % TCP_MAX_SEQ;
s->snd_una = s->iss; s->snd_una = s->iss;
s->snd_nxt = s->iss; s->snd_nxt = s->iss;
s->snd_max = 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", printf("[nconnect] Connecting to %s:%d ...\n",
inet_ntoa(in->sin_addr), ntohs(s->dport)); 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); clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += 5; deadline.tv_sec += 5;
struct inout_ring *ring = ringInstanceFor(s->worker_id); struct inout_ring *ring = ringInstance();
int spin_count = 0; int spin_count = 0;
while (s->status != NG_TCP_STATUS_ESTABLISHED) 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; return -1;
/* Service RX/TX */ /* Service RX/TX */
struct inout_ring *ring = ringInstanceFor(s->worker_id); struct inout_ring *ring = ringInstance();
struct rte_mbuf *tx[BURST_SIZE]; struct rte_mbuf *tx[BURST_SIZE];
unsigned nb_tx = rte_ring_sc_dequeue_burst(ring->out, (void **)tx, BURST_SIZE, NULL); unsigned nb_tx = rte_ring_sc_dequeue_burst(ring->out, (void **)tx, BURST_SIZE, NULL);
if (nb_tx > 0) if (nb_tx > 0)
@@ -498,7 +493,7 @@ int nclose(int fd)
clock_gettime(CLOCK_REALTIME, &deadline); clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += 5; deadline.tv_sec += 5;
int spin = 0; int spin = 0;
struct inout_ring *ring = ringInstanceFor(s->worker_id); struct inout_ring *ring = ringInstance();
while (s->status != NG_TCP_STATUS_CLOSED && while (s->status != NG_TCP_STATUS_CLOSED &&
s->status != NG_TCP_STATUS_TIME_WAIT) s->status != NG_TCP_STATUS_TIME_WAIT)
{ {
@@ -534,7 +529,7 @@ int nclose(int fd)
clock_gettime(CLOCK_REALTIME, &deadline); clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += 5; deadline.tv_sec += 5;
int spin = 0; int spin = 0;
struct inout_ring *ring = ringInstanceFor(s->worker_id); struct inout_ring *ring = ringInstance();
while (s->status != NG_TCP_STATUS_CLOSED) while (s->status != NG_TCP_STATUS_CLOSED)
{ {
/* Service TX */ /* Service TX */