Unexpected behavior with merging a traced Stream with another Stream

[ryanmiville]

Merging a stream (.merge, .mergeHaltBoth, etc.) that uses bayou for tracing with another stream breaks the usage of child spans.

I know this project is just experimental, but I am also following the discussion of adding similar functionally to natchez so I figured I would share anyway. I have not pulled down and tested Ross' branch to verify the same behavior.

Merging a stream that uses bayou with another stream (even untraced streams) causes the underlying IOLocal to not properly update its state. I see no errors, but when I step through the debugger, the IOLocal will call set with the new Span[IO], but the next get still returns the default span. so every Trace[F] method acts upon the original root span used when instantiating the IOTrace. If the same stream is not merged with another, you can create and nest child spans as you would expect.

Here is an example of a single stream with its tracing output, which matches my expectations. (scala-cli https://gist.github.com/ryanmiville/6e4d756eceac70f9919c1de859e48f5a)

And here is the exact same stream merged with another stream, along with its traces. (scala-cli https://gist.github.com/ryanmiville/71d06f76dfe7c62ab96f342f5bed72b5)

Unfortunately, I don't know much about the inner workings of fs2 or IOLocal to provide any insight.

I don't expect you to really do anything about it, but I just wanted to bring it up in case the same behavior ends up making it into natchez, so it can at least be documented.

[armanbilge]

Yes, no worries, thank you very much for trying this out and writing this up! Pinging @rossabaker and @SystemFw so they can have a look as well.

[rossabaker]

Thanks! I've been a little queasy about Trace[IO], and this makes me much more queasy. IOLocal values are visible on the fiber we set them, and are passed to any fibers we start. But they don't magically make it through more complicated types. If the stream interpreter hops fibers between these two lines, "child" will not be created from span (aka "new_root"), but from whatever trace was initialized with ("root"):

            _    <- Stream.resource(trace.spanR(span))
            _    <- Stream.resource(trace.spanR("child"))

I don't really know what to do about this.

[SystemFw]

Generally speaking I'm queasy on *Local because of this type of issues, but also queasy on the alternative (tagless + Kleisli) because it raises the complexity floor so high for a non-core business logic like tracing that it's unjustifiable outside of highly skilled teams for me...now if we're lucky, this problem might be specific to merge, but I suspect Ross' general point will hold

[SystemFw]

For context, all concurrent combinators in fs2 follow a roughly similar pattern:

• create a Channel or Queue
• Add some code to the input streams to manage stream level interruption (typically with onFinalize and interruptWhen)
• compile.drain.start the Streams
• return the output Stream out of that Queue/Channel

The fact that the different streams end up running in different fibers doesn't bode too well for IOLocal propagation, but I haven't done a full analysis on the issue. It'd also be interesting to check what happens with Resource, although the situation there is both simpler, and offers more tools for control like allocatedCase

[rossabaker]

Tracing resources with IOLocal seems ... pretty good? I posted a full gist, but the intersting bits are:

    trace.use { implicit trace =>
      def io(name: String): IO[Unit] =
        trace.span(name)(
          IO(Thread.currentThread.getName)
            .flatMap(t => trace.put("thread" -> t))
        )

      val acquire = trace.span("acquire")(IO.unit)
      val release = trace.span("release")(IO.unit)
      val r = trace.spanR("resource") >> Resource.make(acquire)(_ => release)
      r.use(_ => trace.span("use")(IO.race(io("a"), io("b")).void))
    }

acquire, release, and use are children of resource. a and b are raced on different threads, but still are children of use.

---

The worst I've been able to do is this intentionally stupid thing, where we allocate resources and release them in an unnatural order:

    trace.use { implicit trace =>
      for {
        a <- trace.spanR("a").allocated
        b <- trace.spanR("b").allocated
        _ <- trace.put("1" -> "1")
        _ <- a._2
        _ <- trace.put("2" -> "2")
        _ <- b._2
        _ <- trace.put("3" -> "3")
      } yield ()
    }

a and 2 end up under root. 1, 3, and b all vanish. I clearly voided the warranty here, but I left trace in a horrible state.

[armanbilge]

I've also been experimenting, to see how hard it is to break Stream tracing. So far I've minified down to 2ca914e which basically inserts a Queue between two Streams.

    def run: IO[Unit] = {
      trace.use { implicit trace =>
        foo(stream).compile.drain
      }
    }

  def foo[F[_], O](
    s: Stream[F, O]
  )(implicit F: Concurrent[F]): Stream[F, O] =
    Stream.eval(Queue.unbounded[F, Option[O]]).flatMap { q =>
      Stream.eval(s.enqueueNoneTerminated(q).compile.drain.start).flatMap { _ =>
        Stream.repeatEval[F, Option[O]](q.take).unNoneTerminate
      } 
    }

    val entryPoint = Log.entryPoint[IO]("example")

    val trace = entryPoint.root("root").flatMap(r => Resource.eval(Trace.ioTrace(r)))

    def stream(implicit trace: IOTrace): Stream[IO, Unit] = {
        Stream(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
          .covary[IO]
          .chunkN(3)
          .flatMap { chunk =>
            for {
            _    <- Stream.resource(trace.spanR("green kangaroo"))
            _    <- Stream.resource(trace.spanR("child"))
            _    <- Stream.eval(trace.put("vals" -> chunk.toList.mkString(",")))
            _    <- Stream.sleep[IO](1.second)
            } yield ()
          }
    }

Gives:

{
  "name" : "root",
  "service" : "example",
  "timestamp" : "2022-03-09T01:53:38.047505Z",
  "duration_ms" : 4523,
  "trace.span_id" : "fec8a9f6-9c78-43e1-9de5-e909406f5819",
  "trace.parent_id" : null,
  "trace.trace_id" : "b3396a93-4342-452e-8d53-88653d9e2bed",
  "exit.case" : "succeeded",
  "vals" : "10",
  "children" : [
    {
      "name" : "child",
      "service" : "example",
      "timestamp" : "2022-03-09T01:53:38.475466Z",
      "duration_ms" : 1036,
      "trace.span_id" : "b3b5586e-1573-418f-aaf6-ec249150acdb",
      "trace.parent_id" : "b3396a93-4342-452e-8d53-88653d9e2bed",
      "trace.trace_id" : "b3396a93-4342-452e-8d53-88653d9e2bed",
      "exit.case" : "succeeded",
      "children" : [
      ]
    },
    {
      "name" : "green kangaroo",
      "service" : "example",
      "timestamp" : "2022-03-09T01:53:38.466419Z",
      "duration_ms" : 1074,
      "trace.span_id" : "a0965406-9196-4671-940b-dea74a9e0e8d",
      "trace.parent_id" : "b3396a93-4342-452e-8d53-88653d9e2bed",
      "trace.trace_id" : "b3396a93-4342-452e-8d53-88653d9e2bed",
      "exit.case" : "succeeded",
      "children" : [
      ]
    },
    ...
  ]
}

So sadly, child is not nested under green kangaroo. I really cannot understand why.

[armanbilge]

Ok, minimized down to noneTerminate ...

def run: IO[Unit] = {
  trace.use { implicit trace =>
    stream.noneTerminate.compile.drain
  }
}

Update:

Yikes. Minimized to this?

def run: IO[Unit] = {
  trace.use { implicit trace =>
    stream.map(_ => ()).compile.drain
  }
}

[rossabaker]

stream.map(identity).compile.drain

[rossabaker]

It calls this:

  private[fs2] def mapOutput[F[_], O, P](s: Stream[F, O], f: O => P): Pull[F, P, Unit] =
    interruptScope(mapOutputNoScope(s, f))

And when I reimplement mapOutputNoScope, it's fine. interruptScope sounds ominous.

[rossabaker]

stream.interruptScope.compile.drain does it. I can confirm that it's releasing resources in the reverse order they're acquired, but the child is not seeing the same IOLocal state as the parent. I suspect it's in InterruptionContext, but I'm not familiar enough to have an idea what can be done about it.

[armanbilge]

Thanks. Minimized to this:

//> using scala "3.1.1"
//> using lib "co.fs2::fs2-core:3.2.5"

import cats.effect._
import fs2._

object Bayou extends IOApp.Simple:
  // def run = s.compile.lastOrError.flatMap(IO.println) // List(2, 1)
  def run = s.interruptScope.compile.lastOrError.flatMap(IO.println) // List()

  def s = Stream.eval(IOLocal(List.empty[Int])).flatMap { local =>
    Stream.eval(local.update(1 :: _)).flatMap { _ =>
      Stream.eval(local.update(2 :: _)).flatMap { _ =>
        Stream.eval(local.get)
      }
    }
  }

[armanbilge]

Ok, I think we more-or-less got to the bottom of it in typelevel/fs2#2842 (comment). Whenever interruption is involved, so are multiple fibers. And whenever multiple fibers are involved, you are liable to lose these sorts of modifications to fiber-local state once the fiber is completed/discarded.

[armanbilge]

It looks like there's some hope to fix this in fs2 :) but, what it means for Bayou is that it would probably specifically need a spanS(...): Stream[F, Unit]-style method to make the magic invocations on the Stream.

[ryanmiville]

it would probably specifically need a spanS(...): Stream[F, Unit]-style method

FWIW I have already been adding a spanS as a convenience method in my project. To echo @SystemFw's sentiment, tracing should be a frictionless addition to our codebase, so I personally welcome specific methods for specific use cases. :)