Class: RedisQueuedLocks::Swarm::FlushZombies Private

Inherits:
SwarmElement::Isolated show all
Defined in:
lib/redis_queued_locks/swarm/flush_zombies.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Since:

  • 1.9.0

Constant Summary

Constants included from Utilities

Utilities::RACTOR_LIVENESS_PATTERN, Utilities::RACTOR_STATUS_PATTERN, Utilities::RUBY_OVER_4

Instance Attribute Summary

Attributes inherited from SwarmElement::Isolated

#rql_client, #swarm_element, #sync

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from SwarmElement::Isolated

#initialize, #reswarm_if_dead!, #status, swarm_loop, #try_kill!, #try_swarm!

Methods included from Utilities

clock_gettime, ractor_alive?, ractor_status, ruby_over_4?, run_non_critical, thread_state

Constructor Details

This class inherits a constructor from RedisQueuedLocks::Swarm::SwarmElement::Isolated

Class Method Details

.flush_zombies(redis_client, zombie_ttl, lock_scan_size, queue_scan_size) ⇒ Hash<Symbol,Boolean|Set<String>]

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

rubocop:disable Metrics/MethodLength

Parameters:

  • redis_client (RedisClient)
  • zombie_ttl (Integer)
  • lock_scan_size (Integer)
  • queue_scan_size (Integer)

Returns:

  • (Hash<Symbol,Boolean|Set<String>]) —

    Format: { ok: , deleted_zombie_hosts: <Set>, deleted_zombie_acquirers: <Set>, deleted_zombie_locks: <Set> }

Since:

  • 1.9.0



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/redis_queued_locks/swarm/flush_zombies.rb', line 22

def flush_zombies(
  redis_client,
  zombie_ttl,
  lock_scan_size,
  queue_scan_size
)
  redis_client.with do |rconn|
    # Step 1:
    #   calculate zombie score (the time marker that shows acquirers that
    #   have not announced live probes for a long time)
    zombie_score = RedisQueuedLocks::Resource.calc_zombie_score(zombie_ttl / 1_000.0)

    # Step 2: extract zombie acquirers from the swarm list
    zombie_hosts = rconn.call('HGETALL', RedisQueuedLocks::Resource::SWARM_KEY)
    zombie_hosts = zombie_hosts.each_with_object(Set.new) do |(hst_id, ts), zombies|
      (zombies << hst_id) if (zombie_score > ts.to_f)
    end

    # Step X: exit if we have no any zombie acquirer
    next {
      ok: true,
      deleted_zombie_hosts: Set.new,
      deleted_zombie_acquirers: Set.new,
      deleted_zombie_locks: Set.new
    } if zombie_hosts.empty?

    # Step 3: find zombie locks held by zombies and delete them
    # TODO: indexing (in order to prevent full database scan);
    # NOTE: original redis does not support indexing so we need to use
    #   internal data structers to simulate data indexing (such as sorted sets or lists);
    zombie_locks = Set.new #: Set[String]
    zombie_acquirers = Set.new #: Set[String]

    rconn.scan(
      'MATCH', RedisQueuedLocks::Resource::LOCK_PATTERN, count: lock_scan_size
    ) do |lock_key|
      acquirer_id, host_id = rconn.call('HMGET', lock_key, 'acq_id', 'hst_id')
      if zombie_hosts.include?(host_id)
        zombie_locks << lock_key
        zombie_acquirers << acquirer_id
      end
    end

    # NOTE: (steep) steep can't use <Set>s for splats
    rconn.call('DEL', *zombie_locks) if zombie_locks.any? # steep:ignore

    # Step 4: find zombie requests => and drop them
    # TODO: indexing (in order to prevent full database scan);
    # NOTE: original redis does not support indexing so we need to use
    #   internal data structers to simulate data indexing (such as sorted sets or lists);
    rconn.scan(
      'MATCH', RedisQueuedLocks::Resource::LOCK_QUEUE_PATTERN, count: queue_scan_size
    ) do |lock_queue|
      zombie_acquirers.each do |zombie_acquirer|
        rconn.call('ZREM', lock_queue, zombie_acquirer)
      end
    end

    # Step 5: drop zombies from the swarm
    rconn.call('HDEL', RedisQueuedLocks::Resource::SWARM_KEY, *zombie_hosts)

    # Step 6: inform about deleted zombies
    {
      ok: true,
      deleted_zombie_hosts: zombie_hosts,
      deleted_zombie_acquirers: zombie_acquirers,
      deleted_zombie_locks: zombie_locks
    }
  end
end

Instance Method Details

#enabled? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean)

Since:

  • 1.9.0



99
100
101
 
# File 'lib/redis_queued_locks/swarm/flush_zombies.rb', line 99

def enabled?
  rql_client.config['swarm.flush_zombies.enabled_for_swarm']
end

#swarm! ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Since:

  • 1.9.0



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/redis_queued_locks/swarm/flush_zombies.rb', line 107

def swarm!
  @swarm_element = Ractor.new(
    rql_client.config.slice('swarm.flush_zombies.redis_config'),
    rql_client.config['swarm.flush_zombies.zombie_ttl'],
    rql_client.config['swarm.flush_zombies.zombie_lock_scan_size'],
    rql_client.config['swarm.flush_zombies.zombie_queue_scan_size'],
    rql_client.config['swarm.flush_zombies.zombie_flush_period']
  ) do |rc, z_ttl, z_lss, z_qss, z_fl_prd|
    RedisQueuedLocks::Swarm::FlushZombies.swarm_loop do
      Thread.new do
        redis_client = RedisQueuedLocks::Swarm::RedisClientBuilder.build(
          pooled: rc['pooled'],
          sentinel: rc['sentinel'],
          config: rc['config'],
          pool_config: rc['pool_config']
        )

        loop do
          RedisQueuedLocks::Swarm::FlushZombies.flush_zombies(
            redis_client, z_ttl, z_lss, z_qss
          )
          sleep(z_fl_prd)
        end
      end
    end
  end
end