From 44c90059b650356141f57eb33df5dde5c93b4430 Mon Sep 17 00:00:00 2001
From: Carl Lerche <me@carllerche.com>
Date: Sun, 6 May 2018 11:16:13 -0700
Subject: [PATCH] Assert that stream state is released

This patch adds a debug level assertion checking that state associated
with streams is successfully released. This is done by ensuring that no
stream state remains when the connection inner state is dropped. This
does not happen until all handles to the connection drops, which should
result in all streams being dropped.

This also pulls in a fix for one bug that is exposed with this change.
The fix is first proposed as part of #261.
---
 src/proto/streams/prioritize.rs | 15 ++++++++-------
 src/proto/streams/store.rs      |  4 ++++
 src/proto/streams/streams.rs    | 10 ++++++++++
 3 files changed, 22 insertions(+), 7 deletions(-)

diff --git a/src/proto/streams/prioritize.rs b/src/proto/streams/prioritize.rs
index 0de95c682..8012c6603 100644
--- a/src/proto/streams/prioritize.rs
+++ b/src/proto/streams/prioritize.rs
@@ -613,21 +613,22 @@ impl Prioritize {
                     trace!("pop_frame; stream={:?}; stream.state={:?}",
                         stream.id, stream.state);
 
+                    // It's possible that this stream, besides having data to send,
+                    // is also queued to send a reset, and thus is already in the queue
+                    // to wait for "some time" after a reset.
+                    //
+                    // To be safe, we just always ask the stream.
+                    let is_pending_reset = stream.is_pending_reset_expiration();
+
                     // If the stream receives a RESET from the peer, it may have
                     // had data buffered to be sent, but all the frames are cleared
                     // in clear_queue(). Instead of doing O(N) traversal through queue
                     // to remove, lets just ignore peer_reset streams here.
                     if stream.state.is_peer_reset() {
+                        counts.transition_after(stream, is_pending_reset);
                         continue;
                     }
 
-                    // It's possible that this stream, besides having data to send,
-                    // is also queued to send a reset, and thus is already in the queue
-                    // to wait for "some time" after a reset.
-                    //
-                    // To be safe, we just always ask the stream.
-                    let is_pending_reset = stream.is_pending_reset_expiration();
-
                     trace!(" --> stream={:?}; is_pending_reset={:?};",
                         stream.id, is_pending_reset);
 
diff --git a/src/proto/streams/store.rs b/src/proto/streams/store.rs
index 121b29124..09cefe86e 100644
--- a/src/proto/streams/store.rs
+++ b/src/proto/streams/store.rs
@@ -85,6 +85,10 @@ impl Store {
         }
     }
 
+    pub fn is_empty(&self) -> bool {
+        self.slab.is_empty()
+    }
+
     pub fn find_mut(&mut self, id: &StreamId) -> Option<Ptr> {
         let key = match self.ids.get(id) {
             Some(key) => *key,
diff --git a/src/proto/streams/streams.rs b/src/proto/streams/streams.rs
index 6b3ad8566..b51bfd494 100644
--- a/src/proto/streams/streams.rs
+++ b/src/proto/streams/streams.rs
@@ -1099,3 +1099,13 @@ impl Actions {
         }
     }
 }
+
+impl Drop for Inner {
+    fn drop(&mut self) {
+        use std::thread;
+
+        if !thread::panicking() {
+            debug_assert!(self.store.is_empty());
+        }
+    }
+}