h2/proto/streams/
streams.rs

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