From 3b89c8299b1483b69a981123271cc9ee116020ff Mon Sep 17 00:00:00 2001
From: vandomej <jacob.vandome15@gmail.com>
Date: Sat, 23 Oct 2021 13:27:03 -0700
Subject: [PATCH] Using smol executor

---
 gemla/Cargo.toml      |  6 +++--
 gemla/src/bin/bin.rs  | 55 +++++++++++++++++++++++++++++--------------
 gemla/src/core/mod.rs | 19 ++++++---------
 3 files changed, 48 insertions(+), 32 deletions(-)

diff --git a/gemla/Cargo.toml b/gemla/Cargo.toml
index 4f2df1e..2ea4026 100644
--- a/gemla/Cargo.toml
+++ b/gemla/Cargo.toml
@@ -25,5 +25,7 @@ anyhow = "1.0"
 rand = "0.8.4"
 log = "0.4.14"
 env_logger = "0.9.0"
-tokio = { version = "1.12.0", features = ["full"] }
-futures = "0.3.17"
\ No newline at end of file
+futures = "0.3.17"
+smol = "1.2.5"
+num_cpus = "1.13.0"
+easy-parallel = "3.1.0"
\ No newline at end of file
diff --git a/gemla/src/bin/bin.rs b/gemla/src/bin/bin.rs
index 585ba13..774cc92 100644
--- a/gemla/src/bin/bin.rs
+++ b/gemla/src/bin/bin.rs
@@ -18,32 +18,51 @@ use test_state::TestState;
 ///
 /// Use the -h, --h, or --help flag to see usage syntax.
 /// TODO
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
+fn main() -> anyhow::Result<()> {
     env_logger::init();
     info!("Starting");
 
     let now = Instant::now();
 
-    // Command line arguments are parsed with the clap crate. And this program uses
-    // the yaml method with clap.
-    let yaml = load_yaml!("../../cli.yml");
-    let matches = App::from_yaml(yaml).get_matches();
+    // Obtainning number of threads to use
+    let num_threads = num_cpus::get().max(1);
+    let ex = smol::Executor::new();
+    let (signal, shutdown) = smol::channel::unbounded::<()>();
 
-    // Checking that the first argument <DIRECTORY> is a valid directory
-    let file_path = matches.value_of(gemla::constants::args::FILE).unwrap();
-    let mut gemla = log_error(Gemla::<TestState>::new(
-        &PathBuf::from(file_path),
-        GemlaConfig {
-            generations_per_node: 1,
-            overwrite: false,
-        },
-    ))?;
+    // Create an executor thread pool.
+    let (_, result): (
+        Vec<Result<(), smol::channel::RecvError>>,
+        Result<(), gemla::error::Error>,
+    ) = easy_parallel::Parallel::new()
+        .each(0..num_threads, |_| {
+            smol::future::block_on(ex.run(shutdown.recv()))
+        })
+        .finish(|| {
+            smol::block_on(async {
+                drop(signal);
 
-    log_error(gemla.simulate(100).await)?;
+                // Command line arguments are parsed with the clap crate. And this program uses
+                // the yaml method with clap.
+                let yaml = load_yaml!("../../cli.yml");
+                let matches = App::from_yaml(yaml).get_matches();
 
-    // let mut f = std::fs::File::create("./test")?;
-    // write!(f, "{}", serde_json::to_string(&gemla.data.readonly().0)?)?;
+                // Checking that the first argument <DIRECTORY> is a valid directory
+                let file_path = matches.value_of(gemla::constants::args::FILE).unwrap();
+                let mut gemla = log_error(Gemla::<TestState>::new(
+                    &PathBuf::from(file_path),
+                    GemlaConfig {
+                        generations_per_node: 1,
+                        overwrite: false,
+                    },
+                ))?;
+
+                log_error(gemla.simulate(100).await)?;
+
+                Ok(())
+            })
+        });
+
+    result?;
 
     info!("Finished in {:?}", now.elapsed());
 
diff --git a/gemla/src/core/mod.rs b/gemla/src/core/mod.rs
index 7c16d80..d31ede2 100644
--- a/gemla/src/core/mod.rs
+++ b/gemla/src/core/mod.rs
@@ -91,12 +91,8 @@ where
             if let Some(node) = node_to_process {
                 trace!("Adding node to process list {}", node.get_id());
 
-                // if self.threads.len() > 5 {
-                //     self.join_threads().await?;
-                // } else {
-                    self.threads
-                        .insert(node.get_id(), Box::pin(Gemla::process_node(node)));
-                // }
+                self.threads
+                    .insert(node.get_id(), Box::pin(Gemla::process_node(node)));
             } else {
                 trace!("No node found to process, joining threads");
 
@@ -110,13 +106,13 @@ where
     async fn join_threads(&mut self) -> Result<(), Error> {
         if self.threads.len() > 0 {
             trace!("Joining threads for nodes {:?}", self.threads.keys());
-    
+
             let results = futures::future::join_all(self.threads.values_mut()).await;
             let reduced_results: Result<Vec<GeneticNodeWrapper<T>>, Error> =
                 results.into_iter().collect();
-    
+
             self.threads.clear();
-    
+
             reduced_results.and_then(|r| {
                 if !self
                     .data
@@ -124,13 +120,12 @@ where
                 {
                     warn!("Unable to find nodes to replace in tree")
                 }
-    
+
                 self.data
                     .mutate(|(d, _)| Gemla::merge_completed_nodes(d.as_mut().unwrap()))??;
-    
+
                 Ok(())
             })?;
-    
         }
 
         Ok(())