Proposal: stream resource ownership transfer

[RaasAhsan]

Supersedes #2289 and #2300.

Just to summarize a bunch of GitHub issues and Gitter conversations around fs2/http4s: the main objective here is to arrive at a safe, network API which doesn't couple the lifecycles of server and socket handler streams. This is really useful when we want to build network servers which gracefully shutdown, where we stop accepting new connections and allow remaining connections to run for some time.

This is achievable in CE2/FS2 with a leaky SocketGroup.serverResource in conjunction with a custom forking combinator that doesn't lease resources. #2300 demonstrates it was possible in CE3/FS3 by writing a concurrent data structure Shared that negotiates resource transfer between two streams. That approach was unfavorable since it exposes something on a public API which would be redundant once this functionality is built out.

So what we would like instead is a safe, Stream-native mechanism that can negotiate resource ownership transfer between two (or more) streams that can be leveraged via a set of combinators. I think most of the infrastructure we need is already in place with Scope and ScopedResource, which we just need to augment a little bit to get what we want.

---

Approach 1

The main idea for this approach is to designate some resources as "transferable". Transferable resources contrast with ordinary, nontransferable resources in the sense that they can be transferred between scopes on-demand. Other than that, they are identical; transferable resources can be acquired, leased, released, etc.

Once we have those two pieces, we can write a combinator like parJoin that transfers resources instead of leasing.

object Stream {
  def resourceTransferable[F[_], O](r: Resource[F, O])(implicit F: MonadCancel[F, _]): Stream[F, O]
  def resourceTransferableWeak[F[_], O](r: Resource[F, O])(implicit F: MonadCancel[F, _]): Stream[F, O]

  def forking[F[_], O](streams: Stream[F, Stream[F, O]], maxOpen: Int)(implicit F: Concurrent[F]): Stream[F, Nothing]
}

trait Scope[F[_]] {
  // Transfers all the transferable resources in this scope to the target scope
  def transfer(target: Scope[F]): F[Unit]
}

The SocketGroup public API in fs2.io.tcp now remains the same, however, the individual Sockets can now be transferred to streams.

Approach 2

Same general idea as above, but the API corresponds to Lease:

trait Scope[F[_]] {
  // Transfers all the transferable resources in this scope to the target scope
  def transfer: Transfer[F]
}

trait Transfer[F[_]] {
  def transferToScope(scope: Scope[F]): F[Unit]
}

Approach 3

This approach exploits the fact that Stream already has a resource transfer mechanism in the form of leasing. One problem is that leasing is greedy; it may acquire more resources than we would like. Another problem is that leasing itself inhibits stream termination, even if we don't need any of the scoped resources. In the context of server sockets, this basically means that all socket handler streams lease the server resource, which prevents the server stream from finalizing.

I haven't really fleshed this out much, and I'm not sure how I feel about touching leasing, but the idea would be to expose a weaker form of leasing that only selectively leases resources that we're interested in. We need a way to designate which resources those are, and I think something like resourceTransferable may work.

---

I'm going to continue refining these approaches, but here are some questions we should think about:

1. Should transfer only be allowed once, an arbitrary number of times, or unlimited times? The immediate use case only necessitates single transfer.
2. Is it preferable to be able to reuse parJoin with transfer semantics?
3. What does safety look like? In particular, are there scenarios in which a resource can be used after it has been released? If so, are there ways to hide that in an API?
4. How does transfer interact with leasing? i.e. should they commute
5. Should all SocketGroup implementations provide uniform transferable semantics? Ideally, we could swap out the fs2 standard library for fs2-netty and things should just work. However, if fs2-netty doesn't provide transfer semantics, and we try to use forking with it, the server will break down and fail to process any connections. Arguably a benefit of the Shared approach is that those semantics were enforced by typing.

I'm probably missing better approaches, so please comment them here :)

[mpilquist]

All seem like good options for investigation. Approach 1 seems simplest to me without thinking too much about tradeoffs.