Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
proof-of-concept generic back-pressure provider, enables blocking back-pressure to be controlled outside the code that is performing the actions.
# The BackPressureProvider allows back-pressure to be applied to non-blocking APIs when those APIs also
# provide hooks for identifying when they _should_ block.
class BackPressureProvider
def initialize(desc, logger)
@desc = desc
@logger = logger
@mutex = Mutex.new
@cond = ConditionVariable.new
@back_pressure_engaged = false
@timeout_max = 10
end
# Engages back-pressure; threads using `BackPressureProvider#execute` will be blocked until back-pressure
# is removed.
#
# @param reason [String]: the reason back-pressure is being applied, to be included in the log message.
# @return [void]
def engage_back_pressure(reason)
@mutex.synchronize do
@back_pressure_engaged = true
@logger.warn("#{@desc} back-pressure engaged: #{reason}")
end
end
# Removes back-pressure, waking any threads that were blocked while using `BackPressureProvider#execute`.
#
# @return [void]
def remove_back_pressure
@mutex.synchronize do
@back_pressure_engaged = false
@logger.info("#{@desc} back-pressure removed.")
@cond.broadcast # wakeup _all_ waiting threads
end
end
# Executes the provided block, _after_ waiting out any back-pressure.
#
# @yieldreturn [Object] the value returned from the block is returned by this method
# @return [Object]
def execute
if @back_pressure_engaged
timeout = 1
start = Time.now
thread_id = Thread.current.__id__
loop do
# instead of a blind sleep, wait for a notification with a timeout; this allows
# us to _immediately_ begin sending events when we are unblocked.
should_block = @mutex.synchronize do
@cond.wait(@mutex, @timeout)
@back_pressure_engaged
end
break unless should_block
block_duration = Time.now - start
@logger.warn("#{desc} has been blocked with no movement for #{block_duration.round}s... (#{thread_id})")
timeout = [@timeout_max, (timeout * 2)].min
end
end
yield
end
end
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment