nx_circulation.nx source

1// nx_circulation.nx -- reliable byte stream with flow control (TCP class). 2// 3// Biology: cardiovascular circulation regulates blood flow via heart 4// rate + vessel diameter + smooth-muscle tone. Substrate: reliable 5// ordered byte stream with backpressure (window-based flow control). 6 7import "nx_syscalls.nx" 8import "nx_tier.nx" 9 10const NX_CR_V_OK: nx_int = 0 11const NX_CR_V_BLOCKED: nx_int = 1 // would exceed window 12const NX_CR_V_CLOSED: nx_int = 2 13const NX_CR_V_NULL: nx_int = 3 14const NX_CR_V_N: nx_int = 4 15 16struct NxStream { 17 stream_id: nx_int, 18 state: nx_int, 19 bytes_in_flight: nx_size, 20 window_size: nx_size, 21 bytes_acked: nx_size, 22 bytes_sent_total: nx_size, 23 bytes_received_total: nx_size, 24} 25 26const NX_CR_BYTES: nx_int = 48 27const NX_CR_STATE_OPEN: nx_int = 0 28const NX_CR_STATE_HALF_CLOSED: nx_int = 1 29const NX_CR_STATE_CLOSED: nx_int = 2 30 31func nx_cr_v_is_valid(v: nx_int) -> nx_int { 32 if v < 0 { return 0 } 33 if v >= NX_CR_V_N { return 0 } 34 return 1 35} 36 37func nx_cr_stream_new(stream_id: nx_int, window_size: nx_size) -> *NxStream { 38 if window_size <= 0 { return 0 as *NxStream } 39 let raw: *u8 = sys_mmap(NX_CR_BYTES) 40 let s: *NxStream = raw as *NxStream 41 s.stream_id = stream_id 42 s.state = NX_CR_STATE_OPEN 43 s.bytes_in_flight = 0 44 s.window_size = window_size 45 s.bytes_acked = 0 46 s.bytes_sent_total = 0 47 s.bytes_received_total = 0 48 return s 49} 50 51func nx_cr_send(s: *NxStream, bytes: nx_size) -> nx_int { 52 if (s as i64) == 0 { return NX_CR_V_NULL } 53 if s.state == NX_CR_STATE_CLOSED { return NX_CR_V_CLOSED } 54 let new_in_flight: nx_size = s.bytes_in_flight + bytes 55 if new_in_flight > s.window_size { return NX_CR_V_BLOCKED } 56 s.bytes_in_flight = new_in_flight 57 s.bytes_sent_total = s.bytes_sent_total + bytes 58 return NX_CR_V_OK 59} 60 61func nx_cr_ack(s: *NxStream, bytes: nx_size) -> nx_int { 62 if (s as i64) == 0 { return NX_CR_V_NULL } 63 if bytes > s.bytes_in_flight { return NX_CR_V_BLOCKED } // can't ack more than in-flight 64 s.bytes_in_flight = s.bytes_in_flight - bytes 65 s.bytes_acked = s.bytes_acked + bytes 66 return NX_CR_V_OK 67} 68 69func nx_cr_receive(s: *NxStream, bytes: nx_size) -> nx_int { 70 if (s as i64) == 0 { return NX_CR_V_NULL } 71 if s.state == NX_CR_STATE_CLOSED { return NX_CR_V_CLOSED } 72 s.bytes_received_total = s.bytes_received_total + bytes 73 return NX_CR_V_OK 74} 75 76func nx_cr_available_window(s: *NxStream) -> nx_size { 77 if (s as i64) == 0 { return 0 } 78 if s.bytes_in_flight >= s.window_size { return 0 } 79 return s.window_size - s.bytes_in_flight 80} 81 82func nx_cr_close(s: *NxStream) -> nx_int { 83 if (s as i64) == 0 { return NX_CR_V_NULL } 84 s.state = NX_CR_STATE_CLOSED 85 return NX_CR_V_OK 86}