mirror of
https://github.com/0glabs/0g-storage-node.git
synced 2024-11-10 10:05:17 +00:00
Remove file location cache for connection timeout peer (#96)
This commit is contained in:
parent
fa74a4b9c1
commit
439314372b
@ -126,6 +126,12 @@ impl AnnouncementCache {
|
|||||||
let result = self.items.values().cloned().collect();
|
let result = self.items.values().cloned().collect();
|
||||||
(result, collected)
|
(result, collected)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Removes announcement for the specified `peer_id` if any.
|
||||||
|
fn remove(&mut self, peer_id: &PeerId) -> Option<SignedAnnounceFile> {
|
||||||
|
self.priorities.remove(peer_id)?;
|
||||||
|
self.items.remove(peer_id)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Caches announcements for different files.
|
/// Caches announcements for different files.
|
||||||
@ -230,6 +236,14 @@ impl FileCache {
|
|||||||
self.update_on_announcement_cache_changed(&tx_id, collected);
|
self.update_on_announcement_cache_changed(&tx_id, collected);
|
||||||
Some(result)
|
Some(result)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Removes the announcement of specified file by `tx_id` and `peer_id`.
|
||||||
|
fn remove(&mut self, tx_id: &TxID, peer_id: &PeerId) -> Option<SignedAnnounceFile> {
|
||||||
|
let item = self.files.get_mut(tx_id)?;
|
||||||
|
let result = item.remove(peer_id)?;
|
||||||
|
self.update_on_announcement_cache_changed(tx_id, 1);
|
||||||
|
Some(result)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
@ -281,6 +295,10 @@ impl FileLocationCache {
|
|||||||
self.cache.lock().all(tx_id).unwrap_or_default()
|
self.cache.lock().all(tx_id).unwrap_or_default()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn remove(&self, tx_id: &TxID, peer_id: &PeerId) -> Option<SignedAnnounceFile> {
|
||||||
|
self.cache.lock().remove(tx_id, peer_id)
|
||||||
|
}
|
||||||
|
|
||||||
/// TODO: Trigger chunk_pool/sync to reconstruct if it changes?
|
/// TODO: Trigger chunk_pool/sync to reconstruct if it changes?
|
||||||
pub fn insert_peer_config(
|
pub fn insert_peer_config(
|
||||||
&self,
|
&self,
|
||||||
|
@ -1,5 +1,7 @@
|
|||||||
|
use file_location_cache::FileLocationCache;
|
||||||
use network::{Multiaddr, PeerAction, PeerId};
|
use network::{Multiaddr, PeerAction, PeerId};
|
||||||
use rand::seq::IteratorRandom;
|
use rand::seq::IteratorRandom;
|
||||||
|
use shared_types::TxID;
|
||||||
use std::cmp::Ordering;
|
use std::cmp::Ordering;
|
||||||
use std::collections::{BTreeSet, HashMap};
|
use std::collections::{BTreeSet, HashMap};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
@ -46,13 +48,19 @@ impl PeerInfo {
|
|||||||
pub struct SyncPeers {
|
pub struct SyncPeers {
|
||||||
peers: HashMap<PeerId, PeerInfo>,
|
peers: HashMap<PeerId, PeerInfo>,
|
||||||
ctx: Option<Arc<SyncNetworkContext>>,
|
ctx: Option<Arc<SyncNetworkContext>>,
|
||||||
|
file_location_cache: Option<(TxID, Arc<FileLocationCache>)>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SyncPeers {
|
impl SyncPeers {
|
||||||
pub fn new(ctx: Arc<SyncNetworkContext>) -> Self {
|
pub fn new(
|
||||||
|
ctx: Arc<SyncNetworkContext>,
|
||||||
|
tx_id: TxID,
|
||||||
|
file_location_cache: Arc<FileLocationCache>,
|
||||||
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
peers: Default::default(),
|
peers: Default::default(),
|
||||||
ctx: Some(ctx),
|
ctx: Some(ctx),
|
||||||
|
file_location_cache: Some((tx_id, file_location_cache)),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -192,6 +200,8 @@ impl SyncPeers {
|
|||||||
if info.since.elapsed() >= PEER_CONNECT_TIMEOUT {
|
if info.since.elapsed() >= PEER_CONNECT_TIMEOUT {
|
||||||
info!(%peer_id, %info.addr, "Peer connection timeout");
|
info!(%peer_id, %info.addr, "Peer connection timeout");
|
||||||
bad_peers.push(*peer_id);
|
bad_peers.push(*peer_id);
|
||||||
|
|
||||||
|
// Ban peer in case of continuous connection timeout
|
||||||
if let Some(ctx) = &self.ctx {
|
if let Some(ctx) = &self.ctx {
|
||||||
ctx.report_peer(
|
ctx.report_peer(
|
||||||
*peer_id,
|
*peer_id,
|
||||||
@ -199,6 +209,11 @@ impl SyncPeers {
|
|||||||
"Dail timeout",
|
"Dail timeout",
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Remove cached file announcement if connection timeout
|
||||||
|
if let Some((tx_id, cache)) = &self.file_location_cache {
|
||||||
|
cache.remove(tx_id, peer_id);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -110,7 +110,7 @@ impl SerialSyncController {
|
|||||||
next_chunk: goal.index_start,
|
next_chunk: goal.index_start,
|
||||||
failures: 0,
|
failures: 0,
|
||||||
state: SyncState::Idle,
|
state: SyncState::Idle,
|
||||||
peers: SyncPeers::new(ctx.clone()),
|
peers: SyncPeers::new(ctx.clone(), tx_id, file_location_cache.clone()),
|
||||||
ctx,
|
ctx,
|
||||||
store,
|
store,
|
||||||
file_location_cache,
|
file_location_cache,
|
||||||
|
Loading…
Reference in New Issue
Block a user