1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
// Copyright 2024 Cloudflare, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Pingora tokio runtime.
//!
//! Tokio runtime comes in two flavors: a single-threaded runtime
//! and a multi-threaded one which provides work stealing.
//! Benchmark shows that, compared to the single-threaded runtime, the multi-threaded one
//! has some overhead due to its more sophisticated work steal scheduling.
//!
//! This crate provides a third flavor: a multi-threaded runtime without work stealing.
//! This flavor is as efficient as the single-threaded runtime while allows the async
//! program to use multiple cores.
use pingora_runtime::{current_handle, Runtime};
use std::error::Error;
use std::{thread, time};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
async fn hello_server(port: usize) -> Result<(), Box<dyn Error + Send>> {
let addr = format!("127.0.0.1:{port}");
let listener = TcpListener::bind(&addr).await.unwrap();
println!("Listening on: {}", addr);
loop {
let (mut socket, _) = listener.accept().await.unwrap();
socket.set_nodelay(true).unwrap();
let rt = current_handle();
rt.spawn(async move {
loop {
let mut buf = [0; 1024];
let res = socket.read(&mut buf).await;
let n = match res {
Ok(n) => n,
Err(_) => return,
};
if n == 0 {
return;
}
let _ = socket
.write_all(
b"HTTP/1.1 200 OK\r\ncontent-length: 12\r\nconnection: keep-alive\r\n\r\nHello world!",
)
.await;
}
});
}
}
/* On M1 macbook pro
wrk -t40 -c1000 -d10 http://127.0.0.1:3001 --latency
Running 10s test @ http://127.0.0.1:3001
40 threads and 1000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 3.53ms 0.87ms 17.12ms 84.99%
Req/Sec 7.09k 1.29k 33.11k 93.30%
Latency Distribution
50% 3.56ms
75% 3.95ms
90% 4.37ms
99% 5.38ms
2844034 requests in 10.10s, 203.42MB read
Requests/sec: 281689.27
Transfer/sec: 20.15MB
wrk -t40 -c1000 -d10 http://127.0.0.1:3000 --latency
Running 10s test @ http://127.0.0.1:3000
40 threads and 1000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 12.16ms 16.29ms 112.29ms 83.40%
Req/Sec 5.47k 2.01k 48.85k 83.67%
Latency Distribution
50% 2.09ms
75% 20.23ms
90% 37.11ms
99% 65.16ms
2190869 requests in 10.10s, 156.70MB read
Requests/sec: 216918.71
Transfer/sec: 15.52MB
*/
fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let rt = Runtime::new_steal(2, "");
let handle = rt.get_handle();
handle.spawn(hello_server(3000));
let rt2 = Runtime::new_no_steal(2, "");
let handle = rt2.get_handle();
handle.spawn(hello_server(3001));
thread::sleep(time::Duration::from_secs(999999999));
Ok(())
}
|
