Struct CircHopOutbound 

pub(crate) struct CircHopOutbound {
    ccontrol: Arc<Mutex<CongestionControl>>,
    map: Arc<Mutex<StreamMap>>,
    relay_format: RelayCellFormat,
    flow_ctrl_params: Arc<FlowCtrlParameters>,
    n_outgoing_cells_permitted: Option<NonZeroU32>,
}

The outbound state of a hop.

Fields

ccontrol: Arc<Mutex<CongestionControl>>

Congestion control object.

This object is also in charge of handling circuit level SENDME logic for this hop.

map: Arc<Mutex<StreamMap>>

Map from stream IDs to streams.

We store this with the reactor instead of the circuit, since the reactor needs it for every incoming cell on a stream, whereas the circuit only needs it when allocating new streams.

NOTE: this is behind a mutex because the client reactor polls the StreamMaps of all hops concurrently, in a FuturesUnordered. Without the mutex, this wouldn’t be possible, because it would mean holding multiple mutable references to self (the reactor). Note, however, that there should never be any contention on this mutex: we never create more than one CircHopList::ready_streams_iterator() stream at a time, and we never clone/lock the hop’s StreamMap outside of it.

Additionally, the stream map of the last hop (join point) of a conflux tunnel is shared with all the circuits in the tunnel.

relay_format: RelayCellFormat

Format to use for relay cells.

flow_ctrl_params: Arc<FlowCtrlParameters>

Flow control parameters for new streams.

n_outgoing_cells_permitted: Option<NonZeroU32>

Remaining permitted outgoing relay cells from this hop, plus 1.

If this ever decrements from Some(1), then the circuit must be torn down with an error.

Implementations

impl CircHopOutbound

pub(crate) fn new( ccontrol: Arc<Mutex<CongestionControl>>, relay_format: RelayCellFormat, flow_ctrl_params: Arc<FlowCtrlParameters>, settings: &HopSettings, ) -> Self

Create a new CircHopOutbound.

pub(crate) fn begin_stream( &mut self, hop: Option<HopNum>, message: AnyRelayMsg, sender: StreamQueueSender, rx: Receiver<AnyRelayMsg, MpscSpec>, rate_limit_updater: Sender<StreamRateLimit>, drain_rate_requester: NotifySender<DrainRateRequest>, cmd_checker: Box<dyn CmdChecker + Send + 'static>, ) -> Result<(SendRelayCell, StreamId)>

Start a stream. Creates an entry in the stream map with the given channels, and sends the message to the provided hop.

pub(crate) fn close_stream( &mut self, circ_id: impl Display, id: StreamId, hop: Option<HopNum>, message: CloseStreamBehavior, why: TerminateReason, expiry: Instant, ) -> Result<Option<SendRelayCell>>

Close the stream associated with id because the stream was dropped.

If we have not already received an END cell on this stream, send one. If no END cell is specified, an END cell with the reason byte set to REASON_MISC will be sent.

pub(crate) fn maybe_send_xon( &mut self, rate: XonKbpsEwma, id: StreamId, ) -> Result<Option<Xon>>

Check if we should send an XON message.

If we should, then returns the XON message that should be sent.

pub(crate) fn maybe_send_xoff(&mut self, id: StreamId) -> Result<Option<Xoff>>

Check if we should send an XOFF message.

If we should, then returns the XOFF message that should be sent.

pub(crate) fn relay_cell_format(&self) -> RelayCellFormat

Return the format that is used for relay cells sent to this hop.

For the most part, this format isn’t necessary to interact with a CircHop; it becomes relevant when we are deciding what we can encode for the hop.

pub(crate) fn n_open_streams(&self) -> usize

Return the number of open streams on this hop.

WARNING: because this locks the stream map mutex, it should never be called from a context where that mutex is already locked.

pub(crate) fn ccontrol(&self) -> &Arc<Mutex<CongestionControl>>

Return a reference to our CongestionControl object.

pub(crate) fn about_to_send( &mut self, circ_id: impl Display, stream_id: StreamId, msg: &AnyRelayMsg, ) -> Result<()>

We’re about to send msg.

See OpenStreamEnt::about_to_send.

pub(crate) fn add_ent_with_id( &self, sink: StreamQueueSender, rx: Receiver<AnyRelayMsg, MpscSpec>, rate_limit_updater: Sender<StreamRateLimit>, drain_rate_requester: NotifySender<DrainRateRequest>, stream_id: StreamId, cmd_checker: Box<dyn CmdChecker + Send + 'static>, ) -> Result<()>

Add an entry to this map using the specified StreamId.

fn build_flow_ctrl( &self, params: Arc<FlowCtrlParameters>, rate_limit_updater: Sender<StreamRateLimit>, drain_rate_requester: NotifySender<DrainRateRequest>, ) -> Result<StreamFlowCtrl>

Builds the reactor’s flow control handler for a new stream.

fn deliver_msg_to_stream( streamid: StreamId, ent: &mut OpenStreamEnt, cell_counts_toward_windows: bool, msg: UnparsedRelayMsg, ) -> Result<bool>

Deliver msg to the specified open stream entry ent.

pub(crate) fn ending_msg_received(&self, stream_id: StreamId) -> Result<()>

Note that we received an END message (or other message indicating the end of the stream) on the stream with id.

See StreamMap::ending_msg_received.

pub(crate) fn handle_msg<F>( &self, possible_proto_violation_err: F, cell_counts_toward_windows: bool, streamid: StreamId, msg: UnparsedRelayMsg, now: Instant, ) -> Result<Option<UnparsedRelayMsg>>

where F: FnOnce(StreamId) -> Error,

Handle msg, delivering it to the stream with the specified streamid if appropriate.

Returns back the provided msg, if the message is an incoming stream request that needs to be handled by the calling code.

pub(crate) fn stream_map(&self) -> &Arc<Mutex<StreamMap>>

Get the stream map of this hop.

pub(crate) fn set_stream_map( &mut self, map: Arc<Mutex<StreamMap>>, ) -> StdResult<(), Bug>

Set the stream map of this hop to map.

Returns an error if the existing stream map of the hop has any open stream.

pub(crate) fn decrement_cell_limit(&mut self) -> Result<()>

Decrement the limit of outbound cells that may be sent to this hop; give an error if it would reach zero.