Files
pezkuwi-subxt/substrate/test-utils/test-runner/src/client.rs
T
Bastian Köcher c09d52ead7 Use tokio runtime handle instead of TaskExecutor abstraction (#9737)
* Use tokio runtime handle instead of TaskExecutor abstraction

Before this pr we had the `TaskExecutor` abstraction which theoretically
allowed that any futures executor could have been used. However, this
was never tested and is currently not really required. Anyone running a
node currently only used tokio and nothing else (because this was hard
coded in CLI). So, this pr removes the `TaskExecutor` abstraction and
relies directly on the tokio runtime handle.

Besides this changes, this pr also makes sure that the http and ws rpc
server use the same tokio runtime. This fixes a panic that occurred when
you drop the rpc servers inside an async function (tokio doesn't like
that a tokio runtime is dropped in the async context of another tokio
runtime).

As we don't use any custom runtime in the http rpc server anymore, this
pr also removes the `rpc-http-threads` cli argument. If external parties
complain that there aren't enough threads for the rpc server, we could
bring support for increasing the thread count of the tokio runtime.

* FMT

* Fix try runtime

* Fix integration tests and some other optimizations

* Remove warnings
2021-09-12 12:29:11 +00:00

245 lines
7.6 KiB
Rust

// This file is part of Substrate.
// Copyright (C) 2021 Parity Technologies (UK) Ltd.
// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
//! Client parts
use crate::{default_config, ChainInfo};
use futures::channel::mpsc;
use jsonrpc_core::MetaIoHandler;
use manual_seal::{
consensus::babe::{BabeConsensusDataProvider, SlotTimestampProvider},
import_queue,
rpc::{ManualSeal, ManualSealApi},
run_manual_seal, EngineCommand, ManualSealParams,
};
use sc_client_api::backend::Backend;
use sc_executor::NativeElseWasmExecutor;
use sc_service::{
build_network, new_full_parts, spawn_tasks, BuildNetworkParams, ChainSpec, Configuration,
SpawnTasksParams, TFullBackend, TFullClient, TaskManager,
};
use sc_transaction_pool::BasicPool;
use sc_transaction_pool_api::TransactionPool;
use sp_api::{ApiExt, ConstructRuntimeApi, Core, Metadata};
use sp_block_builder::BlockBuilder;
use sp_consensus_babe::BabeApi;
use sp_finality_grandpa::GrandpaApi;
use sp_keyring::sr25519::Keyring::Alice;
use sp_offchain::OffchainWorkerApi;
use sp_runtime::traits::{Block as BlockT, Header};
use sp_session::SessionKeys;
use sp_transaction_pool::runtime_api::TaggedTransactionQueue;
use std::{str::FromStr, sync::Arc};
type ClientParts<T> = (
Arc<MetaIoHandler<sc_rpc::Metadata, sc_rpc_server::RpcMiddleware>>,
TaskManager,
Arc<
TFullClient<
<T as ChainInfo>::Block,
<T as ChainInfo>::RuntimeApi,
NativeElseWasmExecutor<<T as ChainInfo>::ExecutorDispatch>,
>,
>,
Arc<
dyn TransactionPool<
Block = <T as ChainInfo>::Block,
Hash = <<T as ChainInfo>::Block as BlockT>::Hash,
Error = sc_transaction_pool::error::Error,
InPoolTransaction = sc_transaction_pool::Transaction<
<<T as ChainInfo>::Block as BlockT>::Hash,
<<T as ChainInfo>::Block as BlockT>::Extrinsic,
>,
>,
>,
mpsc::Sender<EngineCommand<<<T as ChainInfo>::Block as BlockT>::Hash>>,
Arc<TFullBackend<<T as ChainInfo>::Block>>,
);
/// Provide the config or chain spec for a given chain
pub enum ConfigOrChainSpec {
/// Configuration object
Config(Configuration),
/// Chain spec object
ChainSpec(Box<dyn ChainSpec>, tokio::runtime::Handle),
}
/// Creates all the client parts you need for [`Node`](crate::node::Node)
pub fn client_parts<T>(
config_or_chain_spec: ConfigOrChainSpec,
) -> Result<ClientParts<T>, sc_service::Error>
where
T: ChainInfo + 'static,
<T::RuntimeApi as ConstructRuntimeApi<
T::Block,
TFullClient<T::Block, T::RuntimeApi, NativeElseWasmExecutor<T::ExecutorDispatch>>,
>>::RuntimeApi: Core<T::Block>
+ Metadata<T::Block>
+ OffchainWorkerApi<T::Block>
+ SessionKeys<T::Block>
+ TaggedTransactionQueue<T::Block>
+ BlockBuilder<T::Block>
+ BabeApi<T::Block>
+ ApiExt<T::Block, StateBackend = <TFullBackend<T::Block> as Backend<T::Block>>::State>
+ GrandpaApi<T::Block>,
<T::Runtime as frame_system::Config>::Call: From<frame_system::Call<T::Runtime>>,
<<T as ChainInfo>::Block as BlockT>::Hash: FromStr + Unpin,
<<T as ChainInfo>::Block as BlockT>::Header: Unpin,
<<<T as ChainInfo>::Block as BlockT>::Header as Header>::Number:
num_traits::cast::AsPrimitive<usize>,
{
use sp_consensus_babe::AuthorityId;
let config = match config_or_chain_spec {
ConfigOrChainSpec::Config(config) => config,
ConfigOrChainSpec::ChainSpec(chain_spec, tokio_handle) =>
default_config(tokio_handle, chain_spec),
};
let executor = NativeElseWasmExecutor::<T::ExecutorDispatch>::new(
config.wasm_method,
config.default_heap_pages,
config.max_runtime_instances,
);
let (client, backend, keystore, mut task_manager) =
new_full_parts::<T::Block, T::RuntimeApi, _>(&config, None, executor)?;
let client = Arc::new(client);
let select_chain = sc_consensus::LongestChain::new(backend.clone());
let (grandpa_block_import, ..) = grandpa::block_import(
client.clone(),
&(client.clone() as Arc<_>),
select_chain.clone(),
None,
)?;
let slot_duration = sc_consensus_babe::Config::get_or_compute(&*client)?;
let (block_import, babe_link) = sc_consensus_babe::block_import(
slot_duration.clone(),
grandpa_block_import,
client.clone(),
)?;
let consensus_data_provider = BabeConsensusDataProvider::new(
client.clone(),
keystore.sync_keystore(),
babe_link.epoch_changes().clone(),
vec![(AuthorityId::from(Alice.public()), 1000)],
)
.expect("failed to create ConsensusDataProvider");
let import_queue =
import_queue(Box::new(block_import.clone()), &task_manager.spawn_essential_handle(), None);
let transaction_pool = BasicPool::new_full(
config.transaction_pool.clone(),
true.into(),
config.prometheus_registry(),
task_manager.spawn_essential_handle(),
client.clone(),
);
let (network, system_rpc_tx, network_starter) = {
let params = BuildNetworkParams {
config: &config,
client: client.clone(),
transaction_pool: transaction_pool.clone(),
spawn_handle: task_manager.spawn_handle(),
import_queue,
on_demand: None,
block_announce_validator_builder: None,
warp_sync: None,
};
build_network(params)?
};
// offchain workers
sc_service::build_offchain_workers(
&config,
task_manager.spawn_handle(),
client.clone(),
network.clone(),
);
// Proposer object for block authorship.
let env = sc_basic_authorship::ProposerFactory::new(
task_manager.spawn_handle(),
client.clone(),
transaction_pool.clone(),
config.prometheus_registry(),
None,
);
// Channel for the rpc handler to communicate with the authorship task.
let (command_sink, commands_stream) = mpsc::channel(10);
let rpc_sink = command_sink.clone();
let rpc_handlers = {
let params = SpawnTasksParams {
config,
client: client.clone(),
backend: backend.clone(),
task_manager: &mut task_manager,
keystore: keystore.sync_keystore(),
on_demand: None,
transaction_pool: transaction_pool.clone(),
rpc_extensions_builder: Box::new(move |_, _| {
let mut io = jsonrpc_core::IoHandler::default();
io.extend_with(ManualSealApi::to_delegate(ManualSeal::new(rpc_sink.clone())));
Ok(io)
}),
remote_blockchain: None,
network,
system_rpc_tx,
telemetry: None,
};
spawn_tasks(params)?
};
let cloned_client = client.clone();
let create_inherent_data_providers = Box::new(move |_, _| {
let client = cloned_client.clone();
async move {
let timestamp =
SlotTimestampProvider::new(client.clone()).map_err(|err| format!("{:?}", err))?;
let babe =
sp_consensus_babe::inherents::InherentDataProvider::new(timestamp.slot().into());
Ok((timestamp, babe))
}
});
// Background authorship future.
let authorship_future = run_manual_seal(ManualSealParams {
block_import,
env,
client: client.clone(),
pool: transaction_pool.clone(),
commands_stream,
select_chain,
consensus_data_provider: Some(Box::new(consensus_data_provider)),
create_inherent_data_providers,
});
// spawn the authorship task as an essential task.
task_manager.spawn_essential_handle().spawn("manual-seal", authorship_future);
network_starter.start_network();
let rpc_handler = rpc_handlers.io_handler();
Ok((rpc_handler, task_manager, client, transaction_pool, command_sink, backend))
}