Struct Reactor 

pub struct Reactor {
    control: UnboundedReceiver<CtrlMsg>,
    command: UnboundedReceiver<CtrlCmd>,
    reactor_closed_tx: Sender<Void>,
    circuits: ConfluxSet,
    tunnel_id: TunnelId,
    cell_handlers: CellHandlers,
    runtime: DynTimeProvider,
    conflux_hs_ctx: Option<ConfluxHandshakeCtx>,
    ooo_msgs: BinaryHeap<ConfluxHeapEntry>,
}

Object to handle incoming cells and background tasks on a circuit

This type is returned when you finish a circuit; you need to spawn a new task that calls run() on it.

Fields

control: UnboundedReceiver<CtrlMsg>

Receiver for control messages for this reactor, sent by ClientCirc objects.

This channel is polled in Reactor::run_once, but only if the chan_sender sink is ready to accept cells.

command: UnboundedReceiver<CtrlCmd>

Receiver for command messages for this reactor, sent by ClientCirc objects.

This channel is polled in Reactor::run_once.

NOTE: this is a separate channel from control, because some messages have higher priority and need to be handled even if the chan_sender is not ready (whereas control messages are not read until the chan_sender sink is ready to accept cells).

reactor_closed_tx: Sender<Void>

A oneshot sender that is used to alert other tasks when this reactor is finally dropped.

It is a sender for Void because we never actually want to send anything here; we only want to generate canceled events.

circuits: ConfluxSet

A set of circuits that form a tunnel.

Contains 1 or more circuits.

Circuits may be added to this set throughout the lifetime of the reactor.

Sometimes, the reactor will remove circuits from this set, for example if the LINKED message takes too long to arrive, or if congestion control negotiation fails. The reactor will continue running with the remaining circuits. It will shut down if all the circuits are removed.

tunnel_id: TunnelId

An identifier for logging about this tunnel reactor.

cell_handlers: CellHandlers

Handlers, shared with Circuit.

runtime: DynTimeProvider

The time provider, used for conflux handshake timeouts.

conflux_hs_ctx: Option<ConfluxHandshakeCtx>

The conflux handshake context, if there is an on-going handshake.

Set to None if this is a single-path tunnel, or if none of the circuit legs from our conflux set are currently in the conflux handshake phase.

ooo_msgs: BinaryHeap<ConfluxHeapEntry>

A min-heap buffering all the out-of-order messages received so far.

TODO(conflux): this becomes a DoS vector unless we impose a limit on its size. We should make this participate in the memquota memory tracking system, somehow.

Implementations

impl Reactor

pub(super) fn new( channel: Arc<Channel>, channel_id: CircId, unique_id: UniqId, input: Receiver<AnyChanMsg, MpscSpec>, runtime: DynTimeProvider, memquota: CircuitAccount, padding_ctrl: PaddingController, padding_stream: PaddingEventStream, timeouts: Arc<dyn TimeoutEstimator + Send>, ) -> (Self, UnboundedSender<CtrlMsg>, UnboundedSender<CtrlCmd>, Receiver<Void>, Arc<TunnelMutableState>)

Create a new circuit reactor.

The reactor will send outbound messages on channel, receive incoming messages on input, and identify this circuit by the channel-local CircId provided.

The internal unique identifier for this circuit will be unique_id.

pub async fn run(self) -> Result<()>

Launch the reactor, and run until the circuit closes or we encounter an error.

Once this method returns, the circuit is dead and cannot be used again.

async fn run_once(&mut self) -> StdResult<(), ReactorError>

Helper for run: doesn’t mark the circuit closed on finish. Only processes one cell or control message.

async fn try_dequeue_ooo_msgs(&mut self) -> StdResult<(), ReactorError>

Try to process the previously-out-of-order messages we might have buffered.

async fn handle_run_once_cmd( &mut self, cmd: RunOnceCmd, ) -> StdResult<(), ReactorError>

Handle a RunOnceCmd.

async fn handle_single_run_once_cmd( &mut self, cmd: RunOnceCmdInner, ) -> StdResult<(), ReactorError>

Handle a RunOnceCmd.

async fn wait_for_create(&mut self) -> StdResult<(), ReactorError>

Wait for a CtrlMsg::Create to come along to set up the circuit.

Returns an error if an unexpected CtrlMsg is received.

fn note_conflux_handshake_result( &mut self, res: StdResult<(), ConfluxHandshakeError>, reactor_is_closing: bool, ) -> StdResult<(), ReactorError>

Add the specified handshake result to our ConfluxHandshakeContext.

If all the circuits we were waiting on have finished the conflux handshake, the ConfluxHandshakeContext is consumed, and the results we have collected are sent to the handshake initiator.

fn prepare_msg_and_install_handler( &mut self, msg: Option<AnyRelayMsgOuter>, handler: Option<Box<dyn MetaCellHandler + Send + 'static>>, ) -> Result<Option<SendRelayCell>>

Prepare a SendRelayCell request, and install the given meta-cell handler.

fn handle_shutdown(&self) -> StdResult<Option<RunOnceCmdInner>, ReactorError>

Handle a shutdown request.

fn handle_shutdown_and_return_circuit( &mut self, answer: Sender<StdResult<Circuit, Bug>>, ) -> StdResult<(), ReactorError>

Handle a request to shutdown the reactor and return the only Circuit in this tunnel.

Returns an error over the answer channel if the reactor has no circuits, or more than one circuit. The reactor will shut down regardless.

fn resolve_target_hop( &self, hop: TargetHop, ) -> StdResult<HopLocation, NoHopsBuiltError>

Resolves a TargetHop to a HopLocation.

After resolving a TargetHop::LastHop, the HopLocation can become stale if a single-path circuit is later extended or truncated. This means that the HopLocation can become stale from one reactor iteration to the next.

It’s generally okay to hold on to a (possibly stale) HopLocation if you need a fixed hop position in the tunnel. For example if we open a stream to TargetHop::LastHop, we would want to store the stream position as a HopLocation and not a TargetHop::LastHop as we don’t want the stream position to change as the tunnel is extended or truncated.

Returns NoHopsBuiltError if trying to resolve TargetHop::LastHop and the tunnel has no hops (either has no legs, or has legs which contain no hops).

fn resolve_hop_location( &self, hop: HopLocation, ) -> StdResult<(UniqId, HopNum), NoJoinPointError>

Resolves a HopLocation to a UniqId and HopNum.

After resolving a HopLocation::JoinPoint, the UniqId and HopNum can become stale if the primary leg changes.

You should try to only resolve to a specific UniqId and HopNum immediately before you need them, and you should not hold on to the resolved UniqId and HopNum between reactor iterations as the primary leg may change from one iteration to the next.

Returns NoJoinPointError if trying to resolve HopLocation::JoinPoint but it does not have a join point.

pub(crate) fn target_hop_to_hopnum_id( &self, hop: TargetHop, ) -> Option<(UniqId, HopNum)>

Resolve a TargetHop directly into a UniqId and HopNum.

This is a helper function that basically calls both resolve_target_hop and resolve_hop_location back to back.

It returns None on failure to resolve meaning that if you want more detailed error on why it failed, explicitly use the resolve_hop_location() and resolve_target_hop() functions.

fn set_padding_at_hop( &self, hop: HopLocation, padder: Option<CircuitPadder>, ) -> Result<()>

Install or remove a padder at a given hop.

fn uses_stream_sendme(&self, leg: UniqId, hop: HopNum) -> Option<bool>

Does congestion control use stream SENDMEs for the given hop?

Returns None if either the leg or hop don’t exist.

async fn handle_link_circuits( &mut self, circuits: Vec<Circuit>, answer: Sender<Result<Vec<Result<(), ConfluxHandshakeError>>>>, ) -> StdResult<(), ReactorError>

Handle a request to link some extra circuits in the reactor’s conflux set.

The circuits are validated, and if they do not have the same length, or if they do not all have the same last hop, an error is returned on the answer channel, and the conflux handshake is not initiated.

If validation succeeds, the circuits are added to this reactor’s conflux set, and the conflux handshake is initiated (by sending a LINK cell on each leg).

NOTE: this blocks the reactor main loop until all the cells are sent.