Class: AceMQ::AMQP::PublishPermits Private

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/transport.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.

How many messages may be on the wire with no confirm back yet.

The ceiling is the only backpressure a publisher that does not wait per message has. Without it a caller who publishes faster than the broker confirms accumulates unconfirmed messages until the process dies, which looks like throughput right up to the moment it does not — and the bigger the array handed to Transport#publish_all, the closer that moment.

A permit is taken before the message is written and given back when the broker has answered for it, whichever way it answered: an ack, a nack and a basic.return are all answers. A message nothing ever answered for keeps its permit, because as far as the broker is concerned that message is still outstanding — which is what makes the ceiling bite when a broker stops keeping up rather than only when somebody hands in a huge array. Java holds the permit in exactly the same place, in a Semaphore released from its confirm listener, and .NET in a SemaphoreSlim.

Nothing waits here. Java can afford to block for its confirm timeout because its publishes are asynchronous and another thread's confirm can release a permit while this one waits; Ruby's transport publishes under the publishing channel's mutex, so the thread that would wait is the only thread that could ever release one, and a wait would be a deadlock dressed up as patience. Exhaustion is therefore reported straight away, in the sentence Java uses for it, rather than stalling silently.

Constant Summary collapse

DEFAULT =

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.

What Java's ConnectionConfig.maxOutstandingPublishes and .NET's MaxOutstandingPublishes default to. Large enough that an ordinary batch never notices it, small enough that a runaway publisher is stopped while the process still has memory to report it with.

1_000

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(limit = DEFAULT) ⇒ PublishPermits

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.

Returns a new instance of PublishPermits.

Parameters:

  • limit (Integer) (defaults to: DEFAULT) —

    how many publishes may await a confirm at once

Raises:



161
162
163
164
165
166
167
168
169
170
 
# File 'lib/acemq/amqp/transport.rb', line 161

def initialize(limit = DEFAULT)
  @limit = Integer(limit)
  if @limit < 1
    raise ConfigurationError,
          "max_outstanding_publishes must be at least 1, was #{limit.inspect}"
  end

  @available = @limit
  @lock = Mutex.new
end

Instance Attribute Details

#limit ⇒ Integer (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.

Returns the ceiling this was built with.

Returns:

  • (Integer) —

    the ceiling this was built with



157
158
159
 
# File 'lib/acemq/amqp/transport.rb', line 157

def limit
  @limit
end

Instance Method Details

#available ⇒ 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.

How many more messages may go out before one has to be answered for.



173
 
# File 'lib/acemq/amqp/transport.rb', line 173

def available = @lock.synchronize { @available }

#exhausted ⇒ 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.

What to say when there is no room left. The second sentence is Java's, word for word, so one runbook covers both.



205
206
207
208
209
 
# File 'lib/acemq/amqp/transport.rb', line 205

def exhausted
  "#{@limit} publishes are already waiting for a confirm and none of them " \
    "completed. The broker is not keeping up; publish more slowly rather " \
    "than buffering more."
end

#release(count) ⇒ 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.

Gives back the permits of messages the broker has answered for.



186
187
188
189
190
191
 
# File 'lib/acemq/amqp/transport.rb', line 186

def release(count)
  return if count.zero?

  @lock.synchronize { @available = [@available + count, @limit].min }
  nil
end

#reopened ⇒ 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.

Forgets everything outstanding, for a publishing channel that has been reopened. The messages those permits were held for went down with the old channel and can no longer be confirmed on the new one, which is the same conclusion Java's shutdown listener reaches when it fails every pending publish.



198
199
200
201
 
# File 'lib/acemq/amqp/transport.rb', line 198

def reopened
  @lock.synchronize { @available = @limit }
  nil
end

#take ⇒ 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.

Takes one permit, or answers false when there are none left.



176
177
178
179
180
181
182
183
 
# File 'lib/acemq/amqp/transport.rb', line 176

def take
  @lock.synchronize do
    next false if @available.zero?

    @available -= 1
    true
  end
end