Class: ForestAdminDatasourceRpc::Utils::SchemaPollingPool

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb

Overview

Thread pool manager for RPC schema polling. Uses a single scheduler thread that dispatches polling tasks to a bounded pool of worker threads, preventing thread exhaustion when many RPC slaves are configured.

Design principles:

  • Minimal mutex hold times to avoid blocking HTTP request threads
  • Workers yield control frequently to prevent GIL starvation
  • Non-blocking queue operations where possible

Constant Summary collapse

DEFAULT_MAX_THREADS = 
5
MIN_THREADS = 
1
MAX_THREADS = 
50
SCHEDULER_INTERVAL = 
1
INITIAL_STAGGER_WINDOW = 
30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ SchemaPollingPool

Returns a new instance of SchemaPollingPool.



25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 25

def initialize
  @mutex = Mutex.new
  @clients = {}
  @work_queue = Queue.new
  @workers = []
  @running = false
  @max_threads = DEFAULT_MAX_THREADS
  @shutdown_requested = false
  @configured = false
  @scheduler_thread = nil
end

Instance Attribute Details

#configured ⇒ Object (readonly)

Returns the value of attribute configured.



23
24
25
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 23

def configured
  @configured
end

#max_threads ⇒ Object (readonly)

Returns the value of attribute max_threads.



23
24
25
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 23

def max_threads
  @max_threads
end

Instance Method Details

#client_count ⇒ Object 



96
97
98
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 96

def client_count
  @mutex.synchronize { @clients.size }
end

#configure(max_threads:) ⇒ Object

Configure the pool before starting. Must be called before any clients register.

Parameters:

  • max_threads (Integer) —

    Maximum number of worker threads (1-20)



39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 39

def configure(max_threads:)
  @mutex.synchronize do
    raise 'Cannot configure pool while running' if @running

    validated_max = max_threads.to_i.clamp(MIN_THREADS, MAX_THREADS)
    @max_threads = validated_max
    @configured = true

    log('Info', "[SchemaPollingPool] Configured with max_threads: #{@max_threads}")
  end
end

#register?(client_id, client) ⇒ Boolean

Returns:

  • (Boolean) 


51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 51

def register?(client_id, client)
  should_start = false

  @mutex.synchronize do
    if @clients.key?(client_id)
      log('Warn', "[SchemaPollingPool] Client #{client_id} already registered, skipping")
      return false
    end

    @clients[client_id] = {
      client: client,
      last_poll_at: nil,
      next_poll_at: calculate_initial_poll_time
    }

    log('Info', "[SchemaPollingPool] Registered client: #{client_id} (#{@clients.size} total clients)")

    should_start = !@running
  end

  start_pool if should_start

  true
end

#reset! ⇒ Object 



119
120
121
122
123
124
125
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 119

def reset!
  shutdown!
  @mutex.synchronize do
    @max_threads = DEFAULT_MAX_THREADS
    @configured = false
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


100
101
102
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 100

def running?
  @mutex.synchronize { @running }
end

#shutdown! ⇒ Object 



104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 104

def shutdown!
  @mutex.synchronize do
    return unless @running

    @shutdown_requested = true
  end

  stop_pool

  @mutex.synchronize do
    @clients.clear
    @shutdown_requested = false
  end
end

#unregister?(client_id) ⇒ Boolean

Returns:

  • (Boolean) 


76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/forest_admin_datasource_rpc/Utils/schema_polling_pool.rb', line 76

def unregister?(client_id)
  should_stop = false

  @mutex.synchronize do
    unless @clients.key?(client_id)
      log('Debug', "[SchemaPollingPool] Client #{client_id} not found for unregister")
      return false
    end

    @clients.delete(client_id)
    log('Info', "[SchemaPollingPool] Unregistered client: #{client_id} (#{@clients.size} remaining)")

    should_stop = @clients.empty? && @running
  end

  stop_pool if should_stop

  true
end