Module: Synthra::PerformanceMode

Defined in:
lib/synthra/performance_mode.rb

Overview

Performance Mode

High-performance data generation for millions of records. Uses parallel processing, memory pooling, and native acceleration.

Examples:

Generate millions of records

Synthra::PerformanceMode.generate(schema, count: 1_000_000)

With progress callback

Synthra::PerformanceMode.generate(schema, count: 10_000_000) do |progress|
  puts "Generated #{progress[:current]} / #{progress[:total]} (#{progress[:percent]}%)"
end

Constant Summary collapse

DEFAULT_CONFIG =

Default configuration

{
  batch_size: 10_000,          # Records per batch
  threads: nil,                 # nil = auto (CPU cores)
  memory_limit_mb: 512,         # Max memory per batch
  use_native: true,             # Use Rust engine if available
  progress_interval: 100_000,   # Report progress every N records
  gc_interval: 500_000          # Force GC every N records
}.freeze

Class Method Summary collapse

Class Method Details

.benchmark(schema, counts: [1000, 10_000, 100_000]) ⇒ Hash

Benchmark generation performance

Parameters:

  • schema (Schema) —

    schema to benchmark

  • counts (Array<Integer>) (defaults to: [1000, 10_000, 100_000]) —

    record counts to test

Returns:

  • (Hash) —

    benchmark results



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/synthra/performance_mode.rb', line 164

def benchmark(schema, counts: [1000, 10_000, 100_000])
  results = {}
  
  counts.each do |count|
    start_time = Time.now
    records = schema.generate_many(count, mode: :random)
    elapsed = Time.now - start_time
    
    results[count] = {
      elapsed: elapsed.round(3),
      rate: safe_rate(count, elapsed),
      memory_mb: estimate_memory_usage
    }
    
    # Force GC between benchmarks
    records = nil
    GC.start
  end
  
  results
end

.estimate_memory_usage ⇒ Object (private) 



345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
 
# File 'lib/synthra/performance_mode.rb', line 345

def estimate_memory_usage
  # Try to get memory usage, fallback to 0 if not available
  if defined?(ObjectSpace) && ObjectSpace.respond_to?(:memsize_of_all)
    (ObjectSpace.memsize_of_all / 1024.0 / 1024.0).round(2)
  elsif RUBY_PLATFORM.include?("darwin") || RUBY_PLATFORM.include?("linux")
    begin
      # Use ps command as fallback
      pid = Process.pid
      mem = `ps -o rss= -p #{pid}`.strip.to_i
      (mem / 1024.0).round(2)
    rescue StandardError
      0.0
    end
  else
    0.0
  end
end

.format_csv_value(value) ⇒ Object (private) 



330
331
332
333
334
335
336
337
338
339
340
341
342
343
 
# File 'lib/synthra/performance_mode.rb', line 330

def format_csv_value(value)
  case value
  when nil
    ""
  when String
    value.include?(",") || value.include?('"') ? "\"#{value.gsub('"', '""')}\"" : value
  when Array
    "\"#{value.to_json}\""
  when Hash
    "\"#{value.to_json.gsub('"', '""')}\""
  else
    value.to_s
  end
end

.generate(schema, count:, **options) {|Hash| ... } ⇒ Array, Enumerator

Generate records in high-performance mode

Parameters:

  • schema (Schema) —

    schema to generate

  • count (Integer) —

    number of records

  • options (Hash) —

    generation options

Yields:

  • (Hash) —

    progress updates

Returns:

  • (Array, Enumerator) —

    generated records or enumerator



39
40
41
42
43
44
45
46
47
48
 
# File 'lib/synthra/performance_mode.rb', line 39

def generate(schema, count:, **options, &progress_callback)
  config = DEFAULT_CONFIG.merge(options)
  config[:threads] ||= optimal_thread_count
  
  if config[:use_native] && native_available?
    generate_native(schema, count, config, &progress_callback)
  else
    generate_parallel(schema, count, config, &progress_callback)
  end
end

.generate_native(schema, count, config, &progress_callback) ⇒ Object (private) 



229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/synthra/performance_mode.rb', line 229

def generate_native(schema, count, config, &progress_callback)
  require_relative "native" if defined?(Synthra::Native)
  
  results = []
  generated = 0
  start_time = Time.now
  seed = config[:seed] || Random.new_seed
  
  batches = (count.to_f / config[:batch_size]).ceil
  
  batches.times do |batch_num|
    batch_count = [config[:batch_size], count - generated].min
    batch_seed = seed + (batch_num * config[:batch_size])
    
    batch = Synthra::Native.generate_batch(
      schema.to_native_spec,
      batch_count,
      batch_seed,
      config[:threads]
    )
    
    results.concat(batch)
    generated += batch_count
    
    if progress_callback && generated % config[:progress_interval] == 0
      report_progress(progress_callback, generated, count, start_time)
    end
  end
  
  results
rescue StandardError
  # Fallback to parallel mode
  generate_parallel(schema, count, config, &progress_callback)
end

.generate_parallel(schema, count, config, &progress_callback) ⇒ Object (private) 



264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
 
# File 'lib/synthra/performance_mode.rb', line 264

def generate_parallel(schema, count, config, &progress_callback)
  results = []
  generated = 0
  start_time = Time.now
  seed = config[:seed] || Random.new_seed
  mutex = Mutex.new
  
  threads = config[:threads]
  records_per_thread = (count.to_f / threads).ceil
  
  worker_threads = threads.times.map do |thread_num|
    Thread.new do
      thread_start = thread_num * records_per_thread
      thread_count = [records_per_thread, count - thread_start].min
      next [] if thread_count <= 0
      
      thread_seed = seed + thread_start
      
      schema.generate_many(
        thread_count,
        seed: thread_seed,
        mode: config[:mode] || :random
      )
    end
  end
  
  worker_threads.each do |thread|
    batch = thread.value
    mutex.synchronize do
      results.concat(batch)
      generated = results.length
      
      if progress_callback && generated % config[:progress_interval] < batch.length
        report_progress(progress_callback, generated, count, start_time)
      end
    end
  end
  
  results
end

.native_available? ⇒ Boolean (private)

Returns:

  • (Boolean) 


202
203
204
205
206
 
# File 'lib/synthra/performance_mode.rb', line 202

def native_available?
  defined?(Synthra::Native) && Synthra::Native.respond_to?(:generate_batch)
rescue StandardError
  false
end

.optimal_thread_count ⇒ Object (private) 



208
209
210
211
 
# File 'lib/synthra/performance_mode.rb', line 208

def optimal_thread_count
  # Leave one core for the main thread
  [Etc.nprocessors - 1, 1].max
end

.recommended_batch_size ⇒ Object (private) 



213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/synthra/performance_mode.rb', line 213

def recommended_batch_size
  # Based on available memory
  available_mb = begin
    # Try to get actual available memory
    if RUBY_PLATFORM.include?("darwin")
      `sysctl -n hw.memsize`.to_i / 1024 / 1024 / 4
    else
      1024
    end
  rescue StandardError
    1024
  end
  
  [available_mb * 100, 100_000].min
end

.report_progress(callback, current, total, start_time) ⇒ Object (private) 



316
317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/synthra/performance_mode.rb', line 316

def report_progress(callback, current, total, start_time)
  elapsed = Time.now - start_time
  rate = safe_rate(current, elapsed)

  callback.call({
    current: current,
    total: total,
    percent: (current * 100.0 / total).round(1),
    rate: rate,
    elapsed: elapsed.round(1),
    eta: safe_eta(total - current, rate)
  })
end

.safe_eta(remaining, rate) ⇒ Object (private)

Seconds remaining, guarded against a 0 rate (which would otherwise divide by zero).



312
313
314
 
# File 'lib/synthra/performance_mode.rb', line 312

def safe_eta(remaining, rate)
  rate.positive? ? (remaining / rate.to_f).round(1) : 0.0
end

.safe_rate(records, elapsed) ⇒ Object (private)

Records/sec, guarded: sub-millisecond runs on fast hardware give elapsed == 0.0, and records / 0.0 is Float::INFINITY whose #round raises FloatDomainError. Return 0 instead.



307
308
309
 
# File 'lib/synthra/performance_mode.rb', line 307

def safe_rate(records, elapsed)
  elapsed.positive? ? (records / elapsed).round(0) : 0
end

.stream(schema, count:, **options) ⇒ Enumerator

Stream records without loading all into memory

Parameters:

  • schema (Schema) —

    schema to generate

  • count (Integer) —

    number of records

  • options (Hash) —

    generation options

Returns:

  • (Enumerator) —

    record enumerator



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
 
# File 'lib/synthra/performance_mode.rb', line 57

def stream(schema, count:, **options)
  config = DEFAULT_CONFIG.merge(options)
  
  Enumerator.new do |yielder|
    generated = 0
    last_gc = 0
    seed = options[:seed] || Random.new_seed

    while generated < count
      batch_size = [config[:batch_size], count - generated].min
      batch_seed = seed + generated

      batch = schema.generate_many(
        batch_size,
        seed: batch_seed,
        mode: options[:mode] || :random
      )

      batch.each { |record| yielder << record }
      generated += batch_size

      # Memory management — trigger on the gc_interval *delta*, not exact divisibility, so a
      # batch_size that doesn't evenly divide gc_interval can't skip every GC.
      if generated - last_gc >= config[:gc_interval]
        GC.start
        last_gc = generated
      end
    end
  end
end

.system_info ⇒ Hash

Get system capabilities

Returns:

  • (Hash) —

    system info



190
191
192
193
194
195
196
197
198
 
# File 'lib/synthra/performance_mode.rb', line 190

def system_info
  {
    cpu_cores: Etc.nprocessors,
    ruby_version: RUBY_VERSION,
    native_available: native_available?,
    optimal_threads: optimal_thread_count,
    recommended_batch_size: recommended_batch_size
  }
end

.to_file(schema, count:, output:, format: :ndjson, **options, &progress_callback) ⇒ Object

Write records directly to file (streaming)

Parameters:

  • schema (Schema) —

    schema to generate

  • count (Integer) —

    number of records

  • output (String, IO) —

    file path or IO object

  • format (Symbol) (defaults to: :ndjson) —

    output format (:json, :ndjson, :csv)

  • options (Hash) —

    generation options



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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/synthra/performance_mode.rb', line 96

def to_file(schema, count:, output:, format: :ndjson, **options, &progress_callback)
  config = DEFAULT_CONFIG.merge(options)
  
  io = output.is_a?(String) ? File.open(output, "w") : output
  generated = 0
  start_time = Time.now
  
  begin
    # Write header for CSV
    if format == :csv
      io.puts schema.fields.map(&:name).join(",")
    end
    
    # JSON array start
    if format == :json
      io.puts "["
    end
    
    stream(schema, count: count, **options).each_with_index do |record, index|
      case format
      when :json
        io.print "  "
        io.print JSON.generate(record)
        io.puts(index < count - 1 ? "," : "")
      when :ndjson
        io.puts JSON.generate(record)
      when :csv
        io.puts schema.fields.map { |f| format_csv_value(record[f.name]) }.join(",")
      end
      
      generated += 1
      
      if progress_callback && generated % config[:progress_interval] == 0
        elapsed = Time.now - start_time
        rate = safe_rate(generated, elapsed)
        progress_callback.call({
          current: generated,
          total: count,
          percent: (generated * 100.0 / count).round(1),
          rate: rate,
          elapsed: elapsed.round(1),
          eta: safe_eta(count - generated, rate)
        })
      end
    end
    
    # JSON array end
    if format == :json
      io.puts "]"
    end
  ensure
    io.close if output.is_a?(String)
  end
  
  final_elapsed = Time.now - start_time
  {
    records: generated,
    elapsed: final_elapsed,
    rate: safe_rate(generated, final_elapsed)
  }
end