octo
/
Zebra
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Stop using ServiceExt::call_all due to buffer bugs 

ServiceExt::call_all leaks Tower::Buffer reservations, so we can't use
it in Zebra.

Instead, use a loop in the returned future.

See #1593 for details.
This commit is contained in:
teor 2021-01-25 14:55:36 +10:00
parent 64bc45cd2e
commit 94eb91305b
2 changed files with 24 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
zebra-state/src/service.rs
View File

@ -530,13 +530,6 @@ impl Service<Request> for StateService {
type Future =
Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
// ## Correctness:
//
// This function must not return Poll::Pending, unless:
// 1. We remove all instances of `call_all` on the state service, or fix the leaked
// service reservation in the `CallAll` implementation:
// https://github.com/tower-rs/tower/blob/master/tower/src/util/call_all/common.rs#L112
// 2. We schedule the current task for wakeup via the `Context`
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
let now = Instant::now();

66
zebrad/src/components/inbound.rs
View File

@ -7,7 +7,7 @@ use std::{
use futures::{
future::{FutureExt, TryFutureExt},
stream::{Stream, TryStreamExt},
stream::Stream,
};
use tokio::sync::oneshot;
use tower::{buffer::Buffer, timeout::Timeout, util::BoxService, Service, ServiceExt};
@ -52,12 +52,6 @@ pub type SetupData = (Outbound, Arc<Mutex<AddressBook>>);
/// behind the current tip, while the `Inbound` service is *externally driven*,
/// responding to block gossip by attempting to download and validate advertised
/// blocks.
///
/// ## Correctness
///
/// The `state` service must not return `Poll::Pending`. If it does, a bug in the
/// `ServiceExt::call_all` implementation might cause the `state` buffer to fill
/// up, and make Zebra hang.
pub struct Inbound {
// invariants:
// * Before setup: address_book and downloads are None, and the *_setup members are Some
@ -160,18 +154,14 @@ impl Service<zn::Request> for Inbound {
}
// TODO:
// * do we want to propagate backpressure from the download queue or its outbound network here?
// currently, the download queue waits for the outbound network in the download future, and
// drops new requests after it reaches a hard-coded limit. This is the "load shed directly"
// pattern from #1618.
// * if we want to propagate backpressure, add a ReadyCache to ensure that each poll_ready
// has a matching call. See #1593 for details.
// Ignore state readiness, to avoid reserving its buffer slots.
// We can't use a state ReadyCache, because call_all uses state directly.
// We can't call state.poll_ready, because:
// * call_all also calls poll_ready
// * some requests don't use the state
// * do we want to propagate backpressure from the download queue or its outbound network?
// currently, the download queue waits for the outbound network in the download future,
// and drops new requests after it reaches a hard-coded limit. This is the
// "load shed directly" pattern from #1618.
// * currently, the state service is always ready, unless its buffer is full.
// So we might also want to propagate backpressure from its buffer.
// * if we want to propagate backpressure, add a ReadyCache for each service, to ensure
// that each poll_ready has a matching call. See #1593 for details.
Poll::Ready(Ok(()))
}
@ -194,34 +184,26 @@ impl Service<zn::Request> for Inbound {
zn::Request::BlocksByHash(hashes) => {
// Correctness:
//
// We don't need to use ServiceExt::oneshot here, because
// call_all uses poll_ready internally.
//
// The state must not return Poll::Pending, because call_all
// leaks a buffer reservation every time that happens
// We can't use `call_all` here, because it leaks buffer slots:
// https://github.com/tower-rs/tower/blob/master/tower/src/util/call_all/common.rs#L112
let state = self.state.clone();
let requests = futures::stream::iter(
hashes
.into_iter()
.map(|hash| zs::Request::Block(hash.into())),
);
state
.call_all(requests)
.try_filter_map(|rsp| {
futures::future::ready(match rsp {
zs::Response::Block(Some(block)) => Ok(Some(block)),
let mut state = self.state.clone();
async move {
let mut blocks = Vec::new();
for hash in hashes {
let request = zs::Request::Block(hash.into());
// we can't use ServiceExt::oneshot here, due to lifetime issues
match state.ready_and().await?.call(request).await? {
zs::Response::Block(Some(block)) => blocks.push(block),
// `zcashd` ignores missing blocks in GetData responses,
// rather than including them in a trailing `NotFound`
// message
zs::Response::Block(None) => Ok(None),
zs::Response::Block(None) => {}
_ => unreachable!("wrong response from state"),
})
})
.try_collect::<Vec<_>>()
.map_ok(zn::Response::Blocks)
.boxed()
}
}
Ok(zn::Response::Blocks(blocks))
}
.boxed()
}
zn::Request::TransactionsByHash(_transactions) => {
// `zcashd` returns a list of found transactions, followed by a