Class: Que::Locker

Inherits:
Object
  • Object
show all
Defined in:
lib/que/locker.rb

Constant Summary collapse

MESSAGE_RESOLVERS = 
{}
RESULT_RESOLVERS = 
{}
DEFAULT_POLL_INTERVAL = 
5.0
DEFAULT_WAIT_PERIOD = 
50
DEFAULT_MAXIMUM_BUFFER_SIZE = 
8
DEFAULT_WORKER_PRIORITIES = 
[10, 30, 50, nil, nil, nil].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(queues: [Que.default_queue], connection_url: nil, listen: true, poll: true, poll_interval: DEFAULT_POLL_INTERVAL, wait_period: DEFAULT_WAIT_PERIOD, maximum_buffer_size: DEFAULT_MAXIMUM_BUFFER_SIZE, worker_priorities: DEFAULT_WORKER_PRIORITIES, on_worker_start: nil) ⇒ Locker

Returns a new instance of Locker.



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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
109
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
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/que/locker.rb', line 55

def initialize(
  queues:              [Que.default_queue],
  connection_url:      nil,
  listen:              true,
  poll:                true,
  poll_interval:       DEFAULT_POLL_INTERVAL,
  wait_period:         DEFAULT_WAIT_PERIOD,
  maximum_buffer_size: DEFAULT_MAXIMUM_BUFFER_SIZE,
  worker_priorities:   DEFAULT_WORKER_PRIORITIES,
  on_worker_start:     nil
)

  # Sanity-check all our arguments, since some users may instantiate Locker
  # directly.
  Que.assert [TrueClass, FalseClass], listen
  Que.assert [TrueClass, FalseClass], poll

  Que.assert Numeric, poll_interval
  Que.assert Numeric, wait_period

  Que.assert Array, worker_priorities
  worker_priorities.each { |p| Que.assert([Integer, NilClass], p) }

  # We assign this globally because we only ever expect one locker to be
  # created per worker process. This can be used by middleware or external
  # code to access the locker during runtime.
  Que.locker = self

  # We use a JobBuffer to track jobs and pass them to workers, and a
  # ResultQueue to receive messages from workers.
  @job_buffer = JobBuffer.new(
    maximum_size: maximum_buffer_size,
    priorities:   worker_priorities.uniq,
  )

  @result_queue = ResultQueue.new

  @stop = false

  Que.internal_log :locker_instantiate, self do
    {
      queues:              queues,
      listen:              listen,
      poll:                poll,
      poll_interval:       poll_interval,
      wait_period:         wait_period,
      maximum_buffer_size: maximum_buffer_size,
      worker_priorities:   worker_priorities,
    }
  end

  # Local cache of which advisory locks are held by this connection.
  @locks = Set.new

  @poll_interval = poll_interval

  if queues.is_a?(Hash)
    @queue_names = queues.keys
    @queues = queues.transform_values do |interval|
      interval || poll_interval
    end
  else
    @queue_names = queues
    @queues = queues.map do |queue_name|
      [queue_name, poll_interval]
    end.to_h
  end

  @wait_period = wait_period.to_f / 1000 # Milliseconds to seconds.

  @workers =
    worker_priorities.map do |priority|
      Worker.new(
        priority:       priority,
        job_buffer:     @job_buffer,
        result_queue:   @result_queue,
        start_callback: on_worker_start,
      )
    end

  # To prevent race conditions, let every worker get into a ready state
  # before starting up the locker thread.
  loop do
    break if job_buffer.waiting_count == workers.count
    sleep 0.001
  end

  # If we weren't passed a specific connection_url, borrow a connection from
  # the pool and derive the connection string from it.
  connection_args =
    if connection_url
      uri = URI.parse(connection_url)

      opts =
        {
          host:     uri.host,
          user:     uri.user,
          password: uri.password,
          port:     uri.port || 5432,
          dbname:   uri.path[1..-1],
        }

      if uri.query
        opts.merge!(Hash[uri.query.split("&").map{|s| s.split('=')}.map{|a,b| [a.to_sym, b]}])
      end

      opts
    else
      Que.pool.checkout do |conn|
        c = conn.wrapped_connection

        {
          host:     c.host,
          user:     c.user,
          password: c.pass,
          port:     c.port,
          dbname:   c.db,
        }
      end
    end

  @connection = Que::Connection.wrap(PG::Connection.open(connection_args))

  @thread =
    Thread.new do
      # An error causing this thread to exit is a bug in Que, which we want
      # to know about ASAP, so propagate the error if it happens.
      Thread.current.abort_on_exception = true

      # Give this thread priority, so it can promptly respond to NOTIFYs.
      Thread.current.priority = 1

      begin
        unless connection_args.has_key?(:application_name)
          @connection.execute(
            "SELECT set_config('application_name', $1, false)",
            ["Que Locker: #{@connection.backend_pid}"]
          )
        end

        Poller.setup(@connection)

        @listener =
          if listen
            Listener.new(connection: @connection)
          end

        @pollers =
          if poll
            @queues.map do |queue_name, interval|
              Poller.new(
                connection:    @connection,
                queue:         queue_name,
                poll_interval: interval,
              )
            end
          end

        work_loop
      ensure
        @connection.wrapped_connection.close
      end
    end
end

Instance Attribute Details

#job_buffer ⇒ Object (readonly)

Returns the value of attribute job_buffer.



36
37
38
 
# File 'lib/que/locker.rb', line 36

def job_buffer
  @job_buffer
end

#locks ⇒ Object (readonly)

Returns the value of attribute locks.



36
37
38
 
# File 'lib/que/locker.rb', line 36

def locks
  @locks
end

#poll_interval ⇒ Object (readonly)

Returns the value of attribute poll_interval.



36
37
38
 
# File 'lib/que/locker.rb', line 36

def poll_interval
  @poll_interval
end

#queues ⇒ Object (readonly)

Returns the value of attribute queues.



36
37
38
 
# File 'lib/que/locker.rb', line 36

def queues
  @queues
end

#thread ⇒ Object (readonly)

Returns the value of attribute thread.



36
37
38
 
# File 'lib/que/locker.rb', line 36

def thread
  @thread
end

#workers ⇒ Object (readonly)

Returns the value of attribute workers.



36
37
38
 
# File 'lib/que/locker.rb', line 36

def workers
  @workers
end

Instance Method Details

#stop ⇒ Object 



224
225
226
227
 
# File 'lib/que/locker.rb', line 224

def stop
  @job_buffer.stop
  @stop = true
end

#stop! ⇒ Object 



220
221
222
 
# File 'lib/que/locker.rb', line 220

def stop!
  stop; wait_for_stop
end

#stopping? ⇒ Boolean

Returns:

  • (Boolean) 


229
230
231
 
# File 'lib/que/locker.rb', line 229

def stopping?
  @stop
end

#wait_for_stop ⇒ Object 



233
234
235
 
# File 'lib/que/locker.rb', line 233

def wait_for_stop
  @thread.join
end