feat(net): 引入kthread，将Inet-网卡处理流程移入内核线程中 (#1130)

kernel/src/driver/net/irq_handle.rs

@@ -1,14 +1,10 @@
 use alloc::sync::Arc;
 use system_error::SystemError;
 
-use crate::{
-    exception::{
-        irqdata::IrqHandlerData,
-        irqdesc::{IrqHandler, IrqReturn},
-        IrqNumber,
-    },
-    // net::net_core::poll_ifaces_try_lock_onetime,
-    net::net_core::poll_ifaces,
+use crate::exception::{
+    irqdata::IrqHandlerData,
+    irqdesc::{IrqHandler, IrqReturn},
+    IrqNumber,
 };
 
 /// 默认的网卡中断处理函数
@@ -22,8 +18,7 @@ impl IrqHandler for DefaultNetIrqHandler {
         _static_data: Option<&dyn IrqHandlerData>,
         _dynamic_data: Option<Arc<dyn IrqHandlerData>>,
     ) -> Result<IrqReturn, SystemError> {
-        // poll_ifaces_try_lock_onetime().ok();
-        poll_ifaces();
+        super::kthread::wakeup_poll_thread();
         Ok(IrqReturn::Handled)
     }
 }

kernel/src/driver/net/kthread.rs

@@ -0,0 +1,47 @@
+use alloc::borrow::ToOwned;
+use alloc::sync::Arc;
+use unified_init::macros::unified_init;
+
+use crate::arch::CurrentIrqArch;
+use crate::exception::InterruptArch;
+use crate::init::initcall::INITCALL_SUBSYS;
+use crate::net::NET_DEVICES;
+use crate::process::kthread::{KernelThreadClosure, KernelThreadMechanism};
+use crate::process::{ProcessControlBlock, ProcessManager};
+use crate::sched::{schedule, SchedMode};
+
+static mut NET_POLL_THREAD: Option<Arc<ProcessControlBlock>> = None;
+
+#[unified_init(INITCALL_SUBSYS)]
+pub fn net_poll_init() -> Result<(), system_error::SystemError> {
+    let closure = KernelThreadClosure::StaticEmptyClosure((&(net_poll_thread as fn() -> i32), ()));
+    let pcb = KernelThreadMechanism::create_and_run(closure, "net_poll".to_owned())
+        .ok_or("")
+        .expect("create net_poll thread failed");
+    log::info!("net_poll thread created");
+    unsafe {
+        NET_POLL_THREAD = Some(pcb);
+    }
+    return Ok(());
+}
+
+fn net_poll_thread() -> i32 {
+    log::info!("net_poll thread started");
+    loop {
+        for (_, iface) in NET_DEVICES.read_irqsave().iter() {
+            iface.poll();
+        }
+        let irq_guard = unsafe { CurrentIrqArch::save_and_disable_irq() };
+        ProcessManager::mark_sleep(true).expect("clocksource_watchdog_kthread:mark sleep failed");
+        drop(irq_guard);
+        schedule(SchedMode::SM_NONE);
+    }
+}
+
+/// 拉起线程
+pub(super) fn wakeup_poll_thread() {
+    if unsafe { NET_POLL_THREAD.is_none() } {
+        return;
+    }
+    let _ = ProcessManager::wakeup(unsafe { NET_POLL_THREAD.as_ref().unwrap() });
+}

kernel/src/driver/net/mod.rs

@@ -14,6 +14,7 @@ pub mod class;
 mod dma;
 pub mod e1000e;
 pub mod irq_handle;
+pub mod kthread;
 pub mod loopback;
 pub mod sysfs;
 pub mod virtio_net;

kernel/src/driver/net/virtio_net.rs

@@ -47,7 +47,7 @@ use crate::{
         rwlock::{RwLockReadGuard, RwLockWriteGuard},
         spinlock::{SpinLock, SpinLockGuard},
     },
-    net::{generate_iface_id, net_core::poll_ifaces, NET_DEVICES},
+    net::{generate_iface_id, NET_DEVICES},
     time::Instant,
 };
 use system_error::SystemError;
@@ -263,8 +263,7 @@ impl Device for VirtIONetDevice {
 
 impl VirtIODevice for VirtIONetDevice {
     fn handle_irq(&self, _irq: IrqNumber) -> Result<IrqReturn, SystemError> {
-        // log::warn!("VirtioInterface: poll_ifaces_try_lock_onetime -> poll_ifaces");
-        poll_ifaces();
+        super::kthread::wakeup_poll_thread();
         return Ok(IrqReturn::Handled);
     }
 

kernel/src/net/net_core.rs

@@ -1,11 +1,8 @@
-use alloc::{collections::BTreeMap, sync::Arc};
-use log::{debug, info, warn};
 use smoltcp::{socket::dhcpv4, wire};
 use system_error::SystemError;
 
 use crate::{
-    driver::net::{Iface, Operstate},
-    libs::rwlock::RwLockReadGuard,
+    driver::net::Operstate,
     net::NET_DEVICES,
     time::{sleep::nanosleep, PosixTimeSpec},
 };
@@ -86,7 +83,7 @@ fn dhcp_query() -> Result<(), SystemError> {
                     if let Some(cidr) = cidr {
                         // 这里先在这里将网卡设置为up，后面等netlink实现了再修改
                         net_face.set_operstate(Operstate::IF_OPER_UP);
-                        info!("Successfully allocated ip by Dhcpv4! Ip:{}", cidr);
+                        log::info!("Successfully allocated ip by Dhcpv4! Ip:{}", cidr);
                         return Ok(());
                     }
                 } else {
@@ -99,7 +96,7 @@ fn dhcp_query() -> Result<(), SystemError> {
             }
 
             Some(dhcpv4::Event::Deconfigured) => {
-                debug!("Dhcp v4 deconfigured");
+                log::debug!("Dhcp v4 deconfigured");
                 net_face
                     .update_ip_addrs(&[smoltcp::wire::IpCidr::Ipv4(wire::Ipv4Cidr::new(
                         wire::Ipv4Address::UNSPECIFIED,
@@ -125,15 +122,3 @@ fn dhcp_query() -> Result<(), SystemError> {
 
     return Err(SystemError::ETIMEDOUT);
 }
-
-pub fn poll_ifaces() {
-    // log::debug!("poll_ifaces");
-    let guard: RwLockReadGuard<BTreeMap<usize, Arc<dyn Iface>>> = NET_DEVICES.read_irqsave();
-    if guard.len() == 0 {
-        warn!("poll_ifaces: No net driver found!");
-        return;
-    }
-    for (_, iface) in guard.iter() {
-        iface.poll();
-    }
-}

kernel/src/net/socket/inet/common/mod.rs

@@ -53,11 +53,11 @@ impl BoundInner {
                 })
                 .expect("No default interface");
 
-            let handle = iface.sockets().lock_irqsave().add(socket);
+            let handle = iface.sockets().lock().add(socket);
             return Ok(Self { handle, iface });
         } else {
             let iface = get_iface_to_bind(address).ok_or(SystemError::ENODEV)?;
-            let handle = iface.sockets().lock_irqsave().add(socket);
+            let handle = iface.sockets().lock().add(socket);
             return Ok(Self { handle, iface });
         }
     }
@@ -72,7 +72,7 @@ impl BoundInner {
     {
         let (iface, address) = get_ephemeral_iface(&remote);
         // let bound_port = iface.port_manager().bind_ephemeral_port(socket_type)?;
-        let handle = iface.sockets().lock_no_preempt().add(socket);
+        let handle = iface.sockets().lock().add(socket);
         // let endpoint = smoltcp::wire::IpEndpoint::new(local_addr, bound_port);
         Ok((Self { handle, iface }, address))
     }

kernel/src/net/socket/inet/datagram/mod.rs

@@ -4,7 +4,6 @@ use system_error::SystemError;
 
 use crate::libs::wait_queue::WaitQueue;
 use crate::net::event_poll::EPollEventType;
-use crate::net::net_core::poll_ifaces;
 use crate::net::socket::{Socket, PMSG};
 use crate::{libs::rwlock::RwLock, net::socket::endpoint::Endpoint};
 use alloc::sync::{Arc, Weak};
@@ -89,7 +88,7 @@ impl UdpSocket {
         match self.inner.read().as_ref().expect("Udp Inner is None") {
             UdpInner::Bound(bound) => {
                 let ret = bound.try_recv(buf);
-                poll_ifaces();
+                bound.inner().iface().poll();
                 ret
             }
             _ => Err(SystemError::ENOTCONN),
@@ -125,10 +124,13 @@ impl UdpSocket {
         };
         // Optimize: 拿两次锁的平均效率是否比一次长时间的读锁效率要高？
         let result = match self.inner.read().as_ref().expect("Udp Inner is None") {
-            UdpInner::Bound(bound) => bound.try_send(buf, to),
+            UdpInner::Bound(bound) => {
+                let ret = bound.try_send(buf, to);
+                bound.inner().iface().poll();
+                ret
+            }
             _ => Err(SystemError::ENOTCONN),
         };
-        poll_ifaces();
         return result;
     }
 

kernel/src/net/socket/inet/stream/inner.rs

@@ -211,7 +211,7 @@ impl Connecting {
     }
 
     pub fn into_result(self) -> (Inner, Result<(), SystemError>) {
-        let result = *self.result.read_irqsave();
+        let result = *self.result.read();
         match result {
             ConnectResult::Connecting => (
                 Inner::Connecting(self),
@@ -245,7 +245,7 @@ impl Connecting {
 
         self.inner
             .with_mut(|socket: &mut smoltcp::socket::tcp::Socket| {
-                let mut result = self.result.write_irqsave();
+                let mut result = self.result.write();
                 if matches!(*result, ConnectResult::Refused | ConnectResult::Connected) {
                     return false; // Already connected or refused
                 }