/src/h2/src/proto/streams/stream.rs

Source
use crate::Reason;

use super::*;

use std::fmt;
use std::task::{Context, Waker};
use std::time::Instant;

/// Tracks Stream related state
///
/// # Reference counting
///
/// There can be a number of outstanding handles to a single Stream. These are
/// tracked using reference counting. The `ref_count` field represents the
/// number of outstanding userspace handles that can reach this stream.
///
/// It's important to note that when the stream is placed in an internal queue
/// (such as an accept queue), this is **not** tracked by a reference count.
/// Thus, `ref_count` can be zero and the stream still has to be kept around.
pub(super) struct Stream {
    /// The h2 stream identifier
    pub id: StreamId,

    /// Current state of the stream
    pub state: State,

    /// Set to `true` when the stream is counted against the connection's max
    /// concurrent streams.
    pub is_counted: bool,

    /// Number of outstanding handles pointing to this stream
    pub ref_count: usize,

    // ===== Fields related to sending =====
    /// Next node in the accept linked list
    pub next_pending_send: Option<store::Key>,

    /// Set to true when the stream is pending accept
    pub is_pending_send: bool,

    /// Send data flow control
    pub send_flow: FlowControl,

    /// Amount of send capacity that has been requested, but not yet allocated.
    pub requested_send_capacity: WindowSize,

    /// Amount of data buffered at the prioritization layer.
    /// TODO: Technically this could be greater than the window size...
    pub buffered_send_data: usize,

    /// Task tracking additional send capacity (i.e. window updates).
    send_task: Option<Waker>,

    /// Frames pending for this stream being sent to the socket
    pub pending_send: buffer::Deque,

    /// Next node in the linked list of streams waiting for additional
    /// connection level capacity.
    pub next_pending_send_capacity: Option<store::Key>,

    /// True if the stream is waiting for outbound connection capacity
    pub is_pending_send_capacity: bool,

    /// Set to true when the send capacity has been incremented
    pub send_capacity_inc: bool,

    /// Next node in the open linked list
    pub next_open: Option<store::Key>,

    /// Set to true when the stream is pending to be opened
    pub is_pending_open: bool,

    /// Set to true when a push is pending for this stream
    pub is_pending_push: bool,

    // ===== Fields related to receiving =====
    /// Next node in the accept linked list
    pub next_pending_accept: Option<store::Key>,

    /// Set to true when the stream is pending accept
    pub is_pending_accept: bool,

    /// Receive data flow control
    pub recv_flow: FlowControl,

    pub in_flight_recv_data: WindowSize,

    /// Next node in the linked list of streams waiting to send window updates.
    pub next_window_update: Option<store::Key>,

    /// True if the stream is waiting to send a window update
    pub is_pending_window_update: bool,

    /// The time when this stream may have been locally reset.
    pub reset_at: Option<Instant>,

    /// Next node in list of reset streams that should expire eventually
    pub next_reset_expire: Option<store::Key>,

    /// Frames pending for this stream to read
    pub pending_recv: buffer::Deque,

    /// When the RecvStream drop occurs, no data should be received.
    pub is_recv: bool,

    /// Task tracking receiving frames
    pub recv_task: Option<Waker>,

    /// Task tracking pushed promises.
    pub push_task: Option<Waker>,

    /// The stream's pending push promises
    pub pending_push_promises: store::Queue<NextAccept>,

    /// Validate content-length headers
    pub content_length: ContentLength,
}

/// State related to validating a stream's content-length
#[derive(Debug)]
pub enum ContentLength {
    Omitted,
    Head,
    Remaining(u64),
}

#[derive(Debug)]
pub(super) struct NextAccept;

#[derive(Debug)]
pub(super) struct NextSend;

#[derive(Debug)]
pub(super) struct NextSendCapacity;

#[derive(Debug)]
pub(super) struct NextWindowUpdate;

#[derive(Debug)]
pub(super) struct NextOpen;

#[derive(Debug)]
pub(super) struct NextResetExpire;

impl Stream {
    pub fn new(id: StreamId, init_send_window: WindowSize, init_recv_window: WindowSize) -> Stream {
        let mut send_flow = FlowControl::new();
        let mut recv_flow = FlowControl::new();

        recv_flow
            .inc_window(init_recv_window)
            .expect("invalid initial receive window");
        // TODO: proper error handling?
        let _res = recv_flow.assign_capacity(init_recv_window);
        debug_assert!(_res.is_ok());

        send_flow
            .inc_window(init_send_window)
            .expect("invalid initial send window size");

        Stream {
            id,
            state: State::default(),
            ref_count: 0,
            is_counted: false,

            // ===== Fields related to sending =====
            next_pending_send: None,
            is_pending_send: false,
            send_flow,
            requested_send_capacity: 0,
            buffered_send_data: 0,
            send_task: None,
            pending_send: buffer::Deque::new(),
            is_pending_send_capacity: false,
            next_pending_send_capacity: None,
            send_capacity_inc: false,
            is_pending_open: false,
            next_open: None,
            is_pending_push: false,

            // ===== Fields related to receiving =====
            next_pending_accept: None,
            is_pending_accept: false,
            recv_flow,
            in_flight_recv_data: 0,
            next_window_update: None,
            is_pending_window_update: false,
            reset_at: None,
            next_reset_expire: None,
            pending_recv: buffer::Deque::new(),
            is_recv: true,
            recv_task: None,
            push_task: None,
            pending_push_promises: store::Queue::new(),
            content_length: ContentLength::Omitted,
        }
    }

    /// Increment the stream's ref count
    pub fn ref_inc(&mut self) {
        assert!(self.ref_count < usize::MAX);
        self.ref_count += 1;
    }

    /// Decrements the stream's ref count
    pub fn ref_dec(&mut self) {
        assert!(self.ref_count > 0);
        self.ref_count -= 1;
    }

    /// Returns true if stream is currently being held for some time because of
    /// a local reset.
    pub fn is_pending_reset_expiration(&self) -> bool {
        self.reset_at.is_some()
    }

    /// Returns true if frames for this stream are ready to be sent over the wire
    pub fn is_send_ready(&self) -> bool {
        // Why do we check pending_open?
        //
        // We allow users to call send_request() which schedules a stream to be pending_open
        // if there is no room according to the concurrency limit (max_send_streams), and we
        // also allow data to be buffered for send with send_data() if there is no capacity for
        // the stream to send the data, which attempts to place the stream in pending_send.
        // If the stream is not open, we don't want the stream to be scheduled for
        // execution (pending_send). Note that if the stream is in pending_open, it will be
        // pushed to pending_send when there is room for an open stream.
        //
        // In pending_push we track whether a PushPromise still needs to be sent
        // from a different stream before we can start sending frames on this one.
        // This is different from the "open" check because reserved streams don't count
        // toward the concurrency limit.
        // See https://httpwg.org/specs/rfc7540.html#rfc.section.5.1.2
        !self.is_pending_open && !self.is_pending_push
    }

    /// Returns true if the stream is closed
    pub fn is_closed(&self) -> bool {
        // The state has fully transitioned to closed.
        self.state.is_closed() &&
            // Because outbound frames transition the stream state before being
            // buffered, we have to ensure that all frames have been flushed.
            self.pending_send.is_empty() &&
            // Sometimes large data frames are sent out in chunks. After a chunk
            // of the frame is sent, the remainder is pushed back onto the send
            // queue to be rescheduled.
            //
            // Checking for additional buffered data lets us catch this case.
            self.buffered_send_data == 0
    }

    /// Returns true if the stream is no longer in use
    pub fn is_released(&self) -> bool {
        // The stream is closed and fully flushed
        self.is_closed() &&
            // There are no more outstanding references to the stream
            self.ref_count == 0 &&
            // The stream is not in any queue
            !self.is_pending_send && !self.is_pending_send_capacity &&
            !self.is_pending_accept && !self.is_pending_window_update &&
            !self.is_pending_open && self.reset_at.is_none()
    }

    /// Returns true when the consumer of the stream has dropped all handles
    /// (indicating no further interest in the stream) and the stream state is
    /// not actually closed.
    ///
    /// In this case, a reset should be sent.
    pub fn is_canceled_interest(&self) -> bool {
        self.ref_count == 0 && !self.state.is_closed()
    }

    /// Current available stream send capacity
    pub fn capacity(&self, max_buffer_size: usize) -> WindowSize {
        let available = self.send_flow.available().as_size() as usize;
        let buffered = self.buffered_send_data;

        available.min(max_buffer_size).saturating_sub(buffered) as WindowSize
    }

    pub fn assign_capacity(&mut self, capacity: WindowSize, max_buffer_size: usize) {
        let prev_capacity = self.capacity(max_buffer_size);
        debug_assert!(capacity > 0);
        // TODO: proper error handling
        let _res = self.send_flow.assign_capacity(capacity);
        debug_assert!(_res.is_ok());

        tracing::trace!(
            "  assigned capacity to stream; available={}; buffered={}; id={:?}; max_buffer_size={} prev={}",
            self.send_flow.available(),
            self.buffered_send_data,
            self.id,
            max_buffer_size,
            prev_capacity,
        );

        if prev_capacity < self.capacity(max_buffer_size) {
            self.notify_capacity();
        }
    }

    pub fn send_data(&mut self, len: WindowSize, max_buffer_size: usize) {
        let prev_capacity = self.capacity(max_buffer_size);

        // TODO: proper error handling
        let _res = self.send_flow.send_data(len);
        debug_assert!(_res.is_ok());

        // Decrement the stream's buffered data counter
        debug_assert!(self.buffered_send_data >= len as usize);
        self.buffered_send_data -= len as usize;
        self.requested_send_capacity -= len;

        tracing::trace!(
            "  sent stream data; available={}; buffered={}; id={:?}; max_buffer_size={} prev={}",
            self.send_flow.available(),
            self.buffered_send_data,
            self.id,
            max_buffer_size,
            prev_capacity,
        );

        if prev_capacity < self.capacity(max_buffer_size) {
            self.notify_capacity();
        }
    }

    /// If the capacity was limited because of the max_send_buffer_size,
    /// then consider waking the send task again...
    pub fn notify_capacity(&mut self) {
        self.send_capacity_inc = true;
        tracing::trace!("  notifying task");
        self.notify_send();
    }

    /// Returns `Err` when the decrement cannot be completed due to overflow.
    pub fn dec_content_length(&mut self, len: usize) -> Result<(), ()> {
        match self.content_length {
            ContentLength::Remaining(ref mut rem) => match rem.checked_sub(len as u64) {
                Some(val) => *rem = val,
                None => return Err(()),
            },
            ContentLength::Head => {
                if len != 0 {
                    return Err(());
                }
            }
            _ => {}
        }

        Ok(())
    }

    pub fn ensure_content_length_zero(&self) -> Result<(), ()> {
        match self.content_length {
            ContentLength::Remaining(0) => Ok(()),
            ContentLength::Remaining(_) => Err(()),
            _ => Ok(()),
        }
    }

    pub fn notify_send(&mut self) {
        if let Some(task) = self.send_task.take() {
            task.wake();
        }
    }

    pub fn wait_send(&mut self, cx: &Context) {
        self.send_task = Some(cx.waker().clone());
    }

    pub fn notify_recv(&mut self) {
        if let Some(task) = self.recv_task.take() {
            task.wake();
        }
    }

    pub(super) fn notify_push(&mut self) {
        if let Some(task) = self.push_task.take() {
            task.wake();
        }
    }

    /// Set the stream's state to `Closed` with the given reason and initiator.
    /// Notify the send, receive, and push tasks, if they exist.
    pub(super) fn set_reset(&mut self, reason: Reason, initiator: Initiator) {
        self.state.set_reset(self.id, reason, initiator);
        self.notify_send();
        self.notify_push();
        self.notify_recv();
    }
}

impl fmt::Debug for Stream {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Stream")
            .field("id", &self.id)
            .field("state", &self.state)
            .field("is_counted", &self.is_counted)
            .field("ref_count", &self.ref_count)
            .h2_field_some("next_pending_send", &self.next_pending_send)
            .h2_field_if("is_pending_send", &self.is_pending_send)
            .field("send_flow", &self.send_flow)
            .field("requested_send_capacity", &self.requested_send_capacity)
            .field("buffered_send_data", &self.buffered_send_data)
            .h2_field_some("send_task", &self.send_task.as_ref().map(|_| ()))
            .h2_field_if_then(
                "pending_send",
                !self.pending_send.is_empty(),
                &self.pending_send,
            )
            .h2_field_some(
                "next_pending_send_capacity",
                &self.next_pending_send_capacity,
            )
            .h2_field_if("is_pending_send_capacity", &self.is_pending_send_capacity)
            .h2_field_if("send_capacity_inc", &self.send_capacity_inc)
            .h2_field_some("next_open", &self.next_open)
            .h2_field_if("is_pending_open", &self.is_pending_open)
            .h2_field_if("is_pending_push", &self.is_pending_push)
            .h2_field_some("next_pending_accept", &self.next_pending_accept)
            .h2_field_if("is_pending_accept", &self.is_pending_accept)
            .field("recv_flow", &self.recv_flow)
            .field("in_flight_recv_data", &self.in_flight_recv_data)
            .h2_field_some("next_window_update", &self.next_window_update)
            .h2_field_if("is_pending_window_update", &self.is_pending_window_update)
            .h2_field_some("reset_at", &self.reset_at)
            .h2_field_some("next_reset_expire", &self.next_reset_expire)
            .h2_field_if_then(
                "pending_recv",
                !self.pending_recv.is_empty(),
                &self.pending_recv,
            )
            .h2_field_if("is_recv", &self.is_recv)
            .h2_field_some("recv_task", &self.recv_task.as_ref().map(|_| ()))
            .h2_field_some("push_task", &self.push_task.as_ref().map(|_| ()))
            .h2_field_if_then(
                "pending_push_promises",
                !self.pending_push_promises.is_empty(),
                &self.pending_push_promises,
            )
            .field("content_length", &self.content_length)
            .finish()
    }
}

impl store::Next for NextAccept {
    fn next(stream: &Stream) -> Option<store::Key> {
        stream.next_pending_accept
    }

    fn set_next(stream: &mut Stream, key: Option<store::Key>) {
        stream.next_pending_accept = key;
    }

    fn take_next(stream: &mut Stream) -> Option<store::Key> {
        stream.next_pending_accept.take()
    }

    fn is_queued(stream: &Stream) -> bool {
        stream.is_pending_accept
    }

    fn set_queued(stream: &mut Stream, val: bool) {
        stream.is_pending_accept = val;
    }
}

impl store::Next for NextSend {
    fn next(stream: &Stream) -> Option<store::Key> {
        stream.next_pending_send
    }

    fn set_next(stream: &mut Stream, key: Option<store::Key>) {
        stream.next_pending_send = key;
    }

    fn take_next(stream: &mut Stream) -> Option<store::Key> {
        stream.next_pending_send.take()
    }

    fn is_queued(stream: &Stream) -> bool {
        stream.is_pending_send
    }

    fn set_queued(stream: &mut Stream, val: bool) {
        if val {
            // ensure that stream is not queued for being opened
            // if it's being put into queue for sending data
            debug_assert!(!stream.is_pending_open);
        }
        stream.is_pending_send = val;
    }
}

impl store::Next for NextSendCapacity {
    fn next(stream: &Stream) -> Option<store::Key> {
        stream.next_pending_send_capacity
    }

    fn set_next(stream: &mut Stream, key: Option<store::Key>) {
        stream.next_pending_send_capacity = key;
    }

    fn take_next(stream: &mut Stream) -> Option<store::Key> {
        stream.next_pending_send_capacity.take()
    }

    fn is_queued(stream: &Stream) -> bool {
        stream.is_pending_send_capacity
    }

    fn set_queued(stream: &mut Stream, val: bool) {
        stream.is_pending_send_capacity = val;
    }
}

impl store::Next for NextWindowUpdate {
    fn next(stream: &Stream) -> Option<store::Key> {
        stream.next_window_update
    }

    fn set_next(stream: &mut Stream, key: Option<store::Key>) {
        stream.next_window_update = key;
    }

    fn take_next(stream: &mut Stream) -> Option<store::Key> {
        stream.next_window_update.take()
    }

    fn is_queued(stream: &Stream) -> bool {
        stream.is_pending_window_update
    }

    fn set_queued(stream: &mut Stream, val: bool) {
        stream.is_pending_window_update = val;
    }
}

impl store::Next for NextOpen {
    fn next(stream: &Stream) -> Option<store::Key> {
        stream.next_open
    }

    fn set_next(stream: &mut Stream, key: Option<store::Key>) {
        stream.next_open = key;
    }

    fn take_next(stream: &mut Stream) -> Option<store::Key> {
        stream.next_open.take()
    }

    fn is_queued(stream: &Stream) -> bool {
        stream.is_pending_open
    }

    fn set_queued(stream: &mut Stream, val: bool) {
        if val {
            // ensure that stream is not queued for being sent
            // if it's being put into queue for opening the stream
            debug_assert!(!stream.is_pending_send);
        }
        stream.is_pending_open = val;
    }
}

impl store::Next for NextResetExpire {
    fn next(stream: &Stream) -> Option<store::Key> {
        stream.next_reset_expire
    }

    fn set_next(stream: &mut Stream, key: Option<store::Key>) {
        stream.next_reset_expire = key;
    }

    fn take_next(stream: &mut Stream) -> Option<store::Key> {
        stream.next_reset_expire.take()
    }

    fn is_queued(stream: &Stream) -> bool {
        stream.reset_at.is_some()
    }

    fn set_queued(stream: &mut Stream, val: bool) {
        if val {
            stream.reset_at = Some(Instant::now());
        } else {
            stream.reset_at = None;
        }
    }
}

// ===== impl ContentLength =====

impl ContentLength {
    pub fn is_head(&self) -> bool {
        matches!(*self, Self::Head)
    }
}

Coverage Report

Created: 2026-06-01 06:40

Line	Count	Source
1		use crate::Reason;
2
3		use super::*;
4
5		use std::fmt;
6		use std::task::{Context, Waker};
7		use std::time::Instant;
8
9		/// Tracks Stream related state
10		///
11		/// # Reference counting
12		///
13		/// There can be a number of outstanding handles to a single Stream. These are
14		/// tracked using reference counting. The `ref_count` field represents the
15		/// number of outstanding userspace handles that can reach this stream.
16		///
17		/// It's important to note that when the stream is placed in an internal queue
18		/// (such as an accept queue), this is not tracked by a reference count.
19		/// Thus, `ref_count` can be zero and the stream still has to be kept around.
20		pub(super) struct Stream {
21		/// The h2 stream identifier
22		pub id: StreamId,
23
24		/// Current state of the stream
25		pub state: State,
26
27		/// Set to `true` when the stream is counted against the connection's max
28		/// concurrent streams.
29		pub is_counted: bool,
30
31		/// Number of outstanding handles pointing to this stream
32		pub ref_count: usize,
33
34		// ===== Fields related to sending =====
35		/// Next node in the accept linked list
36		pub next_pending_send: Option<store::Key>,
37
38		/// Set to true when the stream is pending accept
39		pub is_pending_send: bool,
40
41		/// Send data flow control
42		pub send_flow: FlowControl,
43
44		/// Amount of send capacity that has been requested, but not yet allocated.
45		pub requested_send_capacity: WindowSize,
46
47		/// Amount of data buffered at the prioritization layer.
48		/// TODO: Technically this could be greater than the window size...
49		pub buffered_send_data: usize,
50
51		/// Task tracking additional send capacity (i.e. window updates).
52		send_task: Option<Waker>,
53
54		/// Frames pending for this stream being sent to the socket
55		pub pending_send: buffer::Deque,
56
57		/// Next node in the linked list of streams waiting for additional
58		/// connection level capacity.
59		pub next_pending_send_capacity: Option<store::Key>,
60
61		/// True if the stream is waiting for outbound connection capacity
62		pub is_pending_send_capacity: bool,
63
64		/// Set to true when the send capacity has been incremented
65		pub send_capacity_inc: bool,
66
67		/// Next node in the open linked list
68		pub next_open: Option<store::Key>,
69
70		/// Set to true when the stream is pending to be opened
71		pub is_pending_open: bool,
72
73		/// Set to true when a push is pending for this stream
74		pub is_pending_push: bool,
75
76		// ===== Fields related to receiving =====
77		/// Next node in the accept linked list
78		pub next_pending_accept: Option<store::Key>,
79
80		/// Set to true when the stream is pending accept
81		pub is_pending_accept: bool,
82
83		/// Receive data flow control
84		pub recv_flow: FlowControl,
85
86		pub in_flight_recv_data: WindowSize,
87
88		/// Next node in the linked list of streams waiting to send window updates.
89		pub next_window_update: Option<store::Key>,
90
91		/// True if the stream is waiting to send a window update
92		pub is_pending_window_update: bool,
93
94		/// The time when this stream may have been locally reset.
95		pub reset_at: Option<Instant>,
96
97		/// Next node in list of reset streams that should expire eventually
98		pub next_reset_expire: Option<store::Key>,
99
100		/// Frames pending for this stream to read
101		pub pending_recv: buffer::Deque,
102
103		/// When the RecvStream drop occurs, no data should be received.
104		pub is_recv: bool,
105
106		/// Task tracking receiving frames
107		pub recv_task: Option<Waker>,
108
109		/// Task tracking pushed promises.
110		pub push_task: Option<Waker>,
111
112		/// The stream's pending push promises
113		pub pending_push_promises: store::Queue<NextAccept>,
114
115		/// Validate content-length headers
116		pub content_length: ContentLength,
117		}
118
119		/// State related to validating a stream's content-length
120		#[derive(Debug)]
121		pub enum ContentLength {
122		Omitted,
123		Head,
124		Remaining(u64),
125		}
126
127		#[derive(Debug)]
128		pub(super) struct NextAccept;
129
130		#[derive(Debug)]
131		pub(super) struct NextSend;
132
133		#[derive(Debug)]
134		pub(super) struct NextSendCapacity;
135
136		#[derive(Debug)]
137		pub(super) struct NextWindowUpdate;
138
139		#[derive(Debug)]
140		pub(super) struct NextOpen;
141
142		#[derive(Debug)]
143		pub(super) struct NextResetExpire;
144
145		impl Stream {
146	515k	pub fn new(id: StreamId, init_send_window: WindowSize, init_recv_window: WindowSize) -> Stream {
147	515k	let mut send_flow = FlowControl::new();
148	515k	let mut recv_flow = FlowControl::new();
149
150	515k	recv_flow
151	515k	.inc_window(init_recv_window)
152	515k	.expect("invalid initial receive window");
153		// TODO: proper error handling?
154	515k	let _res = recv_flow.assign_capacity(init_recv_window);
155	515k	debug_assert!(_res.is_ok());
156
157	515k	send_flow
158	515k	.inc_window(init_send_window)
159	515k	.expect("invalid initial send window size");
160
161	515k	Stream {
162	515k	id,
163	515k	state: State::default(),
164	515k	ref_count: 0,
165	515k	is_counted: false,
166	515k
167	515k	// ===== Fields related to sending =====
168	515k	next_pending_send: None,
169	515k	is_pending_send: false,
170	515k	send_flow,
171	515k	requested_send_capacity: 0,
172	515k	buffered_send_data: 0,
173	515k	send_task: None,
174	515k	pending_send: buffer::Deque::new(),
175	515k	is_pending_send_capacity: false,
176	515k	next_pending_send_capacity: None,
177	515k	send_capacity_inc: false,
178	515k	is_pending_open: false,
179	515k	next_open: None,
180	515k	is_pending_push: false,
181	515k
182	515k	// ===== Fields related to receiving =====
183	515k	next_pending_accept: None,
184	515k	is_pending_accept: false,
185	515k	recv_flow,
186	515k	in_flight_recv_data: 0,
187	515k	next_window_update: None,
188	515k	is_pending_window_update: false,
189	515k	reset_at: None,
190	515k	next_reset_expire: None,
191	515k	pending_recv: buffer::Deque::new(),
192	515k	is_recv: true,
193	515k	recv_task: None,
194	515k	push_task: None,
195	515k	pending_push_promises: store::Queue::new(),
196	515k	content_length: ContentLength::Omitted,
197	515k	}
198	515k	}
199
200		/// Increment the stream's ref count
201	975k	pub fn ref_inc(&mut self) {
202	975k	assert!(self.ref_count < usize::MAX);
203	975k	self.ref_count += 1;
204	975k	}
205
206		/// Decrements the stream's ref count
207	975k	pub fn ref_dec(&mut self) {
208	975k	assert!(self.ref_count > 0);
209	975k	self.ref_count -= 1;
210	975k	}
211
212		/// Returns true if stream is currently being held for some time because of
213		/// a local reset.
214	4.69M	pub fn is_pending_reset_expiration(&self) -> bool {
215	4.69M	self.reset_at.is_some()
216	4.69M	}
217
218		/// Returns true if frames for this stream are ready to be sent over the wire
219	847k	pub fn is_send_ready(&self) -> bool {
220		// Why do we check pending_open?
221		//
222		// We allow users to call send_request() which schedules a stream to be pending_open
223		// if there is no room according to the concurrency limit (max_send_streams), and we
224		// also allow data to be buffered for send with send_data() if there is no capacity for
225		// the stream to send the data, which attempts to place the stream in pending_send.
226		// If the stream is not open, we don't want the stream to be scheduled for
227		// execution (pending_send). Note that if the stream is in pending_open, it will be
228		// pushed to pending_send when there is room for an open stream.
229		//
230		// In pending_push we track whether a PushPromise still needs to be sent
231		// from a different stream before we can start sending frames on this one.
232		// This is different from the "open" check because reserved streams don't count
233		// toward the concurrency limit.
234		// See https://httpwg.org/specs/rfc7540.html#rfc.section.5.1.2
235	847k	!self.is_pending_open && !self.is_pending_push
236	847k	}
237
238		/// Returns true if the stream is closed
239	6.46M	pub fn is_closed(&self) -> bool {
240		// The state has fully transitioned to closed.
241	6.46M	self.state.is_closed() &&
242		// Because outbound frames transition the stream state before being
243		// buffered, we have to ensure that all frames have been flushed.
244	4.01M	self.pending_send.is_empty() &&
245		// Sometimes large data frames are sent out in chunks. After a chunk
246		// of the frame is sent, the remainder is pushed back onto the send
247		// queue to be rescheduled.
248		//
249		// Checking for additional buffered data lets us catch this case.
250	3.63M	self.buffered_send_data == 0
251	6.46M	}
252
253		/// Returns true if the stream is no longer in use
254	2.98M	pub fn is_released(&self) -> bool {
255		// The stream is closed and fully flushed
256	2.98M	self.is_closed() &&
257		// There are no more outstanding references to the stream
258	1.61M	self.ref_count == 0 &&
259		// The stream is not in any queue
260	1.06M	!self.is_pending_send && !self.is_pending_send_capacity &&
261	760k	!self.is_pending_accept && !self.is_pending_window_update &&
262	759k	!self.is_pending_open && self.reset_at.is_none()
263	2.98M	}
264
265		/// Returns true when the consumer of the stream has dropped all handles
266		/// (indicating no further interest in the stream) and the stream state is
267		/// not actually closed.
268		///
269		/// In this case, a reset should be sent.
270	975k	pub fn is_canceled_interest(&self) -> bool {
271	975k	self.ref_count == 0 && !self.state.is_closed()
272	975k	}
273
274		/// Current available stream send capacity
275	175k	pub fn capacity(&self, max_buffer_size: usize) -> WindowSize {
276	175k	let available = self.send_flow.available().as_size() as usize;
277	175k	let buffered = self.buffered_send_data;
278
279	175k	available.min(max_buffer_size).saturating_sub(buffered) as WindowSize
280	175k	}
281
282	71.8k	pub fn assign_capacity(&mut self, capacity: WindowSize, max_buffer_size: usize) {
283	71.8k	let prev_capacity = self.capacity(max_buffer_size);
284	71.8k	debug_assert!(capacity > 0);
285		// TODO: proper error handling
286	71.8k	let _res = self.send_flow.assign_capacity(capacity);
287	71.8k	debug_assert!(_res.is_ok());
288
289	71.8k	tracing::trace!(
290	0	" assigned capacity to stream; available={}; buffered={}; id={:?}; max_buffer_size={} prev={}",
291	0	self.send_flow.available(),
292		self.buffered_send_data,
293		self.id,
294		max_buffer_size,
295		prev_capacity,
296		);
297
298	71.8k	if prev_capacity < self.capacity(max_buffer_size) {
299	0	self.notify_capacity();
300	71.8k	}
301	71.8k	}
302
303	15.9k	pub fn send_data(&mut self, len: WindowSize, max_buffer_size: usize) {
304	15.9k	let prev_capacity = self.capacity(max_buffer_size);
305
306		// TODO: proper error handling
307	15.9k	let _res = self.send_flow.send_data(len);
308	15.9k	debug_assert!(_res.is_ok());
309
310		// Decrement the stream's buffered data counter
311	15.9k	debug_assert!(self.buffered_send_data >= len as usize);
312	15.9k	self.buffered_send_data -= len as usize;
313	15.9k	self.requested_send_capacity -= len;
314
315	15.9k	tracing::trace!(
316	0	" sent stream data; available={}; buffered={}; id={:?}; max_buffer_size={} prev={}",
317	0	self.send_flow.available(),
318		self.buffered_send_data,
319		self.id,
320		max_buffer_size,
321		prev_capacity,
322		);
323
324	15.9k	if prev_capacity < self.capacity(max_buffer_size) {
325	0	self.notify_capacity();
326	15.9k	}
327	15.9k	}
328
329		/// If the capacity was limited because of the max_send_buffer_size,
330		/// then consider waking the send task again...
331	0	pub fn notify_capacity(&mut self) {
332	0	self.send_capacity_inc = true;
333	0	tracing::trace!(" notifying task");
334	0	self.notify_send();
335	0	}
336
337		/// Returns `Err` when the decrement cannot be completed due to overflow.
338	7	pub fn dec_content_length(&mut self, len: usize) -> Result<(), ()> {
339	7	match self.content_length {
340	0	ContentLength::Remaining(ref mut rem) => match rem.checked_sub(len as u64) {
341	0	Some(val) => *rem = val,
342	0	None => return Err(()),
343		},
344		ContentLength::Head => {
345	0	if len != 0 {
346	0	return Err(());
347	0	}
348		}
349	7	_ => {}
350		}
351
352	7	Ok(())
353	7	}
354
355	9	pub fn ensure_content_length_zero(&self) -> Result<(), ()> {
356	9	match self.content_length {
357	0	ContentLength::Remaining(0) => Ok(()),
358	0	ContentLength::Remaining(_) => Err(()),
359	9	_ => Ok(()),
360		}
361	9	}
362
363	892k	pub fn notify_send(&mut self) {
364	892k	if let Some(task) = self.send_task.take() {
365	220	task.wake();
366	892k	}
367	892k	}
368
369	1.69k	pub fn wait_send(&mut self, cx: &Context) {
370	1.69k	self.send_task = Some(cx.waker().clone());
371	1.69k	}
372
373	784k	pub fn notify_recv(&mut self) {
374	784k	if let Some(task) = self.recv_task.take() {
375	485k	task.wake();
376	485k	}
377	784k	}
378
379	732k	pub(super) fn notify_push(&mut self) {
380	732k	if let Some(task) = self.push_task.take() {
381	0	task.wake();
382	732k	}
383	732k	}
384
385		/// Set the stream's state to `Closed` with the given reason and initiator.
386		/// Notify the send, receive, and push tasks, if they exist.
387	94.3k	pub(super) fn set_reset(&mut self, reason: Reason, initiator: Initiator) {
388	94.3k	self.state.set_reset(self.id, reason, initiator);
389	94.3k	self.notify_send();
390	94.3k	self.notify_push();
391	94.3k	self.notify_recv();
392	94.3k	}
393		}
394
395		impl fmt::Debug for Stream {
396	0	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
397	0	f.debug_struct("Stream")
398	0	.field("id", &self.id)
399	0	.field("state", &self.state)
400	0	.field("is_counted", &self.is_counted)
401	0	.field("ref_count", &self.ref_count)
402	0	.h2_field_some("next_pending_send", &self.next_pending_send)
403	0	.h2_field_if("is_pending_send", &self.is_pending_send)
404	0	.field("send_flow", &self.send_flow)
405	0	.field("requested_send_capacity", &self.requested_send_capacity)
406	0	.field("buffered_send_data", &self.buffered_send_data)
407	0	.h2_field_some("send_task", &self.send_task.as_ref().map(\|_\| ()))
408	0	.h2_field_if_then(
409	0	"pending_send",
410	0	!self.pending_send.is_empty(),
411	0	&self.pending_send,
412		)
413	0	.h2_field_some(
414	0	"next_pending_send_capacity",
415	0	&self.next_pending_send_capacity,
416		)
417	0	.h2_field_if("is_pending_send_capacity", &self.is_pending_send_capacity)
418	0	.h2_field_if("send_capacity_inc", &self.send_capacity_inc)
419	0	.h2_field_some("next_open", &self.next_open)
420	0	.h2_field_if("is_pending_open", &self.is_pending_open)
421	0	.h2_field_if("is_pending_push", &self.is_pending_push)
422	0	.h2_field_some("next_pending_accept", &self.next_pending_accept)
423	0	.h2_field_if("is_pending_accept", &self.is_pending_accept)
424	0	.field("recv_flow", &self.recv_flow)
425	0	.field("in_flight_recv_data", &self.in_flight_recv_data)
426	0	.h2_field_some("next_window_update", &self.next_window_update)
427	0	.h2_field_if("is_pending_window_update", &self.is_pending_window_update)
428	0	.h2_field_some("reset_at", &self.reset_at)
429	0	.h2_field_some("next_reset_expire", &self.next_reset_expire)
430	0	.h2_field_if_then(
431	0	"pending_recv",
432	0	!self.pending_recv.is_empty(),
433	0	&self.pending_recv,
434		)
435	0	.h2_field_if("is_recv", &self.is_recv)
436	0	.h2_field_some("recv_task", &self.recv_task.as_ref().map(\|_\| ()))
437	0	.h2_field_some("push_task", &self.push_task.as_ref().map(\|_\| ()))
438	0	.h2_field_if_then(
439	0	"pending_push_promises",
440	0	!self.pending_push_promises.is_empty(),
441	0	&self.pending_push_promises,
442		)
443	0	.field("content_length", &self.content_length)
444	0	.finish()
445	0	}
446		}
447
448		impl store::Next for NextAccept {
449	141	fn next(stream: &Stream) -> Option<store::Key> {
450	141	stream.next_pending_accept
451	141	}
452
453	25	fn set_next(stream: &mut Stream, key: Option<store::Key>) {
454	25	stream.next_pending_accept = key;
455	25	}
456
457	22	fn take_next(stream: &mut Stream) -> Option<store::Key> {
458	22	stream.next_pending_accept.take()
459	22	}
460
461	169	fn is_queued(stream: &Stream) -> bool {
462	169	stream.is_pending_accept
463	169	}
464
465	332	fn set_queued(stream: &mut Stream, val: bool) {
466	332	stream.is_pending_accept = val;
467	332	}
468		}
469
470		impl store::Next for NextSend {
471	6.40k	fn next(stream: &Stream) -> Option<store::Key> {
472	6.40k	stream.next_pending_send
473	6.40k	}
474
475	363k	fn set_next(stream: &mut Stream, key: Option<store::Key>) {
476	363k	stream.next_pending_send = key;
477	363k	}
478
479	363k	fn take_next(stream: &mut Stream) -> Option<store::Key> {
480	363k	stream.next_pending_send.take()
481	363k	}
482
483	675k	fn is_queued(stream: &Stream) -> bool {
484	675k	stream.is_pending_send
485	675k	}
486
487	740k	fn set_queued(stream: &mut Stream, val: bool) {
488	740k	if val {
489		// ensure that stream is not queued for being opened
490		// if it's being put into queue for sending data
491	370k	debug_assert!(!stream.is_pending_open);
492	370k	}
493	740k	stream.is_pending_send = val;
494	740k	}
495		}
496
497		impl store::Next for NextSendCapacity {
498	1.73k	fn next(stream: &Stream) -> Option<store::Key> {
499	1.73k	stream.next_pending_send_capacity
500	1.73k	}
501
502	177k	fn set_next(stream: &mut Stream, key: Option<store::Key>) {
503	177k	stream.next_pending_send_capacity = key;
504	177k	}
505
506	177k	fn take_next(stream: &mut Stream) -> Option<store::Key> {
507	177k	stream.next_pending_send_capacity.take()
508	177k	}
509
510	205k	fn is_queued(stream: &Stream) -> bool {
511	205k	stream.is_pending_send_capacity
512	205k	}
513
514	359k	fn set_queued(stream: &mut Stream, val: bool) {
515	359k	stream.is_pending_send_capacity = val;
516	359k	}
517		}
518
519		impl store::Next for NextWindowUpdate {
520	0	fn next(stream: &Stream) -> Option<store::Key> {
521	0	stream.next_window_update
522	0	}
523
524	0	fn set_next(stream: &mut Stream, key: Option<store::Key>) {
525	0	stream.next_window_update = key;
526	0	}
527
528	0	fn take_next(stream: &mut Stream) -> Option<store::Key> {
529	0	stream.next_window_update.take()
530	0	}
531
532	0	fn is_queued(stream: &Stream) -> bool {
533	0	stream.is_pending_window_update
534	0	}
535
536	0	fn set_queued(stream: &mut Stream, val: bool) {
537	0	stream.is_pending_window_update = val;
538	0	}
539		}
540
541		impl store::Next for NextOpen {
542	4.77k	fn next(stream: &Stream) -> Option<store::Key> {
543	4.77k	stream.next_open
544	4.77k	}
545
546	482k	fn set_next(stream: &mut Stream, key: Option<store::Key>) {
547	482k	stream.next_open = key;
548	482k	}
549
550	482k	fn take_next(stream: &mut Stream) -> Option<store::Key> {
551	482k	stream.next_open.take()
552	482k	}
553
554	487k	fn is_queued(stream: &Stream) -> bool {
555	487k	stream.is_pending_open
556	487k	}
557
558	975k	fn set_queued(stream: &mut Stream, val: bool) {
559	975k	if val {
560		// ensure that stream is not queued for being sent
561		// if it's being put into queue for opening the stream
562	487k	debug_assert!(!stream.is_pending_send);
563	487k	}
564	975k	stream.is_pending_open = val;
565	975k	}
566		}
567
568		impl store::Next for NextResetExpire {
569	5.02k	fn next(stream: &Stream) -> Option<store::Key> {
570	5.02k	stream.next_reset_expire
571	5.02k	}
572
573	44.5k	fn set_next(stream: &mut Stream, key: Option<store::Key>) {
574	44.5k	stream.next_reset_expire = key;
575	44.5k	}
576
577	44.5k	fn take_next(stream: &mut Stream) -> Option<store::Key> {
578	44.5k	stream.next_reset_expire.take()
579	44.5k	}
580
581	49.5k	fn is_queued(stream: &Stream) -> bool {
582	49.5k	stream.reset_at.is_some()
583	49.5k	}
584
585	99.1k	fn set_queued(stream: &mut Stream, val: bool) {
586	99.1k	if val {
587	49.5k	stream.reset_at = Some(Instant::now());
588	49.5k	} else {
589	49.5k	stream.reset_at = None;
590	49.5k	}
591	99.1k	}
592		}
593
594		// ===== impl ContentLength =====
595
596		impl ContentLength {
597	282	pub fn is_head(&self) -> bool {
598	282	matches!(*self, Self::Head)
599	282	}
600		}