Module: Polyrun::Partition::Hrw

Defined in:
lib/polyrun/partition/hrw.rb

Overview

Rendezvous / highest-hash shard assignment (spec_queue.md): stateless, stable when m changes.

Class Method Summary collapse

Class Method Details

.normalize_shard_weights(shard_weights, total_shards) ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/polyrun/partition/hrw.rb', line 24

def normalize_shard_weights(shard_weights, total_shards)
  m = Integer(total_shards)
  return Array.new(m, 1.0) if shard_weights.nil? || shard_weights.empty?

  weights = shard_weights.map { |w| w.to_f }
  if weights.size < m
    weights += Array.new(m - weights.size, 1.0)
  elsif weights.size > m
    weights = weights[0, m]
  end
  weights
end

.pick_shard(path:, total_shards:, seed:) ⇒ Object

Raises:



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/polyrun/partition/hrw.rb', line 37

def pick_shard(path:, total_shards:, seed:)
  m = Integer(total_shards)
  raise Polyrun::Error, "total_shards must be >= 1" if m < 1

  best_j = 0
  best = -1.0
  salt = seed.to_s
  p = path.to_s
  m.times do |j|
    h = yield(p, j, salt)
    if h > best
      best = h
      best_j = j
    end
  end
  best_j
end

.score(path, shard_index, salt) ⇒ Object 



55
56
57
58
59
60
61
62
 
# File 'lib/polyrun/partition/hrw.rb', line 55

def score(path, shard_index, salt)
  digest = Digest::SHA256.digest("#{salt}\n#{path}\n#{shard_index}")
  if fast_score?
    digest.unpack1("Q>")
  else
    digest.unpack1("H*").hex
  end
end

.shard_for(path:, total_shards:, seed: "") ⇒ Integer

Returns shard index in 0...m.

Returns:

  • (Integer) —

    shard index in 0...m



10
11
12
 
# File 'lib/polyrun/partition/hrw.rb', line 10

def shard_for(path:, total_shards:, seed: "")
  pick_shard(path: path, total_shards: total_shards, seed: seed) { |p, j, salt| score(p, j, salt) }
end

.weighted_shard_for(path:, total_shards:, seed: "", shard_weights: nil) ⇒ Object

Per-shard weights (heterogeneous nodes). Uniform weights match shard_for.



15
16
17
18
19
20
21
22
 
# File 'lib/polyrun/partition/hrw.rb', line 15

def weighted_shard_for(path:, total_shards:, seed: "", shard_weights: nil)
  weights = normalize_shard_weights(shard_weights, total_shards)
  pick_shard(path: path, total_shards: total_shards, seed: seed) do |p, j, salt|
    base = score(p, j, salt).to_f
    w = weights[j]
    w.positive? ? base / w : base
  end
end