Struct StreamPollSet 

pub struct StreamPollSet<K, P, S>
where
    S: PeekableStream + Unpin,
{
    priorities: HashMap<K, (P, InTunnelActivity)>,
    ready_streams: BTreeMap<(P, K), S>,
    pending_streams: KeyedFuturesUnordered<K, PeekableReady<S>>,
    tunnel_activity: TunnelActivity,
}

Manages a dynamic set of [futures::Stream] with associated keys and priorities.

Notable features:

• Prioritization: streams have an associated priority, and ready-streams are iterated over in ascending priority order.
• Efficient polling: an unready stream won’t be polled again until it’s ready or exhausted (e.g. a corresponding [futures::Sink] is written-to or dropped). A ready stream won’t be polled again until the ready item has been removed.

Fields

priorities: HashMap<K, (P, InTunnelActivity)>

Priority for each stream in the set, and associated InTunnelActivity token.

ready_streams: BTreeMap<(P, K), S>

Streams that have a result ready, in ascending order by priority.

pending_streams: KeyedFuturesUnordered<K, PeekableReady<S>>

Streams for which we’re still waiting for the next result.

tunnel_activity: TunnelActivity

Information about how active this particular hop has been, with respect to tracking overall tunnel activity.

Implementations

impl<K, P, S> StreamPollSet<K, P, S>

where K: Ord + Hash + Clone + Send + Sync + 'static, S: PeekableStream + Unpin, P: Ord + Clone,

pub fn new() -> Self

Create a new, empty, StreamPollSet.

pub fn try_insert( &mut self, key: K, priority: P, stream: S, ) -> Result<(), KeyAlreadyInsertedError<K, P, S>>

Insert a stream, with an associated key and priority.

If the key is already in use, the parameters are returned without altering self.

pub fn remove(&mut self, key: &K) -> Option<(K, P, S)>

Remove the entry for key, if any. This is the key, priority, buffered poll_next result, and stream.

pub fn poll_ready_iter_mut<'a>( &'a mut self, cx: &mut Context<'_>, ) -> impl Iterator<Item = (&'a K, &'a P, &'a mut S)> + 'a + use<'a, K, P, S>

Polls streams that are ready to be polled, and returns an iterator over all streams for which we have a buffered Poll::Ready result, in ascending priority order.

Registers the provided Context to be woken when any of the internal streams that weren’t ready in the previous call to this method (and therefore wouldn’t have appeared in the iterator results) become potentially ready (based on when the inner stream wakes the Context provided to its own poll_next).

The same restrictions apply as for Self::stream_mut. e.g. do not directly call PeekableStream::poll_peek to see what item is available on the stream; instead use Self::peek_mut. (Or tor_async_utils::peekable_stream::UnobtrusivePeekableStream if implemented for the stream).

This method does not drain ready items. Some values can be removed with Self::take_ready_value_and_reprioritize. None values can only be removed by removing the whole stream with Self::remove.

This API is meant to allow callers to find the first stream (in priority order) that is ready, and that the caller is able to process now. i.e. it’s specifically to support the use-case where external factors may prevent the processing of some streams but not others.

Example:

# // We need the `nocompile` since `StreamPollSet` is non-pub.
# // TODO: take away the nocompile if we make this pub or implement some
# // workaround to expose it to doc-tests.
# type Key=u64;
# type Value=u64;
# type Priority=u64;
# type MyStream=Box<dyn futures::Stream<Item=Value> + Unpin>;
# fn can_process(key: &Key, val: &Value) -> bool { true }
# fn process(val: Value) { }
# fn new_priority(priority: &Priority) -> Priority { *priority }
fn process_a_ready_stream(sps: &mut StreamPollSet<Key, Value, Priority, MyStream>, cx: &mut std::task::Context) -> std::task::Poll<()> {
  let mut iter = sps.poll_ready_iter(cx);
  while let Some((key, priority, stream)) = iter.next() {
    let Some(value) = stream.unobtrusive_peek(Pin::new(stream)) else {
       // Stream exhausted. Remove the stream. We have to drop the iterator
       // first, though, so that we can mutate.
       let key = *key;
       drop(iter);
       sps.remove(&key).unwrap();
       return std::task::Poll::Ready(());
    };
    if can_process(key, value) {
       let key = *key;
       let priority = new_priority(priority);
       drop(iter);
       let (_old_priority, value) = sps.take_ready_value_and_reprioritize(&key, priority).unwrap();
       process(value);
       return std::task::Poll::Ready(());
    }
  }
  return std::task::Poll::Pending;
}

pub fn take_ready_value_and_reprioritize( &mut self, key: &K, new_priority: P, ) -> Option<(P, S::Item)>

If the stream for key has Some(value) ready, take that value and set the priority for it to new_priority.

This method doesn’t register a waker with the polled stream. Use poll_ready_iter to ensure streams make progress.

If the key doesn’t exist, the stream isn’t ready, or the stream’s value is None (indicating the end of the stream), this function returns None without mutating anything.

Ended streams should be removed using Self::remove.

pub fn peek_mut<'a>( &'a mut self, key: &K, ) -> Option<Poll<Option<&'a mut S::Item>>>

Get a mut reference to a ready value for key key, if one exists.

This method doesn’t poll the internal streams. Use poll_ready_iter to ensure streams make progress.

pub fn stream(&self, key: &K) -> Option<&S>

Get a reference to the stream for key.

The same restrictions apply as for Self::stream_mut (e.g. using interior mutability).

pub fn stream_mut(&mut self, key: &K) -> Option<&mut S>

Get a mut reference to the stream for key.

Polling the stream through this reference, or otherwise causing its registered Waker to be removed without waking it, will result in unspecified (but not unsound) behavior.

This is mostly intended for accessing non-Stream functionality of the stream object, though it is permitted to mutate it in a way that the stream becomes ready (potentially removing and waking its registered Waker(s)).

pub fn len(&self) -> usize

Number of streams managed by this object.

pub fn tunnel_activity(&self) -> TunnelActivity

Return a TunnelActivity for this hop.

Trait Implementations

impl<K, P, S> Drop for StreamPollSet<K, P, S>

where S: PeekableStream + Unpin,

fn drop(&mut self)

Executes the destructor for this type.