Class: HTTPX::Selector

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/httpx/selector.rb,
sig/selector.rbs

Overview

Implements the selector loop, where it registers and monitors "Selectable" objects.

A Selectable object is an object which can calculate the interests (:r, :w or :rw, respectively "read", "write" or "read-write") it wants to monitor for, and returns (via to_io method) an IO object which can be passed to functions such as IO.select . More exhaustively, a Selectable must implement the following methods:

state

returns the state as a Symbol, must return :closed when disposed of resources.

to_io

returns the IO object.

call

gets called when the IO is ready.

interests

returns the current interests to monitor for, as described above.

timeout

returns nil or an integer, representing how long to wait for interests.

handle_socket_timeout(Numeric)

called when waiting for interest times out.

Constant Summary collapse

READABLE =

Returns:

  • (Array[io_interests]) 
%i[rw r].freeze
WRITABLE =

Returns:

  • (Array[io_interests]) 
%i[rw w].freeze

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Selector

Returns a new instance of Selector.



34
35
36
37
38
 
# File 'lib/httpx/selector.rb', line 34

def initialize
  @timers = Timers.new
  @selectables = []
  @is_timer_interval = false
end

Instance Method Details

#deregister(io) ⇒ selectable?

deregisters io from selectables.

Parameters:

  • io (selectable)

Returns:

  • (selectable, nil) 


122
123
124
 
# File 'lib/httpx/selector.rb', line 122

def deregister(io)
  @selectables.delete(io)
end

#each ⇒ ::Enumerator[selectable, self] #each ⇒ self

Overloads:

  • #each ⇒ ::Enumerator[selectable, self]

    Returns:

    • (::Enumerator[selectable, self])
  • #each ⇒ self

    Returns:

    • (self)

Yields:

Yield Parameters:

  • arg0 (selectable)

Yield Returns:

  • (void) 


42
43
 
# File 'sig/selector.rbs', line 42

def each: () -> ::Enumerator[selectable, self]
| () { (selectable) -> void } -> self

#each_connection ⇒ void #each_connection ⇒ Enumerable[Connection]

Overloads:

  • #each_connection ⇒ void

    This method returns an undefined value.

  • #each_connection ⇒ Enumerable[Connection]

    Returns:

Yields:

Yield Parameters:

Yield Returns:

  • (void) 


96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/httpx/selector.rb', line 96

def each_connection(&block)
  return enum_for(__method__) unless block

  @selectables.each do |c|
    case c
    when Resolver::Resolver
      c.each_connection(&block)
    when Connection
      yield c
    end
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


40
41
42
 
# File 'lib/httpx/selector.rb', line 40

def empty?
  @selectables.empty? && @timers.empty?
end

#find_connection(request_uri, options) ⇒ Connection?

Parameters:

  • request_uri (http_uri)
  • options (Options)

Returns:



109
110
111
112
113
 
# File 'lib/httpx/selector.rb', line 109

def find_connection(request_uri, options)
  each_connection.find do |connection|
    connection.match?(request_uri, options)
  end
end

#find_mergeable_connection(connection) ⇒ Connection?

Parameters:

Returns:



115
116
117
118
119
 
# File 'lib/httpx/selector.rb', line 115

def find_mergeable_connection(connection)
  each_connection.find do |ch|
    ch != connection && ch.mergeable?(connection)
  end
end

#find_resolver(options) ⇒ Resolver::Resolver?

Parameters:

Returns:



87
88
89
90
91
92
93
94
 
# File 'lib/httpx/selector.rb', line 87

def find_resolver(options)
  res = @selectables.find do |c|
    c.is_a?(Resolver::Resolver) &&
      options.resolver_options_match?(c.options)
  end

  res.multi if res
end

#initial_call ⇒ Object

first time the registered selectables are added, there's probably work to do.



45
46
47
 
# File 'lib/httpx/selector.rb', line 45

def initial_call
  @selectables.each(&:initial_call)
end

#next_tick ⇒ void

This method returns an undefined value.



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/httpx/selector.rb', line 49

def next_tick
  catch(:jump_tick) do
    timeout = next_timeout
    if timeout && timeout.negative?
      @timers.fire
      throw(:jump_tick)
    end

    begin
      select(timeout) do |c|
        c.log(level: 2) { "[#{c.state}] selected from selector##{object_id} #{" after #{timeout} secs" unless timeout.nil?}..." }

        c.call
      end

      @timers.fire
    rescue TimeoutError => e
      @timers.fire(e)
    end
  end
end

#next_timeout ⇒ interval?

Returns:

  • (interval, nil) 


291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
 
# File 'lib/httpx/selector.rb', line 291

def next_timeout
  @is_timer_interval = false

  timer_interval = @timers.wait_interval

  connection_interval = @selectables.filter_map(&:timeout).min

  return connection_interval unless timer_interval

  if connection_interval.nil? || timer_interval <= connection_interval
    @is_timer_interval = true

    return timer_interval
  end

  connection_interval
end

#register(io) ⇒ void

This method returns an undefined value.

register io.

Parameters:

  • io (selectable) 


127
128
129
130
131
 
# File 'lib/httpx/selector.rb', line 127

def register(io)
  return if @selectables.include?(io)

  @selectables << io
end

#rw_wait(io, interval) ⇒ Object?

Parameters:

  • (io_select_selectable IO)
  • interval (interval, nil)

Returns:

  • (Object, nil) 


310
311
312
 
# File 'lib/httpx/selector.rb', line 310

def rw_wait(io, interval)
  io.to_io.wait(interval, :read_write)
end

#select(interval) {|arg0| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • interval (interval, nil)

Yields:

Yield Parameters:

  • arg0 (selectable)

Yield Returns:

  • (void) 


135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
 
# File 'lib/httpx/selector.rb', line 135

def select(interval, &block)
  # do not cause an infinite loop here.
  #
  # this may happen if timeout calculation actually triggered an error which causes
  # the connections to be reaped (such as the total timeout error) before #select
  # gets called.
  if @selectables.empty?
    begin
      sleep(interval)
    rescue IOError
      # @fiber-switch-guard
      # in a fiber scheduler scenario, IOs may be closed by the scheduler and raised in a separate fiber
      # on wakeup, which includes a sleep call.
    end if interval
    return
  end

  # @type var r: (selectable | Array[selectable])?
  # @type var w: (selectable | Array[selectable])?
  r, w = nil

  @selectables.delete_if do |io|
    interests = io.interests

    is_closed = io.state == :closed

    if is_closed
      # the process by which io was closed may have already triggered the on_close callback,
      # which already deregistered the io. this check prevents it from deleting the wrong io,
      # because of https://bugs.ruby-lang.org/issues/22021 .
      next(@selectables.include?(io))
    end

    if interests
      io.log(level: 2) do
        "[#{io.state}] registering in selector##{object_id} for select (#{interests})#{" for #{interval} seconds" unless interval.nil?}"
      end

      if READABLE.include?(interests)
        r = r.nil? ? io : (Array(r) << io)
      end

      if WRITABLE.include?(interests)
        w = w.nil? ? io : (Array(w) << io)
      end
    end

    is_closed
  end

  case r
  when Array
    w = Array(w) unless w.nil?

    select_many(r, w, interval, &block)
  when nil
    case w
    when Array
      select_many(r, w, interval, &block)
    when nil
      return unless interval && @selectables.any?

      # no selectables
      # TODO: replace with sleep?
      select_many(r, w, interval, &block)
    else
      select_one(w, :w, interval, &block)
    end

  else
    case w
    when Array
      select_many(Array(r), w, interval, &block)
    when nil
      select_one(r, :r, interval, &block)
    else
      if r == w
        select_one(r, :rw, interval, &block)
      else
        select_many(Array(r), Array(w), interval, &block)
      end
    end
  end
end

#select_many(r, w, interval) {|arg0| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • r (io_select_selectable)
  • w (io_select_selectable)
  • interval (interval, nil)

Yields:

Yield Parameters:

  • arg0 (selectable)

Yield Returns:

  • (void) 


220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/httpx/selector.rb', line 220

def select_many(r, w, interval, &block)
  begin
    readers, writers = ::IO.select(r, w, nil, interval)
  rescue IOError => e
    (Array(r) + Array(w)).each do |sel|
      # TODO: is there a way to cheaply find the IO associated with the error?
      sel.on_io_error(e)
    end
  rescue StandardError => e
    (Array(r) + Array(w)).each do |sel|
      sel.on_error(e)
    end

    return
  rescue Exception => e # rubocop:disable Lint/RescueException
    (Array(r) + Array(w)).each do |sel|
      sel.force_close(true)
    end

    raise e
  end

  if readers.nil? && writers.nil? && interval
    [*r, *w].each { |io| io.handle_socket_timeout(interval) }
    return
  end

  if writers
    readers.each do |io|
      yield io

      # so that we don't yield 2 times
      writers.delete(io)
    end if readers

    writers.each(&block)
  else
    readers.each(&block) if readers
  end
end

#select_one(io, interests, interval) {|io| ... } ⇒ void

This method returns an undefined value.

Parameters:

  • io (selectable)
  • interests (io_interests)
  • interval (interval, nil)

Yields:

  • (io)

Yield Parameters:

  • arg0 (selectable)

Yield Returns:

  • (void) 


261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
 
# File 'lib/httpx/selector.rb', line 261

def select_one(io, interests, interval)
  begin
    result =
      case interests
      when :r then io.to_io.wait_readable(interval)
      when :w then io.to_io.wait_writable(interval)
      when :rw then rw_wait(io, interval)
      end
  rescue IOError => e
    io.on_io_error(e)

    return
  rescue StandardError => e
    io.on_error(e)

    return
  rescue Exception => e # rubocop:disable Lint/RescueException
    io.force_close(true)

    raise e
  end

  unless result || interval.nil?
    io.handle_socket_timeout(interval) unless @is_timer_interval
    return
  end

  yield io
end

#terminate ⇒ void

This method returns an undefined value.



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/httpx/selector.rb', line 71

def terminate
  # array may change during iteration
  selectables = @selectables.reject(&:inflight?)

  selectables.delete_if do |sel|
    sel.terminate
    sel.state == :closed
  end

  until selectables.empty?
    next_tick

    selectables &= @selectables
  end
end