Merge pull request #129 from robinbernon/threadpool_to_take_num_cores…

…_input

Adding input param of threads to ThreadPool::new()

README.md

@@ -98,7 +98,7 @@ struct LogLine {
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
- let pool = ThreadPool::new(None)?;
+ let pool = ThreadPool::new(None, None)?;
 
  let rows = Parquet::new(ParquetDirectory::new(S3Directory::new_with(
  AwsRegion::UsEast1,
@@ -133,7 +133,7 @@ use std::error::Error;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
- let pool = ThreadPool::new(None)?;
+ let pool = ThreadPool::new(None, None)?;
 
  let rows = Parquet::new(ParquetDirectory::new(S3Directory::new_with(
  AwsRegion::UsEast1,
@@ -169,7 +169,7 @@ use std::error::Error;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error>> {
- let pool = ThreadPool::new(None)?;
+ let pool = ThreadPool::new(None, None)?;
 
  let rows = Parquet::new(ParquetDirectory::new(S3Directory::new_with(
  AwsRegion::UsEast1,
@@ -215,7 +215,7 @@ fn main() -> Result<(), Box<dyn Error>> {
  .build()
  .unwrap()
  .block_on(async {
- let pool = ProcessPool::new(None, None, Resources::default())?;
+ let pool = ProcessPool::new(None, None, None, Resources::default())?;
 
  let rows = Parquet::new(ParquetDirectory::new(S3Directory::new_with(
  AwsRegion::UsEast1,

amadeus-postgres/src/lib.rs

@@ -213,9 +213,9 @@ where
 }
 
 #[cfg(not(nightly))]
-type Output<Row> = Pin<Box<dyn Stream<Item = Result<Row, PostgresError>> + Send>>;
+type Output<Row: PostgresData> = Pin<Box<dyn Stream<Item = Result<Row, PostgresError>> + Send>>;
 #[cfg(nightly)]
-type Output<Row> = impl Stream<Item = Result<Row, PostgresError>> + Send;
+type Output<Row: PostgresData> = impl Stream<Item = Result<Row, PostgresError>> + Send;
 
 FnMutNamed! {
  pub type Closure<Row> = |self|(config, tables)=> (ConnectParams, Vec<PostgresSelect>)| -> Output<Row>

amadeus-serde/src/csv.rs

@@ -79,9 +79,9 @@ where
 
 type Error<P, E> = CsvError<E, <P as Partition>::Error, <<P as Partition>::Page as Page>::Error>;
 #[cfg(not(nightly))]
-type Output<P, Row, E> = std::pin::Pin<Box<dyn Stream<Item = Result<Row, Error<P, E>>>>>;
+type Output<P, Row: SerdeData, E> = std::pin::Pin<Box<dyn Stream<Item = Result<Row, Error<P, E>>>>>;
 #[cfg(nightly)]
-type Output<P: Partition, Row, E> = impl Stream<Item = Result<Row, Error<P, E>>>;
+type Output<P: Partition, Row: SerdeData, E> = impl Stream<Item = Result<Row, Error<P, E>>>;
 
 FnMutNamed! {
  pub type Closure<P, Row, E> = |self|partition=> P| -> Output<P, Row, E>

amadeus-serde/src/json.rs

@@ -38,9 +38,9 @@ where
 
 type Error<P, E> = JsonError<E, <P as Partition>::Error, <<P as Partition>::Page as Page>::Error>;
 #[cfg(not(nightly))]
-type Output<P, Row, E> = std::pin::Pin<Box<dyn Stream<Item = Result<Row, Error<P, E>>>>>;
+type Output<P, Row: SerdeData, E> = std::pin::Pin<Box<dyn Stream<Item = Result<Row, Error<P, E>>>>>;
 #[cfg(nightly)]
-type Output<P: Partition, Row, E> = impl Stream<Item = Result<Row, Error<P, E>>>;
+type Output<P: Partition, Row: SerdeData, E> = impl Stream<Item = Result<Row, Error<P, E>>>;
 
 FnMutNamed! {
  pub type Closure<P, Row, E> = |self|partition=> P| -> Output<P, Row, E>

amadeus-types/src/time.rs

@@ -199,7 +199,7 @@ impl Timezone {
  _ => None,
  },
  }
- .map(|tz| tz.name())
+ .map(Tz::name)
  }
  /// Makes a new Timezone for the Eastern Hemisphere with given timezone difference. The negative seconds means the Western Hemisphere.
  pub fn from_offset(seconds: i32) -> Option<Self> {

benches/csv.rs

@@ -19,7 +19,7 @@ static RT: Lazy<Runtime> = Lazy::new(|| {
  .build()
  .unwrap()
 });
-static POOL: Lazy<ThreadPool> = Lazy::new(|| ThreadPool::new(None).unwrap());
+static POOL: Lazy<ThreadPool> = Lazy::new(|| ThreadPool::new(None, None).unwrap());
 
 #[derive(Data, Clone, Deserialize, PartialEq, PartialOrd, Debug)]
 struct GameDerived {

benches/in_memory.rs

@@ -19,7 +19,7 @@ static RT: Lazy<Runtime> = Lazy::new(|| {
  .build()
  .unwrap()
 });
-static POOL: Lazy<ThreadPool> = Lazy::new(|| ThreadPool::new(None).unwrap());
+static POOL: Lazy<ThreadPool> = Lazy::new(|| ThreadPool::new(None, None).unwrap());
 
 #[bench]
 fn vec(b: &mut Bencher) {

benches/parquet.rs

@@ -19,7 +19,7 @@ static RT: Lazy<Runtime> = Lazy::new(|| {
  .build()
  .unwrap()
 });
-static POOL: Lazy<ThreadPool> = Lazy::new(|| ThreadPool::new(None).unwrap());
+static POOL: Lazy<ThreadPool> = Lazy::new(|| ThreadPool::new(None, None).unwrap());
 
 #[derive(Data, Clone, PartialEq, Debug)]
 struct TenKayVeeTwo {

examples/cloudfront_logs.rs

@@ -16,7 +16,7 @@ use amadeus::prelude::*;
 #[allow(unreachable_code)]
 #[tokio::main]
 async fn main() {
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  let rows = Cloudfront::new_with(
  AwsRegion::UsEast1,

examples/columnar.rs

@@ -2,7 +2,7 @@ use amadeus::prelude::*;
 
 #[tokio::main]
 async fn main() {
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  let rows = Cloudfront::new_with(
  AwsRegion::UsEast1,

examples/commoncrawl.rs

@@ -19,7 +19,7 @@ use amadeus::{data::Webpage, dist::prelude::*};
 async fn main() {
  return; // TODO: runs for a long time
 
- let pool = ThreadPool::new(None).unwrap();
+ let pool = ThreadPool::new(None, None).unwrap();
 
  let webpages = CommonCrawl::new("CC-MAIN-2020-24").await.unwrap();
 

examples/commoncrawl_dist.rs

@@ -44,7 +44,7 @@ fn main() {
  // Accept the number of processes at the command line, defaulting to the maximum available
  let processes = env::args().nth(1).and_then(|arg| arg.parse::<usize>().ok());
 
- let pool = ProcessPool::new(processes, None, Resources::default()).unwrap();
+ let pool = ProcessPool::new(processes, None, None, Resources::default()).unwrap();
 
  let webpages = CommonCrawl::new("CC-MAIN-2020-24").await.unwrap();
 

src/pool/process.rs

@@ -113,7 +113,8 @@ struct ProcessPoolInner {
 impl ProcessPoolInner {
  #[allow(clippy::double_parens)] // TODO: work out what's triggering this
  fn new(
- processes: Option<usize>, tasks_per_core: Option<usize>, resources: Resources,
+ processes: Option<usize>, threads: Option<usize>, tasks: Option<usize>,
+ resources: Resources,
  ) -> Result<Self, SpawnError> {
  let processes = processes.unwrap_or(3); // TODO!
  let mut processes_vec = Vec::with_capacity(processes);
@@ -130,7 +131,7 @@ impl ProcessPoolInner {
  let receiver = Receiver::<Option<Request>>::new(parent);
  let sender = Sender::<Result<Response, Panicked>>::new(parent);
 
- let thread_pool = ThreadPool::new(tasks_per_core).unwrap();
+ let thread_pool = ThreadPool::new(threads, tasks).unwrap();
 
  while let Some(work) = receiver.recv().await.unwrap() {
  let ret = panic::catch_unwind(panic::AssertUnwindSafe(|| {
@@ -334,12 +335,11 @@ pub struct ProcessPool(Arc<ProcessPoolInner>);
 #[cfg_attr(not(nightly), serde_closure::desugar)]
 impl ProcessPool {
  pub fn new(
- processes: Option<usize>, tasks_per_core: Option<usize>, resources: Resources,
+ processes: Option<usize>, threads: Option<usize>, tasks: Option<usize>,
+ resources: Resources,
  ) -> Result<Self, SpawnError> {
  Ok(Self(Arc::new(ProcessPoolInner::new(
- processes,
- tasks_per_core,
- resources,
+ processes, threads, tasks, resources,
  )?)))
  }
  pub fn processes(&self) -> usize {

src/pool/thread.rs

@@ -17,33 +17,36 @@ const DEFAULT_TASKS_PER_CORE: usize = 100;
 
 #[derive(Debug)]
 struct ThreadPoolInner {
- logical_cores: usize,
- tasks_per_core: usize,
+ threads: usize,
+ tasks: usize,
  #[cfg(not(target_arch = "wasm32"))]
  pool: Pool,
 }
 
 #[derive(Debug)]
 pub struct ThreadPool(Arc<ThreadPoolInner>);
 impl ThreadPool {
- pub fn new(tasks_per_core: Option<usize>) -> io::Result<Self> {
- let logical_cores = if !cfg!(target_arch = "wasm32") {
+ pub fn new(threads: Option<usize>, tasks: Option<usize>) -> io::Result<Self> {
+ let threads = if let Some(threads) = threads {
+ threads
+ } else if !cfg!(target_arch = "wasm32") {
  num_cpus::get()
  } else {
  1
  };
- let tasks_per_core = tasks_per_core.unwrap_or(DEFAULT_TASKS_PER_CORE);
+
+ let tasks = tasks.unwrap_or(DEFAULT_TASKS_PER_CORE);
  #[cfg(not(target_arch = "wasm32"))]
- let pool = Pool::new(logical_cores);
+ let pool = Pool::new(threads);
  Ok(ThreadPool(Arc::new(ThreadPoolInner {
- logical_cores,
- tasks_per_core,
+ threads,
+ tasks,
  #[cfg(not(target_arch = "wasm32"))]
  pool,
  })))
  }
  pub fn threads(&self) -> usize {
- self.0.logical_cores * self.0.tasks_per_core
+ self.0.threads * self.0.tasks
  }
  pub fn spawn<F, Fut, T>(&self, task: F) -> impl Future<Output = Result<T, Panicked>> + Send
  where

tests/cloudfront.rs

@@ -6,7 +6,7 @@ use std::time::SystemTime;
 #[tokio::test(threaded_scheduler)]
 #[cfg_attr(miri, ignore)]
 async fn cloudfront() {
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  let start = SystemTime::now();
 

tests/cloudfront_dist.rs

@@ -17,12 +17,13 @@ fn main() {
  .unwrap()
  .block_on(async {
  let thread_pool_time = {
- let thread_pool = ThreadPool::new(None).unwrap();
+ let thread_pool = ThreadPool::new(None, None).unwrap();
  run(&thread_pool).await
  };
  #[cfg(feature = "constellation")]
  let process_pool_time = {
- let process_pool = ProcessPool::new(None, None, Resources::default()).unwrap();
+ let process_pool =
+ ProcessPool::new(None, None, None, Resources::default()).unwrap();
  run(&process_pool).await
  };
  #[cfg(not(feature = "constellation"))]

tests/commoncrawl.rs

@@ -20,7 +20,7 @@ use amadeus::{data::Webpage, prelude::*};
 async fn commoncrawl() {
  let start = SystemTime::now();
 
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  let webpages = CommonCrawl::new("CC-MAIN-2020-24").await.unwrap();
  let _ = webpages

tests/commoncrawl_dist.rs

@@ -31,12 +31,13 @@ fn main() {
  .unwrap()
  .block_on(async {
  let thread_pool_time = {
- let thread_pool = ThreadPool::new(None).unwrap();
+ let thread_pool = ThreadPool::new(None, None).unwrap();
  run(&thread_pool).await
  };
  #[cfg(feature = "constellation")]
  let process_pool_time = {
- let process_pool = ProcessPool::new(None, None, Resources::default()).unwrap();
+ let process_pool =
+ ProcessPool::new(None, None, None, Resources::default()).unwrap();
  run(&process_pool).await
  };
  #[cfg(not(feature = "constellation"))]

tests/csv.rs

@@ -9,7 +9,7 @@ use amadeus::prelude::*;
 async fn csv() {
  let start = SystemTime::now();
 
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  #[derive(Data, Clone, PartialEq, PartialOrd, Debug)]
  struct GameDerived {

tests/csv_dist.rs

@@ -20,12 +20,13 @@ fn main() {
  .unwrap()
  .block_on(async {
  let thread_pool_time = {
- let thread_pool = ThreadPool::new(None).unwrap();
+ let thread_pool = ThreadPool::new(None, None).unwrap();
  run(&thread_pool).await
  };
  #[cfg(feature = "constellation")]
  let process_pool_time = {
- let process_pool = ProcessPool::new(None, None, Resources::default()).unwrap();
+ let process_pool =
+ ProcessPool::new(None, None, None, Resources::default()).unwrap();
  run(&process_pool).await
  };
  #[cfg(not(feature = "constellation"))]

tests/csv_wasm.rs

@@ -44,7 +44,7 @@ async fn csv() {
  let timer = web_sys::window().unwrap().performance().unwrap();
  let start = timer.now();
 
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  #[derive(Data, Clone, PartialEq, PartialOrd, Debug)]
  struct GameDerived {

tests/into_par_stream.rs

@@ -5,7 +5,7 @@ use amadeus::prelude::*;
 #[tokio::test(threaded_scheduler)]
 #[cfg_attr(miri, ignore)]
 async fn into_par_stream() {
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  <&[usize] as IntoParallelStream>::into_par_stream(&[1, 2, 3])
  .map(|a: usize| a)

tests/into_par_stream_dist.rs

@@ -19,12 +19,13 @@ fn main() {
  .unwrap()
  .block_on(async {
  let thread_pool_time = {
- let thread_pool = ThreadPool::new(None).unwrap();
+ let thread_pool = ThreadPool::new(None, None).unwrap();
  run(&thread_pool).await
  };
  #[cfg(feature = "constellation")]
  let process_pool_time = {
- let process_pool = ProcessPool::new(None, None, Resources::default()).unwrap();
+ let process_pool =
+ ProcessPool::new(None, None, None, Resources::default()).unwrap();
  run(&process_pool).await
  };
  #[cfg(not(feature = "constellation"))]

tests/json.rs

@@ -9,7 +9,7 @@ use amadeus::prelude::*;
 async fn json() {
  let start = SystemTime::now();
 
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
  let tasks = 100;
 
  #[derive(Data, Clone, PartialEq, PartialOrd, Debug)]

tests/json_dist.rs

@@ -20,12 +20,13 @@ fn main() {
  .unwrap()
  .block_on(async {
  let thread_pool_time = {
- let thread_pool = ThreadPool::new(None).unwrap();
+ let thread_pool = ThreadPool::new(None, None).unwrap();
  run(&thread_pool, 100).await
  };
  #[cfg(feature = "constellation")]
  let process_pool_time = {
- let process_pool = ProcessPool::new(None, None, Resources::default()).unwrap();
+ let process_pool =
+ ProcessPool::new(None, None, None, Resources::default()).unwrap();
  run(&process_pool, 100).await
  };
  #[cfg(not(feature = "constellation"))]

tests/panic.rs

@@ -8,7 +8,7 @@ use amadeus::prelude::*;
 async fn panic() {
  let start = SystemTime::now();
 
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  let res = AssertUnwindSafe((0i32..1_000).into_par_stream().for_each(pool, |i| {
  if i == 500 {

tests/panic_dist.rs

@@ -21,12 +21,13 @@ fn main() {
  .unwrap()
  .block_on(async {
  let thread_pool_time = {
- let thread_pool = ThreadPool::new(None).unwrap();
+ let thread_pool = ThreadPool::new(None, None).unwrap();
  run(&thread_pool).await
  };
  #[cfg(feature = "constellation")]
  let process_pool_time = {
- let process_pool = ProcessPool::new(None, None, Resources::default()).unwrap();
+ let process_pool =
+ ProcessPool::new(None, None, None, Resources::default()).unwrap();
  run(&process_pool).await
  };
  #[cfg(not(feature = "constellation"))]

tests/parquet.rs

@@ -14,7 +14,7 @@ use amadeus::prelude::*;
 async fn parquet() {
  let start = SystemTime::now();
 
- let pool = &ThreadPool::new(None).unwrap();
+ let pool = &ThreadPool::new(None, None).unwrap();
 
  let rows = Parquet::<_, Value>::new(vec![
  PathBuf::from("amadeus-testing/parquet/cf-accesslogs/year=2018/month=11/day=02/part-00176-17868f39-cd99-4b60-bb48-8daf9072122e.c000.snappy.parquet"),