Class: Gru::Adapters::RedisAdapter

Inherits:
Object
  • Object
show all
Defined in:
lib/gru/adapters/redis_adapter.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(settings) ⇒ RedisAdapter

Returns a new instance of RedisAdapter.



8
9
10
11
 
# File 'lib/gru/adapters/redis_adapter.rb', line 8

def initialize(settings)
  @settings = settings
  @client = initialize_client(settings.client_settings)
end

Instance Attribute Details

#client ⇒ Object (readonly)

Returns the value of attribute client.



6
7
8
 
# File 'lib/gru/adapters/redis_adapter.rb', line 6

def client
  @client
end

Instance Method Details

#expire_workers ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/gru/adapters/redis_adapter.rb', line 38

def expire_workers
  removable = {}
  workers = max_host_workers
  workers.each do |worker, count|
    removable[worker] = with_worker_counts(worker,count) do |total|
      if expire_worker?(worker)
        total -= 1 if expire_worker(worker)
      end
      total
    end
  end
  removable
end

#provision_workers ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/gru/adapters/redis_adapter.rb', line 24

def provision_workers
  available = {}
  workers = max_host_workers
  workers.each do |worker, count|
    available[worker] = with_worker_counts(worker,count) do |total|
      if reserve_worker?(worker)
        total += 1 if reserve_worker(worker)
      end
      total
    end
  end
  available
end

#release_presumed_dead_worker_hosts ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/gru/adapters/redis_adapter.rb', line 67

def release_presumed_dead_worker_hosts
  return false unless manage_heartbeat?
  update_heartbeat
  presumed_dead_worker_hosts.each_pair do |hostname,timestamp|
    lock_key = "#{gru_key}:removing_dead_host:#{hostname}"
    if send_message(:setnx,lock_key,Time.now.to_i)
      remove_worker_host(hostname)
      send_message(:hdel,heartbeat_key,hostname)
      send_message(:del,lock_key)
      return true
    end
  end
  false
end

#release_workers ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/gru/adapters/redis_adapter.rb', line 52

def release_workers
  workers = max_host_workers
  workers.keys.each do |worker|
    host_count = local_running_count(worker)
    global_count = host_count
    host_count.times do
      global_count = send_message(:hincrby, global_workers_running_key, worker, -1) if global_count > 0
      host_count = send_message(:hincrby, host_workers_running_key,worker,-1) if host_count > 0
    end
  end
  send_message(:del, host_workers_running_key)
  send_message(:del, host_max_worker_key)
  send_message(:hdel, heartbeat_key, hostname)
end

#set_worker_counts ⇒ Object 



13
14
15
16
17
18
19
20
21
22
 
# File 'lib/gru/adapters/redis_adapter.rb', line 13

def set_worker_counts
  set_rebalance_flag(@settings.rebalance_flag)
  set_presume_host_dead_after(@settings.presume_host_dead_after)
  release_workers
  register_workers(@settings.host_maximums)
  set_max_worker_counts(@settings.host_maximums)
  register_global_workers(@settings.cluster_maximums)
  set_max_global_worker_counts(@settings.cluster_maximums)
  update_heartbeat if manage_heartbeat?
end