Class: Dalli::Protocol::ConnectionManager

Inherits:
Object
  • Object
show all
Defined in:
lib/dalli/protocol/connection_manager.rb

Overview

Manages the socket connection to the server, including ensuring liveness and retries.

Constant Summary collapse

DEFAULTS = 
{
  # seconds between trying to contact a remote server
  down_retry_delay: 30,
  # connect/read/write timeout for socket operations
  socket_timeout: 1,
  # times a socket operation may fail before considering the server dead
  socket_max_failures: 2,
  # amount of time to sleep between retries when a failure occurs
  socket_failure_delay: 0.1,
  # Set keepalive
  keepalive: true
}.freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(hostname, port, socket_type, client_options) ⇒ ConnectionManager

Returns a new instance of ConnectionManager.



30
31
32
33
34
35
36
37
38
39
40
 
# File 'lib/dalli/protocol/connection_manager.rb', line 30

def initialize(hostname, port, socket_type, client_options)
  @hostname = hostname
  @port = port
  @socket_type = socket_type
  @options = DEFAULTS.merge(client_options)
  @request_in_progress = false
  @sock = nil
  @pid = nil

  reset_down_info
end

Instance Attribute Details

#hostname ⇒ Object

Returns the value of attribute hostname.



27
28
29
 
# File 'lib/dalli/protocol/connection_manager.rb', line 27

def hostname
  @hostname
end

#options ⇒ Object

Returns the value of attribute options.



27
28
29
 
# File 'lib/dalli/protocol/connection_manager.rb', line 27

def options
  @options
end

#port ⇒ Object

Returns the value of attribute port.



27
28
29
 
# File 'lib/dalli/protocol/connection_manager.rb', line 27

def port
  @port
end

#sock ⇒ Object (readonly)

Returns the value of attribute sock.



28
29
30
 
# File 'lib/dalli/protocol/connection_manager.rb', line 28

def sock
  @sock
end

#socket_type ⇒ Object

Returns the value of attribute socket_type.



27
28
29
 
# File 'lib/dalli/protocol/connection_manager.rb', line 27

def socket_type
  @socket_type
end

Instance Method Details

#abort_request! ⇒ Object 



150
151
152
 
# File 'lib/dalli/protocol/connection_manager.rb', line 150

def abort_request!
  @request_in_progress = false
end

#close ⇒ Object 



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/dalli/protocol/connection_manager.rb', line 111

def close
  return unless @sock

  begin
    @sock.close
  rescue StandardError
    nil
  ensure
    # A non-StandardError (e.g. a second Async::Stop fired into the
    # fiber while it is already inside this cleanup) can escape
    # @sock.close; run the state cleanup unconditionally so the client
    # isn't returned to the pool with a half-closed socket and
    # @request_in_progress == true.
    @sock = nil
    @pid = nil
    abort_request!
  end
end

#confirm_in_progress! ⇒ Object 



105
106
107
108
109
 
# File 'lib/dalli/protocol/connection_manager.rb', line 105

def confirm_in_progress!
  raise '[Dalli] No request in progress. This may be a bug in Dalli.' unless request_in_progress?

  reconnect_on_fork if fork_detected?
end

#confirm_ready! ⇒ Object 



100
101
102
103
 
# File 'lib/dalli/protocol/connection_manager.rb', line 100

def confirm_ready!
  close if request_in_progress?
  reconnect_on_fork if fork_detected?
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


130
131
132
 
# File 'lib/dalli/protocol/connection_manager.rb', line 130

def connected?
  !@sock.nil?
end

#down! ⇒ Object

Marks the server instance as down. Updates the down_at state and raises an Dalli::NetworkError that includes the underlying error in the message. Calls close to clean up socket state



83
84
85
86
87
88
89
90
 
# File 'lib/dalli/protocol/connection_manager.rb', line 83

def down!
  close
  log_down_detected

  @error = $ERROR_INFO&.class&.name
  @msg ||= $ERROR_INFO&.message
  raise_down_error
end

#error_on_request!(err_or_string) ⇒ Object 



212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/dalli/protocol/connection_manager.rb', line 212

def error_on_request!(err_or_string)
  log_warn_message(err_or_string)

  @fail_count += 1
  if @fail_count >= max_allowed_failures
    down!
  else
    # Closes the existing socket, setting up for a reconnect
    # on next request
    reconnect!('Socket operation failed, retrying...')
  end
end

#establish_connection ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/dalli/protocol/connection_manager.rb', line 50

def establish_connection
  Dalli.logger.debug { "Dalli::Server#connect #{name}" }

  @sock = memcached_socket
  @sock.sync = false # Enable buffered I/O for better performance
  @pid = Process.pid
  @request_in_progress = false
rescue SystemCallError, *TIMEOUT_ERRORS, EOFError, SocketError => e
  # SocketError = DNS resolution failure
  error_on_request!(e)
end

#finish_request! ⇒ Object 



144
145
146
147
148
 
# File 'lib/dalli/protocol/connection_manager.rb', line 144

def finish_request!
  raise '[Dalli] No request in progress. This may be a bug in Dalli.' unless @request_in_progress

  @request_in_progress = false
end

#flush ⇒ Object 



188
189
190
191
192
 
# File 'lib/dalli/protocol/connection_manager.rb', line 188

def flush
  @sock.flush
rescue SystemCallError, *TIMEOUT_ERRORS, *SSL_ERRORS, IOError => e
  error_on_request!(e)
end

#flushed_write(bytes) ⇒ Object 



194
195
196
197
198
199
200
 
# File 'lib/dalli/protocol/connection_manager.rb', line 194

def flushed_write(bytes)
  written = @sock.write(bytes)
  @sock.flush
  written
rescue SystemCallError, *TIMEOUT_ERRORS, *SSL_ERRORS, IOError => e
  error_on_request!(e)
end

#fork_detected? ⇒ Boolean

Returns:

  • (Boolean) 


262
263
264
 
# File 'lib/dalli/protocol/connection_manager.rb', line 262

def fork_detected?
  @pid && @pid != Process.pid
end

#log_down_detected ⇒ Object 



266
267
268
269
270
271
272
273
274
275
276
 
# File 'lib/dalli/protocol/connection_manager.rb', line 266

def log_down_detected
  @last_down_at = Time.now

  if @down_at
    time = Time.now - @down_at
    Dalli.logger.debug { format('%<name>s is still down (for %<time>.3f seconds now)', name: name, time: time) }
  else
    @down_at = @last_down_at
    Dalli.logger.warn("#{name} is down")
  end
end

#log_up_detected ⇒ Object 



278
279
280
281
282
283
 
# File 'lib/dalli/protocol/connection_manager.rb', line 278

def log_up_detected
  return unless @down_at

  time = Time.now - @down_at
  Dalli.logger.warn { format('%<name>s is back (downtime was %<time>.3f seconds)', name: name, time: time) }
end

#log_warn_message(err_or_string) ⇒ Object 



247
248
249
250
251
252
 
# File 'lib/dalli/protocol/connection_manager.rb', line 247

def log_warn_message(err_or_string)
  Dalli.logger.warn do
    detail = err_or_string.is_a?(String) ? err_or_string : "#{err_or_string.class}: #{err_or_string.message}"
    "#{name} failed (count: #{@fail_count}) #{detail}"
  end
end

#max_allowed_failures ⇒ Object 



208
209
210
 
# File 'lib/dalli/protocol/connection_manager.rb', line 208

def max_allowed_failures
  @max_allowed_failures ||= @options[:socket_max_failures] || 2
end

#memcached_socket ⇒ Object 



239
240
241
242
243
244
245
 
# File 'lib/dalli/protocol/connection_manager.rb', line 239

def memcached_socket
  if socket_type == :unix
    Dalli::Socket::UNIX.open(hostname, options)
  else
    Dalli::Socket::TCP.open(hostname, port, options)
  end
end

#name ⇒ Object 



42
43
44
45
46
47
48
 
# File 'lib/dalli/protocol/connection_manager.rb', line 42

def name
  if socket_type == :unix
    hostname
  else
    "#{hostname}:#{port}"
  end
end

#raise_down_error ⇒ Object

Raises:



92
93
94
 
# File 'lib/dalli/protocol/connection_manager.rb', line 92

def raise_down_error
  raise Dalli::NetworkError, "#{name} is down: #{@error} #{@msg}"
end

#read(count) ⇒ Object Also known as: read_exact 



165
166
167
168
169
 
# File 'lib/dalli/protocol/connection_manager.rb', line 165

def read(count)
  @sock.readfull(count)
rescue SystemCallError, *TIMEOUT_ERRORS, *SSL_ERRORS, EOFError => e
  error_on_request!(e)
end

#read_available ⇒ Object

Non-blocking read. Here to support the operation of the get_multi operation



204
205
206
 
# File 'lib/dalli/protocol/connection_manager.rb', line 204

def read_available(...)
  @sock.read_available(...)
end

#read_line ⇒ Object 



154
155
156
157
158
159
160
 
# File 'lib/dalli/protocol/connection_manager.rb', line 154

def read_line
  data = @sock.gets("\r\n")
  error_on_request!('EOF in read_line') if data.nil?
  data
rescue SystemCallError, *TIMEOUT_ERRORS, *SSL_ERRORS, EOFError => e
  error_on_request!(e)
end

#reconnect!(message) ⇒ Object

Raises:



225
226
227
228
229
 
# File 'lib/dalli/protocol/connection_manager.rb', line 225

def reconnect!(message)
  close
  sleep(options[:socket_failure_delay]) if options[:socket_failure_delay]
  raise Dalli::RetryableNetworkError, message
end

#reconnect_down_server? ⇒ Boolean

Returns:

  • (Boolean) 


62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/dalli/protocol/connection_manager.rb', line 62

def reconnect_down_server?
  return true unless @last_down_at

  time_to_next_reconnect = @last_down_at + options[:down_retry_delay] - Time.now
  return true unless time_to_next_reconnect.positive?

  Dalli.logger.debug do
    format('down_retry_delay not reached for %<name>s (%<time>.3f seconds left)', name: name,
                                                                                  time: time_to_next_reconnect)
  end
  false
end

#reconnect_on_fork ⇒ Object 



254
255
256
257
258
259
260
 
# File 'lib/dalli/protocol/connection_manager.rb', line 254

def reconnect_on_fork
  message = 'Fork detected, re-connecting child process...'
  Dalli.logger.info { message }
  # Close socket on a fork and reconnect immediately
  close
  establish_connection
end

#request_in_progress? ⇒ Boolean

Returns:

  • (Boolean) 


134
135
136
 
# File 'lib/dalli/protocol/connection_manager.rb', line 134

def request_in_progress?
  @request_in_progress
end

#reset_down_info ⇒ Object 



231
232
233
234
235
236
237
 
# File 'lib/dalli/protocol/connection_manager.rb', line 231

def reset_down_info
  @fail_count = 0
  @down_at = nil
  @last_down_at = nil
  @msg = nil
  @error = nil
end

#socket_timeout ⇒ Object 



96
97
98
 
# File 'lib/dalli/protocol/connection_manager.rb', line 96

def socket_timeout
  @socket_timeout ||= @options[:socket_timeout]
end

#start_request! ⇒ Object 



138
139
140
141
142
 
# File 'lib/dalli/protocol/connection_manager.rb', line 138

def start_request!
  raise '[Dalli] Request already in progress. This may be a bug in Dalli.' if @request_in_progress

  @request_in_progress = true
end

#up! ⇒ Object 



75
76
77
78
 
# File 'lib/dalli/protocol/connection_manager.rb', line 75

def up!
  log_up_detected
  reset_down_info
end

#write(bytes) ⇒ Object 



182
183
184
185
186
 
# File 'lib/dalli/protocol/connection_manager.rb', line 182

def write(bytes)
  @sock.write(bytes)
rescue SystemCallError, *TIMEOUT_ERRORS, *SSL_ERRORS, IOError => e
  error_on_request!(e)
end