Class: Collective::Pool

Inherits:
Object
  • Object
show all
Includes:
Log
Defined in:
lib/collective/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 Log

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

Constructor Details

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

Returns a new instance of Pool.



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

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

  # type checks
  policy.pool_min_workers
  registry.workers
end

Instance Attribute Details

#kind ⇒ Object (readonly)

job class



14
15
16
 
# File 'lib/collective/pool.rb', line 14

def kind
  @kind
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



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

def name
  @name
end

#policy ⇒ Object (readonly)

Returns the value of attribute policy.



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

def policy
  @policy
end

#registry ⇒ Object (readonly)

Returns the value of attribute registry.



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

def registry
  @registry
end

#storage ⇒ Object (readonly)

where to store worker details



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

def storage
  @storage
end

Instance Method Details

#mq ⇒ Object 



73
74
75
76
77
78
 
# File 'lib/collective/pool.rb', line 73

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

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

shut down a worker



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

def reap( key, options = {} )
  wait = options.delete(:wait)
  raise if options.size > 0

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

  if wait then
    Collective::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



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

def spawn( options = {} )
  wait = options.delete(:wait)
  raise if options.size > 0

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

  before = registry.checked_workers( policy ).live

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

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

#stop_all ⇒ Object

tell all workers to quit



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

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



38
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
 
# File 'lib/collective/pool.rb', line 38

def synchronize( options = {} )
  do_log = options.delete(:log)
  raise if options.size > 0

  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 == Collective::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