use std::collections::HashMap;
use std::os::unix::prelude::RawFd;
use std::sync::{Arc, Barrier, Mutex, OnceLock};
use std::{process, thread};
use anyhow::{bail, Result};
use log::{error, info, warn};
use super::config::IothreadConfig;
use crate::machine::IOTHREADS;
use crate::qmp::qmp_schema::IothreadInfo;
use crate::signal_handler::get_signal;
use crate::temp_cleaner::TempCleaner;
use util::loop_context::{
gen_delete_notifiers, get_notifiers_fds, EventLoopContext, EventLoopManager, EventNotifier,
};
pub struct EventLoop {
main_loop: EventLoopContext,
io_threads: HashMap<String, EventLoopContext>,
}
static GLOBAL_EVENT_LOOP: OnceLock<EventLoop> = OnceLock::new();
impl EventLoop {
pub fn object_init(iothreads: &Option<Vec<IothreadConfig>>) -> Result<()> {
let mut io_threads = HashMap::new();
let cnt = match iothreads {
Some(thrs) => thrs.len(),
None => 0,
};
let thread_exit_barrier = Arc::new(Barrier::new(cnt + 1));
if let Some(thrs) = iothreads {
for thr in thrs {
io_threads.insert(
thr.id.clone(),
EventLoopContext::new(thread_exit_barrier.clone()),
);
}
}
let res = GLOBAL_EVENT_LOOP.set(EventLoop {
main_loop: EventLoopContext::new(thread_exit_barrier),
io_threads,
});
if res.is_err() {
warn!("GLOBAL_EVENT_LOOP has been initialized");
return Ok(());
}
let event_loop = GLOBAL_EVENT_LOOP.get().unwrap();
for (id, ctx) in &event_loop.io_threads {
thread::Builder::new().name(id.to_string()).spawn(move || {
let iothread_info = IothreadInfo {
shrink: 0,
pid: process::id(),
grow: 0,
max: 0,
id: id.to_string(),
};
IOTHREADS.lock().unwrap().push(iothread_info);
while ctx.iothread_run().is_ok() {
if TempCleaner::is_cleaned() {
break;
}
}
if let Err(e) = ctx.clean_event_loop() {
error!("Failed to clean event loop {:?}", e);
}
ctx.thread_exit_barrier.wait();
})?;
}
Ok(())
}
pub fn get_ctx(name: Option<&String>) -> Option<&EventLoopContext> {
if let Some(event_loop) = GLOBAL_EVENT_LOOP.get() {
if let Some(name) = name {
return event_loop.io_threads.get(name);
}
return Some(&event_loop.main_loop);
}
panic!("Global Event Loop have not been initialized.");
}
pub fn set_manager(manager: Arc<Mutex<dyn EventLoopManager>>) {
if let Some(event_loop) = GLOBAL_EVENT_LOOP.get() {
event_loop.main_loop.set_manager(manager.clone());
for (_name, io_thread) in event_loop.io_threads.iter() {
io_thread.set_manager(manager.clone());
}
}
}
pub fn update_event(notifiers: Vec<EventNotifier>, name: Option<&String>) -> Result<()> {
if let Some(ctx) = Self::get_ctx(name) {
ctx.update_events(notifiers)
} else {
bail!("Loop Context not found in EventLoop.")
}
}
pub fn loop_run() -> Result<()> {
if let Some(event_loop) = GLOBAL_EVENT_LOOP.get() {
loop {
let sig_num = get_signal();
if sig_num != 0 {
info!("MainLoop exits due to receive signal {}", sig_num);
return Ok(());
}
if !event_loop.main_loop.run()? {
info!("MainLoop exits due to guest internal operation.");
return Ok(());
}
}
} else {
bail!("Global Event Loop have not been initialized.")
}
}
pub fn loop_clean() {
EventLoop::kick_iothreads();
if let Some(event_loop) = GLOBAL_EVENT_LOOP.get() {
event_loop.main_loop.thread_exit_barrier.wait();
}
}
pub fn kick_iothreads() {
if let Some(event_loop) = GLOBAL_EVENT_LOOP.get() {
for (_name, io_thread) in event_loop.io_threads.iter() {
io_thread.kick();
}
}
}
}
pub fn register_event_helper(
notifiers: Vec<EventNotifier>,
ctx_name: Option<&String>,
record_evts: &mut Vec<RawFd>,
) -> Result<()> {
let mut notifiers_fds = get_notifiers_fds(¬ifiers);
EventLoop::update_event(notifiers, ctx_name)?;
record_evts.append(&mut notifiers_fds);
Ok(())
}
pub fn unregister_event_helper(
ctx_name: Option<&String>,
record_evts: &mut Vec<RawFd>,
) -> Result<()> {
EventLoop::update_event(gen_delete_notifiers(record_evts), ctx_name)?;
record_evts.clear();
Ok(())
}