feat: 初步实现Unix stream socket (#916)

* 修复mprotect系统调用未正确设置vm_flags的错误 (#847)

* fix(time): modify update wall time (#836)

更改了时间子系统的update_wall_time函数，通过读取当前周期数，计算delta值进行更新，而不是通过传入delta值进行更新

* chore: 调整triagebot.toml以适应新的组织架构 (#848)

* doc: 完善README.md (#849)

* doc: 完善README.md

* chore: 更新sphinx相关配置，适应read the docs的更新 (#850)

根据read the docs在7月15日blog,进行此修改

https://about.readthedocs.com/blog/2024/07/addons-by-default/

* feat(driver/net): 实现Loopback网卡接口 (#845)

* 初步实现loopback设备

* fix: build-scripts和tools目录下的make check指定工具链版本 (#861)

* fix: tcp poll没有正确处理posix socket的listen状态的问题 (#859)

* chore: 将工具链更新到2024-07-23 (#864)

* chore: 将工具链更新到2024-07-23

* feat(fs): add eventfd syscall support (#858)

* feat(fs): add eventfd syscall support

* refactor: 删除过时的va-pa转换函数，改为统一使用MMArch (#862)

* 默认nightly-2024-07-23 & config改为config.toml (#872)

* fix: 修复由于升级到2024-07-23工具链之后，某些机器上面内核运行一直fault的问题。 (#870)

* fix: 修复由于升级到2024-07-23工具链之后，某些机器上面内核运行一直fault的问题。

* feat(cred): 初步实现Cred (#846)

* 初步实现Cred

* 添加seteuid和setegid

* 添加cred测试程序

* 修改Cred::fscmp返回结果为CredFsCmp枚举

* 完善root用户相关信息

* fix: 修复键盘码解析器没能正确处理类似ctrl C的控制字符的问题 (#877)

* fix: 修复键盘码解析器没能正确处理类似ctrl C的控制字符的问题

* fix: 解决ntty潜在的panic问题

* ci: enable ci workflow on branches other than master (#891)

* 修复unlink、unlinkat系统调用的路径错误 (#892)

* fix: socket shutdown wrong implement (#893)

* feat: 增加tokio异步运行时支持 (#894)

* fix the EventFdFlags error

* feat: support tokio (Single thread version)

Fix deadlock issue on closing file.
Add function for PipeInode and EventFdInode.

* feat: 实现unix stream sock和其状态

* 握手支持

* feat：建立连接功能实现

* feat:unix stream socket 初版

* fix: 消除标红

* feat: 阻塞式读写buffer

* feat: 实现unix socket buffer

* 不小心改了inode

---------

Co-authored-by: MemoryShore <[email protected]>
Co-authored-by: 黄铭涛 <[email protected]>
Co-authored-by: LoGin <[email protected]>
Co-authored-by: linfeng <[email protected]>
Co-authored-by: Jomo <[email protected]>
Co-authored-by: Chiichen <[email protected]>
Co-authored-by: Samuel Dai <[email protected]>

Author: 8 people

.vscode/settings.json

@@ -161,4 +161,5 @@
     ],
     "rust-analyzer.showUnlinkedFileNotification": false,
     "editor.accessibilityPageSize": 15,
+    "makefile.configureOnOpen": false,
 }

kernel/src/net/socket/base.rs

@@ -3,7 +3,6 @@
 use core::fmt::Debug;
 use alloc::sync::Arc;
 use system_error::SystemError::{self, *};
-use crate::filesystem::vfs::IndexNode;
 use crate::net::socket::*;
 use crate::net::syscall_util::MsgHdr;
 
@@ -27,7 +26,7 @@ pub trait Socket: Sync + Send + Debug {
     /// 接受连接，仅用于listening stream socket
     /// ## Block
     /// 如果没有连接到来，会阻塞
-    fn accept(&self) -> Result<(Arc<dyn IndexNode>, Endpoint), SystemError> {
+    fn accept(&self) -> Result<(Arc<Inode>, Endpoint), SystemError> {
         Err(ENOSYS)
     }
 

kernel/src/net/socket/buffer.rs

@@ -0,0 +1,70 @@
+use alloc::vec::Vec;
+
+use alloc::sync::Arc;
+use system_error::SystemError;
+
+use crate::libs::spinlock::SpinLock;
+
+#[derive(Debug)]
+pub struct Buffer {
+    metadata: Metadata,
+    read_buffer: SpinLock<Vec<u8>>,
+    write_buffer: SpinLock<Vec<u8>>,
+}
+
+impl Buffer {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            metadata: Metadata::default(),
+            read_buffer: SpinLock::new(Vec::new()),
+            write_buffer: SpinLock::new(Vec::new())
+        })
+    }
+
+    pub fn is_read_buf_empty(&self) -> bool {
+        return self.read_buffer.lock().is_empty();
+    }
+
+    pub fn is_write_buf_empty(&self) -> bool {
+        return self.write_buffer.lock().is_empty();
+    }
+
+    pub fn read_read_buffer(&self, buf: &mut [u8]) -> Result<usize, SystemError> {
+        let mut read_buffer = self.read_buffer.lock_irqsave();
+        let len = core::cmp::min(buf.len(), read_buffer.len());
+        buf[..len].copy_from_slice(&read_buffer[..len]);
+        read_buffer.split_off(len);
+        return Ok(len);
+    }
+
+    pub fn write_read_buffer(&self, buf: &[u8]) -> Result<usize, SystemError> {
+        let mut buffer = self.read_buffer.lock_irqsave();
+
+        let len = buf.len();
+        if self.metadata.buf_size - buffer.len() < len {
+            return Err(SystemError::ENOBUFS);
+        }
+        buffer.extend_from_slice(buf);
+
+        Ok(len)
+    }
+
+}
+
+#[derive(Debug)]
+pub struct Metadata {
+    /// 默认的元数据缓冲区大小
+    metadata_buf_size: usize,
+    /// 默认的缓冲区大小
+    buf_size: usize,
+}
+
+impl Default for Metadata {
+    fn default() -> Self {
+        Self { 
+            metadata_buf_size: 1024, 
+            buf_size: 64 * 1024, 
+        }
+    }
+}
+

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

@@ -329,9 +329,9 @@ impl Socket for TcpSocket {
         }
     }
 
-    fn accept(&self) -> Result<(Arc<dyn IndexNode>, crate::net::Endpoint), SystemError> {
+    fn accept(&self) -> Result<(Arc<Inode>, crate::net::socket::Endpoint), SystemError> {
         self.try_accept().map(|(stream, remote)| 
-            (stream as Arc<dyn IndexNode>, crate::net::Endpoint::from(remote))
+            (stream as Arc<Inode>, crate::net::socket::Endpoint::from(remote))
         )
     }
 }
@@ -376,7 +376,7 @@ impl TcpStream {
     }
 }
 
-use crate::net::socket::Socket;
+use crate::net::socket::{Inode, Socket};
 use crate::filesystem::vfs::IndexNode;
 
 impl IndexNode for TcpStream {

kernel/src/net/socket/inode.rs

@@ -71,6 +71,10 @@ impl Socket for Inode {
 }
 
 impl Inode {
+    pub fn new(socket: Arc<dyn Socket>) -> Arc<Self>{
+        return Arc::new(Self{inner: socket.clone()});
+    }
+
     pub fn set_nonblock(&self, nonblock: bool) {
         todo!()
     }

kernel/src/net/socket/mod.rs

@@ -7,6 +7,7 @@ mod inode;
 mod family;
 mod utils;
 mod base;
+mod buffer;
 mod endpoint;
 
 pub use define::*;

kernel/src/net/socket/unix/mod.rs

@@ -1,235 +1 @@
-use core::any::Any;
-
-use alloc::{boxed::Box, sync::Arc, vec::Vec};
-use system_error::SystemError;
-
-use crate::{libs::spinlock::SpinLock, net::Endpoint};
-
-use super::{
-    handle::GlobalSocketHandle, PosixSocketHandleItem, Socket, inode::Inode, SocketMetadata,
-    Options, InetSocketType,
-};
-
-#[derive(Debug, Clone)]
-pub struct StreamSocket {
-    metadata: SocketMetadata,
-    buffer: Arc<SpinLock<Vec<u8>>>,
-    peer_inode: Option<Arc<Inode>>,
-    handle: GlobalSocketHandle,
-    posix_item: Arc<PosixSocketHandleItem>,
-}
-
-impl StreamSocket {
-    /// 默认的元数据缓冲区大小
-    pub const DEFAULT_METADATA_BUF_SIZE: usize = 1024;
-    /// 默认的缓冲区大小
-    pub const DEFAULT_BUF_SIZE: usize = 64 * 1024;
-
-    /// # 创建一个 Stream Socket
-    ///
-    /// ## 参数
-    /// - `options`: socket选项
-    pub fn new(options: Options) -> Self {
-        let buffer = Arc::new(SpinLock::new(Vec::with_capacity(Self::DEFAULT_BUF_SIZE)));
-
-        let metadata = SocketMetadata::new(
-            InetSocketType::Unix,
-            Self::DEFAULT_BUF_SIZE,
-            Self::DEFAULT_BUF_SIZE,
-            Self::DEFAULT_METADATA_BUF_SIZE,
-            options,
-        );
-
-        let posix_item = Arc::new(PosixSocketHandleItem::new(None));
-
-        Self {
-            metadata,
-            buffer,
-            peer_inode: None,
-            handle: GlobalSocketHandle::new_kernel_handle(),
-            posix_item,
-        }
-    }
-}
-
-impl Socket for StreamSocket {
-    fn socket_handle(&self) -> GlobalSocketHandle {
-        self.handle
-    }
-
-    fn close(&mut self) {}
-
-    fn read(&self, buf: &mut [u8]) -> (Result<usize, SystemError>, Endpoint) {
-        let mut buffer = self.buffer.lock_irqsave();
-
-        let len = core::cmp::min(buf.len(), buffer.len());
-        buf[..len].copy_from_slice(&buffer[..len]);
-
-        let _ = buffer.split_off(len);
-
-        (Ok(len), Endpoint::Inode(self.peer_inode.clone()))
-    }
-
-    fn write(&self, buf: &[u8], _to: Option<Endpoint>) -> Result<usize, SystemError> {
-        if self.peer_inode.is_none() {
-            return Err(SystemError::ENOTCONN);
-        }
-
-        let peer_inode = self.peer_inode.clone().unwrap();
-        let len = peer_inode.inner().write_buffer(buf)?;
-        Ok(len)
-    }
-
-    fn connect(&mut self, endpoint: Endpoint) -> Result<(), SystemError> {
-        if self.peer_inode.is_some() {
-            return Err(SystemError::EISCONN);
-        }
-
-        if let Endpoint::Inode(inode) = endpoint {
-            self.peer_inode = inode;
-            Ok(())
-        } else {
-            Err(SystemError::EINVAL)
-        }
-    }
-
-    fn write_buffer(&self, buf: &[u8]) -> Result<usize, SystemError> {
-        let mut buffer = self.buffer.lock_irqsave();
-
-        let len = buf.len();
-        if buffer.capacity() - buffer.len() < len {
-            return Err(SystemError::ENOBUFS);
-        }
-        buffer.extend_from_slice(buf);
-
-        Ok(len)
-    }
-
-    fn metadata(&self) -> SocketMetadata {
-        self.metadata.clone()
-    }
-
-    fn box_clone(&self) -> Box<dyn Socket> {
-        Box::new(self.clone())
-    }
-
-    fn as_any_ref(&self) -> &dyn core::any::Any {
-        self
-    }
-
-    fn as_any_mut(&mut self) -> &mut dyn core::any::Any {
-        self
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct SeqpacketSocket {
-    metadata: SocketMetadata,
-    buffer: Arc<SpinLock<Vec<u8>>>,
-    peer_inode: Option<Arc<Inode>>,
-    handle: GlobalSocketHandle,
-    posix_item: Arc<PosixSocketHandleItem>,
-}
-
-impl SeqpacketSocket {
-    /// 默认的元数据缓冲区大小
-    pub const DEFAULT_METADATA_BUF_SIZE: usize = 1024;
-    /// 默认的缓冲区大小
-    pub const DEFAULT_BUF_SIZE: usize = 64 * 1024;
-
-    /// # 创建一个 Seqpacket Socket
-    ///
-    /// ## 参数
-    /// - `options`: socket选项
-    pub fn new(options: Options) -> Self {
-        let buffer = Arc::new(SpinLock::new(Vec::with_capacity(Self::DEFAULT_BUF_SIZE)));
-
-        let metadata = SocketMetadata::new(
-            InetSocketType::Unix,
-            Self::DEFAULT_BUF_SIZE,
-            Self::DEFAULT_BUF_SIZE,
-            Self::DEFAULT_METADATA_BUF_SIZE,
-            options,
-        );
-
-        let posix_item = Arc::new(PosixSocketHandleItem::new(None));
-
-        Self {
-            metadata,
-            buffer,
-            peer_inode: None,
-            handle: GlobalSocketHandle::new_kernel_handle(),
-            posix_item,
-        }
-    }
-}
-
-impl Socket for SeqpacketSocket {
-    fn close(&mut self) {}
-
-    fn read(&self, buf: &mut [u8]) -> (Result<usize, SystemError>, Endpoint) {
-        let mut buffer = self.buffer.lock_irqsave();
-
-        let len = core::cmp::min(buf.len(), buffer.len());
-        buf[..len].copy_from_slice(&buffer[..len]);
-
-        let _ = buffer.split_off(len);
-
-        (Ok(len), Endpoint::Inode(self.peer_inode.clone()))
-    }
-
-    fn write(&self, buf: &[u8], _to: Option<Endpoint>) -> Result<usize, SystemError> {
-        if self.peer_inode.is_none() {
-            return Err(SystemError::ENOTCONN);
-        }
-
-        let peer_inode = self.peer_inode.clone().unwrap();
-        let len = peer_inode.inner().write_buffer(buf)?;
-        Ok(len)
-    }
-
-    fn connect(&mut self, endpoint: Endpoint) -> Result<(), SystemError> {
-        if self.peer_inode.is_some() {
-            return Err(SystemError::EISCONN);
-        }
-
-        if let Endpoint::Inode(inode) = endpoint {
-            self.peer_inode = inode;
-            Ok(())
-        } else {
-            Err(SystemError::EINVAL)
-        }
-    }
-
-    fn write_buffer(&self, buf: &[u8]) -> Result<usize, SystemError> {
-        let mut buffer = self.buffer.lock_irqsave();
-
-        let len = buf.len();
-        if buffer.capacity() - buffer.len() < len {
-            return Err(SystemError::ENOBUFS);
-        }
-        buffer.extend_from_slice(buf);
-
-        Ok(len)
-    }
-
-    fn socket_handle(&self) -> GlobalSocketHandle {
-        self.handle
-    }
-
-    fn metadata(&self) -> SocketMetadata {
-        self.metadata.clone()
-    }
-
-    fn box_clone(&self) -> Box<dyn Socket> {
-        Box::new(self.clone())
-    }
-
-    fn as_any_ref(&self) -> &dyn core::any::Any {
-        self
-    }
-
-    fn as_any_mut(&mut self) -> &mut dyn core::any::Any {
-        self
-    }
-}
+mod stream;