Class: Wurk::Topology

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/topology.rb

Overview

Worker topology DSL (Wurk extension on top of Ent's flat swarm). Lets users declare specialized slots: e.g. 2 forks dedicated to the critical queue with low concurrency, 2 forks for bulk + low with high concurrency. Stronger queue isolation than a flat swarm.

Each Slot describes a kind of fork; count is how many identical forks of that kind to spawn. Swarm consumes assignments (the flat list of forks to spawn, in order) so a slot with count=2 yields two assignment entries pointing at the same Slot.

See docs/idea/03-process-model.md §Worker topology.

Defined Under Namespace

Classes: Slot

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Topology

Returns a new instance of Topology.



20
21
22
 
# File 'lib/wurk/topology.rb', line 20

def initialize
  @slots = []
end

Class Method Details

.flat(count:, queues:, concurrency:) ⇒ Object

Convenience: build a flat topology of count identical forks consuming queues with concurrency threads each. Used by the railtie when the host hasn't declared a custom topology.



52
53
54
 
# File 'lib/wurk/topology.rb', line 52

def self.flat(count:, queues:, concurrency:)
  new.slot(count: count, queues: queues, concurrency: concurrency)
end

Instance Method Details

#assignments ⇒ Object

Flat ordered list of slots to fork, one per child process. A slot with count=N contributes N entries.



41
42
43
 
# File 'lib/wurk/topology.rb', line 41

def assignments
  @slots.flat_map { |s| Array.new(s.count, s) }
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


35
36
37
 
# File 'lib/wurk/topology.rb', line 35

def empty?
  @slots.empty?
end

#slot(count:, queues:, concurrency:) ⇒ Object

Declare one slot kind. Returns self so calls chain.



25
26
27
28
29
 
# File 'lib/wurk/topology.rb', line 25

def slot(count:, queues:, concurrency:)
  queue_list = validate_slot!(count, queues, concurrency)
  @slots << Slot.new(count: count, queues: queue_list, concurrency: concurrency).freeze
  self
end

#slots ⇒ Object 



31
32
33
 
# File 'lib/wurk/topology.rb', line 31

def slots
  @slots.dup.freeze
end

#total_processes ⇒ Object 



45
46
47
 
# File 'lib/wurk/topology.rb', line 45

def total_processes
  @slots.sum(&:count)
end