Backpressure and load shedding are methods you can use to mitigate queue overload.
These are methods that kick in automatically therefore your system needs to have enough instrumentation to know if a queue is overloaded. For example, we can have a queue
object which responds to the overloaded?
message, (i.e. queue.overloaded?
).
The definition of queue overload (i.e. overloaded?
) is up to you to define.
A queue can be overloaded if the queue latency is above a certain threshold.
class Queue
LATENCY_THRESHOLD = 10.seconds
def overloaded?
@latency > LATENCY_THRESHOLD
end
end
A queue can be overloaded if queue size is above a certain threshold.
class Queue
SIZE_THRESHOLD = 1000
def overloaded?
@size > SIZE_THRESHOLD
end
end
A queue can be overloaded if produce rate is above a certain threshold.
class Queue
PRODUCE_RATE_THRESHOLD = 200000 # messages per second
def overloaded?
@produce_rate > PRODUCE_RATE_THRESHOLD
end
end
You need to pick a definition of overload for which you are able to instrument.
The difference between backpressure and load shedding becomes the action that is taken by the consumers/producers in the event of overload.
Load shedding
Load shedding is about dropping/rejecting messages in the event of overload. You can decide to drop them at the consumer and/or at the producer.
class Consumer
def consume
queue.each_message do |message|
next if queue.overloaded? # ignore received message during overload
process(message)
end
end
def process(message)
# process message
end
end
class Producer
def produce(message)
return if queue.overloaded? # don't produce message during overload
queue.add(message)
end
end
Backpressure
Backpressure is about reducing produce rate by slowing down the producers. Backpressure allows the client of the producer to decide how to deal with the increased delay (i.e. abort the operation, wait until the message can be produced, retry the operation).
You can implement backpressure in the producers.
class Producer
def produce(message)
while queue.overloaded? do
# slow down producer by blocking it until the message can be added
sleep 1
end
queue.add(message)
end
end
class Producer
def produce(message)
# reject the message but properly notify the client of produce so it
# can choose to retry or not
raise "Queue is overloaded" if queue.overloaded?
queue.add(message)
end
end
You can find other posts of my series on queues here. I can send you my posts straight to your e-mail inbox if you sign up for my newsletter.