smeb_y/0g-storage-node
1
0
Fork 0
mirror of https://github.com/0glabs/0g-storage-node.git synced 2025-04-04 15:35:18 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix random test failure

Browse Source
This commit is contained in:
boqiu 2024-10-25 12:23:08 +08:00
parent da903fefe7
commit 98c62b314e
1 changed files with 28 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
node/sync/src/auto_sync/manager.rs
View File

@ -35,17 +35,21 @@ impl AutoSyncManager {
executor: &TaskExecutor, executor: &TaskExecutor,
store: Store, store: Store,
sync_send: SyncSender, sync_send: SyncSender,
_log_sync_recv: broadcast::Receiver<LogSyncEvent>, log_sync_recv: broadcast::Receiver<LogSyncEvent>,
catch_up_end_recv: oneshot::Receiver<()>, catch_up_end_recv: oneshot::Receiver<()>,
) -> Result<Self> { ) -> Result<Self> {
let (file_announcement_send, _file_announcement_recv) = unbounded_channel(); let (file_announcement_send, file_announcement_recv) = unbounded_channel();
let (new_file_send, new_file_recv) = unbounded_channel(); let (new_file_send, new_file_recv) = unbounded_channel();
// use v2 db to avoid reading v1 files that announced from the whole network instead of neighbors let sync_store = if config.neighbors_only {
let sync_store = Arc::new(SyncStore::new_with_name( // use v2 db to avoid reading v1 files that announced from the whole network instead of neighbors
store.clone(), Arc::new(SyncStore::new_with_name(
"pendingv2", store.clone(),
"readyv2", "pendingv2",
)); "readyv2",
))
} else {
Arc::new(SyncStore::new(store.clone()))
};
let catched_up = Arc::new(AtomicBool::new(false)); let catched_up = Arc::new(AtomicBool::new(false));
// handle new file // handle new file
@ -55,15 +59,21 @@ impl AutoSyncManager {
); );
// sync in sequence // sync in sequence
// let serial = let serial = if config.neighbors_only {
// SerialBatcher::new(config, store.clone(), sync_send.clone(), sync_store.clone()) None
// .await?; } else {
// executor.spawn( let serial =
// serial SerialBatcher::new(config, store.clone(), sync_send.clone(), sync_store.clone())
// .clone() .await?;
// .start(file_announcement_recv, log_sync_recv, catched_up.clone()), executor.spawn(
// "auto_sync_serial", serial
// ); .clone()
.start(file_announcement_recv, log_sync_recv, catched_up.clone()),
"auto_sync_serial",
);
Some(serial)
};
// sync randomly // sync randomly
let random = RandomBatcher::new(config, store, sync_send, sync_store); let random = RandomBatcher::new(config, store, sync_send, sync_store);
@ -76,7 +86,7 @@ impl AutoSyncManager {
); );
Ok(Self { Ok(Self {
serial: None, serial,
random, random,
file_announcement_send, file_announcement_send,
new_file_send, new_file_send,