非阻塞的网络通信

尽管上一章节中的 HTTP 连接实现起来很容易，但他们并不适合在生产环境使用。如果程序一次只允许一个连接开启（阻塞），CPU 将不得不等待缓慢的网络。非阻塞 I/O 意味着程序可以同时保持多个连接开启，并处理这些连接传输的数据。程序可以轮询这些开启的连接，或是等待输入的数据触发异步函数。这将让 I/O 密集的程序在单线程环境中运行得非常快。在这一章中，我们将介绍轮询和异步编程模型。

非阻塞 HTTP 客户端示例

非阻塞的 HTTP 客户端程序的源代码在这里。下面的 main() 函数开启了两个 HTTP 连接。它同时保持两个连接开启，并轮流查看是否有数据传输进来。换句话说，这两个连接并不会相互阻塞。他们的数据在传输进来的时候被同时（或者轮流）处理。

  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. }

使用如下命令来编译 Rust 程序。

cargo build --target wasm32-wasi --release

使用如下命令在 WasmEdge 中运行程序。

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

非阻塞 HTTP 服务器示例

非阻塞的 HTTP 服务器程序的 源代码在这里。下面的 main() 函数开启了一个 HTTP 服务器。它同时从多个开启的连接中接收事件，并通过调用注册在每个连接的异步处理函数来处理这些事件。服务器可以同时从多个开启的连接中处理事件。

  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. }

handle_connection() 函数处理来自于开启的连接的数据。当前它只是将请求的内容写回响应中。这同样是以异步的形式完成的，意味着 handle_connection() 函数为响应创建了一个事件，然后将事件放入了一个队列中。主程序循环处理这些事件，并在等待来自其他连接的数据时发送这些响应。

  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. }

使用如下命令来编译 Rust 程序。

cargo build --target wasm32-wasi --release

使用如下命令在 WasmEdge 中运行程序。

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

你可以用 curl 发送一个 HTTP 请求，来测试 HTTP 服务器。

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