run "load new crates" and "build crates from queue" in different threads

QuietMisdreavus · QuietMisdreavus · commit 11b587c21f57 · 2019-05-01T13:56:45.000-05:00
diff --git a/src/docbuilder/mod.rs b/src/docbuilder/mod.rs
@@ -110,6 +110,11 @@ impl DocBuilder {
         Ok(())
     }
 
+    /// Checks for the lock file and returns whether it currently exists.
+    pub fn is_locked(&self) -> bool {
+        self.lock_path().exists()
+    }
+
     /// Returns a reference of options
     pub fn options(&self) -> &DocBuilderOptions {
         &self.options
diff --git a/src/docbuilder/queue.rs b/src/docbuilder/queue.rs
@@ -10,10 +10,11 @@ use crates_index_diff::{ChangeKind, Index};
 impl DocBuilder {
     /// Updates crates.io-index repository and adds new crates into build queue.
     /// Returns size of queue
-    pub fn get_new_crates(&mut self) -> Result<i64> {
+    pub fn get_new_crates(&mut self) -> Result<usize> {
         let conn = try!(connect_db());
         let index = try!(Index::from_path_or_cloned(&self.options.crates_io_index_path));
         let mut changes = try!(index.fetch_changes());
+        let mut add_count: usize = 0;
 
         // I belive this will fix ordering of queue if we get more than one crate from changes
         changes.reverse();
@@ -23,17 +24,20 @@ impl DocBuilder {
                          &[&krate.name, &krate.version])
                 .ok();
             debug!("{}-{} added into build queue", krate.name, krate.version);
+            add_count += 1;
         }
 
-        let queue_count = conn.query("SELECT COUNT(*) FROM queue WHERE attempt < 5", &[])
+        Ok(add_count)
+    }
+
+    pub fn get_queue_count(&self) -> Result<i64> {
+        let conn = try!(connect_db());
+        Ok(conn.query("SELECT COUNT(*) FROM queue WHERE attempt < 5", &[])
             .unwrap()
             .get(0)
-            .get(0);
-
-        Ok(queue_count)
+            .get(0))
     }
 
-
     /// Builds packages from queue
     pub fn build_packages_queue(&mut self) -> Result<usize> {
         let conn = try!(connect_db());
@@ -67,6 +71,44 @@ impl DocBuilder {
 
         Ok(build_count)
     }
+
+    /// Builds the top package from the queue. Returns whether the queue was empty.
+    pub fn build_next_queue_package(&mut self) -> Result<bool> {
+        let conn = try!(connect_db());
+
+        let query = try!(conn.query("SELECT id, name, version
+                                     FROM queue
+                                     WHERE attempt < 5
+                                     ORDER BY attempt ASC, id ASC
+                                     LIMIT 1",
+                                    &[]));
+
+        if query.is_empty() {
+            // nothing in the queue; bail
+            return Ok(false);
+        }
+
+        let id: i32 = query.get(0).get(0);
+        let name: String = query.get(0).get(1);
+        let version: String = query.get(0).get(2);
+
+        match self.build_package(&name[..], &version[..]) {
+            Ok(_) => {
+                let _ = conn.execute("DELETE FROM queue WHERE id = $1", &[&id]);
+            }
+            Err(e) => {
+                // Increase attempt count
+                let _ = conn.execute("UPDATE queue SET attempt = attempt + 1 WHERE id = $1",
+                                     &[&id]);
+                error!("Failed to build package {}-{} from queue: {}",
+                       name,
+                       version,
+                       e)
+            }
+        }
+
+        Ok(true)
+    }
 }
 
 #[cfg(test)]
diff --git a/src/utils/daemon.rs b/src/utils/daemon.rs
@@ -4,6 +4,7 @@
 
 
 use std::{env, thread};
+use std::panic::{catch_unwind, AssertUnwindSafe};
 use std::process::exit;
 use std::fs::File;
 use std::io::Write;
@@ -44,80 +45,160 @@ pub fn start_daemon() {
         exit(0);
     }
 
-
     // check new crates every minute
     thread::spawn(move || {
+        // space this out to prevent it from clashing against the queue-builder thread on launch
+        thread::sleep(Duration::from_secs(30));
         loop {
+            let opts = opts();
+            let mut doc_builder = DocBuilder::new(opts);
+
+            debug!("Checking new crates");
+            match doc_builder.get_new_crates() {
+                Ok(n) => debug!("{} crates added to queue", n),
+                Err(e) => error!("Failed to get new crates: {}", e),
+            }
+
             thread::sleep(Duration::from_secs(60));
+        }
+    });
+
+    // build new crates every minute
+    thread::spawn(move || {
+        let mut opts = opts();
+        opts.skip_if_exists = true;
+        let mut doc_builder = DocBuilder::new(opts);
+
+        /// Represents the current state of the builder thread.
+        enum BuilderState {
+            /// The builder thread has just started, and hasn't built any crates yet.
+            Fresh,
+            /// The builder has just seen an empty build queue.
+            EmptyQueue,
+            /// The builder has just seen the lock file.
+            Locked,
+            /// The builder has just finished building a crate. The enclosed count is the number of
+            /// crates built since the caches have been refreshed.
+            QueueInProgress(usize),
+        }
 
-            let mut opts = opts();
-            opts.skip_if_exists = true;
+        let mut status = BuilderState::Fresh;
+
+        loop {
+            if !status.is_in_progress() {
+                thread::sleep(Duration::from_secs(60));
+            }
 
             // check lock file
-            if opts.prefix.join("cratesfyi.lock").exists() {
+            if doc_builder.is_locked() {
                 warn!("Lock file exits, skipping building new crates");
+                status = BuilderState::Locked;
                 continue;
             }
 
-            let mut doc_builder = DocBuilder::new(opts);
+            if status.count() > 10 {
+                // periodically, we need to flush our caches and ping the hubs
+                debug!("10 builds in a row; flushing caches");
+                status = BuilderState::QueueInProgress(0);
 
-            debug!("Checking new crates");
-            let queue_count = match doc_builder.get_new_crates() {
-                Ok(size) => size,
-                Err(e) => {
-                    error!("Failed to get new crates: {}", e);
-                    continue;
+                match pubsubhubbub::ping_hubs() {
+                    Err(e) => error!("Failed to ping hub: {}", e),
+                    Ok(n) => debug!("Succesfully pinged {} hubs", n)
                 }
-            };
 
-            // Only build crates if there is any
-            if queue_count == 0 {
-                debug!("Queue is empty, going back to sleep");
-                continue;
-            }
+                if let Err(e) = doc_builder.load_cache() {
+                    error!("Failed to load cache: {}", e);
+                }
 
-            info!("Building {} crates from queue", queue_count);
+                if let Err(e) = doc_builder.save_cache() {
+                    error!("Failed to save cache: {}", e);
+                }
 
-            // update index
-            if let Err(e) = update_sources() {
-                error!("Failed to update sources: {}", e);
-                continue;
+                if let Err(e) = update_sources() {
+                    error!("Failed to update sources: {}", e);
+                    continue;
+                }
             }
 
-            if let Err(e) = doc_builder.load_cache() {
-                error!("Failed to load cache: {}", e);
-                continue;
+            // Only build crates if there are any to build
+            debug!("Checking build queue");
+            match doc_builder.get_queue_count() {
+                Err(e) => {
+                    error!("Failed to read the number of crates in the queue: {}", e);
+                    continue;
+                }
+                Ok(0) => {
+                    if status.is_in_progress() {
+                        // ping the hubs before continuing
+                        match pubsubhubbub::ping_hubs() {
+                            Err(e) => error!("Failed to ping hub: {}", e),
+                            Ok(n) => debug!("Succesfully pinged {} hubs", n)
+                        }
+
+                        if let Err(e) = doc_builder.save_cache() {
+                            error!("Failed to save cache: {}", e);
+                        }
+                    }
+                    debug!("Queue is empty, going back to sleep");
+                    status = BuilderState::EmptyQueue;
+                    continue;
+                }
+                Ok(queue_count) => {
+                    info!("Starting build with {} crates in queue (currently on a {} crate streak)",
+                          queue_count, status.count());
+                }
             }
 
+            // if we're starting a new batch, reload our caches and sources
+            if !status.is_in_progress() {
+                if let Err(e) = doc_builder.load_cache() {
+                    error!("Failed to load cache: {}", e);
+                    continue;
+                }
+
+                if let Err(e) = update_sources() {
+                    error!("Failed to update sources: {}", e);
+                    continue;
+                }
+            }
 
-            // Run build_packages_queue in it's own thread to catch panics
+            // Run build_packages_queue under `catch_unwind` to catch panics
             // This only panicked twice in the last 6 months but its just a better
             // idea to do this.
-            let res = thread::spawn(move || {
-                    match doc_builder.build_packages_queue() {
-                        Err(e) => error!("Failed build new crates: {}", e),
-                        Ok(n) => {
-                            if n > 0 {
-                                match pubsubhubbub::ping_hubs() {
-                                    Err(e) => error!("Failed to ping hub: {}", e),
-                                    Ok(n) => debug!("Succesfully pinged {} hubs", n)
-                                }
-                            }
-                        }
-                    }
-
-                    if let Err(e) = doc_builder.save_cache() {
-                        error!("Failed to save cache: {}", e);
+            let res = catch_unwind(AssertUnwindSafe(|| {
+                match doc_builder.build_next_queue_package() {
+                    Err(e) => error!("Failed to build crate from queue: {}", e),
+                    Ok(crate_built) => if crate_built {
+                        status.increment();
                     }
 
-                    debug!("Finished building new crates, going back to sleep");
-                })
-                .join();
+                }
+            }));
 
             if let Err(e) = res {
                 error!("GRAVE ERROR Building new crates panicked: {:?}", e);
             }
         }
+
+        impl BuilderState {
+            fn count(&self) -> usize {
+                match *self {
+                    BuilderState::QueueInProgress(n) => n,
+                    _ => 0,
+                }
+            }
+
+            fn is_in_progress(&self) -> bool {
+                match *self {
+                    BuilderState::QueueInProgress(_) => true,
+                    _ => false,
+                }
+            }
+
+            fn increment(&mut self) {
+                *self = BuilderState::QueueInProgress(self.count() + 1);
+            }
+        }
     });
 
 
