Fs2: Timed operations for Queue

[dmgcodevil]

Where is timed operations implemented in this PR https://github.com/functional-streams-for-scala/fs2/pull/924 ?

[SystemFw]

You don't need them anymore, you can directly call .timeout on enqueue1, dequeue1

[dmgcodevil]

@SystemFw if you are talking about IO.timeout then it doesn't guarantee that the element wont added if enqueue1 has timed out. take a look at the code below:

object RaceSpec extends IOApp {

  val counter = new AtomicInteger()

  val inc1 = IO.sleep(100.millis) >> IO(counter.incrementAndGet())
  val inc2 = IO.sleep(100.millis) >> IO(counter.incrementAndGet())

  val n = 100

  override def run(args: List[String]): IO[ExitCode] = {

    (0 until n).map(_ => IO.race(inc1, inc2)).toList.sequence_ >> IO(println(counter.get)) >> IO(ExitCode.Success)

  }

}

prints to the console: 168

cats-effect contributors saying that:

so you might have a different printed value every time, but from looking at the code I reckon it'll always be between 100 and 200
because sometimes both IOs will complete at around the same time and cancelation won't hit either

because the race might cancel the loser after incrementAndGet executes

[SystemFw]

Cancelation is not magic :)

When you race, you start two fibers, and once the first increment (say inc1) terminates, it sends a signal to inc2 which is already running. So you have a race condition between inc2 waking up and incrementing and the cancelation signal reaching inc2, and non deterministically sometimes inc2 will complete before the cancelation signal can reach it.

You need to build additional synchronisation if you want to ensure that only one of the two sets something

[dmgcodevil]

@SystemFw yes, i understand that but enqueue1.timeout is broken, you might get a failed result based on this code:

  final def timeout(duration: FiniteDuration)
    (implicit timer: Timer[IO], cs: ContextShift[IO]): IO[A] =
    timeoutTo(duration, IO.raiseError(new TimeoutException(duration.toString)))

but nonetheless the element might be added into the queue. Using tryEnqueue in a loop fixed the issue, but I'd like to block asynchronously rather that burn cpu .
imho timed operations should be supported natively in Queue, otherwise it might lead to unpredictable results

[SystemFw]

enqueue1.timeout is not broken, it's the same thing as above.
enqueu1.timeout is race(enqueue1, IO.raiseError), raiseError wins, sends a cancelation signal but enqueue1 gets interrupted after having enqueued the element but before finishing.

Using tryEnqueue in a loop fixed the issue, but I'd like to block asynchronously rather that burn cpu

Unfortunately this is an unsolvable contradiction: concurrent interruption is _always_ a race (not in cats-effect, in life, because information does not transmit instantaneously), and the only way to make it deterministic is to remove the concurrency, i.e. a synchronous loop.

What you could do is put something in Queue than on cancelation checks if you did manage to enqueue the element and removes it, but then you have a race with someone else dequeueing that element anyway.

And even if you somehow manage to make that work for timedEnqueue1, race(timedEnqueue1 >> F.unit, anything) can be broken in the same fashion anyway.

[dmgcodevil]

but I still don't understand the behavior below.

the code below outputs :

"Completed" printed  8 times
"Canceled" printed 92 times
"inc" printed 57
counter value  is 57

i.e. delayedTask was canceled but nevertheless the source task was executed, why ?

import java.util.concurrent.atomic.AtomicInteger

import cats.effect.ExitCase.{Canceled, Completed}
import cats.effect.{ExitCode, IO, IOApp}
import cats.implicits._

import scala.concurrent.duration._

object RaceSpec extends IOApp {

  val counter = new AtomicInteger()

  val n = 100

  def inc(): Unit = {
    println("inc")
    counter.incrementAndGet()
  }

  override def run(args: List[String]): IO[ExitCode] = {
    (0 until n).map(_ => run(inc())).toList.sequence_ >> IO(println(counter.get)) >> IO(ExitCode.Success)
  }

  def run(task: => Unit): IO[Unit] = {
    val delayedTask = IO.sleep(100.millis) >> IO(task)
    for {
      _ <- IO.race(
        delayedTask.guaranteeCase {
          case Completed => IO(println("Completed"))
          case Canceled => IO(println("Canceled"))
        },
        IO.sleep(100.millis))

    } yield ()
  }
}

[SystemFw]

Canceled means "it received the cancelation signal".
In this case, there is a race.
If delayedTask receives the cancelation signal during sleep, or between sleep and IO(task), it gets Canceled and the counter does not get incremented.
If delayedTask receives the cancelation signal during IO(task), that cannot be interrupted (it's a synchronous task) and so you get Canceled but it's too late to stop the counter being incremented

[dmgcodevil]

I see. looks like there is nothing I can do but redesign my code to address duplicated events. thanks a lot.

[SystemFw]

Unfortunately yes

[SystemFw]

What I do in a similar scenario is keep track of valid events through a Ref (AtomicReference basically), so that duplicated events are not canceled but become no-ops. If you look at the implementation of groupWithin here in fs2 you'll see what I mean (warning, it's a complicated combinator for several other reasons)

[dmgcodevil]

that would be really tricky to implement in my case, but I don't see other options. BTW is possible to implement _correct_ timed operations for Queue ?

[SystemFw]

Not for your definition of correct because you always have a race between the sleep completing and trying to interrupt the enqueueing. Again, this is just a limitation of reality because information takes some time (however short) to transmit. The only way is to have a synchronous loop that tries to enqueue and checks the time, tries to enqueue and checks the time, and so on, because you don't have concurrency and interruption, but now you're wasting cycles, and you might still consider the lack of resolution as _incorrect_ (e.g. I only checked for interruption after 101 millis because I started enqueuing at 99 millis and it took me 2 millis, and I wanted a timeout at 100 millis).
Also just fyi, a timed enqueue is supposed to be used with a bounded queue where there is actually some asynchronous waiting if the queue is full. If the queue is not full enqueue1 is basically uncancelable (very fast essentially atomic operation), which will expose you to this kind of race more often.

[dmgcodevil]

yes, I'm using bounded queue. however java's blockingqueue provides offer(e, time, unit), I know that it uses locks and condition variables and bounded Queue in fs2 might have a completely different implementation, however I'm not aware of any lock-free (based on atomic operations) implementations of bounded blocking queues. looks like it's impossible in theory
https://stackoverflow.com/questions/46482378/is-there-a-bounded-lock-free-blocking-queue

[SystemFw]

JDK doesn't provide a queue implementation that, (1) is mostly lock-free, (2) doesn't require busy waiting, and (3) bounded.

This is what fs2 bounded queue does, although it kinda depends on what you mean by lock-free.
The definition in that question:

Does anyone know a queue implementation that is mostly lock-free (doesn't use a lock in the hot path),

Yes, fs2 Queue checks this (no locks)

blocks when empty (no need to busy waiting)

Yes, fs2 Queue checks this (uses semantic blocking so the underlying JVM threads is free to do other things)

and is bounded (blocking when full)?

yes, fs2 Queue checks this (with semantic blocking as well)

[dmgcodevil]

so , in other words , you somehow managed to implement mostly lock-free (no locks used, no threads blocked) bounded queue which _blocks_ when full ? is there any documentation about underlying algorithm / implementation ? which algorithm did you use as a starting point ? classical Simple, Fast, and Practical Non-Blocking and Blocking Concurrent Queue Algorithms by Michael L. Scott ?

[SystemFw]

I'll think about whether there is a way to implement what I think it is you want, i.e. that you never get a "this timed out" signal if you do manage to put things in the queue

[SystemFw]

as I said, it depends on what you mean by lock-free. The queue does have _blocking_, but it does not block a _thread_, just a fiber

[dmgcodevil]

Ok, I think this discussion is going beyond the original question , thanks for your help.
ps: where i can read about _blocking fibers_ ? just curious how it's implemented , particularly in cats-effect

[SystemFw]

There isn't a comprehensive description right now, you need to go read the code. I'm planning on giving a talk on how cats-effect concurrency works under the hood in a few months