RFC: Opening new streams via a context-aware, global API

[thomaseizinger]

The problem

At the moment, opening a new stream and sending data on it from within a ConnectionHandler is somewhat tedious:

1. Some action (likely a message from the behaviour) triggers the opening of a new stream. For this, we need to queue a ConnectionHandlerEvent::OutboundSubstreamRequest.
2. This event is popped off from some queue in the handler and emitted in ConnectionHandler::poll.
3. Opening of the new stream either succeeds or fails, both of them resulting in a ConnectionEvent.
4. The new stream is composed with some other data and likely added to a Futures{Set,Map} where it is polled to completion.

This is a lot of state tracking that one needs to get right.

Given the linear execution flow of the above, it would be much easier (and in line with the coding guidelines) to offer a async-based API for requesting a new stream. The problem with such an API is that it needs to be .awaited and nothing within ConnectionHandler is marked async (on purpose).

The actual execution of the protocol using the stream is however async so it could be added to the beginning of that. Think something like:

// Inside some fn of `ConnectionHandler`:

self.worker_streams.try_push(async move {
	let stream = open_stream("/my-protocol/1.0.0").await?;

	// Execute protocol on stream
});

The question is, what is open_stream? The function somehow needs a reference to libp2p_swarm::Connection, which could be achieved using a channel. Said channel would have to be passed into NetworkBehaviour::handle_established_*_connection such that it can be passed on to the ConnectionHandler implementation. That is somewhat tedious and a breaking change.

The proposal

Instead of accessing the Connection explicitly, we could access it implicitly via thread-local storage. Let's remember that:

• A Connection runs within a task that combines the channel from the behaviour.
• The Connection itself interacts with the ConnectionHandler and the muxer within its poll function.

1. Similar to how we initialize the tracing span as part of the Connection, we could set a thread-local storage variable at the beginning of poll that provides access to a channel within the Connection itself:

   thread_local! {
       static CONNECTION_CONTEXT: RefCell<...> ;
   }
   
   // Connection::poll
   fn poll(&mut self, cx: &mut Context<'_>) -> ... {
   	let new = ...; // initialize state depending on current `Connection`, maybe just the connection ID or some other reference / shared hashmap / channel so we can retrieve the requested streams again.
   
   	let old = CONNECTION_CONTEXT.replace(new);
   
   	// Set `old` back at the end of the function (probably using a RAII guard)
   }

2. open_stream would access this TLS and thereby implicitly communicate with "its" Connection instance, being able to e.g. queue a oneshot::Sender into a list for pending streams. The corresponding oneshot::Receiver would be returned from open_stream.

   fn open_stream(p: &'static str) -> impl Future<Output = Result<Stream, OpenStreamError>> {
   	let receiver = CONNECTION_CONTEXT.with_borrow_mut(|state| state.request_new_stream(p));
   
   	receiver
   }

3. The Connection would then proceed to open a new stream via the muxer, negotiate the protocol and - if successful - send the stream into the channel, back to the ConnectionHandler.

Advantages

• No state tracking for opening of new streams within ConnectionHandler, instead, any function can directly create a new Future that opens a stream, runs the protocols and returns the result.
• Smaller API surface within ConnectionHandler.
• Can we shipped in a backwards-compatible way alongside the current API. Initially, we might even want to ship it as an "experimental" API that is #[doc(hidden)] as we explore it within the codebase first.

Disadvantages

• "Global" functions can be perceived as confusing and the data flow is harder to understand.

[mxinden]

Documenting an out-of-band discussion here.

I don't think this is a good idea for the following reason:

• implicit instead of explicit
  • As a user I would ask myself: Where does the stream come from? How does it know which remote peer to open the stream with?
• global variable
  • Say that a user uses the open_stream function in an auxiliary function foo. What happens when the user unit tests this function? The above thread local context will not be available.
  • Similar what happens when I spawn a new task within my ConnectionHandler and call open_stream in that task? Again, the thread local context might not be available, as the task might be executed on a different thread.