Class: ActiveRecord::Bogacs::ShareablePool

Inherits:
DefaultPool
  • Object
show all
Includes:
Concurrent::ThreadSafe::Util::CheapLockable
Defined in:
lib/active_record/bogacs/shareable_pool.rb

Constant Summary collapse

AtomicReference = 
::Concurrent::AtomicReference
DEFAULT_SHARED_POOL =

only allow 25% of the pool size to be shared

0.25
MAX_THREAD_SHARING =

not really a strict limit but should hold

5

Instance Attribute Summary collapse

Attributes inherited from DefaultPool

#automatic_reconnect, #checkout_timeout, #initial_size, #reaper, #size, #spec, #validator

Attributes included from PoolSupport

#schema_cache

Instance Method Summary collapse

Methods inherited from DefaultPool

#checkin, #checkout, #clear_stale_cached_connections!, #connected?, #connections, #discard!, #discarded?, #flush, #flush!, #logger, #reap, #reaper?, #reaping?, #stat, #validating?, #validator?, #verify_active_connections!, #with_connection

Methods included from PoolSupport

#connection_cache_key, included, #lock_thread=, #new_connection, #reap

Constructor Details

#initialize(spec) ⇒ ShareablePool

Returns a new instance of ShareablePool.



29
30
31
32
33
34
35
36
37
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 29

def initialize(spec)
  super(spec)
  shared_size = spec.config[:shared_pool]
  shared_size = shared_size ? shared_size.to_f : DEFAULT_SHARED_POOL
  # size 0.0 - 1.0 assumes percentage of the pool size
  shared_size = ( @size * shared_size ).round if shared_size <= 1.0
  @shared_size = shared_size.to_i
  @shared_connections = ThreadSafe::Map.new # initial_capacity: @shared_size
end

Instance Attribute Details

#shared_size ⇒ Object (readonly)

Returns the value of attribute shared_size.



26
27
28
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 26

def shared_size
  @shared_size
end

Instance Method Details

#active_connection? ⇒ Boolean

Returns:



45
46
47
48
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 45

def active_connection?
  return true if current_thread[shared_connection_key]
  has_active_connection? # super
end

#clear_reloadable_connections! ⇒ Object 



74
75
76
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 74

def clear_reloadable_connections!
  synchronize { @shared_connections.clear; super }
end

#connection ⇒ Object 



40
41
42
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 40

def connection
  current_thread[shared_connection_key] || super
end

#disconnect! ⇒ Object 



69
70
71
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 69

def disconnect!
  synchronize { @shared_connections.clear; super }
end

#release_connection(owner_thread = Thread.current) ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 51

def release_connection(owner_thread = Thread.current)
  conn_id = connection_cache_key(owner_thread)
  if reserved_conn = @thread_cached_conns.delete(conn_id)
    if shared_count = @shared_connections[reserved_conn]
      cheap_synchronize do # lock due #get_shared_connection ... not needed ?!
        # NOTE: the other option is to not care about shared here at all ...
        if shared_count.get == 0 # releasing a shared connection
          release_shared_connection(reserved_conn, owner_thread)
        #else return false
        end
      end
    else # check back-in non-shared connections
      checkin reserved_conn # (what super does)
    end
  end
end

#release_shared_connection(connection, owner_thread = Thread.current) ⇒ Object

Custom API :



100
101
102
103
104
105
106
107
108
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 100

def release_shared_connection(connection, owner_thread = Thread.current)
  shared_conn_key = shared_connection_key
  if connection == owner_thread[shared_conn_key]
    owner_thread[shared_conn_key] = nil
  end

  @shared_connections.delete(connection)
  shared_checkin connection # synchronized
end

#remove(conn) ⇒ Object

Note:

called from #reap thus the pool should work with reaper



80
81
82
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 80

def remove(conn)
  synchronize { @shared_connections.delete(conn); super }
end

#with_shared_connection ⇒ Object 



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/active_record/bogacs/shareable_pool.rb', line 110

def with_shared_connection
  shared_conn_key = shared_connection_key
  # with_shared_connection call nested in the same thread
  if connection = Thread.current[shared_conn_key]
    emulated_checkout(connection)
    return yield connection
  end

  start = Time.now if DEBUG
  begin
    # if there's a 'regular' connection on the thread use it as super
    if has_active_connection? # for current thread
      connection = self.connection # do not mark as shared
      DEBUG && debug("with_shared_conn 10 got active = #{connection.to_s}")
    # otherwise if we have a shared connection - use that one :
    elsif connection = get_shared_connection
      emulated_checkout(connection); shared = true
      DEBUG && debug("with_shared_conn 20 got shared = #{connection.to_s}")
    else
      shared = true
      synchronize do
        # check shared again as/if threads end up sync-ing up here :
        if connection = get_shared_connection
          emulated_checkout(connection)
          DEBUG && debug("with_shared_conn 21 got shared = #{connection.to_s}")
        end # here we acquire but a connection from the pool
        # TODO the bottle-neck for concurrency doing sync { checkout } :
        unless connection # here we acquire a connection from the pool
          connection = self.checkout # might block if pool fully used
          add_shared_connection(connection)
          DEBUG && debug("with_shared_conn 30 acq shared = #{connection.to_s}")
        end
      end
    end

    Thread.current[shared_conn_key] = connection if shared

    DEBUG && debug("with_shared_conn obtaining a connection took #{(Time.now - start) * 1000}ms")
    yield connection
  ensure
    Thread.current[shared_conn_key] = nil if shared
    rem_shared_connection(connection) if shared && connection
  end
end