`Queue.synchronous` may suffer from the "lost item" problem

[djspiewak]

Specifically, when a take is canceled at the moment an offer completes, the value may be lost.

[djspiewak]

I'm not convinced this is actually happening:

    "not lose offer when taker is canceled during exchange" in real {
      val test = for {
        q <- Queue.synchronous[IO, Unit]
        latch <- CountDownLatch[IO](2)
        offererDone <- IO.ref(false)

        _ <- (latch.release *> q.offer(())).guarantee(offererDone.set(true)).start
        taker <- (latch.release *> latch.await *> q.take).start

        _ <- latch.await
        _ <- taker.cancel

        // we should either have received the value successfully, or we left the value in queue
        // what we *don't* want is to remove the value and then lose it due to cancelation
        oc <- taker.join

        _ <- if (oc.isCanceled) {
          // we (maybe) hit the race condition
          // if we lost the value, q.take will hang
          offererDone.get.flatMap(b => IO(b must beFalse)) *> q.take
        } else {
          // we definitely didn't hit the race condition, because we got the value in taker
          IO.unit
        }
      } yield ok

      test.parReplicateA_(1000000).as(ok)
    }

This test really should reproduce the problem if it exists, but it doesn't seem to do so.

Update: I was wrong. It's happening. Just needed to adjust the test a bit to make the offerer await the latch.

[durban]

In this comment I described why I think there is still a problem (although not exactly a "lost item"), but I'm not sure if this issue needs reopening...