fix: fix some msg left on buffer

secio/src/codec/secure_stream.rs

@@ -1,5 +1,5 @@
 use bytes::{Bytes, BytesMut};
-use futures::{Sink, StreamExt};
+use futures::{SinkExt, StreamExt};
 use log::{debug, trace};
 use tokio::{
     io::AsyncReadExt,
@@ -161,13 +161,12 @@ where
     ) -> Poll<io::Result<usize>> {
         debug!("start sending plain data: {:?}", buf);
 
-        let frame = self.encode_buffer(buf);
-        trace!("start sending encrypted data size: {:?}", frame.len());
-        let mut sink = Pin::new(&mut self.socket);
-        match sink.as_mut().poll_ready(cx) {
+        match self.socket.poll_ready_unpin(cx) {
             Poll::Ready(Ok(_)) => {
-                sink.as_mut().start_send(frame)?;
-                let _ignore = sink.as_mut().poll_flush(cx)?;
+                let frame = self.encode_buffer(buf);
+                trace!("start sending encrypted data size: {:?}", frame.len());
+                self.socket.start_send_unpin(frame)?;
+                let _ignore = self.socket.poll_flush_unpin(cx)?;
                 Poll::Ready(Ok(buf.len()))
             }
             Poll::Pending => Poll::Pending,
@@ -176,11 +175,11 @@ where
     }
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
-        Pin::new(&mut self.socket).as_mut().poll_flush(cx)
+        self.socket.poll_flush_unpin(cx)
     }
 
     fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
-        Pin::new(&mut self.socket).as_mut().poll_close(cx)
+        self.socket.poll_close_unpin(cx)
     }
 }
 

tentacle/src/channel/queue.rs

@@ -173,3 +173,16 @@ impl<T> Queue<T> {
         }
     }
 }
+
+impl<T> Drop for Queue<T> {
+    fn drop(&mut self) {
+        unsafe {
+            let mut cur = *self.tail.get();
+            while !cur.is_null() {
+                let next = (*cur).next.load(Ordering::Relaxed);
+                drop(Box::from_raw(cur));
+                cur = next;
+            }
+        }
+    }
+}

tentacle/src/substream.rs

@@ -147,6 +147,7 @@ where
             Poll::Pending => {
                 debug!("framed_stream NotReady, frame len: {:?}", frame.len());
                 self.push_front(priority, frame);
+                self.poll_complete(cx)?;
                 Ok(true)
             }
         }
@@ -191,6 +192,10 @@ where
             log::trace!("sub stream poll shutdown err {}", e)
         }
 
+        if !self.keep_buffer {
+            self.event_sender.clear()
+        }
+
         if let Some(ref mut service_proto_sender) = self.service_proto_sender {
             let (mut sender, mut events) = service_proto_sender.take();
             events.push_back(ServiceProtocolEvent::Disconnected {
@@ -246,9 +251,6 @@ where
             | ErrorKind::UnexpectedEof => return,
             _ => (),
         }
-        if !self.keep_buffer {
-            self.event_sender.clear()
-        }
         self.event_sender.push(ProtocolEvent::Error {
             id: self.id,
             proto_id: self.proto_id,
@@ -422,6 +424,7 @@ where
 
     #[inline]
     fn flush(&mut self, cx: &mut Context) -> Result<(), io::Error> {
+        self.poll_complete(cx)?;
         if !self
             .service_proto_sender
             .as_ref()
@@ -488,18 +491,6 @@ where
 
         is_pending &= self.recv_event(cx).is_pending();
 
-        if self.dead || self.context.closed.load(Ordering::SeqCst) {
-            debug!(
-                "Substream({}) finished, self.dead || self.context.closed.load(Ordering::SeqCst), tail",
-                self.id
-            );
-            if !self.keep_buffer {
-                self.event_sender.clear()
-            }
-            self.close_proto_stream(cx);
-            return Poll::Ready(None);
-        }
-
         if is_pending {
             Poll::Pending
         } else {
@@ -680,6 +671,7 @@ where
             Poll::Pending => {
                 debug!("framed_stream NotReady, frame len: {:?}", frame.len());
                 self.push_front(priority, frame);
+                self.poll_complete(cx)?;
                 Ok(true)
             }
         }
@@ -714,6 +706,7 @@ where
 
     #[inline]
     fn flush(&mut self, cx: &mut Context) -> Result<(), io::Error> {
+        self.poll_complete(cx)?;
         if !self.event_sender.is_empty()
             || !self.write_buf.is_empty()
             || !self.high_write_buf.is_empty()
@@ -880,15 +873,6 @@ where
 
         let is_pending = self.recv_event(cx).is_pending();
 
-        if self.dead || self.context.closed.load(Ordering::SeqCst) {
-            debug!(
-                "Substream({}) finished, self.dead || self.context.closed.load(Ordering::SeqCst), tail",
-                self.id
-            );
-            self.close_proto_stream(cx);
-            return Poll::Ready(None);
-        }
-
         if is_pending {
             Poll::Pending
         } else {

yamux/src/session.rs

@@ -13,7 +13,7 @@ use std::{
 use timer::Instant;
 
 use futures::{
-    channel::mpsc::{channel, Receiver, Sender},
+    channel::mpsc::{channel, unbounded, Receiver, Sender, UnboundedReceiver, UnboundedSender},
     Sink, Stream,
 };
 use log::debug;
@@ -82,6 +82,9 @@ pub struct Session<T> {
     event_sender: Sender<StreamEvent>,
     // For receive events from sub streams
     event_receiver: Receiver<StreamEvent>,
+    // The control information must be sent successfully and will not cause memory problems
+    unbound_event_sender: UnboundedSender<StreamEvent>,
+    unbound_event_receiver: UnboundedReceiver<StreamEvent>,
 
     /// use to async open stream/close session
     control_sender: Sender<Command>,
@@ -122,6 +125,7 @@ where
             SessionType::Server => 2,
         };
         let (event_sender, event_receiver) = channel(32);
+        let (unbound_event_sender, unbound_event_receiver) = unbounded();
         let (control_sender, control_receiver) = channel(32);
         let framed_stream = Framed::new(
             raw_stream,
@@ -149,6 +153,8 @@ where
             read_pending_frames: VecDeque::default(),
             event_sender,
             event_receiver,
+            unbound_event_sender,
+            unbound_event_receiver,
             control_sender,
             control_receiver,
             keepalive,
@@ -280,12 +286,13 @@ where
         let mut stream = StreamHandle::new(
             stream_id,
             self.event_sender.clone(),
+            self.unbound_event_sender.clone(),
             frame_receiver,
             state,
             self.config.max_stream_window_size,
             self.config.max_stream_window_size,
         );
-        if let Err(err) = stream.send_window_update(None) {
+        if let Err(err) = stream.send_window_update() {
             debug!("[{:?}] stream.send_window_update error={:?}", self.ty, err);
         }
         Ok(stream)
@@ -516,11 +523,6 @@ where
                     self.streams.remove(&stream_id);
                 }
             }
-            StreamEvent::Flush(stream_id) => {
-                debug!("[{}] session flushing.....", stream_id);
-                self.flush(cx)?;
-                debug!("[{}] session flushed", stream_id);
-            }
             StreamEvent::GoAway => self.send_go_away_with_code(cx, GoAwayCode::ProtocolError)?,
         }
         Ok(())
@@ -543,6 +545,24 @@ where
         }
     }
 
+    fn recv_unbound_events(&mut self, cx: &mut Context) -> Poll<Option<Result<(), io::Error>>> {
+        match Pin::new(&mut self.unbound_event_receiver)
+            .as_mut()
+            .poll_next(cx)
+        {
+            Poll::Ready(Some(event)) => {
+                self.handle_event(cx, event)?;
+                Poll::Ready(Some(Ok(())))
+            }
+            Poll::Ready(None) => {
+                // Since session hold one event sender,
+                // the channel can not be disconnected.
+                unreachable!()
+            }
+            Poll::Pending => Poll::Pending,
+        }
+    }
+
     fn control_poll(&mut self, cx: &mut Context) -> Poll<Option<Result<(), io::Error>>> {
         match Pin::new(&mut self.control_receiver).as_mut().poll_next(cx) {
             Poll::Ready(Some(event)) => {
@@ -609,6 +629,7 @@ where
 
             let mut is_pending = self.control_poll(cx)?.is_pending();
             is_pending &= self.recv_frames(cx)?.is_pending();
+            is_pending &= self.recv_unbound_events(cx)?.is_pending();
             is_pending &= self.recv_events(cx)?.is_pending();
             if is_pending {
                 break;