frost/bdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make esplora call in parallel

Browse Source
This commit is contained in:
Riccardo Casatta 2020-11-03 22:09:32 +01:00
parent fe7ecd3dd2
commit 4c59809f8e
No known key found for this signature in database
GPG Key ID: FD986A969E450397
1 changed files with 34 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
src/blockchain/esplora.rs
View File

@ -38,7 +38,7 @@
use std::collections::{HashMap, HashSet};
use std::fmt;
use futures::stream::{self, StreamExt, TryStreamExt};
use futures::stream::{self, FuturesOrdered, StreamExt, TryStreamExt};
#[allow(unused_imports)]
use log::{debug, error, info, trace};
@ -56,8 +56,11 @@ use self::utils::{ELSGetHistoryRes, ElectrumLikeSync};
use super::*;
use crate::database::BatchDatabase;
use crate::error::Error;
use crate::wallet::utils::ChunksIterator;
use crate::FeeRate;
const CONCURRENT: usize = 4;
#[derive(Debug)]
struct UrlClient {
url: String,
@ -301,10 +304,16 @@ impl ElectrumLikeSync for UrlClient {
scripts: I,
) -> Result<Vec<Vec<ELSGetHistoryRes>>, Error> {
let future = async {
Ok(stream::iter(scripts)
.then(|script| self._script_get_history(&script))
.try_collect()
.await?)
let mut results = vec![];
for chunk in ChunksIterator::new(scripts.into_iter(), CONCURRENT) {
let mut futs = FuturesOrdered::new();
for script in chunk {
futs.push(self._script_get_history(&script));
}
let partial_results: Vec<Vec<ELSGetHistoryRes>> = futs.try_collect().await?;
results.extend(partial_results);
}
Ok(stream::iter(results).collect().await)
};
await_or_block!(future)
@ -315,10 +324,16 @@ impl ElectrumLikeSync for UrlClient {
txids: I,
) -> Result<Vec<Transaction>, Error> {
let future = async {
Ok(stream::iter(txids)
.then(|txid| self._get_tx_no_opt(&txid))
.try_collect()
.await?)
let mut results = vec![];
for chunk in ChunksIterator::new(txids.into_iter(), CONCURRENT) {
let mut futs = FuturesOrdered::new();
for txid in chunk {
futs.push(self._get_tx_no_opt(&txid));
}
let partial_results: Vec<Transaction> = futs.try_collect().await?;
results.extend(partial_results);
}
Ok(stream::iter(results).collect().await)
};
await_or_block!(future)
@ -329,10 +344,16 @@ impl ElectrumLikeSync for UrlClient {
heights: I,
) -> Result<Vec<BlockHeader>, Error> {
let future = async {
Ok(stream::iter(heights)
.then(|h| self._get_header(h))
.try_collect()
.await?)
let mut results = vec![];
for chunk in ChunksIterator::new(heights.into_iter(), CONCURRENT) {
let mut futs = FuturesOrdered::new();
for height in chunk {
futs.push(self._get_header(height));
}
let partial_results: Vec<BlockHeader> = futs.try_collect().await?;
results.extend(partial_results);
}
Ok(stream::iter(results).collect().await)
};
await_or_block!(future)