Fix #380: Deferred, Semaphore and MVar

[alexandru]

Fixes #380.

TL;DR about the issue:

1. when we have an acquire and release pair, like we have in Deferred (get / complete), Semaphore (acquire / release) and MVar (take / put), all of them being similar, the release part can block indefinitely in case the bind continuation of acquire is a never ending loop
2. the fix would be for acquire to always fork after an await, however we do not have shift available because we don't have a ContextShift restriction specified
3. F.unit.start.flatMap(_.join) is NOT a replacement for shift

The solution is, when notifying consumers that are waiting for a value, to start and then forget about the result, something like this:

def notifyReaders(a: A, r: Iterable[A => Unit]): F[Unit] = {
  r.foldLeft(F.unit) { (acc, cb) => 
    acc.flatMap { _ =>
      F.delay(cb(a)).start.map(_ => ())
    }
  }
}

This is the only reasonable solution for the Concurrent-enabled implementations. Some notes:

• ordering callback calls is wrong, because a single callback that blocks will block all the rest, just like it blocks the thread calling release (this is related to a remark by @SystemFw that Semaphore relies on listeners waking up in order and I really hope you only rely on the priority of the listeners and not on how the actual callbacks are being called, because that would be very wrong)
• we really need startAndForget in the type class hierarchy (Cats-Effect 2.0) ... unfortunately we cannot add it for Scala 2.11 without breaking binary compatibility, but maybe we can find alternative ways until then
• I believe we need a suspendWithContext[A](f: ExecutionContext => F[A]): F[A] in ContextShift; this too breaks binary compatibility, plus it involves some work in IO's run-loop, but that part is I believe doable without binary compat changes

---

Unfortunately in fixing MVar I needed this function and it made sense to expose it:

object Concurrent {
  // ...

  def cancelableF[F[_], A](k: (Either[Throwable, A] => Unit) => F[CancelToken[F]])
    (implicit F: Concurrent[F]): F[A] = {

    F.asyncF { cb =>
      val state = new AtomicReference[Either[Throwable, Unit] => Unit](null)
      val cb1 = (a: Either[Throwable, A]) => {
        try {
          cb(a)
        } finally {
          // This CAS can only succeed in case the operation is already finished
          // and no cancellation token was installed yet
          if (!state.compareAndSet(null, Callback.dummy1)) {
            val cb2 = state.get()
            state.lazySet(null)
            cb2(Callback.rightUnit)
          }
        }
      }
      // Until we've got a cancellation token, the task needs to be evaluated
      // uninterruptedly, otherwise risking a leak, hence the bracket
      F.bracketCase(k(cb1)) { _ =>
        F.async[Unit] { cb =>
          if (!state.compareAndSet(null, cb)) {
            cb(Callback.rightUnit)
          }
        }
      } { (token, e) =>
        e match {
          case ExitCase.Canceled => token
          case _ => F.unit
        }
      }
    }
  }  
}

[alexandru]

I broke MVar apparently. Have to fix it.

[codecov-io]

Codecov Report

Merging #424 into master will increase coverage by 2.63%.
The diff coverage is 96.61%.

@@            Coverage Diff             @@
##           master     #424      +/-   ##
==========================================
+ Coverage   87.03%   89.66%   +2.63%     
==========================================
  Files          69       70       +1     
  Lines        1905     2003      +98     
  Branches      199      199              
==========================================
+ Hits         1658     1796     +138     
+ Misses        247      207      -40

[mpilquist]

F.void instead of F.map(..)(mapUnit)?

[alexandru]

F.void is more wasteful, because it translates to F.map(task)(_ => ()) and the Scala compiler won't reuse the created _ => ().

When doing performance-sensitive work, in general I care about these issues in about this order:

1. avoiding extraneous async or lazy boundaries
2. avoiding extraneous compareAndSet operations
3. avoiding extraneous memory allocations

Caching dummy functions like _ => () falls into point 3, and sometimes I screw up due to the code becoming unclear, but it doesn't happen that often 🙂

[mpilquist]

What guarantees are provided w.r.t. cancelation of the returned action?

[alexandru]

This is a very good question.

I realized that the task generated by the provided generator function should be evaluated uninterruptedly, otherwise it can leak.

I fixed it and added a ScalaDoc comment about it.

It's now using bracketCase, which tbh isn't ideal for performance, but it does so only after the first put or take attempt failed, so it's not on the happy path.