非阻塞的网络通信

尽管上一章节中的 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