Non-blocking networking sockets

While the simple HTTP connections from the previous chapter are easy to implement, they are not ready for production use. If the program can only have one connection open at a time (e.g., blocking), the fast CPU would be waiting for the slow network. Non-blocking I/O means that the application program can keep multiple connections open at the same time, and process data in and out of those connections as they come in. The program can either alternatingly poll those open connections or wait for incoming data to trigger async functions. That allows I/O intensive programs to run much faster even in a single-threaded environment. In this chapter, we will cover both polling and async programming models.

A non-blocking HTTP client example

The source code for a non-blocking HTTP client application is available. The following main() function starts two HTTP connections. The program keeps both connections open, and alternatingly checks for incoming data from them. In another word, the two connections are not blocking each other. Their data are handled concurrently (or alternatingly) as the data comes in.

  1. use httparse::{Response, EMPTY_HEADER};
  2. use std::io::{self, Read, Write};
  3. use std::str::from_utf8;
  4. use wasmedge_wasi_socket::TcpStream;
  6. fn main() {
  7.     let req = "GET / HTTP/1.0\n\n";
  8.     let mut first_connection = TcpStream::connect("127.0.0.1:80").unwrap();
  9.     first_connection.set_nonblocking(true).unwrap();
  10.     first_connection.write_all(req.as_bytes()).unwrap();
  12.     let mut second_connection = TcpStream::connect("127.0.0.1:80").unwrap();
  13.     second_connection.set_nonblocking(true).unwrap();
  14.     second_connection.write_all(req.as_bytes()).unwrap();
  16.     let mut first_buf = vec![0; 4096];
  17.     let mut first_bytes_read = 0;
  18.     let mut second_buf = vec![0; 4096];
  19.     let mut second_bytes_read = 0;
  21.     loop {
  22.         let mut first_complete = false;
  23.         let mut second_complete = false;
  24.         if !first_complete {
  25.             match read_data(&mut first_connection, &mut first_buf, first_bytes_read) {
  26.                 Ok((bytes_read, false)) => {
  27.                     first_bytes_read = bytes_read;
  28.                 }
  29.                 Ok((bytes_read, true)) => {
  30.                     println!("First connection completed");
  31.                     if bytes_read != 0 {
  32.                         parse_data(&first_buf, bytes_read);
  33.                     }
  34.                     first_complete = true;
  35.                 }
  36.                 Err(e) => {
  37.                     println!("First connection error: {}", e);
  38.                     first_complete = true;
  39.                 }
  40.             }
  41.         }
  42.         if !second_complete {
  43.             match read_data(&mut second_connection, &mut second_buf, second_bytes_read) {
  44.                 Ok((bytes_read, false)) => {
  45.                     second_bytes_read = bytes_read;
  46.                 }
  47.                 Ok((bytes_read, true)) => {
  48.                     println!("Second connection completed");
  49.                     if bytes_read != 0 {
  50.                         parse_data(&second_buf, bytes_read);
  51.                     }
  52.                     second_complete = true;
  53.                 }
  54.                 Err(e) => {
  55.                     println!("Second connection error: {}", e);
  56.                     second_complete = true;
  57.                 }
  58.             }
  59.         }
  60.         if first_complete && second_complete {
  61.             break;
  62.         }
  63.     }
  64. }

The following command compiles the Rust program.

cargo build --target wasm32-wasi --release

The following command runs the application in WasmEdge.

  2. #![allow(unused)]
  3. fn main() {
  4. wasmedge target/wasm32-wasi/release/nonblock_http_client.wasm
  5. }

A non-blocking HTTP server example

The source code for a non-blocking HTTP server application is available. The following main() function starts an HTTP server. It receives events from multiple open connections, and processes those events as they are received by calling the async handler functions registered to each connection. This server can process events from multiple open connections concurrently.

  1. fn main() -> std::io::Result<()> {
  2.     let mut poll = Poll::new();
  3.     let server = TcpListener::bind("127.0.0.1:1234", true)?;
  4.     println!("Listening on 127.0.0.1:1234");
  5.     let mut connections = HashMap::new();
  6.     let mut handlers = HashMap::new();
  7.     const SERVER: Token = Token(0);
  8.     let mut unique_token = Token(SERVER.0 + 1);
  10.     poll.register(&server, SERVER, Interest::Read);
  12.     loop {
  13.         let events = poll.poll().unwrap();
  15.         for event in events {
  16.             match event.token {
  17.                 SERVER => loop {
  18.                     let (connection, address) = match server.accept(FDFLAGS_NONBLOCK) {
  19.                         Ok((connection, address)) => (connection, address),
  20.                         Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
  21.                         Err(e) => panic!("accept error: {}", e),
  22.                     };
  24.                     println!("Accepted connection from: {}", address);
  26.                     let token = unique_token.add();
  27.                     poll.register(&connection, token, Interest::Read);
  28.                     connections.insert(token, connection);
  29.                 },
  30.                 token => {
  31.                     let done = if let Some(connection) = connections.get_mut(&token) {
  32.                         let handler = match handlers.get_mut(&token) {
  33.                             Some(handler) => handler,
  34.                             None => {
  35.                                 let handler = Handler::new();
  36.                                 handlers.insert(token, handler);
  37.                                 handlers.get_mut(&token).unwrap()
  38.                             }
  39.                         };
  40.                         handle_connection(&mut poll, connection, handler, &event)?
  41.                     } else {
  42.                         false
  43.                     };
  44.                     if done {
  45.                         if let Some(connection) = connections.remove(&token) {
  46.                             connection.shutdown(Shutdown::Both)?;
  47.                             poll.unregister(&connection);
  48.                             handlers.remove(&token);
  49.                         }
  50.                     }
  51.                 }
  52.             }
  53.         }
  54.     }
  55. }

The handle_connection() function processes the data from those open connections. In this case, it just writes the request body into the response. It is also done asynchronously — meaning that the handle_connection() function creates an event for the response, and puts it in the queue. The main application loop processes the event and sends the response when it is waiting for data from other connections.

  2. #![allow(unused)]
  3. fn main() {
  4. fn handle_connection(
  5.     poll: &mut Poll,
  6.     connection: &mut TcpStream,
  7.     handler: &mut Handler,
  8.     event: &Event,
  9. ) -> io::Result<bool> {
  10.     if event.is_readable() {
  11.         let mut connection_closed = false;
  12.         let mut received_data = vec![0; 4096];
  13.         let mut bytes_read = 0;
  14.         loop {
  15.             match connection.read(&mut received_data[bytes_read..]) {
  16.                 Ok(0) => {
  17.                     connection_closed = true;
  18.                     break;
  19.                 }
  20.                 Ok(n) => {
  21.                     bytes_read += n;
  22.                     if bytes_read == received_data.len() {
  23.                         received_data.resize(received_data.len() + 1024, 0);
  24.                     }
  25.                 }
  26.                 Err(ref err) if would_block(err) => {
  27.                     if bytes_read != 0 {
  28.                         let received_data = &received_data[..bytes_read];
  29.                         let mut bs: parsed::stream::ByteStream =
  30.                             match String::from_utf8(received_data.to_vec()) {
  31.                                 Ok(s) => s,
  32.                                 Err(_) => {
  33.                                     continue;
  34.                                 }
  35.                             }
  36.                             .into();
  37.                         let req = match parsed::http::parse_http_request(&mut bs) {
  38.                             Some(req) => req,
  39.                             None => {
  40.                                 break;
  41.                             }
  42.                         };
  43.                         for header in req.headers.iter() {
  44.                             if header.name.eq("Conntent-Length") {
  45.                                 let content_length = header.value.parse::<usize>().unwrap();
  46.                                 if content_length > received_data.len() {
  47.                                     return Ok(true);
  48.                                 }
  49.                             }
  50.                         }
  51.                         println!(
  52.                             "{:?} request: {:?} {:?}",
  53.                             connection.peer_addr().unwrap(),
  54.                             req.method,
  55.                             req.path
  56.                         );
  57.                         let res = Response {
  58.                             protocol: "HTTP/1.1".to_string(),
  59.                             code: 200,
  60.                             message: "OK".to_string(),
  61.                             headers: vec![
  62.                                 Header {
  63.                                     name: "Content-Length".to_string(),
  64.                                     value: req.content.len().to_string(),
  65.                                 },
  66.                                 Header {
  67.                                     name: "Connection".to_string(),
  68.                                     value: "close".to_string(),
  69.                                 },
  70.                             ],
  71.                             content: req.content,
  72.                         };
  74.                         handler.response = Some(res.into());
  76.                         poll.reregister(connection, event.token, Interest::Write);
  77.                         break;
  78.                     } else {
  79.                         println!("Empty request");
  80.                         return Ok(true);
  81.                     }
  82.                 }
  83.                 Err(ref err) if interrupted(err) => continue,
  84.                 Err(err) => return Err(err),
  85.             }
  86.         }
  88.         if connection_closed {
  89.             println!("Connection closed");
  90.             return Ok(true);
  91.         }
  92.     }
  94.     if event.is_writable() && handler.response.is_some() {
  95.         let resp = handler.response.clone().unwrap();
  96.         match connection.write(resp.as_bytes()) {
  97.             Ok(n) if n < resp.len() => return Err(io::ErrorKind::WriteZero.into()),
  98.             Ok(_) => {
  99.                 return Ok(true);
  100.             }
  101.             Err(ref err) if would_block(err) => {}
  102.             Err(ref err) if interrupted(err) => {
  103.                 return handle_connection(poll, connection, handler, event)
  104.             }
  105.             Err(err) => return Err(err),
  106.         }
  107.     }
  109.     Ok(false)
  110. }
  111. }

The following command compiles the Rust program.

cargo build --target wasm32-wasi --release

The following command runs the application in WasmEdge.

$ wasmedge target/wasm32-wasi/release/poll_http_server.wasm new connection at 1234

To test the HTTP server, you can submit a HTTP request to it via curl.

$ curl -d "name=WasmEdge" -X POST http://127.0.0.1:1234 echo: name=WasmEdge