Rewrote parTraverseN and parTraverseN_ for better performance

[djspiewak]

This shifts to a fully bespoke implementation of parTraverseN and such. There are a few things left to clean up, such as a few more tests and running some comparative benchmarks, but early results are very promising. In particular, the failure case from #4434 appears to be around two to three orders of magnitude faster with this implementation (which makes sense, since it handles early abort correctly). Kudos to @SystemFw for the core idea which makes this possible.

One of the things I'm doing here is giving up entirely on universal fairness and merely focusing on in-batch fairness. A simpler way of saying this is that we are hardened against head of line blocking, both for actions and cancelation.

Fixes #4434

[djspiewak]

Pros and cons on performance, though I think it's possible to do better here. It's a little bit slower than the previous implementation in the happy path, but it's several orders of magnitude faster in the error path so I'll call that a win.

Before

[info] Benchmark                             (cpuTokens)  (size)   Mode  Cnt    Score    Error  Units
[info] ParallelBenchmark.parTraverse               10000    1000  thrpt   10  292.924 ±  1.496  ops/s
[info] ParallelBenchmark.parTraverseN              10000    1000  thrpt   10  277.978 ±  1.280  ops/s
[info] ParallelBenchmark.parTraverseNCancel        10000    1000  thrpt   10    0.006 ±  0.001  ops/s
[info] ParallelBenchmark.traverse                  10000    1000  thrpt   10   48.015 ±  0.016  ops/s

After

[info] Benchmark                             (cpuTokens)  (size)   Mode  Cnt    Score   Error  Units
[info] ParallelBenchmark.parTraverse               10000    1000  thrpt   10  293.834 ± 1.152  ops/s
[info] ParallelBenchmark.parTraverseN              10000    1000  thrpt   10  233.868 ± 0.309  ops/s
[info] ParallelBenchmark.parTraverseNCancel        10000    1000  thrpt   10    7.859 ± 0.014  ops/s
[info] ParallelBenchmark.traverse                  10000    1000  thrpt   10   48.059 ± 0.014  ops/s

[djspiewak]

So I haven't golfed the failure down yet, but it really looks like we're hitting a bug in Scala.js, probably stemming from the "null safe" test. @durban you may be amused

I think we could just remove the null safe test now since we're not using an ArrayBuffer internally, but it's kind of a neat surprise.

[durban]

Well, "amused" is one word for it :-) So it's not a bug in Scala.js, as in, it behaves as documented: dereferencing null is undefined behavior in Scala.js (LOL, what? Seriously.), so literally any behavior is "behaving as documented". Apparently scalaJSLinkerConfig could be configured to behave properly for nulls. But removing that very specific test is also fine I think.

[djspiewak]

Well that's fun. I actually thought we had some special checking for when the cur0 action became null in the runloop, but apparently not.

[durban]

That we do check. It's this line: https://github.com/typelevel/cats-effect/blob/series/3.x/core/shared/src/main/scala/cats/effect/IO.scala#L2024 (and fa is null).

[djspiewak]

Ahhhhhhh that makes sense. Okay, by that token, I think it's fair to say that a lot of our combinators just aren't null-safe and that's how it's going to be. :P

[durban]

It's annoying, because in Scala they are null safe. (The test passed before, it just failed on JS.) We'd have to do something like this (everywhere), to make it work on JS:

def combinator(fa: F[A], ...) = {
  if (fa eq null) throw new NullPointerException
}

Which is (1) annoying, (2) very redundant, except on Scala.js, and (3) apparently has performance problems in Scala.js (or maybe that's only the linker setting?).

I don't propose we do this. There is a Scala.js linker setting which fixes the problem. In Scala and Scala Native it works by default.

[durban]

(Just some context about the null test you've removed: I've added that on my old branch, because I've had a bug previously in my implementation, where it didn't handle correctly if f(a) was null. I don't remember exactly, but I think it just ignored it. So I've added the test to make sure we see the NPE. As I've said, I think it's fine to remove it here.)

[mr-git]

Could this fix hit 3.6.4?

[djspiewak]

Could this fix hit 3.6.4?

There are a couple failing tests related to early termination that I'm still trying to track down. Am trying to find the spare time needed to push on it. Help definitely welcome! Otherwise I'll probably get to it within the next few weeks. Sorry :(

[domaspoliakas]

I see that last CI is green, do you mean that you want to reintroduce the tests removed in this commit? 599b790

[djspiewak]

@durban Would you mind taking another look here? I think I got this all sorted out.

[durban]

Yes, I'll take another look (hopefully sometime this week).

[durban]

@djspiewak I've pushed 2 failing tests. The idea is the same for them: let every fiber start; then one of them self-cancels/fails; let awaitAll observe this through preempt; and the problem is, awaitAll succeeds in this case, and thus the fibers will not be cancelled.

(I've also added another comment.)

[durban]

Minor thing, but this means we can get cancelled when we're already done. If work completes successfully, i.e., we're literally done, there is no reason really, to observe a cancellation. (It feels a little bit like the timeout changes.)

[djspiewak]

This is also a correctness issue actually because we could lose data. I'll correct it.

[durban]

Also: I thought the idea was, that there is no need to release the semaphore in case there is an error/cancel, because we're shutting down all the things anyway... This latest versions seems to release the semaphore always.

(Also: sorry for pushing to your branch, I've just didn't feel like going through the ceremony of opening a PR-for-the-PR. You can obviously roll back if I've messed up something.)

[djspiewak]

Also: I thought the idea was, that there is no need to release the semaphore in case there is an error/cancel, because we're shutting down all the things anyway... This latest versions seems to release the semaphore always.

It was initially easier to structure it that way, but this is a good point. I should move this back to the happy path exclusively.

(Also: sorry for pushing to your branch, I've just didn't feel like going through the ceremony of opening a PR-for-the-PR. You can obviously roll back if I've messed up something.)

No worries at all! I think that's totally reasonable

[djspiewak]

@durban Ready for another look

[durban]

@djspiewak I looked over parTraverseN_, left 2 comments. I'll get to parTraverseN "soon". (It's interesting how different parTraverseN and parTraverseN_ is here.)

[durban]

@djspiewak I've looked at parTraverseN too now, left more comments. (Please let me know if I'm being unclear, today is not my best day). In general, I find it interesting, that this essentially implements a queue with semaphore+start, but without an actual queue. Which is fair enough, although I'm wondering if maybe an actual queue would be easier to reason about. (I'm genuinely not sure.)

[durban]

Isn't there a possible race condition here? Let's say 2 fibers are starting in parallel:

• fiber1 is started, observes preempt empty, starts working (but fiber1 is not yet inserted into supervision, that happens concurrently)
• fiber2 is started, encounters an error, so:
• fiber2 is completing preempt, and calling cancelAll
• however, fiber1 is not yet in supervision, so cancelAll will not cancel it. That's not good.

[djspiewak]

Oh that's an interesting observation. I think you're right.

[djspiewak]

        "interrupt never on error" in ticked { implicit ticker =>
          case object TestException extends RuntimeException
          val test = List(0, 1).parTraverseN(2) {
            case 0 => IO.never[Unit]
            case 1 => IO.raiseError(TestException)
          }

          test.attempt.void.parReplicateA_(100000) must completeAs(())
        }

Weirdly this passes. It feels like we need a double-check on preempt here but the evidence doesn't seem to bear that out.

[durban]

Yeah, it is weird that doesn't hang. However this variant does hang for me (at least I think Succeeded(None) means a hang for ticked):

        "interrupt never on error" in ticked { implicit ticker =>
          case object TestException extends RuntimeException
          val test = (0 to 10).toList.parTraverseN(2) { i =>
            if ((i % 2) != 0) IO.raiseError(TestException)
            else IO.never[Unit]
          }

          test.attempt.void.parReplicateA_(100000) must completeAs(())
        }

[durban]

I'm sorry, please disregard my previous comment. As written, that test is expected to sometimes hang.

[djspiewak]

Why would this test sometimes hang? I think I agree with you that it should pass

[durban]

So, about my previous test here, with the odd-even never/raiseError: I think it's allowed to hang, because if scheduling happens to work out in a way that 2 instances of never get the 2 available permits, then hanging is expected.

However, since then, I've found this, which also hangs, but really shouldn't:

        "interrupt never on error" in ticked { implicit ticker =>
          case object TestException extends RuntimeException
          val test = (0 to 10).toList.parTraverseN(2) { i =>
            if (i == 5) IO.never[Unit]
            else IO.raiseError(TestException)
          }

          test.attempt.void.parReplicateA_(100000) must completeAs(())
        }

(There is only 1 never here, but 2 permits.)

[djspiewak]

Excellent minimization. It seems very plausible to me that this reproduces the lack of a double-check after adding to the supervision set.

[djspiewak]

Fascinatingly, adding a double-check doesn't fix it. I think this test is actually identifying something else.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request