nx_socket_pingpong_test.nx source
↩ module page · 142 lines · 5943 B
1// nx_socket_pingpong_test.nx -- prove TCP listen/accept/connect
2// loopback works via real Linux syscalls. Foundation for L9
3// distributed MIMD (nx_net_chan, nx_remote_spawn, nx_rpc).
4//
5// Design: parent thread is the SERVER (listens + accepts + echoes
6// "ping" -> "pong" once + closes); a spawned child thread is the
7// CLIENT (sleeps briefly to let parent bind, then connects + sends
8// "ping" + receives "pong" + closes + sets done flag). Main spin-
9// waits on the done flag.
10//
11// Uses port NX_PORT (chosen high to avoid privileged + commonly-used);
12// if port is already in use the bind() returns -EADDRINUSE which we
13// surface as a clear failure code rather than hanging.
14
15import "nx_kernel_v2.nx"
16import "nx_log.nx"
17import "nx_atom.nx"
18import "nx_thread.nx"
19import "nx_socket.nx"
20
21const NX_PORT: i64 = 38917 // arbitrary high port
22
23// Shared state passed to client thread.
24struct PingPongState {
25 done_flag: i64, // atomic; client sets to 1 on success, -1 on fail
26 verdict: i64, // pass=1, fail=N for diagnostic
27}
28
29func _sleep_ms(ms: i64) -> i64 {
30 let ts_raw: *u8 = sys_mmap(16)
31 let ts: *i64 = ts_raw as *i64
32 ts[0] = ms / 1000 // tv_sec
33 ts[1] = (ms % 1000) * 1000000 // tv_nsec
34 return __syscall(SYS_CLOCK_NANOSLEEP, 1, 0, ts as i64, 0, 0, 0)
35}
36
37// Client thread. ctx is *PingPongState.
38func client_main(arg: *u8) -> i64 {
39 let s: *PingPongState = arg as *PingPongState
40 let done_addr: *i64 = (arg as *i64)
41
42 // Give server time to bind+listen+accept. In a real system the
43 // client would retry connect-with-backoff; the smoke uses a
44 // generous fixed sleep.
45 _sleep_ms(200)
46
47 let fd: i64 = nx_sock_socket(NX_AF_INET, NX_SOCK_STREAM, NX_IPPROTO_TCP)
48 if fd < 0 { s.verdict = 10; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
49
50 let addr_raw: *u8 = sys_mmap(16)
51 // 127.0.0.1 as little-endian-stored u32 bytes 0x7F.0.0.1 in net
52 // order is 7F 00 00 01. inet_aton would convert "127.0.0.1"
53 // into a packed BE u32; we build it inline since we know the
54 // constant.
55 nx_sock_sin_init(addr_raw, 0x0100007F, NX_PORT)
56 let cr: i64 = nx_sock_connect(fd, addr_raw, 16)
57 if cr < 0 { s.verdict = 20; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
58
59 let buf: *u8 = sys_mmap(64)
60 buf[0] = 0x70; buf[1] = 0x69; buf[2] = 0x6E; buf[3] = 0x67 // "ping"
61 let sr: i64 = __syscall(NX_SYS_SENDTO, fd, buf as i64, 4, 0, 0, 0)
62 if sr != 4 { s.verdict = 30; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
63
64 let resp: *u8 = sys_mmap(64)
65 let rr: i64 = __syscall(NX_SYS_RECVFROM, fd, resp as i64, 64, 0, 0, 0)
66 if rr != 4 { s.verdict = 40; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
67 if resp[0] != 0x70 { s.verdict = 41; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
68 if resp[1] != 0x6F { s.verdict = 42; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
69 if resp[2] != 0x6E { s.verdict = 43; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
70 if resp[3] != 0x67 { s.verdict = 44; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
71
72 nx_sock_shutdown(fd, NX_SHUT_RDWR)
73 sys_close(fd)
74
75 s.verdict = 1
76 nx_atom_store_i64(done_addr, 1, NX_MO_SEQ_CST)
77 return 0
78}
79
80func main() -> nx_exit {
81 println("=== nx_socket TCP loopback ping/pong ===" as *u8)
82 println("port:" as *u8); print_i64(NX_PORT); println("" as *u8)
83
84 let state_raw: *u8 = sys_mmap(64)
85 let s: *PingPongState = state_raw as *PingPongState
86 s.done_flag = 0
87 s.verdict = 0
88
89 // Build the listener BEFORE spawning client so accept is ready
90 // by the time client tries to connect.
91 let lfd: i64 = nx_sock_socket(NX_AF_INET, NX_SOCK_STREAM, NX_IPPROTO_TCP)
92 if lfd < 0 { println("FAIL: socket" as *u8); return 1 }
93 nx_sock_reuseaddr(lfd)
94 let laddr: *u8 = sys_mmap(16)
95 nx_sock_sin_init(laddr, 0, NX_PORT) // bind to 0.0.0.0
96 let br: i64 = nx_sock_bind(lfd, laddr, 16)
97 if br < 0 { println("FAIL: bind (port in use?)" as *u8); print_i64(br); println("" as *u8); return 2 }
98 let lr: i64 = nx_sock_listen(lfd, 1)
99 if lr < 0 { println("FAIL: listen" as *u8); return 3 }
100
101 // Spawn client thread NOW so accept() below can pair with its connect.
102 let tid: i64 = nx_thread_spawn_fn(client_main, state_raw, 65536)
103 if tid <= 0 { println("FAIL: spawn" as *u8); return 4 }
104
105 // Accept on parent.
106 let cfd: i64 = nx_sock_accept(lfd, 0 as *u8, 0 as *i64)
107 if cfd < 0 { println("FAIL: accept" as *u8); print_i64(cfd); println("" as *u8); return 5 }
108
109 let recv_buf: *u8 = sys_mmap(64)
110 let n: i64 = __syscall(NX_SYS_RECVFROM, cfd, recv_buf as i64, 64, 0, 0, 0)
111 if n != 4 { println("FAIL: server recv != 4" as *u8); return 6 }
112 if recv_buf[0] != 0x70 { println("FAIL: not 'p'" as *u8); return 7 }
113
114 // Reply with "pong".
115 let send_buf: *u8 = sys_mmap(64)
116 send_buf[0] = 0x70; send_buf[1] = 0x6F; send_buf[2] = 0x6E; send_buf[3] = 0x67
117 let sr: i64 = __syscall(NX_SYS_SENDTO, cfd, send_buf as i64, 4, 0, 0, 0)
118 if sr != 4 { println("FAIL: server send != 4" as *u8); return 8 }
119
120 nx_sock_shutdown(cfd, NX_SHUT_RDWR)
121 sys_close(cfd)
122 sys_close(lfd)
123
124 // Wait for client to finish.
125 let done_addr: *i64 = (state_raw as *i64)
126 var spins: i64 = 0
127 while nx_atom_load_i64(done_addr, NX_MO_SEQ_CST) == 0 {
128 nx_thread_yield()
129 spins = spins + 1
130 if spins > 100000000 { println("FAIL: client timeout" as *u8); return 9 }
131 }
132
133 let final_state: i64 = nx_atom_load_i64(done_addr, NX_MO_SEQ_CST)
134 if final_state != 1 {
135 println("FAIL: client verdict not PASS" as *u8)
136 print_i64(s.verdict); println("" as *u8)
137 return 10
138 }
139 println("PASS: TCP ping/pong over loopback completed" as *u8)
140 println("Foundation for L9 distributed MIMD: nx_net_chan + nx_remote_spawn." as *u8)
141 return 0
142}