Rename crates from jinux-* to aster-*

2025-06-28 11:53:24 +00:00 · 2023-12-25 03:12:25 +00:00
parent 6dbf5d560d
commit 93781df27b
460 changed files with 596 additions and 595 deletions
--- a/services/libs/aster-std/src/net/socket/ip/always_some.rs
+++ b/services/libs/aster-std/src/net/socket/ip/always_some.rs
@ -0,0 +1,51 @@
+use crate::prelude::*;
+use core::ops::{Deref, DerefMut};
+
+/// AlwaysSome is a wrapper for Option.
+///
+/// AlwaysSome should always be Some(T), so we can treat it as a smart pointer.
+/// If it becomes None, the AlwaysSome should be viewed invalid and cannot be used anymore.
+pub struct AlwaysSome<T>(Option<T>);
+
+impl<T> AlwaysSome<T> {
+    pub fn new(value: T) -> Self {
+        AlwaysSome(Some(value))
+    }
+
+    pub fn try_take_with<R, E: Into<Error>, F: FnOnce(T) -> core::result::Result<R, (E, T)>>(
+        &mut self,
+        f: F,
+    ) -> Result<R> {
+        let value = if let Some(value) = self.0.take() {
+            value
+        } else {
+            return_errno_with_message!(Errno::EINVAL, "the take cell is none");
+        };
+        match f(value) {
+            Ok(res) => Ok(res),
+            Err((err, t)) => {
+                self.0 = Some(t);
+                Err(err.into())
+            }
+        }
+    }
+
+    /// Takes inner value
+    pub fn take(&mut self) -> T {
+        debug_assert!(self.0.is_some());
+        self.0.take().unwrap()
+    }
+}
+
+impl<T> Deref for AlwaysSome<T> {
+    type Target = T;
+    fn deref(&self) -> &Self::Target {
+        self.0.as_ref().unwrap()
+    }
+}
+
+impl<T> DerefMut for AlwaysSome<T> {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.0.as_mut().unwrap()
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/common.rs
+++ b/services/libs/aster-std/src/net/socket/ip/common.rs
@ -0,0 +1,65 @@
+use crate::net::iface::BindPortConfig;
+use crate::net::iface::Iface;
+use crate::net::iface::{AnyBoundSocket, AnyUnboundSocket};
+use crate::net::iface::{IpAddress, IpEndpoint};
+use crate::net::IFACES;
+use crate::prelude::*;
+
+pub fn get_iface_to_bind(ip_addr: &IpAddress) -> Option<Arc<dyn Iface>> {
+    let ifaces = IFACES.get().unwrap();
+    let IpAddress::Ipv4(ipv4_addr) = ip_addr;
+    ifaces
+        .iter()
+        .find(|iface| {
+            if let Some(iface_ipv4_addr) = iface.ipv4_addr() {
+                iface_ipv4_addr == *ipv4_addr
+            } else {
+                false
+            }
+        })
+        .map(Clone::clone)
+}
+
+/// Get a suitable iface to deal with sendto/connect request if the socket is not bound to an iface.
+/// If the remote address is the same as that of some iface, we will use the iface.
+/// Otherwise, we will use a default interface.
+fn get_ephemeral_iface(remote_ip_addr: &IpAddress) -> Arc<dyn Iface> {
+    let ifaces = IFACES.get().unwrap();
+    let IpAddress::Ipv4(remote_ipv4_addr) = remote_ip_addr;
+    if let Some(iface) = ifaces.iter().find(|iface| {
+        if let Some(iface_ipv4_addr) = iface.ipv4_addr() {
+            iface_ipv4_addr == *remote_ipv4_addr
+        } else {
+            false
+        }
+    }) {
+        return iface.clone();
+    }
+    // FIXME: use the virtio-net as the default interface
+    ifaces[0].clone()
+}
+
+pub(super) fn bind_socket(
+    unbound_socket: Box<AnyUnboundSocket>,
+    endpoint: IpEndpoint,
+    can_reuse: bool,
+) -> core::result::Result<Arc<AnyBoundSocket>, (Error, Box<AnyUnboundSocket>)> {
+    let iface = match get_iface_to_bind(&endpoint.addr) {
+        Some(iface) => iface,
+        None => {
+            let err = Error::with_message(Errno::EADDRNOTAVAIL, "Request iface is not available");
+            return Err((err, unbound_socket));
+        }
+    };
+    let bind_port_config = match BindPortConfig::new(endpoint.port, can_reuse) {
+        Ok(config) => config,
+        Err(e) => return Err((e, unbound_socket)),
+    };
+    iface.bind_socket(unbound_socket, bind_port_config)
+}
+
+pub fn get_ephemeral_endpoint(remote_endpoint: &IpEndpoint) -> IpEndpoint {
+    let iface = get_ephemeral_iface(&remote_endpoint.addr);
+    let ip_addr = iface.ipv4_addr().unwrap();
+    IpEndpoint::new(IpAddress::Ipv4(ip_addr), 0)
+}
--- a/services/libs/aster-std/src/net/socket/ip/datagram/bound.rs
+++ b/services/libs/aster-std/src/net/socket/ip/datagram/bound.rs
@ -0,0 +1,108 @@
+use crate::events::{IoEvents, Observer};
+use crate::net::iface::IpEndpoint;
+
+use crate::net::poll_ifaces;
+use crate::process::signal::{Pollee, Poller};
+use crate::{
+    net::{
+        iface::{AnyBoundSocket, RawUdpSocket},
+        socket::util::send_recv_flags::SendRecvFlags,
+    },
+    prelude::*,
+};
+
+pub struct BoundDatagram {
+    bound_socket: Arc<AnyBoundSocket>,
+    remote_endpoint: RwLock<Option<IpEndpoint>>,
+    pollee: Pollee,
+}
+
+impl BoundDatagram {
+    pub fn new(bound_socket: Arc<AnyBoundSocket>, pollee: Pollee) -> Arc<Self> {
+        let bound = Arc::new(Self {
+            bound_socket,
+            remote_endpoint: RwLock::new(None),
+            pollee,
+        });
+        bound.bound_socket.set_observer(Arc::downgrade(&bound) as _);
+        bound
+    }
+
+    pub fn remote_endpoint(&self) -> Result<IpEndpoint> {
+        self.remote_endpoint
+            .read()
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "remote endpoint is not specified"))
+    }
+
+    pub fn set_remote_endpoint(&self, endpoint: IpEndpoint) {
+        *self.remote_endpoint.write() = Some(endpoint);
+    }
+
+    pub fn local_endpoint(&self) -> Result<IpEndpoint> {
+        self.bound_socket.local_endpoint().ok_or_else(|| {
+            Error::with_message(Errno::EINVAL, "socket does not bind to local endpoint")
+        })
+    }
+
+    pub fn try_recvfrom(
+        &self,
+        buf: &mut [u8],
+        flags: &SendRecvFlags,
+    ) -> Result<(usize, IpEndpoint)> {
+        poll_ifaces();
+        let recv_slice = |socket: &mut RawUdpSocket| {
+            socket
+                .recv_slice(buf)
+                .map_err(|_| Error::with_message(Errno::EAGAIN, "recv buf is empty"))
+        };
+        self.bound_socket.raw_with(recv_slice)
+    }
+
+    pub fn try_sendto(
+        &self,
+        buf: &[u8],
+        remote: Option<IpEndpoint>,
+        flags: SendRecvFlags,
+    ) -> Result<usize> {
+        let remote_endpoint = remote
+            .or_else(|| self.remote_endpoint().ok())
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "udp should provide remote addr"))?;
+        let send_slice = |socket: &mut RawUdpSocket| {
+            socket
+                .send_slice(buf, remote_endpoint)
+                .map(|_| buf.len())
+                .map_err(|_| Error::with_message(Errno::EAGAIN, "send udp packet fails"))
+        };
+        let len = self.bound_socket.raw_with(send_slice)?;
+        poll_ifaces();
+        Ok(len)
+    }
+
+    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    fn update_io_events(&self) {
+        self.bound_socket.raw_with(|socket: &mut RawUdpSocket| {
+            let pollee = &self.pollee;
+
+            if socket.can_recv() {
+                pollee.add_events(IoEvents::IN);
+            } else {
+                pollee.del_events(IoEvents::IN);
+            }
+
+            if socket.can_send() {
+                pollee.add_events(IoEvents::OUT);
+            } else {
+                pollee.del_events(IoEvents::OUT);
+            }
+        });
+    }
+}
+
+impl Observer<()> for BoundDatagram {
+    fn on_events(&self, _: &()) {
+        self.update_io_events();
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/datagram/mod.rs
+++ b/services/libs/aster-std/src/net/socket/ip/datagram/mod.rs
@ -0,0 +1,213 @@
+use core::sync::atomic::{AtomicBool, Ordering};
+
+use crate::events::IoEvents;
+use crate::fs::utils::StatusFlags;
+use crate::net::iface::IpEndpoint;
+
+use crate::process::signal::Poller;
+use crate::{
+    fs::file_handle::FileLike,
+    net::socket::{
+        util::{send_recv_flags::SendRecvFlags, sockaddr::SocketAddr},
+        Socket,
+    },
+    prelude::*,
+};
+
+use self::bound::BoundDatagram;
+use self::unbound::UnboundDatagram;
+
+use super::always_some::AlwaysSome;
+use super::common::get_ephemeral_endpoint;
+
+mod bound;
+mod unbound;
+
+pub struct DatagramSocket {
+    nonblocking: AtomicBool,
+    inner: RwLock<Inner>,
+}
+
+enum Inner {
+    Unbound(AlwaysSome<UnboundDatagram>),
+    Bound(Arc<BoundDatagram>),
+}
+
+impl Inner {
+    fn is_bound(&self) -> bool {
+        matches!(self, Inner::Bound { .. })
+    }
+
+    fn bind(&mut self, endpoint: IpEndpoint) -> Result<Arc<BoundDatagram>> {
+        let unbound = match self {
+            Inner::Unbound(unbound) => unbound,
+            Inner::Bound(..) => return_errno_with_message!(
+                Errno::EINVAL,
+                "the socket is already bound to an address"
+            ),
+        };
+        let bound = unbound.try_take_with(|unbound| unbound.bind(endpoint))?;
+        *self = Inner::Bound(bound.clone());
+        Ok(bound)
+    }
+
+    fn bind_to_ephemeral_endpoint(
+        &mut self,
+        remote_endpoint: &IpEndpoint,
+    ) -> Result<Arc<BoundDatagram>> {
+        let endpoint = get_ephemeral_endpoint(remote_endpoint);
+        self.bind(endpoint)
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        match self {
+            Inner::Unbound(unbound) => unbound.poll(mask, poller),
+            Inner::Bound(bound) => bound.poll(mask, poller),
+        }
+    }
+}
+
+impl DatagramSocket {
+    pub fn new(nonblocking: bool) -> Self {
+        let unbound = UnboundDatagram::new();
+        Self {
+            inner: RwLock::new(Inner::Unbound(AlwaysSome::new(unbound))),
+            nonblocking: AtomicBool::new(nonblocking),
+        }
+    }
+
+    pub fn is_bound(&self) -> bool {
+        self.inner.read().is_bound()
+    }
+
+    pub fn is_nonblocking(&self) -> bool {
+        self.nonblocking.load(Ordering::SeqCst)
+    }
+
+    pub fn set_nonblocking(&self, nonblocking: bool) {
+        self.nonblocking.store(nonblocking, Ordering::SeqCst);
+    }
+
+    fn bound(&self) -> Result<Arc<BoundDatagram>> {
+        if let Inner::Bound(bound) = &*self.inner.read() {
+            Ok(bound.clone())
+        } else {
+            return_errno_with_message!(Errno::EINVAL, "socket does not bind to local endpoint")
+        }
+    }
+
+    fn try_bind_empheral(&self, remote_endpoint: &IpEndpoint) -> Result<Arc<BoundDatagram>> {
+        // Fast path
+        if let Inner::Bound(bound) = &*self.inner.read() {
+            return Ok(bound.clone());
+        }
+
+        // Slow path
+        let mut inner = self.inner.write();
+        if let Inner::Bound(bound) = &*inner {
+            return Ok(bound.clone());
+        }
+        inner.bind_to_ephemeral_endpoint(remote_endpoint)
+    }
+}
+
+impl FileLike for DatagramSocket {
+    fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        // FIXME: respect flags
+        let flags = SendRecvFlags::empty();
+        let (recv_len, _) = self.recvfrom(buf, flags)?;
+        Ok(recv_len)
+    }
+
+    fn write(&self, buf: &[u8]) -> Result<usize> {
+        // FIXME: set correct flags
+        let flags = SendRecvFlags::empty();
+        self.sendto(buf, None, flags)
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.inner.read().poll(mask, poller)
+    }
+
+    fn as_socket(&self) -> Option<&dyn Socket> {
+        Some(self)
+    }
+
+    fn status_flags(&self) -> StatusFlags {
+        if self.is_nonblocking() {
+            StatusFlags::O_NONBLOCK
+        } else {
+            StatusFlags::empty()
+        }
+    }
+
+    fn set_status_flags(&self, new_flags: StatusFlags) -> Result<()> {
+        if new_flags.contains(StatusFlags::O_NONBLOCK) {
+            self.set_nonblocking(true);
+        } else {
+            self.set_nonblocking(false);
+        }
+        Ok(())
+    }
+}
+
+impl Socket for DatagramSocket {
+    fn bind(&self, sockaddr: SocketAddr) -> Result<()> {
+        let endpoint = sockaddr.try_into()?;
+        self.inner.write().bind(endpoint)?;
+        Ok(())
+    }
+
+    fn connect(&self, sockaddr: SocketAddr) -> Result<()> {
+        let endpoint = sockaddr.try_into()?;
+        let bound = self.try_bind_empheral(&endpoint)?;
+        bound.set_remote_endpoint(endpoint);
+        Ok(())
+    }
+
+    fn addr(&self) -> Result<SocketAddr> {
+        self.bound()?.local_endpoint()?.try_into()
+    }
+
+    fn peer_addr(&self) -> Result<SocketAddr> {
+        self.bound()?.remote_endpoint()?.try_into()
+    }
+
+    // FIXME: respect RecvFromFlags
+    fn recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<(usize, SocketAddr)> {
+        debug_assert!(flags.is_all_supported());
+        let bound = self.bound()?;
+        let poller = Poller::new();
+        loop {
+            if let Ok((recv_len, remote_endpoint)) = bound.try_recvfrom(buf, &flags) {
+                let remote_addr = remote_endpoint.try_into()?;
+                return Ok((recv_len, remote_addr));
+            }
+            let events = bound.poll(IoEvents::IN, Some(&poller));
+            if !events.contains(IoEvents::IN) {
+                if self.is_nonblocking() {
+                    return_errno_with_message!(Errno::EAGAIN, "try to receive again");
+                }
+                // FIXME: deal with recvfrom timeout
+                poller.wait()?;
+            }
+        }
+    }
+
+    fn sendto(
+        &self,
+        buf: &[u8],
+        remote: Option<SocketAddr>,
+        flags: SendRecvFlags,
+    ) -> Result<usize> {
+        debug_assert!(flags.is_all_supported());
+        let (bound, remote_endpoint) = if let Some(addr) = remote {
+            let endpoint = addr.try_into()?;
+            (self.try_bind_empheral(&endpoint)?, Some(endpoint))
+        } else {
+            let bound = self.bound()?;
+            (bound, None)
+        };
+        bound.try_sendto(buf, remote_endpoint, flags)
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/datagram/unbound.rs
+++ b/services/libs/aster-std/src/net/socket/ip/datagram/unbound.rs
@ -0,0 +1,52 @@
+use crate::events::IoEvents;
+use crate::net::iface::IpEndpoint;
+
+use crate::net::socket::ip::common::bind_socket;
+use crate::process::signal::{Pollee, Poller};
+use crate::{
+    net::iface::{AnyUnboundSocket, RawUdpSocket},
+    prelude::*,
+};
+
+use super::bound::BoundDatagram;
+
+pub struct UnboundDatagram {
+    unbound_socket: Box<AnyUnboundSocket>,
+    pollee: Pollee,
+}
+
+impl UnboundDatagram {
+    pub fn new() -> Self {
+        Self {
+            unbound_socket: Box::new(AnyUnboundSocket::new_udp()),
+            pollee: Pollee::new(IoEvents::empty()),
+        }
+    }
+
+    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    pub fn bind(
+        self,
+        endpoint: IpEndpoint,
+    ) -> core::result::Result<Arc<BoundDatagram>, (Error, Self)> {
+        let bound_socket = match bind_socket(self.unbound_socket, endpoint, false) {
+            Ok(bound_socket) => bound_socket,
+            Err((err, unbound_socket)) => {
+                return Err((
+                    err,
+                    Self {
+                        unbound_socket,
+                        pollee: self.pollee,
+                    },
+                ))
+            }
+        };
+        let bound_endpoint = bound_socket.local_endpoint().unwrap();
+        bound_socket.raw_with(|socket: &mut RawUdpSocket| {
+            socket.bind(bound_endpoint).unwrap();
+        });
+        Ok(BoundDatagram::new(bound_socket, self.pollee))
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/mod.rs
+++ b/services/libs/aster-std/src/net/socket/ip/mod.rs
@ -0,0 +1,7 @@
+mod always_some;
+mod common;
+mod datagram;
+mod stream;
+
+pub use datagram::DatagramSocket;
+pub use stream::StreamSocket;
--- a/services/libs/aster-std/src/net/socket/ip/stream/connected.rs
+++ b/services/libs/aster-std/src/net/socket/ip/stream/connected.rs
@ -0,0 +1,169 @@
+use core::sync::atomic::{AtomicBool, Ordering};
+
+use crate::events::{IoEvents, Observer};
+use crate::net::iface::IpEndpoint;
+use crate::process::signal::{Pollee, Poller};
+use crate::{
+    net::{
+        iface::{AnyBoundSocket, RawTcpSocket},
+        poll_ifaces,
+        socket::util::{send_recv_flags::SendRecvFlags, shutdown_cmd::SockShutdownCmd},
+    },
+    prelude::*,
+};
+
+pub struct ConnectedStream {
+    nonblocking: AtomicBool,
+    bound_socket: Arc<AnyBoundSocket>,
+    remote_endpoint: IpEndpoint,
+    pollee: Pollee,
+}
+
+impl ConnectedStream {
+    pub fn new(
+        is_nonblocking: bool,
+        bound_socket: Arc<AnyBoundSocket>,
+        remote_endpoint: IpEndpoint,
+        pollee: Pollee,
+    ) -> Arc<Self> {
+        let connected = Arc::new(Self {
+            nonblocking: AtomicBool::new(is_nonblocking),
+            bound_socket,
+            remote_endpoint,
+            pollee,
+        });
+        connected
+            .bound_socket
+            .set_observer(Arc::downgrade(&connected) as _);
+        connected
+    }
+
+    pub fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
+        // TODO: deal with cmd
+        self.bound_socket.raw_with(|socket: &mut RawTcpSocket| {
+            socket.close();
+        });
+        poll_ifaces();
+        Ok(())
+    }
+
+    pub fn recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<(usize, IpEndpoint)> {
+        debug_assert!(flags.is_all_supported());
+
+        let poller = Poller::new();
+        loop {
+            let recv_len = self.try_recvfrom(buf, flags)?;
+            if recv_len > 0 {
+                let remote_endpoint = self.remote_endpoint()?;
+                return Ok((recv_len, remote_endpoint));
+            }
+            let events = self.poll(IoEvents::IN, Some(&poller));
+            if events.contains(IoEvents::HUP) || events.contains(IoEvents::ERR) {
+                return_errno_with_message!(Errno::ENOTCONN, "recv packet fails");
+            }
+            if !events.contains(IoEvents::IN) {
+                if self.is_nonblocking() {
+                    return_errno_with_message!(Errno::EAGAIN, "try to recv again");
+                }
+                // FIXME: deal with receive timeout
+                poller.wait()?;
+            }
+        }
+    }
+
+    fn try_recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<usize> {
+        poll_ifaces();
+        let res = self.bound_socket.raw_with(|socket: &mut RawTcpSocket| {
+            socket
+                .recv_slice(buf)
+                .map_err(|_| Error::with_message(Errno::ENOTCONN, "fail to recv packet"))
+        });
+        self.update_io_events();
+        res
+    }
+
+    pub fn sendto(&self, buf: &[u8], flags: SendRecvFlags) -> Result<usize> {
+        debug_assert!(flags.is_all_supported());
+
+        let poller = Poller::new();
+        loop {
+            let sent_len = self.try_sendto(buf, flags)?;
+            if sent_len > 0 {
+                return Ok(sent_len);
+            }
+            let events = self.poll(IoEvents::OUT, Some(&poller));
+            if events.contains(IoEvents::HUP) || events.contains(IoEvents::ERR) {
+                return_errno_with_message!(Errno::ENOBUFS, "fail to send packets");
+            }
+            if !events.contains(IoEvents::OUT) {
+                if self.is_nonblocking() {
+                    return_errno_with_message!(Errno::EAGAIN, "try to send again");
+                }
+                // FIXME: deal with send timeout
+                poller.wait()?;
+            }
+        }
+    }
+
+    fn try_sendto(&self, buf: &[u8], flags: SendRecvFlags) -> Result<usize> {
+        let res = self
+            .bound_socket
+            .raw_with(|socket: &mut RawTcpSocket| socket.send_slice(buf))
+            .map_err(|_| Error::with_message(Errno::ENOBUFS, "cannot send packet"));
+        match res {
+            // We have to explicitly invoke `update_io_events` when the send buffer becomes
+            // full. Note that smoltcp does not think it is an interface event, so calling
+            // `poll_ifaces` alone is not enough.
+            Ok(0) => self.update_io_events(),
+            Ok(_) => poll_ifaces(),
+            _ => (),
+        };
+        res
+    }
+
+    pub fn local_endpoint(&self) -> Result<IpEndpoint> {
+        self.bound_socket
+            .local_endpoint()
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "does not has remote endpoint"))
+    }
+
+    pub fn remote_endpoint(&self) -> Result<IpEndpoint> {
+        Ok(self.remote_endpoint)
+    }
+
+    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    fn update_io_events(&self) {
+        self.bound_socket.raw_with(|socket: &mut RawTcpSocket| {
+            let pollee = &self.pollee;
+
+            if socket.can_recv() {
+                pollee.add_events(IoEvents::IN);
+            } else {
+                pollee.del_events(IoEvents::IN);
+            }
+
+            if socket.can_send() {
+                pollee.add_events(IoEvents::OUT);
+            } else {
+                pollee.del_events(IoEvents::OUT);
+            }
+        });
+    }
+
+    pub fn is_nonblocking(&self) -> bool {
+        self.nonblocking.load(Ordering::Relaxed)
+    }
+
+    pub fn set_nonblocking(&self, nonblocking: bool) {
+        self.nonblocking.store(nonblocking, Ordering::Relaxed);
+    }
+}
+
+impl Observer<()> for ConnectedStream {
+    fn on_events(&self, _: &()) {
+        self.update_io_events();
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/stream/connecting.rs
+++ b/services/libs/aster-std/src/net/socket/ip/stream/connecting.rs
@ -0,0 +1,155 @@
+use core::sync::atomic::{AtomicBool, Ordering};
+
+use alloc::sync::Arc;
+
+use crate::events::{IoEvents, Observer};
+use crate::net::iface::RawTcpSocket;
+use crate::net::poll_ifaces;
+use crate::prelude::*;
+
+use crate::net::iface::{AnyBoundSocket, IpEndpoint};
+use crate::process::signal::{Pollee, Poller};
+
+use super::connected::ConnectedStream;
+use super::init::InitStream;
+
+pub struct ConnectingStream {
+    nonblocking: AtomicBool,
+    bound_socket: Arc<AnyBoundSocket>,
+    remote_endpoint: IpEndpoint,
+    conn_result: RwLock<Option<ConnResult>>,
+    pollee: Pollee,
+}
+
+enum ConnResult {
+    Connected,
+    Refused,
+}
+
+impl ConnectingStream {
+    pub fn new(
+        nonblocking: bool,
+        bound_socket: Arc<AnyBoundSocket>,
+        remote_endpoint: IpEndpoint,
+        pollee: Pollee,
+    ) -> Result<Arc<Self>> {
+        bound_socket.do_connect(remote_endpoint)?;
+
+        let connecting = Arc::new(Self {
+            nonblocking: AtomicBool::new(nonblocking),
+            bound_socket,
+            remote_endpoint,
+            conn_result: RwLock::new(None),
+            pollee,
+        });
+        connecting.pollee.reset_events();
+        connecting
+            .bound_socket
+            .set_observer(Arc::downgrade(&connecting) as _);
+        Ok(connecting)
+    }
+
+    pub fn wait_conn(
+        &self,
+    ) -> core::result::Result<Arc<ConnectedStream>, (Error, Arc<InitStream>)> {
+        debug_assert!(!self.is_nonblocking());
+
+        let poller = Poller::new();
+        loop {
+            poll_ifaces();
+
+            match *self.conn_result.read() {
+                Some(ConnResult::Connected) => {
+                    return Ok(ConnectedStream::new(
+                        self.is_nonblocking(),
+                        self.bound_socket.clone(),
+                        self.remote_endpoint,
+                        self.pollee.clone(),
+                    ));
+                }
+                Some(ConnResult::Refused) => {
+                    return Err((
+                        Error::with_message(Errno::ECONNREFUSED, "connection refused"),
+                        InitStream::new_bound(
+                            self.is_nonblocking(),
+                            self.bound_socket.clone(),
+                            self.pollee.clone(),
+                        ),
+                    ));
+                }
+                None => (),
+            };
+
+            let events = self.poll(IoEvents::OUT, Some(&poller));
+            if !events.contains(IoEvents::OUT) {
+                // FIXME: deal with nonblocking mode & connecting timeout
+                poller.wait().expect("async connect() not implemented");
+            }
+        }
+    }
+
+    pub fn local_endpoint(&self) -> Result<IpEndpoint> {
+        self.bound_socket
+            .local_endpoint()
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "no local endpoint"))
+    }
+
+    pub fn remote_endpoint(&self) -> Result<IpEndpoint> {
+        Ok(self.remote_endpoint)
+    }
+
+    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    pub fn is_nonblocking(&self) -> bool {
+        self.nonblocking.load(Ordering::Relaxed)
+    }
+
+    pub fn set_nonblocking(&self, nonblocking: bool) {
+        self.nonblocking.store(nonblocking, Ordering::Relaxed);
+    }
+
+    fn update_io_events(&self) {
+        if self.conn_result.read().is_some() {
+            return;
+        }
+
+        let became_writable = self.bound_socket.raw_with(|socket: &mut RawTcpSocket| {
+            let mut result = self.conn_result.write();
+            if result.is_some() {
+                return false;
+            }
+
+            // Connected
+            if socket.can_send() {
+                *result = Some(ConnResult::Connected);
+                return true;
+            }
+            // Connecting
+            if socket.is_open() {
+                return false;
+            }
+            // Refused
+            *result = Some(ConnResult::Refused);
+            true
+        });
+
+        // Either when the connection is established, or when the connection fails, the socket
+        // shall indicate that it is writable.
+        //
+        // TODO: Find a way to turn `ConnectingStream` into `ConnectedStream` or `InitStream`
+        // here, so non-blocking `connect()` can work correctly. Meanwhile, the latter should
+        // be responsible to initialize all the I/O events including `IoEvents::OUT`, so the
+        // following hard-coded event addition can be removed.
+        if became_writable {
+            self.pollee.add_events(IoEvents::OUT);
+        }
+    }
+}
+
+impl Observer<()> for ConnectingStream {
+    fn on_events(&self, _: &()) {
+        self.update_io_events();
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/stream/init.rs
+++ b/services/libs/aster-std/src/net/socket/ip/stream/init.rs
@ -0,0 +1,153 @@
+use core::sync::atomic::{AtomicBool, Ordering};
+
+use crate::events::IoEvents;
+use crate::net::iface::Iface;
+use crate::net::iface::IpEndpoint;
+use crate::net::iface::{AnyBoundSocket, AnyUnboundSocket};
+use crate::net::socket::ip::always_some::AlwaysSome;
+use crate::net::socket::ip::common::{bind_socket, get_ephemeral_endpoint};
+use crate::prelude::*;
+use crate::process::signal::Pollee;
+use crate::process::signal::Poller;
+
+use super::connecting::ConnectingStream;
+use super::listen::ListenStream;
+
+pub struct InitStream {
+    inner: RwLock<Inner>,
+    is_nonblocking: AtomicBool,
+    pollee: Pollee,
+}
+
+enum Inner {
+    Unbound(AlwaysSome<Box<AnyUnboundSocket>>),
+    Bound(AlwaysSome<Arc<AnyBoundSocket>>),
+}
+
+impl Inner {
+    fn new() -> Inner {
+        let unbound_socket = Box::new(AnyUnboundSocket::new_tcp());
+        Inner::Unbound(AlwaysSome::new(unbound_socket))
+    }
+
+    fn is_bound(&self) -> bool {
+        match self {
+            Self::Unbound(_) => false,
+            Self::Bound(_) => true,
+        }
+    }
+
+    fn bind(&mut self, endpoint: IpEndpoint) -> Result<()> {
+        let unbound_socket = if let Inner::Unbound(unbound_socket) = self {
+            unbound_socket
+        } else {
+            return_errno_with_message!(Errno::EINVAL, "the socket is already bound to an address");
+        };
+        let bound_socket =
+            unbound_socket.try_take_with(|raw_socket| bind_socket(raw_socket, endpoint, false))?;
+        *self = Inner::Bound(AlwaysSome::new(bound_socket));
+        Ok(())
+    }
+
+    fn bind_to_ephemeral_endpoint(&mut self, remote_endpoint: &IpEndpoint) -> Result<()> {
+        let endpoint = get_ephemeral_endpoint(remote_endpoint);
+        self.bind(endpoint)
+    }
+
+    fn bound_socket(&self) -> Option<&Arc<AnyBoundSocket>> {
+        match self {
+            Inner::Bound(bound_socket) => Some(bound_socket),
+            Inner::Unbound(_) => None,
+        }
+    }
+
+    fn iface(&self) -> Option<Arc<dyn Iface>> {
+        match self {
+            Inner::Bound(bound_socket) => Some(bound_socket.iface().clone()),
+            Inner::Unbound(_) => None,
+        }
+    }
+
+    fn local_endpoint(&self) -> Option<IpEndpoint> {
+        self.bound_socket()
+            .and_then(|socket| socket.local_endpoint())
+    }
+}
+
+impl InitStream {
+    // FIXME: In Linux we have the `POLLOUT` event for a newly created socket, while calling
+    // `write()` on it triggers `SIGPIPE`/`EPIPE`. No documentation found yet, but confirmed by
+    // experimentation and Linux source code.
+    pub fn new(nonblocking: bool) -> Arc<Self> {
+        Arc::new(Self {
+            inner: RwLock::new(Inner::new()),
+            is_nonblocking: AtomicBool::new(nonblocking),
+            pollee: Pollee::new(IoEvents::empty()),
+        })
+    }
+
+    pub fn new_bound(
+        nonblocking: bool,
+        bound_socket: Arc<AnyBoundSocket>,
+        pollee: Pollee,
+    ) -> Arc<Self> {
+        bound_socket.set_observer(Weak::<()>::new());
+        let inner = Inner::Bound(AlwaysSome::new(bound_socket));
+        Arc::new(Self {
+            is_nonblocking: AtomicBool::new(nonblocking),
+            inner: RwLock::new(inner),
+            pollee,
+        })
+    }
+
+    pub fn bind(&self, endpoint: IpEndpoint) -> Result<()> {
+        self.inner.write().bind(endpoint)
+    }
+
+    pub fn connect(&self, remote_endpoint: &IpEndpoint) -> Result<Arc<ConnectingStream>> {
+        if !self.inner.read().is_bound() {
+            self.inner
+                .write()
+                .bind_to_ephemeral_endpoint(remote_endpoint)?
+        }
+        ConnectingStream::new(
+            self.is_nonblocking(),
+            self.inner.read().bound_socket().unwrap().clone(),
+            *remote_endpoint,
+            self.pollee.clone(),
+        )
+    }
+
+    pub fn listen(&self, backlog: usize) -> Result<Arc<ListenStream>> {
+        let bound_socket = if let Some(bound_socket) = self.inner.read().bound_socket() {
+            bound_socket.clone()
+        } else {
+            return_errno_with_message!(Errno::EINVAL, "cannot listen without bound")
+        };
+        ListenStream::new(
+            self.is_nonblocking(),
+            bound_socket,
+            backlog,
+            self.pollee.clone(),
+        )
+    }
+
+    pub fn local_endpoint(&self) -> Result<IpEndpoint> {
+        self.inner
+            .read()
+            .local_endpoint()
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "does not has local endpoint"))
+    }
+
+    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    pub fn is_nonblocking(&self) -> bool {
+        self.is_nonblocking.load(Ordering::Relaxed)
+    }
+
+    pub fn set_nonblocking(&self, nonblocking: bool) {
+        self.is_nonblocking.store(nonblocking, Ordering::Relaxed);
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/stream/listen.rs
+++ b/services/libs/aster-std/src/net/socket/ip/stream/listen.rs
@ -0,0 +1,179 @@
+use core::sync::atomic::{AtomicBool, Ordering};
+
+use crate::events::{IoEvents, Observer};
+use crate::net::iface::{AnyUnboundSocket, BindPortConfig, IpEndpoint};
+
+use crate::net::iface::{AnyBoundSocket, RawTcpSocket};
+use crate::process::signal::{Pollee, Poller};
+use crate::{net::poll_ifaces, prelude::*};
+
+use super::connected::ConnectedStream;
+
+pub struct ListenStream {
+    is_nonblocking: AtomicBool,
+    backlog: usize,
+    /// A bound socket held to ensure the TCP port cannot be released
+    bound_socket: Arc<AnyBoundSocket>,
+    /// Backlog sockets listening at the local endpoint
+    backlog_sockets: RwLock<Vec<BacklogSocket>>,
+    pollee: Pollee,
+}
+
+impl ListenStream {
+    pub fn new(
+        nonblocking: bool,
+        bound_socket: Arc<AnyBoundSocket>,
+        backlog: usize,
+        pollee: Pollee,
+    ) -> Result<Arc<Self>> {
+        let listen_stream = Arc::new(Self {
+            is_nonblocking: AtomicBool::new(nonblocking),
+            backlog,
+            bound_socket,
+            backlog_sockets: RwLock::new(Vec::new()),
+            pollee,
+        });
+        listen_stream.fill_backlog_sockets()?;
+        listen_stream.pollee.reset_events();
+        listen_stream
+            .bound_socket
+            .set_observer(Arc::downgrade(&listen_stream) as _);
+        Ok(listen_stream)
+    }
+
+    pub fn accept(&self) -> Result<(Arc<ConnectedStream>, IpEndpoint)> {
+        // wait to accept
+        let poller = Poller::new();
+        loop {
+            poll_ifaces();
+            let accepted_socket = if let Some(accepted_socket) = self.try_accept() {
+                accepted_socket
+            } else {
+                let events = self.poll(IoEvents::IN, Some(&poller));
+                if !events.contains(IoEvents::IN) {
+                    if self.is_nonblocking() {
+                        return_errno_with_message!(Errno::EAGAIN, "try accept again");
+                    }
+                    // FIXME: deal with accept timeout
+                    poller.wait()?;
+                }
+                continue;
+            };
+            let remote_endpoint = accepted_socket.remote_endpoint().unwrap();
+            let connected_stream = {
+                let BacklogSocket {
+                    bound_socket: backlog_socket,
+                } = accepted_socket;
+                ConnectedStream::new(
+                    false,
+                    backlog_socket,
+                    remote_endpoint,
+                    Pollee::new(IoEvents::empty()),
+                )
+            };
+            return Ok((connected_stream, remote_endpoint));
+        }
+    }
+
+    /// Append sockets listening at LocalEndPoint to support backlog
+    fn fill_backlog_sockets(&self) -> Result<()> {
+        let backlog = self.backlog;
+        let mut backlog_sockets = self.backlog_sockets.write();
+        let current_backlog_len = backlog_sockets.len();
+        debug_assert!(backlog >= current_backlog_len);
+        if backlog == current_backlog_len {
+            return Ok(());
+        }
+        for _ in current_backlog_len..backlog {
+            let backlog_socket = BacklogSocket::new(&self.bound_socket)?;
+            backlog_sockets.push(backlog_socket);
+        }
+        Ok(())
+    }
+
+    fn try_accept(&self) -> Option<BacklogSocket> {
+        let backlog_socket = {
+            let mut backlog_sockets = self.backlog_sockets.write();
+            let index = backlog_sockets
+                .iter()
+                .position(|backlog_socket| backlog_socket.is_active())?;
+            backlog_sockets.remove(index)
+        };
+        self.fill_backlog_sockets().unwrap();
+        self.update_io_events();
+        Some(backlog_socket)
+    }
+
+    pub fn local_endpoint(&self) -> Result<IpEndpoint> {
+        self.bound_socket
+            .local_endpoint()
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "does not has remote endpoint"))
+    }
+
+    pub fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+
+    fn update_io_events(&self) {
+        // The lock should be held to avoid data races
+        let backlog_sockets = self.backlog_sockets.read();
+
+        let can_accept = backlog_sockets.iter().any(|socket| socket.is_active());
+        if can_accept {
+            self.pollee.add_events(IoEvents::IN);
+        } else {
+            self.pollee.del_events(IoEvents::IN);
+        }
+    }
+
+    pub fn is_nonblocking(&self) -> bool {
+        self.is_nonblocking.load(Ordering::Relaxed)
+    }
+
+    pub fn set_nonblocking(&self, nonblocking: bool) {
+        self.is_nonblocking.store(nonblocking, Ordering::Relaxed);
+    }
+}
+
+impl Observer<()> for ListenStream {
+    fn on_events(&self, _: &()) {
+        self.update_io_events();
+    }
+}
+
+struct BacklogSocket {
+    bound_socket: Arc<AnyBoundSocket>,
+}
+
+impl BacklogSocket {
+    fn new(bound_socket: &Arc<AnyBoundSocket>) -> Result<Self> {
+        let local_endpoint = bound_socket.local_endpoint().ok_or(Error::with_message(
+            Errno::EINVAL,
+            "the socket is not bound",
+        ))?;
+        let unbound_socket = Box::new(AnyUnboundSocket::new_tcp());
+        let bound_socket = {
+            let iface = bound_socket.iface();
+            let bind_port_config = BindPortConfig::new(local_endpoint.port, true)?;
+            iface
+                .bind_socket(unbound_socket, bind_port_config)
+                .map_err(|(e, _)| e)?
+        };
+        bound_socket.raw_with(|raw_tcp_socket: &mut RawTcpSocket| {
+            raw_tcp_socket
+                .listen(local_endpoint)
+                .map_err(|_| Error::with_message(Errno::EINVAL, "fail to listen"))
+        })?;
+        Ok(Self { bound_socket })
+    }
+
+    fn is_active(&self) -> bool {
+        self.bound_socket
+            .raw_with(|socket: &mut RawTcpSocket| socket.is_active())
+    }
+
+    fn remote_endpoint(&self) -> Option<IpEndpoint> {
+        self.bound_socket
+            .raw_with(|socket: &mut RawTcpSocket| socket.remote_endpoint())
+    }
+}
--- a/services/libs/aster-std/src/net/socket/ip/stream/mod.rs
+++ b/services/libs/aster-std/src/net/socket/ip/stream/mod.rs
@ -0,0 +1,257 @@
+use crate::events::IoEvents;
+use crate::fs::{file_handle::FileLike, utils::StatusFlags};
+use crate::net::iface::IpEndpoint;
+use crate::net::socket::{
+    util::{
+        send_recv_flags::SendRecvFlags, shutdown_cmd::SockShutdownCmd,
+        sock_options::SockOptionName, sockaddr::SocketAddr,
+    },
+    Socket,
+};
+use crate::prelude::*;
+use crate::process::signal::Poller;
+
+use self::{
+    connected::ConnectedStream, connecting::ConnectingStream, init::InitStream,
+    listen::ListenStream,
+};
+
+mod connected;
+mod connecting;
+mod init;
+mod listen;
+
+pub struct StreamSocket {
+    state: RwLock<State>,
+}
+
+enum State {
+    // Start state
+    Init(Arc<InitStream>),
+    // Intermediate state
+    Connecting(Arc<ConnectingStream>),
+    // Final State 1
+    Connected(Arc<ConnectedStream>),
+    // Final State 2
+    Listen(Arc<ListenStream>),
+}
+
+impl StreamSocket {
+    pub fn new(nonblocking: bool) -> Self {
+        let state = State::Init(InitStream::new(nonblocking));
+        Self {
+            state: RwLock::new(state),
+        }
+    }
+
+    fn is_nonblocking(&self) -> bool {
+        match &*self.state.read() {
+            State::Init(init) => init.is_nonblocking(),
+            State::Connecting(connecting) => connecting.is_nonblocking(),
+            State::Connected(connected) => connected.is_nonblocking(),
+            State::Listen(listen) => listen.is_nonblocking(),
+        }
+    }
+
+    fn set_nonblocking(&self, nonblocking: bool) {
+        match &*self.state.read() {
+            State::Init(init) => init.set_nonblocking(nonblocking),
+            State::Connecting(connecting) => connecting.set_nonblocking(nonblocking),
+            State::Connected(connected) => connected.set_nonblocking(nonblocking),
+            State::Listen(listen) => listen.set_nonblocking(nonblocking),
+        }
+    }
+
+    fn do_connect(&self, remote_endpoint: &IpEndpoint) -> Result<Arc<ConnectingStream>> {
+        let mut state = self.state.write();
+        let init_stream = match &*state {
+            State::Init(init_stream) => init_stream,
+            State::Listen(_) | State::Connecting(_) | State::Connected(_) => {
+                return_errno_with_message!(Errno::EINVAL, "cannot connect")
+            }
+        };
+
+        let connecting = init_stream.connect(remote_endpoint)?;
+        *state = State::Connecting(connecting.clone());
+        Ok(connecting)
+    }
+}
+
+impl FileLike for StreamSocket {
+    fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        // FIXME: set correct flags
+        let flags = SendRecvFlags::empty();
+        let (recv_len, _) = self.recvfrom(buf, flags)?;
+        Ok(recv_len)
+    }
+
+    fn write(&self, buf: &[u8]) -> Result<usize> {
+        // FIXME: set correct flags
+        let flags = SendRecvFlags::empty();
+        self.sendto(buf, None, flags)
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        let state = self.state.read();
+        match &*state {
+            State::Init(init) => init.poll(mask, poller),
+            State::Connecting(connecting) => connecting.poll(mask, poller),
+            State::Connected(connected) => connected.poll(mask, poller),
+            State::Listen(listen) => listen.poll(mask, poller),
+        }
+    }
+
+    fn status_flags(&self) -> StatusFlags {
+        if self.is_nonblocking() {
+            StatusFlags::O_NONBLOCK
+        } else {
+            StatusFlags::empty()
+        }
+    }
+
+    fn set_status_flags(&self, new_flags: StatusFlags) -> Result<()> {
+        if new_flags.contains(StatusFlags::O_NONBLOCK) {
+            self.set_nonblocking(true);
+        } else {
+            self.set_nonblocking(false);
+        }
+        Ok(())
+    }
+
+    fn as_socket(&self) -> Option<&dyn Socket> {
+        Some(self)
+    }
+}
+
+impl Socket for StreamSocket {
+    fn bind(&self, sockaddr: SocketAddr) -> Result<()> {
+        let endpoint = sockaddr.try_into()?;
+        let state = self.state.read();
+        match &*state {
+            State::Init(init_stream) => init_stream.bind(endpoint),
+            _ => return_errno_with_message!(Errno::EINVAL, "cannot bind"),
+        }
+    }
+
+    fn connect(&self, sockaddr: SocketAddr) -> Result<()> {
+        let remote_endpoint = sockaddr.try_into()?;
+
+        let connecting_stream = self.do_connect(&remote_endpoint)?;
+        match connecting_stream.wait_conn() {
+            Ok(connected_stream) => {
+                *self.state.write() = State::Connected(connected_stream);
+                Ok(())
+            }
+            Err((err, init_stream)) => {
+                *self.state.write() = State::Init(init_stream);
+                Err(err)
+            }
+        }
+    }
+
+    fn listen(&self, backlog: usize) -> Result<()> {
+        let mut state = self.state.write();
+        let init_stream = match &*state {
+            State::Init(init_stream) => init_stream,
+            State::Connecting(connecting_stream) => {
+                return_errno_with_message!(Errno::EINVAL, "cannot listen for a connecting stream")
+            }
+            State::Listen(listen_stream) => {
+                return_errno_with_message!(Errno::EINVAL, "cannot listen for a listening stream")
+            }
+            State::Connected(_) => return_errno_with_message!(Errno::EINVAL, "cannot listen"),
+        };
+
+        let listener = init_stream.listen(backlog)?;
+        *state = State::Listen(listener);
+        Ok(())
+    }
+
+    fn accept(&self) -> Result<(Arc<dyn FileLike>, SocketAddr)> {
+        let listen_stream = match &*self.state.read() {
+            State::Listen(listen_stream) => listen_stream.clone(),
+            _ => return_errno_with_message!(Errno::EINVAL, "the socket is not listening"),
+        };
+
+        let (connected_stream, remote_endpoint) = {
+            let listen_stream = listen_stream.clone();
+            listen_stream.accept()?
+        };
+
+        let accepted_socket = {
+            let state = RwLock::new(State::Connected(connected_stream));
+            Arc::new(StreamSocket { state })
+        };
+
+        let socket_addr = remote_endpoint.try_into()?;
+        Ok((accepted_socket, socket_addr))
+    }
+
+    fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
+        let state = self.state.read();
+        match &*state {
+            State::Connected(connected_stream) => connected_stream.shutdown(cmd),
+            // TDOD: shutdown listening stream
+            _ => return_errno_with_message!(Errno::EINVAL, "cannot shutdown"),
+        }
+    }
+
+    fn addr(&self) -> Result<SocketAddr> {
+        let state = self.state.read();
+        let local_endpoint = match &*state {
+            State::Init(init_stream) => init_stream.local_endpoint(),
+            State::Connecting(connecting_stream) => connecting_stream.local_endpoint(),
+            State::Listen(listen_stream) => listen_stream.local_endpoint(),
+            State::Connected(connected_stream) => connected_stream.local_endpoint(),
+        }?;
+        local_endpoint.try_into()
+    }
+
+    fn peer_addr(&self) -> Result<SocketAddr> {
+        let state = self.state.read();
+        let remote_endpoint = match &*state {
+            State::Init(init_stream) => {
+                return_errno_with_message!(Errno::EINVAL, "init socket does not have peer")
+            }
+            State::Connecting(connecting_stream) => connecting_stream.remote_endpoint(),
+            State::Listen(listen_stream) => {
+                return_errno_with_message!(Errno::EINVAL, "listening socket does not have peer")
+            }
+            State::Connected(connected_stream) => connected_stream.remote_endpoint(),
+        }?;
+        remote_endpoint.try_into()
+    }
+
+    fn sock_option(&self, optname: &SockOptionName) -> Result<&[u8]> {
+        return_errno_with_message!(Errno::EINVAL, "getsockopt not implemented");
+    }
+
+    fn recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<(usize, SocketAddr)> {
+        let connected_stream = match &*self.state.read() {
+            State::Connected(connected_stream) => connected_stream.clone(),
+            _ => return_errno_with_message!(Errno::EINVAL, "the socket is not connected"),
+        };
+
+        let (recv_size, remote_endpoint) = connected_stream.recvfrom(buf, flags)?;
+        let socket_addr = remote_endpoint.try_into()?;
+        Ok((recv_size, socket_addr))
+    }
+
+    fn sendto(
+        &self,
+        buf: &[u8],
+        remote: Option<SocketAddr>,
+        flags: SendRecvFlags,
+    ) -> Result<usize> {
+        debug_assert!(remote.is_none());
+        if remote.is_some() {
+            return_errno_with_message!(Errno::EINVAL, "tcp socked should not provide remote addr");
+        }
+
+        let connected_stream = match &*self.state.read() {
+            State::Connected(connected_stream) => connected_stream.clone(),
+            _ => return_errno_with_message!(Errno::EINVAL, "the socket is not connected"),
+        };
+        connected_stream.sendto(buf, flags)
+    }
+}
--- a/services/libs/aster-std/src/net/socket/mod.rs
+++ b/services/libs/aster-std/src/net/socket/mod.rs
@ -0,0 +1,78 @@
+use crate::{fs::file_handle::FileLike, prelude::*};
+
+pub use self::util::send_recv_flags::SendRecvFlags;
+pub use self::util::shutdown_cmd::SockShutdownCmd;
+pub use self::util::sock_options::{SockOptionLevel, SockOptionName};
+pub use self::util::sockaddr::SocketAddr;
+
+pub mod ip;
+pub mod unix;
+mod util;
+
+/// Operations defined on a socket.
+pub trait Socket: FileLike + Send + Sync {
+    /// Assign the address specified by sockaddr to the socket
+    fn bind(&self, sockaddr: SocketAddr) -> Result<()> {
+        return_errno_with_message!(Errno::EINVAL, "bind not implemented");
+    }
+
+    /// Build connection for a given address
+    fn connect(&self, sockaddr: SocketAddr) -> Result<()> {
+        return_errno_with_message!(Errno::EINVAL, "connect not implemented");
+    }
+
+    /// Listen for connections on a socket
+    fn listen(&self, backlog: usize) -> Result<()> {
+        return_errno_with_message!(Errno::EINVAL, "connect not implemented");
+    }
+
+    /// Accept a connection on a socket
+    fn accept(&self) -> Result<(Arc<dyn FileLike>, SocketAddr)> {
+        return_errno_with_message!(Errno::EINVAL, "accept not implemented");
+    }
+
+    /// Shut down part of a full-duplex connection
+    fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
+        return_errno_with_message!(Errno::EINVAL, "shutdown not implemented");
+    }
+
+    /// Get address of this socket.
+    fn addr(&self) -> Result<SocketAddr> {
+        return_errno_with_message!(Errno::EINVAL, "getsockname not implemented");
+    }
+
+    /// Get address of peer socket
+    fn peer_addr(&self) -> Result<SocketAddr> {
+        return_errno_with_message!(Errno::EINVAL, "getpeername not implemented");
+    }
+
+    /// Get options on the socket
+    fn sock_option(&self, optname: &SockOptionName) -> Result<&[u8]> {
+        return_errno_with_message!(Errno::EINVAL, "getsockopt not implemented");
+    }
+
+    /// Set options on the socket
+    fn set_sock_option(
+        &self,
+        opt_level: SockOptionLevel,
+        optname: SockOptionName,
+        option_val: &[u8],
+    ) -> Result<()> {
+        return_errno_with_message!(Errno::EINVAL, "setsockopt not implemented");
+    }
+
+    /// Receive a message from a socket
+    fn recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<(usize, SocketAddr)> {
+        return_errno_with_message!(Errno::EINVAL, "recvfrom not implemented");
+    }
+
+    /// Send a message on a socket
+    fn sendto(
+        &self,
+        buf: &[u8],
+        remote: Option<SocketAddr>,
+        flags: SendRecvFlags,
+    ) -> Result<usize> {
+        return_errno_with_message!(Errno::EINVAL, "recvfrom not implemented");
+    }
+}
--- a/services/libs/aster-std/src/net/socket/unix/addr.rs
+++ b/services/libs/aster-std/src/net/socket/unix/addr.rs
@ -0,0 +1,55 @@
+use crate::fs::utils::Dentry;
+use crate::net::socket::util::sockaddr::SocketAddr;
+use crate::prelude::*;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum UnixSocketAddr {
+    Path(String),
+    Abstract(String),
+}
+
+#[derive(Clone)]
+pub(super) enum UnixSocketAddrBound {
+    Path(Arc<Dentry>),
+    Abstract(String),
+}
+
+impl PartialEq for UnixSocketAddrBound {
+    fn eq(&self, other: &Self) -> bool {
+        match (self, other) {
+            (Self::Abstract(l0), Self::Abstract(r0)) => l0 == r0,
+            (Self::Path(l0), Self::Path(r0)) => Arc::ptr_eq(l0.inode(), r0.inode()),
+            _ => false,
+        }
+    }
+}
+
+impl TryFrom<SocketAddr> for UnixSocketAddr {
+    type Error = Error;
+
+    fn try_from(value: SocketAddr) -> Result<Self> {
+        match value {
+            SocketAddr::Unix(unix_socket_addr) => Ok(unix_socket_addr),
+            _ => return_errno_with_message!(Errno::EINVAL, "Invalid unix socket addr"),
+        }
+    }
+}
+
+impl From<UnixSocketAddrBound> for UnixSocketAddr {
+    fn from(value: UnixSocketAddrBound) -> Self {
+        match value {
+            UnixSocketAddrBound::Path(dentry) => {
+                let abs_path = dentry.abs_path();
+                Self::Path(abs_path)
+            }
+            UnixSocketAddrBound::Abstract(name) => Self::Abstract(name),
+        }
+    }
+}
+
+impl From<UnixSocketAddrBound> for SocketAddr {
+    fn from(value: UnixSocketAddrBound) -> Self {
+        let unix_socket_addr = UnixSocketAddr::from(value);
+        SocketAddr::Unix(unix_socket_addr)
+    }
+}
--- a/services/libs/aster-std/src/net/socket/unix/mod.rs
+++ b/services/libs/aster-std/src/net/socket/unix/mod.rs
@ -0,0 +1,5 @@
+mod addr;
+mod stream;
+
+pub use addr::UnixSocketAddr;
+pub use stream::UnixStreamSocket;
--- a/services/libs/aster-std/src/net/socket/unix/stream/connected.rs
+++ b/services/libs/aster-std/src/net/socket/unix/stream/connected.rs
@ -0,0 +1,51 @@
+use super::endpoint::Endpoint;
+use crate::events::IoEvents;
+use crate::net::socket::{unix::addr::UnixSocketAddrBound, SockShutdownCmd};
+use crate::prelude::*;
+use crate::process::signal::Poller;
+
+pub(super) struct Connected {
+    local_endpoint: Arc<Endpoint>,
+}
+
+impl Connected {
+    pub(super) fn new(local_endpoint: Arc<Endpoint>) -> Self {
+        Connected { local_endpoint }
+    }
+
+    pub(super) fn addr(&self) -> Option<UnixSocketAddrBound> {
+        self.local_endpoint.addr()
+    }
+
+    pub(super) fn peer_addr(&self) -> Option<UnixSocketAddrBound> {
+        self.local_endpoint.peer_addr()
+    }
+
+    pub(super) fn is_bound(&self) -> bool {
+        self.addr().is_some()
+    }
+
+    pub(super) fn write(&self, buf: &[u8]) -> Result<usize> {
+        self.local_endpoint.write(buf)
+    }
+
+    pub(super) fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        self.local_endpoint.read(buf)
+    }
+
+    pub(super) fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
+        self.local_endpoint.shutdown(cmd)
+    }
+
+    pub(super) fn is_nonblocking(&self) -> bool {
+        self.local_endpoint.is_nonblocking()
+    }
+
+    pub(super) fn set_nonblocking(&self, is_nonblocking: bool) {
+        self.local_endpoint.set_nonblocking(is_nonblocking).unwrap();
+    }
+
+    pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.local_endpoint.poll(mask, poller)
+    }
+}
--- a/services/libs/aster-std/src/net/socket/unix/stream/endpoint.rs
+++ b/services/libs/aster-std/src/net/socket/unix/stream/endpoint.rs
@ -0,0 +1,125 @@
+use crate::events::IoEvents;
+use crate::process::signal::Poller;
+use crate::{
+    fs::utils::{Channel, Consumer, Producer, StatusFlags},
+    net::socket::{unix::addr::UnixSocketAddrBound, SockShutdownCmd},
+    prelude::*,
+};
+
+pub(super) struct Endpoint(Inner);
+
+struct Inner {
+    addr: RwLock<Option<UnixSocketAddrBound>>,
+    reader: Consumer<u8>,
+    writer: Producer<u8>,
+    peer: Weak<Endpoint>,
+}
+
+impl Endpoint {
+    pub(super) fn new_pair(is_nonblocking: bool) -> Result<(Arc<Endpoint>, Arc<Endpoint>)> {
+        let flags = if is_nonblocking {
+            StatusFlags::O_NONBLOCK
+        } else {
+            StatusFlags::empty()
+        };
+        let (writer_a, reader_b) =
+            Channel::with_capacity_and_flags(DAFAULT_BUF_SIZE, flags)?.split();
+        let (writer_b, reader_a) =
+            Channel::with_capacity_and_flags(DAFAULT_BUF_SIZE, flags)?.split();
+        let mut endpoint_b = None;
+        let endpoint_a = Arc::new_cyclic(|endpoint_a_ref| {
+            let peer = Arc::new(Endpoint::new(reader_b, writer_b, endpoint_a_ref.clone()));
+            let endpoint_a = Endpoint::new(reader_a, writer_a, Arc::downgrade(&peer));
+            endpoint_b = Some(peer);
+            endpoint_a
+        });
+        Ok((endpoint_a, endpoint_b.unwrap()))
+    }
+
+    fn new(reader: Consumer<u8>, writer: Producer<u8>, peer: Weak<Endpoint>) -> Self {
+        Self(Inner {
+            addr: RwLock::new(None),
+            reader,
+            writer,
+            peer,
+        })
+    }
+
+    pub(super) fn addr(&self) -> Option<UnixSocketAddrBound> {
+        self.0.addr.read().clone()
+    }
+
+    pub(super) fn set_addr(&self, addr: UnixSocketAddrBound) {
+        *self.0.addr.write() = Some(addr);
+    }
+
+    pub(super) fn peer_addr(&self) -> Option<UnixSocketAddrBound> {
+        self.0.peer.upgrade().and_then(|peer| peer.addr())
+    }
+
+    pub(super) fn is_nonblocking(&self) -> bool {
+        let reader_status = self.0.reader.is_nonblocking();
+        let writer_status = self.0.writer.is_nonblocking();
+        debug_assert!(reader_status == writer_status);
+        reader_status
+    }
+
+    pub(super) fn set_nonblocking(&self, is_nonblocking: bool) -> Result<()> {
+        let reader_flags = self.0.reader.status_flags();
+        self.0
+            .reader
+            .set_status_flags(reader_flags | StatusFlags::O_NONBLOCK)?;
+        let writer_flags = self.0.writer.status_flags();
+        self.0
+            .writer
+            .set_status_flags(writer_flags | StatusFlags::O_NONBLOCK)?;
+        Ok(())
+    }
+
+    pub(super) fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        self.0.reader.read(buf)
+    }
+
+    pub(super) fn write(&self, buf: &[u8]) -> Result<usize> {
+        self.0.writer.write(buf)
+    }
+
+    pub(super) fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
+        if !self.is_connected() {
+            return_errno_with_message!(Errno::ENOTCONN, "The socket is not connected.");
+        }
+
+        if cmd.shut_read() {
+            self.0.reader.shutdown();
+        }
+
+        if cmd.shut_write() {
+            self.0.writer.shutdown();
+        }
+
+        Ok(())
+    }
+
+    pub(super) fn is_connected(&self) -> bool {
+        self.0.peer.upgrade().is_some()
+    }
+
+    pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        let mut events = IoEvents::empty();
+        // FIXME: should reader and writer use the same mask?
+        let reader_events = self.0.reader.poll(mask, poller);
+        let writer_events = self.0.writer.poll(mask, poller);
+
+        if reader_events.contains(IoEvents::HUP) || self.0.reader.is_shutdown() {
+            events |= IoEvents::RDHUP | IoEvents::IN;
+            if writer_events.contains(IoEvents::ERR) || self.0.writer.is_shutdown() {
+                events |= IoEvents::HUP | IoEvents::OUT;
+            }
+        }
+
+        events |= (reader_events & IoEvents::IN) | (writer_events & IoEvents::OUT);
+        events
+    }
+}
+
+const DAFAULT_BUF_SIZE: usize = 4096;
--- a/services/libs/aster-std/src/net/socket/unix/stream/init.rs
+++ b/services/libs/aster-std/src/net/socket/unix/stream/init.rs
@ -0,0 +1,101 @@
+use core::sync::atomic::{AtomicBool, Ordering};
+
+use crate::events::IoEvents;
+use crate::fs::fs_resolver::{split_path, FsPath};
+use crate::fs::utils::{Dentry, InodeMode, InodeType};
+use crate::net::socket::unix::addr::{UnixSocketAddr, UnixSocketAddrBound};
+use crate::prelude::*;
+use crate::process::signal::{Pollee, Poller};
+
+use super::connected::Connected;
+use super::endpoint::Endpoint;
+use super::listener::push_incoming;
+
+pub(super) struct Init {
+    is_nonblocking: AtomicBool,
+    addr: Mutex<Option<UnixSocketAddrBound>>,
+    pollee: Pollee,
+}
+
+impl Init {
+    pub(super) fn new(is_nonblocking: bool) -> Self {
+        Self {
+            is_nonblocking: AtomicBool::new(is_nonblocking),
+            addr: Mutex::new(None),
+            pollee: Pollee::new(IoEvents::empty()),
+        }
+    }
+
+    pub(super) fn bind(&self, addr_to_bind: &UnixSocketAddr) -> Result<()> {
+        let mut addr = self.addr.lock();
+        if addr.is_some() {
+            return_errno_with_message!(Errno::EINVAL, "the socket is already bound");
+        }
+
+        let bound_addr = match addr_to_bind {
+            UnixSocketAddr::Abstract(_) => todo!(),
+            UnixSocketAddr::Path(path) => {
+                let dentry = create_socket_file(path)?;
+                UnixSocketAddrBound::Path(dentry)
+            }
+        };
+
+        *addr = Some(bound_addr);
+        Ok(())
+    }
+
+    pub(super) fn connect(&self, remote_addr: &UnixSocketAddrBound) -> Result<Connected> {
+        let addr = self.addr();
+
+        if let Some(ref addr) = addr {
+            if *addr == *remote_addr {
+                return_errno_with_message!(Errno::EINVAL, "try to connect to self is invalid");
+            }
+        }
+
+        let (this_end, remote_end) = Endpoint::new_pair(self.is_nonblocking())?;
+        remote_end.set_addr(remote_addr.clone());
+        if let Some(addr) = addr {
+            this_end.set_addr(addr.clone());
+        };
+
+        push_incoming(remote_addr, remote_end)?;
+        Ok(Connected::new(this_end))
+    }
+
+    pub(super) fn is_bound(&self) -> bool {
+        self.addr.lock().is_some()
+    }
+
+    pub(super) fn addr(&self) -> Option<UnixSocketAddrBound> {
+        self.addr.lock().clone()
+    }
+
+    pub(super) fn is_nonblocking(&self) -> bool {
+        self.is_nonblocking.load(Ordering::Acquire)
+    }
+
+    pub(super) fn set_nonblocking(&self, is_nonblocking: bool) {
+        self.is_nonblocking.store(is_nonblocking, Ordering::Release);
+    }
+
+    pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        self.pollee.poll(mask, poller)
+    }
+}
+
+fn create_socket_file(path: &str) -> Result<Arc<Dentry>> {
+    let (parent_pathname, file_name) = split_path(path);
+    let parent = {
+        let current = current!();
+        let fs = current.fs().read();
+        let parent_path = FsPath::try_from(parent_pathname)?;
+        fs.lookup(&parent_path)?
+    };
+    let dentry = parent.create(
+        file_name,
+        InodeType::Socket,
+        InodeMode::S_IRUSR | InodeMode::S_IWUSR,
+    )?;
+    Ok(dentry)
+}
--- a/services/libs/aster-std/src/net/socket/unix/stream/listener.rs
+++ b/services/libs/aster-std/src/net/socket/unix/stream/listener.rs
@ -0,0 +1,219 @@
+use super::{connected::Connected, endpoint::Endpoint, UnixStreamSocket};
+use crate::events::IoEvents;
+use crate::fs::file_handle::FileLike;
+use crate::fs::utils::{Dentry, Inode};
+use crate::net::socket::unix::addr::{UnixSocketAddr, UnixSocketAddrBound};
+use crate::net::socket::SocketAddr;
+use crate::prelude::*;
+use crate::process::signal::{Pollee, Poller};
+use core::sync::atomic::{AtomicBool, Ordering};
+use keyable_arc::KeyableWeak;
+
+pub(super) struct Listener {
+    addr: UnixSocketAddrBound,
+    is_nonblocking: AtomicBool,
+}
+
+impl Listener {
+    pub(super) fn new(
+        addr: UnixSocketAddrBound,
+        backlog: usize,
+        nonblocking: bool,
+    ) -> Result<Self> {
+        BACKLOG_TABLE.add_backlog(&addr, backlog)?;
+        Ok(Self {
+            addr,
+            is_nonblocking: AtomicBool::new(nonblocking),
+        })
+    }
+
+    pub(super) fn addr(&self) -> &UnixSocketAddrBound {
+        &self.addr
+    }
+
+    pub(super) fn is_nonblocking(&self) -> bool {
+        self.is_nonblocking.load(Ordering::Acquire)
+    }
+
+    pub(super) fn set_nonblocking(&self, is_nonblocking: bool) {
+        self.is_nonblocking.store(is_nonblocking, Ordering::Release);
+    }
+
+    pub(super) fn accept(&self) -> Result<(Arc<dyn FileLike>, SocketAddr)> {
+        let addr = self.addr().clone();
+        let is_nonblocking = self.is_nonblocking();
+
+        let connected = {
+            let local_endpoint = BACKLOG_TABLE.pop_incoming(is_nonblocking, &addr)?;
+            Connected::new(local_endpoint)
+        };
+
+        let peer_addr = match connected.peer_addr() {
+            None => SocketAddr::Unix(UnixSocketAddr::Path(String::new())),
+            Some(addr) => SocketAddr::from(addr.clone()),
+        };
+
+        let socket = Arc::new(UnixStreamSocket::new_connected(connected));
+
+        Ok((socket, peer_addr))
+    }
+
+    pub(super) fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        let addr = self.addr();
+        let backlog = BACKLOG_TABLE.get_backlog(addr).unwrap();
+        backlog.poll(mask, poller)
+    }
+}
+
+static BACKLOG_TABLE: BacklogTable = BacklogTable::new();
+
+struct BacklogTable {
+    backlog_sockets: RwLock<BTreeMap<KeyableWeak<dyn Inode>, Arc<Backlog>>>,
+    // TODO: For linux, there is also abstract socket domain that a socket addr is not bound to an inode.
+}
+
+impl BacklogTable {
+    const fn new() -> Self {
+        Self {
+            backlog_sockets: RwLock::new(BTreeMap::new()),
+        }
+    }
+
+    fn add_backlog(&self, addr: &UnixSocketAddrBound, backlog: usize) -> Result<()> {
+        let inode = {
+            let UnixSocketAddrBound::Path(dentry) = addr else {
+                todo!()
+            };
+            create_keyable_inode(dentry)
+        };
+
+        let mut backlog_sockets = self.backlog_sockets.write();
+        if backlog_sockets.contains_key(&inode) {
+            return_errno_with_message!(Errno::EADDRINUSE, "the addr is already used");
+        }
+        let new_backlog = Arc::new(Backlog::new(backlog));
+        backlog_sockets.insert(inode, new_backlog);
+        Ok(())
+    }
+
+    fn get_backlog(&self, addr: &UnixSocketAddrBound) -> Result<Arc<Backlog>> {
+        let inode = {
+            let UnixSocketAddrBound::Path(dentry) = addr else {
+                todo!()
+            };
+            create_keyable_inode(dentry)
+        };
+
+        let backlog_sockets = self.backlog_sockets.read();
+        backlog_sockets
+            .get(&inode)
+            .map(Arc::clone)
+            .ok_or_else(|| Error::with_message(Errno::EINVAL, "the socket is not listened"))
+    }
+
+    fn pop_incoming(&self, nonblocking: bool, addr: &UnixSocketAddrBound) -> Result<Arc<Endpoint>> {
+        let poller = Poller::new();
+        loop {
+            let backlog = self.get_backlog(addr)?;
+
+            if let Some(endpoint) = backlog.pop_incoming() {
+                return Ok(endpoint);
+            }
+
+            if nonblocking {
+                return_errno_with_message!(Errno::EAGAIN, "no connection comes");
+            }
+
+            let events = {
+                let mask = IoEvents::IN;
+                backlog.poll(mask, Some(&poller))
+            };
+
+            if events.contains(IoEvents::ERR) | events.contains(IoEvents::HUP) {
+                return_errno_with_message!(Errno::ECONNABORTED, "connection is aborted");
+            }
+
+            // FIXME: deal with accept timeout
+            if events.is_empty() {
+                poller.wait()?;
+            }
+        }
+    }
+
+    fn push_incoming(&self, addr: &UnixSocketAddrBound, endpoint: Arc<Endpoint>) -> Result<()> {
+        let backlog = self.get_backlog(addr).map_err(|_| {
+            Error::with_message(
+                Errno::ECONNREFUSED,
+                "no socket is listened at the remote address",
+            )
+        })?;
+
+        backlog.push_incoming(endpoint)
+    }
+
+    fn remove_backlog(&self, addr: &UnixSocketAddrBound) {
+        let UnixSocketAddrBound::Path(dentry) = addr else {
+            todo!()
+        };
+
+        let inode = create_keyable_inode(dentry);
+        self.backlog_sockets.write().remove(&inode);
+    }
+}
+
+struct Backlog {
+    pollee: Pollee,
+    backlog: usize,
+    incoming_endpoints: Mutex<VecDeque<Arc<Endpoint>>>,
+}
+
+impl Backlog {
+    fn new(backlog: usize) -> Self {
+        Self {
+            pollee: Pollee::new(IoEvents::empty()),
+            backlog,
+            incoming_endpoints: Mutex::new(VecDeque::with_capacity(backlog)),
+        }
+    }
+
+    fn push_incoming(&self, endpoint: Arc<Endpoint>) -> Result<()> {
+        let mut endpoints = self.incoming_endpoints.lock();
+        if endpoints.len() >= self.backlog {
+            return_errno_with_message!(Errno::ECONNREFUSED, "incoming_endpoints is full");
+        }
+        endpoints.push_back(endpoint);
+        self.pollee.add_events(IoEvents::IN);
+        Ok(())
+    }
+
+    fn pop_incoming(&self) -> Option<Arc<Endpoint>> {
+        let mut incoming_endpoints = self.incoming_endpoints.lock();
+        let endpoint = incoming_endpoints.pop_front();
+        if incoming_endpoints.is_empty() {
+            self.pollee.del_events(IoEvents::IN);
+        }
+        endpoint
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        // Lock to avoid any events may change pollee state when we poll
+        let _lock = self.incoming_endpoints.lock();
+        self.pollee.poll(mask, poller)
+    }
+}
+
+fn create_keyable_inode(dentry: &Arc<Dentry>) -> KeyableWeak<dyn Inode> {
+    let weak_inode = Arc::downgrade(dentry.inode());
+    KeyableWeak::from(weak_inode)
+}
+
+pub(super) fn unregister_backlog(addr: &UnixSocketAddrBound) {
+    BACKLOG_TABLE.remove_backlog(addr);
+}
+
+pub(super) fn push_incoming(
+    remote_addr: &UnixSocketAddrBound,
+    remote_end: Arc<Endpoint>,
+) -> Result<()> {
+    BACKLOG_TABLE.push_incoming(remote_addr, remote_end)
+}
--- a/services/libs/aster-std/src/net/socket/unix/stream/mod.rs
+++ b/services/libs/aster-std/src/net/socket/unix/stream/mod.rs
@ -0,0 +1,7 @@
+mod connected;
+mod endpoint;
+mod init;
+mod listener;
+mod socket;
+
+pub use socket::UnixStreamSocket;
--- a/services/libs/aster-std/src/net/socket/unix/stream/socket.rs
+++ b/services/libs/aster-std/src/net/socket/unix/stream/socket.rs
@ -0,0 +1,299 @@
+use crate::events::IoEvents;
+use crate::fs::file_handle::FileLike;
+use crate::fs::fs_resolver::FsPath;
+use crate::fs::utils::{Dentry, InodeType, StatusFlags};
+use crate::net::socket::unix::addr::UnixSocketAddrBound;
+use crate::net::socket::unix::UnixSocketAddr;
+use crate::net::socket::util::send_recv_flags::SendRecvFlags;
+use crate::net::socket::util::sockaddr::SocketAddr;
+use crate::net::socket::{SockShutdownCmd, Socket};
+use crate::prelude::*;
+use crate::process::signal::Poller;
+
+use super::connected::Connected;
+use super::endpoint::Endpoint;
+use super::init::Init;
+use super::listener::{unregister_backlog, Listener};
+
+pub struct UnixStreamSocket(RwLock<State>);
+
+impl UnixStreamSocket {
+    pub(super) fn new_init(init: Init) -> Self {
+        Self(RwLock::new(State::Init(Arc::new(init))))
+    }
+
+    pub(super) fn new_listen(listen: Listener) -> Self {
+        Self(RwLock::new(State::Listen(Arc::new(listen))))
+    }
+
+    pub(super) fn new_connected(connected: Connected) -> Self {
+        Self(RwLock::new(State::Connected(Arc::new(connected))))
+    }
+}
+
+enum State {
+    Init(Arc<Init>),
+    Listen(Arc<Listener>),
+    Connected(Arc<Connected>),
+}
+
+impl UnixStreamSocket {
+    pub fn new(nonblocking: bool) -> Self {
+        let init = Init::new(nonblocking);
+        Self::new_init(init)
+    }
+
+    pub fn new_pair(nonblocking: bool) -> Result<(Arc<Self>, Arc<Self>)> {
+        let (end_a, end_b) = Endpoint::new_pair(nonblocking)?;
+        let connected_a = {
+            let connected = Connected::new(end_a);
+            Self::new_connected(connected)
+        };
+        let connected_b = {
+            let connected = Connected::new(end_b);
+            Self::new_connected(connected)
+        };
+        Ok((Arc::new(connected_a), Arc::new(connected_b)))
+    }
+
+    fn bound_addr(&self) -> Option<UnixSocketAddrBound> {
+        let status = self.0.read();
+        match &*status {
+            State::Init(init) => init.addr(),
+            State::Listen(listen) => Some(listen.addr().clone()),
+            State::Connected(connected) => connected.addr(),
+        }
+    }
+
+    fn mask_flags(status_flags: &StatusFlags) -> StatusFlags {
+        const SUPPORTED_FLAGS: StatusFlags = StatusFlags::O_NONBLOCK;
+        const UNSUPPORTED_FLAGS: StatusFlags = SUPPORTED_FLAGS.complement();
+
+        if status_flags.intersects(UNSUPPORTED_FLAGS) {
+            warn!("ignore unsupported flags");
+        }
+
+        status_flags.intersection(SUPPORTED_FLAGS)
+    }
+}
+
+impl FileLike for UnixStreamSocket {
+    fn as_socket(&self) -> Option<&dyn Socket> {
+        Some(self)
+    }
+
+    fn read(&self, buf: &mut [u8]) -> Result<usize> {
+        self.recvfrom(buf, SendRecvFlags::empty())
+            .map(|(read_size, _)| read_size)
+    }
+
+    fn write(&self, buf: &[u8]) -> Result<usize> {
+        self.sendto(buf, None, SendRecvFlags::empty())
+    }
+
+    fn poll(&self, mask: IoEvents, poller: Option<&Poller>) -> IoEvents {
+        let inner = self.0.read();
+        match &*inner {
+            State::Init(init) => init.poll(mask, poller),
+            State::Listen(listen) => listen.poll(mask, poller),
+            State::Connected(connected) => connected.poll(mask, poller),
+        }
+    }
+
+    fn status_flags(&self) -> StatusFlags {
+        let inner = self.0.read();
+        let is_nonblocking = match &*inner {
+            State::Init(init) => init.is_nonblocking(),
+            State::Listen(listen) => listen.is_nonblocking(),
+            State::Connected(connected) => connected.is_nonblocking(),
+        };
+
+        if is_nonblocking {
+            StatusFlags::O_NONBLOCK
+        } else {
+            StatusFlags::empty()
+        }
+    }
+
+    fn set_status_flags(&self, new_flags: StatusFlags) -> Result<()> {
+        let is_nonblocking = {
+            let supported_flags = Self::mask_flags(&new_flags);
+            supported_flags.contains(StatusFlags::O_NONBLOCK)
+        };
+
+        let mut inner = self.0.write();
+        match &mut *inner {
+            State::Init(init) => init.set_nonblocking(is_nonblocking),
+            State::Listen(listen) => listen.set_nonblocking(is_nonblocking),
+            State::Connected(connected) => connected.set_nonblocking(is_nonblocking),
+        }
+        Ok(())
+    }
+}
+
+impl Socket for UnixStreamSocket {
+    fn bind(&self, sockaddr: SocketAddr) -> Result<()> {
+        let addr = UnixSocketAddr::try_from(sockaddr)?;
+
+        let init = match &*self.0.read() {
+            State::Init(init) => init.clone(),
+            _ => return_errno_with_message!(
+                Errno::EINVAL,
+                "cannot bind a listening or connected socket"
+            ),
+            // FIXME: Maybe binding a connected socket should also be allowed?
+        };
+
+        init.bind(&addr)
+    }
+
+    fn connect(&self, sockaddr: SocketAddr) -> Result<()> {
+        let remote_addr = {
+            let unix_socket_addr = UnixSocketAddr::try_from(sockaddr)?;
+            match unix_socket_addr {
+                UnixSocketAddr::Abstract(abstract_name) => {
+                    UnixSocketAddrBound::Abstract(abstract_name)
+                }
+                UnixSocketAddr::Path(path) => {
+                    let dentry = lookup_socket_file(&path)?;
+                    UnixSocketAddrBound::Path(dentry)
+                }
+            }
+        };
+
+        let init = match &*self.0.read() {
+            State::Init(init) => init.clone(),
+            State::Listen(_) => return_errno_with_message!(Errno::EINVAL, "the socket is listened"),
+            State::Connected(_) => {
+                return_errno_with_message!(Errno::EISCONN, "the socket is connected")
+            }
+        };
+
+        let connected = init.connect(&remote_addr)?;
+
+        *self.0.write() = State::Connected(Arc::new(connected));
+        Ok(())
+    }
+
+    fn listen(&self, backlog: usize) -> Result<()> {
+        let init = match &*self.0.read() {
+            State::Init(init) => init.clone(),
+            State::Listen(_) => {
+                return_errno_with_message!(Errno::EINVAL, "the socket is already listening")
+            }
+            State::Connected(_) => {
+                return_errno_with_message!(Errno::EISCONN, "the socket is already connected")
+            }
+        };
+
+        let addr = init.addr().ok_or(Error::with_message(
+            Errno::EINVAL,
+            "the socket is not bound",
+        ))?;
+
+        let listener = Listener::new(addr.clone(), backlog, init.is_nonblocking())?;
+        *self.0.write() = State::Listen(Arc::new(listener));
+        Ok(())
+    }
+
+    fn accept(&self) -> Result<(Arc<dyn FileLike>, SocketAddr)> {
+        let listen = match &*self.0.read() {
+            State::Listen(listen) => listen.clone(),
+            _ => return_errno_with_message!(Errno::EINVAL, "the socket is not listening"),
+        };
+
+        listen.accept()
+    }
+
+    fn shutdown(&self, cmd: SockShutdownCmd) -> Result<()> {
+        let connected = match &*self.0.read() {
+            State::Connected(connected) => connected.clone(),
+            _ => return_errno_with_message!(Errno::ENOTCONN, "the socked is not connected"),
+        };
+
+        connected.shutdown(cmd)
+    }
+
+    fn addr(&self) -> Result<SocketAddr> {
+        let addr = match &*self.0.read() {
+            State::Init(init) => init.addr(),
+            State::Listen(listen) => Some(listen.addr().clone()),
+            State::Connected(connected) => connected.addr(),
+        };
+
+        addr.map(Into::<SocketAddr>::into)
+            .ok_or(Error::with_message(
+                Errno::EINVAL,
+                "the socket does not bind to addr",
+            ))
+    }
+
+    fn peer_addr(&self) -> Result<SocketAddr> {
+        let connected = match &*self.0.read() {
+            State::Connected(connected) => connected.clone(),
+            _ => return_errno_with_message!(Errno::ENOTCONN, "the socket is not connected"),
+        };
+
+        match connected.peer_addr() {
+            None => Ok(SocketAddr::Unix(UnixSocketAddr::Path(String::new()))),
+            Some(peer_addr) => Ok(SocketAddr::from(peer_addr.clone())),
+        }
+    }
+
+    fn recvfrom(&self, buf: &mut [u8], flags: SendRecvFlags) -> Result<(usize, SocketAddr)> {
+        let connected = match &*self.0.read() {
+            State::Connected(connected) => connected.clone(),
+            _ => return_errno_with_message!(Errno::ENOTCONN, "the socket is not connected"),
+        };
+
+        let peer_addr = self.peer_addr()?;
+        let read_size = connected.read(buf)?;
+        Ok((read_size, peer_addr))
+    }
+
+    fn sendto(
+        &self,
+        buf: &[u8],
+        remote: Option<SocketAddr>,
+        flags: SendRecvFlags,
+    ) -> Result<usize> {
+        debug_assert!(remote.is_none());
+        // TODO: deal with flags
+        let connected = match &*self.0.read() {
+            State::Connected(connected) => connected.clone(),
+            _ => return_errno_with_message!(Errno::ENOTCONN, "the socket is not connected"),
+        };
+
+        connected.write(buf)
+    }
+}
+
+impl Drop for UnixStreamSocket {
+    fn drop(&mut self) {
+        let Some(bound_addr) = self.bound_addr() else {
+            return;
+        };
+
+        if let State::Listen(_) = &*self.0.read() {
+            unregister_backlog(&bound_addr);
+        }
+    }
+}
+
+fn lookup_socket_file(path: &str) -> Result<Arc<Dentry>> {
+    let dentry = {
+        let current = current!();
+        let fs = current.fs().read();
+        let fs_path = FsPath::try_from(path)?;
+        fs.lookup(&fs_path)?
+    };
+
+    if dentry.inode_type() != InodeType::Socket {
+        return_errno_with_message!(Errno::ENOTSOCK, "not a socket file")
+    }
+
+    if !dentry.inode_mode().is_readable() || !dentry.inode_mode().is_writable() {
+        return_errno_with_message!(Errno::EACCES, "the socket cannot be read or written")
+    }
+    Ok(dentry)
+}
--- a/services/libs/aster-std/src/net/socket/util/mod.rs
+++ b/services/libs/aster-std/src/net/socket/util/mod.rs
@ -0,0 +1,4 @@
+pub mod send_recv_flags;
+pub mod shutdown_cmd;
+pub mod sock_options;
+pub mod sockaddr;
--- a/services/libs/aster-std/src/net/socket/util/send_recv_flags.rs
+++ b/services/libs/aster-std/src/net/socket/util/send_recv_flags.rs
@ -0,0 +1,45 @@
+use crate::prelude::*;
+
+bitflags! {
+    /// Flags used for send/recv.
+    /// The definiton is from https://elixir.bootlin.com/linux/v6.0.9/source/include/linux/socket.h
+    #[repr(C)]
+    #[derive(Pod)]
+    pub struct SendRecvFlags: i32 {
+        const MSG_OOB = 1;
+        const MSG_PEEK	= 2;
+        const MSG_DONTROUTE	= 4;
+        // const MSG_TRYHARD  =   4;       /* Synonym for MSG_DONTROUTE for DECnet */
+        const MSG_CTRUNC	= 8;
+        const MSG_PROBE	= 0x10;	/* Do not send. Only probe path f.e. for MTU */
+        const MSG_TRUNC	= 0x20;
+        const MSG_DONTWAIT	= 0x40;	/* Nonblocking io		 */
+        const MSG_EOR       = 0x80;	/* End of record */
+        const MSG_WAITALL	= 0x100;	/* Wait for a full request */
+        const MSG_FIN        = 0x200;
+        const MSG_SYN		= 0x400;
+        const MSG_CONFIRM	= 0x800;	/* Confirm path validity */
+        const MSG_RST		= 0x1000;
+        const MSG_ERRQUEUE	= 0x2000;	/* Fetch message from error queue */
+        const MSG_NOSIGNAL	= 0x4000;	/* Do not generate SIGPIPE */
+        const MSG_MORE	= 0x8000;	/* Sender will send more */
+        const MSG_WAITFORONE	= 0x10000;	/* recvmmsg(): block until 1+ packets avail */
+        const MSG_SENDPAGE_NOPOLICY = 0x10000; /* sendpage() internal : do no apply policy */
+        const MSG_SENDPAGE_NOTLAST = 0x20000; /* sendpage() internal : not the last page */
+        const MSG_BATCH	= 0x40000; /* sendmmsg(): more messages coming */
+        // const MSG_EOF         MSG_FIN
+        const MSG_NO_SHARED_FRAGS = 0x80000; /* sendpage() internal : page frags are not shared */
+        const MSG_SENDPAGE_DECRYPTED	= 0x100000; /* sendpage() internal : page may carry plain text and require encryption */
+    }
+}
+
+impl SendRecvFlags {
+    fn supported_flags() -> Self {
+        SendRecvFlags::empty()
+    }
+
+    pub fn is_all_supported(&self) -> bool {
+        let supported_flags = Self::supported_flags();
+        supported_flags.contains(*self)
+    }
+}
--- a/services/libs/aster-std/src/net/socket/util/shutdown_cmd.rs
+++ b/services/libs/aster-std/src/net/socket/util/shutdown_cmd.rs
@ -0,0 +1,25 @@
+use crate::prelude::*;
+
+/// Shutdown types
+/// From https://elixir.bootlin.com/linux/v6.0.9/source/include/linux/net.h
+#[repr(i32)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, TryFromInt)]
+#[allow(non_camel_case_types)]
+pub enum SockShutdownCmd {
+    /// Shutdown receptions
+    SHUT_RD = 0,
+    /// Shutdown transmissions
+    SHUT_WR = 1,
+    /// Shutdown receptions and transmissions
+    SHUT_RDWR = 2,
+}
+
+impl SockShutdownCmd {
+    pub fn shut_read(&self) -> bool {
+        *self == Self::SHUT_RD || *self == Self::SHUT_RDWR
+    }
+
+    pub fn shut_write(&self) -> bool {
+        *self == Self::SHUT_WR || *self == Self::SHUT_RDWR
+    }
+}
--- a/services/libs/aster-std/src/net/socket/util/sock_options.rs
+++ b/services/libs/aster-std/src/net/socket/util/sock_options.rs
@ -0,0 +1,41 @@
+use crate::prelude::*;
+
+/// The definition is from https://elixir.bootlin.com/linux/v6.0.9/source/include/uapi/asm-generic/socket.h.
+/// We do not include all options here
+#[repr(i32)]
+#[derive(Debug, Clone, Copy, TryFromInt, PartialEq, Eq, PartialOrd, Ord)]
+#[allow(non_camel_case_types)]
+pub enum SockOptionName {
+    SO_DEBUG = 1,
+    SO_REUSEADDR = 2,
+    SO_TYPE = 3,
+    SO_ERROR = 4,
+    SO_DONTROUTE = 5,
+    SO_BROADCAST = 6,
+    SO_SNDBUF = 7,
+    SO_RCVBUF = 8,
+    SO_SNDBUFFORCE = 32,
+    SO_RCVBUFFORCE = 33,
+    SO_KEEPALIVE = 9,
+    SO_OOBINLINE = 10,
+    SO_NO_CHECK = 11,
+    SO_PRIORITY = 12,
+    SO_LINGER = 13,
+    SO_BSDCOMPAT = 14,
+    SO_REUSEPORT = 15,
+    SO_RCVTIMEO_NEW = 66,
+    SO_SNDTIMEO_NEW = 67,
+}
+
+/// Sock Opt level. The definition is from https://elixir.bootlin.com/linux/v6.0.9/source/include/linux/socket.h#L343
+#[repr(i32)]
+#[derive(Debug, Clone, Copy, TryFromInt, PartialEq, Eq)]
+#[allow(non_camel_case_types)]
+pub enum SockOptionLevel {
+    SOL_IP = 0,
+    SOL_SOCKET = 1,
+    SOL_TCP = 6,
+    SOL_UDP = 17,
+    SOL_IPV6 = 41,
+    SOL_RAW = 255,
+}
--- a/services/libs/aster-std/src/net/socket/util/sockaddr.rs
+++ b/services/libs/aster-std/src/net/socket/util/sockaddr.rs
@ -0,0 +1,52 @@
+use crate::net::iface::{IpAddress, Ipv4Address};
+use crate::net::iface::{IpEndpoint, IpListenEndpoint};
+use crate::net::socket::unix::UnixSocketAddr;
+use crate::prelude::*;
+
+type PortNum = u16;
+
+#[derive(Debug)]
+pub enum SocketAddr {
+    Unix(UnixSocketAddr),
+    IPv4(Ipv4Address, PortNum),
+    IPv6,
+}
+
+impl TryFrom<SocketAddr> for IpEndpoint {
+    type Error = Error;
+
+    fn try_from(value: SocketAddr) -> Result<Self> {
+        match value {
+            SocketAddr::IPv4(addr, port) => Ok(IpEndpoint::new(addr.into_address(), port)),
+            _ => return_errno_with_message!(
+                Errno::EINVAL,
+                "sock addr cannot be converted as IpEndpoint"
+            ),
+        }
+    }
+}
+
+impl TryFrom<IpEndpoint> for SocketAddr {
+    type Error = Error;
+
+    fn try_from(endpoint: IpEndpoint) -> Result<Self> {
+        let port = endpoint.port;
+        let socket_addr = match endpoint.addr {
+            IpAddress::Ipv4(addr) => SocketAddr::IPv4(addr, port), // TODO: support IPv6
+        };
+        Ok(socket_addr)
+    }
+}
+
+impl TryFrom<IpListenEndpoint> for SocketAddr {
+    type Error = Error;
+
+    fn try_from(value: IpListenEndpoint) -> Result<Self> {
+        let port = value.port;
+        let socket_addr = match value.addr {
+            None => return_errno_with_message!(Errno::EINVAL, "address is unspecified"),
+            Some(IpAddress::Ipv4(address)) => SocketAddr::IPv4(address, port),
+        };
+        Ok(socket_addr)
+    }
+}