4
\$\begingroup\$

This is my attempt at implementing debounce in Scala without any 3rd party library. How can I improve the code/make it more idiomatic (and any bugs I might have missed)?

  import scala.compat.Platform.{currentTime => now}
  import scala.concurrent.duration._
  import scala.util.control.NonFatal

  /**
   * Return a function f that it can only be invoked iff it is not already running
   * and atleast wait duration has passed since it last stopped running
   * Usage:
   *    def plus1(x: Int) = println(x + 1)
   *    val f = debounce(1.second)(plus1)
   *    f(1)
   *    f(2)
   *    Thread.sleep(2000)
   *    f(3)
   * 
   * @return a function such that it returns Some(original output) if it was invoked 
   *         or else if it failed to run because of above rules, None
   */
  def debounce[A, B](wait: Duration)(f: A => B): A => Option[B] = {
    var (isRunning, lastStopTime) = (false, Long.MinValue)
    (input: A) => {
      if (!isRunning && lastStopTime + wait.toMillis <= now) {
        try {
          isRunning = true
          Some(f(input))
        } catch {
          case NonFatal(e) => throw e
        } finally {
          isRunning = false
          lastStopTime = now
        }
      } else {
        None
      }
    }
  }
\$\endgroup\$
3
\$\begingroup\$

The most noticeable problem is the race condition on the access of isRunning within the returned anonymous function. It is possible that two different threads can call the function at the same exact time, so they both see that isRunning == true before it is flipped to false.

For example:

def test(i: Int): Int = { println(i); i }
val db = debounce(1.seconds)(test)

When running making many concurrent calls to db, it occasionally will leak an extra Some, noted by the extra printed number (occurred less than 1ms after the original call).

scala> (0 to 10000000).par.map(db)
0
5001723

The easiest way to fix this is to synchronize access to isRunning (and really, the whole function), which can be done by surrounding the result of the function with this.synchronized:

def debounce[A, B](wait: Duration)(f: A => B): A => Option[B] = {
  var (isRunning, lastStopTime) = (false, Long.MinValue)

  (input: A) => this.synchronized {
    if (!isRunning && lastStopTime + wait.toMillis <= now) {
      try {
        isRunning = true
        Some(f(input))
      } finally {
        isRunning = false
        lastStopTime = now
      }
    } else {
      None
    }
  }
}

This will have some performance implications, particularly if f is slow, since we are locking on it's evaluation. That is, you can end up blocking the calling threads when making multiple concurrent calls while waiting for the first f to evaluate.

I also removed the catch block, as it was superfluous.


I should note there is also a fundamental difference between the way your method behaves, and the linked javascript function. Yours intends to prevent f from being evaluated more than once during a specific duration after an evaluation (rate limiting). Underscore's debounce method will not evaluate the function at all until one wait period after the last call to it has expired. That is, if wait was one second and my function db was called once every 0.5 seconds repeatedly, the contained function would never be evaluated (with the underscore behavior). The author of that post does not seem to understand that underscore's debounce is not for rate limiting, but for calling a function after a stream of input has been interrupted for an arbitrary amount of time.


For fun, here is a version that uses AtomicBoolean, similarly to a ReentrantLock. It should be better than synchronize in that it won't block calling threads while f is evaluating. It uses compareAndSet to first check that isRunning is false, and then sets it to true if it was already false, but only if the wait hasn't expired. This ensures that it isn't currently running, and it doesn't prematurely re-lock before the time has elapsed.

import java.util.concurrent.atomic.AtomicBoolean

def debounce[A, B](wait: Duration)(f: A => B): A => Option[B] = {
    var (isRunning, lastStopTime) = (new AtomicBoolean(false), Long.MinValue)

    (input: A) => {
        val doneWaiting = lastStopTime + wait.toMillis <= now
        if (isRunning.compareAndSet(false, doneWaiting) && doneWaiting) {
            try {
                Some(f(input))
            } finally {
                lastStopTime = now
                isRunning.set(false)
            }
        } else {
            None
        }
    }
}

Similarly to the ReentrantLock though, this suffers from contention to check the status of isRunning. Which means that if there are many concurrent calls to compareAndSet, it will actually delay the call to set. This shouldn't really be a problem unless a ridiculous amount of calls are being made, though.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Thank you for the review. And, yes, I actually prefer this behaviour over underscore's :) Would it be better if we make isRunning an atomic boolean and not put synchronized? \$\endgroup\$ – pathikrit Jul 29 '15 at 4:21
  • \$\begingroup\$ It's possible, but that opens up the possibility for a different race condition that I've been trying to resolve. The wait period is what complicates it, so I need to think about it more. \$\endgroup\$ – m-z Jul 29 '15 at 4:25
  • \$\begingroup\$ Regarding the performance optimization when f is slow, an improvement could be to do an eager check outsize the sync block and do once more inside e.g. if (!isRunning) synchornized { if(!isRunning) .... } \$\endgroup\$ – pathikrit Jul 29 '15 at 5:52
  • \$\begingroup\$ @wrick Added a version using AtomicBoolean, which won't synchronize over f. \$\endgroup\$ – m-z Jul 29 '15 at 16:43
  • \$\begingroup\$ Why didn't you use if(!isRunning.get() && lastStopTime + wait.toMillis <= now) { isRunning.set(true) ... ? \$\endgroup\$ – pathikrit Jul 29 '15 at 17:14
1
\$\begingroup\$

Debouncing a function like that requires mutable state, which is never going to be very idiomatic. To get more idiomatic (and get rid of the mutable state), I'd reconsider the architecture and probably use something like ScalaRX. Debouncing is something that works quite naturally with Streams.

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Can you explain how to debounce with streams? \$\endgroup\$ – pathikrit Aug 5 '15 at 20:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.