Compare commits

...

3 Commits

Author SHA1 Message Date
0g-peterzhb
b5eec9ae46
Merge f456773b72 into 53449e1faa 2024-08-09 17:13:41 +08:00
peilun-conflux
53449e1faa
Half log_page_size when it queries too many logs. (#152)
Some checks failed
abi-consistent-check / build-and-compare (push) Has been cancelled
code-coverage / unittest-cov (push) Has been cancelled
rust / check (push) Has been cancelled
rust / test (push) Has been cancelled
rust / lints (push) Has been cancelled
functional-test / test (push) Has been cancelled
* Half log_page_size when it queries too many logs.

* fmt.

* Increase log broadcast channel size.
2024-08-09 17:13:20 +08:00
Peter Zhang
f456773b72 remove update_config.sh, users can pass in cli params to the program 2024-07-17 17:43:19 +08:00
3 changed files with 39 additions and 42 deletions

View File

@ -14,6 +14,8 @@ use thiserror::Error;
pub(crate) type PinBoxFut<'a, T> = pub(crate) type PinBoxFut<'a, T> =
Pin<Box<dyn Future<Output = Result<T, ProviderError>> + Send + 'a>>; Pin<Box<dyn Future<Output = Result<T, ProviderError>> + Send + 'a>>;
const TOO_MANY_LOGS_ERROR_MSG: &str = "query returned more than";
/// A log query provides streaming access to historical logs via a paginated /// A log query provides streaming access to historical logs via a paginated
/// request. For streaming access to future logs, use [`Middleware::watch`] or /// request. For streaming access to future logs, use [`Middleware::watch`] or
/// [`Middleware::subscribe_logs`] /// [`Middleware::subscribe_logs`]
@ -21,6 +23,9 @@ pub struct LogQuery<'a, P> {
provider: &'a Provider<P>, provider: &'a Provider<P>,
filter: Filter, filter: Filter,
from_block: Option<U64>, from_block: Option<U64>,
expected_page_size: u64,
/// It may be smaller than `expected_page_size` if the server cannot return all the logs.
page_size: u64, page_size: u64,
current_logs: VecDeque<Log>, current_logs: VecDeque<Log>,
last_block: Option<U64>, last_block: Option<U64>,
@ -31,7 +36,8 @@ pub struct LogQuery<'a, P> {
enum LogQueryState<'a> { enum LogQueryState<'a> {
Initial, Initial,
LoadLastBlock(PinBoxFut<'a, U64>), LoadLastBlock(PinBoxFut<'a, U64>),
LoadLogs(PinBoxFut<'a, Vec<Log>>), /// `(from_block, get_logs_fut)`. `from_block` is used to resume if the request fails.
LoadLogs((Option<U64>, PinBoxFut<'a, Vec<Log>>)),
Consume, Consume,
} }
@ -45,6 +51,7 @@ where
provider, provider,
filter: filter.clone(), filter: filter.clone(),
from_block: filter.get_from_block(), from_block: filter.get_from_block(),
expected_page_size: 10000,
page_size: 10000, page_size: 10000,
current_logs: VecDeque::new(), current_logs: VecDeque::new(),
last_block: None, last_block: None,
@ -56,6 +63,7 @@ where
/// set page size for pagination /// set page size for pagination
pub fn with_page_size(mut self, page_size: u64) -> Self { pub fn with_page_size(mut self, page_size: u64) -> Self {
self.page_size = page_size; self.page_size = page_size;
self.expected_page_size = page_size;
self self
} }
} }
@ -98,7 +106,7 @@ where
tokio::time::sleep(delay).await; tokio::time::sleep(delay).await;
provider.get_logs(&filter).await provider.get_logs(&filter).await
}); });
rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs(fut)); rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs((None, fut)));
} else { } else {
// if paginatable, load last block // if paginatable, load last block
let fut = match self.filter.get_to_block() { let fut = match self.filter.get_to_block() {
@ -134,18 +142,33 @@ where
tokio::time::sleep(delay).await; tokio::time::sleep(delay).await;
provider.get_logs(&filter).await provider.get_logs(&filter).await
}); });
rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs(fut)); rewake_with_new_state!(
ctx,
self,
LogQueryState::LoadLogs((Some(from_block), fut))
);
} }
Err(err) => Poll::Ready(Some(Err(LogQueryError::LoadLastBlockError(err)))), Err(err) => Poll::Ready(Some(Err(LogQueryError::LoadLastBlockError(err)))),
} }
} }
LogQueryState::LoadLogs(fut) => match futures_util::ready!(fut.as_mut().poll(ctx)) { LogQueryState::LoadLogs((from_block, fut)) => {
Ok(logs) => { match futures_util::ready!(fut.as_mut().poll(ctx)) {
self.current_logs = VecDeque::from(logs); Ok(logs) => {
rewake_with_new_state!(ctx, self, LogQueryState::Consume); self.current_logs = VecDeque::from(logs);
self.page_size = self.expected_page_size;
rewake_with_new_state!(ctx, self, LogQueryState::Consume);
}
Err(err) => {
if err.to_string().contains(TOO_MANY_LOGS_ERROR_MSG) {
self.from_block = *from_block;
self.page_size /= 2;
rewake_with_new_state!(ctx, self, LogQueryState::Consume);
} else {
Poll::Ready(Some(Err(LogQueryError::LoadLogsError(err))))
}
}
} }
Err(err) => Poll::Ready(Some(Err(LogQueryError::LoadLogsError(err)))), }
},
LogQueryState::Consume => { LogQueryState::Consume => {
let log = self.current_logs.pop_front(); let log = self.current_logs.pop_front();
if log.is_none() { if log.is_none() {
@ -183,7 +206,11 @@ where
provider.get_logs(&filter).await provider.get_logs(&filter).await
}); });
rewake_with_new_state!(ctx, self, LogQueryState::LoadLogs(fut)); rewake_with_new_state!(
ctx,
self,
LogQueryState::LoadLogs((Some(from_block), fut))
);
} }
} else { } else {
Poll::Ready(log.map(Ok)) Poll::Ready(log.map(Ok))

View File

@ -21,7 +21,8 @@ use tokio::sync::{oneshot, RwLock};
const RETRY_WAIT_MS: u64 = 500; const RETRY_WAIT_MS: u64 = 500;
// A RPC query can return at most 10000 entries. // A RPC query can return at most 10000 entries.
const BROADCAST_CHANNEL_CAPACITY: usize = 10000; // Each tx has less than 10KB, so the cache size should be acceptable.
const BROADCAST_CHANNEL_CAPACITY: usize = 25000;
const CATCH_UP_END_GAP: u64 = 10; const CATCH_UP_END_GAP: u64 = 10;
#[derive(Clone, Debug)] #[derive(Clone, Debug)]

View File

@ -1,31 +0,0 @@
#!/bin/bash
MINER_KEY=""
MINE_CONTRACT=""
BLOCKCHAIN_RPC=""
FLOW_CONTRACT=""
BLOCK_NUMBER=0
PUBLIC_IP=$(curl -s https://ipinfo.io/ip)
FILE=run/config.toml
# enable sync
sed -in-place='' 's/# \[sync\]/\[sync\]/g' $FILE
# enable auto_sync
sed -in-place='' 's/# auto_sync_enabled = false/auto_sync_enabled = true/g' $FILE
# reduce timeout for finding peers
sed -in-place='' 's/# find_peer_timeout = .*/find_peer_timeout = "10s"/g' $FILE
# set public ip
sed -in-place='' "s/# network_enr_address = .*/network_enr_address = \"$PUBLIC_IP\"/g" $FILE
# set miner key
sed -in-place='' "s/miner_key = \"\"/miner_key = \"$MINER_KEY\"/g" $FILE
# set miner contract address
sed -in-place='' "s/mine_contract_address = .*/mine_contract_address = \"$MINE_CONTRACT\"/g" $FILE
# set blockchain rpc endpoint
sed -in-place='' "s|blockchain_rpc_endpoint = .*|blockchain_rpc_endpoint = \"$BLOCKCHAIN_RPC\"|g" $FILE
# set flow contract address
sed -in-place='' "s/log_contract_address = .*/log_contract_address = \"$FLOW_CONTRACT\"/g" $FILE
# set contract deployed block number
sed -in-place='' "s/log_sync_start_block_number = .*/log_sync_start_block_number = $BLOCK_NUMBER/g" $FILE
# update the boot node ids
sed -in-place='' 's|network_boot_nodes = .*|network_boot_nodes = ["/ip4/54.219.26.22/udp/1234/p2p/16Uiu2HAmTVDGNhkHD98zDnJxQWu3i1FL1aFYeh9wiQTNu4pDCgps","/ip4/52.52.127.117/udp/1234/p2p/16Uiu2HAkzRjxK2gorngB1Xq84qDrT4hSVznYDHj6BkbaE4SGx9oS","/ip4/18.167.69.68/udp/1234/p2p/16Uiu2HAm2k6ua2mGgvZ8rTMV8GhpW71aVzkQWy7D37TTDuLCpgmX"]|g' $FILE