Fix #3076 parEvalMap resource scoping

[reardonj]

Updates parEvalMap*, broadcastThrough and prefetch to extend the resource scope past the channel/topic used to implement concurrency for these operators.

In all cases the solution is the same:

1. use underlying.uncons.flatMap to get into a Pull context with access to the source stream
2. setup the new foreground stream with Pull.extendScopeTo to pull the scope across to the new stream

I don't see an obvious way to extract this to something general. The extra finalization in the background stream for parEvalMapUnorderedUnbounded means I can't just make a concurrently variant that takes the background stream, since the foreground needs to consume the transformed stream, while only taking the original stream's scope.

Edit: I have abstracted this solution to a new extendScopeThrough method, which works as through except propagating the scope to the new stream. It appears we should be doing this for any stream combinator where the resulting stream is not directly derived from the current stream (e.g. any combinator that uses an internal buffer, channel, topic, or such).

I also had to fix the cancallation safety of extendScopeTo (see #3474). The StreamSuite "resource safety test 4" started failing after my changes, presumably because the scope started propagating far enough to actually get hit by cancellation.

There is at least conflateChunks broken the same way but I'd like to validate this solution is correct before I continue trying to chase down all the places the fix should be applied.

[yurique]

Is it guaranteed to be safe (in the context of scopes) to use .peek before Pull.extendScopeTo?

[reardonj]

I'm assuming scopes work properly for simple streams. 🤞 Do you have a specific scenario in mind?

The fundamental issue is that I need to get ahold of a scope before we go adding more finalizers to source to avoid deadlocking in parEvalMap and I don't see a way to do that without uncons. peek is just a convenience helper.

[yurique]

Do you have a specific scenario in mind?

No, not really, it's just me being paranoid :)

I don't see a way to do it without uncons either, I was thinking about swapping the order of those Pulls:

def extendScopeThrough[F2[x] >: F[x], O2](
      f: Stream[F, O] => Stream[F2, O2]
  )(implicit F: MonadError[F2, Throwable]): Stream[F2, O2] =
    Pull
      .extendScopeTo(this.covary[F2])
      .flatMap { stream =>
        stream.pull.peek.flatMap {
          case Some((_, tl)) => f(tl).underlying // <---------------
          case None          => f(Stream.empty).underlying
        }
      }
      .stream

but the problem is tl is not a Stream[F, O] (it's a Stream[F2, O]), and .covary[F2] at the beginning has to be there because we only have MonadError for F2.

[reardonj]

Just make it f: Stream[F2, O] => Stream[F2, O2]. Requires type annotations at the call sites, but otherwise, this version also passes the tests.

Unless we can find an observable difference I'd rather keep the current version to avoid the extra annotations.

[yurique]

I agree :)

[reardonj]

This solution is inadequate. It only handles scopes open at the first pull.

[yurique]

@reardonj

wouldn't the Scope.lease (that is used in the Pull.extendScopeTo) handle the subsequent (child) scopes?

Leases the resources of this scope until the returned lease is cancelled.
Note that this leases all resources in this scope, resources in all parent scopes (up to root)
and resources of all child scopes.

(unless, of course, I'm mis-understanding what the above means)

[yurique]

I tried some ridiculous things like the following:

  fs2.Stream.unit
    .covary[IO]
    .flatMap { _ =>
      fs2.Stream.bracket {
        IO.println("making resource 1").as("TempFile")
      } { res =>
        IO.println(s"!! releasing resource 1: $res")
      }
    }
    .flatMap { _ =>
      fs2.Stream.bracket {
        IO.println("making resource 2").as("TempFile")
      } { res =>
        IO.println(s"!! releasing resource 2: $res")
      }
    }
    .parEvalMap(2) { _ =>
      IO.println("eval")
    }
    .flatMap { _ =>
      fs2.Stream.bracket {
        IO.println("making resource 3").as("TempFile")
      } { res =>
        IO.println(s"!! releasing resource 3: $res")
      }
    }
    .parEvalMap(2) { _ =>
      IO.println("eval 2")
    }
    .flatMap { _ =>
      fs2.Stream.bracket {
        IO.println("making resource 4").as("TempFile")
      } { res =>
        IO.println(s"!! releasing resource 4: $res")
      }
    }
    .parEvalMap(2) { _ =>
      IO.println("eval 3")
    }
    .compile
    .drain
    .unsafeRunSync()

and it works like a clock

making resource 1
making resource 2
eval
making resource 3
eval 2
making resource 4
eval 3
!! releasing resource 4: TempFile
!! releasing resource 3: TempFile
!! releasing resource 2: TempFile
!! releasing resource 1: TempFile

[reardonj]

@yurique , try a stream that creates multiple scopes. I just pushed the test that I broke based on the Network[IO].server conversation:

      Stream(1, 2, 3, 4, 5, 6)
        .flatMap(i => Stream.bracket(Deferred[IO, Int])(_.complete(i).void)) // 1
        .parEvalMap(2)(d => IO.sleep(1.second) >> d.complete(0))
        .evalMap(completed => IO.raiseWhen(!completed)(new RuntimeException("already completed")))
        .timeout(5.seconds)
        .compile
        .last
        .assertEquals(Some(()))

As I understand, each element passing through // 1 opens another scope, but only the first scope is actually propagated