1
0

request_pool.rb 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  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. WAIT_TIMEOUT = 5
  25. MAX_POOL_SIZE = ENV.fetch('MAX_REQUEST_POOL_SIZE', 512).to_i
  26. class Connection
  27. attr_reader :site, :last_used_at, :created_at, :in_use, :dead, :fresh
  28. def initialize(site)
  29. @site = site
  30. @http_client = http_client
  31. @last_used_at = nil
  32. @created_at = current_time
  33. @dead = false
  34. @fresh = true
  35. end
  36. def use
  37. @last_used_at = current_time
  38. @in_use = true
  39. retries = 0
  40. begin
  41. yield @http_client
  42. rescue HTTP::ConnectionError
  43. # It's possible the connection was closed, so let's
  44. # try re-opening it once
  45. close
  46. if @fresh || retries.positive?
  47. raise
  48. else
  49. @http_client = http_client
  50. retries += 1
  51. retry
  52. end
  53. rescue StandardError
  54. # If this connection raises errors of any kind, it's
  55. # better if it gets reaped as soon as possible
  56. close
  57. @dead = true
  58. raise
  59. end
  60. ensure
  61. @fresh = false
  62. @in_use = false
  63. end
  64. def seconds_idle
  65. current_time - (@last_used_at || @created_at)
  66. end
  67. def close
  68. @http_client.close
  69. end
  70. private
  71. def http_client
  72. Request.http_client.persistent(@site, timeout: MAX_IDLE_TIME)
  73. end
  74. def current_time
  75. Process.clock_gettime(Process::CLOCK_MONOTONIC)
  76. end
  77. end
  78. def initialize
  79. @pool = ConnectionPool::SharedConnectionPool.new(size: MAX_POOL_SIZE, timeout: WAIT_TIMEOUT) { |site| Connection.new(site) }
  80. @reaper = Reaper.new(self, 30)
  81. @reaper.run
  82. end
  83. def with(site, &block)
  84. @pool.with(site) do |connection|
  85. ActiveSupport::Notifications.instrument('with.request_pool', miss: connection.fresh, host: connection.site) do
  86. connection.use(&block)
  87. end
  88. end
  89. end
  90. delegate :size, :flush, to: :@pool
  91. end