Class: Langsmith::BatchProcessor

Inherits:
Object
  • Object
show all
Defined in:
lib/langsmith/batch_processor.rb

Overview

Background processor that batches trace runs and sends them to LangSmith. Uses concurrent-ruby for thread-safe operations.

Thread Safety:

  • Uses AtomicBoolean for atomic start/shutdown
  • Uses @pending_mutex to protect all pending array access (add + extract)
  • Uses @flush_mutex to ensure only one flush operation runs at a time
  • HTTP calls happen outside locks to avoid blocking the worker

Constant Summary collapse

CREATE =

Entry types for the queue

:create
UPDATE = 
:update
SHUTDOWN = 
:shutdown

Instance Method Summary collapse

Constructor Details

#initialize(client: nil, batch_size: nil, flush_interval: nil, max_pending_entries: nil) ⇒ BatchProcessor

Returns a new instance of BatchProcessor.



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/langsmith/batch_processor.rb', line 20

def initialize(client: nil, batch_size: nil, flush_interval: nil, max_pending_entries: nil)
  config = Langsmith.configuration
  @client = client || Client.new
  @batch_size = batch_size || config.batch_size
  @flush_interval = flush_interval || config.flush_interval
  @max_pending_entries = max_pending_entries || config.max_pending_entries

  @queue = Queue.new
  @running = Concurrent::AtomicBoolean.new(false)
  @worker_thread = Concurrent::AtomicReference.new(nil)

  # Use regular arrays protected by mutex (simpler than Concurrent::Array)
  @pending_creates = []
  @pending_updates = []
  @pending_mutex = Mutex.new

  # Separate mutex for flush operations to prevent concurrent flushes
  @flush_mutex = Mutex.new

  @flush_task = nil
  @shutdown_hook_registered = false
end

Instance Method Details

#enqueue_create(run) ⇒ Object 



68
69
70
 
# File 'lib/langsmith/batch_processor.rb', line 68

def enqueue_create(run)
  enqueue(CREATE, run)
end

#enqueue_update(run) ⇒ Object 



72
73
74
 
# File 'lib/langsmith/batch_processor.rb', line 72

def enqueue_update(run)
  enqueue(UPDATE, run)
end

#flush ⇒ Object 



76
77
78
79
80
81
82
83
84
85
 
# File 'lib/langsmith/batch_processor.rb', line 76

def flush
  ensure_started

  # Drain anything currently in the queue into pending, then flush.
  # Run a second drain pass to catch items enqueued while we were flushing.
  2.times do
    drain_queue_non_blocking
    flush_pending
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


87
88
89
 
# File 'lib/langsmith/batch_processor.rb', line 87

def running?
  @running.true?
end

#shutdown ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/langsmith/batch_processor.rb', line 53

def shutdown
  return unless @running.make_false

  @flush_task&.shutdown
  @queue << { type: SHUTDOWN }

  worker = @worker_thread.get
  if worker&.alive? && !worker.join(5)
    # Give the worker time to drain the queue gracefully
    log_error("Worker thread did not terminate within timeout", force: true)
  end

  flush_pending
end

#start ⇒ Object 



43
44
45
46
47
48
49
50
51
 
# File 'lib/langsmith/batch_processor.rb', line 43

def start
  return unless @running.make_true

  @worker_thread.set(create_worker_thread)
  @flush_task = create_flush_task
  @flush_task.execute

  register_shutdown_hook
end