request_pool.rb 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115
  1. # frozen_string_literal: true
  2. require_relative 'connection_pool/shared_connection_pool'
  3. class RequestPool
  4. def self.current
  5. @current ||= RequestPool.new
  6. end
  7. class Reaper
  8. attr_reader :pool, :frequency
  9. def initialize(pool, frequency)
  10. @pool = pool
  11. @frequency = frequency
  12. end
  13. def run
  14. return unless frequency&.positive?
  15. Thread.new(frequency, pool) do |t, p|
  16. loop do
  17. sleep t
  18. p.flush
  19. end
  20. end
  21. end
  22. end
  23. MAX_IDLE_TIME = 30
  24. MAX_POOL_SIZE = ENV.fetch('MAX_REQUEST_POOL_SIZE', 512).to_i
  25. REAPER_FREQUENCY = 30
  26. WAIT_TIMEOUT = 5
  27. class Connection
  28. attr_reader :site, :last_used_at, :created_at, :in_use, :dead, :fresh
  29. def initialize(site)
  30. @site = site
  31. @http_client = http_client
  32. @last_used_at = nil
  33. @created_at = current_time
  34. @dead = false
  35. @fresh = true
  36. end
  37. def use
  38. @last_used_at = current_time
  39. @in_use = true
  40. retries = 0
  41. begin
  42. yield @http_client
  43. rescue HTTP::ConnectionError
  44. # It's possible the connection was closed, so let's
  45. # try re-opening it once
  46. close
  47. if @fresh || retries.positive?
  48. raise
  49. else
  50. @http_client = http_client
  51. retries += 1
  52. retry
  53. end
  54. rescue
  55. # If this connection raises errors of any kind, it's
  56. # better if it gets reaped as soon as possible
  57. close
  58. @dead = true
  59. raise
  60. end
  61. ensure
  62. @fresh = false
  63. @in_use = false
  64. end
  65. def seconds_idle
  66. current_time - (@last_used_at || @created_at)
  67. end
  68. def close
  69. @http_client.close
  70. end
  71. private
  72. def http_client
  73. Request.http_client.persistent(@site, timeout: MAX_IDLE_TIME)
  74. end
  75. def current_time
  76. Process.clock_gettime(Process::CLOCK_MONOTONIC)
  77. end
  78. end
  79. def initialize
  80. @pool = ConnectionPool::SharedConnectionPool.new(size: MAX_POOL_SIZE, timeout: WAIT_TIMEOUT) { |site| Connection.new(site) }
  81. @reaper = Reaper.new(self, REAPER_FREQUENCY)
  82. @reaper.run
  83. end
  84. def with(site, &block)
  85. @pool.with(site) do |connection|
  86. ActiveSupport::Notifications.instrument('with.request_pool', miss: connection.fresh, host: connection.site) do
  87. connection.use(&block)
  88. end
  89. end
  90. end
  91. delegate :size, :flush, to: :@pool
  92. end