Close H2Stream readBuffer on data.endStream

[armanbilge]

Fixes #7146. The Channel introduced in #7096 was not being closed in the right place.

[armanbilge]

The test revealed another bug 😅

[danicheg]

Not an Ember expert, but changes look good to me

[danicheg]

Probably it's kind of bikeshedding, but is it intentional that H2Stream#receiveData doesn't close readBuffer on cancelation?

[armanbilge]

It's a good question, cancelation is handled in various places by sending an exception to the readBuffer.

http4s/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Stream.scala

Lines 440 to 448 in a1687d9

	private[H2Stream] def cancelWith(msg: String)(implicit F: Monad[F]): F[Unit] = {
	// Unsure of this, but also unsure about exposing custom throwable
	val t = new CancellationException(msg)
	writeBlock.complete(Left(t)) >>
	request.complete(Left(t)) >>
	response.complete(Left(t)) >>
	readBuffer.send(Left(t)) >>
	trailers.complete(Left(t)).void
	}

[danicheg]

Yeah, but does it work for receiveData as well?

[armanbilge]

receiveData is called only from the readLoop. The readLoop runs forever when you create a server or a client: the only way to cancel it is to close the server/client resource. But to close the server/client resource you must have first closed all on-going requests.

http4s/ember-core/shared/src/main/scala/org/http4s/ember/core/h2/H2Connection.scala

Line 204 in fe80361

def readLoop: F[Unit] = {

[danicheg]

Thanks for the explanation. I can't beat my feeling suspicious about this. But this kind of leak perhaps would be noticeable.

[armanbilge]

The H2 state machine is rather complex and difficult to reason about, it's not impossible there are some bugs here. This PR and the previous one it fixes (which itself was a fix) proves that 😅 probably we are missing some test coverage in this area, that would help address your suspicions, and part of that question is how to cancel receiveData in the first place.

[Masynchin]

Hello all! I just discovered this PR and took a look at its changes. I found that whenA may be used instead of if then ... else Applicative[F].unit:

- _ <-
-    if (data.endStream) s.readBuffer.close *> s.trailWith(List.empty).void
-    else Applicative[F].unit

+ _ <- data.endStream.whenA(s.readBuffer.close *> s.trailWith(List.empty).void)

I am not that experienced at scala, so can I ask you, is whenA version more readable than if then else or it is not?

[armanbilge]

Yes, you are right, this would be a good place to use whenA: it is correct and probably more readable :)

It would look like this:

_ <- (s.readBuffer.close *> s.trailWith(List.empty)).void.whenA(data.endStream)

There is a (very small) performance penalty to use it instead of an if, which is one reason why I didn't consider changing it here.

[Masynchin]

There is a (very small) performance penalty to use it instead of an if

This performance penalty caused because of method calling rather then directly using if then else? Because the whenA's body is just if (cond) void(f) else unit. Correct me if I am wrong, I am just trying to learn more about Scala :)

[armanbilge]

The signature for whenA looks like this:

def whenA[A](cond: Boolean)(f: => F[A]): F[Unit]

The => F[A] is a "by-name" parameter. This is used to preserve laziness in the case when the condition is false.

https://docs.scala-lang.org/tour/by-name-parameters.html

What this means is that instead of directly passing F[A] to whenA, it allocates a by-name wrapper for F[A], and passes that instead. Creating this wrapper incurs a very small penalty.

[Masynchin]

Thanks for the explanation, @armanbilge!