nx_rpc_test.nx source
↩ module page · 162 lines · 6198 B
1// nx_rpc_test.nx -- end-to-end typed RPC over loopback.
2//
3// Server registers 2 method handlers:
4// method 1 (echo): copy request -> response unchanged
5// method 2 (length): respond with the request length as 8-byte i64
6//
7// Client makes 3 calls (echo "ping" + echo "hello world" + length
8// "this is a longer message" -> 24), verifies each response, then
9// sends shutdown sentinel.
10
11import "nx_kernel_v2.nx"
12import "nx_log.nx"
13import "nx_atom.nx"
14import "nx_thread.nx"
15import "nx_socket.nx"
16import "nx_net_chan.nx"
17import "nx_rpc.nx"
18
19const NX_PORT: i64 = 38927
20
21struct RpcTestState {
22 done_flag: i64,
23 diag: i64,
24}
25
26func _sleep_ms(ms: i64) -> i64 {
27 let ts_raw: *u8 = sys_mmap(16)
28 let ts: *i64 = ts_raw as *i64
29 ts[0] = ms / 1000
30 ts[1] = (ms % 1000) * 1000000
31 return __syscall(SYS_CLOCK_NANOSLEEP, 1, 0, ts as i64, 0, 0, 0)
32}
33
34// Server-side handler: echo (copy req -> resp). Returns resp_n.
35func handler_echo(req: *u8, req_n: i64, resp: *u8, max_resp: i64) -> i64 {
36 if req_n > max_resp { return 0 - 1 }
37 var i: i64 = 0
38 while i < req_n { resp[i] = req[i]; i = i + 1 }
39 return req_n
40}
41
42// Server-side handler: length (write 8-byte big-endian length).
43func handler_length(req: *u8, req_n: i64, resp: *u8, max_resp: i64) -> i64 {
44 if max_resp < 8 { return 0 - 1 }
45 let v: i64 = req_n
46 resp[0] = ((v >> 56) & 0xFF) as u8
47 resp[1] = ((v >> 48) & 0xFF) as u8
48 resp[2] = ((v >> 40) & 0xFF) as u8
49 resp[3] = ((v >> 32) & 0xFF) as u8
50 resp[4] = ((v >> 24) & 0xFF) as u8
51 resp[5] = ((v >> 16) & 0xFF) as u8
52 resp[6] = ((v >> 8) & 0xFF) as u8
53 resp[7] = (v & 0xFF) as u8
54 return 8
55}
56
57// Client thread: connect, make 3 RPC calls, verify responses, signal done.
58func rpc_client(arg: *u8) -> i64 {
59 let s: *RpcTestState = arg as *RpcTestState
60 let done_addr: *i64 = (arg as *i64)
61
62 _sleep_ms(200)
63 let fd: i64 = nx_sock_socket(NX_AF_INET, NX_SOCK_STREAM, NX_IPPROTO_TCP)
64 if fd < 0 { s.diag = 1; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
65 let addr: *u8 = sys_mmap(16)
66 nx_sock_sin_init(addr, 0x0100007F, NX_PORT)
67 if nx_sock_connect(fd, addr, 16) < 0 {
68 s.diag = 2; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0
69 }
70 let nc: *NxNetChan = nx_net_chan_from_fd(fd)
71
72 // Call 1: echo "ping" (4 bytes)
73 let req1: *u8 = sys_mmap(16)
74 req1[0] = 0x70; req1[1] = 0x69; req1[2] = 0x6E; req1[3] = 0x67
75 let resp1: *u8 = sys_mmap(64)
76 let n1: i64 = nx_rpc_call(nc, 1, req1, 4, resp1, 64)
77 if n1 != 4 { s.diag = 10; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
78 if resp1[0] != 0x70 { s.diag = 11; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
79 if resp1[3] != 0x67 { s.diag = 14; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
80
81 // Call 2: echo "hello world" (11 bytes)
82 let req2: *u8 = sys_mmap(32)
83 req2[0] = 0x68; req2[1] = 0x65; req2[2] = 0x6C; req2[3] = 0x6C
84 req2[4] = 0x6F; req2[5] = 0x20; req2[6] = 0x77; req2[7] = 0x6F
85 req2[8] = 0x72; req2[9] = 0x6C; req2[10] = 0x64
86 let resp2: *u8 = sys_mmap(64)
87 let n2: i64 = nx_rpc_call(nc, 1, req2, 11, resp2, 64)
88 if n2 != 11 { s.diag = 20; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
89 if resp2[10] != 0x64 { s.diag = 21; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
90
91 // Call 3: length of a 24-byte request -> should respond with 24.
92 let req3: *u8 = sys_mmap(64)
93 var k: i64 = 0
94 while k < 24 { req3[k] = (0x41 + k) as u8; k = k + 1 }
95 let resp3: *u8 = sys_mmap(64)
96 let n3: i64 = nx_rpc_call(nc, 2, req3, 24, resp3, 64)
97 if n3 != 8 { s.diag = 30; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
98 // unpack BE i64 from resp3
99 let b7: i64 = resp3[7]
100 if b7 != 24 { s.diag = 31; nx_atom_store_i64(done_addr, -1, NX_MO_SEQ_CST); return 0 }
101
102 // Shutdown server cleanly.
103 nx_rpc_shutdown_call(nc)
104 nx_net_chan_close(nc)
105 s.diag = 1
106 nx_atom_store_i64(done_addr, 1, NX_MO_SEQ_CST)
107 return 0
108}
109
110func main() -> nx_exit {
111 println("=== nx_rpc typed RPC smoke (3 calls) ===" as *u8)
112
113 let state_raw: *u8 = sys_mmap(64)
114 let s: *RpcTestState = state_raw as *RpcTestState
115 s.done_flag = 0; s.diag = 0
116
117 // Listener
118 let lfd: i64 = nx_sock_socket(NX_AF_INET, NX_SOCK_STREAM, NX_IPPROTO_TCP)
119 if lfd < 0 { println("FAIL: socket" as *u8); return 1 }
120 nx_sock_reuseaddr(lfd)
121 let laddr: *u8 = sys_mmap(16)
122 nx_sock_sin_init(laddr, 0, NX_PORT)
123 if nx_sock_bind(lfd, laddr, 16) < 0 { println("FAIL: bind" as *u8); return 2 }
124 if nx_sock_listen(lfd, 1) < 0 { println("FAIL: listen" as *u8); return 3 }
125
126 // Spawn client thread
127 let tid: i64 = nx_thread_spawn_fn(rpc_client, state_raw, 65536)
128 if tid <= 0 { println("FAIL: spawn" as *u8); return 4 }
129
130 // Accept connection + bind to RPC server
131 let cfd: i64 = nx_sock_accept(lfd, 0 as *u8, 0 as *i64)
132 if cfd < 0 { println("FAIL: accept" as *u8); return 5 }
133 let nc: *NxNetChan = nx_net_chan_from_fd(cfd)
134 let server: *NxRpcServer = nx_rpc_server_new(nc)
135 nx_rpc_register(server, 1, handler_echo)
136 nx_rpc_register(server, 2, handler_length)
137
138 // Run server loop until shutdown
139 let scratch: *u8 = sys_mmap(4096)
140 nx_rpc_server_run(server, scratch, 4096)
141
142 nx_net_chan_close(nc)
143 sys_close(lfd)
144
145 // Wait for client verdict
146 let done_addr: *i64 = (state_raw as *i64)
147 var spins: i64 = 0
148 while nx_atom_load_i64(done_addr, NX_MO_SEQ_CST) == 0 {
149 nx_thread_yield()
150 spins = spins + 1
151 if spins > 100000000 { println("FAIL: client timeout" as *u8); return 6 }
152 }
153 if nx_atom_load_i64(done_addr, NX_MO_SEQ_CST) != 1 {
154 println("FAIL: client diag" as *u8); print_i64(s.diag); println("" as *u8); return 7
155 }
156 println("PASS: 3 typed RPC calls completed end-to-end" as *u8)
157 println(" echo 4B -> 4B response (bit-exact)" as *u8)
158 println(" echo 11B -> 11B response (bit-exact)" as *u8)
159 println(" length(24B request) -> 24 in last byte of i64 BE response" as *u8)
160 println("Server dispatched 2 distinct method_ids; clean sentinel shutdown." as *u8)
161 return 0
162}