Skip to content

Commit e04fbbb

Browse files
eastspireeastspire
authored
Upgrade hyperlane (#9748)
* feat: update & fortune * feat: randomNumber * feat: max row 500 * feat: update * feat: Fortune * feat: fortunes * feat: update * feat: update * feat: update * feat: update * feat: fortunes * feat: fortunes * feat: fortunes * feat: fortunes * feat: fortunes * feat: fortunes * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: update * feat: cache * feat: cache * feat: config * feat: config * feat: config * feat: v4.36.1 * docs: readme * feat: remove dyn * docs: readme * feat: lock * feat: lock * feat: lock * feat: lock * feat: db pool * feat: db pool * feat: lock * feat: lock * feat: db * feat: db * feat: db * feat: db * feat: db * feat: db * feat: db * feat: db * feat: rand * feat: rand * feat: rand * feat: rand * feat: rand * feat: port * feat: port * feat: port * feat: db * feat: db * feat: db * feat: lock * feat: lock * feat: v4.41.0 * feat: v4.42.0 * Merge remote-tracking branch 'upstream/master' * Merge remote-tracking branch 'upstream/master' * feat: inline * feat: dockerfile * feat: v4.52.1 * feat: remove key * remove: log * remove: log * feat: async * remove: empty loop * feat: utf8 * change: pool_size * remove: utf8 * feat: log * feat: log * feat: v3.14.1 * feat: 4.56.3 * feat: 4.56.4 * feat: 4.56.5 * feat: rename * Merge branch 'master' of github.com:TechEmpower/FrameworkBenchmarks * feat: speed * feat: speed * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * lock: toml * feat: runtime * feat: runtime * feat: runtime
1 parent b16b0d8 commit e04fbbb

File tree

8 files changed

+103
-91
lines changed

8 files changed

+103
-91
lines changed

‎frameworks/Rust/hyperlane/Cargo.toml‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,11 @@ exclude = [
1818
]
1919

2020
[dependencies]
21-
hyperlane = "4.56.5"
22-
rand = "0.9.0"
23-
serde = "1.0.219"
24-
sqlx = { version = "0.8.3", features = ["runtime-tokio", "postgres"] }
21+
hyperlane = "=4.65.0"
22+
num_cpus = "=1.16.0"
23+
rand = "=0.9.0"
24+
serde = "=1.0.219"
25+
sqlx = { version = "=0.8.3", features = ["runtime-tokio", "postgres"] }
2526

2627
[profile.dev]
2728
incremental = false

‎frameworks/Rust/hyperlane/src/db.rs‎

Lines changed: 39 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,20 @@
11
use crate::*;
22

3-
pub async fn get_db_connection() -> DbPoolConnection {
4-
if let Some(db_pool) = DB.get() {
5-
return db_pool.clone();
6-
};
7-
let db_pool: DbPoolConnection = connection_db().await;
8-
DB.set(db_pool.clone())
9-
.expect("Failed to initialize DB_POOL");
10-
db_pool
3+
pub fn get_db_connection() -> &'static DbPoolConnection {
4+
&DB
115
}
126

137
#[cfg(feature = "dev")]
148
pub async fn create_database() {
15-
let db_pool: DbPoolConnection = get_db_connection().await;
9+
let db_pool: &DbPoolConnection = get_db_connection();
1610
let _ = query(&format!("CREATE DATABASE {};", DATABASE_NAME))
1711
.execute(&db_pool)
1812
.await;
1913
}
2014

2115
#[cfg(feature = "dev")]
2216
pub async fn create_table() {
23-
let db_pool: DbPoolConnection = get_db_connection().await;
17+
let db_pool: &DbPoolConnection = get_db_connection();
2418
let _ = query(&format!(
2519
"CREATE TABLE IF NOT EXISTS {} (
2620
id SERIAL PRIMARY KEY, randomNumber INT NOT NULL
@@ -41,7 +35,7 @@ pub async fn create_table() {
4135

4236
#[cfg(feature = "dev")]
4337
pub async fn insert_records() {
44-
let db_pool: DbPoolConnection = get_db_connection().await;
38+
let db_pool: &DbPoolConnection = get_db_connection();
4539
let row: PgRow = query(&format!("SELECT COUNT(*) FROM {}", TABLE_NAME_WORLD))
4640
.fetch_one(&db_pool)
4741
.await
@@ -76,21 +70,21 @@ pub async fn insert_records() {
7670
let _ = query(&sql).execute(&db_pool).await;
7771
}
7872

79-
pub async fn init_cache() {
73+
pub async fn init_cache() -> Vec<QueryRow> {
8074
let mut res: Vec<QueryRow> = Vec::with_capacity(RANDOM_MAX as usize);
81-
let db_pool: DbPoolConnection = get_db_connection().await;
75+
let db_pool: &DbPoolConnection = get_db_connection();
8276
let sql: String = format!(
8377
"SELECT id, randomNumber FROM {} LIMIT {}",
8478
TABLE_NAME_WORLD, RANDOM_MAX
8579
);
86-
if let Ok(rows) = query(&sql).fetch_all(&db_pool).await {
80+
if let Ok(rows) = query(&sql).fetch_all(db_pool).await {
8781
for row in rows {
8882
let id: i32 = row.get(KEY_ID);
8983
let random_number: i32 = row.get(KEY_RANDOM_NUMBER);
9084
res.push(QueryRow::new(id, random_number));
9185
}
9286
}
93-
let _ = CACHE.set(res);
87+
res
9488
}
9589

9690
pub async fn connection_db() -> DbPoolConnection {
@@ -106,18 +100,10 @@ pub async fn connection_db() -> DbPoolConnection {
106100
DATABASE_NAME
107101
),
108102
};
109-
let pool_size: u32 = (get_thread_count() << 2).max(10).min(100) as u32;
110-
let max_pool_size: u32 = option_env!("POSTGRES_MAX_POOL_SIZE")
111-
.unwrap_or(&pool_size.to_string())
112-
.parse::<u32>()
113-
.unwrap_or(pool_size);
114-
let min_pool_size: u32 = option_env!("POSTGRES_MIN_POOL_SIZE")
115-
.unwrap_or(&pool_size.to_string())
116-
.parse::<u32>()
117-
.unwrap_or(pool_size);
103+
let pool_size: u32 = num_cpus::get() as u32;
118104
let pool: DbPoolConnection = PgPoolOptions::new()
119-
.max_connections(max_pool_size)
120-
.min_connections(min_pool_size)
105+
.max_connections(100)
106+
.min_connections(pool_size)
121107
.max_lifetime(None)
122108
.test_before_acquire(false)
123109
.idle_timeout(None)
@@ -130,10 +116,10 @@ pub async fn connection_db() -> DbPoolConnection {
130116
pub async fn get_update_data(
131117
limit: Queries,
132118
) -> (String, Vec<QueryRow>, Vec<Queries>, Vec<Queries>) {
133-
let db_pool: DbPoolConnection = get_db_connection().await;
119+
let db_pool: &DbPoolConnection = get_db_connection();
134120
let mut query_res_list: Vec<QueryRow> = Vec::with_capacity(limit as usize);
135-
let rows: Vec<QueryRow> = get_some_row_id(limit, &db_pool).await;
136-
let mut sql = format!("UPDATE {} SET randomNumber = CASE id ", TABLE_NAME_WORLD);
121+
let rows: Vec<QueryRow> = get_some_row_id(limit, db_pool).await;
122+
let mut sql: String = format!("UPDATE {} SET randomNumber = CASE id ", TABLE_NAME_WORLD);
137123
let mut id_list: Vec<i32> = Vec::with_capacity(rows.len());
138124
let mut value_list: Vec<String> = Vec::with_capacity(rows.len() * 2);
139125
let mut random_numbers: Vec<i32> = Vec::with_capacity(rows.len());
@@ -159,14 +145,13 @@ pub async fn get_update_data(
159145
}
160146

161147
pub async fn init_db() {
162-
get_db_connection().await;
163148
#[cfg(feature = "dev")]
164149
{
165150
create_database().await;
166151
create_table().await;
167152
insert_records().await;
168153
}
169-
init_cache().await;
154+
black_box(init_cache().await);
170155
}
171156

172157
pub async fn random_world_row(db_pool: &DbPoolConnection) -> QueryRow {
@@ -176,7 +161,7 @@ pub async fn random_world_row(db_pool: &DbPoolConnection) -> QueryRow {
176161

177162
pub async fn query_world_row(db_pool: &DbPoolConnection, id: Queries) -> QueryRow {
178163
let sql: String = format!(
179-
"SELECT id, randomNumber FROM {} WHERE id = {} LIMIT 1",
164+
"SELECT id, randomNumber FROM {} WHERE id = {}",
180165
TABLE_NAME_WORLD, id
181166
);
182167
if let Ok(rows) = query(&sql).fetch_one(db_pool).await {
@@ -187,7 +172,7 @@ pub async fn query_world_row(db_pool: &DbPoolConnection, id: Queries) -> QueryRo
187172
}
188173

189174
pub async fn update_world_rows(limit: Queries) -> Vec<QueryRow> {
190-
let db_pool: DbPoolConnection = get_db_connection().await;
175+
let db_pool: &DbPoolConnection = get_db_connection();
191176
let (sql, data, id_list, random_numbers) = get_update_data(limit).await;
192177
let mut query_builder: query::Query<'_, Postgres, postgres::PgArguments> = query(&sql);
193178
for (id, random_number) in id_list.iter().zip(random_numbers.iter()) {
@@ -196,23 +181,34 @@ pub async fn update_world_rows(limit: Queries) -> Vec<QueryRow> {
196181
for id in &id_list {
197182
query_builder = query_builder.bind(id);
198183
}
199-
let _ = query_builder.execute(&db_pool).await;
184+
let _ = query_builder.execute(db_pool).await;
200185
data
201186
}
202187

203188
pub async fn all_world_row() -> Vec<PgRow> {
204-
let db_pool: DbPoolConnection = get_db_connection().await;
189+
let db_pool: &DbPoolConnection = get_db_connection();
205190
let sql: String = format!("SELECT id, message FROM {}", TABLE_NAME_FORTUNE);
206-
let res: Vec<PgRow> = query(&sql).fetch_all(&db_pool).await.unwrap_or_default();
191+
let res: Vec<PgRow> = query(&sql).fetch_all(db_pool).await.unwrap_or_default();
207192
return res;
208193
}
209194

210195
pub async fn get_some_row_id(limit: Queries, db_pool: &DbPoolConnection) -> Vec<QueryRow> {
211-
let futures: Vec<_> = (0..limit)
212-
.map(|_| async {
213-
let id: i32 = get_random_id();
214-
query_world_row(db_pool, id).await
215-
})
216-
.collect();
217-
join_all(futures).await
196+
let semaphore: Arc<Semaphore> = Arc::new(Semaphore::new(32));
197+
let mut tasks: Vec<JoinHandle<QueryRow>> = Vec::with_capacity(limit as usize);
198+
for _ in 0..limit {
199+
let _ = semaphore.clone().acquire_owned().await.map(|permit| {
200+
let db_pool: DbPoolConnection = db_pool.clone();
201+
tasks.push(spawn(async move {
202+
let id: i32 = get_random_id();
203+
let res: QueryRow = query_world_row(&db_pool, id).await;
204+
drop(permit);
205+
res
206+
}));
207+
});
208+
}
209+
join_all(tasks)
210+
.await
211+
.into_iter()
212+
.filter_map(Result::ok)
213+
.collect()
218214
}

‎frameworks/Rust/hyperlane/src/lazy.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
11
use crate::*;
22

3-
pub static DB: OnceCell<DbPoolConnection> = OnceCell::new();
4-
pub static CACHE: OnceCell<Vec<QueryRow>> = OnceCell::new();
3+
pub static DB: Lazy<DbPoolConnection> = Lazy::new(|| block_on(async { connection_db().await }));
4+
pub static CACHE: Lazy<Vec<QueryRow>> = Lazy::new(|| block_on(async { init_cache().await }));

‎frameworks/Rust/hyperlane/src/main.rs‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,12 @@ pub(crate) mod utils;
1010

1111
pub(crate) use constant::*;
1212
pub(crate) use db::*;
13-
pub(crate) use futures::future::join_all;
1413
pub(crate) use hyperlane::{
15-
once_cell::sync::OnceCell,
14+
futures::{executor::block_on, future::join_all},
15+
once_cell::sync::Lazy,
1616
serde::*,
1717
serde_json::{Value, json},
18+
tokio::{spawn, sync::Semaphore, task::JoinHandle},
1819
*,
1920
};
2021
pub(crate) use lazy::*;
@@ -27,12 +28,10 @@ pub(crate) use sqlx::{
2728
postgres::{PgPoolOptions, PgRow},
2829
*,
2930
};
30-
pub(crate) use std::fmt;
31+
pub(crate) use std::{fmt, hint::black_box, sync::Arc};
3132
pub(crate) use r#type::*;
3233
pub(crate) use utils::*;
3334

34-
#[tokio::main]
35-
async fn main() {
36-
init_db().await;
37-
run_server().await;
35+
fn main() {
36+
run_server();
3837
}

‎frameworks/Rust/hyperlane/src/request_middleware.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
use crate::*;
22

3-
pub async fn request(controller_data: ControllerData) {
4-
let _ = controller_data
3+
pub async fn request(ctx: Context) {
4+
let _ = ctx
55
.set_response_header(CONNECTION, CONNECTION_KEEP_ALIVE)
66
.await
77
.set_response_header(CONTENT_TYPE, APPLICATION_JSON)

‎frameworks/Rust/hyperlane/src/response_middleware.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
use crate::*;
22

3-
pub async fn response(controller_data: ControllerData) {
4-
let _ = controller_data.send().await;
3+
pub async fn response(ctx: Context) {
4+
let _ = ctx.send().await;
55
}

‎frameworks/Rust/hyperlane/src/route.rs‎

Lines changed: 22 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,46 +1,46 @@
11
use crate::*;
22

3-
pub async fn json(controller_data: ControllerData) {
3+
pub async fn json(ctx: Context) {
44
let json: Value = json!({
55
"message": RESPONSEDATA_STR
66
});
7-
let _ = controller_data
7+
let _ = ctx
88
.set_response_body(serde_json::to_string(&json).unwrap_or_default())
99
.await;
1010
}
1111

12-
pub async fn plaintext(controller_data: ControllerData) {
13-
let _ = controller_data
12+
pub async fn plaintext(ctx: Context) {
13+
let _ = ctx
1414
.set_response_header(CONTENT_TYPE, TEXT_PLAIN)
1515
.await
1616
.set_response_body(RESPONSEDATA_BIN)
1717
.await;
1818
}
1919

20-
pub async fn db(controller_data: ControllerData) {
21-
let db_connection: DbPoolConnection = get_db_connection().await;
22-
let query_row: QueryRow = random_world_row(&db_connection).await;
23-
let _ = controller_data
20+
pub async fn db(ctx: Context) {
21+
let db_connection: &DbPoolConnection = get_db_connection();
22+
let query_row: QueryRow = random_world_row(db_connection).await;
23+
let _ = ctx
2424
.set_response_body(serde_json::to_string(&query_row).unwrap_or_default())
2525
.await;
2626
}
2727

28-
pub async fn queries(controller_data: ControllerData) {
29-
let queries: Queries = controller_data
28+
pub async fn queries(ctx: Context) {
29+
let queries: Queries = ctx
3030
.get_request_query("q")
3131
.await
3232
.and_then(|queries| queries.parse::<Queries>().ok())
3333
.unwrap_or_default()
3434
.min(ROW_LIMIT as Queries)
3535
.max(1);
36-
let db_pool: DbPoolConnection = get_db_connection().await;
37-
let data: Vec<QueryRow> = get_some_row_id(queries, &db_pool).await;
38-
let _ = controller_data
36+
let db_pool: &DbPoolConnection = get_db_connection();
37+
let data: Vec<QueryRow> = get_some_row_id(queries, db_pool).await;
38+
let _ = ctx
3939
.set_response_body(serde_json::to_string(&data).unwrap_or_default())
4040
.await;
4141
}
4242

43-
pub async fn fortunes(controller_data: ControllerData) {
43+
pub async fn fortunes(ctx: Context) {
4444
let all_rows: Vec<PgRow> = all_world_row().await;
4545
let mut fortunes_list: Vec<Fortunes> = all_rows
4646
.iter()
@@ -56,43 +56,36 @@ pub async fn fortunes(controller_data: ControllerData) {
5656
));
5757
fortunes_list.sort_by(|it, next| it.message.cmp(&next.message));
5858
let res: String = FortunesTemplate::new(fortunes_list).to_string();
59-
controller_data
60-
.set_response_header(CONTENT_TYPE, content_type_charset(TEXT_HTML, UTF8))
59+
ctx.set_response_header(CONTENT_TYPE, content_type_charset(TEXT_HTML, UTF8))
6160
.await
6261
.set_response_body(res)
6362
.await;
6463
}
6564

66-
pub async fn updates(controller_data: ControllerData) {
67-
let queries: Queries = controller_data
65+
pub async fn updates(ctx: Context) {
66+
let queries: Queries = ctx
6867
.get_request_query("q")
6968
.await
7069
.and_then(|queries| queries.parse::<Queries>().ok())
7170
.unwrap_or_default()
7271
.min(ROW_LIMIT as Queries)
7372
.max(1);
7473
let res: Vec<QueryRow> = update_world_rows(queries).await;
75-
let _ = controller_data
74+
let _ = ctx
7675
.set_response_body(serde_json::to_string(&res).unwrap_or_default())
7776
.await;
7877
}
7978

80-
pub async fn cached_queries(controller_data: ControllerData) {
81-
let count: Queries = controller_data
79+
pub async fn cached_queries(ctx: Context) {
80+
let count: Queries = ctx
8281
.get_request_query("c")
8382
.await
8483
.and_then(|queries| queries.parse::<Queries>().ok())
8584
.unwrap_or_default()
8685
.min(ROW_LIMIT as Queries)
8786
.max(1);
88-
let res: Vec<QueryRow> = CACHE
89-
.get()
90-
.unwrap_or(&Vec::new())
91-
.iter()
92-
.take(count as usize)
93-
.cloned()
94-
.collect();
95-
let _ = controller_data
87+
let res: Vec<QueryRow> = CACHE.iter().take(count as usize).cloned().collect();
88+
let _ = ctx
9689
.set_response_body(serde_json::to_string(&res).unwrap_or_default())
9790
.await;
9891
}

0 commit comments

Comments
 (0)