Unverified Commit 1a55ca95 authored by Fedor Sakharov's avatar Fedor Sakharov Committed by GitHub
Browse files

runtime-api subsystem lru cache (#2309)

* Add memory-lru cache to runtime-api

* Add cache.rs

* Adds MallocSizeOf

* Review nits

* Add a cached requests metric

* More review nits

* Some more review nits
parent 93fcd8fa
Pipeline #121841 canceled with stages
in 17 minutes and 35 seconds
......@@ -1317,6 +1317,33 @@ dependencies = [
"libc",
]
[[package]]
name = "ethbloom"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22a621dcebea74f2a6f2002d0a885c81ccf6cbdf86760183316a7722b5707ca4"
dependencies = [
"crunchy",
"fixed-hash",
"impl-rlp",
"impl-serde",
"tiny-keccak",
]
[[package]]
name = "ethereum-types"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05dc5f0df4915fa6dff7f975a8366ecfaaa8959c74235469495153e7bb1b280e"
dependencies = [
"ethbloom",
"fixed-hash",
"impl-rlp",
"impl-serde",
"primitive-types",
"uint",
]
[[package]]
name = "event-listener"
version = "2.5.1"
......@@ -2350,6 +2377,15 @@ dependencies = [
"parity-scale-codec",
]
[[package]]
name = "impl-rlp"
version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f28220f89297a075ddc7245cd538076ee98b01f2a9c23a53a4f1105d5a322808"
dependencies = [
"rlp",
]
[[package]]
name = "impl-serde"
version = "0.3.1"
......@@ -3438,6 +3474,15 @@ dependencies = [
"parity-util-mem",
]
[[package]]
name = "memory-lru"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "beeb98b3d1ed2c0054bd81b5ba949a0243c3ccad751d45ea898fa8059fa2860a"
dependencies = [
"lru",
]
[[package]]
name = "memory_units"
version = "0.3.0"
......@@ -4548,9 +4593,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f17f15cb05897127bf36a240085a1f0bbef7bce3024849eccf7f93f6171bc27"
dependencies = [
"cfg-if 1.0.0",
"ethereum-types",
"hashbrown",
"impl-trait-for-tuples 0.2.0",
"jemallocator",
"lru",
"parity-util-mem-derive",
"parking_lot 0.11.1",
"primitive-types",
......@@ -5028,6 +5075,7 @@ name = "polkadot-core-primitives"
version = "0.7.30"
dependencies = [
"parity-scale-codec",
"parity-util-mem",
"sp-core",
"sp-runtime",
"sp-std",
......@@ -5263,6 +5311,8 @@ name = "polkadot-node-core-runtime-api"
version = "0.1.0"
dependencies = [
"futures 0.3.12",
"memory-lru",
"parity-util-mem",
"polkadot-node-subsystem",
"polkadot-node-subsystem-test-helpers",
"polkadot-node-subsystem-util",
......@@ -5429,6 +5479,7 @@ dependencies = [
"futures 0.3.12",
"log",
"parity-scale-codec",
"parity-util-mem",
"parking_lot 0.11.1",
"polkadot-core-primitives",
"sc-executor",
......@@ -5471,6 +5522,7 @@ dependencies = [
"frame-system",
"hex-literal",
"parity-scale-codec",
"parity-util-mem",
"polkadot-core-primitives",
"polkadot-parachain",
"pretty_assertions",
......@@ -6029,6 +6081,7 @@ checksum = "b3824ae2c5e27160113b9e029a10ec9e3f0237bad8029f69c7724393c9fdefd8"
dependencies = [
"fixed-hash",
"impl-codec",
"impl-rlp",
"impl-serde",
"uint",
]
......@@ -6567,6 +6620,16 @@ dependencies = [
"winapi 0.3.9",
]
[[package]]
name = "rlp"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e54369147e3e7796c9b885c7304db87ca3d09a0a98f72843d532868675bbfba8"
dependencies = [
"bytes 1.0.1",
"rustc-hex",
]
[[package]]
name = "rocksdb"
version = "0.15.0"
......
......@@ -9,6 +9,7 @@ sp-core = { git = "https://github.com/paritytech/substrate", branch = "master",
sp-std = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
parity-scale-codec = { version = "1.3.6", default-features = false, features = [ "derive" ] }
parity-util-mem = { version = "0.8.0", default-features = false, optional = true }
[features]
default = [ "std" ]
......@@ -17,4 +18,5 @@ std = [
"sp-runtime/std",
"sp-std/std",
"parity-scale-codec/std",
"parity-util-mem",
]
......@@ -22,6 +22,8 @@
use sp_runtime::{generic, MultiSignature, traits::{Verify, BlakeTwo256, IdentifyAccount}};
use parity_scale_codec::{Encode, Decode};
#[cfg(feature = "std")]
use parity_util_mem::MallocSizeOf;
/// The block number type used by Polkadot.
/// 32-bits will allow for 136 years of blocks assuming 1 block per second.
......@@ -57,6 +59,7 @@ pub type Hash = sp_core::H256;
///
/// This type makes it easy to enforce that a hash is a candidate hash on the type level.
#[derive(Clone, Copy, Encode, Decode, Hash, Eq, PartialEq, Debug, Default)]
#[cfg_attr(feature = "std", derive(MallocSizeOf))]
pub struct CandidateHash(pub Hash);
#[cfg(feature="std")]
......@@ -103,6 +106,7 @@ pub type DownwardMessage = sp_std::vec::Vec<u8>;
/// A wrapped version of `DownwardMessage`. The difference is that it has attached the block number when
/// the message was sent.
#[derive(Encode, Decode, Clone, sp_runtime::RuntimeDebug, PartialEq)]
#[cfg_attr(feature = "std", derive(MallocSizeOf))]
pub struct InboundDownwardMessage<BlockNumber = crate::BlockNumber> {
/// The block number at which this messages was put into the downward message queue.
pub sent_at: BlockNumber,
......@@ -112,6 +116,7 @@ pub struct InboundDownwardMessage<BlockNumber = crate::BlockNumber> {
/// An HRMP message seen from the perspective of a recipient.
#[derive(Encode, Decode, Clone, sp_runtime::RuntimeDebug, PartialEq)]
#[cfg_attr(feature = "std", derive(MallocSizeOf))]
pub struct InboundHrmpMessage<BlockNumber = crate::BlockNumber> {
/// The block number at which this message was sent.
/// Specifically, it is the block number at which the candidate that sends this message was
......@@ -123,6 +128,7 @@ pub struct InboundHrmpMessage<BlockNumber = crate::BlockNumber> {
/// An HRMP message seen from the perspective of a sender.
#[derive(Encode, Decode, Clone, sp_runtime::RuntimeDebug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "std", derive(MallocSizeOf))]
pub struct OutboundHrmpMessage<Id> {
/// The para that will get this message in its downward message queue.
pub recipient: Id,
......
......@@ -8,6 +8,9 @@ edition = "2018"
futures = "0.3.12"
tracing = "0.1.22"
tracing-futures = "0.2.4"
memory-lru = "0.1.0"
parity-util-mem = { version = "0.8.0", default-features = false }
sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
......
// Copyright 2020 Parity Technologies (UK) Ltd.
// This file is part of Polkadot.
// Polkadot 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.
// Polkadot 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 Polkadot. If not, see <http://www.gnu.org/licenses/>.
use polkadot_primitives::v1::{
BlockNumber, CandidateCommitments, CommittedCandidateReceipt, CandidateEvent,
CoreState, GroupRotationInfo, InboundDownwardMessage, InboundHrmpMessage, Hash,
PersistedValidationData, Id as ParaId, OccupiedCoreAssumption,
SessionIndex, SessionInfo, ValidationCode, ValidatorId, ValidatorIndex,
};
use parity_util_mem::{MallocSizeOf, MallocSizeOfExt};
use memory_lru::{MemoryLruCache, ResidentSize};
use std::collections::btree_map::BTreeMap;
const VALIDATORS_CACHE_SIZE: usize = 64 * 1024;
const VALIDATOR_GROUPS_CACHE_SIZE: usize = 64 * 1024;
const AVAILABILITY_CORES_CACHE_SIZE: usize = 64 * 1024;
const PERSISTED_VALIDATION_DATA_CACHE_SIZE: usize = 64 * 1024;
const CHECK_VALIDATION_OUTPUTS_CACHE_SIZE: usize = 64 * 1024;
const SESSION_INDEX_FOR_CHILD_CACHE_SIZE: usize = 64 * 1024;
const VALIDATION_CODE_CACHE_SIZE: usize = 10 * 1024 * 1024;
const HISTORICAL_VALIDATION_CODE_CACHE_SIZE: usize = 10 * 1024 * 1024;
const CANDIDATE_PENDING_AVAILABILITY_CACHE_SIZE: usize = 64 * 1024;
const CANDIDATE_EVENTS_CACHE_SIZE: usize = 64 * 1024;
const SESSION_INFO_CACHE_SIZE: usize = 64 * 1024;
const DMQ_CONTENTS_CACHE_SIZE: usize = 64 * 1024;
const INBOUND_HRMP_CHANNELS_CACHE_SIZE: usize = 64 * 1024;
struct ResidentSizeOf<T>(T);
impl<T: MallocSizeOf> ResidentSize for ResidentSizeOf<T> {
fn resident_size(&self) -> usize {
std::mem::size_of::<Self>() + self.0.malloc_size_of()
}
}
pub(crate) struct RequestResultCache {
validators: MemoryLruCache<Hash, ResidentSizeOf<Vec<ValidatorId>>>,
validator_groups: MemoryLruCache<Hash, ResidentSizeOf<(Vec<Vec<ValidatorIndex>>, GroupRotationInfo)>>,
availability_cores: MemoryLruCache<Hash, ResidentSizeOf<Vec<CoreState>>>,
persisted_validation_data: MemoryLruCache<(Hash, ParaId, OccupiedCoreAssumption), ResidentSizeOf<Option<PersistedValidationData>>>,
check_validation_outputs: MemoryLruCache<(Hash, ParaId, CandidateCommitments), ResidentSizeOf<bool>>,
session_index_for_child: MemoryLruCache<Hash, ResidentSizeOf<SessionIndex>>,
validation_code: MemoryLruCache<(Hash, ParaId, OccupiedCoreAssumption), ResidentSizeOf<Option<ValidationCode>>>,
historical_validation_code: MemoryLruCache<(Hash, ParaId, BlockNumber), ResidentSizeOf<Option<ValidationCode>>>,
candidate_pending_availability: MemoryLruCache<(Hash, ParaId), ResidentSizeOf<Option<CommittedCandidateReceipt>>>,
candidate_events: MemoryLruCache<Hash, ResidentSizeOf<Vec<CandidateEvent>>>,
session_info: MemoryLruCache<(Hash, SessionIndex), ResidentSizeOf<Option<SessionInfo>>>,
dmq_contents: MemoryLruCache<(Hash, ParaId), ResidentSizeOf<Vec<InboundDownwardMessage<BlockNumber>>>>,
inbound_hrmp_channels_contents: MemoryLruCache<(Hash, ParaId), ResidentSizeOf<BTreeMap<ParaId, Vec<InboundHrmpMessage<BlockNumber>>>>>,
}
impl Default for RequestResultCache {
fn default() -> Self {
Self {
validators: MemoryLruCache::new(VALIDATORS_CACHE_SIZE),
validator_groups: MemoryLruCache::new(VALIDATOR_GROUPS_CACHE_SIZE),
availability_cores: MemoryLruCache::new(AVAILABILITY_CORES_CACHE_SIZE),
persisted_validation_data: MemoryLruCache::new(PERSISTED_VALIDATION_DATA_CACHE_SIZE),
check_validation_outputs: MemoryLruCache::new(CHECK_VALIDATION_OUTPUTS_CACHE_SIZE),
session_index_for_child: MemoryLruCache::new(SESSION_INDEX_FOR_CHILD_CACHE_SIZE),
validation_code: MemoryLruCache::new(VALIDATION_CODE_CACHE_SIZE),
historical_validation_code: MemoryLruCache::new(HISTORICAL_VALIDATION_CODE_CACHE_SIZE),
candidate_pending_availability: MemoryLruCache::new(CANDIDATE_PENDING_AVAILABILITY_CACHE_SIZE),
candidate_events: MemoryLruCache::new(CANDIDATE_EVENTS_CACHE_SIZE),
session_info: MemoryLruCache::new(SESSION_INFO_CACHE_SIZE),
dmq_contents: MemoryLruCache::new(DMQ_CONTENTS_CACHE_SIZE),
inbound_hrmp_channels_contents: MemoryLruCache::new(INBOUND_HRMP_CHANNELS_CACHE_SIZE),
}
}
}
impl RequestResultCache {
pub(crate) fn validators(&mut self, relay_parent: &Hash) -> Option<&Vec<ValidatorId>> {
self.validators.get(relay_parent).map(|v| &v.0)
}
pub(crate) fn cache_validators(&mut self, relay_parent: Hash, validators: Vec<ValidatorId>) {
self.validators.insert(relay_parent, ResidentSizeOf(validators));
}
pub(crate) fn validator_groups(&mut self, relay_parent: &Hash) -> Option<&(Vec<Vec<ValidatorIndex>>, GroupRotationInfo)> {
self.validator_groups.get(relay_parent).map(|v| &v.0)
}
pub(crate) fn cache_validator_groups(&mut self, relay_parent: Hash, groups: (Vec<Vec<ValidatorIndex>>, GroupRotationInfo)) {
self.validator_groups.insert(relay_parent, ResidentSizeOf(groups));
}
pub(crate) fn availability_cores(&mut self, relay_parent: &Hash) -> Option<&Vec<CoreState>> {
self.availability_cores.get(relay_parent).map(|v| &v.0)
}
pub(crate) fn cache_availability_cores(&mut self, relay_parent: Hash, cores: Vec<CoreState>) {
self.availability_cores.insert(relay_parent, ResidentSizeOf(cores));
}
pub(crate) fn persisted_validation_data(&mut self, key: (Hash, ParaId, OccupiedCoreAssumption)) -> Option<&Option<PersistedValidationData>> {
self.persisted_validation_data.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_persisted_validation_data(&mut self, key: (Hash, ParaId, OccupiedCoreAssumption), data: Option<PersistedValidationData>) {
self.persisted_validation_data.insert(key, ResidentSizeOf(data));
}
pub(crate) fn check_validation_outputs(&mut self, key: (Hash, ParaId, CandidateCommitments)) -> Option<&bool> {
self.check_validation_outputs.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_check_validation_outputs(&mut self, key: (Hash, ParaId, CandidateCommitments), value: bool) {
self.check_validation_outputs.insert(key, ResidentSizeOf(value));
}
pub(crate) fn session_index_for_child(&mut self, relay_parent: &Hash) -> Option<&SessionIndex> {
self.session_index_for_child.get(relay_parent).map(|v| &v.0)
}
pub(crate) fn cache_session_index_for_child(&mut self, relay_parent: Hash, index: SessionIndex) {
self.session_index_for_child.insert(relay_parent, ResidentSizeOf(index));
}
pub(crate) fn validation_code(&mut self, key: (Hash, ParaId, OccupiedCoreAssumption)) -> Option<&Option<ValidationCode>> {
self.validation_code.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_validation_code(&mut self, key: (Hash, ParaId, OccupiedCoreAssumption), value: Option<ValidationCode>) {
self.validation_code.insert(key, ResidentSizeOf(value));
}
pub(crate) fn historical_validation_code(&mut self, key: (Hash, ParaId, BlockNumber)) -> Option<&Option<ValidationCode>> {
self.historical_validation_code.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_historical_validation_code(&mut self, key: (Hash, ParaId, BlockNumber), value: Option<ValidationCode>) {
self.historical_validation_code.insert(key, ResidentSizeOf(value));
}
pub(crate) fn candidate_pending_availability(&mut self, key: (Hash, ParaId)) -> Option<&Option<CommittedCandidateReceipt>> {
self.candidate_pending_availability.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_candidate_pending_availability(&mut self, key: (Hash, ParaId), value: Option<CommittedCandidateReceipt>) {
self.candidate_pending_availability.insert(key, ResidentSizeOf(value));
}
pub(crate) fn candidate_events(&mut self, relay_parent: &Hash) -> Option<&Vec<CandidateEvent>> {
self.candidate_events.get(relay_parent).map(|v| &v.0)
}
pub(crate) fn cache_candidate_events(&mut self, relay_parent: Hash, events: Vec<CandidateEvent>) {
self.candidate_events.insert(relay_parent, ResidentSizeOf(events));
}
pub(crate) fn session_info(&mut self, key: (Hash, SessionIndex)) -> Option<&Option<SessionInfo>> {
self.session_info.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_session_info(&mut self, key: (Hash, SessionIndex), value: Option<SessionInfo>) {
self.session_info.insert(key, ResidentSizeOf(value));
}
pub(crate) fn dmq_contents(&mut self, key: (Hash, ParaId)) -> Option<&Vec<InboundDownwardMessage<BlockNumber>>> {
self.dmq_contents.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_dmq_contents(&mut self, key: (Hash, ParaId), value: Vec<InboundDownwardMessage<BlockNumber>>) {
self.dmq_contents.insert(key, ResidentSizeOf(value));
}
pub(crate) fn inbound_hrmp_channels_contents(&mut self, key: (Hash, ParaId)) -> Option<&BTreeMap<ParaId, Vec<InboundHrmpMessage<BlockNumber>>>> {
self.inbound_hrmp_channels_contents.get(&key).map(|v| &v.0)
}
pub(crate) fn cache_inbound_hrmp_channel_contents(&mut self, key: (Hash, ParaId), value: BTreeMap<ParaId, Vec<InboundHrmpMessage<BlockNumber>>>) {
self.inbound_hrmp_channels_contents.insert(key, ResidentSizeOf(value));
}
}
pub(crate) enum RequestResult {
Validators(Hash, Vec<ValidatorId>),
ValidatorGroups(Hash, (Vec<Vec<ValidatorIndex>>, GroupRotationInfo)),
AvailabilityCores(Hash, Vec<CoreState>),
PersistedValidationData(Hash, ParaId, OccupiedCoreAssumption, Option<PersistedValidationData>),
CheckValidationOutputs(Hash, ParaId, CandidateCommitments, bool),
SessionIndexForChild(Hash, SessionIndex),
ValidationCode(Hash, ParaId, OccupiedCoreAssumption, Option<ValidationCode>),
HistoricalValidationCode(Hash, ParaId, BlockNumber, Option<ValidationCode>),
CandidatePendingAvailability(Hash, ParaId, Option<CommittedCandidateReceipt>),
CandidateEvents(Hash, Vec<CandidateEvent>),
SessionInfo(Hash, SessionIndex, Option<SessionInfo>),
DmqContents(Hash, ParaId, Vec<InboundDownwardMessage<BlockNumber>>),
InboundHrmpChannelsContents(Hash, ParaId, BTreeMap<ParaId, Vec<InboundHrmpMessage<BlockNumber>>>),
}
......@@ -38,6 +38,9 @@ use sp_core::traits::SpawnNamed;
use futures::{prelude::*, stream::FuturesUnordered, channel::oneshot, select};
use std::{sync::Arc, collections::VecDeque, pin::Pin};
use cache::{RequestResult, RequestResultCache};
mod cache;
const LOG_TARGET: &str = "runtime_api";
......@@ -53,9 +56,14 @@ pub struct RuntimeApiSubsystem<Client> {
metrics: Metrics,
spawn_handle: Box<dyn SpawnNamed>,
/// If there are [`MAX_PARALLEL_REQUESTS`] requests being executed, we buffer them in here until they can be executed.
waiting_requests: VecDeque<(Pin<Box<dyn Future<Output = ()> + Send>>, oneshot::Receiver<()>)>,
waiting_requests: VecDeque<(
Pin<Box<dyn Future<Output = ()> + Send>>,
oneshot::Receiver<Option<RequestResult>>,
)>,
/// All the active runtime api requests that are currently being executed.
active_requests: FuturesUnordered<oneshot::Receiver<()>>,
active_requests: FuturesUnordered<oneshot::Receiver<Option<RequestResult>>>,
/// Requests results cache
requests_cache: RequestResultCache,
}
impl<Client> RuntimeApiSubsystem<Client> {
......@@ -67,6 +75,7 @@ impl<Client> RuntimeApiSubsystem<Client> {
spawn_handle: Box::new(spawn_handle),
waiting_requests: Default::default(),
active_requests: Default::default(),
requests_cache: RequestResultCache::default(),
}
}
}
......@@ -88,6 +97,102 @@ impl<Client> RuntimeApiSubsystem<Client> where
Client: ProvideRuntimeApi<Block> + Send + 'static + Sync,
Client::Api: ParachainHost<Block>,
{
fn store_cache(&mut self, result: RequestResult) {
use RequestResult::*;
match result {
Validators(relay_parent, validators) =>
self.requests_cache.cache_validators(relay_parent, validators),
ValidatorGroups(relay_parent, groups) =>
self.requests_cache.cache_validator_groups(relay_parent, groups),
AvailabilityCores(relay_parent, cores) =>
self.requests_cache.cache_availability_cores(relay_parent, cores),
PersistedValidationData(relay_parent, para_id, assumption, data) =>
self.requests_cache.cache_persisted_validation_data((relay_parent, para_id, assumption), data),
CheckValidationOutputs(relay_parent, para_id, commitments, b) =>
self.requests_cache.cache_check_validation_outputs((relay_parent, para_id, commitments), b),
SessionIndexForChild(relay_parent, session_index) =>
self.requests_cache.cache_session_index_for_child(relay_parent, session_index),
ValidationCode(relay_parent, para_id, assumption, code) =>
self.requests_cache.cache_validation_code((relay_parent, para_id, assumption), code),
HistoricalValidationCode(relay_parent, para_id, n, code) =>
self.requests_cache.cache_historical_validation_code((relay_parent, para_id, n), code),
CandidatePendingAvailability(relay_parent, para_id, candidate) =>
self.requests_cache.cache_candidate_pending_availability((relay_parent, para_id), candidate),
CandidateEvents(relay_parent, events) =>
self.requests_cache.cache_candidate_events(relay_parent, events),
SessionInfo(relay_parent, session_index, info) =>
self.requests_cache.cache_session_info((relay_parent, session_index), info),
DmqContents(relay_parent, para_id, messages) =>
self.requests_cache.cache_dmq_contents((relay_parent, para_id), messages),
InboundHrmpChannelsContents(relay_parent, para_id, contents) =>
self.requests_cache.cache_inbound_hrmp_channel_contents((relay_parent, para_id), contents),
}
}
fn query_cache(&mut self, relay_parent: Hash, request: Request) -> Option<Request> {
macro_rules! query {
// Just query by relay parent
($cache_api_name:ident (), $sender:expr) => {{
let sender = $sender;
if let Some(value) = self.requests_cache.$cache_api_name(&relay_parent) {
let _ = sender.send(Ok(value.clone()));
self.metrics.on_cached_request();
None
} else {
Some(sender)
}
}};
// Query by relay parent + additional parameters
($cache_api_name:ident ($($param:expr),+), $sender:expr) => {{
let sender = $sender;
if let Some(value) = self.requests_cache.$cache_api_name((relay_parent.clone(), $($param.clone()),+)) {
self.metrics.on_cached_request();
let _ = sender.send(Ok(value.clone()));
None
} else {
Some(sender)
}
}}
}
match request {
Request::Validators(sender) => query!(validators(), sender)
.map(|sender| Request::Validators(sender)),
Request::ValidatorGroups(sender) => query!(validator_groups(), sender)
.map(|sender| Request::ValidatorGroups(sender)),
Request::AvailabilityCores(sender) => query!(availability_cores(), sender)
.map(|sender| Request::AvailabilityCores(sender)),
Request::PersistedValidationData(para, assumption, sender) =>
query!(persisted_validation_data(para, assumption), sender)
.map(|sender| Request::PersistedValidationData(para, assumption, sender)),
Request::CheckValidationOutputs(para, commitments, sender) =>
query!(check_validation_outputs(para, commitments), sender)
.map(|sender| Request::CheckValidationOutputs(para, commitments, sender)),
Request::SessionIndexForChild(sender) =>
query!(session_index_for_child(), sender)
.map(|sender| Request::SessionIndexForChild(sender)),
Request::ValidationCode(para, assumption, sender) =>
query!(validation_code(para, assumption), sender)
.map(|sender| Request::ValidationCode(para, assumption, sender)),
Request::HistoricalValidationCode(para, at, sender) =>
query!(historical_validation_code(para, at), sender)
.map(|sender| Request::HistoricalValidationCode(para, at, sender)),
Request::CandidatePendingAvailability(para, sender) =>
query!(candidate_pending_availability(para), sender)
.map(|sender| Request::CandidatePendingAvailability(para, sender)),
Request::CandidateEvents(sender) => query!(candidate_events(), sender)
.map(|sender| Request::CandidateEvents(sender)),
Request::SessionInfo(index, sender) => query!(session_info(index), sender)
.map(|sender| Request::SessionInfo(index, sender)),
Request::DmqContents(id, sender) => query!(dmq_contents(id), sender)
.map(|sender| Request::DmqContents(id, sender)),
Request::InboundHrmpChannelsContents(id, sender) =>
query!(inbound_hrmp_channels_contents(id), sender)
.map(|sender| Request::InboundHrmpChannelsContents(id, sender))
}
}
/// Spawn a runtime api request.
///
/// If there are already [`MAX_PARALLEL_REQUESTS`] requests being executed, the request will be buffered.
......@@ -96,14 +201,19 @@ impl<Client> RuntimeApiSubsystem<Client> where
let metrics = self.metrics.clone();
let (sender, receiver) = oneshot::channel();
let request = match self.query_cache(relay_parent.clone(), request) {
Some(request) => request,
None => return,
};
let request = async move {
make_runtime_api_request(
let result = make_runtime_api_request(
client,
metrics,
relay_parent,
request,
);
let _ = sender.send(());
let _ = sender.send(result);
}.boxed();
if self.active_requests.len() >= MAX_PARALLEL_REQUESTS {
......@@ -130,7 +240,9 @@ impl<Client> RuntimeApiSubsystem<Client> where
}
// If there are active requests, this will always resolve to `Some(_)` when a request is finished.
let _ = self.active_requests.next().await;
if let Some(Ok(Some(result))) = self.active_requests.next().await {
self.store_cache(result);
}
if let Some((req, recv)) = self.waiting_requests.pop_front() {
self.spawn_handle.spawn_blocking(API_REQUEST_TASK_NAME, req);
......@@ -170,42 +282,63 @@ fn make_runtime_api_request<Client>(
metrics: Metrics,
relay_parent: Hash,
request: Request,
) where
) -> Option<RequestResult>
where
Client: ProvideRuntimeApi<Block>,
Client::Api: ParachainHost<Block>,
{
let _timer = metrics.time_make_runtime_api_request();
macro_rules! query {
($api_name:ident ($($param:expr),*), $sender:expr) => {{
($req_variant:ident, $api_name:ident (), $sender:expr) => {{
let sender = $sender;
let api = client.runtime_api();
let res = api.$api_name(&BlockId::Hash(relay_parent), $($param),*)
let res = api.$api_name(&BlockId::Hash(relay_parent))
.map_err(|e| RuntimeApiError::from(format!("{:?}", e)));
metrics.on_request(res.is_ok());
let _ = sender.send(res);
let _ = sender.send(res.clone());
if let Ok(res) = res {
Some(RequestResult::$req_variant(