Producer-Consumer with Futures in Scala

Question

Suppose I have a "producer-consumer" problem: the producer sends messages to the consumer and the consumer processes them asynchronously using Scala Futures: for instance future { /* do the processing */ }.

Suppose now that the producer produces 100 messages per sec. but the consumer processes only 10 messages per sec. What will happen ? I guess there will be a memory leak. There will be a lot of Future objects and the internal message queue of the thread pool will grow as well. Does it make sense ?

What is the best way to handle it ?

Note - I was confused by this - this question is about Futures and not actors. — JasonG

JasonG JasonG · Accepted Answer · 2013-08-06T13:41:43

In akka, an execution context is used but there appears to be no mailbox - it would be worthwhile to read the source but I can answer your question by experimentation:

Future's do not have a 'mailbox' and I'm not 100% sure exactly what Akka does under the hood or what the execution context actually contains, but we can see that akka will run out of memory when using futures directly:

scala> import scala.concurrent.Future
import scala.concurrent.Future

scala> import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.ExecutionContext.Implicits.global                  ^

scala> while(1==1) Future(Thread.sleep(100))
java.lang.OutOfMemoryError: Java heap space

If we're talking about messages, then there is a mailbox that describes the behaviour of the actor message queue (which would fill as only one message is processed at a time) - I'll explain this below.

Assuming a bounded mailbox (eg one with a size limit) what happens to the messages. The answer is it depends on the mailbox. First, a bounded mailbox has a few settings, such as size limit:

bounded-mailbox {
  mailbox-type = "akka.dispatch.BoundedMailbox"
  mailbox-capacity = 1000
  mailbox-push-timeout-time = 10s
}

Now when that limit is hit, akka will either drop the old or the new messages depending on how the mailbox is configured - eg with this setting

# whether to drop older items (instead of newer) when the queue is full
discard-old-when-full = on

Obviously if there are other resource issues like out of memory then your application can crash meaning that the messages will be lost as they are stored in memory. An unbounded mailbox will continue to stack messages until error conditions occur which is why you may want to use bounded mailboxes.

If the loss of messages in error conditions in undesirable, there is another option = a durable mailbox can be used which will store the messages somewhere more persistent such as in a file. Here is an example mailbox configuration that uses a file for more durable message storage.

akka {
  actor {
    mailbox {
      file-based {
        # directory below which this queue resides
        directory-path = "./_mb"

        # attempting to add an item after the queue reaches this size (in items)
        # will fail.
        max-items = 2147483647

        # attempting to add an item after the queue reaches this size (in bytes)
        # will fail.
        max-size = 2147483647 bytes

        # attempting to add an item larger than this size (in bytes) will fail.
        max-item-size = 2147483647 bytes

        # maximum expiration time for this queue (seconds).
        max-age = 0s

        # maximum journal size before the journal should be rotated.
        max-journal-size = 16 MiB

        # maximum size of a queue before it drops into read-behind mode.
        max-memory-size = 128 MiB

        # maximum overflow (multiplier) of a journal file before we re-create it.
        max-journal-overflow = 10

        # absolute maximum size of a journal file until we rebuild it,
        # no matter what.
        max-journal-size-absolute = 9223372036854775807 bytes

        # whether to drop older items (instead of newer) when the queue is full
        discard-old-when-full = on

        # whether to keep a journal file at all
        keep-journal = on

        # whether to sync the journal after each transaction
        sync-journal = off

        # circuit breaker configuration
        circuit-breaker {
          # maximum number of failures before opening breaker
          max-failures = 3

          # duration of time beyond which a call is assumed to be timed out and
          # considered a failure
          call-timeout = 3 seconds

          # duration of time to wait until attempting to reset the breaker during
          # which all calls fail-fast
          reset-timeout = 30 seconds
        }
      }
    }
  }
}

Producer-Consumer with Futures in Scala

3 Answers