Class: Datadog::SymbolDatabase::ScopeBatcher Private

Inherits:
Object
  • Object
show all
Defined in:
lib/datadog/symbol_database/scope_batcher.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.

Batches extracted scopes and triggers uploads at appropriate times.

Implements two upload triggers:

  1. Size-based: Immediate upload when 400 scopes collected (MAX_SCOPES)
  2. Time-based: Upload after 1 second of inactivity (debounce timer, not periodic)

Also provides:

  • Deduplication: Tracks uploaded module names to prevent re-uploads
  • File limiting: Stops after 10,000 files to prevent runaway extraction
  • Thread safety: Mutex-protected state for concurrent access

Timer implementation: A single long-lived thread waits on a ConditionVariable with a timeout. Each add_scope signals the CV to reset the deadline. When the timeout expires without a signal, the timer fires and flushes the batch. This avoids creating/destroying a thread per add_scope call.

Flow: Extractor → add_scope → (batch or timer) → Uploader Created by: Component (during initialization) Calls: Uploader.upload_scopes when batch full or timer fires

Constant Summary collapse

MAX_SCOPES =

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

Maximum scopes per batch before triggering immediate upload. This matches the batch size used in Java and Python tracers to ensure consistent upload behavior across languages.

400
INACTIVITY_TIMEOUT =

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

seconds

1.0
MAX_FILES =

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

Maximum unique files to track before stopping extraction. This prevents runaway memory usage in applications with very large numbers of loaded classes (e.g., heavily modularized Rails apps).

10_000
TIMER_JOIN_TIMEOUT =

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

Seconds to wait for the timer thread to exit when joining during shutdown or reset. Bounded so a misbehaving thread cannot hang the caller indefinitely.

5

Instance Method Summary collapse

Constructor Details

#initialize(uploader, logger:, on_upload: nil, timer_enabled: true) ⇒ ScopeBatcher

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.

Initialize batching context.

Parameters:

  • uploader (Uploader) —

    Uploader instance for triggering uploads

  • logger (Logger) —

    Logger for diagnostics

  • on_upload (Proc, nil) (defaults to: nil) —

    Optional callback called after upload (for testing)

  • timer_enabled (Boolean) (defaults to: true) —

    Enable async timer (default true, false for tests)



49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/datadog/symbol_database/scope_batcher.rb', line 49

def initialize(uploader, logger:, on_upload: nil, timer_enabled: true)
  @uploader = uploader
  @logger = logger
  @on_upload = on_upload
  @timer_enabled = timer_enabled
  @scopes = []
  @mutex = Mutex.new
  @file_count = 0
  @uploaded_modules = Set.new

  # Timer state: single long-lived thread + ConditionVariable for debounce.
  # @timer_signaled is set to true on each add_scope and cleared by the timer
  # thread after waking. This flag is needed because ConditionVariable#wait
  # does not distinguish signal vs timeout on Ruby < 3.2 (returns self in both
  # cases). The flag gives a portable way to detect whether the wakeup was a
  # signal (reset deadline) or a timeout (fire the timer).
  @timer_cv = ConditionVariable.new
  @timer_thread = nil
  @timer_stopped = false
  @timer_signaled = false
end

Instance Method Details

#add_scope(scope) ⇒ 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.

Add a scope to the batch. Triggers immediate upload if batch reaches 400 scopes. Resets inactivity timer if batch not full.

Parameters:

  • scope (Scope) —

    The scope to add



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/datadog/symbol_database/scope_batcher.rb', line 76

def add_scope(scope)
  # @type var scopes_to_upload: ::Array[Scope]?
  scopes_to_upload = nil

  @mutex.synchronize do
    # Check file limit (counts only unique accepted files; duplicates are
    # filtered by the dedup check below and do not consume the budget).
    if @file_count >= MAX_FILES
      @logger.debug { "symdb: file limit (#{MAX_FILES}) reached, ignoring scope: #{scope.name}" }
      return
    end

    # Check if already uploaded — duplicates do not count toward MAX_FILES
    # so a re-extraction scenario does not exhaust the budget for unique scopes.
    if @uploaded_modules.include?(scope.name)
      @logger.trace { "symdb: skipping #{scope.name}: already uploaded" }
      return
    end

    # Marked uploaded before perform_upload runs. This is intentional:
    # symdb extraction is a one-shot operation per process — extract_all
    # walks ObjectSpace once and never revisits a module within the same
    # process. The Set deduplicates within a single extraction run, not
    # across upload attempts. Failed uploads are not retried; symbols
    # from a failed batch are lost until the next process restart, by
    # design (matches Python and Go; Java retries via OkHttp, .NET via
    # exponential backoff — Ruby does neither).
    @uploaded_modules.add(scope.name)
    @file_count += 1

    # Add the scope
    @scopes << scope

    # Check if batch size reached (AFTER adding)
    if @scopes.size >= MAX_SCOPES
      # Prepare for upload (clear within mutex)
      scopes_to_upload = @scopes.dup
      @scopes.clear
    end

    # Signal the timer thread to reset its inactivity deadline.
    # If batch was full, this is harmless — the timer will just
    # re-check and find an empty batch if it fires.
    ensure_timer_running
    @timer_signaled = true
    @timer_cv.signal
  end

  # Upload outside mutex (if batch was full)
  perform_upload(scopes_to_upload) if scopes_to_upload
rescue Exception => e # standard:disable Lint/RescueException
  Datadog::DI.reraise_if_fatal(e)
  @logger.debug { "symdb: failed to add scope: #{e.class}: #{e.message}" }
  # Don't propagate, continue operation
end

#flush ⇒ 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.

Force upload of current batch immediately.



134
135
136
137
138
139
140
141
142
143
144
145
146
 
# File 'lib/datadog/symbol_database/scope_batcher.rb', line 134

def flush
  # @type var scopes_to_upload: ::Array[Scope]?
  scopes_to_upload = nil

  @mutex.synchronize do
    return if @scopes.empty?

    scopes_to_upload = @scopes.dup
    @scopes.clear
  end

  perform_upload(scopes_to_upload)
end

#scopes_pending? ⇒ 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.

Check if scopes are pending upload.

Returns:

  • (Boolean) —

    true if scopes waiting in batch



180
181
182
 
# File 'lib/datadog/symbol_database/scope_batcher.rb', line 180

def scopes_pending?
  @mutex.synchronize { @scopes.any? }
end

#shutdown ⇒ 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.

Shutdown and upload remaining scopes.



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/datadog/symbol_database/scope_batcher.rb', line 150

def shutdown
  # @type var scopes_to_upload: ::Array[Scope]?
  scopes_to_upload = nil
  # @type var thread_to_join: ::Thread?
  thread_to_join = nil

  @mutex.synchronize do
    @timer_stopped = true
    @timer_cv.signal  # Wake the timer thread so it exits

    # Capture the timer thread under the mutex so a concurrent add_scope
    # cannot create a new thread that we'd accidentally orphan when we
    # nil the field below.
    thread_to_join = @timer_thread
    @timer_thread = nil

    scopes_to_upload = @scopes.dup
    @scopes.clear
  end

  # Join the timer thread outside the mutex.
  # The thread checks @timer_stopped and exits when signaled.
  thread_to_join&.join(TIMER_JOIN_TIMEOUT)

  # Upload outside mutex
  perform_upload(scopes_to_upload) unless scopes_to_upload.nil? || scopes_to_upload.empty?
end

#size ⇒ Integer

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.

Get current batch size.

Returns:

  • (Integer) —

    Number of scopes in current batch



186
187
188
 
# File 'lib/datadog/symbol_database/scope_batcher.rb', line 186

def size
  @mutex.synchronize { @scopes.size }
end