Module: Legion::Extensions::Llm::Vllm::Runners::DiscoveryRefresh::WeightPublication

Included in:
Legion::Extensions::Llm::Vllm::Runners::DiscoveryRefresh
Defined in:
lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb

Overview

Module-runner bindings for the shared writer reconciler. The existing discovery cadence remains the only execution path.

Instance Method Summary collapse

Instance Method Details

#activate_tracked_state(instance_id:, state:, probe_token:) ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 46

def activate_tracked_state(instance_id:, state:, probe_token:)
  Legion::Extensions::Llm::Inventory::WeightReconciler.activate_tracked!(
    settings: Legion::Settings,
    instance_id: instance_id,
    state_key: state.fetch(:name),
    state: state,
    states: instance_states,
    mutex: state_mutex,
    probe_token: probe_token,
    activate: method(:activate_weight_snapshot),
    activation_sequence: ->(tracked) { tracked.fetch(:sequence) }
  )
end

#activate_weight_snapshot(instance_id:, state:, offerings:, sequence:, probe_token:) ⇒ Object 



36
37
38
39
40
41
42
43
44
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 36

def activate_weight_snapshot(instance_id:, state:, offerings:, sequence:, probe_token:)
  publisher.activate_instance_snapshot(
    instance_id: instance_id,
    publisher_token: state.fetch(:publisher_token),
    offerings: offerings,
    sequence: sequence,
    probe_token: probe_token
  )
end

#dormant_weight_tracker ⇒ Object 



79
80
81
82
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 79

def dormant_weight_tracker
  @dormant_weight_tracker ||=
    Legion::Extensions::Llm::Inventory::DormantWeightTracker.new
end

#observe_dormant_weights ⇒ Object 



60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 60

def observe_dormant_weights
  Legion::Extensions::Llm::Inventory::WeightReconciler.observe_dormant!(
    settings: Legion::Settings,
    provider_family: :vllm,
    states: instance_states,
    mutex: state_mutex,
    tracker: dormant_weight_tracker,
    dormant_logger: lambda do |key|
      log.info(
        "[llm][vllm] action=dormant_weight weight_key=#{key.inspect} no_lane_published=true"
      )
    end
  )
end

#reconcile_weight_snapshot(instance_id:, state:, discovered_offerings:) ⇒ Object 



12
13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 12

def reconcile_weight_snapshot(instance_id:, state:, discovered_offerings:)
  Legion::Extensions::Llm::Inventory::WeightReconciler.commit_if_changed!(
    settings: Legion::Settings,
    instance_id: instance_id,
    state: state,
    discovered_offerings: discovered_offerings,
    mutex: state_mutex,
    equivalent: lambda do |previous, current|
      !instance_states[instance_id].equal?(state) ||
        offerings_equivalent?(previous, current)
    end,
    replace: method(:replace_weight_snapshot)
  )
end

#replace_weight_snapshot(instance_id:, state:, offerings:, sequence:) ⇒ Object 



27
28
29
30
31
32
33
34
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 27

def replace_weight_snapshot(instance_id:, state:, offerings:, sequence:)
  publisher.replace_instance_snapshot(
    instance_id: instance_id,
    publisher_token: state.fetch(:publisher_token),
    offerings: offerings,
    sequence: sequence
  )
end

#state_mutex ⇒ Object 



75
76
77
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/weight_publication.rb', line 75

def state_mutex
  @state_mutex ||= Mutex.new
end