Class: Hive::Pool

Inherits:
Object
  • Object
show all
Includes:
Log, Utilities::Hash
Defined in:
lib/hive/pool.rb

Overview

A pool is a collection of workers, each of which is a separate process.

All workers are of the same kind (class).

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Utilities::Hash

#assert_valid_keys

Methods included from Log

#format_for_logging, #log, #logger, #logger=

Constructor Details

#initialize(kind, policy_prototype = {}) ⇒ Pool

Returns a new instance of Pool.



21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/hive/pool.rb', line 21

def initialize( kind, policy_prototype = {} )
  if kind.kind_of?(Array) then
    kind, policy_prototype = kind.first, kind.last
  end
  @kind     = kind
  @policy   = Hive::Policy.resolve(policy_prototype) or raise
  @name     = @policy.name || kind.name or raise Hive::ConfigurationError, "Pool or Job must have a name"
  @storage  = policy.storage
  @registry = Hive::Registry.new( name, storage )

  # type checks
  policy.pool_min_workers
  registry.workers
end

Instance Attribute Details

#kind ⇒ Object (readonly)

job class



15
16
17
 
# File 'lib/hive/pool.rb', line 15

def kind
  @kind
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



16
17
18
 
# File 'lib/hive/pool.rb', line 16

def name
  @name
end

#policy ⇒ Object (readonly)

Returns the value of attribute policy.



17
18
19
 
# File 'lib/hive/pool.rb', line 17

def policy
  @policy
end

#registry ⇒ Object (readonly)

Returns the value of attribute registry.



18
19
20
 
# File 'lib/hive/pool.rb', line 18

def registry
  @registry
end

#storage ⇒ Object (readonly)

where to store worker details



19
20
21
 
# File 'lib/hive/pool.rb', line 19

def storage
  @storage
end

Instance Method Details

#mq ⇒ Object 



74
75
76
77
78
79
 
# File 'lib/hive/pool.rb', line 74

def mq
  @mq ||= begin
    key = Hive::Key.new( "#{name}-pool", Process.pid )
    me  = Hive::Messager.new storage, my_address: key
  end
end

#reap(key, options = {}) ⇒ Object

shut down a worker



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/hive/pool.rb', line 116

def reap( key, options = {} )
  assert_valid_keys( options, :wait )
  wait = options[:wait]

  if key.host == Hive::Key.local_host then
    ::Process.kill( "TERM", key.pid )
    Hive::Utilities::Process.wait_and_terminate key.pid, timeout: 10
  else
    mq.send "Quit", to: key
  end

  if wait then
    Hive::Idler.wait_until( 10 ) do
      live = registry.checked_workers( policy ).live
      ! live.member? key
    end
  end
end

#spawn(options = {}) ⇒ Object

this really should be protected but it’s convenient to be able to force a spawn param options can true to wait until after the process is spawned



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/hive/pool.rb', line 94

def spawn( options = {} )
  assert_valid_keys( options, :wait )
  wait = options[:wait]

  if ! wait then
    Hive::Worker.spawn kind, registry: registry, policy: policy, name: name
    return
  end

  before = registry.checked_workers( policy ).live

  Hive::Worker.spawn kind, registry: registry, policy: policy, name: name

  Hive::Idler.wait_until( 10 ) do
    after = registry.checked_workers( policy ).live
    diff  = ( after - before ).select { |k| k.host == Hive::Key.local_host }
    diff.size > 0
  end
end

#stop_all ⇒ Object

tell all workers to quit



83
84
85
86
87
88
89
 
# File 'lib/hive/pool.rb', line 83

def stop_all
  checklist = registry.checked_workers( policy )
  checklist.live.each { |key| reap(key) }
  checklist.late.each { |key| reap(key) }
  checklist.hung.each { |key| reap(key) }
  checklist.dead.each { |key| reap(key) }
end

#synchronize(options = {}) ⇒ Object

Parameters:

  • options (:log) (defaults to: {}) —

    can be true



39
40
41
42
43
44
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
 
# File 'lib/hive/pool.rb', line 39

def synchronize( options = {} )
  assert_valid_keys( options, :log )
  do_log = options[:log]

  checklist  = registry.checked_workers( policy )
  live_count = checklist.live.size

  if do_log then
    check_live_workers( checklist )
    check_late_workers( checklist )
    check_hung_workers( checklist )
    check_dead_workers( checklist )
  end

  if (need = policy.pool_min_workers - live_count) > 0 then
    # launch workers
    need.times do
      spawn wait: true
    end

  elsif (excess = live_count - policy.pool_max_workers) > 0 then
    # spin down some workers
    # try to find LOCAL workers to spin down first
    locals = checklist.live.select { |k| k.host == Hive::Key.local_host }
    if locals.size > 0 then
      reap locals.first, wait: true
    else
      reap checklist.live.first, wait: true
    end
  end

  checklist = registry.checked_workers( policy )
end