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