h2/proto/streams/
streams.rs

1use super::recv::RecvHeaderBlockError;
2use super::store::{self, Entry, Resolve, Store};
3use super::{Buffer, BufferStatus, Config, Counts, Prioritized, Recv, Send, Stream, StreamId};
4use crate::codec::{Codec, SendError, UserError};
5use crate::ext::Protocol;
6use crate::frame::{self, Frame, Reason};
7use crate::proto::{peer, Error, Initiator, Open, Peer, WindowSize};
8use crate::{client, proto, server};
9
10use bytes::{Buf, Bytes};
11use http::{HeaderMap, Request, Response};
12use std::task::{Context, Poll, Waker};
13use tokio::io::AsyncWrite;
14
15use std::sync::{Arc, Mutex};
16use std::{fmt, io};
17
18#[derive(Debug)]
19pub(crate) struct Streams<B, P>
20where
21    P: Peer,
22{
23    /// Holds most of the connection and stream related state for processing
24    /// HTTP/2 frames associated with streams.
25    inner: Arc<Mutex<Inner>>,
26
27    /// This is the queue of frames to be written to the wire. This is split out
28    /// to avoid requiring a `B` generic on all public API types even if `B` is
29    /// not technically required.
30    ///
31    /// Currently, splitting this out requires a second `Arc` + `Mutex`.
32    /// However, it should be possible to avoid this duplication with a little
33    /// bit of unsafe code. This optimization has been postponed until it has
34    /// been shown to be necessary.
35    send_buffer: Arc<SendBuffer<B>>,
36
37    _p: ::std::marker::PhantomData<P>,
38}
39
40// Like `Streams` but with a `peer::Dyn` field instead of a static `P: Peer` type parameter.
41// Ensures that the methods only get one instantiation, instead of two (client and server)
42#[derive(Debug)]
43pub(crate) struct DynStreams<'a, B> {
44    inner: &'a Mutex<Inner>,
45
46    send_buffer: &'a SendBuffer<B>,
47
48    peer: peer::Dyn,
49}
50
51/// Outcome of receiving a frame that can open a stream.
52#[derive(Debug)]
53#[must_use = "a refused stream must be reset by the connection"]
54pub(crate) enum RecvOutcome {
55    /// The frame was handled, including any intentional ignoring of it.
56    Processed,
57    /// The connection must send REFUSED_STREAM for this stream ID.
58    Refused(StreamId),
59}
60
61/// Reference to the stream state
62#[derive(Debug)]
63pub(crate) struct StreamRef<B> {
64    opaque: OpaqueStreamRef,
65    send_buffer: Arc<SendBuffer<B>>,
66}
67
68/// Reference to the stream state that hides the send data chunk generic
69pub(crate) struct OpaqueStreamRef {
70    inner: Arc<Mutex<Inner>>,
71    key: store::Key,
72}
73
74/// Fields needed to manage state related to managing the set of streams. This
75/// is mostly split out to make ownership happy.
76///
77/// TODO: better name
78#[derive(Debug)]
79struct Inner {
80    /// Tracks send & recv stream concurrency.
81    counts: Counts,
82
83    /// Connection level state and performs actions on streams
84    actions: Actions,
85
86    /// Stores stream state
87    store: Store,
88
89    /// The number of stream refs to this shared state.
90    refs: usize,
91}
92
93#[derive(Debug)]
94struct Actions {
95    /// Manages state transitions initiated by receiving frames
96    recv: Recv,
97
98    /// Manages state transitions initiated by sending frames
99    send: Send,
100
101    /// Task that calls `poll_complete`.
102    task: Option<Waker>,
103
104    /// If the connection errors, a copy is kept for any StreamRefs.
105    conn_error: Option<proto::Error>,
106}
107
108/// Contains the buffer of frames to be written to the wire.
109#[derive(Debug)]
110struct SendBuffer<B> {
111    inner: Mutex<Buffer<Frame<B>>>,
112}
113
114// ===== impl Streams =====
115
116impl<B, P> Streams<B, P>
117where
118    B: Buf,
119    P: Peer,
120{
121    pub fn new(config: Config) -> Self {
122        let peer = P::r#dyn();
123
124        Streams {
125            inner: Inner::new(peer, config),
126            send_buffer: Arc::new(SendBuffer::new()),
127            _p: ::std::marker::PhantomData,
128        }
129    }
130
131    pub fn set_target_connection_window_size(&mut self, size: WindowSize) -> Result<(), Reason> {
132        let mut me = self.inner.lock().unwrap();
133        let me = &mut *me;
134
135        me.actions
136            .recv
137            .set_target_connection_window(size, &mut me.actions.task)?;
138        me.counts.set_target_connection_window_size(size);
139        Ok(())
140    }
141
142    pub fn next_incoming(&mut self) -> Option<StreamRef<B>> {
143        let mut me = self.inner.lock().unwrap();
144        let me = &mut *me;
145        me.actions.recv.next_incoming(&mut me.store).map(|key| {
146            let stream = &mut me.store.resolve(key);
147            tracing::trace!(
148                "next_incoming; id={:?}, state={:?}",
149                stream.id,
150                stream.state
151            );
152            // TODO: ideally, OpaqueStreamRefs::new would do this, but we're holding
153            // the lock, so it can't.
154            me.refs += 1;
155
156            // Pending-accepted remotely-reset streams are counted.
157            if stream.state.is_remote_reset() {
158                me.counts.dec_num_remote_reset_streams();
159            }
160
161            StreamRef {
162                opaque: OpaqueStreamRef::new(self.inner.clone(), stream),
163                send_buffer: self.send_buffer.clone(),
164            }
165        })
166    }
167
168    /// Count a refused stream toward the local error reset limit.
169    pub fn count_pending_refusal(&mut self) -> Result<(), Error> {
170        let mut me = self.inner.lock().unwrap();
171        if !me.counts.can_inc_num_local_error_resets() {
172            return Err(Error::library_go_away_data(
173                Reason::ENHANCE_YOUR_CALM,
174                "too_many_internal_resets",
175            ));
176        }
177        me.counts.inc_num_local_error_resets();
178        Ok(())
179    }
180
181    pub fn clear_expired_reset_streams(&mut self) {
182        let mut me = self.inner.lock().unwrap();
183        let me = &mut *me;
184        me.actions
185            .recv
186            .clear_expired_reset_streams(&mut me.store, &mut me.counts);
187    }
188
189    pub fn poll_complete<T>(
190        &mut self,
191        cx: &mut Context,
192        dst: &mut Codec<T, Prioritized<B>>,
193    ) -> Poll<io::Result<()>>
194    where
195        T: AsyncWrite + Unpin,
196    {
197        loop {
198            // Make any required socket progress before taking stream locks.
199            ready!(dst.poll_ready(cx))?;
200
201            let status = {
202                let mut me = self.inner.lock().unwrap();
203                let status = me.buffer_pending(&self.send_buffer, dst)?;
204
205                // Register the task while holding the same lock used to
206                // observe that all pending frames have been buffered. A
207                // producer that queues another frame while the codec is being
208                // flushed will then take and wake this task.
209                if status == BufferStatus::Complete {
210                    me.actions.task = Some(cx.waker().clone());
211                }
212
213                status
214            };
215
216            match status {
217                BufferStatus::Complete => {}
218                BufferStatus::CodecFull => continue,
219            }
220
221            // Flush any frames staged by `buffer_pending` without holding the
222            // stream-state or send-buffer mutexes.
223            ready!(dst.flush(cx))?;
224
225            let reclaimed = {
226                let mut me = self.inner.lock().unwrap();
227                me.reclaim_written_frame(&self.send_buffer, dst)
228            };
229
230            if !reclaimed {
231                return Poll::Ready(Ok(()));
232            }
233        }
234    }
235
236    pub fn apply_remote_settings(
237        &mut self,
238        frame: &frame::Settings,
239        is_initial: bool,
240    ) -> Result<(), Error> {
241        let mut me = self.inner.lock().unwrap();
242        let me = &mut *me;
243
244        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
245        let send_buffer = &mut *send_buffer;
246
247        me.counts.apply_remote_settings(frame, is_initial);
248
249        me.actions.send.apply_remote_settings(
250            frame,
251            send_buffer,
252            &mut me.store,
253            &mut me.counts,
254            &mut me.actions.task,
255        )
256    }
257
258    pub fn apply_local_settings(&mut self, frame: &frame::Settings) -> Result<(), Error> {
259        let mut me = self.inner.lock().unwrap();
260        let me = &mut *me;
261
262        me.actions.recv.apply_local_settings(frame, &mut me.store)
263    }
264
265    pub fn send_request(
266        &mut self,
267        mut request: Request<()>,
268        end_of_stream: bool,
269        pending: Option<&OpaqueStreamRef>,
270    ) -> Result<(StreamRef<B>, bool), SendError> {
271        use super::stream::ContentLength;
272        use http::Method;
273
274        let protocol = request.extensions_mut().remove::<Protocol>();
275
276        // Clear before taking lock, incase extensions contain a StreamRef.
277        request.extensions_mut().clear();
278
279        // TODO: There is a hazard with assigning a stream ID before the
280        // prioritize layer. If prioritization reorders new streams, this
281        // implicitly closes the earlier stream IDs.
282        //
283        // See: hyperium/h2#11
284        let mut me = self.inner.lock().unwrap();
285        let me = &mut *me;
286
287        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
288        let send_buffer = &mut *send_buffer;
289
290        me.actions.ensure_no_conn_error()?;
291        me.actions.send.ensure_next_stream_id()?;
292
293        // The `pending` argument is provided by the `Client`, and holds
294        // a store `Key` of a `Stream` that may have been not been opened
295        // yet.
296        //
297        // If that stream is still pending, the Client isn't allowed to
298        // queue up another pending stream. They should use `poll_ready`.
299        if let Some(stream) = pending {
300            if me.store.resolve(stream.key).is_pending_open {
301                return Err(UserError::Rejected.into());
302            }
303        }
304
305        if me.counts.peer().is_server() {
306            // Servers cannot open streams. PushPromise must first be reserved.
307            return Err(UserError::UnexpectedFrameType.into());
308        }
309
310        let stream_id = me.actions.send.open()?;
311
312        let mut stream = Stream::new(
313            stream_id,
314            me.actions.send.init_window_sz(),
315            me.actions.recv.init_window_sz(),
316        );
317
318        if *request.method() == Method::HEAD {
319            stream.content_length = ContentLength::Head;
320        }
321
322        // Convert the message
323        let headers =
324            client::Peer::convert_send_message(stream_id, request, protocol, end_of_stream)?;
325
326        let mut stream = me.store.insert(stream.id, stream);
327
328        let sent = me.actions.send.send_headers(
329            headers,
330            send_buffer,
331            &mut stream,
332            &mut me.counts,
333            &mut me.actions.task,
334        );
335
336        // send_headers can return a UserError, if it does,
337        // we should forget about this stream.
338        if let Err(err) = sent {
339            stream.unlink();
340            stream.remove();
341            return Err(err.into());
342        }
343
344        // Given that the stream has been initialized, it should not be in the
345        // closed state.
346        debug_assert!(!stream.state.is_closed());
347
348        // TODO: ideally, OpaqueStreamRefs::new would do this, but we're holding
349        // the lock, so it can't.
350        me.refs += 1;
351
352        let is_full = me.counts.next_send_stream_will_reach_capacity();
353        Ok((
354            StreamRef {
355                opaque: OpaqueStreamRef::new(self.inner.clone(), &mut stream),
356                send_buffer: self.send_buffer.clone(),
357            },
358            is_full,
359        ))
360    }
361
362    pub(crate) fn is_extended_connect_protocol_enabled(&self) -> bool {
363        self.inner
364            .lock()
365            .unwrap()
366            .actions
367            .send
368            .is_extended_connect_protocol_enabled()
369    }
370
371    pub fn current_max_send_streams(&self) -> usize {
372        let me = self.inner.lock().unwrap();
373        me.counts.max_send_streams()
374    }
375
376    pub fn current_max_recv_streams(&self) -> usize {
377        let me = self.inner.lock().unwrap();
378        me.counts.max_recv_streams()
379    }
380}
381
382impl<B> DynStreams<'_, B> {
383    pub fn is_buffer_empty(&self) -> bool {
384        self.send_buffer.is_empty()
385    }
386
387    pub fn is_server(&self) -> bool {
388        self.peer.is_server()
389    }
390
391    /// Reports whether the connection must refuse the stream opened by the peer.
392    pub fn recv_headers(&mut self, frame: frame::Headers) -> Result<RecvOutcome, Error> {
393        let mut me = self.inner.lock().unwrap();
394
395        me.recv_headers(self.peer, self.send_buffer, frame)
396    }
397
398    pub fn recv_data(&mut self, frame: frame::Data) -> Result<(), Error> {
399        let mut me = self.inner.lock().unwrap();
400        me.recv_data(self.peer, self.send_buffer, frame)
401    }
402
403    pub fn recv_reset(&mut self, frame: frame::Reset) -> Result<(), Error> {
404        let mut me = self.inner.lock().unwrap();
405
406        me.recv_reset(self.send_buffer, frame)
407    }
408
409    /// Notify all streams that a connection-level error happened.
410    pub fn handle_error(&mut self, err: proto::Error) -> StreamId {
411        let mut me = self.inner.lock().unwrap();
412        me.handle_error(self.send_buffer, err)
413    }
414
415    pub fn recv_go_away(&mut self, frame: &frame::GoAway) -> Result<(), Error> {
416        let mut me = self.inner.lock().unwrap();
417        me.recv_go_away(self.send_buffer, frame)
418    }
419
420    pub fn last_processed_id(&self) -> StreamId {
421        self.inner.lock().unwrap().actions.recv.last_processed_id()
422    }
423
424    pub fn recv_window_update(&mut self, frame: frame::WindowUpdate) -> Result<(), Error> {
425        let mut me = self.inner.lock().unwrap();
426        me.recv_window_update(self.send_buffer, frame)
427    }
428
429    /// Reports whether the connection must refuse the promised stream.
430    pub fn recv_push_promise(&mut self, frame: frame::PushPromise) -> Result<RecvOutcome, Error> {
431        let mut me = self.inner.lock().unwrap();
432        me.recv_push_promise(self.send_buffer, frame)
433    }
434
435    pub fn recv_eof(&mut self, clear_pending_accept: bool) -> Result<(), ()> {
436        let mut me = self.inner.lock().map_err(|_| ())?;
437        me.recv_eof(self.send_buffer, clear_pending_accept)
438    }
439
440    pub fn send_reset(
441        &mut self,
442        id: StreamId,
443        reason: Reason,
444    ) -> Result<(), crate::proto::error::GoAway> {
445        let mut me = self.inner.lock().unwrap();
446        me.send_reset(self.send_buffer, id, reason)
447    }
448
449    pub fn send_go_away(&mut self, last_processed_id: StreamId) {
450        let mut me = self.inner.lock().unwrap();
451        me.actions.recv.go_away(last_processed_id);
452    }
453}
454
455impl Inner {
456    fn new(peer: peer::Dyn, config: Config) -> Arc<Mutex<Self>> {
457        Arc::new(Mutex::new(Inner {
458            counts: Counts::new(peer, &config),
459            actions: Actions {
460                recv: Recv::new(peer, &config),
461                send: Send::new(&config),
462                task: None,
463                conn_error: None,
464            },
465            store: Store::new(),
466            refs: 1,
467        }))
468    }
469
470    fn recv_headers<B>(
471        &mut self,
472        peer: peer::Dyn,
473        send_buffer: &SendBuffer<B>,
474        frame: frame::Headers,
475    ) -> Result<RecvOutcome, Error> {
476        let id = frame.stream_id();
477
478        // The GOAWAY process has begun. All streams with a greater ID than
479        // specified as part of GOAWAY should be ignored.
480        if id > self.actions.recv.max_stream_id() {
481            tracing::trace!(
482                "id ({:?}) > max_stream_id ({:?}), ignoring HEADERS",
483                id,
484                self.actions.recv.max_stream_id()
485            );
486            return Ok(RecvOutcome::Processed);
487        }
488
489        let key = match self.store.find_entry(id) {
490            Entry::Occupied(e) => e.key(),
491            Entry::Vacant(e) => {
492                // Client: it's possible to send a request, and then send
493                // a RST_STREAM while the response HEADERS were in transit.
494                //
495                // Server: we can't reset a stream before having received
496                // the request headers, so don't allow.
497                if !peer.is_server() {
498                    // This may be response headers for a stream we've already
499                    // forgotten about...
500                    if self.actions.may_have_forgotten_stream(peer, id) {
501                        tracing::debug!(
502                            "recv_headers for old stream={:?}, sending STREAM_CLOSED",
503                            id,
504                        );
505                        return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
506                    }
507                }
508
509                match self
510                    .actions
511                    .recv
512                    .open(id, Open::Headers, &mut self.counts)?
513                {
514                    Some(stream_id) => {
515                        let stream = Stream::new(
516                            stream_id,
517                            self.actions.send.init_window_sz(),
518                            self.actions.recv.init_window_sz(),
519                        );
520
521                        e.insert(stream)
522                    }
523                    None => return Ok(RecvOutcome::Refused(id)),
524                }
525            }
526        };
527
528        let stream = self.store.resolve(key);
529
530        if stream.is_pending_open {
531            proto_err!(conn: "recv_headers: received frame on idle stream {:?}", id);
532            return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
533        }
534
535        if stream.state.is_local_error() {
536            // Locally reset streams must ignore frames "for some time".
537            // This is because the remote may have sent trailers before
538            // receiving the RST_STREAM frame.
539            tracing::trace!("recv_headers; ignoring trailers on {:?}", stream.id);
540            return Ok(RecvOutcome::Processed);
541        }
542
543        let actions = &mut self.actions;
544
545        self.counts.transition(stream, |counts, stream| {
546            tracing::trace!(
547                "recv_headers; stream={:?}; state={:?}",
548                stream.id,
549                stream.state
550            );
551
552            let res = if stream.state.is_recv_headers() {
553                match actions.recv.recv_headers(frame, stream, counts) {
554                    Ok(()) => Ok(()),
555                    Err(RecvHeaderBlockError::Oversize) => {
556                        if peer.is_server() {
557                            let mut resp = frame::Headers::new(
558                                stream.id,
559                                frame::Pseudo::response(
560                                    ::http::StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE,
561                                ),
562                                HeaderMap::new(),
563                            );
564                            resp.set_end_stream();
565                            let mut send_buffer = send_buffer.inner.lock().unwrap();
566                            let sent = actions.send.send_headers(
567                                resp, &mut send_buffer, stream, counts, &mut actions.task);
568                            debug_assert!(sent.is_ok(), "oversize response should not fail");
569
570                            actions.send.schedule_implicit_reset(
571                                stream,
572                                Reason::PROTOCOL_ERROR,
573                                counts,
574                                &mut actions.task);
575
576                            actions.recv.enqueue_reset_expiration(stream, counts);
577
578                            Ok(())
579                        } else {
580                            Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR))
581                        }
582                    },
583                    Err(RecvHeaderBlockError::State(err)) => Err(err),
584                }
585            } else {
586                if !frame.is_end_stream() {
587                    // Receiving trailers that don't set EOS is a "malformed"
588                    // message. Malformed messages are a stream error.
589                    proto_err!(stream: "recv_headers: trailers frame was not EOS; stream={:?}", stream.id);
590                    return Err(Error::library_reset(stream.id, Reason::PROTOCOL_ERROR));
591                }
592
593                actions.recv.recv_trailers(frame, stream)
594            };
595
596            actions.reset_on_recv_stream_err_deferred(send_buffer, stream, counts, res)
597        })?;
598
599        Ok(RecvOutcome::Processed)
600    }
601
602    fn recv_data<B>(
603        &mut self,
604        peer: peer::Dyn,
605        send_buffer: &SendBuffer<B>,
606        frame: frame::Data,
607    ) -> Result<(), Error> {
608        let id = frame.stream_id();
609
610        let stream = match self.store.find_mut(&id) {
611            Some(stream) => stream,
612            None => {
613                // The GOAWAY process has begun. All streams with a greater ID
614                // than specified as part of GOAWAY should be ignored.
615                if id > self.actions.recv.max_stream_id() {
616                    tracing::trace!(
617                        "id ({:?}) > max_stream_id ({:?}), ignoring DATA",
618                        id,
619                        self.actions.recv.max_stream_id()
620                    );
621
622                    // We still need to account for connection-level flow control.
623                    let sz = frame.flow_controlled_len();
624                    assert!(sz <= super::MAX_WINDOW_SIZE as usize);
625                    let sz = sz as WindowSize;
626                    self.actions.recv.ignore_data(sz)?;
627
628                    return Ok(());
629                }
630
631                if self.actions.may_have_forgotten_stream(peer, id) {
632                    tracing::debug!("recv_data for old stream={:?}, sending STREAM_CLOSED", id,);
633
634                    let sz = frame.flow_controlled_len();
635                    // This should have been enforced at the codec::FramedRead layer, so
636                    // this is just a sanity check.
637                    assert!(sz <= super::MAX_WINDOW_SIZE as usize);
638                    let sz = sz as WindowSize;
639                    self.actions.recv.ignore_data(sz)?;
640
641                    return Err(Error::library_reset(id, Reason::STREAM_CLOSED));
642                }
643
644                proto_err!(conn: "recv_data: stream not found; id={:?}", id);
645                return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
646            }
647        };
648
649        let actions = &mut self.actions;
650
651        self.counts.transition(stream, |counts, stream| {
652            let sz = frame.flow_controlled_len();
653            let is_end_stream = frame.is_end_stream();
654            let payload_len = frame.payload().len();
655            let mut res = actions.recv.recv_data(frame, stream);
656            // A stream can receive at most one final DATA frame, so it cannot
657            // be used to create unbounded framing overhead on that stream.
658            if res.is_ok() && !is_end_stream {
659                res = counts.record_data_frame(payload_len).map_err(|_| {
660                    tracing::debug!("too many small DATA frames");
661                    Error::library_go_away_data(Reason::ENHANCE_YOUR_CALM, "too_many_data_frames")
662                });
663            }
664
665            // Any stream error after receiving a DATA frame means
666            // we won't give the data to the user, and so they can't
667            // release the capacity. We do it automatically.
668            if let Err(Error::Reset(..)) = res {
669                actions
670                    .recv
671                    .release_connection_capacity(sz as WindowSize, &mut None);
672            }
673            actions.reset_on_recv_stream_err_deferred(send_buffer, stream, counts, res)
674        })
675    }
676
677    fn recv_reset<B>(
678        &mut self,
679        send_buffer: &SendBuffer<B>,
680        frame: frame::Reset,
681    ) -> Result<(), Error> {
682        let id = frame.stream_id();
683
684        if id.is_zero() {
685            proto_err!(conn: "recv_reset: invalid stream ID 0");
686            return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
687        }
688
689        // The GOAWAY process has begun. All streams with a greater ID than
690        // specified as part of GOAWAY should be ignored.
691        if id > self.actions.recv.max_stream_id() {
692            tracing::trace!(
693                "id ({:?}) > max_stream_id ({:?}), ignoring RST_STREAM",
694                id,
695                self.actions.recv.max_stream_id()
696            );
697            return Ok(());
698        }
699
700        let stream = match self.store.find_mut(&id) {
701            Some(stream) => stream,
702            None => {
703                // TODO: Are there other error cases?
704                self.actions
705                    .ensure_not_idle(self.counts.peer(), id)
706                    .map_err(Error::library_go_away)?;
707
708                return Ok(());
709            }
710        };
711
712        if stream.is_pending_open {
713            proto_err!(conn: "recv_reset: received frame on idle stream {:?}", id);
714            return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
715        }
716
717        let mut send_buffer = send_buffer.inner.lock().unwrap();
718        let send_buffer = &mut *send_buffer;
719
720        let actions = &mut self.actions;
721
722        self.counts.transition(stream, |counts, stream| {
723            actions.recv.recv_reset(frame, stream, counts)?;
724            actions.send.handle_error(send_buffer, stream, counts);
725            assert!(stream.state.is_closed());
726            Ok(())
727        })
728    }
729
730    fn recv_window_update<B>(
731        &mut self,
732        send_buffer: &SendBuffer<B>,
733        frame: frame::WindowUpdate,
734    ) -> Result<(), Error> {
735        let id = frame.stream_id();
736
737        let mut send_buffer = send_buffer.inner.lock().unwrap();
738        let send_buffer = &mut *send_buffer;
739
740        if id.is_zero() {
741            self.actions
742                .send
743                .recv_connection_window_update(frame, &mut self.store, &mut self.counts)
744                .map_err(Error::library_go_away)?;
745        } else {
746            // The remote may send window updates for streams that the local now
747            // considers closed. It's ok...
748            if let Some(mut stream) = self.store.find_mut(&id) {
749                if stream.is_pending_open {
750                    proto_err!(conn: "recv_window_update: received frame on idle stream {:?}", id);
751                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
752                }
753
754                let res = self
755                    .actions
756                    .send
757                    .recv_stream_window_update(
758                        frame.size_increment(),
759                        send_buffer,
760                        &mut stream,
761                        &mut self.counts,
762                        &mut self.actions.task,
763                    )
764                    .map_err(|reason| Error::library_reset(id, reason));
765
766                return self.actions.reset_on_recv_stream_err(
767                    send_buffer,
768                    &mut stream,
769                    &mut self.counts,
770                    res,
771                );
772            } else {
773                self.actions
774                    .ensure_not_idle(self.counts.peer(), id)
775                    .map_err(Error::library_go_away)?;
776            }
777        }
778
779        Ok(())
780    }
781
782    fn handle_error<B>(&mut self, send_buffer: &SendBuffer<B>, err: proto::Error) -> StreamId {
783        let actions = &mut self.actions;
784        let counts = &mut self.counts;
785        let mut send_buffer = send_buffer.inner.lock().unwrap();
786        let send_buffer = &mut *send_buffer;
787
788        let last_processed_id = actions.recv.last_processed_id();
789
790        self.store.for_each(|stream| {
791            counts.transition(stream, |counts, stream| {
792                actions.recv.handle_error(&err, &mut *stream);
793                actions.send.handle_error(send_buffer, stream, counts);
794            })
795        });
796
797        actions.conn_error = Some(err);
798
799        last_processed_id
800    }
801
802    fn recv_go_away<B>(
803        &mut self,
804        send_buffer: &SendBuffer<B>,
805        frame: &frame::GoAway,
806    ) -> Result<(), Error> {
807        let actions = &mut self.actions;
808        let counts = &mut self.counts;
809        let mut send_buffer = send_buffer.inner.lock().unwrap();
810        let send_buffer = &mut *send_buffer;
811
812        let last_stream_id = frame.last_stream_id();
813
814        actions.send.recv_go_away(last_stream_id)?;
815
816        let err = Error::remote_go_away(frame.debug_data().clone(), frame.reason());
817
818        let peer = counts.peer();
819        self.store.for_each(|stream| {
820            if stream.id > last_stream_id && peer.is_local_init(stream.id) {
821                counts.transition(stream, |counts, stream| {
822                    actions.recv.handle_error(&err, &mut *stream);
823                    actions.send.handle_error(send_buffer, stream, counts);
824                })
825            }
826        });
827
828        actions.conn_error = Some(err);
829
830        Ok(())
831    }
832
833    fn recv_push_promise<B>(
834        &mut self,
835        send_buffer: &SendBuffer<B>,
836        frame: frame::PushPromise,
837    ) -> Result<RecvOutcome, Error> {
838        let id = frame.stream_id();
839        let promised_id = frame.promised_id();
840
841        // First, ensure that the initiating stream is still in a valid state.
842        let parent_key = match self.store.find_mut(&id) {
843            Some(stream) => {
844                // The GOAWAY process has begun. All streams with a greater ID
845                // than specified as part of GOAWAY should be ignored.
846                if id > self.actions.recv.max_stream_id() {
847                    tracing::trace!(
848                        "id ({:?}) > max_stream_id ({:?}), ignoring PUSH_PROMISE",
849                        id,
850                        self.actions.recv.max_stream_id()
851                    );
852                    return Ok(RecvOutcome::Processed);
853                }
854
855                // The stream must be receive open
856                if !stream.state.ensure_recv_open()? {
857                    proto_err!(conn: "recv_push_promise: initiating stream is not opened");
858                    return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
859                }
860
861                stream.key()
862            }
863            None => {
864                proto_err!(conn: "recv_push_promise: initiating stream is in an invalid state");
865                return Err(Error::library_go_away(Reason::PROTOCOL_ERROR));
866            }
867        };
868
869        // Ensure that we can reserve streams
870        self.actions.recv.ensure_can_reserve()?;
871
872        // Next, open the stream.
873        //
874        // If `None` is returned, then the stream is being refused. There is no
875        // further work to be done.
876        if self
877            .actions
878            .recv
879            .open(promised_id, Open::PushPromise, &mut self.counts)?
880            .is_none()
881        {
882            return Ok(RecvOutcome::Refused(promised_id));
883        }
884
885        // Try to handle the frame and create a corresponding key for the pushed stream
886        // this requires a bit of indirection to make the borrow checker happy.
887        let child_key: Option<store::Key> = {
888            // Create state for the stream
889            let stream = self.store.insert(promised_id, {
890                Stream::new(
891                    promised_id,
892                    self.actions.send.init_window_sz(),
893                    self.actions.recv.init_window_sz(),
894                )
895            });
896
897            let actions = &mut self.actions;
898
899            self.counts.transition(stream, |counts, stream| {
900                // RFC 9113 excludes reserved streams from the concurrent stream
901                // count. Still, account reserved remote streams against the local
902                // receive limit so a peer cannot create an unbounded number of
903                // them by withholding the response HEADERS. `recv_headers` will
904                // see `is_counted` and avoid counting the stream a second time.
905                counts.inc_num_recv_streams(stream);
906                let stream_valid = actions.recv.recv_push_promise(frame, stream);
907
908                match stream_valid {
909                    Ok(()) => Ok(Some(stream.key())),
910                    _ => {
911                        let mut send_buffer = send_buffer.inner.lock().unwrap();
912                        actions
913                            .reset_on_recv_stream_err(
914                                &mut *send_buffer,
915                                stream,
916                                counts,
917                                stream_valid,
918                            )
919                            .map(|()| None)
920                    }
921                }
922            })?
923        };
924        // If we're successful, push the headers and stream...
925        if let Some(child) = child_key {
926            let mut ppp = self.store[parent_key].pending_push_promises.take();
927            ppp.push(&mut self.store.resolve(child));
928
929            let parent = &mut self.store.resolve(parent_key);
930            parent.pending_push_promises = ppp;
931            parent.notify_push();
932        };
933
934        Ok(RecvOutcome::Processed)
935    }
936
937    fn recv_eof<B>(
938        &mut self,
939        send_buffer: &SendBuffer<B>,
940        clear_pending_accept: bool,
941    ) -> Result<(), ()> {
942        let actions = &mut self.actions;
943        let counts = &mut self.counts;
944        let mut send_buffer = send_buffer.inner.lock().unwrap();
945        let send_buffer = &mut *send_buffer;
946
947        if actions.conn_error.is_none() {
948            actions.conn_error = Some(
949                io::Error::new(
950                    io::ErrorKind::BrokenPipe,
951                    "connection closed because of a broken pipe",
952                )
953                .into(),
954            );
955        }
956
957        tracing::trace!("Streams::recv_eof");
958
959        self.store.for_each(|stream| {
960            counts.transition(stream, |counts, stream| {
961                actions.recv.recv_eof(stream);
962
963                // This handles resetting send state associated with the
964                // stream
965                actions.send.handle_error(send_buffer, stream, counts);
966            })
967        });
968
969        actions.clear_queues(clear_pending_accept, &mut self.store, counts);
970        Ok(())
971    }
972
973    fn buffer_pending<T, B>(
974        &mut self,
975        send_buffer: &SendBuffer<B>,
976        dst: &mut Codec<T, Prioritized<B>>,
977    ) -> io::Result<BufferStatus>
978    where
979        T: AsyncWrite + Unpin,
980        B: Buf,
981    {
982        let mut send_buffer = send_buffer.inner.lock().unwrap();
983        let send_buffer = &mut *send_buffer;
984
985        // Send WINDOW_UPDATE frames first
986        //
987        // TODO: It would probably be better to interleave updates w/ data
988        // frames.
989        if self
990            .actions
991            .recv
992            .buffer_pending(&mut self.store, &mut self.counts, dst)?
993            == BufferStatus::CodecFull
994        {
995            return Ok(BufferStatus::CodecFull);
996        }
997
998        // Send any other pending frames
999        if self
1000            .actions
1001            .send
1002            .buffer_pending(send_buffer, &mut self.store, &mut self.counts, dst)?
1003            == BufferStatus::CodecFull
1004        {
1005            return Ok(BufferStatus::CodecFull);
1006        }
1007
1008        Ok(BufferStatus::Complete)
1009    }
1010
1011    fn reclaim_written_frame<T, B>(
1012        &mut self,
1013        send_buffer: &SendBuffer<B>,
1014        dst: &mut Codec<T, Prioritized<B>>,
1015    ) -> bool
1016    where
1017        B: Buf,
1018    {
1019        let mut send_buffer = send_buffer.inner.lock().unwrap();
1020        let send_buffer = &mut *send_buffer;
1021
1022        self.actions
1023            .send
1024            .reclaim_written_frame(send_buffer, &mut self.store, dst)
1025    }
1026
1027    fn send_reset<B>(
1028        &mut self,
1029        send_buffer: &SendBuffer<B>,
1030        id: StreamId,
1031        reason: Reason,
1032    ) -> Result<(), crate::proto::error::GoAway> {
1033        let key = match self.store.find_entry(id) {
1034            Entry::Occupied(e) => e.key(),
1035            Entry::Vacant(e) => {
1036                // Resetting a stream we don't know about? That could be OK...
1037                //
1038                // 1. As a server, we just received a request, but that request
1039                //    was bad, so we're resetting before even accepting it.
1040                //    This is totally fine.
1041                //
1042                // 2. The remote may have sent us a frame on new stream that
1043                //    it's *not* supposed to have done, and thus, we don't know
1044                //    the stream. In that case, sending a reset will "open" the
1045                //    stream in our store. Maybe that should be a connection
1046                //    error instead? At least for now, we need to update what
1047                //    our vision of the next stream is.
1048                if self.counts.peer().is_local_init(id) {
1049                    // We normally would open this stream, so update our
1050                    // next-send-id record.
1051                    self.actions.send.maybe_reset_next_stream_id(id);
1052                } else {
1053                    // We normally would recv this stream, so update our
1054                    // next-recv-id record.
1055                    self.actions.recv.maybe_reset_next_stream_id(id);
1056                }
1057
1058                let stream = Stream::new(id, 0, 0);
1059
1060                e.insert(stream)
1061            }
1062        };
1063
1064        let stream = self.store.resolve(key);
1065        let mut send_buffer = send_buffer.inner.lock().unwrap();
1066        let send_buffer = &mut *send_buffer;
1067        self.actions.send_reset(
1068            stream,
1069            reason,
1070            Initiator::Library,
1071            &mut self.counts,
1072            send_buffer,
1073        )
1074    }
1075}
1076
1077impl<B> Streams<B, client::Peer>
1078where
1079    B: Buf,
1080{
1081    pub fn poll_pending_open(
1082        &mut self,
1083        cx: &Context,
1084        pending: Option<&OpaqueStreamRef>,
1085    ) -> Poll<Result<(), crate::Error>> {
1086        let mut me = self.inner.lock().unwrap();
1087        let me = &mut *me;
1088
1089        me.actions.ensure_no_conn_error()?;
1090        me.actions.send.ensure_next_stream_id()?;
1091
1092        if let Some(pending) = pending {
1093            let mut stream = me.store.resolve(pending.key);
1094            tracing::trace!("poll_pending_open; stream = {:?}", stream.is_pending_open);
1095            if stream.is_pending_open {
1096                stream.wait_send(cx);
1097                return Poll::Pending;
1098            }
1099        }
1100        Poll::Ready(Ok(()))
1101    }
1102}
1103
1104impl<B, P> Streams<B, P>
1105where
1106    P: Peer,
1107{
1108    pub fn as_dyn(&self) -> DynStreams<'_, B> {
1109        let Self {
1110            inner,
1111            send_buffer,
1112            _p,
1113        } = self;
1114        DynStreams {
1115            inner,
1116            send_buffer,
1117            peer: P::r#dyn(),
1118        }
1119    }
1120
1121    /// This function is safe to call multiple times.
1122    ///
1123    /// A `Result` is returned to avoid panicking if the mutex is poisoned.
1124    pub fn recv_eof(&mut self, clear_pending_accept: bool) -> Result<(), ()> {
1125        self.as_dyn().recv_eof(clear_pending_accept)
1126    }
1127
1128    pub(crate) fn max_send_streams(&self) -> usize {
1129        self.inner.lock().unwrap().counts.max_send_streams()
1130    }
1131
1132    pub(crate) fn max_recv_streams(&self) -> usize {
1133        self.inner.lock().unwrap().counts.max_recv_streams()
1134    }
1135
1136    #[cfg(feature = "unstable")]
1137    pub fn num_active_streams(&self) -> usize {
1138        let me = self.inner.lock().unwrap();
1139        me.store.num_active_streams()
1140    }
1141
1142    pub fn has_streams(&self) -> bool {
1143        let me = self.inner.lock().unwrap();
1144        me.counts.has_streams()
1145    }
1146
1147    pub fn has_streams_or_other_references(&self) -> bool {
1148        let me = self.inner.lock().unwrap();
1149        me.counts.has_streams() || me.refs > 1
1150    }
1151
1152    #[cfg(feature = "unstable")]
1153    pub fn num_wired_streams(&self) -> usize {
1154        let me = self.inner.lock().unwrap();
1155        me.store.num_wired_streams()
1156    }
1157}
1158
1159// no derive because we don't need B and P to be Clone.
1160impl<B, P> Clone for Streams<B, P>
1161where
1162    P: Peer,
1163{
1164    fn clone(&self) -> Self {
1165        self.inner.lock().unwrap().refs += 1;
1166        Streams {
1167            inner: self.inner.clone(),
1168            send_buffer: self.send_buffer.clone(),
1169            _p: ::std::marker::PhantomData,
1170        }
1171    }
1172}
1173
1174impl<B, P> Drop for Streams<B, P>
1175where
1176    P: Peer,
1177{
1178    fn drop(&mut self) {
1179        if let Ok(mut inner) = self.inner.lock() {
1180            inner.refs -= 1;
1181            if inner.refs == 1 {
1182                if let Some(task) = inner.actions.task.take() {
1183                    task.wake();
1184                }
1185            }
1186        }
1187    }
1188}
1189
1190// ===== impl StreamRef =====
1191
1192impl<B> StreamRef<B> {
1193    pub fn send_data(&mut self, data: B, end_stream: bool) -> Result<(), UserError>
1194    where
1195        B: Buf,
1196    {
1197        let mut me = self.opaque.inner.lock().unwrap();
1198        let me = &mut *me;
1199
1200        let stream = me.store.resolve(self.opaque.key);
1201        let actions = &mut me.actions;
1202        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
1203        let send_buffer = &mut *send_buffer;
1204
1205        me.counts.transition(stream, |counts, stream| {
1206            // Create the data frame
1207            let mut frame = frame::Data::new(stream.id, data);
1208            frame.set_end_stream(end_stream);
1209
1210            // Send the data frame
1211            actions
1212                .send
1213                .send_data(frame, send_buffer, stream, counts, &mut actions.task)
1214        })
1215    }
1216
1217    pub fn send_trailers(&mut self, trailers: HeaderMap) -> Result<(), UserError> {
1218        let mut me = self.opaque.inner.lock().unwrap();
1219        let me = &mut *me;
1220
1221        let stream = me.store.resolve(self.opaque.key);
1222        let actions = &mut me.actions;
1223        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
1224        let send_buffer = &mut *send_buffer;
1225
1226        me.counts.transition(stream, |counts, stream| {
1227            // Create the trailers frame
1228            let frame = frame::Headers::trailers(stream.id, trailers);
1229
1230            // Send the trailers frame
1231            actions
1232                .send
1233                .send_trailers(frame, send_buffer, stream, counts, &mut actions.task)
1234        })
1235    }
1236
1237    pub fn send_reset(&mut self, reason: Reason) {
1238        let mut me = self.opaque.inner.lock().unwrap();
1239        let me = &mut *me;
1240
1241        let stream = me.store.resolve(self.opaque.key);
1242        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
1243        let send_buffer = &mut *send_buffer;
1244
1245        match me
1246            .actions
1247            .send_reset(stream, reason, Initiator::User, &mut me.counts, send_buffer)
1248        {
1249            Ok(()) => (),
1250            Err(crate::proto::error::GoAway { .. }) => {
1251                // this should never happen, because Initiator::User resets do
1252                // not count toward the local limit.
1253                // we could perhaps make this state impossible, if we made the
1254                // initiator argument a generic, and so this could return
1255                // Infallible instead of an impossible GoAway, but oh well.
1256                unreachable!("Initiator::User should not error sending reset");
1257            }
1258        }
1259    }
1260
1261    pub fn send_informational_headers(&mut self, frame: frame::Headers) -> Result<(), UserError> {
1262        let mut me = self.opaque.inner.lock().unwrap();
1263        let me = &mut *me;
1264
1265        let stream = me.store.resolve(self.opaque.key);
1266        let actions = &mut me.actions;
1267        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
1268        let send_buffer = &mut *send_buffer;
1269
1270        me.counts.transition(stream, |counts, stream| {
1271            // For informational responses (1xx), we need to send headers without
1272            // changing the stream state. This allows multiple informational responses
1273            // to be sent before the final response.
1274
1275            // Validate that this is actually an informational response
1276            debug_assert!(
1277                frame.is_informational(),
1278                "Frame must be informational after conversion from informational response"
1279            );
1280
1281            // Ensure the frame is not marked as end_stream for informational responses
1282            if frame.is_end_stream() {
1283                return Err(UserError::UnexpectedFrameType);
1284            }
1285
1286            // Send the interim informational headers directly to the buffer without state changes
1287            // This bypasses the normal send_headers flow that would transition the stream state
1288            actions.send.send_interim_informational_headers(
1289                frame,
1290                send_buffer,
1291                stream,
1292                counts,
1293                &mut actions.task,
1294            )
1295        })
1296    }
1297
1298    pub fn send_response(
1299        &mut self,
1300        mut response: Response<()>,
1301        end_of_stream: bool,
1302    ) -> Result<(), UserError> {
1303        // Clear before taking lock, incase extensions contain a StreamRef.
1304        response.extensions_mut().clear();
1305        let mut me = self.opaque.inner.lock().unwrap();
1306        let me = &mut *me;
1307
1308        let stream = me.store.resolve(self.opaque.key);
1309        let actions = &mut me.actions;
1310        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
1311        let send_buffer = &mut *send_buffer;
1312
1313        me.counts.transition(stream, |counts, stream| {
1314            let frame = server::Peer::convert_send_message(stream.id, response, end_of_stream);
1315
1316            actions
1317                .send
1318                .send_headers(frame, send_buffer, stream, counts, &mut actions.task)
1319        })
1320    }
1321
1322    pub fn send_push_promise(
1323        &mut self,
1324        mut request: Request<()>,
1325    ) -> Result<StreamRef<B>, UserError> {
1326        // Clear before taking lock, incase extensions contain a StreamRef.
1327        request.extensions_mut().clear();
1328        let mut me = self.opaque.inner.lock().unwrap();
1329        let me = &mut *me;
1330
1331        let mut send_buffer = self.send_buffer.inner.lock().unwrap();
1332        let send_buffer = &mut *send_buffer;
1333
1334        let actions = &mut me.actions;
1335        let promised_id = actions.send.reserve_local()?;
1336
1337        let child_key = {
1338            let mut child_stream = me.store.insert(
1339                promised_id,
1340                Stream::new(
1341                    promised_id,
1342                    actions.send.init_window_sz(),
1343                    actions.recv.init_window_sz(),
1344                ),
1345            );
1346            child_stream.state.reserve_local()?;
1347            child_stream.is_pending_push = true;
1348            child_stream.key()
1349        };
1350
1351        let pushed = {
1352            let mut stream = me.store.resolve(self.opaque.key);
1353
1354            let frame = crate::server::Peer::convert_push_message(stream.id, promised_id, request)?;
1355
1356            actions
1357                .send
1358                .send_push_promise(frame, send_buffer, &mut stream, &mut actions.task)
1359        };
1360
1361        if let Err(err) = pushed {
1362            let mut child_stream = me.store.resolve(child_key);
1363            child_stream.unlink();
1364            child_stream.remove();
1365            return Err(err);
1366        }
1367
1368        me.refs += 1;
1369        let opaque =
1370            OpaqueStreamRef::new(self.opaque.inner.clone(), &mut me.store.resolve(child_key));
1371
1372        Ok(StreamRef {
1373            opaque,
1374            send_buffer: self.send_buffer.clone(),
1375        })
1376    }
1377
1378    /// Called by the server after the stream is accepted. Given that clients
1379    /// initialize streams by sending HEADERS, the request will always be
1380    /// available.
1381    ///
1382    /// # Panics
1383    ///
1384    /// This function panics if the request isn't present.
1385    pub fn take_request(&self) -> Request<()> {
1386        let mut me = self.opaque.inner.lock().unwrap();
1387        let me = &mut *me;
1388
1389        let mut stream = me.store.resolve(self.opaque.key);
1390        me.actions.recv.take_request(&mut stream)
1391    }
1392
1393    /// Called by a client to see if the current stream is pending open
1394    pub fn is_pending_open(&self) -> bool {
1395        let mut me = self.opaque.inner.lock().unwrap();
1396        me.store.resolve(self.opaque.key).is_pending_open
1397    }
1398
1399    /// Request capacity to send data
1400    pub fn reserve_capacity(&mut self, capacity: WindowSize) {
1401        let mut me = self.opaque.inner.lock().unwrap();
1402        let me = &mut *me;
1403
1404        let mut stream = me.store.resolve(self.opaque.key);
1405
1406        me.actions
1407            .send
1408            .reserve_capacity(capacity, &mut stream, &mut me.counts)
1409    }
1410
1411    /// Returns the stream's current send capacity.
1412    pub fn capacity(&self) -> WindowSize {
1413        let mut me = self.opaque.inner.lock().unwrap();
1414        let me = &mut *me;
1415
1416        let mut stream = me.store.resolve(self.opaque.key);
1417
1418        me.actions.send.capacity(&mut stream)
1419    }
1420
1421    /// Request to be notified when the stream's capacity increases
1422    pub fn poll_capacity(&mut self, cx: &Context) -> Poll<Option<Result<WindowSize, UserError>>> {
1423        let mut me = self.opaque.inner.lock().unwrap();
1424        let me = &mut *me;
1425
1426        let mut stream = me.store.resolve(self.opaque.key);
1427
1428        me.actions.send.poll_capacity(cx, &mut stream)
1429    }
1430
1431    /// Request to be notified for if a `RST_STREAM` is received for this stream.
1432    pub(crate) fn poll_reset(
1433        &mut self,
1434        cx: &Context,
1435        mode: proto::PollReset,
1436    ) -> Poll<Result<Reason, crate::Error>> {
1437        let mut me = self.opaque.inner.lock().unwrap();
1438        let me = &mut *me;
1439
1440        let mut stream = me.store.resolve(self.opaque.key);
1441
1442        me.actions.send.poll_reset(cx, &mut stream, mode)
1443    }
1444
1445    pub fn clone_to_opaque(&self) -> OpaqueStreamRef {
1446        self.opaque.clone()
1447    }
1448
1449    pub fn stream_id(&self) -> StreamId {
1450        self.opaque.stream_id()
1451    }
1452}
1453
1454impl<B> Clone for StreamRef<B> {
1455    fn clone(&self) -> Self {
1456        StreamRef {
1457            opaque: self.opaque.clone(),
1458            send_buffer: self.send_buffer.clone(),
1459        }
1460    }
1461}
1462
1463// ===== impl OpaqueStreamRef =====
1464
1465impl OpaqueStreamRef {
1466    fn new(inner: Arc<Mutex<Inner>>, stream: &mut store::Ptr) -> OpaqueStreamRef {
1467        stream.ref_inc();
1468        OpaqueStreamRef {
1469            inner,
1470            key: stream.key(),
1471        }
1472    }
1473    /// Called by a client to check for a received response.
1474    pub fn poll_response(&mut self, cx: &Context) -> Poll<Result<Response<()>, proto::Error>> {
1475        let mut me = self.inner.lock().unwrap();
1476        let me = &mut *me;
1477
1478        let mut stream = me.store.resolve(self.key);
1479
1480        me.actions.recv.poll_response(cx, &mut stream)
1481    }
1482
1483    /// Called by a client to check for informational responses (1xx status codes)
1484    pub fn poll_informational(
1485        &mut self,
1486        cx: &Context,
1487    ) -> Poll<Option<Result<Response<()>, proto::Error>>> {
1488        let mut me = self.inner.lock().unwrap();
1489        let me = &mut *me;
1490
1491        let mut stream = me.store.resolve(self.key);
1492
1493        me.actions.recv.poll_informational(cx, &mut stream)
1494    }
1495    /// Called by a client to check for a pushed request.
1496    pub fn poll_pushed(
1497        &mut self,
1498        cx: &Context,
1499    ) -> Poll<Option<Result<(Request<()>, OpaqueStreamRef), proto::Error>>> {
1500        let mut me = self.inner.lock().unwrap();
1501        let me = &mut *me;
1502
1503        let mut stream = me.store.resolve(self.key);
1504        me.actions
1505            .recv
1506            .poll_pushed(cx, &mut stream)
1507            .map_ok(|(h, key)| {
1508                me.refs += 1;
1509                let opaque_ref =
1510                    OpaqueStreamRef::new(self.inner.clone(), &mut me.store.resolve(key));
1511                (h, opaque_ref)
1512            })
1513    }
1514
1515    pub fn is_end_stream(&self) -> bool {
1516        let mut me = self.inner.lock().unwrap();
1517        let me = &mut *me;
1518
1519        let stream = me.store.resolve(self.key);
1520
1521        me.actions.recv.is_end_stream(&stream)
1522    }
1523
1524    pub fn poll_data(&mut self, cx: &Context) -> Poll<Option<Result<Bytes, proto::Error>>> {
1525        let mut me = self.inner.lock().unwrap();
1526        let me = &mut *me;
1527
1528        let mut stream = me.store.resolve(self.key);
1529
1530        me.actions
1531            .recv
1532            .poll_data(cx, &mut stream)
1533            .map(|result| match result {
1534                Some(Ok(data)) => {
1535                    if data.is_budgeted {
1536                        me.counts.release_data_frame(data.payload.len());
1537                    }
1538                    Some(Ok(data.payload))
1539                }
1540                Some(Err(err)) => Some(Err(err)),
1541                None => None,
1542            })
1543    }
1544
1545    pub fn poll_trailers(&mut self, cx: &Context) -> Poll<Option<Result<HeaderMap, proto::Error>>> {
1546        let mut me = self.inner.lock().unwrap();
1547        let me = &mut *me;
1548
1549        let mut stream = me.store.resolve(self.key);
1550
1551        me.actions.recv.poll_trailers(cx, &mut stream)
1552    }
1553
1554    pub(crate) fn available_recv_capacity(&self) -> isize {
1555        let me = self.inner.lock().unwrap();
1556        let me = &*me;
1557
1558        let stream = &me.store[self.key];
1559        stream.recv_flow.available().into()
1560    }
1561
1562    pub(crate) fn used_recv_capacity(&self) -> WindowSize {
1563        let me = self.inner.lock().unwrap();
1564        let me = &*me;
1565
1566        let stream = &me.store[self.key];
1567        stream.in_flight_recv_data
1568    }
1569
1570    /// Releases recv capacity back to the peer. This may result in sending
1571    /// WINDOW_UPDATE frames on both the stream and connection.
1572    pub fn release_capacity(&mut self, capacity: WindowSize) -> Result<(), UserError> {
1573        let mut me = self.inner.lock().unwrap();
1574        let me = &mut *me;
1575
1576        let mut stream = me.store.resolve(self.key);
1577
1578        me.actions
1579            .recv
1580            .release_capacity(capacity, &mut stream, &mut me.actions.task)
1581    }
1582
1583    /// Clear the receive queue and set the status to no longer receive data frames.
1584    pub(crate) fn clear_recv_buffer(&mut self) {
1585        let mut me = self.inner.lock().unwrap();
1586        let me = &mut *me;
1587
1588        let mut stream = me.store.resolve(self.key);
1589        stream.is_recv = false;
1590        me.actions
1591            .recv
1592            .clear_recv_buffer(&mut stream, &mut me.actions.task, &mut me.counts);
1593    }
1594
1595    pub fn stream_id(&self) -> StreamId {
1596        self.inner.lock().unwrap().store[self.key].id
1597    }
1598}
1599
1600impl fmt::Debug for OpaqueStreamRef {
1601    fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
1602        use std::sync::TryLockError::*;
1603
1604        match self.inner.try_lock() {
1605            Ok(me) => {
1606                let stream = &me.store[self.key];
1607                fmt.debug_struct("OpaqueStreamRef")
1608                    .field("stream_id", &stream.id)
1609                    .field("ref_count", &stream.ref_count)
1610                    .finish()
1611            }
1612            Err(Poisoned(_)) => fmt
1613                .debug_struct("OpaqueStreamRef")
1614                .field("inner", &"<Poisoned>")
1615                .finish(),
1616            Err(WouldBlock) => fmt
1617                .debug_struct("OpaqueStreamRef")
1618                .field("inner", &"<Locked>")
1619                .finish(),
1620        }
1621    }
1622}
1623
1624impl Clone for OpaqueStreamRef {
1625    fn clone(&self) -> Self {
1626        // Increment the ref count
1627        let mut inner = self.inner.lock().unwrap();
1628        inner.store.resolve(self.key).ref_inc();
1629        inner.refs += 1;
1630
1631        OpaqueStreamRef {
1632            inner: self.inner.clone(),
1633            key: self.key,
1634        }
1635    }
1636}
1637
1638impl Drop for OpaqueStreamRef {
1639    fn drop(&mut self) {
1640        drop_stream_ref(&self.inner, self.key);
1641    }
1642}
1643
1644// TODO: Move back in fn above
1645fn drop_stream_ref(inner: &Mutex<Inner>, key: store::Key) {
1646    let mut me = match inner.lock() {
1647        Ok(inner) => inner,
1648        Err(_) => {
1649            if ::std::thread::panicking() {
1650                tracing::trace!("StreamRef::drop; mutex poisoned");
1651                return;
1652            } else {
1653                panic!("StreamRef::drop; mutex poisoned");
1654            }
1655        }
1656    };
1657
1658    let me = &mut *me;
1659    me.refs -= 1;
1660    let mut stream = me.store.resolve(key);
1661
1662    tracing::trace!("drop_stream_ref; stream={:?}", stream);
1663
1664    // decrement the stream's ref count by 1.
1665    stream.ref_dec();
1666
1667    let actions = &mut me.actions;
1668
1669    // If the stream is not referenced and it is already
1670    // closed (does not have to go through logic below
1671    // of canceling the stream), we should notify the task
1672    // (connection) so that it can close properly
1673    if stream.ref_count == 0 && stream.is_closed() {
1674        if let Some(task) = actions.task.take() {
1675            task.wake();
1676        }
1677    }
1678
1679    me.counts.transition(stream, |counts, stream| {
1680        maybe_cancel(stream, actions, counts);
1681
1682        if stream.ref_count == 0 {
1683            // Release any recv window back to connection, no one can access
1684            // it anymore.
1685            actions
1686                .recv
1687                .release_closed_capacity(stream, &mut actions.task, counts);
1688
1689            // We won't be able to reach our push promises anymore
1690            let mut ppp = stream.pending_push_promises.take();
1691            while let Some(promise) = ppp.pop(stream.store_mut()) {
1692                counts.transition(promise, |counts, stream| {
1693                    maybe_cancel(stream, actions, counts);
1694                });
1695            }
1696        }
1697    });
1698}
1699
1700fn maybe_cancel(stream: &mut store::Ptr, actions: &mut Actions, counts: &mut Counts) {
1701    if stream.is_canceled_interest() {
1702        // Server is allowed to early respond without fully consuming the client input stream
1703        // But per the RFC, must send a RST_STREAM(NO_ERROR) in such cases. https://www.rfc-editor.org/rfc/rfc7540#section-8.1
1704        // Some other http2 implementation may interpret other error code as fatal if not respected (i.e: nginx https://trac.nginx.org/nginx/ticket/2376)
1705        let reason = if counts.peer().is_server()
1706            && stream.state.is_send_closed()
1707            && stream.state.is_recv_streaming()
1708        {
1709            Reason::NO_ERROR
1710        } else {
1711            Reason::CANCEL
1712        };
1713
1714        actions
1715            .send
1716            .schedule_implicit_reset(stream, reason, counts, &mut actions.task);
1717        actions.recv.enqueue_reset_expiration(stream, counts);
1718    }
1719}
1720
1721// ===== impl SendBuffer =====
1722
1723impl<B> SendBuffer<B> {
1724    fn new() -> Self {
1725        let inner = Mutex::new(Buffer::new());
1726        SendBuffer { inner }
1727    }
1728
1729    pub fn is_empty(&self) -> bool {
1730        let buf = self.inner.lock().unwrap();
1731        buf.is_empty()
1732    }
1733}
1734
1735// ===== impl Actions =====
1736
1737impl Actions {
1738    fn send_reset<B>(
1739        &mut self,
1740        stream: store::Ptr,
1741        reason: Reason,
1742        initiator: Initiator,
1743        counts: &mut Counts,
1744        send_buffer: &mut Buffer<Frame<B>>,
1745    ) -> Result<(), crate::proto::error::GoAway> {
1746        counts.transition(stream, |counts, stream| {
1747            if initiator.is_library() {
1748                if counts.can_inc_num_local_error_resets() {
1749                    counts.inc_num_local_error_resets();
1750                } else {
1751                    tracing::warn!(
1752                        "locally-reset streams reached limit ({:?})",
1753                        counts.max_local_error_resets().unwrap(),
1754                    );
1755                    return Err(crate::proto::error::GoAway {
1756                        reason: Reason::ENHANCE_YOUR_CALM,
1757                        debug_data: "too_many_internal_resets".into(),
1758                    });
1759                }
1760            }
1761
1762            self.send.send_reset(
1763                reason,
1764                initiator,
1765                send_buffer,
1766                stream,
1767                counts,
1768                &mut self.task,
1769            );
1770            self.recv.enqueue_reset_expiration(stream, counts);
1771            // if a RecvStream is parked, ensure it's notified
1772            stream.notify_recv();
1773
1774            Ok(())
1775        })
1776    }
1777
1778    fn reset_on_recv_stream_err_deferred<B>(
1779        &mut self,
1780        send_buffer: &SendBuffer<B>,
1781        stream: &mut store::Ptr,
1782        counts: &mut Counts,
1783        res: Result<(), Error>,
1784    ) -> Result<(), Error> {
1785        if matches!(res, Err(Error::Reset(..))) {
1786            let mut send_buffer = send_buffer.inner.lock().unwrap();
1787            self.reset_on_recv_stream_err(&mut send_buffer, stream, counts, res)
1788        } else {
1789            res
1790        }
1791    }
1792
1793    fn reset_on_recv_stream_err<B>(
1794        &mut self,
1795        buffer: &mut Buffer<Frame<B>>,
1796        stream: &mut store::Ptr,
1797        counts: &mut Counts,
1798        res: Result<(), Error>,
1799    ) -> Result<(), Error> {
1800        if let Err(Error::Reset(stream_id, reason, initiator)) = res {
1801            debug_assert_eq!(stream_id, stream.id);
1802
1803            if counts.can_inc_num_local_error_resets() {
1804                counts.inc_num_local_error_resets();
1805
1806                // Reset the stream.
1807                self.send
1808                    .send_reset(reason, initiator, buffer, stream, counts, &mut self.task);
1809                self.recv.enqueue_reset_expiration(stream, counts);
1810                // if a RecvStream is parked, ensure it's notified
1811                stream.notify_recv();
1812                Ok(())
1813            } else {
1814                tracing::warn!(
1815                    "reset_on_recv_stream_err; locally-reset streams reached limit ({:?})",
1816                    counts.max_local_error_resets().unwrap(),
1817                );
1818                Err(Error::library_go_away_data(
1819                    Reason::ENHANCE_YOUR_CALM,
1820                    "too_many_internal_resets",
1821                ))
1822            }
1823        } else {
1824            res
1825        }
1826    }
1827
1828    fn ensure_not_idle(&mut self, peer: peer::Dyn, id: StreamId) -> Result<(), Reason> {
1829        if peer.is_local_init(id) {
1830            self.send.ensure_not_idle(id)
1831        } else {
1832            self.recv.ensure_not_idle(id)
1833        }
1834    }
1835
1836    fn ensure_no_conn_error(&self) -> Result<(), proto::Error> {
1837        if let Some(ref err) = self.conn_error {
1838            Err(err.clone())
1839        } else {
1840            Ok(())
1841        }
1842    }
1843
1844    /// Check if we possibly could have processed and since forgotten this stream.
1845    ///
1846    /// If we send a RST_STREAM for a stream, we will eventually "forget" about
1847    /// the stream to free up memory. It's possible that the remote peer had
1848    /// frames in-flight, and by the time we receive them, our own state is
1849    /// gone. We *could* tear everything down by sending a GOAWAY, but it
1850    /// is more likely to be latency/memory constraints that caused this,
1851    /// and not a bad actor. So be less catastrophic, the spec allows
1852    /// us to send another RST_STREAM of STREAM_CLOSED.
1853    fn may_have_forgotten_stream(&self, peer: peer::Dyn, id: StreamId) -> bool {
1854        if id.is_zero() {
1855            return false;
1856        }
1857        if peer.is_local_init(id) {
1858            self.send.may_have_created_stream(id)
1859        } else {
1860            self.recv.may_have_created_stream(id)
1861        }
1862    }
1863
1864    fn clear_queues(&mut self, clear_pending_accept: bool, store: &mut Store, counts: &mut Counts) {
1865        self.recv.clear_queues(clear_pending_accept, store, counts);
1866        self.send.clear_queues(store, counts);
1867    }
1868}
h2/proto/streams/streams.rs

h2/proto/streams/
streams.rs
