Class: Ticketing::Broker

Inherits:
Object
  • Object
show all
Defined in:
lib/ticketing/broker.rb

Overview

A shared handle to a network distributed-lock service.

A broker opens one persistent TCP connection per address and load-balances requests across them round-robin. Connections are managed in the background: creation never blocks or fails, and a dropped connection is retried automatically without disturbing the others.

Create it once and share it for the whole process — it is safe to use from many threads at once.

broker = Ticketing::Broker.connect("127.0.0.1:5225", "127.0.0.1:5226")
ticket = broker.acquire("order-42", wait: 5, lease: 30)
# ... critical section, guarded by ticket.token at the resource ...
ticket.release

Constant Summary collapse

RELEASE_RETRY_WINDOW = 
5.0
RELEASE_RETRY_INTERVAL = 
0.2

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(conns, logger) ⇒ Broker

Returns a new instance of Broker.



52
53
54
55
56
57
 
# File 'lib/ticketing/broker.rb', line 52

def initialize(conns, logger)
  @conns = conns
  @logger = logger
  @rr = 0
  @rr_mutex = Mutex.new
end

Class Method Details

.connect(*addrs) ⇒ Object

Builds a broker for the given server addresses (as "host:port" strings, varargs or an array) and starts connecting in the background.

One address means a single node; two or more means a cluster, and the broker fails over between them. A single address is internally held with two connections to the same node, so a transient drop on one socket does not interrupt service.

Raises:

  • (ArgumentError) 


35
36
37
38
39
40
41
42
43
44
 
# File 'lib/ticketing/broker.rb', line 35

def self.connect(*addrs)
  list = addrs.flatten.map(&:to_s)
  raise ArgumentError, "at least one address is required" if list.empty?

  list = [list[0], list[0]] if list.size == 1
  logger = default_logger
  conns = list.map { |a| Conn.new(a, logger) }
  conns.each(&:start_connector)
  new(conns, logger)
end

.default_logger ⇒ Object 



46
47
48
49
50
 
# File 'lib/ticketing/broker.rb', line 46

def self.default_logger
  logger = Logger.new($stderr)
  logger.level = Logger::WARN
  logger
end

Instance Method Details

#acquire(key, wait:, lease:) ⇒ Object

Acquires the lock named key, waiting up to wait seconds for it to become free.

  • wait: how long to wait for the lock. 0 waits indefinitely. The wait is enforced by the server, so giving up never leaves a lock stranded. Sub-second values are rounded up to whole seconds.
  • lease: how long the server keeps the lock before auto-releasing it if this client dies without releasing (a safety net, not the expected hold time). Rounded up to whole seconds, minimum one second.

On success returns a Ticket carrying a monotonic fencing token. Raises TimeoutError if wait elapses first.

Raises:



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/ticketing/broker.rb', line 84

def acquire(key, wait:, lease:)
  Protocol.validate_key(key)
  frame = Protocol.acquire_frame(key, wait, lease)
  [@conns.size, 1].max.times do
    conn = pick or raise NotConnectedError

    begin
      reply = conn.request(Protocol::OP_ACQUIRE, key, frame)
      case reply
      when Reply::Acquired then return Ticket.new(self, conn, key, reply.token, @logger)
      when Reply::TIMED_OUT then raise TimeoutError
      else raise ProtocolError, "unexpected reply for acquire"
      end
    rescue IoError => e
      raise e unless e.cause_error.is_a?(ConnectionClosed)
    end
  end
  raise NotConnectedError
end

#close ⇒ Object

Cancels the background connections. The broker is unusable afterwards.



105
106
107
 
# File 'lib/ticketing/broker.rb', line 105

def close
  @conns.each(&:stop)
end

#release_lock(granted, key) ⇒ Object

Releases key on the connection that granted it, retrying within RELEASE_RETRY_WINDOW seconds. Returns true if the server still held the lock, false if it had already expired. Intended for internal use by Ticket.



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
 
# File 'lib/ticketing/broker.rb', line 113

def release_lock(granted, key)
  frame = Protocol.release_frame(key)
  deadline = monotonic + RELEASE_RETRY_WINDOW
  last_error = NotConnectedError.new
  loop do
    conn = granted.usable? ? granted : pick
    if conn
      begin
        reply = conn.request(Protocol::OP_RELEASE, key, frame)
        case reply
        when Reply::RELEASED then return true
        when Reply::NOT_FOUND then return false
        else raise ProtocolError, "unexpected reply for release"
        end
      rescue IoError => e
        raise e unless e.cause_error.is_a?(ConnectionClosed)

        last_error = Ticketing.closed_error
      end
    end
    raise last_error if monotonic >= deadline

    sleep RELEASE_RETRY_INTERVAL
  end
end

#wait_ready(timeout) ⇒ Object

Waits up to timeout seconds for at least one connection to come up. Returns true if the broker is ready, false if the timeout elapsed first. Best-effort convenience for startup; the broker works without it.



62
63
64
65
66
67
68
69
70
 
# File 'lib/ticketing/broker.rb', line 62

def wait_ready(timeout)
  deadline = monotonic + timeout
  loop do
    return true if pick
    return false if monotonic >= deadline

    sleep 0.02
  end
end