BackPressureHandling

Automated back-pressure handling is based on the idea that pressure is created by the consumer but experienced at the producer side. E.g. for http that means that a too big number of incoming requests is the ultimate cause of an experienced bottleneck on the response sending side.

The principle of applying back-pressure means that the best way of handling pressure is by handling it at the root cause which means throttling the rate at which work requests are coming in. That's the underlying assumption here: work is generated on the incoming network side. If that's not true, e.g. when the network stream is a truly bi-directional one (e.g. websockets) the strategy presented here won't be optimal.

How it works:

No pressure:

forward all incoming data
send out n responses with NoAcks
send one response with Ack
once that ack was received we know all the former unacknowledged writes have been successful as well and don't need any further handling

Pressure:

a Write fails, we know now that all former writes were successful, all latter ones, including the failed one were discarded (but we'll still receive CommandFailed messages for them as well)
the incoming side is informed to SuspendReading
we send ResumeWriting which is queued after all the Writes that will be discarded as well
once we receive WritingResumed go back to the no pressure mode and retry all of the buffered writes
we schedule a final write probe which will trigger ResumeReading when no lowWatermark is defined
once we receive the ack for that probe or the buffer size falls below a lowWatermark after an acknowledged Write, we ResumeReading

Possible improvement: (see http://doc.akka.io/docs/akka/2.2.0-RC1/scala/io-tcp.html)

go into Ack based mode for a while after WritingResumed

Linear Supertypes

AnyRef, Any

Type Members

case class Ack(offset: Int) extends akka.io.Tcp.Event with Product with Serializable
class OutQueue extends AnyRef

A mutable queue of outgoing write requests

Value Members

final def !=(arg0: AnyRef): Boolean

Definition Classes
AnyRef
final def !=(arg0: Any): Boolean

Definition Classes
Any
final def ##(): Int

Definition Classes
AnyRef → Any
final def ==(arg0: AnyRef): Boolean

Definition Classes
AnyRef
final def ==(arg0: Any): Boolean

Definition Classes
Any
object CanCloseNow extends akka.io.Tcp.Event
val ProbeForEndOfWriting: Write
val ProbeForWriteQueueEmpty: Write
object ResumeReadingNow extends akka.io.Tcp.Event
def apply(ackRate: Int, lowWatermark: Int = Int.MaxValue): PipelineStage
final def asInstanceOf[T0]: T0

Definition Classes
Any
def clone(): AnyRef

Attributes
protected[java.lang]
Definition Classes
AnyRef
Annotations
@throws( ... )
final def eq(arg0: AnyRef): Boolean

Definition Classes
AnyRef
def equals(arg0: Any): Boolean

Definition Classes
AnyRef → Any
def finalize(): Unit

Attributes
protected[java.lang]
Definition Classes
AnyRef
Annotations
@throws( classOf[java.lang.Throwable] )
final def getClass(): Class[_]

Definition Classes
AnyRef → Any
def hashCode(): Int

Definition Classes
AnyRef → Any
final def isInstanceOf[T0]: Boolean

Definition Classes
Any
final def ne(arg0: AnyRef): Boolean

Definition Classes
AnyRef
final def notify(): Unit

Definition Classes
AnyRef
final def notifyAll(): Unit

Definition Classes
AnyRef
final def synchronized[T0](arg0: ⇒ T0): T0

Definition Classes
AnyRef
def toString(): String

Definition Classes
AnyRef → Any
final def wait(): Unit

Definition Classes
AnyRef
Annotations
@throws( ... )
final def wait(arg0: Long, arg1: Int): Unit

Definition Classes
AnyRef
Annotations
@throws( ... )
final def wait(arg0: Long): Unit

Definition Classes
AnyRef
Annotations
@throws( ... )

object BackPressureHandling

Type Members

case class Ack(offset: Int) extends akka.io.Tcp.Event with Product with Serializable

class OutQueue extends AnyRef

Value Members

final def !=(arg0: AnyRef): Boolean

final def !=(arg0: Any): Boolean

final def ##(): Int

final def ==(arg0: AnyRef): Boolean

final def ==(arg0: Any): Boolean

object CanCloseNow extends akka.io.Tcp.Event

val ProbeForEndOfWriting: Write

val ProbeForWriteQueueEmpty: Write

object ResumeReadingNow extends akka.io.Tcp.Event

def apply(ackRate: Int, lowWatermark: Int = Int.MaxValue): PipelineStage

final def asInstanceOf[T0]: T0

def clone(): AnyRef

final def eq(arg0: AnyRef): Boolean

def equals(arg0: Any): Boolean

def finalize(): Unit

final def getClass(): Class[_]

def hashCode(): Int

final def isInstanceOf[T0]: Boolean

final def ne(arg0: AnyRef): Boolean

final def notify(): Unit

final def notifyAll(): Unit

final def synchronized[T0](arg0: ⇒ T0): T0

def toString(): String

final def wait(): Unit

final def wait(arg0: Long, arg1: Int): Unit

final def wait(arg0: Long): Unit

Inherited from AnyRef

Inherited from Any

Ungrouped