Class: Attio::ConnectionPool Private

Inherits:
Object
  • Object
show all
Defined in:
lib/attio/connection_pool.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Thread-safe connection pool for high-throughput operations

Examples:

Basic usage

pool = ConnectionPool.new(size: 5) do
  Attio::HttpClient.new(base_url: API_URL, headers: headers)
end

pool.with do |connection|
  connection.get("records")
end

Since:

  • 1.0.0

Defined Under Namespace

Classes: PoolShuttingDownError, TimeoutError

Constant Summary collapse

DEFAULT_SIZE =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0

5
DEFAULT_TIMEOUT =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0

5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size: DEFAULT_SIZE, timeout: DEFAULT_TIMEOUT) { ... } ⇒ ConnectionPool

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Initialize a new connection pool

Yields:

  • Block that creates a new connection

Raises:

  • (ArgumentError)

Since:

  • 1.0.0



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/attio/connection_pool.rb', line 30

def initialize(size: DEFAULT_SIZE, timeout: DEFAULT_TIMEOUT, &block)
  raise ArgumentError, "Block required to create connections" unless block_given?

  @size = size
  @timeout = timeout
  @create_block = block
  @available = []
  @allocated = {}
  @mutex = Mutex.new
  @resource = ConditionVariable.new
  @shutting_down = false
  @created = 0

  # Stats tracking
  @stats = {
    requests: 0,
    timeouts: 0,
    wait_time: 0,
    active: 0,
    created: 0,
    destroyed: 0,
  }
end

Instance Attribute Details

#allocated ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0



23
24
25
 
# File 'lib/attio/connection_pool.rb', line 23

def allocated
  @allocated
end

#available ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0



23
24
25
 
# File 'lib/attio/connection_pool.rb', line 23

def available
  @available
end

#size ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0



23
24
25
 
# File 'lib/attio/connection_pool.rb', line 23

def size
  @size
end

#timeout ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0



23
24
25
 
# File 'lib/attio/connection_pool.rb', line 23

def timeout
  @timeout
end

Instance Method Details

#checkin(connection) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Return a connection to the pool

Since:

  • 1.0.0



113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/attio/connection_pool.rb', line 113

def checkin(connection)
  @mutex.synchronize do
    if @allocated[Thread.current] == connection
      @allocated.delete(Thread.current)
      @stats[:active] -= 1

      if @shutting_down
        destroy_connection(connection)
      else
        @available.push(connection)
        @resource.signal
      end
    end
  end
end

#checkout ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Check out a connection from the pool

Raises:

  • (TimeoutError) —

    if no connection available within timeout

Since:

  • 1.0.0



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/attio/connection_pool.rb', line 71

def checkout
  start_time = Time.now
  deadline = start_time + @timeout

  @mutex.synchronize do
    raise PoolShuttingDownError, "Pool is shutting down" if @shutting_down

    @stats[:requests] += 1

    loop do
      # Return available connection
      if (connection = @available.pop)
        @allocated[Thread.current] = connection
        @stats[:active] += 1
        @stats[:wait_time] += (Time.now - start_time)
        return connection
      end

      # Create new connection if under limit
      if @created < @size
        connection = create_connection
        @allocated[Thread.current] = connection
        @stats[:active] += 1
        @stats[:wait_time] += (Time.now - start_time)
        return connection
      end

      # Wait for available connection
      remaining = deadline - Time.now
      if remaining <= 0
        @stats[:timeouts] += 1
        raise TimeoutError, "Timed out waiting for connection after #{@timeout}s"
      end

      @resource.wait(@mutex, remaining)
    end
  end
end

#create_connection ⇒ Object (private)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0



193
194
195
196
197
198
 
# File 'lib/attio/connection_pool.rb', line 193

private def create_connection
  connection = @create_block.call
  @created += 1
  @stats[:created] += 1
  connection
end

#destroy_connection(connection) ⇒ Object (private)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Since:

  • 1.0.0



200
201
202
203
204
205
206
207
 
# File 'lib/attio/connection_pool.rb', line 200

private def destroy_connection(connection)
  # Call close if connection responds to it
  connection.close if connection.respond_to?(:close)
  @stats[:destroyed] += 1
rescue StandardError => e
  # Log but don't raise on close errors
  warn "Error closing connection: #{e.message}"
end

#healthy? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Check if pool is healthy

Since:

  • 1.0.0



186
187
188
189
190
191
 
# File 'lib/attio/connection_pool.rb', line 186

def healthy?
  return false if @shutting_down
  return true if @stats[:requests] == 0

  @stats[:timeouts] < (@stats[:requests] * 0.01)
end

#reset! ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Reset the pool by closing all connections

Since:

  • 1.0.0



145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/attio/connection_pool.rb', line 145

def reset!
  @mutex.synchronize do
    # Close all available connections
    while (connection = @available.pop)
      destroy_connection(connection)
    end

    @created = 0
    @stats[:created] = 0
    @stats[:destroyed] = 0
  end
end

#shutdown ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Shutdown the pool and close all connections

Since:

  • 1.0.0



130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/attio/connection_pool.rb', line 130

def shutdown
  @mutex.synchronize do
    @shutting_down = true

    # Close available connections
    while (connection = @available.pop)
      destroy_connection(connection)
    end

    # NOTE: allocated connections will be closed when checked in
    @resource.broadcast
  end
end

#stats ⇒ Hash

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get pool statistics

Since:

  • 1.0.0



161
162
163
164
165
166
167
168
169
170
 
# File 'lib/attio/connection_pool.rb', line 161

def stats
  @mutex.synchronize do
    @stats.merge(
      size: @size,
      available: @available.size,
      allocated: @allocated.size,
      created: @created
    )
  end
end

#utilization ⇒ Float

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Current pool utilization (0.0 to 1.0)

Since:

  • 1.0.0



175
176
177
178
179
180
181
 
# File 'lib/attio/connection_pool.rb', line 175

def utilization
  @mutex.synchronize do
    return 0.0 if @size == 0

    @allocated.size.to_f / @size
  end
end

#with {|connection| ... } ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Execute a block with a connection from the pool

Yields:

  • (connection) —

    Block to execute with connection

Since:

  • 1.0.0



58
59
60
61
62
63
64
65
 
# File 'lib/attio/connection_pool.rb', line 58

def with
  connection = checkout
  begin
    yield connection
  ensure
    checkin(connection)
  end
end