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

Returns:

  • (Boolean) 


156
157
158
 
# File 'lib/politics/static_queue_worker.rb', line 156

def alive?
  true
end

#as_dictator(&block) ⇒ Object 



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

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 



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

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 



170
171
172
 
# File 'lib/politics/static_queue_worker.rb', line 170

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

#followers_to_stop ⇒ Object 



124
125
126
 
# File 'lib/politics/static_queue_worker.rb', line 124

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

#leader ⇒ Object 



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

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 



143
144
145
 
# File 'lib/politics/static_queue_worker.rb', line 143

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

#perform_leader_duties ⇒ Object 



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

def perform_leader_duties
  before_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 



120
121
122
 
# File 'lib/politics/static_queue_worker.rb', line 120

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) 


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
79
 
# File 'lib/politics/static_queue_worker.rb', line 46

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
43
 
# 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
    internal_cleanup
    cleanup
  end
end

#seize_leadership(duration = iteration_length) ⇒ Object 



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

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 



147
148
149
 
# File 'lib/politics/static_queue_worker.rb', line 147

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

#until_next_iteration ⇒ Object 



151
152
153
154
 
# File 'lib/politics/static_queue_worker.rb', line 151

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