1 use crate::wasi::clocks::monotonic_clock; 2 use crate::wasi::io::poll::{self, Pollable}; 3 use crate::wasi::io::streams::{InputStream, OutputStream, StreamError}; 4 use crate::wasi::sockets::instance_network; 5 use crate::wasi::sockets::ip_name_lookup; 6 use crate::wasi::sockets::network::{ 7 ErrorCode, IpAddress, IpAddressFamily, IpSocketAddress, Ipv4SocketAddress, Ipv6SocketAddress, 8 Network, 9 }; 10 use crate::wasi::sockets::tcp::TcpSocket; 11 use crate::wasi::sockets::udp::{ 12 IncomingDatagram, IncomingDatagramStream, OutgoingDatagram, OutgoingDatagramStream, UdpSocket, 13 }; 14 use crate::wasi::sockets::{tcp_create_socket, udp_create_socket}; 15 use std::ops::Range; 16 17 const TIMEOUT_NS: u64 = 1_000_000_000; 18 19 impl Pollable { 20 pub fn block_until(&self, timeout: &Pollable) -> Result<(), ErrorCode> { 21 let ready = poll::poll(&[self, timeout]); 22 assert!(ready.len() > 0); 23 match ready[0] { 24 0 => Ok(()), 25 1 => Err(ErrorCode::Timeout), 26 _ => unreachable!(), 27 } 28 } 29 } 30 31 impl OutputStream { 32 pub fn blocking_write_util(&self, mut bytes: &[u8]) -> Result<(), StreamError> { 33 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 34 let pollable = self.subscribe(); 35 36 while !bytes.is_empty() { 37 pollable.block_until(&timeout).expect("write timed out"); 38 39 let permit = self.check_write()?; 40 41 let len = bytes.len().min(permit as usize); 42 let (chunk, rest) = bytes.split_at(len); 43 44 self.write(chunk)?; 45 46 self.blocking_flush()?; 47 48 bytes = rest; 49 } 50 Ok(()) 51 } 52 } 53 54 impl Network { 55 pub fn default() -> Network { 56 instance_network::instance_network() 57 } 58 59 pub fn blocking_resolve_addresses(&self, name: &str) -> Result<Vec<IpAddress>, ErrorCode> { 60 let stream = ip_name_lookup::resolve_addresses(&self, name)?; 61 62 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 63 let pollable = stream.subscribe(); 64 65 let mut addresses = vec![]; 66 67 loop { 68 match stream.resolve_next_address() { 69 Ok(Some(addr)) => { 70 addresses.push(addr); 71 } 72 Ok(None) => match addresses[..] { 73 [] => return Err(ErrorCode::NameUnresolvable), 74 _ => return Ok(addresses), 75 }, 76 Err(ErrorCode::WouldBlock) => { 77 pollable.block_until(&timeout)?; 78 } 79 Err(err) => return Err(err), 80 } 81 } 82 } 83 84 /// Same as `Network::blocking_resolve_addresses` but ignores post validation errors 85 /// 86 /// The ignored error codes signal that the input passed validation 87 /// and a lookup was actually attempted, but failed. These are ignored to 88 /// make the CI tests less flaky. 89 pub fn permissive_blocking_resolve_addresses( 90 &self, 91 name: &str, 92 ) -> Result<Vec<IpAddress>, ErrorCode> { 93 match self.blocking_resolve_addresses(name) { 94 Err(ErrorCode::NameUnresolvable | ErrorCode::TemporaryResolverFailure) => Ok(vec![]), 95 r => r, 96 } 97 } 98 } 99 100 impl TcpSocket { 101 pub fn new(address_family: IpAddressFamily) -> Result<TcpSocket, ErrorCode> { 102 tcp_create_socket::create_tcp_socket(address_family) 103 } 104 105 pub fn blocking_bind( 106 &self, 107 network: &Network, 108 local_address: IpSocketAddress, 109 ) -> Result<(), ErrorCode> { 110 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 111 let sub = self.subscribe(); 112 113 self.start_bind(&network, local_address)?; 114 115 loop { 116 match self.finish_bind() { 117 Err(ErrorCode::WouldBlock) => sub.block_until(&timeout)?, 118 result => return result, 119 } 120 } 121 } 122 123 pub fn blocking_listen(&self) -> Result<(), ErrorCode> { 124 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 125 let sub = self.subscribe(); 126 127 self.start_listen()?; 128 129 loop { 130 match self.finish_listen() { 131 Err(ErrorCode::WouldBlock) => sub.block_until(&timeout)?, 132 result => return result, 133 } 134 } 135 } 136 137 pub fn blocking_connect( 138 &self, 139 network: &Network, 140 remote_address: IpSocketAddress, 141 ) -> Result<(InputStream, OutputStream), ErrorCode> { 142 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 143 let sub = self.subscribe(); 144 145 self.start_connect(&network, remote_address)?; 146 147 loop { 148 match self.finish_connect() { 149 Err(ErrorCode::WouldBlock) => sub.block_until(&timeout)?, 150 result => return result, 151 } 152 } 153 } 154 155 pub fn blocking_accept(&self) -> Result<(TcpSocket, InputStream, OutputStream), ErrorCode> { 156 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 157 let sub = self.subscribe(); 158 159 loop { 160 match self.accept() { 161 Err(ErrorCode::WouldBlock) => sub.block_until(&timeout)?, 162 result => return result, 163 } 164 } 165 } 166 } 167 168 impl UdpSocket { 169 pub fn new(address_family: IpAddressFamily) -> Result<UdpSocket, ErrorCode> { 170 udp_create_socket::create_udp_socket(address_family) 171 } 172 173 pub fn blocking_bind( 174 &self, 175 network: &Network, 176 local_address: IpSocketAddress, 177 ) -> Result<(), ErrorCode> { 178 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 179 let sub = self.subscribe(); 180 181 self.start_bind(&network, local_address)?; 182 183 loop { 184 match self.finish_bind() { 185 Err(ErrorCode::WouldBlock) => sub.block_until(&timeout)?, 186 result => return result, 187 } 188 } 189 } 190 191 pub fn blocking_bind_unspecified(&self, network: &Network) -> Result<(), ErrorCode> { 192 let ip = IpAddress::new_unspecified(self.address_family()); 193 let port = 0; 194 195 self.blocking_bind(network, IpSocketAddress::new(ip, port)) 196 } 197 } 198 199 impl OutgoingDatagramStream { 200 fn blocking_check_send(&self, timeout: &Pollable) -> Result<u64, ErrorCode> { 201 let sub = self.subscribe(); 202 203 loop { 204 match self.check_send() { 205 Ok(0) => sub.block_until(timeout)?, 206 result => return result, 207 } 208 } 209 } 210 211 pub fn blocking_send(&self, mut datagrams: &[OutgoingDatagram]) -> Result<(), ErrorCode> { 212 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 213 214 while !datagrams.is_empty() { 215 let permit = self.blocking_check_send(&timeout)?; 216 let chunk_len = datagrams.len().min(permit as usize); 217 match self.send(&datagrams[..chunk_len]) { 218 Ok(0) => {} 219 Ok(packets_sent) => { 220 let packets_sent = packets_sent as usize; 221 datagrams = &datagrams[packets_sent..]; 222 } 223 Err(err) => return Err(err), 224 } 225 } 226 227 Ok(()) 228 } 229 } 230 231 impl IncomingDatagramStream { 232 pub fn blocking_receive(&self, count: Range<u64>) -> Result<Vec<IncomingDatagram>, ErrorCode> { 233 let timeout = monotonic_clock::subscribe_duration(TIMEOUT_NS); 234 let pollable = self.subscribe(); 235 let mut datagrams = vec![]; 236 237 loop { 238 match self.receive(count.end - datagrams.len() as u64) { 239 Ok(mut chunk) => { 240 datagrams.append(&mut chunk); 241 242 if datagrams.len() >= count.start as usize { 243 return Ok(datagrams); 244 } else { 245 pollable.block_until(&timeout)?; 246 } 247 } 248 Err(err) => return Err(err), 249 } 250 } 251 } 252 } 253 254 impl IpAddress { 255 pub const IPV4_BROADCAST: IpAddress = IpAddress::Ipv4((255, 255, 255, 255)); 256 257 pub const IPV4_LOOPBACK: IpAddress = IpAddress::Ipv4((127, 0, 0, 1)); 258 pub const IPV6_LOOPBACK: IpAddress = IpAddress::Ipv6((0, 0, 0, 0, 0, 0, 0, 1)); 259 260 pub const IPV4_UNSPECIFIED: IpAddress = IpAddress::Ipv4((0, 0, 0, 0)); 261 pub const IPV6_UNSPECIFIED: IpAddress = IpAddress::Ipv6((0, 0, 0, 0, 0, 0, 0, 0)); 262 263 pub const IPV4_MAPPED_LOOPBACK: IpAddress = 264 IpAddress::Ipv6((0, 0, 0, 0, 0, 0xFFFF, 0x7F00, 0x0001)); 265 266 pub const fn new_loopback(family: IpAddressFamily) -> IpAddress { 267 match family { 268 IpAddressFamily::Ipv4 => Self::IPV4_LOOPBACK, 269 IpAddressFamily::Ipv6 => Self::IPV6_LOOPBACK, 270 } 271 } 272 273 pub const fn new_unspecified(family: IpAddressFamily) -> IpAddress { 274 match family { 275 IpAddressFamily::Ipv4 => Self::IPV4_UNSPECIFIED, 276 IpAddressFamily::Ipv6 => Self::IPV6_UNSPECIFIED, 277 } 278 } 279 280 pub const fn family(&self) -> IpAddressFamily { 281 match self { 282 IpAddress::Ipv4(_) => IpAddressFamily::Ipv4, 283 IpAddress::Ipv6(_) => IpAddressFamily::Ipv6, 284 } 285 } 286 } 287 288 impl PartialEq for IpAddress { 289 fn eq(&self, other: &Self) -> bool { 290 match (self, other) { 291 (Self::Ipv4(left), Self::Ipv4(right)) => left == right, 292 (Self::Ipv6(left), Self::Ipv6(right)) => left == right, 293 _ => false, 294 } 295 } 296 } 297 298 impl IpSocketAddress { 299 pub const fn new(ip: IpAddress, port: u16) -> IpSocketAddress { 300 match ip { 301 IpAddress::Ipv4(addr) => IpSocketAddress::Ipv4(Ipv4SocketAddress { 302 port: port, 303 address: addr, 304 }), 305 IpAddress::Ipv6(addr) => IpSocketAddress::Ipv6(Ipv6SocketAddress { 306 port: port, 307 address: addr, 308 flow_info: 0, 309 scope_id: 0, 310 }), 311 } 312 } 313 314 pub const fn ip(&self) -> IpAddress { 315 match self { 316 IpSocketAddress::Ipv4(addr) => IpAddress::Ipv4(addr.address), 317 IpSocketAddress::Ipv6(addr) => IpAddress::Ipv6(addr.address), 318 } 319 } 320 321 pub const fn port(&self) -> u16 { 322 match self { 323 IpSocketAddress::Ipv4(addr) => addr.port, 324 IpSocketAddress::Ipv6(addr) => addr.port, 325 } 326 } 327 328 pub const fn family(&self) -> IpAddressFamily { 329 match self { 330 IpSocketAddress::Ipv4(_) => IpAddressFamily::Ipv4, 331 IpSocketAddress::Ipv6(_) => IpAddressFamily::Ipv6, 332 } 333 } 334 } 335 336 impl PartialEq for Ipv4SocketAddress { 337 fn eq(&self, other: &Self) -> bool { 338 self.port == other.port && self.address == other.address 339 } 340 } 341 342 impl PartialEq for Ipv6SocketAddress { 343 fn eq(&self, other: &Self) -> bool { 344 self.port == other.port 345 && self.flow_info == other.flow_info 346 && self.address == other.address 347 && self.scope_id == other.scope_id 348 } 349 } 350 351 impl PartialEq for IpSocketAddress { 352 fn eq(&self, other: &Self) -> bool { 353 match (self, other) { 354 (Self::Ipv4(l0), Self::Ipv4(r0)) => l0 == r0, 355 (Self::Ipv6(l0), Self::Ipv6(r0)) => l0 == r0, 356 _ => false, 357 } 358 } 359 } 360