Add more control config

Signed-off-by: koushiro <koushiro.cqx@gmail.com>

Author: koushiro

bin/node-template-archive/archive.toml

@@ -44,13 +44,25 @@ std = "DEBUG"
 #name = "archive.log"
 
 [control]
+# Number of database actors to be spawned.
+# Optional, default: 4
+db_actor_pool_size = 4
+
 # Number of threads to dedicate for executing tasks
 # Optional, default: the number of logical system threads
 task_workers = 8
 
+# Timeout to wait for a task to start execution.
+# Optional, default: 20 seconds
+task_timeout = 20
+
+# Maximum number of tasks to queue in the threadpool.
+# Optional, default: 64
+max_tasks = 64
+
 # Maximium number of blocks to load and insert into database at a time.
 # Useful for controlling memory usage.
-# Optional, defaults: 100,000.
+# Optional, defaults: 100,000
 max_block_load = 100000
 
 [wasm_tracing]

bin/polkadot-archive/archive.toml

@@ -51,13 +51,25 @@ std = "DEBUG"
 #name = "archive.log"
 
 [control]
+# Number of database actors to be spawned.
+# Optional, default: 4
+db_actor_pool_size = 4
+
 # Number of threads to dedicate for executing tasks
 # Optional, default: the number of logical system threads
 task_workers = 8
 
+# Timeout to wait for a task to start execution.
+# Optional, default: 20 seconds
+task_timeout = 20
+
+# Maximum number of tasks to queue in the threadpool.
+# Optional, default: 64
+max_tasks = 64
+
 # Maximium number of blocks to load and insert into database at a time.
 # Useful for controlling memory usage.
-# Optional, defaults: 100,000.
+# Optional, defaults: 100,000
 max_block_load = 100000
 
 [wasm_tracing]

substrate-archive/src/actors.rs

@@ -24,7 +24,7 @@ use std::{marker::PhantomData, panic::AssertUnwindSafe, sync::Arc, time::Duratio
 use coil::Job as _;
 use futures::{future::BoxFuture, FutureExt};
 use hashbrown::HashSet;
-use serde::de::DeserializeOwned;
+use serde::{de::DeserializeOwned, Deserialize};
 use xtra::{prelude::*, spawn::Smol, Disconnected};
 
 use sc_client_api::backend;
@@ -60,8 +60,7 @@ where
 	pub backend: Arc<ReadOnlyBackend<B, D>>,
 	pub pg_url: String,
 	pub meta: Meta<B>,
-	pub workers: usize,
-	pub max_block_load: u32,
+	pub control: ControlConfig,
 	pub tracing_targets: Option<String>,
 }
 
@@ -74,13 +73,64 @@ where
 			backend: Arc::clone(&self.backend),
 			pg_url: self.pg_url.clone(),
 			meta: self.meta.clone(),
-			workers: self.workers,
-			max_block_load: self.max_block_load,
+			control: self.control,
 			tracing_targets: self.tracing_targets.clone(),
 		}
 	}
 }
 
+#[derive(Copy, Clone, Debug, Deserialize)]
+pub struct ControlConfig {
+	/// number of database actors to be spawned in the actor pool.
+	#[serde(default = "default_db_actor_pool_size")]
+	pub(crate) db_actor_pool_size: usize,
+	/// number of threads to spawn for task execution.
+	#[serde(default = "default_task_workers")]
+	pub(crate) task_workers: usize,
+	/// maximum amount of time coil will wait for a task to begin.
+	/// times out if tasks don't start execution in the threadpool within `task_timeout` seconds.
+	#[serde(default = "default_task_timeout")]
+	pub(crate) task_timeout: u64,
+	/// maximum tasks to queue in the threadpool.
+	#[serde(default = "default_task_workers")]
+	pub(crate) max_tasks: usize,
+	/// maximum amount of blocks to index at once
+	#[serde(default = "default_max_block_load")]
+	pub(crate) max_block_load: u32,
+}
+
+impl Default for ControlConfig {
+	fn default() -> Self {
+		Self {
+			db_actor_pool_size: default_db_actor_pool_size(),
+			task_workers: default_task_workers(),
+			task_timeout: default_task_timeout(),
+			max_tasks: default_max_tasks(),
+			max_block_load: default_max_block_load(),
+		}
+	}
+}
+
+const fn default_db_actor_pool_size() -> usize {
+	4
+}
+
+fn default_task_workers() -> usize {
+	num_cpus::get()
+}
+
+const fn default_task_timeout() -> u64 {
+	20
+}
+
+const fn default_max_tasks() -> usize {
+	64
+}
+
+const fn default_max_block_load() -> u32 {
+	100_000
+}
+
 impl<B: BlockT + Unpin, D: ReadOnlyDB> SystemConfig<B, D>
 where
 	B::Hash: Unpin,
@@ -89,11 +139,10 @@ where
 		backend: Arc<ReadOnlyBackend<B, D>>,
 		pg_url: String,
 		meta: Meta<B>,
-		workers: usize,
-		max_block_load: u32,
+		control: ControlConfig,
 		tracing_targets: Option<String>,
 	) -> Self {
-		Self { backend, pg_url, meta, workers, max_block_load, tracing_targets }
+		Self { backend, pg_url, meta, control, tracing_targets }
 	}
 
 	pub fn backend(&self) -> &Arc<ReadOnlyBackend<B, D>> {
@@ -207,10 +256,10 @@ where
 
 		let runner = coil::Runner::builder(env, TaskExecutor, &pool)
 			.register_job::<crate::tasks::execute_block::Job<B, R, C, D>>()
-			.num_threads(conf.workers)
+			.num_threads(conf.control.task_workers)
 			// times out if tasks don't start execution on the threadpool within 20 seconds.
-			.timeout(Duration::from_secs(20))
-			.max_tasks(64)
+			.timeout(Duration::from_secs(conf.control.task_timeout))
+			.max_tasks(conf.control.max_tasks)
 			.build()?;
 
 		loop {
@@ -237,7 +286,8 @@ where
 
 	async fn spawn_actors(conf: SystemConfig<B, D>) -> Result<Actors<B, D>> {
 		let db = workers::DatabaseActor::<B>::new(conf.pg_url().into()).await?;
-		let db_pool = actor_pool::ActorPool::new(db, 4).create(None).spawn(&mut Smol::Global);
+		let db_pool =
+			actor_pool::ActorPool::new(db, conf.control.db_actor_pool_size).create(None).spawn(&mut Smol::Global);
 		let storage = workers::StorageAggregator::new(db_pool.clone()).create(None).spawn(&mut Smol::Global);
 		let metadata = workers::MetadataActor::new(db_pool.clone(), conf.meta().clone())
 			.await?
@@ -345,7 +395,7 @@ where
 		Ok(())
 	}
 
-	fn context(&self) -> Result<super::actors::SystemConfig<B, D>> {
-		Ok(self.config.clone())
+	fn context(&self) -> &SystemConfig<B, D> {
+		&self.config
 	}
 }

substrate-archive/src/actors/workers/blocks.rs

@@ -70,7 +70,7 @@ where
 			backend: conf.backend().clone(),
 			db,
 			meta,
-			max_block_load: conf.max_block_load,
+			max_block_load: conf.control.max_block_load,
 		}
 	}
 

substrate-archive/src/archive.rs

@@ -34,7 +34,7 @@ use substrate_archive_backend::{
 };
 
 use crate::{
-	actors::{System, SystemConfig},
+	actors::{ControlConfig, System, SystemConfig},
 	database::{self, DatabaseConfig},
 	error::Result,
 	logger::{self, FileLoggerConfig, LoggerConfig},
@@ -88,31 +88,6 @@ pub struct TracingConfig {
 	pub folder: Option<PathBuf>,
 }
 
-#[derive(Clone, Debug, Deserialize)]
-pub struct ControlConfig {
-	/// number of threads to spawn for task execution.
-	#[serde(default = "default_task_workers")]
-	pub(crate) task_workers: usize,
-	/// maximum amount of blocks to index at once
-	#[serde(default = "default_max_block_load")]
-	pub(crate) max_block_load: u32,
-}
-
-impl Default for ControlConfig {
-	fn default() -> Self {
-		Self { task_workers: default_task_workers(), max_block_load: default_max_block_load() }
-	}
-}
-
-fn default_task_workers() -> usize {
-	num_cpus::get()
-}
-
-// 100_000 blocks to index at once
-const fn default_max_block_load() -> u32 {
-	100_000
-}
-
 #[derive(Clone, Debug, Default, Deserialize)]
 pub struct ArchiveConfig {
 	/// chain spec and database
@@ -144,7 +119,7 @@ where
 	fn drive(&mut self) -> Result<()>;
 
 	/// this method will block indefinitely
-	async fn block_until_stopped(&self) -> ();
+	async fn block_until_stopped(&self);
 
 	/// shutdown the system
 	fn shutdown(self) -> Result<()>;
@@ -153,7 +128,7 @@ where
 	fn boxed_shutdown(self: Box<Self>) -> Result<()>;
 
 	/// Get a reference to the context the actors are using
-	fn context(&self) -> Result<SystemConfig<B, D>>;
+	fn context(&self) -> &SystemConfig<B, D>;
 }
 
 pub struct ArchiveBuilder<B, R, D, DB> {
@@ -241,7 +216,7 @@ impl<B, R, D, DB> ArchiveBuilder<B, R, D, DB> {
 		self
 	}
 
-	/// Number of 64KB Heap Pages to allocate for WASM execution
+	/// Set the number of 64KB Heap Pages to allocate for WASM execution
 	///
 	/// # Default
 	/// defaults to 64 * (number of logic cpu's)
@@ -250,6 +225,15 @@ impl<B, R, D, DB> ArchiveBuilder<B, R, D, DB> {
 		self
 	}
 
+	/// Set the number of database actors to be spawned in the actor pool.
+	///
+	/// # Default
+	/// defaults to 4
+	pub fn db_actor_pool_size(mut self, size: usize) -> Self {
+		self.config.control.db_actor_pool_size = size;
+		self
+	}
+
 	/// Set the  number of threads spawn for task execution.
 	///
 	/// # Default
@@ -259,6 +243,24 @@ impl<B, R, D, DB> ArchiveBuilder<B, R, D, DB> {
 		self
 	}
 
+	/// Set the timeout to wait for a task to start execution.
+	///
+	/// # Default
+	/// defaults to 20 seconds
+	pub fn task_timeout(mut self, timeout: u64) -> Self {
+		self.config.control.task_timeout = timeout;
+		self
+	}
+
+	/// Set the maximum tasks to queue in the threadpool.
+	///
+	/// # Default
+	/// defaults to 64
+	pub fn max_tasks(mut self, max: usize) -> Self {
+		self.config.control.max_tasks = max;
+		self
+	}
+
 	/// Set the number of blocks to index at once
 	///
 	/// # Default
@@ -392,8 +394,7 @@ where
 			backend,
 			pg_url,
 			client.clone(),
-			self.config.control.task_workers,
-			self.config.control.max_block_load,
+			self.config.control,
 			self.config.wasm_tracing.map(|t| t.targets),
 		);
 		let sys = System::<_, R, _, _>::new(client, config)?;

substrate-archive/src/lib.rs

@@ -32,8 +32,8 @@ mod tasks;
 mod types;
 mod wasm_tracing;
 
-pub use self::actors::System;
-pub use self::archive::{Archive, ArchiveBuilder, ArchiveConfig, ChainConfig, ControlConfig, TracingConfig};
+pub use self::actors::{ControlConfig, System};
+pub use self::archive::{Archive, ArchiveBuilder, ArchiveConfig, ChainConfig, TracingConfig};
 pub use self::database::{queries, DatabaseConfig};
 pub use self::error::ArchiveError;