Using smol executor

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"
+futures = "0.3.17"
+smol = "1.2.5"
+num_cpus = "1.13.0"
+easy-parallel = "3.1.0"

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());
 

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(())