Module: Politics::StaticQueueWorker

Defined in:
lib/politics/static_queue_worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#buckets ⇒ Object (readonly)

Returns the value of attribute buckets.



18
19
20
 
# File 'lib/politics/static_queue_worker.rb', line 18

def buckets
  @buckets
end

#dictatorship_length ⇒ Object (readonly)

Returns the value of attribute dictatorship_length.



18
19
20
 
# File 'lib/politics/static_queue_worker.rb', line 18

def dictatorship_length
  @dictatorship_length
end

#group_name ⇒ Object (readonly)

Returns the value of attribute group_name.



18
19
20
 
# File 'lib/politics/static_queue_worker.rb', line 18

def group_name
  @group_name
end

#iteration_length ⇒ Object (readonly)

Returns the value of attribute iteration_length.



18
19
20
 
# File 'lib/politics/static_queue_worker.rb', line 18

def iteration_length
  @iteration_length
end

#uri ⇒ Object (readonly)

Returns the value of attribute uri.



18
19
20
 
# File 'lib/politics/static_queue_worker.rb', line 18

def uri
  @uri
end

Instance Method Details

#alive? ⇒ Boolean 



154
155
156
 
# File 'lib/politics/static_queue_worker.rb', line 154

def alive?
  true
end

#as_dictator(&block) ⇒ Object 



80
81
82
83
84
85
86
87
 
# File 'lib/politics/static_queue_worker.rb', line 80

def as_dictator(&block)
  duration = dictatorship_length || (iteration_length * 10)
  log.debug { "become dictator for up to #{duration} seconds" }
  seize_leadership duration
  yield
  raise "lost leadership while being dictator for #{duration} seconds" unless leader?
  seize_leadership
end

#bucket_request(requestor_uri, context) ⇒ Object 



126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/politics/static_queue_worker.rb', line 126

def bucket_request(requestor_uri, context)
  if leader?
    log.debug "delivering bucket request"
    bucket_spec = next_bucket(requestor_uri, context)
    if !bucket_spec[0] && @followers_to_stop.include?(requestor_uri)
      bucket_spec = [:stop, 0]
      @followers_to_stop.delete(requestor_uri)
    end
    bucket_spec
  else
    log.debug "received request for bucket but am not leader - delivering :not_leader"
    [:not_leader, 0]
  end
end

#find_workers ⇒ Object 



168
169
170
 
# File 'lib/politics/static_queue_worker.rb', line 168

def find_workers
  raise "Please provide a method ”find_workers” returning a list of all other worker URIs"
end

#followers_to_stop ⇒ Object 



122
123
124
 
# File 'lib/politics/static_queue_worker.rb', line 122

def followers_to_stop
  @followers_to_stop.select {|u| DRbObject.new(nil, u).alive? rescue DRb::DRbConnError && false}
end

#leader ⇒ Object 



158
159
160
161
162
163
164
165
166
 
# File 'lib/politics/static_queue_worker.rb', line 158

def leader
  2.times do
    break if leader_uri
    log.debug "could not determine leader - relaxing until next iteration"
    relax until_next_iteration
  end
  raise "cannot determine leader" unless leader_uri
  DRbObject.new(nil, leader_uri)
end

#next_bucket(requestor_uri, context) ⇒ Object 



141
142
143
 
# File 'lib/politics/static_queue_worker.rb', line 141

def next_bucket(requestor_uri, context)
  [@buckets.pop, sleep_until_next_bucket_time]
end

#perform_leader_duties ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/politics/static_queue_worker.rb', line 94

def perform_leader_duties
  # Drb thread handles requests to leader
  as_dictator do
    initialize_buckets
  end
  # keeping leader state as long as buckets are available by renominating before nomination
  # times out
  while !buckets.empty? do
    log.debug { "relaxes half the time until next iteration" }
    relax(until_next_iteration / 2)
    as_dictator do
      update_buckets unless restart_wanted?
    end
  end
  as_dictator() {populate_followers_to_stop} if restart_wanted?
  # keeping leader state as long as there are followers to stop
  while !followers_to_stop.empty? do
    relax(until_next_iteration / 2)
    seize_leadership
  end
  exit 0 if restart_wanted?
  relax until_next_iteration
end

#populate_followers_to_stop ⇒ Object 



118
119
120
 
# File 'lib/politics/static_queue_worker.rb', line 118

def populate_followers_to_stop
  @followers_to_stop.replace(find_workers)
end

#process_bucket(&block) ⇒ Object

Fetch a bucket out of the queue and pass it to the given block to be processed.

Raises:

  • (ArgumentError) 


45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/politics/static_queue_worker.rb', line 45

def process_bucket(&block)
  log.debug "start bucket processing"
  raise ArgumentError, "process_bucket requires a block!" unless block_given?
  unless @memcache_client
    raise ArgumentError, "You must call register_worker before processing!"
  end

  begin
    begin
      raise "self is not alive via drb" unless DRbObject.new(nil, uri).alive?
    rescue Exception => e
      raise "cannot reach self via drb: #{e.message}"
    end
    begin
      nominate
      if leader?
        log.info { "has been elected leader" }
        perform_leader_duties
      else
        # Get a bucket from the leader and process it
        begin
          log.debug "getting bucket request from leader (#{leader_uri}) and processing it"
          bucket_process(*leader.bucket_request(uri, bucket_request_context), &block)
        rescue DRb::DRbError => dre
          log.error { "Error talking to leader: #{dre.message}" }
          relax until_next_iteration
        end
      end
    rescue MemCache::MemCacheError => e
      log.error { "Unexpected MemCacheError: #{e.message}" }
      relax until_next_iteration
    end
  end while loop?
end

#register_worker(name, bucket_count, config = {}) ⇒ Object

Register this process as able to work on buckets.



21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/politics/static_queue_worker.rb', line 21

def register_worker(name, bucket_count, config = {})
  options = config.dup
  options[:iteration_length] ||= 10
  options[:servers] ||= ['127.0.0.1:11211']

  @group_name = name
  @iteration_length = options[:iteration_length]
  @memcache_client = client_for(Array(options[:servers]))
  @nominated_at = Time.now
  @dictatorship_length = options[:dictatorship_length]

  @buckets = []
  @bucket_count = bucket_count
  @followers_to_stop = Set.new

  start_druby_service
  log.progname = uri
  log.info { "Registered in group #{group_name} at #{uri}" }
  at_exit do
    cleanup
  end
end

#seize_leadership(duration = iteration_length) ⇒ Object 



89
90
91
92
 
# File 'lib/politics/static_queue_worker.rb', line 89

def seize_leadership(duration = iteration_length)
  @memcache_client.set(token, uri, duration)
  @nominated_at = Time.now + duration - iteration_length
end

#sleep_until_next_bucket_time ⇒ Object 



145
146
147
 
# File 'lib/politics/static_queue_worker.rb', line 145

def sleep_until_next_bucket_time
  [[until_next_iteration / 2, 1].max, iteration_length / 2].min
end

#until_next_iteration ⇒ Object 



149
150
151
152
 
# File 'lib/politics/static_queue_worker.rb', line 149

def until_next_iteration
  left = iteration_length - (Time.now - @nominated_at)
  left > 0 ? left : 0
end