Add peek method to all queue interfaces and implementations

[sbly]

This implements #2639

This only adds methods, but it seems it may break backwards-compatibility according to that discussion. Is that because it adds methods to an interface?

[armanbilge]

Thanks for the PR!

Is that because it adds methods to an interface?

Right, exactly. We can only add new methods if we also can provide a default implementation. Unfortnately I don't think we can do that here.

Also, I was a bit confused about the value of peek for a concurrent queue, see https://github.com/typelevel/cats-effect/issues/2637#issuecomment-991954827, but that's just my opinion :)

[sbly]

@armanbilge No problem! Thanks for the explanation

Yeah you're right that it might not be that useful in concurrent situations. It seems like the original motivation was simply to make the migration to CE3 easier. That said, if it's decided that this isn't worth it i'm OK with that too

[armanbilge]

It seems like the original motivation was simply to make the migration to CE3 easier. That said, if it's decided that this isn't worth it i'm OK with that too

It's certainly worth considering if we can provide a default implementation, so that this change is binary-compatible :)

[sbly]

@armanbilge Unfortunately, I don't see a reasonable way to do that. Even tryTake.flatMap(tryOffer) wouldn't work. Any suggestions?

[LMnet]

I want to add a few thoughts about necessity of the peek method. I have a pretty complex concurrent component in my codebase which is heavily rely on a this method. I will not go too much into details about my component, but in general it uses peek to wait for the first new element in an empty queue. And after that it use take to get an actual element. I can't use take without a peek, because I have more than 1 queue, and I want to get an element only from the first filled queue. With take I will get elements from all queues, but it's not what I want.

So, peek is usable sometimes. Although, for my specific use-case method like waitForNonEmpty will be enough.

[armanbilge]

@LMnet If I understand your problem correctly, I think you are looking for race :)

IO.race(queue1.take, queue2.take)

The first queue that gets an element will win the race and take successfully, and the other take will be cancelled.

If you want to race take on multiple queues you can have a look at https://github.com/typelevel/cats-effect/issues/512#issuecomment-1072337740.

[djspiewak]

Sorry for the delay in getting to this…

My concern about this method is there really isn't a way to do it atomically in a default implementation. Now, seeing as we control all realistic Queue implementations, maybe that's not such a huge concern, but it's definitely something to think about. Are we okay with it being non-atomic (and thus potentially corrupting ordering and/or fiber blocking) on third-party implementations of Queue?

[armanbilge]

Now, seeing as we control all realistic Queue implementations, maybe that's not such a huge concern

If we really feel that way, a "safer" way forward might be to seal Queue in Cats Effect 3.4, but not make any other changes. This will be source-breaking, but not binary-breaking. Assuming that nobody complains about this, I'd say from 3.5 forward we can freely add new abstract methods to the Queue interface, including peek.

[LMnet]

@armanbilge since it's off-topic, I have hided my answer under a spoiler

If I understand your problem correctly, I think you are looking for race :)

In theory it should work. But in practice IO.race(queue1.take, queue2.take) will return the first element, but will not cancel the other take. Because take is uncancellable inside. You can check it through the simple test:

for {
  queue1 <- Queue.unbounded[IO, String]
  queue2 <- Queue.unbounded[IO, String]
  _ <- queue1.offer("1")
  _ <- queue2.offer("2")
  res <- IO.race(queue1.take, queue2.take)
  size1 <- queue1.size
  size2 <- queue2.size
} yield {
  assert(size1 + size2 == 1)
}

This test will always fail, because both take will dequeue elements from their queues. But because of race only one of the results will be returned. Another one will be dropped.

Maybe it's somehow possible to support cancellation in the queue, but now it's not working.

[djspiewak]

If we really feel that way, a "safer" way forward might be to seal Queue in Cats Effect 3.4, but not make any other changes. This will be source-breaking, but not binary-breaking. Assuming that nobody complains about this, I'd say from 3.5 forward we can freely add new abstract methods to the Queue interface, including peek.

I mean, not really. :-) It still breaks bincompat with anything that was compiled before the sealed modifier was added. In particular, anyone calling a hypothetical abstract peek on Queue where the underlying implementation predates the addition of the sealed modifier would get a link error.

[armanbilge]

In particular, anyone calling a hypothetical abstract peek on Queue where the underlying implementation predates the addition of the sealed modifier would get a link error.

Is that better or worse than silently corrupting said Queue? :)

[LMnet]

Is there any chances that this pr will be merged and released with the next patch version?

[Jasper-M]

IMO the method being added here should be called tryPeek. peek should be like take without actually removing the element.

[djspiewak]

I genuinely don't think that there's a way this can be done safely in the current API. Given the tradeoffs involved in this type of function, I would rather it exist in userspace (q.tryTake.flatMap(_.traverse(q.offer(_))).uncancelable), where those tradeoffs are more directly visible.