controllers/summary: Use tokio::try_join!() to perform database queries in parallel (#10082)

Author: Turbo87

src/controllers/summary.rs

@@ -9,85 +9,98 @@ use axum_extra::json;
 use axum_extra::response::ErasedJson;
 use diesel::prelude::*;
 use diesel_async::{AsyncPgConnection, RunQueryDsl};
+use futures_util::FutureExt;
+use std::future::Future;
 
 /// Handles the `GET /summary` route.
 pub async fn summary(state: AppState) -> AppResult<ErasedJson> {
     let mut conn = state.db_read().await?;
 
-    let popular_categories = Category::toplevel(&mut conn, "crates", 10, 0)
-        .await?
-        .into_iter()
-        .map(Category::into)
-        .collect::<Vec<EncodableCategory>>();
-
-    let num_crates: i64 = crates::table.count().get_result(&mut conn).await?;
-    let num_downloads: i64 = metadata::table
-        .select(metadata::total_downloads)
-        .get_result(&mut conn)
-        .await?;
-
     let config = &state.config;
 
-    let new_crates = crates::table
-        .inner_join(crate_downloads::table)
-        .left_join(recent_crate_downloads::table)
-        .left_join(default_versions::table)
-        .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
-        .order(crates::created_at.desc())
-        .select(Record::as_select())
-        .limit(10)
-        .load(&mut conn)
-        .await?;
-    let just_updated = crates::table
-        .inner_join(crate_downloads::table)
-        .left_join(recent_crate_downloads::table)
-        .left_join(default_versions::table)
-        .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
-        .filter(crates::updated_at.ne(crates::created_at))
-        .order(crates::updated_at.desc())
-        .select(Record::as_select())
-        .limit(10)
-        .load(&mut conn)
-        .await?;
+    let (
+        num_crates,
+        num_downloads,
+        new_crates,
+        just_updated,
+        most_downloaded,
+        most_recently_downloaded,
+        popular_categories,
+        popular_keywords,
+    ) = tokio::try_join!(
+        crates::table.count().get_result::<i64>(&mut conn).boxed(),
+        metadata::table
+            .select(metadata::total_downloads)
+            .get_result::<i64>(&mut conn)
+            .boxed(),
+        crates::table
+            .inner_join(crate_downloads::table)
+            .left_join(recent_crate_downloads::table)
+            .left_join(default_versions::table)
+            .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
+            .order(crates::created_at.desc())
+            .select(Record::as_select())
+            .limit(10)
+            .load(&mut conn)
+            .boxed(),
+        crates::table
+            .inner_join(crate_downloads::table)
+            .left_join(recent_crate_downloads::table)
+            .left_join(default_versions::table)
+            .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
+            .filter(crates::updated_at.ne(crates::created_at))
+            .order(crates::updated_at.desc())
+            .select(Record::as_select())
+            .limit(10)
+            .load(&mut conn)
+            .boxed(),
+        crates::table
+            .inner_join(crate_downloads::table)
+            .left_join(recent_crate_downloads::table)
+            .left_join(default_versions::table)
+            .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
+            .filter(crates::name.ne_all(&config.excluded_crate_names))
+            .then_order_by(crate_downloads::downloads.desc())
+            .select(Record::as_select())
+            .limit(10)
+            .load(&mut conn)
+            .boxed(),
+        crates::table
+            .inner_join(crate_downloads::table)
+            .inner_join(recent_crate_downloads::table)
+            .left_join(default_versions::table)
+            .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
+            .filter(crates::name.ne_all(&config.excluded_crate_names))
+            .then_order_by(recent_crate_downloads::downloads.desc())
+            .select(Record::as_select())
+            .limit(10)
+            .load(&mut conn)
+            .boxed(),
+        Category::toplevel(&mut conn, "crates", 10, 0),
+        keywords::table
+            .order(keywords::crates_cnt.desc())
+            .limit(10)
+            .load(&mut conn)
+            .boxed(),
+    )?;
 
-    let most_downloaded = crates::table
-        .inner_join(crate_downloads::table)
-        .left_join(recent_crate_downloads::table)
-        .left_join(default_versions::table)
-        .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
-        .filter(crates::name.ne_all(&config.excluded_crate_names))
-        .then_order_by(crate_downloads::downloads.desc())
-        .select(Record::as_select())
-        .limit(10)
-        .load(&mut conn)
-        .await?;
+    let (new_crates, most_downloaded, most_recently_downloaded, just_updated) = tokio::try_join!(
+        encode_crates(&mut conn, new_crates),
+        encode_crates(&mut conn, most_downloaded),
+        encode_crates(&mut conn, most_recently_downloaded),
+        encode_crates(&mut conn, just_updated),
+    )?;
 
-    let most_recently_downloaded = crates::table
-        .inner_join(crate_downloads::table)
-        .inner_join(recent_crate_downloads::table)
-        .left_join(default_versions::table)
-        .left_join(versions::table.on(default_versions::version_id.eq(versions::id)))
-        .filter(crates::name.ne_all(&config.excluded_crate_names))
-        .then_order_by(recent_crate_downloads::downloads.desc())
-        .select(Record::as_select())
-        .limit(10)
-        .load(&mut conn)
-        .await?;
+    let popular_categories = popular_categories
+        .into_iter()
+        .map(Category::into)
+        .collect::<Vec<EncodableCategory>>();
 
-    let popular_keywords = keywords::table
-        .order(keywords::crates_cnt.desc())
-        .limit(10)
-        .load(&mut conn)
-        .await?
+    let popular_keywords = popular_keywords
         .into_iter()
         .map(Keyword::into)
         .collect::<Vec<EncodableKeyword>>();
 
-    let new_crates = encode_crates(&mut conn, new_crates).await?;
-    let most_downloaded = encode_crates(&mut conn, most_downloaded).await?;
-    let most_recently_downloaded = encode_crates(&mut conn, most_recently_downloaded).await?;
-    let just_updated = encode_crates(&mut conn, just_updated).await?;
-
     Ok(json!({
         "num_downloads": num_downloads,
         "num_crates": num_crates,
@@ -115,32 +128,41 @@ struct Record {
     yanked: Option<bool>,
 }
 
-async fn encode_crates(
+fn encode_crates(
     conn: &mut AsyncPgConnection,
     data: Vec<Record>,
-) -> AppResult<Vec<EncodableCrate>> {
-    let krates = data.iter().map(|record| &record.krate).collect::<Vec<_>>();
-    let versions: Vec<Version> = Version::belonging_to(&krates)
+) -> impl Future<Output = AppResult<Vec<EncodableCrate>>> {
+    let crate_ids = data
+        .iter()
+        .map(|record| record.krate.id)
+        .collect::<Vec<_>>();
+
+    let future = versions::table
+        .filter(versions::crate_id.eq_any(crate_ids))
         .filter(versions::yanked.eq(false))
         .select(Version::as_select())
-        .load(conn)
-        .await?;
+        .load(conn);
 
-    versions
-        .grouped_by(&krates)
-        .into_iter()
-        .map(TopVersions::from_versions)
-        .zip(data)
-        .map(|(top_versions, record)| {
-            Ok(EncodableCrate::from_minimal(
-                record.krate,
-                record.default_version.as_deref(),
-                record.yanked,
-                Some(&top_versions),
-                false,
-                record.total_downloads,
-                record.recent_downloads,
-            ))
-        })
-        .collect()
+    async move {
+        let versions: Vec<Version> = future.await?;
+
+        let krates = data.iter().map(|record| &record.krate).collect::<Vec<_>>();
+        versions
+            .grouped_by(&krates)
+            .into_iter()
+            .map(TopVersions::from_versions)
+            .zip(data)
+            .map(|(top_versions, record)| {
+                Ok(EncodableCrate::from_minimal(
+                    record.krate,
+                    record.default_version.as_deref(),
+                    record.yanked,
+                    Some(&top_versions),
+                    false,
+                    record.total_downloads,
+                    record.recent_downloads,
+                ))
+            })
+            .collect()
+    }
 }

src/models/category.rs

@@ -5,6 +5,7 @@ use diesel::{
 };
 use diesel_async::scoped_futures::ScopedFutureExt;
 use diesel_async::{AsyncConnection, AsyncPgConnection, RunQueryDsl};
+use std::future::Future;
 
 use crate::models::Crate;
 use crate::schema::*;
@@ -97,12 +98,12 @@ impl Category {
             .await
     }
 
-    pub async fn toplevel(
+    pub fn toplevel(
         conn: &mut AsyncPgConnection,
         sort: &str,
         limit: i64,
         offset: i64,
-    ) -> QueryResult<Vec<Category>> {
+    ) -> impl Future<Output = QueryResult<Vec<Category>>> {
         use diesel::sql_types::Int8;
 
         let sort_sql = match sort {
@@ -116,7 +117,6 @@ impl Category {
             .bind::<Int8, _>(limit)
             .bind::<Int8, _>(offset)
             .load(conn)
-            .await
     }
 
     pub async fn subcategories(&self, conn: &mut AsyncPgConnection) -> QueryResult<Vec<Category>> {