Merge pull request #3298 from Pana/bump/mio4

bump: mio version to 1.0.4

Author: Pana

Cargo.lock

@@ -322,8 +322,9 @@ hyper-util = "0.1"
 async-trait = "0.1"
 tower = "0.4"
 pin-project = "1.0"
-mio = "0.6"
-mio-extras = "2.0.6"
+mio = { version = "1.0.4", features = ["os-ext", "net", "os-poll"] }
+mio-timer = { git = "https://github.com/Pana/mio-timer", rev = "251170b" }
+mio-misc = { git = "https://github.com/Pana/mio-misc", rev = "27ad80f" }
 reqwest = "0.12"
 
 # crypto & hash

Cargo.toml

@@ -9,7 +9,7 @@ cfx-addr = { workspace = true }
 cfx-bytes = { workspace = true }
 cfx-types = { workspace = true }
 cfxkey = { workspace = true }
-mio = { workspace = true }
+mio = { workspace = true, features = ["os-ext", "net", "os-poll"] }
 parking_lot = { workspace = true }
 log = { workspace = true }
 slab = { workspace = true }

crates/network/Cargo.toml

@@ -13,7 +13,7 @@ use log::{debug, trace};
 use metrics::{
     register_meter_with_group, Gauge, GaugeUsize, Histogram, Meter, Sample,
 };
-use mio::{tcp::TcpStream, Poll, PollOpt, Ready, Token};
+use mio::{net::TcpStream, Interest, Registry, Token};
 use priority_send_queue::{PrioritySendQueue, SendQueuePriority};
 use serde::Deserialize;
 use serde_derive::Serialize;
@@ -183,7 +183,7 @@ pub struct GenericConnection<Socket: GenericSocket> {
     /// Sending packet.
     sending_packet: Option<Packet>,
     /// Event flags this connection interested
-    interest: Ready,
+    interest: Interest,
     /// Registered flag
     registered: AtomicBool,
     /// Assemble packet with extra information before sending out.
@@ -324,7 +324,10 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
         let status = self.write_next_from_queue()?;
 
         if self.sending_packet.is_none() && self.send_queue.is_empty() {
-            self.interest.remove(Ready::writable());
+            self.interest = self
+                .interest
+                .remove(Interest::WRITABLE)
+                .expect("Interest::WRITABLE should not be empty");
         }
         NETWORK_SEND_QUEUE_SIZE.update(self.send_queue.len());
         io.update_registration(self.token)?;
@@ -361,7 +364,7 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
             }
 
             if !self.interest.is_writable() {
-                self.interest.insert(Ready::writable());
+                self.interest = self.interest.add(Interest::WRITABLE);
             }
             io.update_registration(self.token).ok();
         }
@@ -384,15 +387,18 @@ impl Connection {
             recv_buf: BytesMut::new(),
             send_queue: PrioritySendQueue::default(),
             sending_packet: None,
-            interest: Ready::hup() | Ready::readable(),
+            interest: Interest::READABLE, /* previously(mio0.6) use
+                                           * Ready::hup() |
+                                           * Ready::readable(), Ready::hub()
+                                           * is removed from 0.7 */
             registered: AtomicBool::new(false),
             assembler: Box::new(PacketWithLenAssembler::default()),
         }
     }
 
     /// Register this connection with the IO event loop.
     pub fn register_socket(
-        &self, reg: Token, event_loop: &Poll,
+        &mut self, reg: Token, poll_registry: &Registry,
     ) -> io::Result<()> {
         if self.registered.load(AtomicOrdering::SeqCst) {
             return Ok(());
@@ -402,12 +408,9 @@ impl Connection {
             self.token,
             reg
         );
-        if let Err(e) = event_loop.register(
-            &self.socket,
-            reg,
-            self.interest,
-            PollOpt::edge(),
-        ) {
+        if let Err(e) =
+            poll_registry.register(&mut self.socket, reg, self.interest)
+        {
             trace!(
                 "Failed to register socket, token = {}, reg = {:?}, err = {:?}",
                 self.token,
@@ -422,18 +425,18 @@ impl Connection {
     /// Update connection registration. Should be called at the end of the IO
     /// handler.
     pub fn update_socket(
-        &self, reg: Token, event_loop: &Poll,
+        &mut self, reg: Token, poll_registry: &Registry,
     ) -> io::Result<()> {
         trace!(
             "Connection reregister, token = {}, reg = {:?}",
             self.token,
             reg
         );
         if !self.registered.load(AtomicOrdering::SeqCst) {
-            self.register_socket(reg, event_loop)
+            self.register_socket(reg, poll_registry)
         } else {
-            event_loop
-                .reregister(&self.socket, reg, self.interest, PollOpt::edge())
+            poll_registry
+                .reregister(&mut self.socket, reg, self.interest)
                 .unwrap_or_else(|e| {
                     trace!("Failed to reregister socket, token = {}, reg = {:?}, err = {:?}", self.token, reg, e);
                 });
@@ -443,9 +446,11 @@ impl Connection {
 
     /// Delete connection registration. Should be called at the end of the IO
     /// handler.
-    pub fn deregister_socket(&self, event_loop: &Poll) -> io::Result<()> {
+    pub fn deregister_socket(
+        &mut self, poll_registry: &Registry,
+    ) -> io::Result<()> {
         trace!("Connection deregister, token = {}", self.token);
-        event_loop.deregister(&self.socket).ok();
+        poll_registry.deregister(&mut self.socket).ok();
         Ok(())
     }
 
@@ -591,7 +596,7 @@ impl PacketAssembler for PacketWithLenAssembler {
 mod tests {
     use super::*;
     use crate::iolib::IoChannel;
-    use mio::Ready;
+    use mio::Interest;
     use std::{
         cmp,
         io::{Read, Result, Write},
@@ -672,7 +677,7 @@ mod tests {
                 send_queue: PrioritySendQueue::default(),
                 sending_packet: None,
                 recv_buf: BytesMut::new(),
-                interest: Ready::hup() | Ready::readable(),
+                interest: Interest::READABLE,
                 registered: AtomicBool::new(false),
                 assembler: Box::new(PacketWithLenAssembler::new(1, None)),
             }

crates/network/src/connection.rs

@@ -28,7 +28,7 @@ use crate::{
 use cfx_types::{Public, H256};
 use io::{IoContext, StreamToken};
 use log::{debug, error, trace};
-use mio::tcp::TcpStream;
+use mio::net::TcpStream;
 use priority_send_queue::SendQueuePriority;
 use std::{
     sync::atomic::{AtomicBool, Ordering},