Fix the centralize disconnection bug. (#1896)
* fixing the first bug * backtick
This commit is contained in:
parent
d96a1426d5
commit
95bed5b016
@ -290,7 +290,7 @@ where
|
|||||||
) -> Result<(), Error> {
|
) -> Result<(), Error> {
|
||||||
if !self.is_main {
|
if !self.is_main {
|
||||||
// secondary node
|
// secondary node
|
||||||
let is_nt = match &mut event {
|
let is_nt_or_heartbeat = match &mut event {
|
||||||
Event::NewTestcase {
|
Event::NewTestcase {
|
||||||
input: _,
|
input: _,
|
||||||
client_config: _,
|
client_config: _,
|
||||||
@ -304,9 +304,14 @@ where
|
|||||||
*forward_id = Some(ClientId(self.inner.mgr_id().0 as u32));
|
*forward_id = Some(ClientId(self.inner.mgr_id().0 as u32));
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
|
Event::UpdateExecStats {
|
||||||
|
time: _,
|
||||||
|
executions: _,
|
||||||
|
phantom: _,
|
||||||
|
} => true,
|
||||||
_ => false,
|
_ => false,
|
||||||
};
|
};
|
||||||
if is_nt {
|
if is_nt_or_heartbeat {
|
||||||
return self.forward_to_main(&event);
|
return self.forward_to_main(&event);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -533,10 +533,12 @@ where
|
|||||||
self.shmem_provider.post_fork(false)?;
|
self.shmem_provider.post_fork(false)?;
|
||||||
handles.push(child.pid);
|
handles.push(child.pid);
|
||||||
#[cfg(feature = "std")]
|
#[cfg(feature = "std")]
|
||||||
log::info!("centralized broker spawned");
|
log::info!("PID: {:#?} centralized broker spawned", std::process::id());
|
||||||
}
|
}
|
||||||
ForkResult::Child => {
|
ForkResult::Child => {
|
||||||
log::info!("{:?} PostFork", unsafe { libc::getpid() });
|
log::info!("{:?} PostFork", unsafe { libc::getpid() });
|
||||||
|
#[cfg(feature = "std")]
|
||||||
|
log::info!("PID: {:#?} I am centralized broker", std::process::id());
|
||||||
self.shmem_provider.post_fork(true)?;
|
self.shmem_provider.post_fork(true)?;
|
||||||
|
|
||||||
let mut broker: CentralizedLlmpEventBroker<S::Input, SP> =
|
let mut broker: CentralizedLlmpEventBroker<S::Input, SP> =
|
||||||
|
@ -2272,7 +2272,7 @@ where
|
|||||||
{
|
{
|
||||||
self.clients_to_remove.push(i);
|
self.clients_to_remove.push(i);
|
||||||
#[cfg(feature = "llmp_debug")]
|
#[cfg(feature = "llmp_debug")]
|
||||||
println!("Client #{i} timed out. Removing.");
|
log::info!("Client #{:#?} timed out. Removing.", client_id);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
new_messages = has_messages;
|
new_messages = has_messages;
|
||||||
@ -2284,7 +2284,8 @@ where
|
|||||||
|
|
||||||
// After brokering, remove all clients we don't want to keep.
|
// After brokering, remove all clients we don't want to keep.
|
||||||
for i in self.clients_to_remove.iter().rev() {
|
for i in self.clients_to_remove.iter().rev() {
|
||||||
log::debug!("Client #{i} disconnected.");
|
let client_id = self.llmp_clients[*i].id;
|
||||||
|
log::info!("Client #{:#?} disconnected.", client_id);
|
||||||
self.llmp_clients.remove(*i);
|
self.llmp_clients.remove(*i);
|
||||||
}
|
}
|
||||||
self.clients_to_remove.clear();
|
self.clients_to_remove.clear();
|
||||||
|
Loading…
x
Reference in New Issue
Block a user