Class: Lich::Gemstone::Combat::AsyncProcessor

Inherits:
Object
  • Object
show all
Defined in:
documented/gemstone/combat/async_processor.rb

Overview

AsyncProcessor handles combat processing asynchronously.

This class is designed to manage multiple threads for combat processing, ensuring thread safety and performance.

See Also:

Instance Method Summary collapse

Constructor Details

#initialize(max_threads = 2) ⇒ void

Initializes a new AsyncProcessor instance.

Parameters:

  • max_threads (Integer) (defaults to: 2) —

    the maximum number of threads to use for processing



22
23
24
25
26
27
28
 
# File 'documented/gemstone/combat/async_processor.rb', line 22

def initialize(max_threads = 2)
  @max_threads = max_threads
  @active_count = Concurrent::AtomicFixnum.new(0)
  @thread_pool = []
  @last_cleanup = Time.now
  @last_compact = Time.now
end

Instance Method Details

#process_async(chunk) ⇒ Thread

Processes a chunk of combat data asynchronously.

This method will create a new thread to handle the processing of the provided chunk of data, ensuring that the number of active threads does not exceed the specified maximum.

Examples:

processor = Lich::Gemstone::Combat::AsyncProcessor.new
processor.process_async(["line1", "line2"])

Parameters:

  • chunk (Array<String>) —

    the chunk of combat data to process

Returns:

  • (Thread) —

    the thread that is processing the chunk



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
 
# File 'documented/gemstone/combat/async_processor.rb', line 40

def process_async(chunk)
  return if chunk.empty?

  # Periodic cleanup of dead threads (safety net)
  cleanup_dead_threads if Time.now - @last_cleanup > 30

  # Wait if at capacity
  while @active_count.value >= @max_threads
    sleep(0.01)
  end

  @active_count.increment

  # Create thread and store reference for self-cleanup
  thread = Thread.new do
    begin
      Thread.current[:start_time] = Time.now
      Thread.current[:line_count] = chunk.size

      Processor.process(chunk)

      elapsed = Time.now - Thread.current[:start_time]
      if elapsed > 0.5 && Tracker.debug?
        respond "[Combat] Processed #{chunk.size} lines in #{elapsed.round(3)}s"
      end
    rescue => e
      respond "[Combat] Processing error: #{e.message}" if Tracker.debug?
      respond e.backtrace.first(3) if Tracker.debug?
    ensure
      @active_count.decrement
      # Thread cleans itself up from pool when done (use Thread.current to avoid race)
      @thread_pool.delete(Thread.current)
    end
  end

  @thread_pool << thread
  thread
end

#shutdown ⇒ void

This method returns an undefined value.

Shuts down the AsyncProcessor, waiting for all threads to finish.

This method will block until all active threads have completed their execution and will perform garbage collection to help with memory management.



85
86
87
88
89
90
91
92
93
 
# File 'documented/gemstone/combat/async_processor.rb', line 85

def shutdown
  respond "[Combat] Waiting for #{@thread_pool.count(&:alive?)} threads..." if Tracker.debug?
  @thread_pool.each(&:join)
  @thread_pool.clear

  # Force GC after shutdown to help with memory fragmentation
  GC.start
  GC.compact if GC.respond_to?(:compact) # Ruby 2.7+
end

#stats ⇒ Hash

Returns statistics about the current state of the AsyncProcessor.

This includes the number of active threads, total alive threads, and other relevant metrics.

Returns:

  • (Hash) —

    a hash containing statistics about the processor's state



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'documented/gemstone/combat/async_processor.rb', line 100

def stats
  {
    active: @active_count.value,
    total_alive: @thread_pool.count(&:alive?),
    pool_size: @thread_pool.size, # ACTUAL array size (includes dead threads if not cleaned)
    dead_threads: @thread_pool.count { |t| !t.alive? }, # Count dead threads still in pool
    max_threads: @max_threads,
    processing: @thread_pool.select(&:alive?).map do |thread|
      {
        lines: thread[:line_count] || 0,
        elapsed: thread[:start_time] ? (Time.now - thread[:start_time]).round(2) : 0
      }
    end
  }
end