Async db

4 files changed, 78 insertions, 58 deletions

diff --git a/front_end/src/cat_page.rs b/front_end/src/cat_page.rs
index cef868a..dd1d191 100644
--- a/front_end/src/cat_page.rs
+++ b/front_end/src/cat_page.rs
@@ -1,10 +1,10 @@
+use futures::stream::StreamExt;
 use crate::templates;
 use crate::Page;
 use categories::Category;
 use categories::CATEGORIES;
 use failure::Error;
 use kitchen_sink::KitchenSink;
-use rayon::prelude::*;
 use render_readme::Renderer;
 use rich_crate::RichCrateVersion;
 use std::collections::HashSet;
@@ -21,15 +21,18 @@ pub struct CatPage<'a> {
 
 impl<'a> CatPage<'a> {
     pub async fn new(cat: &'a Category, crates: &'a KitchenSink, markup: &'a Renderer) -> Result<CatPage<'a>, Error> {
+        let (count, keywords, related) = futures::join!(
+            crates.category_crate_count(&cat.slug),
+            crates.top_keywords_in_category(cat),
+            crates.related_categories(&cat.slug),
+        );
         Ok(Self {
-            count: crates.category_crate_count(&cat.slug)? as usize,
-            keywords: crates.top_keywords_in_category(cat)?,
-            related: crates.related_categories(&cat.slug)?,
-            crates: crates
-                .top_crates_in_category(&cat.slug).await?
-                .par_iter()
-                .with_max_len(1)
-                .filter_map(|o| {
+            count: count? as usize,
+            keywords: keywords?,
+            related: related?,
+            crates: futures::stream::iter(crates
+                .top_crates_in_category(&cat.slug).await?.iter())
+                .filter_map(|o| async move {
                     let c = match crates.rich_crate_version(&o) {
                         Ok(c) => c,
                         Err(e) => {
@@ -40,7 +43,7 @@ impl<'a> CatPage<'a> {
                     if c.is_yanked() {
                         return None;
                     }
-                    let d = match crates.downloads_per_month_or_equivalent(&o) {
+                    let d = match crates.downloads_per_month_or_equivalent(&o).await {
                         Ok(d) => d.unwrap_or(0) as u32,
                         Err(e) => {
                             eprintln!("Skipping {:?} because dl {}", o, e);
@@ -49,7 +52,7 @@ impl<'a> CatPage<'a> {
                     };
                     Some((c, d))
                 })
-                .collect::<Vec<_>>(),
+                .collect::<Vec<_>>().await,
             cat,
             markup,
         })
diff --git a/front_end/src/crate_page.rs b/front_end/src/crate_page.rs
index 2fdf934..370f6c2 100644
--- a/front_end/src/crate_page.rs
+++ b/front_end/src/crate_page.rs
@@ -1,3 +1,4 @@
+use futures::future::Future;
 use tokio::runtime::Handle;
 use futures::future::join_all;
 use crate::download_graph::DownloadsGraph;
@@ -105,14 +106,13 @@ impl<'a> CratePage<'a> {
             .and_then(|(top, slug)| CATEGORIES.from_slug(slug).last().map(|c| (top, c)));
         let is_build_or_dev = kitchen_sink.is_build_or_dev(ver.origin()).await?;
 
-        let (top_keyword, (all_contributors, deps)) = rayon::join(
-            || kitchen_sink.top_keyword(all),
-            || rayon::join(
-                || kitchen_sink.all_contributors(ver),
-                || kitchen_sink.all_dependencies_flattened(ver)));
+        let (top_keyword, all_contributors) = futures::try_join!(
+            kitchen_sink.top_keyword(all),
+            kitchen_sink.all_contributors(ver))?;
+        let deps = kitchen_sink.all_dependencies_flattened(ver);
         let mut page = Self {
-            top_keyword: top_keyword?,
-            all_contributors: all_contributors?,
+            top_keyword,
+            all_contributors,
             all,
             ver,
             kitchen_sink,
@@ -185,7 +185,7 @@ impl<'a> CratePage<'a> {
     }
 
     pub fn changelog_url(&self) -> Option<String> {
-        self.kitchen_sink.changelog_url(self.ver)
+        self.handle.enter(|| futures::executor::block_on(self.kitchen_sink.changelog_url(self.ver)))
     }
 
     pub fn is_build_or_dev(&self) -> (bool, bool) {
@@ -206,18 +206,22 @@ impl<'a> CratePage<'a> {
 
     /// If true, there are many other crates with this keyword. Populated first.
     pub fn keywords_populated(&self) -> Option<Vec<(String, bool)>> {
-        let k = self.kitchen_sink.keywords_populated(self.ver);
-        if k.is_empty() {
-            None
-        } else {
-            Some(k)
-        }
+        self.handle.enter(|| futures::executor::block_on(async {
+            let k = self.kitchen_sink.keywords_populated(self.ver).await;
+            if k.is_empty() {
+                None
+            } else {
+                Some(k)
+            }
+        }))
     }
 
     pub fn parent_crate(&self) -> Option<RichCrateVersion> {
-        let origin = self.kitchen_sink.parent_crate(self.ver)?;
-        self.kitchen_sink.rich_crate_version(&origin)
-            .map_err(|e| eprintln!("parent crate: {} {:?}", e, origin)).ok()
+        self.handle.enter(|| futures::executor::block_on(async {
+            let origin = self.kitchen_sink.parent_crate(self.ver).await?;
+            self.kitchen_sink.rich_crate_version(&origin)
+                .map_err(|e| eprintln!("parent crate: {} {:?}", e, origin)).ok()
+        }))
     }
 
     pub fn render_markdown_str(&self, s: &str) -> templates::Html<String> {
@@ -259,7 +263,11 @@ impl<'a> CratePage<'a> {
 
     pub fn nofollow(&self) -> bool {
         // TODO: take multiple factors into account, like # of contributors, author reputation, dependents
-        self.kitchen_sink.downloads_per_month_or_equivalent(self.all.origin()).ok().and_then(|x| x).unwrap_or(0) < 50
+        self.block(self.kitchen_sink.downloads_per_month_or_equivalent(self.all.origin())).ok().and_then(|x| x).unwrap_or(0) < 50
+    }
+
+    fn block<O>(&self, f: impl Future<Output=O>) -> O {
+        self.handle.enter(|| futures::executor::block_on(f))
     }
 
     pub(crate) fn all_contributors(&self) -> Contributors<'_> {
@@ -355,22 +363,21 @@ impl<'a> CratePage<'a> {
     }
 
     pub fn up_to_date_class(&self, richdep: &RichDep) -> &str {
-        if richdep.dep.req() == "*" {
+        if richdep.dep.req() == "*" || !richdep.dep.is_crates_io() {
             return "common";
         }
-        self.handle.enter(|| {
-            let (matches_latest, pop) = richdep.dep.req().parse().ok().and_then(|req| {
-                if !richdep.dep.is_crates_io() {
-                    return None;
+        self.block(async {
+            if let Ok(req) = richdep.dep.req().parse() {
+                if let Ok(Some((matches_latest, pop))) = self.kitchen_sink.version_popularity(&richdep.package, &req).await {
+                    return match pop {
+                        x if x >= 0.5 && matches_latest => "top",
+                        x if x >= 0.75 || matches_latest => "common",
+                        x if x >= 0.25 => "outdated",
+                        _ => "obsolete",
+                    }
                 }
-                futures::executor::block_on(self.kitchen_sink.version_popularity(&richdep.package, &req)).expect("deps")
-            }).unwrap_or((false, 0.));
-            match pop {
-                x if x >= 0.5 && matches_latest => "top",
-                x if x >= 0.75 || matches_latest => "common",
-                x if x >= 0.25 => "outdated",
-                _ => "obsolete",
             }
+            "obsolete"
         })
     }
 
@@ -675,16 +682,18 @@ impl<'a> CratePage<'a> {
     }
 
     pub fn github_stargazers_and_watchers(&self) -> Option<(u32, u32)> {
-        self.kitchen_sink.github_stargazers_and_watchers(self.all.origin()).ok().and_then(|x| x)
+        self.block(self.kitchen_sink.github_stargazers_and_watchers(self.all.origin())).ok().and_then(|x| x)
     }
 
     pub fn related_crates(&self) -> Option<Vec<Origin>> {
         // require some level of downloads to avoid recommending spam
         // but limit should be relative to the current crate, so that minor crates
         // get related suggestions too
-        let dl = self.kitchen_sink.downloads_per_month_or_equivalent(self.all.origin()).ok().and_then(|x| x).unwrap_or(100);
-        let min_recent_downloads = (dl as u32/2).min(200);
-        self.kitchen_sink.related_crates(&self.ver, min_recent_downloads).map_err(|e| eprintln!("related crates fail: {}", e)).ok()
+        self.block(async {
+            let dl = self.kitchen_sink.downloads_per_month_or_equivalent(self.all.origin()).await.ok().and_then(|x| x).unwrap_or(100);
+            let min_recent_downloads = (dl as u32/2).min(200);
+            self.kitchen_sink.related_crates(&self.ver, min_recent_downloads).await.map_err(|e| eprintln!("related crates fail: {}", e)).ok()
+        })
     }
 
     /// data for piechart
diff --git a/front_end/src/front_end.rs b/front_end/src/front_end.rs
index 7a01ed9..92ab7c3 100644
--- a/front_end/src/front_end.rs
+++ b/front_end/src/front_end.rs
@@ -90,21 +90,21 @@ pub async fn render_category(out: &mut impl Write, cat: &Category, crates: &Kitc
 /// See `homepage.rs.html`
 pub async fn render_homepage<W>(out: &mut W, crates: &KitchenSink) -> Result<(), failure::Error> where W: ?Sized, for<'a> &'a mut W: Write {
     let urler = Urler::new(None);
-    let home = home_page::HomePage::new(crates)?;
+    let home = home_page::HomePage::new(crates).await?;
     let all = home.all_categories().await;
     templates::homepage(out, &home, &all, &urler)?;
     Ok(())
 }
 
 /// See `atom.rs.html`
-pub fn render_feed(out: &mut impl Write, crates: &KitchenSink) -> Result<(), failure::Error> {
+pub async fn render_feed(out: &mut impl Write, crates: &KitchenSink) -> Result<(), failure::Error> {
     let urler = Urler::new(None);
-    templates::atom(out, &home_page::HomePage::new(crates)?, &urler)?;
+    templates::atom(out, &home_page::HomePage::new(crates).await?, &urler)?;
     Ok(())
 }
 
-pub fn render_sitemap(sitemap: &mut impl Write, crates: &KitchenSink) -> Result<(), failure::Error> {
-    let all_crates = crates.sitemap_crates()?;
+pub async fn render_sitemap(sitemap: &mut impl Write, crates: &KitchenSink) -> Result<(), failure::Error> {
+    let all_crates = crates.sitemap_crates().await?;
     let urler = Urler::new(None);
 
     sitemap.write_all(br#"<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">"#)?;
diff --git a/front_end/src/home_page.rs b/front_end/src/home_page.rs
index 589006c..1b8bf8e 100644
--- a/front_end/src/home_page.rs
+++ b/front_end/src/home_page.rs
@@ -32,11 +32,15 @@ pub struct HomeCategory {
 /// Computes data used on the home page on https://lib.rs/
 pub struct HomePage<'a> {
     crates: &'a KitchenSink,
+    handle: tokio::runtime::Handle,
 }
 
 impl<'a> HomePage<'a> {
-    pub fn new(crates: &'a KitchenSink) -> Result<Self, failure::Error> {
-        Ok(Self { crates })
+    pub async fn new(crates: &'a KitchenSink) -> Result<HomePage<'a>, failure::Error> {
+        Ok(Self {
+            crates,
+            handle: tokio::runtime::Handle::current(),
+        })
     }
 
     pub fn total_crates(&self) -> String {
@@ -86,7 +90,7 @@ impl<'a> HomePage<'a> {
                 if stopped() { return Vec::new(); }
                     // depth first - important!
                     let sub = self.make_all_categories(&cat.sub, seen).await;
-                    let own_pop = self.crates.category_crate_count(&cat.slug).unwrap_or(0) as usize;
+                    let own_pop = self.crates.category_crate_count(&cat.slug).await.unwrap_or(0) as usize;
 
                     c.push(HomeCategory {
                         // make container as popular as its best child (already sorted), because homepage sorts by top-level only
@@ -119,7 +123,7 @@ impl<'a> HomePage<'a> {
                 // skip topmost popular, because some categories have literally just 1 super-popular crate,
                 // which elevates the whole category
                 for c in top.iter().skip(1) {
-                    if let Ok(Some(d)) = self.crates.downloads_per_month_or_equivalent(c) {
+                    if let Ok(Some(d)) = self.crates.downloads_per_month_or_equivalent(c).await {
                         dl += d;
                     }
                 }
@@ -181,8 +185,8 @@ impl<'a> HomePage<'a> {
     }
 
     pub fn all_contributors<'c>(&self, krate: &'c RichCrateVersion) -> Option<Vec<CrateAuthor<'c>>> {
-        self.crates
-            .all_contributors(krate)
+        self.block(self.crates
+            .all_contributors(krate))
             .map(|(mut a, mut o, ..)| {
                 a.append(&mut o);
                 a
@@ -191,13 +195,17 @@ impl<'a> HomePage<'a> {
     }
 
     pub fn recently_updated_crates<'z>(&'z self) -> impl Iterator<Item = (RichCrate, RichCrateVersion)> + 'z {
-        self.crates
-            .recently_updated_crates()
+        self.block(self.crates
+            .recently_updated_crates())
             .expect("recent crates")
             .into_iter()
             .map(move |o| (self.crates.rich_crate(&o).unwrap(), self.crates.rich_crate_version(&o).unwrap()))
     }
 
+    fn block<O>(&self, f: impl Future<Output=O>) -> O {
+        self.handle.enter(|| futures::executor::block_on(f))
+    }
+
     pub fn page(&self) -> Page {
         Page {
             title: "Lib.rs — home for Rust crates".to_owned(),