Occlum 异步网络框架

IoHandle
	-> IoToken
    	->Inner
    		->  state
    			completion_callback
    			token_key

 let file_ref = current!().file(fd as FileDesc)?;
    if let Ok(socket) = file_ref.as_host_socket() {
        let raw_addr = addr.to_raw();
        socket.bind(&raw_addr)?;
    } else if let Ok(unix_socket) = file_ref.as_unix_socket() {
        let unix_addr = addr.to_unix()?;
        unix_socket.bind(unix_addr)?;
    } else if let Ok(uring_socket) = file_ref.as_uring_socket() {
        uring_socket.bind(&addr)?;
    } else {
        return_errno!(ENOTSOCK, "not a socket");
    }

pub trait UringSocketType {
    fn as_uring_socket(&self) -> Result<&SocketFile>;
}

impl UringSocketType for FileRef {
    fn as_uring_socket(&self) -> Result<&SocketFile> {
        self.as_any()
            .downcast_ref::<SocketFile>()
            .ok_or_else(|| errno!(ENOTSOCK, "not a uring socket"))
    }
}

pub struct SocketFile {
    socket: AnySocket,
}

enum AnySocket {
    Ipv4Stream(Ipv4Stream),
    Ipv6Stream(Ipv6Stream),
    Ipv4Datagram(Ipv4Datagram),
    Ipv6Datagram(Ipv6Datagram),
}

// A 泛型标识地址类型，比如 IPv4、IPv6，R 泛型表示运行时，这里主要用于 io_uring
pub struct Common<A: Addr + 'static, R: Runtime> {
    host_fd: FileDesc,					// host_fd：主机上的文件描述符
    type_: SocketType,					// socket 类型
    nonblocking: AtomicBool,			// 是否是阻塞类型
    is_closed: AtomicBool,				// socket 是否关闭
    pollee: Pollee,						// 用于 Pollee-Poller 机制，实现异步 IO
    inner: Mutex<Inner<A>>,				// 保存了一个 Stream Socket 连接（bind）的 C/S 地址
    timeout: Mutex<Timeout>,			// 等待时延
    errno: Mutex<Option<Errno>>,		// 错误码
    io_uring: Arc<IoUring>,				// socket 绑定的 IoUring 实例
    phantom_data: PhantomData<(A, R)>,	// 标识 Common 结构体拥有和泛型一样的生命周期
}

// 泛型类型和 Common 类似
pub struct DatagramSocket<A: Addr + 'static, R: Runtime> {
    common: Arc<Common<A, R>>,	// 持有一个 common 结构
    state: RwLock<State>,				// 当前 DG socket 的状态，包含绑定状态和是否连接
    sender: Arc<Sender<A, R>>,	// socket 的发送端，封装发送行为，对应 output 事件
    receiver: Arc<Receiver<A, R>>,	// socket 的接受端，封装接收行为，对应 input 事件
}

struct State {
    bind_state: BindState,	// 这里的绑定状态有三种：没有绑定、显式和隐式
    is_connected: bool,			// 是否建立连接
}

pub fn bind(&self, addr: &A) -> Result<()> {
  let mut state = self.state.write().unwrap();
  if state.is_bound() {
    return_errno!(EINVAL, "The socket is already bound to an address");
  }

  do_bind(self.host_fd(), addr)?;

  self.common.set_addr(addr);
  state.mark_explicit_bind();
  // Start async recv after explicit binding or implicit binding
  self.receiver.initiate_async_recv();

  Ok(())
}

pub struct StreamSocket<A: Addr + 'static, R: Runtime> {
    state: RwLock<State<A, R>>,
    common: Arc<Common<A, R>>,
}

enum State<A: Addr + 'static, R: Runtime> {
    // Start state 初始状态
    Init(Arc<InitStream<A, R>>),
    // Intermediate state 瞬时状态
    Connect(Arc<ConnectingStream<A, R>>),
    // Final state 1 终态 1
    Connected(Arc<ConnectedStream<A, R>>),
    // Final state 2 终态 2
    Listen(Arc<ListenerStream<A, R>>),
}

pub struct InitStream<A: Addr + 'static, R: Runtime> {
    common: Arc<Common<A, R>>,
    inner: Mutex<Inner>,
}
struct Inner {
    has_bound: bool,
}

pub struct ConnectingStream<A: Addr + 'static, R: Runtime> {
    common: Arc<Common<A, R>>,
    peer_addr: A,	// 连接对侧地址
    req: Mutex<ConnectReq<A>>,	// 连接请求
    connected: AtomicBool, // Mainly use for nonblocking socket to update status asynchronously
}

struct ConnectReq<A: Addr> {
    io_handle: Option<IoHandle>,	// 请求的 io_handle
    c_addr: UntrustedBox<libc::sockaddr_storage>,	// socket 地址
    c_addr_len: usize,	// 地址长度
    errno: Option<Errno>,
    phantom_data: PhantomData<A>,
}

pub struct ConnectedStream<A: Addr + 'static, R: Runtime> {
    common: Arc<Common<A, R>>,
    sender: Sender,
    receiver: Receiver,
}

pub struct ListenerStream<A: Addr + 'static, R: Runtime> {
    common: Arc<Common<A, R>>,
    inner: Mutex<Inner<A>>,
}
struct Inner<A: Addr> {
    backlog: Backlog<A>,
    fatal: Option<Errno>,
}
struct Backlog<A: Addr> {
    // The entries in the backlog.
    entries: Box<[Entry]>,
    // Arguments of the io_uring requests submitted for the entries in the backlog.
    reqs: UntrustedBox<[AcceptReq]>,
    // The indexes of completed entries.
    completed: VecDeque<usize>,
    // The number of free entries.
    num_free: usize,
    phantom_data: PhantomData<A>,
}

let (init_stream, connecting_stream) = {
  let mut state = self.state.write().unwrap();
  match &*state {
    State::Init(init_stream) => {
      let connecting_stream = {
        let common = init_stream.common().clone();
        ConnectingStream::new(peer_addr, common)?
      };
      let init_stream = init_stream.clone();
      *state = State::Connect(connecting_stream.clone());
      (init_stream, connecting_stream)
    }
    State::Connect(connecting_stream) => {
      if let Some(connected_stream) =
      Self::try_switch_to_connected_state(connecting_stream)
      {
        *state = State::Connected(connected_stream);
        return_errno!(EISCONN, "the socket is already connected");
      } else {
        // Not connected, keep the connecting state and try connect
        let init_stream =
        InitStream::new_with_common(connecting_stream.common().clone())?;
        (init_stream, connecting_stream.clone())
      }
    }
    State::Connected(_) => {
      return_errno!(EISCONN, "the socket is already connected");
    }
    State::Listen(_) => {
      return_errno!(EINVAL, "the socket is listening");
    }
  }
};

let res = connecting_stream.connect();
  // If success, then the state is switched to connected; otherwise, for blocking socket
  // the state is restored to the init state, and for non-blocking socket, the state
  // keeps in connecting state.
  match &res {
    Ok(()) => {
      let connected_stream = {
        let common = init_stream.common().clone();
        common.set_peer_addr(peer_addr);
        ConnectedStream::new(common)
      };

      let mut state = self.state.write().unwrap();
      *state = State::Connected(connected_stream);
    }
    Err(_) => {
      if !connecting_stream.common().nonblocking() {
        let mut state = self.state.write().unwrap();
        *state = State::Init(init_stream);
      }
    }
  }

pub fn new(backlog: u32, common: Arc<Common<A, R>>) -> Result<Arc<Self>> {
  // Here we use different variables for the backlog. For the libos, as we will issue async accept request
  // ahead of time, and cacel when the socket closes, we set the libos backlog to a reasonable value which
  // is no greater than the max value we set to save resources and make it more efficient. For the host,
  // we just use the backlog value for maximum connection.
  let libos_backlog = std::cmp::min(backlog, PENDING_ASYNC_ACCEPT_NUM_MAX as u32);
  let host_backlog = backlog;

  let inner = Inner::new(libos_backlog)?;
  Self::do_listen(common.host_fd(), host_backlog)?;

  common.pollee().reset_events();
  let new_self = Arc::new(Self {
    common,
    inner: Mutex::new(inner),
  });

  // Start async accept requests right as early as possible to improve performance
  {
    let inner = new_self.inner.lock();
    new_self.initiate_async_accepts(inner);
  }

  Ok(new_self)
}

// 部分
let host_fd = retval as FileDesc;
inner.backlog.entries[entry_idx] = Entry::Completed { host_fd };
inner.backlog.completed.push_back(entry_idx);

stream.common.pollee().add_events(IoEvents::IN);

stream.initiate_async_accepts(inner);