Merge branch 'move-stream-peek-prelim' into 'main' (9aaa1d1b) · Commits · NoisyCoil / Arti

Cargo.toml

+3 −0

Original line number	Diff line number	Diff line
		@@ -5,6 +5,9 @@
		#
		# Please keep this list topologically sorted by dependency relation, so
		# that every crate appears _before_ any other crate that depends on it.
		# We depend on this for publishing to crates.io. e.g.
		# see
		# https://blog.iany.me/2020/10/gotchas-to-publish-rust-crates-in-a-workspace/#cyclic-dependencies
		members = [
		"crates/slotmap-careful",
		"crates/test-temp-dir",

+2 −2

Original line number	Diff line number	Diff line
		@@ -30,7 +30,7 @@ pub(crate) use std::{
		pub(crate) use futures::{
		channel::mpsc,
		stream::FusedStream,
		task::{Spawn, SpawnError, SpawnExt as _},
		task::{noop_waker_ref, Spawn, SpawnError, SpawnExt as _},
		FutureExt as _, Sink, SinkExt as _, Stream, StreamExt as _,
		};

		@@ -67,5 +67,5 @@ pub(crate) use crate::{
		private::Sealed,
		refcount,
		stream_peek::StreamUnobtrusivePeeker,
		utils::{DefaultExtTake, NoopWaker},
		utils::DefaultExtTake,
		};

+2 −3

Original line number	Diff line number	Diff line
		@@ -66,7 +66,7 @@ use tor_async_utils::peekable_stream::UnobtrusivePeekableStream;

		use crate::internal_prelude::*;

		use std::task::{Context, Poll, Poll::*, Waker};
		use std::task::{Context, Poll, Poll::*};

		//---------- Sender ----------

		@@ -522,8 +522,7 @@ impl<T: Debug + Send + 'static, C: ChannelSpec> Drop for ReceiverState<T, C> {

		// try to free whatever is in the queue, in case the stream doesn't do that itself
		// No-one can poll us any more, so we are no longer interested in wakeups
		let noop_waker = Waker::from(Arc::new(NoopWaker));
		let mut noop_cx = Context::from_waker(&noop_waker);
		let mut noop_cx = Context::from_waker(noop_waker_ref());

		// prevent further sends, so that our drain doesn't race indefinitely with the sender
		if let Some(mut rx_inner) =

+2 −4

Original line number	Diff line number	Diff line
		@@ -124,12 +124,10 @@ impl<S: Stream> UnobtrusivePeekableStream for StreamUnobtrusivePeeker<S> {
		return None;
		};

		let waker_buf;
		let waker = if let Some(waker) = self_.poll_waker.as_ref() {
		waker
		} else {
		waker_buf = Waker::from(Arc::new(NoopWaker));
		&waker_buf
		noop_waker_ref()
		};

		match inner.poll_next(&mut Context::from_waker(waker)) {
		@@ -270,7 +268,7 @@ impl<S: Stream> StreamUnobtrusivePeeker<S> {
		/// This method must be used with care!
		/// Whatever you do mustn't interfere with polling and peeking.
		/// Careless use can result in wrong behaviour including deadlocks.
		pub(crate) fn as_raw_inner_pin_mut<'s>(self: Pin<&'s mut Self>) -> Option<Pin<&'s mut S>> {
		pub fn as_raw_inner_pin_mut<'s>(self: Pin<&'s mut Self>) -> Option<Pin<&'s mut S>> {
		self.project().inner.as_pin_mut()
		}
		}

+0 −13

Original line number	Diff line number	Diff line
		@@ -11,16 +11,3 @@ pub(crate) trait DefaultExtTake: Default {
		mem::take(self)
		}
		}

		/// `std::task::Waker::noop` but that's nightly
		///
		/// Note that no-op wakers must be used with care,
		/// so don't just move or copy this elsewhere without consideration.
		/// See <https://github.com/rust-lang/rust/pull/128064>.
		//
		// TODO if that MR is merged in some form, refer to the final version in the actual docs.
		// If that MR is not merged, put some version of the warning here.
		pub(crate) struct NoopWaker;
		impl std::task::Wake for NoopWaker {
		fn wake(self: Arc<Self>) {}
		}